第14章 Prompt 编排和优化技术

当面对复杂的生成任务时，单次提示往往难以获得理想的结果。Prompt 编排技术通过将任务分解为多个子任务、并行执行多个提示、或动态调整提示策略，显著提升了大语言模型处理复杂问题的能力。

本章将介绍五种先进的 Prompt 编排与优化技术，从串行链式提示到自动优化工具，帮助您构建高效、可靠的提示工程流水线。

14.1 Prompt Chaining：串行链式提示设计

14.1.1 核心概念与原理

Prompt Chaining（提示链） 是将复杂任务分解为一系列顺序执行的子任务，每个子任务的输出作为下一个子任务的输入。这种模式模拟了人类解决问题的分步思维过程。

┌─────────────────────────────────────────────────────────────────┐│ Prompt Chaining 模式│├─────────────────────────────────────────────────────────────────┤│ ││输入│││ ││▼ ││┌─────────────┐┌─────────────┐┌─────────────┐ │││Prompt 1 │───▶│Prompt 2 │───▶│Prompt 3 │ │││(子任务1) ││(子任务2) ││(子任务3) │ ││└──────┬──────┘└──────┬──────┘└──────┬──────┘ ││ │││││ ▼▼▼││ 输出1输出2输出3 ││ │││││ └──────────────────┴──────────────────┘│││││▼││最终输出││ │└─────────────────────────────────────────────────────────────────┘

核心优势：

• ✅ 降低复杂度：每个子任务更简单，模型更容易正确完成
• ✅ 可解释性强：中间结果可见，便于调试和优化
• ✅ 质量可控：可以在每个步骤进行质量检查
• ✅ 灵活组合：不同子任务可以使用不同模型或参数

14.1.2 实战案例：营销文案生成

场景：生成一篇完整的营销文案，包括大纲、正文和润色。

from typing import List, Dictimport openaiclass MarketingCopyChain:"""营销文案生成链"""def __init__(self, model: str = "gpt-4"):self.model = modelself.client = openai.OpenAI()def step1_generate_outline(self, product: str, target_audience: str) -> Dict:"""步骤1：生成文案大纲"""prompt = f"""为以下产品生成营销文案大纲。产品：{product}目标受众：{target_audience}请生成包含以下要素的大纲：1. 标题选项（3个）2. 核心卖点（3-5个）3. 文案结构（开头、主体、结尾）4. 呼吁行动（CTA）以JSON格式输出：{{"headlines": [...],"key_selling_points": [...],"structure": {{"intro": "...", "body": "...", "conclusion": "..."}},"cta": "..."}}"""response = self.client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],temperature=0.7)import jsonreturn json.loads(response.choices[0].message.content)def step2_write_draft(self, outline: Dict, tone: str = "professional") -> str:"""步骤2：根据大纲撰写初稿"""prompt = f"""根据以下大纲撰写营销文案初稿。大纲：- 标题：{outline['headlines'][0]}- 核心卖点：{', '.join(outline['key_selling_points'])}- 结构：{outline['structure']}- CTA：{outline['cta']}要求：- 语气：{tone}- 字数：500-800字- 包含所有核心卖点- 有吸引力的开头和有力的结尾文案："""response = self.client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],temperature=0.8)return response.choices[0].message.contentdef step3_polish(self, draft: str, improvement_areas: List[str]) -> str:"""步骤3：润色文案"""prompt = f"""润色以下营销文案。原文：{draft}需要改进的方面：{chr(10).join(['- ' + area for area in improvement_areas])}润色要求：1. 保持原意和核心信息2. 提升语言流畅度和吸引力3. 确保语气一致4. 优化段落过渡润色后的文案："""response = self.client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],temperature=0.5)return response.choices[0].message.contentdef generate(self, product: str, target_audience: str) -> Dict:"""执行完整链"""# 步骤1：生成大纲outline = self.step1_generate_outline(product, target_audience)# 步骤2：撰写初稿draft = self.step2_write_draft(outline)# 步骤3：润色final = self.step3_polish(draft, ["增强情感共鸣", "优化标题吸引力"])return {"outline": outline,"draft": draft,"final": final}# 使用示例chain = MarketingCopyChain()result = chain.generate(product="智能降噪耳机 Pro",target_audience="25-40岁城市白领")

14.1.3 错误处理与重试机制

class RobustPromptChain:"""带错误处理的健壮提示链"""def __init__(self, max_retries: int = 3):self.max_retries = max_retriesdef execute_step(self, prompt: str, validator=None) -> str:"""执行单步，带重试和验证"""for attempt in range(self.max_retries):try:response = self.call_llm(prompt)# 如果有验证器，验证输出if validator:is_valid, error_msg = validator(response)if not is_valid:if attempt < self.max_retries - 1:prompt += f"nn注意：上次输出存在问题：{error_msg}。请修正。"continueelse:raise ValueError(f"验证失败：{error_msg}")return responseexcept Exception as e:if attempt == self.max_retries - 1:raise eprint(f"尝试 {attempt + 1} 失败，重试...")raise RuntimeError("所有重试都失败了")def validate_json_output(self, response: str) -> tuple:"""验证JSON输出"""try:import jsonjson.loads(response)return True, ""except json.JSONDecodeError as e:return False, f"无效的JSON格式：{str(e)}"

14.1.4 Anthropic 的 Chain Complex Prompts 最佳实践

Anthropic 官方建议的链式提示设计原则：

1. 明确每个步骤的输出格式：使用 XML 标签或 JSON 结构化输出
2. 传递足够的上下文：每个步骤的提示应包含完成该步骤所需的所有信息
3. 设置检查点：在关键步骤后进行质量验证
4. 保持步骤独立性：每个步骤应尽可能独立，减少错误传播

14.2 并行提示与结果聚合策略

14.2.1 核心概念

并行提示（Parallelization） 是同时执行多个独立的 LLM 调用，然后聚合结果。这种模式适用于需要多视角评估或高置信度结果的场景。

┌─────────────────────────────────────────────────────────────────┐│ Parallelization 模式│├─────────────────────────────────────────────────────────────────┤│ ││ 输入 ││ │││ ┌─────────────────┼─────────────────┐ ││ │ │ │││ ▼ ▼ ▼││ ┌─────────┐ ┌─────────┐ ┌─────────┐││ │Prompt A │ │Prompt B │ │Prompt C │││ │(角度1)│ │(角度2)│ │(角度3)│││ └────┬────┘ └────┬────┘ └────┬────┘│││ │ │││▼ ▼ ▼││ 结果A结果B结果C │││ │ │││ └─────────────────┼─────────────────┘ ││ │││ ▼││┌─────────────┐│││聚合策略││││ (投票/合并)│││└──────┬──────┘││ │││ ▼││最终输出││ │└─────────────────────────────────────────────────────────────────┘

14.2.2 实战案例：代码安全审查

场景：从多个安全角度并行审查代码，然后聚合结果。

import asynciofrom concurrent.futures import ThreadPoolExecutorfrom typing import List, Dictimport openaiclass ParallelSecurityReviewer:"""并行代码安全审查器"""def __init__(self):self.client = openai.OpenAI()self.review_angles = [{"name": "SQL注入检查","prompt": "从SQL注入角度审查以下代码，识别潜在漏洞。列出发现的问题和风险等级（高/中/低）。"},{"name": "XSS攻击检查","prompt": "从XSS（跨站脚本）攻击角度审查以下代码，识别潜在漏洞。列出发现的问题和风险等级。"},{"name": "权限控制检查","prompt": "从权限控制和访问控制角度审查以下代码，识别潜在问题。列出发现的问题和风险等级。"},{"name": "敏感数据检查","prompt": "从敏感数据保护角度审查以下代码，检查是否有数据泄露风险。列出发现的问题和风险等级。"}]async def review_from_angle(self, code: str, angle: Dict) -> Dict:"""从特定角度审查代码"""prompt = f"""{angle['prompt']}

代码：

请以JSON格式输出：{{"angle": "{angle['name']}","issues_found": [{{"line": "行号或代码片段","description": "问题描述","risk_level": "高/中/低","recommendation": "修复建议"}}],"summary": "总体评估"}}"""response = await asyncio.to_thread(self.client.chat.completions.create,model="gpt-4",messages=[{"role": "user", "content": prompt}],temperature=0)import jsonresult = json.loads(response.choices[0].message.content)return resultasync def parallel_review(self, code: str) -> Dict:"""并行执行多角度审查"""# 并发执行所有审查角度tasks = [self.review_from_angle(code, angle)for angle in self.review_angles]results = await asyncio.gather(*tasks)# 聚合结果return self.aggregate_results(results)def aggregate_results(self, results: List[Dict]) -> Dict:"""聚合多角度审查结果"""all_issues = []high_risk_count = 0for result in results:for issue in result.get("issues_found", []):all_issues.append({**issue,"detected_by": result["angle"]})if issue.get("risk_level") == "高":high_risk_count += 1# 去重：相同位置的问题合并unique_issues = self.deduplicate_issues(all_issues)# 生成综合报告return {"total_issues": len(unique_issues),"high_risk_issues": high_risk_count,"issues": unique_issues,"overall_risk": "高" if high_risk_count > 0 else "中" if unique_issues else "低","recommendation": "立即修复高风险问题" if high_risk_count > 0 else "建议修复中低风险问题"}def deduplicate_issues(self, issues: List[Dict]) -> List[Dict]:"""去重：基于代码位置"""seen = set()unique = []for issue in issues:key = issue.get("line", "") + issue.get("description", "")[:50]if key not in seen:seen.add(key)unique.append(issue)return unique# 使用示例async def main():reviewer = ParallelSecurityReviewer()code = """def login(username, password):query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"return db.execute(query)"""result = await reviewer.parallel_review(code)print(f"发现 {result['total_issues']} 个问题，{result['high_risk_issues']} 个高风险")asyncio.run(main())

14.2.3 结果聚合策略

策略一：投票（Voting）

def voting_aggregation(responses: List[str]) -> str:"""多数投票选择最常见答案"""from collections import Countervote_counts = Counter(responses)return vote_counts.most_common(1)[0][0]

策略二：加权合并（Weighted Merge）

def weighted_merge(responses: List[Dict], weights: List[float]) -> Dict:"""根据置信度加权合并结果"""merged = {}for response, weight in zip(responses, weights):for key, value in response.items():if key not in merged:merged[key] = []merged[key].append((value, weight))# 加权平均或选择最高权重result = {}for key, weighted_values in merged.items():if isinstance(weighted_values[0][0], (int, float)):# 数值类型：加权平均total_weight = sum(w for _, w in weighted_values)result[key] = sum(v * w for v, w in weighted_values) / total_weightelse:# 非数值类型：选择最高权重result[key] = max(weighted_values, key=lambda x: x[1])[0]return result

策略三：最佳结果选择（Best-of-N）

def best_of_n_selection(responses: List[str], evaluator) -> str:"""选择评分最高的结果"""scored_responses = [(response, evaluator.score(response))for response in responses]return max(scored_responses, key=lambda x: x[1])[0]

14.2.4 Anthropic 的 Parallelization 最佳实践

Anthropic 官方《Building Effective Agents》中关于并行化的建议：

1. 适用场景：

   • 需要多视角评估（如代码安全审查、内容审核）
   • 需要高置信度结果（如医疗诊断、法律分析）
   • 可以独立处理的子任务

2. 实现要点：

   • 使用 asyncio 或线程池实现真正的并行
   • 设置合理的超时时间
   • 实现优雅的错误处理（部分失败仍可继续）

14.3 条件分支与动态提示生成

14.3.1 核心概念

条件分支（Routing） 是根据输入的特征动态选择不同的处理路径。这种模式类似于编程中的 if-else 或 switch-case，让系统能够根据输入类型采用最合适的提示策略。

┌─────────────────────────────────────────────────────────────────┐│ Routing 模式 │├─────────────────────────────────────────────────────────────────┤│ ││输入│││ ││▼ ││ ┌─────────────────┐ ││ │分类器│ ││ │ (Classifier)│ ││ └────────┬────────┘ │││ ││┌─────────────────┼─────────────────┐ │││ │ │││▼ ▼ ▼││ ┌──────────┐┌──────────┐┌──────────┐││ │ 类型 A ││ 类型 B ││ 类型 C │││ │ 处理器 ││ 处理器 ││ 处理器 │││ └────┬─────┘└────┬─────┘└────┬─────┘│││ │ │ ││└─────────────────┼─────────────────┘ │││││▼││ 最终输出││ │└─────────────────────────────────────────────────────────────────┘

14.3.2 实战案例：智能客服路由

场景：根据用户问题类型，路由到不同的处理流程。

from enum import Enumfrom typing import Dict, Callableimport openaiclass QueryType(Enum):"""查询类型枚举"""GENERAL = "general" # 一般咨询BILLING = "billing" # 账单问题TECHNICAL = "technical" # 技术支持REFUND = "refund" # 退款申请SALES = "sales" # 销售咨询class IntelligentRouter:"""智能客服路由器"""def __init__(self):self.client = openai.OpenAI()self.handlers: Dict[QueryType, Callable] = {QueryType.GENERAL: self.handle_general,QueryType.BILLING: self.handle_billing,QueryType.TECHNICAL: self.handle_technical,QueryType.REFUND: self.handle_refund,QueryType.SALES: self.handle_sales,}def classify_query(self, query: str) -> QueryType:"""分类用户查询"""classification_prompt = f"""将以下用户查询分类为以下类别之一：- general: 一般咨询、问候、闲聊- billing: 账单问题、支付问题、发票- technical: 技术支持、故障排查、配置帮助- refund: 退款申请、退货、赔偿- sales: 销售咨询、产品信息、价格询问用户查询：{query}只回复类别名称（general/billing/technical/refund/sales），不要其他内容。"""response = self.client.chat.completions.create(model="gpt-3.5-turbo",# 分类任务可用轻量级模型messages=[{"role": "user", "content": classification_prompt}],temperature=0)category = response.choices[0].message.content.strip().lower()try:return QueryType(category)except ValueError:return QueryType.GENERALdef handle_general(self, query: str) -> str:"""处理一般咨询"""prompt = f"""你是一位友好的客服代表。请回答用户的一般咨询。用户：{query}回答："""response = self.client.chat.completions.create(model="gpt-3.5-turbo",messages=[{"role": "user", "content": prompt}])return response.choices[0].message.contentdef handle_billing(self, query: str) -> str:"""处理账单问题"""prompt = f"""你是一位专业的账单客服。请帮助用户解决账单相关问题。你可以：查询账单详情、解释费用明细、协助处理支付问题。用户：{query}回答："""response = self.client.chat.completions.create(model="gpt-4",# 账单问题可能需要更强的推理能力messages=[{"role": "user", "content": prompt}])return response.choices[0].message.contentdef handle_technical(self, query: str) -> str:"""处理技术支持"""prompt = f"""你是一位技术支持专家。请帮助用户解决技术问题。请提供清晰的步骤指导，如果问题复杂，可以询问更多信息。用户：{query}回答："""response = self.client.chat.completions.create(model="gpt-4",messages=[{"role": "user", "content": prompt}])return response.choices[0].message.contentdef handle_refund(self, query: str) -> str:"""处理退款申请"""prompt = f"""你是一位售后客服。请处理用户的退款申请。请收集必要信息：订单号、退款原因、购买时间等。用户：{query}回答："""response = self.client.chat.completions.create(model="gpt-4",messages=[{"role": "user", "content": prompt}])return response.choices[0].message.contentdef handle_sales(self, query: str) -> str:"""处理销售咨询"""prompt = f"""你是一位销售顾问。请回答用户的产品咨询。突出产品优势，提供个性化推荐。用户：{query}回答："""response = self.client.chat.completions.create(model="gpt-4",messages=[{"role": "user", "content": prompt}])return response.choices[0].message.contentdef route(self, query: str) -> Dict:"""主路由方法"""# 分类query_type = self.classify_query(query)# 路由到对应处理器handler = self.handlers.get(query_type, self.handle_general)response = handler(query)return {"query_type": query_type.value,"response": response}# 使用示例router = IntelligentRouter()result = router.route("我的信用卡被扣了两次款，怎么办？")print(f"类型：{result['query_type']}")print(f"回答：{result['response']}")

14.3.3 动态提示生成

根据输入特征动态生成提示词：

class DynamicPromptGenerator:"""动态提示生成器"""def __init__(self):self.templates = {"beginner": {"style": "简单易懂","examples": True,"jargon": False,"step_by_step": True},"intermediate": {"style": "专业但清晰","examples": True,"jargon": True,"step_by_step": False},"expert": {"style": "简洁专业","examples": False,"jargon": True,"step_by_step": False}}def detect_expertise_level(self, query: str) -> str:"""检测用户专业水平"""expert_keywords = ["implementation", "architecture", "optimization", "algorithm"]beginner_keywords = ["what is", "how to", "explain", "basic"]query_lower = query.lower()expert_score = sum(1 for kw in expert_keywords if kw in query_lower)beginner_score = sum(1 for kw in beginner_keywords if kw in query_lower)if expert_score > beginner_score:return "expert"elif beginner_score > expert_score:return "beginner"else:return "intermediate"def generate_prompt(self, query: str, context: str = "") -> str:"""生成动态提示"""level = self.detect_expertise_level(query)config = self.templates[level]prompt_parts = [f"请以{config['style']}的方式回答以下问题。",]if config['examples']:prompt_parts.append("请提供具体示例。")if not config['jargon']:prompt_parts.append("避免使用专业术语，或在使用时进行解释。")if config['step_by_step']:prompt_parts.append("请分步骤说明。")if context:prompt_parts.append(f"n背景信息：{context}")prompt_parts.append(f"n问题：{query}")prompt_parts.append("n回答：")return "n".join(prompt_parts)

14.4 Prompt 压缩技术：LLMLingua 与信息密度优化

14.4.1 为什么需要 Prompt 压缩

核心问题：

• 长上下文成本高（输入 token 按量计费）
• 长上下文可能降低模型注意力质量
• 存在"Lost in the Middle"现象

压缩目标：

• 保留关键信息
• 减少 token 数量
• 保持语义完整性

14.4.2 LLMLingua 技术

LLMLingua 是微软研究院提出的提示压缩技术，通过识别和移除冗余 token 来压缩提示。

核心原理：

1. 使用小型语言模型（如 GPT-2）计算每个 token 的困惑度（perplexity）
2. 困惑度低的 token 对模型理解贡献小，可以移除
3. 保留困惑度高的关键 token

# LLMLingua 使用示例from llmlingua import PromptCompressorllm_lingua = PromptCompressor(model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",use_llmlingua2=True,)# 原始长提示original_prompt = """[系统提示] 你是一个专业助手...[历史对话] 用户：你好。助手：你好！...（很长的对话历史）[文档] 这是一篇很长的参考文档...[当前问题] 总结以上内容。"""# 压缩compressed_prompt = llm_lingua.compress_prompt(original_prompt,rate=0.5,# 压缩到原长度的 50%force_tokens=["总结", "以上"],# 强制保留的 token)print(f"原始长度：{len(original_prompt)} tokens")print(f"压缩后：{len(compressed_prompt)} tokens")

14.4.3 信息密度优化策略

策略一：结构化压缩

def structured_compress(text: str, max_tokens: int = 1000) -> str:"""结构化压缩：保留标题、列表，压缩段落"""lines = text.split('n')compressed = []current_length = 0for line in lines:# 保留结构元素（标题、列表项）if line.startswith('#') or line.startswith('- ') or line.startswith('* '):compressed.append(line)current_length += len(line.split())else:# 压缩普通段落words = line.split()if current_length + len(words) > max_tokens:# 只取前几个词remaining = max_tokens - current_lengthcompressed.append(' '.join(words[:remaining]) + '...')breakelse:compressed.append(line)current_length += len(words)return 'n'.join(compressed)

策略二：语义压缩（基于嵌入）

def semantic_compress(sentences: list[str], target_count: int = 5) -> list[str]:"""语义压缩：选择最具代表性的句子"""import numpy as npfrom sklearn.cluster import KMeans# 获取句子嵌入embeddings = get_embeddings(sentences)# 聚类kmeans = KMeans(n_clusters=target_count, random_state=42)kmeans.fit(embeddings)# 从每个聚类中选择最接近中心的句子selected = []for i in range(target_count):cluster_indices = np.where(kmeans.labels_ == i)[0]center = kmeans.cluster_centers_[i]# 找到最接近中心的句子closest_idx = min(cluster_indices,key=lambda idx: np.linalg.norm(embeddings[idx] - center))selected.append(sentences[closest_idx])# 按原文顺序排序selected.sort(key=lambda s: sentences.index(s))return selected

策略三：关键信息提取

def extract_key_information(text: str) -> str:"""提取关键信息，移除冗余内容"""extraction_prompt = f"""从以下文本中提取关键信息，移除冗余内容。保留：- 核心观点和结论- 关键数据和事实- 重要定义和概念移除：- 重复信息- 修饰性语言- 示例和类比（除非对理解至关重要）原文：{text}提取的关键信息："""response = client.chat.completions.create(model="gpt-3.5-turbo",# 使用轻量级模型进行压缩messages=[{"role": "user", "content": extraction_prompt}],max_tokens=1000)return response.choices[0].message.content

14.4.4 Anthropic 的 Compaction 功能

Anthropic 提供了原生的服务端压缩功能（Beta）：

import anthropicclient = anthropic.Anthropic()messages = []def chat(user_message: str) -> str:messages.append({"role": "user", "content": user_message})response = client.beta.messages.create(betas=["compact-2026-01-12"],model="claude-opus-4-6",max_tokens=4096,messages=messages,context_management={"edits": [{"type": "compact_20260112","trigger": {"type": "input_tokens", "value": 100000},}]},)# 自动包含 compaction 块messages.append({"role": "assistant", "content": response.content})return next(block.text for block in response.content if block.type == "text")

14.5 自动提示词优化工具：DSPy、AutoPrompt 实践

14.5.1 DSPy 框架

DSPy（Declarative Self-improving Python）是斯坦福大学 NLP 团队开发的声明式提示优化框架。

核心理念：

• 将提示工程视为优化问题
• 自动搜索最优提示词和示例
• 支持多种优化器（BootstrapFewShot、MIPRO 等）

# DSPy 基础示例import dspy# 配置模型lm = dspy.LM('openai/gpt-4')dspy.configure(lm=lm)# 定义签名（输入输出规范）class GenerateAnswer(dspy.Signature):"""根据上下文回答问题"""context = dspy.InputField()question = dspy.InputField()answer = dspy.OutputField(desc="简洁准确的答案")# 定义模块class RAG(dspy.Module):def __init__(self, num_passages=3):super().__init__()self.retrieve = dspy.Retrieve(k=num_passages)self.generate_answer = dspy.ChainOfThought(GenerateAnswer)def forward(self, question):context = self.retrieve(question).passagesresponse = self.generate_answer(context=context, question=question)return dspy.Prediction(context=context, answer=response.answer)# 使用rag = RAG()result = rag("What is the capital of France?")print(result.answer)

DSPy 优化器：

from dspy.teleprompt import BootstrapFewShot, MIPROv2# 准备训练数据trainset = [dspy.Example(question="What is 2+2?", answer="4").with_inputs("question"),# ... 更多示例]# 使用 BootstrapFewShot 优化器optimizer = BootstrapFewShot(metric=accuracy_metric, max_bootstrapped_demos=4)optimized_rag = optimizer.compile(RAG(), trainset=trainset)# 或使用 MIPROv2（更强大的优化器）mipro_optimizer = MIPROv2(metric=accuracy_metric, num_candidates=10)optimized_rag = mipro_optimizer.compile(RAG(), trainset=trainset)

14.5.2 AutoPrompt

AutoPrompt 是一种自动发现提示词模板的方法，通过梯度优化找到最优的提示词。

核心思想：

• 将提示词视为可学习的参数
• 使用梯度下降优化提示词 token
• 自动发现有效的提示模式

# AutoPrompt 概念示例（简化版）class AutoPrompt:"""自动提示词发现"""def __init__(self, base_prompt: str, num_tokens: int = 10):self.base_prompt = base_promptself.trigger_tokens = ["[MASK]"] * num_tokensdef optimize(self, train_examples: list, num_iterations: int = 100):"""优化触发 token"""for iteration in range(num_iterations):# 构建当前提示current_prompt = self.base_prompt + " ".join(self.trigger_tokens)# 在训练集上评估scores = []for example in train_examples:score = self.evaluate(current_prompt, example)scores.append(score)avg_score = sum(scores) / len(scores)# 基于梯度（或启发式）更新 trigger tokensself.update_trigger_tokens(scores)if iteration % 10 == 0:print(f"Iteration {iteration}: Score = {avg_score:.4f}")return " ".join(self.trigger_tokens)def evaluate(self, prompt: str, example: dict) -> float:"""评估提示词效果"""# 实际实现会调用 LLM 并计算准确率passdef update_trigger_tokens(self, scores: list):"""更新触发 token（简化实现）"""# 实际实现会使用梯度或离散优化pass

14.5.3 提示词优化最佳实践

评估驱动的优化循环：

class PromptOptimizer:"""提示词优化器"""def __init__(self, evaluator):self.evaluator = evaluatorself.history = []def optimize(self, base_prompt: str, train_set: list,max_iterations: int = 10) -> str:"""优化提示词"""best_prompt = base_promptbest_score = self.evaluate(base_prompt, train_set)for iteration in range(max_iterations):# 生成变体variants = self.generate_variants(best_prompt)# 评估每个变体for variant in variants:score = self.evaluate(variant, train_set)self.history.append({"iteration": iteration,"prompt": variant,"score": score})if score > best_score:best_score = scorebest_prompt = variantprint(f"Iteration {iteration}: New best score = {score:.4f}")# 早停检查if not self.should_continue(self.history):breakreturn best_promptdef generate_variants(self, prompt: str) -> list[str]:"""生成提示词变体"""variants = []# 变体1：调整语气variants.append(self.adjust_tone(prompt, "professional"))variants.append(self.adjust_tone(prompt, "friendly"))# 变体2：添加/移除示例variants.append(self.add_examples(prompt, n=2))variants.append(self.remove_examples(prompt))# 变体3：调整指令明确性variants.append(self.make_more_specific(prompt))variants.append(self.make_more_concise(prompt))return variantsdef evaluate(self, prompt: str, train_set: list) -> float:"""评估提示词效果"""scores = []for example in train_set:prediction = self.run_prompt(prompt, example["input"])score = self.evaluator.score(prediction, example["expected"])scores.append(score)return sum(scores) / len(scores)

14.5.4 工具对比

本章小结

核心概念回顾

关键要点

1. Prompt Chaining：将复杂任务分解为简单子任务，提高质量和可控性
2. Parallelization：多视角并行评估，提升结果可靠性和置信度
3. Routing：根据输入特征动态选择最优处理路径
4. Prompt Compression：LLMLingua、语义压缩、Anthropic Compaction 等技术减少 token 消耗
5. Auto-optimization：DSPy 等框架将提示工程转化为可优化的机器学习问题

技术选型指南

参考来源：