提示词工程完全指南：从零基础到高级优化

1
2
3
4

prompt = """请判断这条评论的情感：
"这电影又无聊又老套，不推荐。"

情感："""

1
2
3
4
5
6
7

def zero_shot_classify(text: str, labels: list[str]) -> str:
    return f"""任务：把下面的文本分类到这些类别中的一个：
{', '.join(labels)}

文本：{text}

类别（只回复类别名称）："""

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

prompt = """判断下列评论的情感：

评论："简直爱了！今年看过最好的电影。"
情感：positive

评论："还行吧，没什么特别但也不差。"
情感：neutral

评论："纯粹浪费时间。演技稀烂。"
情感：negative

评论："摄影非常出色，剧情也抓人。"
情感："""

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

from sklearn.metrics.pairwise import cosine_similarity

def select_examples(query: str, pool: list[dict], k: int = 5) -> list[dict]:
    """挑 k 个例子：既要和问题相关，又要彼此差异大。"""
    q_emb = get_embedding(query)
    embs = [get_embedding(ex["text"]) for ex in pool]

    sims = cosine_similarity([q_emb], embs)[0]
    candidates = sims.argsort()[-k * 2:][::-1]   # 先取 2k 个最相关的

    selected = [candidates[0]]
    for idx in candidates[1:]:
        if len(selected) >= k:
            break
        if all(cosine_similarity([embs[idx]], [embs[s]])[0][0] < 0.9
               for s in selected):
            selected.append(idx)
    return [pool[i] for i in selected]

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

prompt = """请检查以下代码的反模式。

[示例 1]
代码：def get_data(): return db.query('SELECT * FROM users')
问题：原生 SQL，注入风险。改用参数化查询。
严重程度：高

[示例 2]
代码：result = api.call(); result.field
问题：没有错误处理。包一层 try/except。
严重程度：中

…… 还有 198 个示例 ……

现在请评审这段代码：
{user_code}

问题："""

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

sections = extract_sections(contract)               # 1) 拆章节

obligations = []                                    # 2) 逐节抽义务
for section in sections:
    prompt = f"""法律文本：{section}

列出所有义务，按以下格式：
[当事方] 必须 [行为] [条件]"""
    obligations.extend(llm_call(prompt))

classified = classify_obligations(obligations)      # 3) 按类型分类
summary = summarize_obligations(classified)         # 4) 汇总

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

def etl_pattern(data, extract_p, transform_p):
    extracted = llm_call(extract_p.format(data=data))
    return llm_call(transform_p.format(data=extracted))

def map_reduce(items, map_p, reduce_p):
    mapped = [llm_call(map_p.format(item=i)) for i in items]
    return llm_call(reduce_p.format(items=mapped))

def validate_retry(prompt, validator, max_retries=3):
    for _ in range(max_retries):
        result = llm_call(prompt)
        if validator(result):
            return result
        prompt += f"\n\n上一次结果不合法：{result}\n再试一次："
    raise ValueError("超过最大重试次数")

1
2
3
4

prompt = """问：罗杰有 5 个网球。他又买了 2 罐网球，每罐 3 个。
他现在有多少个网球？
答："""
# 经常输出："10"（错）

1
2
3
4
5
6
7
8

prompt = """问：罗杰有 5 个网球。他又买了 2 罐网球，每罐 3 个。
他现在有多少个网球？
答：让我们一步步想。"""
# 输出：
# 罗杰一开始有 5 个。
# 又买了 2 罐，每罐 3 个 = 6 个。
# 总共 5 + 6 = 11 个。
# 答案：11（对了）

1
2

def zero_shot_cot(question: str) -> str:
    return f"{question}\n\n 让我们一步步想。"

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

prompt = """问：5 台机器 5 分钟做 5 个零件，100 台机器做 100 个零件需要多久？
答：让我们一步步想。
   - 5 台机器 5 分钟做 5 个零件
   - 那么 1 台机器 5 分钟做 1 个零件
   - 100 台机器 5 分钟做 100 个零件
答案：5 分钟

问：农场主有 15 只羊，除了 8 只都死了，还剩几只？
答：让我们一步步想。
   - "除了 8 只都死了"意思是有 8 只活下来
   - 所以剩下 8 只
答案：8

问：{new_question}
答：让我们一步步想。"""

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

class ChainOfThoughtEngine:
    def __init__(self, model, temperature: float = 0.7):
        self.model = model
        self.temperature = temperature

    def solve(self, question: str, mode: str = "zero-shot"):
        if mode == "zero-shot":
            prompt = f"{question}\n\n 让我们一步步想。"
        elif mode == "few-shot":
            prompt = self._build_few_shot_prompt(question)
        else:
            raise ValueError(f"未知模式：{mode}")

        response = self.model.generate(
            prompt, temperature=self.temperature, max_tokens=512,
        )
        return self._extract_answer(response), response

    def _extract_answer(self, response: str) -> str:
        for marker in ("答案：", "Answer:", "因此，", "所以"):
            if marker in response:
                tail = response.split(marker)[-1].strip()
                return tail.split("。")[0].split("\n")[0].strip()
        return response.strip().split("\n")[-1]

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

from collections import Counter

def self_consistency(question: str, n: int = 5,
                     temperature: float = 0.7) -> tuple[str, float]:
    """采样 n 条推理链，对答案多数投票。"""
    prompt = f"{question}\n\n 让我们一步步想。"
    answers = []
    for _ in range(n):
        response = llm_call(prompt, temperature=temperature, max_tokens=512)
        answers.append(extract_answer(response))

    most_common, count = Counter(answers).most_common(1)[0]
    return most_common, count / n

1
2
3
4
5

def adaptive_self_consistency(question: str, threshold: float = 0.7):
    answer, conf = self_consistency(question, n=3)
    if conf >= threshold:
        return answer, conf
    return self_consistency(question, n=10)

 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

def tree_of_thoughts(problem: str, depth: int = 3, breadth: int = 3):
    def explore(state: str, d: int) -> int:
        if d >= depth:
            return evaluate_solution(state)
        thoughts = generate_thoughts(state, k=breadth)
        scored = [(t, explore(state + "\n" + t, d + 1)) for t in thoughts]
        return max(scored, key=lambda x: x[1])[1]

    return explore(problem, 0)


def generate_thoughts(state: str, k: int = 3) -> list[str]:
    prompt = f"""当前推理：
{state}

请生成 {k} 个可能的下一步：
1."""
    response = llm_call(prompt, temperature=0.8, max_tokens=300)
    return [t.strip() for t in response.split("\n") if t.strip()][:k]


def evaluate_solution(state: str) -> int:
    prompt = f"""按 1-10 给这段推理打分（逻辑连贯性、向解的进展、正确可能性）：

{state}

分数（1-10 的整数）："""
    return int(llm_call(prompt, temperature=0, max_tokens=5))

1
2
3
4
5

def graph_of_thoughts_summarize(documents: list[str]) -> str:
    summaries = [llm_call(f"摘要：\n{d}") for d in documents]
    themes = [llm_call(f"提取主题：\n{s}") for s in summaries]
    merged = llm_call("合并以下主题：\n" + "\n\n".join(themes))
    return llm_call(f"基于以下内容写最终综述：\n{merged}")

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

class ReActAgent:
    def __init__(self, model, tools: dict, max_steps: int = 10):
        self.model = model
        self.tools = tools
        self.max_steps = max_steps

    def run(self, task: str) -> str:
        trajectory = [f"Task: {task}"]
        for _ in range(self.max_steps):
            response = self.model.generate(
                self._build_prompt(trajectory), temperature=0,
            )
            thought, action, action_input = self._parse(response)
            trajectory.append(f"Thought: {thought}")
            trajectory.append(f"Action: {action}[{action_input}]")

            if action == "finish":
                return action_input

            tool = self.tools.get(action)
            obs = tool(action_input) if tool else f"未知动作：{action}"
            trajectory.append(f"Observation: {obs}")
        return "步数耗尽，未完成。"

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

def automatic_prompt_engineering(task_description, train_examples,
                                 val_examples, num_candidates=20):
    meta_prompt = f"""任务：{task_description}

示例：
{format_examples(train_examples[:5])}

请生成 {num_candidates} 个不同的提示词，用不同的措辞和角度。

提示词："""
    candidates = parse_prompts(
        llm_call(meta_prompt, temperature=1.0, max_tokens=2000)
    )

    results = []
    for prompt in candidates:
        correct = sum(
            normalize(llm_call(f"{prompt}\n\n 输入：{x}\n 输出：",
                               temperature=0)) == normalize(y)
            for x, y in val_examples
        )
        results.append((prompt, correct / len(val_examples)))

    return max(results, key=lambda x: x[1])

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

import dspy

class SentimentSignature(dspy.Signature):
    """对文本做情感分类。"""
    text = dspy.InputField()
    sentiment = dspy.OutputField(desc="positive、negative 或 neutral")

class SentimentClassifier(dspy.Module):
    def __init__(self):
        super().__init__()
        self.predictor = dspy.Predict(SentimentSignature)

    def forward(self, text: str):
        return self.predictor(text=text)

dspy.settings.configure(lm=dspy.OpenAI(model="gpt-3.5-turbo"))
classifier = SentimentClassifier()
print(classifier("这部电影简直太棒了！").sentiment)  # positive

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from dspy.teleprompt import BootstrapFewShot

train_data = [
    dspy.Example(text="产品很赞！", sentiment="positive"),
    dspy.Example(text="服务太差。", sentiment="negative"),
    # ...
]

optimized = BootstrapFewShot(metric=exact_match).compile(
    SentimentClassifier(), trainset=train_data,
)

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

class MultiHopQA(dspy.Module):
    """需要多步推理的问答。"""
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)
        self.gen_query = dspy.ChainOfThought("question -> search_query")
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question: str):
        q = self.gen_query(question=question).search_query
        ctx = self.retrieve(q).passages
        return self.answer(context=ctx, question=question).answer

1
2
3
4
5
6
7
8

from llmlingua import PromptCompressor

compressor = PromptCompressor()
compressed = compressor.compress(
    original_prompt,
    rate=0.5,            # 目标压缩率 50%
    target_token=200,    # 或者直接给一个 token 预算
)

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

class AdaptiveCompressor:
    def __init__(self, base): self.base = base

    def compress(self, prompt: str, budget: int) -> str:
        sections = split_sections(prompt)
        budgets = {
            "instruction": int(budget * 0.3),  # 永远不饿死
            "examples":    int(budget * 0.4),
            "context":     budget - int(budget * 0.7),  # 压得最狠
        }
        return merge({
            k: self.base.compress(sections[k], target_token=b)
            for k, b in budgets.items()
        })

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

def generate_structured(data: str, schema: dict) -> dict:
    """Schema 优先 + 失败时带反馈重试。"""
    prompt = f"""根据下面的 schema 生成一个 JSON 对象：
{json.dumps(schema, indent=2)}

要求：
- 必填字段不能漏
- 类型要对
- 枚举字段必须用允许的值
- 不要加多余字段

输入数据：
{data}

只输出合法 JSON："""
    response = llm_call(prompt, temperature=0)
    try:
        parsed = json.loads(response)
        validate_against_schema(parsed, schema)
        return parsed
    except Exception as e:
        return retry_with_feedback(prompt, response, str(e))

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

def generate_code(task: str, language: str = "python", tests=None) -> str:
    prompt = f"""请用 {language} 完成下面的任务。

任务：{task}

要求：
- 注释清晰
- 处理边界情况
- 遵循 {language} 最佳实践
{format_test_cases(tests) if tests else ''}

提供完整可运行的代码：
```{language}
"""
    code = extract_code_block(llm_call(prompt, temperature=0.3))
    if tests:
        results = run_tests(code, tests, language)
        if not all(r.passed for r in results):
            code = debug_and_fix(code, results, prompt)
    return code

 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

class ConversationManager:
    def __init__(self, model, system_prompt: str, max_tokens: int = 4000):
        self.model = model
        self.system_prompt = system_prompt
        self.max_tokens = max_tokens
        self.history: list[dict] = []

    def chat(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})
        context = self._get_context()
        response = self.model.generate(self._build_prompt(context))
        self.history.append({"role": "assistant", "content": response})
        return response

    def _get_context(self) -> list[dict]:
        ctx = [{"role": "system", "content": self.system_prompt}]
        used = count_tokens(self.system_prompt)
        for msg in reversed(self.history):
            t = count_tokens(msg["content"])
            if used + t > self.max_tokens:
                break
            ctx.insert(1, msg)
            used += t
        return ctx

    def summarize_old_turns(self) -> None:
        if len(self.history) < 10:
            return
        old = self.history[:6]
        summary = self.model.generate(
            f"简明地总结以下对话：\n{format_messages(old)}\n\n 总结："
        )
        self.history = (
            [{"role": "system", "content": f"早期对话摘要：{summary}"}]
            + self.history[6:]
        )

 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

def exact_match(pred, truth):
    return normalize(pred) == normalize(truth)

def f1(pred, truth):
    p = set(normalize(pred).split())
    t = set(normalize(truth).split())
    if not p or not t:
        return float(p == t)
    common = p & t
    if not common:
        return 0.0
    prec = len(common) / len(p)
    rec = len(common) / len(t)
    return 2 * prec * rec / (prec + rec)

def semantic_similarity(pred, truth, model="all-MiniLM-L6-v2"):
    from sentence_transformers import SentenceTransformer, util
    m = SentenceTransformer(model)
    return util.cos_sim(m.encode(pred), m.encode(truth)).item()

def llm_as_judge(pred, truth, criteria):
    return parse_score(llm_call(f"""请评估以下输出。
任务：{criteria['task']}
期望：{truth}
实际：{pred}

请给 0-10 分并简要说明："""))

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

class PromptExperiment:
    def __init__(self, test_set, metrics):
        self.test_set = test_set
        self.metrics = metrics

    def evaluate(self, prompt_fn):
        return {
            m.__name__: sum(
                m(llm_call(prompt_fn(x), temperature=0), y)
                for x, y in self.test_set
            ) / len(self.test_set)
            for m in self.metrics
        }

    def compare(self, variants: dict):
        import pandas as pd
        rows = [{"prompt_name": name, **self.evaluate(fn)}
                for name, fn in variants.items()]
        df = pd.DataFrame(rows)
        return df.sort_values(by=df.columns[1], ascending=False)

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

class PromptDebugger:
    AMBIGUOUS = {"相关", "合适", "好", "差", "一些", "若干", "大量",
                 "stuff", "things"}
    CONFLICT  = {"但", "然而", "尽管", "除了"}

    def check_ambiguity(self, prompt: str) -> bool:
        return any(w in prompt for w in self.AMBIGUOUS)

    def check_conflicts(self, prompt: str) -> bool:
        return sum(w in prompt for w in self.CONFLICT) >= 2

    def infer_format(self, text: str) -> str:
        t = text.strip()
        if t.startswith("{") and t.endswith("}"): return "json_object"
        if t.startswith("[") and t.endswith("]"): return "json_array"
        if "\n-" in t or "\n*" in t or "\n1." in t: return "list"
        return "prose"

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

def categorize_error(pred: str, truth: str) -> str:
    p, t = normalize(pred), normalize(truth)
    if not p:
        return "empty_output"
    if p in t or t in p:
        return "partial_match"
    p_w, t_w = set(p.split()), set(t.split())
    overlap = len(p_w & t_w) / max(len(p_w), len(t_w))
    if overlap > 0.5:
        return "semantic_error"
    if overlap > 0:
        return "partial_hallucination"
    return "complete_hallucination"