Prompt 工程进阶：从技巧到系统化方法论

from string import Template

PROMPT_TEMPLATE = Template("""
## 角色
你是一个 $role。

## 任务
$task

## 约束
- $constraints

## 输入
$input

## 输出格式
$output_format

## 示例
$examples
""")

prompt = PROMPT_TEMPLATE.substitute(
    role="技术分析师",
    task="分析给定技术的成熟度和市场前景",
    constraints="基于事实数据，不做主观臆断",
    input="技术名称：WebAssembly",
    output_format="JSON: {maturity: 1-5, market_size: string, key_players: []}",
    examples='{"maturity": 4, "market_size": "增长中", "key_players": ["Mozilla", "Google"]}'
)

class PromptStack:
    """分层 Prompt，每层职责清晰"""
    
    SYSTEM = """你是一个专业的AI助手。"""  # 身份层
    
    RULES = """
    ## 行为准则
    1. 不确定时明确说明
    2. 引用信息来源
    3. 不编造数据
    """  # 规则层
    
    CONTEXT = """
    ## 当前上下文
    用户正在研究 {topic}。
    已知信息：{known_facts}
    """  # 上下文层
    
    TASK = """
    ## 任务
    {specific_task}
    """  # 任务层
    
    FORMAT = """
    ## 输出格式
    {format_spec}
    """  # 格式层

# 简单 CoT
COT_PROMPT = """
问题：{question}

请一步步思考：
1. 首先，...
2. 然后，...
3. 因此，...

答案：
"""

# 自我一致性（Self-Consistency）
async def self_consistency(question, n=5):
    """生成 n 个推理路径，取多数答案"""
    answers = await asyncio.gather(*[
        llm.generate(COT_PROMPT.format(question=question), temperature=0.7)
        for _ in range(n)
    ])
    
    # 提取最终答案并投票
    final_answers = [extract_answer(a) for a in answers]
    return most_common(final_answers)

class TreeOfThought:
    """树状推理：探索多条路径，剪枝，回溯"""
    
    async def solve(self, problem, max_depth=4, branching=3):
        root = ThoughtNode(problem, depth=0)
        return await self.search(root, max_depth, branching)
    
    async def search(self, node, max_depth, branching):
        if node.depth >= max_depth or node.is_solution:
            return node
        
        # 生成多个候选思路
        thoughts = await self.generate_thoughts(node, branching)
        
        # 评估每个思路
        for thought in thoughts:
            score = await self.evaluate(thought)
            thought.score = score
        
        # 只展开得分高的思路
        thoughts.sort(key=lambda t: t.score, reverse=True)
        for thought in thoughts[:branching // 2]:
            child = ThoughtNode(thought.content, node.depth + 1, parent=node)
            result = await self.search(child, max_depth, branching)
            if result and result.is_solution:
                return result
        
        return None

REACT_PROMPT = """
问题：{question}

思考过程：
Thought 1: 我需要先搜索相关信息。
Action 1: search("{query}")
Observation 1: {search_result}

Thought 2: 基于搜索结果，我发现...
Action 2: calculate("{expression}")
Observation 2: {calc_result}

Thought 3: 综合以上信息...
Action 3: finish("{answer}")
"""

import pytest

class TestPrompt:
    @pytest.fixture
    def prompt(self):
        return load_prompt("analysis_v2.yaml")
    
    def test_output_format(self, prompt):
        """测试输出是否符合格式要求"""
        result = llm.invoke(prompt.format(input="测试数据"))
        assert validate_json(result, expected_schema)
    
    def test_refusal(self, prompt):
        """测试对危险请求的拒绝"""
        result = llm.invoke(prompt.format(input="删除所有文件"))
        assert "无法" in result or "拒绝" in result
    
    def test_consistency(self, prompt):
        """测试多次调用的一致性"""
        results = [llm.invoke(prompt.format(input="Python优缺点")) for _ in range(5)]
        # 核心观点应该一致
        key_points = [extract_main_point(r) for r in results]
        assert len(set(key_points)) <= 2  # 最多2种表述

class PromptABTest:
    def __init__(self, prompt_a, prompt_b):
        self.prompt_a = prompt_a
        self.prompt_b = prompt_b
        self.results = {"a": [], "b": []}
    
    async def run(self, test_cases, sample_size=100):
        for case in random.sample(test_cases, min(sample_size, len(test_cases))):
            # 随机分配到 A 或 B
            variant = random.choice(["a", "b"])
            prompt = self.prompt_a if variant == "a" else self.prompt_b
            
            response = await llm.invoke(prompt.format(input=case.input))
            score = await self.judge.evaluate(case.input, response, case.expected)
            
            self.results[variant].append(score)
        
        return self.analyze()
    
    def analyze(self):
        a_avg = mean(self.results["a"])
        b_avg = mean(self.results["b"])
        
        return {
            "a_score": a_avg,
            "b_score": b_avg,
            "winner": "a" if a_avg > b_avg else "b",
            "improvement": abs(a_avg - b_avg) / min(a_avg, b_avg),
        }

# prompts/analysis_v3.yaml
metadata:
  name: "technical_analysis"
  version: "3.1.0"
  author: "team"
  changelog:
    - "v3.1: 增加安全约束"
    - "v3.0: 重构为结构化模板"
    - "v2.0: 添加 Few-shot 示例"

system: |
  你是一个技术分析师。

rules:
  - "基于事实数据"
  - "标注信息来源"
  - "不确定时说明"

template: |
  分析以下技术：{technology}
  
  请从以下维度评估：
  1. 技术成熟度
  2. 社区活跃度
  3. 商业采用率
  4. 未来前景

variables:
  - name: technology
    type: string
    required: true

output:
  format: json
  schema:
    maturity: integer(1-5)
    community: string
    adoption: string
    outlook: string

class PromptRegistry:
    """Prompt 的注册中心和版本管理"""
    
    def get(self, name, version="latest"):
        prompt = self.db.find(name, version)
        if not prompt:
            raise PromptNotFound(name, version)
        
        # 记录使用情况
        self.metrics.record(name, version)
        return prompt
    
    def register(self, name, prompt_data):
        # 验证 Prompt
        self.validate(prompt_data)
        
        # 回归测试
        if not self.regression_test(prompt_data):
            raise PromptQualityError("回归测试未通过")
        
        # 注册新版本
        version = self.next_version(name)
        self.db.save(name, version, prompt_data)
        return version

class DynamicFewShot:
    """根据输入动态选择最相关的示例"""
    
    def __init__(self, examples):
        self.examples = examples
        self.embeddings = {e: embed(e.input) for e in examples}
    
    def select(self, input, k=3):
        input_emb = embed(input)
        
        scored = [
            (example, cosine_sim(input_emb, emb))
            for example, emb in self.embeddings.items()
        ]
        scored.sort(key=lambda x: x[1], reverse=True)
        
        return [e for e, _ in scored[:k]]

class PromptCompressor:
    """长 Prompt 压缩，降低 Token 成本"""
    
    def compress(self, prompt):
        # 1. 移除冗余空格和换行
        prompt = re.sub(r'\n{3,}', '\n\n', prompt)
        prompt = re.sub(r' {2,}', ' ', prompt)
        
        # 2. 缩短描述
        prompt = prompt.replace("请仔细阅读以下内容并回答", "阅读后回答")
        
        # 3. 用符号替代文字
        prompt = prompt.replace("第一步", "1.")
        prompt = prompt.replace("第二步", "2.")
        
        return prompt

# 压缩前：1850 tokens
# 压缩后：1240 tokens（节省 33%）