Prompt工程

Prompt as Code：理念 代码有 Git，有 CI/CD，有 code review，有单元测试。Prompt 呢？大多数团队的 Prompt 管理方式相当于把代码写在记事本里，用文件名标记版本。 Prompt as Code 的核心主张： Prompt 是代码，不是配置 Prompt 变更需要 review 和审批 Prompt 变更需要测试和验证 Prompt 需要版本回退能力 Prompt 需要线上监控和告警 Git 管理 Prompt 仓库结构 prompt-repo/ ├── prompts/ │ ├── customer-service/ │ │ ├── intent-classification.yaml │ │ ├── response-generation.yaml │ │ └── escalation.yaml │ ├── data-analysis/ │ │ ├── sql-generation.yaml │ │ └── insight-summary.yaml │ └── _shared/ │ ├── system-prompts.yaml │ └── safety-rules.yaml ├── tests/ │ ├── golden-sets/ │ │ ├── customer-service-golden.jsonl │ │ └── data-analysis-golden.jsonl │ └── regression/ │ └── test_regression.py ├── eval/ │ ├── evaluators.py │ └── metrics.py ├── .promptlab.yaml # 工具配置 └── CHANGELOG.md Prompt 文件规范 # prompts/customer-service/intent-classification.yaml id: cs-intent-classification name: "客服意图分类" version: "2.3.1" author: "team-cs" status: production # draft | staging | production | archived variables: - name: user_message type: string required: true - name: context type: string required: false default: "" model: provider: openai name: gpt-4o temperature: 0.1 max_tokens: 256 template: | 系统：你是客服意图分类器。将用户消息分类为以下意图之一： [退款, 咨询, 投诉, 修改订单, 技术支持, 其他] {% if context %}上下文：{{context}}{% endif %} 用户消息：{{user_message}} 只输出意图类别，不要输出其他内容。 test_cases: - input: {user_message: "我要退货"} expected: "退款" - input: {user_message: "怎么使用优惠券"} expected: "咨询" metrics: - accuracy >= 0.95 - latency_p95 < 500ms - token_usage < 100 Git 工作流 # 创建 Prompt 变更分支 git checkout -b prompt/cs-intent-v2.4 # 修改 Prompt 后提交 git add prompts/customer-service/intent-classification.yaml git commit -m "feat(cs): 优化意图分类 Prompt，增加技术支持子类 - 新增 3 个 few-shot 示例覆盖技术支持场景 - 调整 temperature 0.2 → 0.1 减少随机性 - 黄金集准确率 92.3% → 96.1% - Closes #142" # CI 自动跑回归测试 git push origin prompt/cs-intent-v2.4 A/B 测试框架 架构 class PromptABTest: def __init__(self, config): self.control = config['control'] # 当前生产版本 self.treatment = config['treatment'] # 候选版本 self.split_ratio = config.get('split', 0.1) # 10% 流量到 treatment self.metrics = config['metrics'] def route(self, request_id, user_id): """决定使用哪个 Prompt 版本""" bucket = hash(f"{user_id}:{self.experiment_id}") % 100 if bucket < self.split_ratio * 100: return self.treatment return self.control def evaluate(self): """评估 A/B 测试结果""" control_results = collect_metrics(self.control) treatment_results = collect_metrics(self.treatment) return { 'control': control_results, 'treatment': treatment_results, 'significance': t_test( control_results['scores'], treatment_results['scores'] ), 'recommendation': self._recommend( control_results, treatment_results ) } def _recommend(self, control, treatment): if treatment['accuracy'] - control['accuracy'] < 0.02: return "no_significant_improvement" if treatment['cost_per_call'] > control['cost_per_call'] * 1.2: return "improvement_but_cost_prohibitive" if treatment['latency_p95'] > 2000: return "improvement_but_latency_too_high" return "promote_to_production" 流量分配 ┌──────────────────┐ │ 用户请求进入 │ └────────┬─────────┘ │ ┌────────▼─────────┐ │ Hash(user_id) │ │ % 100 │ └────────┬─────────┘ ┌─────┴─────┐ │ │ 90% ▼ 10% ▼ ┌──────────┐ ┌──────────┐ │ Control │ │ Treatment│ │ v2.3.1 │ │ v2.4.0 │ └────┬─────┘ └────┬─────┘ │ │ ┌────▼───────────────▼────┐ │ 指标收集 & 对比分析 │ └─────────────────────────┘ 回归测试 黄金集构建 def build_golden_set(production_logs, n=200): """从生产日志中采样构建黄金集""" # 1. 采样 samples = stratified_sample(production_logs, n) # 2. 人工标注/确认 golden = [] for sample in samples: golden.append({ 'input': sample.input, 'expected_output': sample.human_verified_output, 'min_quality_score': 0.85, 'category': sample.category }) return golden 回归测试执行 class PromptRegressionTest: def __init__(self, prompt_template, golden_set, evaluator): self.template = prompt_template self.golden_set = golden_set self.evaluator = evaluator def run(self, model_config): results = [] for case in self.golden_set: prompt = self.template.render(**case['input']) output = llm_call(prompt, **model_config) score = self.evaluator(output, case['expected_output']) results.append({ 'case_id': case.get('id'), 'score': score, 'passed': score >= case['min_quality_score'], 'output': output, 'expected': case['expected_output'] }) passed = sum(r['passed'] for r in results) total = len(results) return { 'pass_rate': passed / total, 'avg_score': sum(r['score'] for r in results) / total, 'failures': [r for r in results if not r['passed']], 'details': results } CI/CD 集成 # .github/workflows/prompt-ci.yml name: Prompt CI on: pull_request: paths: ['prompts/**'] jobs: regression-test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Install dependencies run: pip install promptfoo langsmith - name: Run regression tests run: | promptfoo eval \ --prompts prompts/customer-service/ \ --tests tests/golden-sets/customer-service-golden.jsonl \ --threshold 0.95 \ --output results.json - name: Check for regressions run: | python scripts/check_regression.py results.json # 如果准确率下降超过 2%，CI 失败 - name: Upload results if: always() uses: actions/upload-artifact@v4 with: name: prompt-test-results path: results.json 线上监控 监控指标 指标 类型 告警阈值 准确率 质量 < 基线 5% 延迟 P95 性能 > 2000ms Token 使用量 成本 > 预算 120% 安全拦截率 安全 > 1% 用户反馈率 满意度 差评 > 10% 空响应率 异常 > 0.5% class PromptMonitor: def __init__(self, prompt_id, version): self.prompt_id = prompt_id self.version = version self.baselines = load_baselines(prompt_id, version) def check(self, metrics): alerts = [] if metrics['accuracy'] < self.baselines['accuracy'] - 0.05: alerts.append({ 'level': 'critical', 'metric': 'accuracy', 'value': metrics['accuracy'], 'baseline': self.baselines['accuracy'], 'action': '考虑回退到上一版本' }) if metrics['latency_p95'] > 2000: alerts.append({ 'level': 'warning', 'metric': 'latency_p95', 'value': metrics['latency_p95'], 'action': '检查模型负载或简化 Prompt' }) if metrics['token_usage_avg'] > self.baselines['token_usage'] * 1.2: alerts.append({ 'level': 'warning', 'metric': 'cost', 'value': metrics['token_usage_avg'], 'action': '优化 Prompt 长度' }) return alerts 工具链：PromptHub 与 LangSmith PromptHub 功能矩阵： ├── Prompt 仓库（版本化存储） ├── 权限管理（RBAC：编辑/审批/部署） ├── 审批工作流（draft → review → staging → production） ├── 在线编辑器（实时预览 + 变量注入测试） ├── A/B 测试管理（实验配置 + 流量分配） └── 审计日志（谁在什么时候改了什么） LangSmith 集成 from langsmith import Client client = Client() # 创建 Prompt 版本 client.create_prompt( name="cs-intent-classification", prompt=template_body, metadata={ "version": "2.4.0", "author": "team-cs", "change_type": "minor" } ) # 线上追踪 @client.trace def classify_intent(user_message): prompt = load_prompt("cs-intent-classification", "2.4.0") response = llm_call(prompt.render(user_message=user_message)) client.record_evaluation( run_id=run.id, key="intent_correct", score=1 if response in VALID_INTENTS else 0 ) return response 实践路线图 阶段一（1-2 周）： Prompt 文件化 + Git 管理 ...

核心问题 “我应该用 Prompt 工程还是微调？“答案是：取决于瓶颈在哪。 模型理解能力不足 → 微调 上下文信息不足 → RAG / Prompt 工程 输出格式不稳定 → Prompt 工程（或少量微调） 领域知识缺失 → RAG（而非微调） Prompt 工程的能力边界 能做什么 # 1. 角色设定和行为控制 SYSTEM_PROMPT = "你是专业法律顾问，仅基于提供的条文回答" # 2. 输出格式控制 FORMAT_PROMPT = '输出JSON: {"risk_level":"high|medium|low", "factors":[]}' # 3. 少样本学习 FEW_SHOT = "示例：\n输入：订单没到\n输出：{\"intent\":\"logistics\"}\n现在处理：{input}" # 4. 思维链推理 COT = "请一步步思考：首先分析...然后...最后..." 天花板 # 解决不了的问题： # 1. 风格深度定制 — 需要大量token示例，成本高且不稳定 style_prompt = "请模仿以下风格（附10000字示例）..." # token成本爆炸 # 2. 领域术语理解 — Prompt过长，模型注意力分散 medical_prompt = "医学缩写对照表（500个）..." # 效果差 # 3. 复杂工具调用模式 — 规则太多时遵循率下降 tool_prompt = "调用工具的50条注意事项..." # 遵循率低 微调的适用场景 场景一：风格和格式定制 # 微调数据：客服风格定制 training_data = [ {"messages": [ {"role": "system", "content": "你是XX品牌客服"}, {"role": "user", "content": "你们的东西太贵了"}, {"role": "assistant", "content": "理解您的感受～我们坚持品质路线，现有新人9折优惠😊"}, ]}, # 500-1000条这样的数据 ] # 微调后：无需长Prompt，模型自然使用品牌语气 场景二：降低推理成本 # 微调前：GPT-4o + 2000 token system prompt → $0.017/请求 # 微调后：GPT-4o-mini fine-tuned → $0.0015/请求 # 成本降低91%，回本周期约0.5个月 # 蒸馏：用GPT-4生成数据训练小模型 distillation_data = [] for input_text in training_inputs: gpt4_output = await gpt4.generate(input_text, system_prompt=LONG_PROMPT) distillation_data.append({"input": input_text, "output": gpt4_output}) 成本对比 一次性成本 项目 Prompt 工程 微调 数据标注 $0 $500-$5000 开发时间 数小时-数天 1-2 周 训练计算 $0 $10-$100（LoRA） 运行时成本 每天10000次请求对比：Prompt工程(GPT-4o+2000token)月$5100 vs 微调(GPT-4o-mini-ft)月$472，回本约13天。 ...

Prompt 工程不是"写咒语" 很多人以为 Prompt 工程就是"找到对的咒语让 AI 听话"。这是 2023 年的认知。2026 年的 Prompt 工程是一门系统工程：有设计原则、有测试方法、有版本管理、有评估指标。 第一层：结构化 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} """ # 格式层 第二层：推理增强 Chain-of-Thought（CoT） # 简单 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) Tree-of-Thought（ToT） 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（推理+行动） 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}") """ 第三层：Prompt 测试 单元测试 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种表述 A/B 测试 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), } 第四层：Prompt 管理 版本控制 # 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 Prompt 注册中心 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 高级技巧 动态 Few-shot 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]] Prompt 压缩 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%） 评估指标 指标 说明 目标 准确率 答案正确的比例 >90% 格式合规率 输出格式符合要求 >98% 拒绝率 对危险请求的拒绝率 100% 一致性 同输入多次输出一致 >85% Token 效率 输出质量/Token数 越高越好 延迟 首字节延迟 <2s 结论 Prompt 工程在 2026 年已经从"调参数"演进为"系统工程"： ...