自然语言处理（七）：提示工程与 In-Context Learning

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from dataclasses import dataclass, field

@dataclass
class Prompt:
    system: str = ""
    instruction: str = ""
    examples: list[tuple[str, str]] = field(default_factory=list)
    query: str = ""
    format_spec: str = ""

    def render(self) -> str:
        parts: list[str] = []
        if self.system:
            parts.append(f"[SYSTEM]\n{self.system}")
        if self.instruction:
            parts.append(f"[TASK]\n{self.instruction}")
        if self.examples:
            shots = "\n\n".join(
                f"Input: {x}\nOutput: {y}" for x, y in self.examples
            )
            parts.append(f"[EXAMPLES]\n{shots}")
        if self.format_spec:
            parts.append(f"[FORMAT]\n{self.format_spec}")
        if self.query:
            parts.append(f"[INPUT]\n{self.query}\nOutput:")
        return "\n\n".join(parts)

1
2
3
4

zero_shot = """将句子的情感分类为 positive、negative 或 neutral。

句子：这部电影剧情精彩、表演出色。
情感："""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

few_shot = """将句子的情感分类为 positive、negative 或 neutral：

句子：今天天气真好，阳光明媚。
情感：positive

句子：这个产品质量差、价格还贵。
情感：negative

句子：明天会下雨。
情感：neutral

句子：这家餐厅的服务令人印象深刻。
情感："""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

问题：一本书有 120 页。第一天读了 30 页，第二天读了第一天的两倍，
第三天读了剩下页数的一半。第三天读了多少页？

一步步推导：
1. 第一天：30 页。
2. 第二天：2 × 30 = 60 页。
3. 累计已读：30 + 60 = 90 页。
4. 剩下：120 − 90 = 30 页。
5. 第三天：30 / 2 = 15 页。

答案：15 页。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import statistics, random
from typing import Callable

def evaluate(
    build_prompt: Callable[[str, list], str],
    model_call: Callable[[str], str],
    cases: list[dict],
    examples_pool: list[dict],
    k: int = 4,
    n_seeds: int = 5,
) -> dict:
    """通过 n_seeds 次不同的示例采样，返回准确率的中位数及其波动范围。"""
    accs = []
    for seed in range(n_seeds):
        rng = random.Random(seed)
        shots = rng.sample(examples_pool, k)
        correct = 0
        for case in cases:
            out = model_call(build_prompt(case["input"], shots))
            if out.strip() == case["expected"].strip():
                correct += 1
        accs.append(correct / len(cases))
    return {
        "median": statistics.median(accs),
        "min": min(accs),
        "max": max(accs),
        "spread": max(accs) - min(accs),
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from collections import Counter

def self_consistency(
    model_call,
    prompt: str,
    k: int = 8,
    temperature: float = 0.7,
    extract_answer=lambda s: s.strip().split()[-1],
) -> dict:
    """采样 k 条 CoT 推理路径，并对最终答案进行多数投票"""
    answers = []
    for _ in range(k):
        out = model_call(prompt, temperature=temperature)
        answers.append(extract_answer(out))
    counts = Counter(answers)
    top, n = counts.most_common(1)[0]
    return {"answer": top, "confidence": n / k, "votes": dict(counts)}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import re
from typing import Callable

class ReActAgent:
    def __init__(self, model_call: Callable[[str], str],
                 tools: dict[str, Callable[[str], str]]):
        self.model_call = model_call
        self.tools = tools

    def _system(self) -> str:
        names = ", ".join(self.tools)
        return (
            "可用工具：" + names + "。\n"
            "请严格遵循以下格式：\n"
            "Thought: <推理内容>\n"
            "Action: <工具名>(<参数>)\n"
            "Observation: <工具返回结果>\n"
            "...重复上述步骤...\n"
            "Answer: <最终答案>"
        )

    def run(self, question: str, max_steps: int = 6) -> str:
        ctx = f"{self._system()}\n\n问题：{question}\n"
        for _ in range(max_steps):
            out = self.model_call(ctx, stop=["\nObservation:"])
            ctx += out
            if "Answer:" in out:
                return out.split("Answer:")[-1].strip()
            m = re.search(r"Action:\s*(\w+)\((.*?)\)", out)
            if not m:
                return "(无法解析 Action)"
            name, arg = m.group(1), m.group(2).strip()
            obs = self.tools[name](arg) if name in self.tools \
                  else f"未知工具：{name}"
            ctx += f"\nObservation: {obs}\n"
        return "(达到最大步数限制)"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from pathlib import Path

class PromptRegistry:
    def __init__(self, root: Path):
        self.root = root
        self._cache: dict[str, str] = {}

    def get(self, name: str, version: str = "latest") -> str:
        key = f"{name}@{version}"
        if key not in self._cache:
            path = self.root / name / f"{version}.txt"
            self._cache[key] = path.read_text()
        return self._cache[key]

1
2
3
4
5
6
7

def regression_check(old_prompt: str, new_prompt: str,
                     cases, model_call, judge) -> bool:
    old_acc = mean(judge(case, model_call(old_prompt + case.input))
                   for case in cases)
    new_acc = mean(judge(case, model_call(new_prompt + case.input))
                   for case in cases)
    return new_acc >= old_acc - 0.01   # 允许 1% 的准确率下降

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def build_advanced(question: str, examples: list[dict],
                   *, role: str = "", use_cot: bool = True) -> str:
    parts: list[str] = []
    if role:
        parts.append(f"你是 {role}。")
    parts.append("请解决以下问题。")
    for ex in examples:
        parts.append(f"Q: {ex['question']}")
        if use_cot and "reasoning" in ex:
            parts.append(f"Reasoning: {ex['reasoning']}")
        parts.append(f"A: {ex['answer']}\n")
    if use_cot:
        parts.append("先一步步推理，再在以 'A:' 开头的一行里给出最终答案。")
    parts.append(f"Q: {question}\n")
    return "\n".join(parts)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

import json, re
from collections import Counter

SYSTEM = (
    "你是一个严谨的文本分类器。只返回合法的 JSON 格式结果，不要添加任何注释或说明。"
)

EXAMPLES = [
    ("这部电影太棒了，演员表现出色，剧情扣人心弦。",
     {"sentiment": "positive", "confidence": 0.95}),
    ("电池续航不到三小时，软件还频繁崩溃，体验极差。",
     {"sentiment": "negative", "confidence": 0.92}),
    ("包裹周二准时送达，没什么特别的问题。",
     {"sentiment": "neutral", "confidence": 0.88}),
]

FORMAT_SPEC = (
    '输出格式必须为：{"sentiment": "<positive|negative|neutral>", '
    '"confidence": <0.0-1.0>}'
)

def build_prompt(text: str) -> str:
    shots = "\n".join(
        f"Text: {x}\nOutput: {json.dumps(y, ensure_ascii=False)}"
        for x, y in EXAMPLES
    )
    return (
        f"[SYSTEM]\n{SYSTEM}\n\n"
        f"[FORMAT]\n{FORMAT_SPEC}\n\n"
        f"[EXAMPLES]\n{shots}\n\n"
        f"[INPUT]\nText: {text}\nOutput:"
    )

def parse(raw: str) -> dict | None:
    m = re.search(r"\{.*\}", raw, re.DOTALL)
    try:
        return json.loads(m.group(0)) if m else None
    except json.JSONDecodeError:
        return None

def classify(model_call, text: str, *, k: int = 5) -> dict:
    """通过 Self-Consistency 提升情感分类的稳定性。"""
    prompt = build_prompt(text)
    labels, confs = [], []
    for _ in range(k):
        parsed = parse(model_call(prompt, temperature=0.5))
        if parsed and "sentiment" in parsed:
            labels.append(parsed["sentiment"])
            confs.append(float(parsed.get("confidence", 0.5)))
    if not labels:
        return {"sentiment": "unknown", "confidence": 0.0, "votes": {}}
    counts = Counter(labels)
    top, n = counts.most_common(1)[0]
    return {"sentiment": top,
            "confidence": n / len(labels),
            "model_confidence": sum(confs) / len(confs),
            "votes": dict(counts)}