Prompt工程

1. In-Context Learning 原理 In-Context Learning（ICL）是大语言模型的核心涌现能力之一：无需梯度更新，仅通过上下文中的示例就能学会新任务。 1.1 ICL 的工作机制 当你在 Prompt 中提供示例时，模型的注意力机制会将示例的模式映射到查询上： # Zero-shot 任务：将以下句子分类为正面/负面/中性 输入：这家餐厅的服务态度很好 输出： # Few-shot（1-shot） 任务：将以下句子分类为正面/负面/中性 输入：这道菜太难吃了 → 输出：负面 输入：这家餐厅的服务态度很好 → 输出： 模型通过示例"理解"了任务的输入输出映射规则，而非死记硬背。 1.2 ICL vs Fine-tuning 维度 ICL (Few-shot) Fine-tuning 参数更新 无 有 数据需求 1-20 个示例 数百到数万 部署成本 低（同一模型） 高（需部署多版本） 灵活性 高（随时改示例） 低（需重新训练） 性能上限 中等 高 延迟 较高（长 Prompt） 较低 2. 示例数量的影响 2.1 边际收益递减曲线 示例数量与效果的关系遵循对数增长曲线： 准确率 | ___________ ← 8-shot | / | / ← 4-shot | / |/ ← 1-shot |________ | ← 0-shot +--------- 示例数量 0 1 4 8 16 经验法则： ...

1. System Prompt 的结构 System Prompt 是 LLM 交互中优先级最高的指令层。一个工程化的 System Prompt 应包含以下模块： ┌──────────────────────────────────┐ │ 1. 角色定义 (Role) │ │ 2. 任务描述 (Task) │ │ 3. 行为约束 (Constraints) │ │ 4. 输出格式 (Output Format) │ │ 5. 知识边界 (Knowledge Boundary)│ │ 6. 示例 (Examples) │ │ 7. 异常处理 (Error Handling) │ └──────────────────────────────────┘ 1.1 完整模板 # 角色 你是一个高级数据分析师，擅长从结构化数据中提取洞察。 # 任务 用户会提供数据描述或 CSV 数据，你需要： 1. 理解数据结构和业务含义 2. 进行统计分析 3. 生成可视化建议 4. 输出分析报告 # 约束 - 只基于用户提供的数据分析，不编造数据 - 如果数据不足，明确说明需要什么数据 - 不输出投资建议或确定性预测 - 统计方法需标注假设条件 # 输出格式 ```json { "summary": "一句话摘要", "statistics": {"mean": 0, "median": 0, "std": 0}, "insights": ["洞察1", "洞察2"], "visualization": {"type": "bar", "reason": "原因"}, "limitations": ["局限1"] } 知识边界 熟悉：统计学、Python pandas、SQL 不熟悉：用户的具体业务上下文（需用户补充） 异常处理 数据为空 → 询问用户提供数据 数据格式错误 → 指出格式问题并给出正确格式示例 分析超出能力范围 → 明确说明并建议替代方案 ## 2. 角色定义的科学方法 ### 2.1 角色三要素 ```python ROLE_TEMPLATE = """ # 身份 你是{profession}，拥有{experience}年经验。 # 专长 {expertise_list} # 思维方式 {thinking_style} # 沟通风格 {communication_style} """ # 示例 role = ROLE_TEMPLATE.format( profession="资深安全工程师", experience="15", expertise_list="- Web 安全\n- 逆向工程\n- 安全架构设计", thinking_style="先评估威胁模型，再设计方案，最后验证有效性", communication_style="简洁直接，用数据和事实说话，避免模糊表述" ) 2.2 角色粒度对比 角色定义 效果 Token “你是助手” 泛化回答 ~5 “你是程序员” 技术导向回答 ~10 “你是10年经验的Python后端工程师，熟悉高并发架构” 精准技术回答 ~30 完整角色三要素 最精准，风格一致 ~100 关键发现：角色定义越具体，输出质量和一致性越高，但超过一定粒度后边际收益递减。 ...

1. 上下文窗口管理 1.1 上下文窗口的构成 ┌─────────────────────────────────────────┐ │ System Prompt ~500 tokens │ │ 对话摘要（注入） ~300 tokens │ │ 用户画像/偏好 ~200 tokens │ │ 历史对话（裁剪后） ~2000 tokens │ │ 当前用户输入 ~200 tokens │ │ ────────────────────────────────────── │ │ 总计 ~3200 tokens │ │ 模型上下文窗口 8K~128K tokens │ │ 预留给输出 ~1000 tokens │ └─────────────────────────────────────────┘ 1.2 Token 预算分配 class ContextWindowManager: def __init__(self, model_context_size=8192, output_reserve=1024): self.total = model_context_size self.output = output_reserve self.input_budget = self.total - self.output def allocate(self, components): """分配 Token 预算""" budget = self.input_budget allocation = {} # 优先级分配 priorities = [ ("system", 0.15), # 系统提示固定 15% ("summary", 0.10), # 摘要 10% ("user_profile", 0.05), # 用户画像 5% ("history", 0.50), # 历史对话 50% ("current", 0.20), # 当前输入 20% ] for name, ratio in priorities: allocation[name] = min( components.get(name, ""), key=lambda x: self._truncate(x, int(budget * ratio)) ) return allocation def _truncate(self, text, max_tokens): """截断到指定 token 数""" tokens = tokenizer.encode(text) if len(tokens) <= max_tokens: return text return tokenizer.decode(tokens[:max_tokens]) 1.3 不同模型窗口大小 模型 上下文窗口 建议历史轮数 GPT-3.5 16K 8-12 轮 GPT-4 8K/32K/128K 5-8/20-30/100+ Claude 3.5 200K 100+ 轮 Gemini 1.5 1M 极长对话 Llama 3 8K 4-6 轮 2. 对话摘要 2.1 何时触发摘要 class SummarizationTrigger: def __init__(self, threshold_ratio=0.7): self.threshold_ratio = threshold_ratio # 上下文使用率达到70%时触发 def should_summarize(self, messages, max_context): current_tokens = count_tokens(messages) return current_tokens > max_context * self.threshold_ratio def summarize_strategy(self, messages): """分层摘要策略""" total_tokens = count_tokens(messages) if total_tokens < 2000: return None # 不需要摘要 # 保留最近 2 轮，摘要其余 recent = messages[-4:] # 最近2轮（user+assistant各1） to_summarize = messages[:-4] return { "to_summarize": to_summarize, "keep_recent": recent, "target_summary_tokens": min(300, total_tokens // 4) } 2.2 摘要 Prompt 设计 SUMMARY_PROMPT = """请将以下对话浓缩为关键信息摘要。 要求： 1. 保留用户的核心需求和意图 2. 保留已确定的关键事实（人名、数值、决策等） 3. 保留未解决的问题 4. 用简洁的条目式表达 5. 省略寒暄和重复内容 对话内容： {conversation} 输出格式： - 用户意图：... - 已知事实：... - 待解决问题：... - 上下文要点：... """ def generate_summary(conversation, llm): prompt = SUMMARY_PROMPT.format( conversation=format_conversation(conversation) ) return llm.generate(prompt, temperature=0.0, max_tokens=300) 2.3 增量摘要 class IncrementalSummarizer: """增量更新摘要，避免每次全量重算""" def __init__(self, llm): self.llm = llm self.current_summary = "" self.summarized_until = 0 # 已摘要到的消息索引 def update(self, messages): """增量更新摘要""" # 只处理新增的消息 new_messages = messages[self.summarized_until:] if len(new_messages) < 4: # 少于2轮不更新 return self.current_summary # 保留最近2轮不摘要 to_summarize = new_messages[:-4] recent = new_messages[-4:] if not to_summarize: return self.current_summary # 增量摘要 if self.current_summary: prompt = f""" 现有摘要：{self.current_summary} 新增对话：{format_conversation(to_summarize)} 请将新增对话信息整合到现有摘要中。保持简洁。 """ else: prompt = SUMMARY_PROMPT.format( conversation=format_conversation(to_summarize) ) self.current_summary = self.llm.generate(prompt, max_tokens=300) self.summarized_until = len(messages) - len(recent) return self.current_summary 3. 状态追踪 3.1 对话状态管理 from dataclasses import dataclass, field from typing import Dict, List, Optional from enum import Enum class ConversationState(Enum): GREETING = "greeting" GATHERING_INFO = "gathering_info" PROCESSING = "processing" AWAITING_CONFIRMATION = "awaiting_confirmation" COMPLETED = "completed" ERROR_RECOVERY = "error_recovery" @dataclass class DialogueState: state: ConversationState = ConversationState.GREETING topic: str = "" collected_info: Dict[str, str] = field(default_factory=dict) pending_questions: List[str] = field(default_factory=list) user_preferences: Dict[str, str] = field(default_factory=dict) turn_count: int = 0 last_intent: str = "" def to_prompt_context(self): """将状态转为 Prompt 上下文""" return f""" [对话状态] 当前阶段: {self.state.value} 话题: {self.topic} 已收集信息: {json.dumps(self.collected_info, ensure_ascii=False)} 待确认问题: {self.pending_questions} 用户偏好: {json.dumps(self.user_preferences, ensure_ascii=False)} 对话轮次: {self.turn_count} """ 3.2 意图检测与状态转移 class StateTracker: def __init__(self, llm): self.llm = llm self.state = DialogueState() def update(self, user_input, assistant_output): """更新对话状态""" self.state.turn_count += 1 # 意图检测 intent = self.detect_intent(user_input) self.state.last_intent = intent # 状态转移 transitions = { (ConversationState.GREETING, "ask_question"): ConversationState.GATHERING_INFO, (ConversationState.GATHERING_INFO, "provide_info"): ConversationState.GATHERING_INFO, (ConversationState.GATHERING_INFO, "request_action"): ConversationState.PROCESSING, (ConversationState.PROCESSING, "confirm"): ConversationState.AWAITING_CONFIRMATION, (ConversationState.AWAITING_CONFIRMATION, "yes"): ConversationState.COMPLETED, (ConversationState.AWAITING_CONFIRMATION, "no"): ConversationState.GATHERING_INFO, (ConversationState.COMPLETED, "new_topic"): ConversationState.GREETING, } new_state = transitions.get((self.state.state, intent)) if new_state: self.state.state = new_state # 信息提取 self.extract_info(user_input, assistant_output) # 话题切换检测 if self.is_topic_change(user_input): self.state.collected_info.clear() self.state.topic = self.detect_topic(user_input) def detect_intent(self, text): prompt = f""" 判断用户意图，从以下选项中选择： ask_question, provide_info, request_action, confirm, new_topic, other 用户输入: {text} 意图: """ return self.llm.generate(prompt, temperature=0.0).strip() 4. 话题切换处理 4.1 话题切换检测 class TopicDetector: def __init__(self, embedder, threshold=0.5): self.embedder = embedder self.threshold = threshold self.topic_history = [] def detect(self, current_input): if not self.topic_history: self.topic_history.append(current_input) return False, None # 计算与最近话题的相似度 current_emb = self.embedder.encode(current_input) recent_emb = self.embedder.encode(self.topic_history[-1]) similarity = cosine_similarity(current_emb, recent_emb)[0][0] if similarity < self.threshold: # 话题切换 self.topic_history.append(current_input) return True, self.extract_topic(current_input) return False, None def extract_topic(self, text): """用 LLM 提取话题关键词""" prompt = f"用一个短语概括以下内容的主题：\n{text}" return llm.generate(prompt, max_tokens=20) 4.2 话题切换时的上下文处理 def handle_topic_switch(new_topic, conversation_manager): """话题切换时保存当前上下文，开始新上下文""" # 1. 保存当前话题的对话摘要 old_summary = conversation_manager.summarize_current() conversation_manager.saved_topics.append({ "topic": conversation_manager.state.topic, "summary": old_summary, "turn_count": conversation_manager.state.turn_count }) # 2. 重置状态 conversation_manager.state = DialogueState() conversation_manager.state.topic = new_topic # 3. 注入话题切换提示 context = f""" [话题切换] 用户从"{conversation_manager.saved_topics[-1]['topic']}"切换到"{new_topic}" 之前的话题已归档，如有需要可以引用。 """ return context 5. 记忆注入 5.1 长期记忆与短期记忆 class MemorySystem: def __init__(self): self.short_term = [] # 最近对话（原文） self.working_memory = {} # 当前任务状态 self.long_term = [] # 跨会话记忆 def inject(self, user_input): """构造记忆注入上下文""" context_parts = [] # 1. 长期记忆检索（基于相似度） relevant_memories = self.retrieve_long_term(user_input, top_k=3) if relevant_memories: context_parts.append( "[长期记忆]\n" + "\n".join(f"- {m}" for m in relevant_memories) ) # 2. 工作记忆 if self.working_memory: context_parts.append( "[工作记忆]\n" + json.dumps(self.working_memory, ensure_ascii=False, indent=2) ) # 3. 短期记忆（最近对话摘要） if self.short_term: context_parts.append( "[近期对话摘要]\n" + self.short_term[-1] ) return "\n\n".join(context_parts) def retrieve_long_term(self, query, top_k=3): """从长期记忆中检索相关内容""" if not self.long_term: return [] query_emb = embedder.encode(query) memory_embs = [embedder.encode(m["content"]) for m in self.long_term] scores = cosine_similarity(query_emb.reshape(1,-1), memory_embs)[0] top_indices = np.argsort(scores)[-top_k:][::-1] return [self.long_term[i]["content"] for i in top_indices if scores[i] > 0.3] # 相似度阈值 6. 长对话策略 6.1 分段处理策略 class LongConversationStrategy: """长对话的分段处理策略""" def __init__(self, max_turns_per_segment=10): self.max_turns = max_turns_per_segment self.segments = [] def process_turn(self, user_input, assistant_output, turn_count): # 检查是否需要分段 if turn_count > 0 and turn_count % self.max_turns == 0: # 生成段摘要 segment = self.get_current_segment() summary = self.summarize_segment(segment) self.segments.append({ "turn_range": f"{turn_count - self.max_turns + 1}-{turn_count}", "summary": summary }) self.reset_current_segment() self.add_to_current_segment(user_input, assistant_output) def get_context(self): """获取分段上下文""" if not self.segments: return "" context = "[历史分段摘要]\n" for seg in self.segments[-3:]: # 只保留最近3个分段 context += f"轮次{seg['turn_range']}: {seg['summary']}\n" return context 6.2 Redis 上下文存储 import redis import json from datetime import datetime, timedelta class RedisContextStore: """Redis 存储对话上下文""" def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.ttl = 86400 * 7 # 7天过期 def save_context(self, session_id, context): """保存对话上下文""" key = f"chat:context:{session_id}" self.redis.setex( key, self.ttl, json.dumps({ "context": context, "updated_at": datetime.now().isoformat() }, ensure_ascii=False) ) def load_context(self, session_id): """加载对话上下文""" key = f"chat:context:{session_id}" data = self.redis.get(key) if data: return json.loads(data)["context"] return None def save_summary(self, session_id, summary): """保存对话摘要""" key = f"chat:summary:{session_id}" self.redis.setex(key, self.ttl, summary) def append_message(self, session_id, role, content): """追加消息到 Redis List""" key = f"chat:messages:{session_id}" message = json.dumps({ "role": role, "content": content, "timestamp": datetime.now().isoformat() }, ensure_ascii=False) self.redis.rpush(key, message) self.redis.expire(key, self.ttl) # 控制列表长度 if self.redis.llen(key) > 100: # 最多保留100条 self.redis.ltrim(key, -80, -1) # 保留最近80条 6.3 完整多轮对话架构 class MultiTurnConversationEngine: def __init__(self, llm, redis_store, system_prompt): self.llm = llm self.redis = redis_store self.system_prompt = system_prompt self.summarizer = IncrementalSummarizer(llm) self.state_tracker = StateTracker(llm) self.memory = MemorySystem() def chat(self, session_id, user_input): # 1. 加载历史上下文 context = self.redis.load_context(session_id) or [] # 2. 更新摘要 summary = self.summarizer.update(context) # 3. 更新对话状态 self.state_tracker.update(user_input, "") # 4. 记忆注入 memory_context = self.memory.inject(user_input) # 5. 构造完整 Prompt messages = self.build_messages( system_prompt=self.system_prompt, summary=summary, state=self.state_tracker.state, memory=memory_context, history=context[-6:], # 最近3轮 user_input=user_input ) # 6. 生成回复 response = self.llm.chat(messages) # 7. 更新状态 self.state_tracker.update(user_input, response) # 8. 保存上下文 context.append({"role": "user", "content": user_input}) context.append({"role": "assistant", "content": response}) self.redis.save_context(session_id, context) return response def build_messages(self, **kwargs): messages = [{"role": "system", "content": kwargs["system_prompt"]}] if kwargs.get("summary"): messages.append({ "role": "system", "content": f"[对话摘要]\n{kwargs['summary']}" }) if kwargs.get("state"): messages.append({ "role": "system", "content": kwargs["state"].to_prompt_context() }) if kwargs.get("memory"): messages.append({ "role": "system", "content": kwargs["memory"] }) for msg in kwargs.get("history", []): messages.append(msg) messages.append({"role": "user", "content": kwargs["user_input"]}) return messages 7. 总结 多轮对话 Prompt 设计的核心挑战是在有限的上下文窗口内保持信息完整性： ...

1. 为什么需要结构化输出 LLM 默认输出自然语言，但生产系统需要结构化数据来做下游处理。JSON 是最常见的结构化格式。 1.1 常见问题 # 期望输出 {"name": "张三", "age": 25, "skills": ["Python", "SQL"]} # 实际可能出现的各种问题 1. 包含 Markdown 代码块标记：```json ... ``` 2. 字段名不一致：{"姓名": "张三", "年龄": "25"} 3. 类型错误：{"age": "25"} ← 字符串而非数字 4. 多余字段：{"name": "张三", "age": 25, "extra": "..."} 5. 缺失字段：{"name": "张三"} 6. 嵌套错误：{"skills": "Python, SQL"} ← 应为数组 7. 幻觉内容：{"name": "张三", "age": 25, "ssn": "..."} ← 泄露敏感信息 1.2 结构化输出方法对比 方法 可靠性 灵活性 实现复杂度 纯 Prompt 描述 ~70% 高 低 JSON Mode ~90% 中 低 Function Calling ~99% 中 中 Schema 约束 + 验证 ~99% 高 中高 Constrained Decoding ~100% 低 高 2. JSON Schema 约束 2.1 在 Prompt 中描述 Schema SCHEMA_PROMPT = """ 从用户输入中提取人员信息，输出必须严格符合以下 JSON Schema： { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "name": { "type": "string", "description": "全名", "minLength": 1, "maxLength": 50 }, "age": { "type": "integer", "minimum": 0, "maximum": 150 }, "skills": { "type": "array", "items": {"type": "string"}, "minItems": 0, "maxItems": 20 }, "department": { "type": "string", "enum": ["工程", "产品", "设计", "运营"] } }, "required": ["name", "age"], "additionalProperties": false } 规则： - 只输出 JSON，不要 Markdown 标记 - 不要输出注释 - 字段名用英文 - 严格遵循 Schema 类型约束 """ 2.2 嵌套结构处理 NESTED_SCHEMA = """ 提取项目信息，支持嵌套结构： { "project": { "name": "string", "status": "enum: planning, active, completed, archived", "team": { "lead": {"name": "string", "email": "string"}, "members": [ {"name": "string", "role": "string"} ] }, "milestones": [ { "name": "string", "due_date": "string (YYYY-MM-DD)", "status": "enum: pending, in_progress, done" } ] } } 示例输入："项目Alpha，进行中，负责人张三 zhangsan@example.com， 团队成员李四（开发）和王五（测试），里程碑：M1设计评审 2026-07-01 待开始， M2开发完成 2026-08-15 进行中" 示例输出： { "project": { "name": "项目Alpha", "status": "active", "team": { "lead": {"name": "张三", "email": "zhangsan@example.com"}, "members": [ {"name": "李四", "role": "开发"}, {"name": "王五", "role": "测试"} ] }, "milestones": [ {"name": "M1设计评审", "due_date": "2026-07-01", "status": "pending"}, {"name": "M2开发完成", "due_date": "2026-08-15", "status": "in_progress"} ] } } """ 3. Function Calling vs JSON Mode 3.1 OpenAI Function Calling from openai import OpenAI client = OpenAI() # 定义函数 schema tools = [ { "type": "function", "function": { "name": "extract_person", "description": "从文本中提取人员信息", "parameters": { "type": "object", "properties": { "name": {"type": "string"}, "age": {"type": "integer"}, "skills": { "type": "array", "items": {"type": "string"} } }, "required": ["name", "age"] } } } ] response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "张三，28岁，擅长Python和Go"}], tools=tools, tool_choice={"type": "function", "function": {"name": "extract_person"}} ) # 结果保证是合法 JSON 且符合 Schema import json result = json.loads(response.choices[0].message.tool_calls[0].function.arguments) print(result) # {"name": "张三", "age": 28, "skills": ["Python", "Go"]} 3.2 JSON Mode response = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": "从用户输入提取人员信息，输出JSON。"}, {"role": "user", "content": "张三，28岁，擅长Python和Go"} ], response_format={"type": "json_object"} # JSON Mode ) # 保证输出是合法 JSON，但不保证字段结构 3.3 对比 特性 Function Calling JSON Mode JSON 合法性 ✅ 保证 ✅ 保证 Schema 遵守 ✅ 保证 ❌ 不保证 字段类型 ✅ 强制 ❌ 依赖 Prompt 多函数选择 ✅ 支持 ❌ 不适用 多模态支持 ✅ ❌ Token 开销 较高 较低 4. Pydantic 验证 4.1 定义模型 from pydantic import BaseModel, Field, field_validator, model_validator from typing import List, Optional from datetime import date from enum import Enum class Department(str, Enum): ENGINEERING = "工程" PRODUCT = "产品" DESIGN = "设计" OPERATIONS = "运营" class TeamMember(BaseModel): name: str = Field(min_length=1, max_length=50) role: str = Field(min_length=1, max_length=30) email: Optional[str] = Field(default=None, pattern=r'^[\w.-]+@[\w.-]+\.\w+$') class Project(BaseModel): name: str = Field(min_length=1, max_length=100) status: str = Field(pattern=r'^(planning|active|completed|archived)$') team_lead: TeamMember members: List[TeamMember] = Field(default_factory=list, max_length=50) start_date: Optional[date] = None budget: Optional[float] = Field(default=None, ge=0) @field_validator('status') @classmethod def validate_status(cls, v): valid = {'planning', 'active', 'completed', 'archived'} if v not in valid: raise ValueError(f'status must be one of {valid}') return v @model_validator(mode='after') def validate_team_size(self): if len(self.members) > 0 and self.team_lead.name in [m.name for m in self.members]: raise ValueError('team_lead should not be in members') return self 4.2 验证循环 import json from pydantic import ValidationError class StructuredOutputEngine: def __init__(self, llm_client, model_class, max_retries=3): self.llm = llm_client self.model = model_class self.max_retries = max_retries def generate(self, user_input: str): prompt = self._build_prompt(user_input) for attempt in range(self.max_retries): # 1. LLM 生成 raw = self.llm.generate(prompt, response_format={"type": "json_object"}) # 2. JSON 解析 try: data = json.loads(raw) except json.JSONDecodeError as e: prompt = self._fix_prompt(raw, f"JSON解析错误: {e}", user_input) continue # 3. Schema 验证 try: validated = self.model(**data) return validated except ValidationError as e: errors = self._format_errors(e) prompt = self._fix_prompt(raw, errors, user_input) continue raise RuntimeError(f"Failed after {self.max_retries} attempts. Last output: {raw}") def _build_prompt(self, user_input): schema = self.model.model_json_schema() return f""" 从用户输入中提取信息并输出 JSON。 Schema: {json.dumps(schema, indent=2, ensure_ascii=False)} 用户输入：{user_input} 只输出 JSON，不要任何额外文本。 """ def _fix_prompt(self, bad_output, errors, user_input): return f""" 之前的输出有错误，请修正。 用户输入：{user_input} 上次输出：{bad_output} 错误信息：{errors} 请修正后重新输出 JSON。只输出 JSON，不要其他内容。 """ def _format_errors(self, e: ValidationError): lines = [] for err in e.errors(): loc = ".".join(str(x) for x in err['loc']) lines.append(f"- {loc}: {err['msg']}") return "\n".join(lines) 4.3 使用示例 # 使用 engine = StructuredOutputEngine(llm_client, Project) try: project = engine.generate( "项目Alpha，状态active，负责人张三 zhang@corp.com，" "成员李四(开发)和王五(测试)，预算50万" ) print(project.model_dump_json(indent=2)) except RuntimeError as e: print(f"提取失败: {e}") 5. 错误修复循环 5.1 常见错误与修复策略 ERROR_FIX_STRATEGIES = { "json_decode_error": { "cause": "输出不是合法 JSON", "fix": "提示模型去掉 Markdown 标记，只输出纯 JSON" }, "missing_field": { "cause": "必填字段缺失", "fix": "明确指出缺失的字段名，要求补充" }, "type_error": { "cause": "字段类型不匹配（如 string vs int）", "fix": "指出字段名和期望类型，给出正确示例" }, "enum_violation": { "cause": "枚举值不在允许范围内", "fix": "列出所有允许值" }, "extra_field": { "cause": "包含 Schema 中不存在的字段", "fix": "提示移除未定义字段" } } 5.2 带重试的完整流程 def extract_with_retry(llm, user_input, schema_model, max_retries=3): """带自动修复的结构化提取""" messages = [ {"role": "system", "content": f"提取信息为JSON。Schema: {schema_model.model_json_schema()}"}, {"role": "user", "content": user_input} ] for i in range(max_retries): response = llm.chat.completions.create( model="gpt-4", messages=messages, response_format={"type": "json_object"} ) output = response.choices[0].message.content try: data = json.loads(output) return schema_model(**data) except (json.JSONDecodeError, ValidationError) as e: # 将错误反馈给模型 messages.append({"role": "assistant", "content": output}) messages.append({ "role": "user", "content": f"输出有误：{e}\n请修正后重新输出。" }) raise RuntimeError(f"Failed after {max_retries} retries") 6. 数组处理技巧 6.1 固定长度数组 # Schema 中限制数组长度 { "type": "array", "minItems": 3, "maxItems": 3, "items": {"type": "string"} } 6.2 动态数组与分页 # 当结果可能很多时，使用分页结构 PAGINATED_SCHEMA = { "type": "object", "properties": { "items": {"type": "array", "items": {"type": "object"}}, "total_estimated": {"type": "integer"}, "has_more": {"type": "boolean"}, "next_page_hint": {"type": "string"} } } # Prompt 中引导分页 EXTRACTION_PROMPT = """ 从文档中提取所有人员信息。 如果人员超过20个，先输出前20个，设置 has_more=true， 并在 next_page_hint 中说明如何获取下一批。 """ 6.3 异构数组 # 数组中包含不同类型的对象 HETEROGENEOUS_SCHEMA = { "type": "array", "items": { "oneOf": [ {"type": "object", "properties": {"type": {"const": "text"}, "content": {"type": "string"}}}, {"type": "object", "properties": {"type": {"const": "image"}, "url": {"type": "string"}}}, {"type": "object", "properties": {"type": {"const": "code"}, "language": {"type": "string"}, "content": {"type": "string"}}} ] } } 7. 性能与可靠性数据 7.1 各方法可靠性对比 方法 成功率 平均重试次数 Token 消耗 纯 Prompt 72% 1.8 350 JSON Mode 89% 1.2 320 Function Calling 97% 1.0 420 Schema + Pydantic + 重试 99.5% 1.1 480 Constrained Decoding 100% 1.0 350 7.2 错误分布 JSON解析失败 ████████████ 12% 字段缺失 ██████████ 25% 类型错误 ████████ 20% 枚举值错误 ████ 10% 多余字段 ████ 8% 嵌套结构错误 █████ 15% 数组长度错误 ██ 5% 其他 █ 5% 8. 总结 结构化输出是 LLM 从"聊天玩具"到"生产工具"的关键一步。核心要点： ...

ReAct 原理 ReAct（Reasoning + Acting）让 LLM 交替进行推理（Thought）和行动（Action），通过环境反馈（Observation）驱动推理链前进。 问题 → Thought → Action → Observation → Thought → Action → ... → Answer ↑ ↑ 内部推理 环境反馈 为什么 ReAct 有效 纯推理(CoT) 纯行动 ReAct 基于错误事实推理 盲目尝试 推理指导行动，行动修正推理 无法获取外部信息 无法利用中间结论 动态获取并整合信息 错误沿链传播 每步独立 观察反馈纠正方向 Thought-Action-Observation 循环 Thought 1: 分析问题，决定第一步 Action 1: 调用工具 Observation 1: 工具返回结果 Thought 2: 基于观察决定下一步 Action 2: 调用工具 Observation 2: 工具返回结果 Thought N: 已有足够信息 Action: FINISH Answer: 最终答案 示例 问题: 2024年奥斯卡最佳影片导演出生城市人口？ Thought 1: 先找最佳影片和导演 Action 1: search("2024 Oscar best picture") Observation 1: 《奥本海默》获最佳影片，导演诺兰。 Thought 2: 诺兰出生城市 Action 2: search("Nolan birthplace") Observation 2: 出生于伦敦。 Thought 3: 伦敦人口 Action 3: search("London population") Observation 3: 约898万。 Thought 4: 信息齐全 Action 4: FINISH Answer: 诺兰出生于伦敦，人口约898万。 ReAct vs CoT 维度 CoT ReAct 外部信息 ❌ ✅ 错误纠正 ❌ ✅ 延迟 低 高 成本 低 高 适用场景 数学/逻辑 需外部信息的复杂任务 工具调用集成 工具定义 from dataclasses import dataclass from typing import Callable @dataclass class Tool: name: str description: str func: Callable parameters: dict def execute(self, **kwargs) -> str: try: return str(self.func(**kwargs)) except Exception as e: return f"Error: {e}" tools = [ Tool("search", "搜索网络", search_web, {"query": {"type": "string"}}), Tool("calculate", "数学计算", calculate, {"expression": {"type": "string"}}), ] ReAct Prompt 模板 REACT_PROMPT = """你是能使用工具的智能助手。 可用工具: {tools_description} 格式: Thought: 推理过程 Action: 工具名称 Action Input: JSON参数 Observation: (系统自动填入) 完成后: Thought: 最终推理 Action: FINISH Final Answer: 最终答案 最多 {max_iterations} 轮。 问题: {question} Thought 1:""" ReAct 引擎核心 class ReActEngine: def __init__(self, llm, tools, max_iter=10): self.llm = llm self.tools = {t.name: t for t in tools} self.max_iter = max_iter def run(self, question: str) -> dict: history = "" for i in range(self.max_iter): response = self.llm.generate(self._build_prompt(question, history), temperature=0) parsed = self._parse(response) if parsed.get('action') == 'FINISH': return {'answer': parsed.get('final_answer', ''), 'iterations': i+1} name = parsed.get('action', '') obs = self.tools[name].execute(**parsed.get('input', {})) if name in self.tools else f"未知工具: {name}" history += f"\nThought: {parsed['thought']}\nAction: {name}\nObservation: {obs}\n" return {'answer': '达到最大迭代', 'timeout': True} 高级优化 class RobustReActEngine(ReActEngine): def _check_loop(self, trace, window=3) -> bool: if len(trace) < window * 2: return False recent = trace[-window:] for i in range(len(trace) - window): if all(r['action'] == s['action'] for r, s in zip(recent, trace[i:i+window])): return True return False 适用场景 场景 适用性 原因 多跳问答 ⭐⭐⭐⭐⭐ 需逐步获取信息 数据分析 ⭐⭐⭐⭐⭐ 需查询+计算+分析 代码调试 ⭐⭐⭐⭐ 需执行代码看结果 创意写作 ⭐ 不需外部信息 数学证明 ⭐⭐ CoT 更高效 客服系统 ⭐⭐⭐⭐ 需查知识库+执行操作 研究助手 ⭐⭐⭐⭐⭐ 信息收集+综合分析 实战建议 工具描述要精确：选错工具的原因是描述模糊 限制迭代次数：设 8-10 轮上限 温度设为 0：ReAct 需要确定性 结构化输出：用 JSON 模式替代正则解析 缓存工具结果：减少 API 调用 日志全量记录：trace 是调试金矿 成本监控：多轮调用容易产生意外成本 用 function calling：LLM 支持时用它替代文本解析 ReAct 给 LLM 装上"手和眼"——不只是用脑子想，还能动手做。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

核心原理 Self-Consistency (SC) 的核心思想极其简洁：对同一个问题生成多条推理路径，通过投票选出最一致的答案。 传统 Chain-of-Thought (CoT) 只采样一次推理路径，若该路径在某一步出错，最终答案就会错。SC 利用了一个关键观察——正确答案往往比错误答案更容易被多种推理路径到达。 # 单次 CoT：一条路径 问题 → [推理路径 A] → 答案 A # Self-Consistency：多条路径 + 投票 问题 → [推理路径 A] → 答案 A ─┐ → [推理路径 B] → 答案 B ─┼→ 多数投票 → 最终答案 → [推理路径 C] → 答案 A ─┘ → [推理路径 D] → 答案 A ─┘ 采样策略详解 温度参数 (Temperature) 温度控制采样的随机性，是 SC 效果的关键变量： 温度范围 效果 适用场景 0.0-0.3 路径高度相似，多样性不足 简单问题 0.5-0.7 多样性与质量的最佳平衡 推荐默认 0.8-1.0 路径差异大，但质量下降 复杂推理需高多样性 Top-p 采样 Top-p (nucleus sampling) 限制了候选 token 的概率质量： ...

链式调用原理 单次 LLM 调用的能力是有限的。链式调用 (Prompt Chaining) 将复杂任务分解为多个步骤，每步的输出作为下一步的输入，形成流水线： [用户输入] → [Step 1: 理解意图] → [Step 2: 检索知识] → [Step 3: 生成答案] → [最终输出] 核心假设：分解后的子任务更简单、更可靠，错误隔离更容易。 链式架构模式 1. 顺序链 (Sequential Chain) 最简单的模式——线性传递： from typing import Dict, Any import openai def llm_call(prompt: str, model: str = "gpt-4") -> str: resp = openai.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content class SequentialChain: def __init__(self): self.steps = [] def add_step(self, name: str, prompt_template: str): """prompt_template 中用 {prev_output} 引用上一步输出""" self.steps.append({"name": name, "template": prompt_template}) return self def run(self, initial_input: str) -> Dict[str, str]: results = {} current_output = initial_input for step in self.steps: prompt = step["template"].format(prev_output=current_output) current_output = llm_call(prompt) results[step["name"]] = current_output print(f"[{step['name']}] 完成") results["final"] = current_output return results # 示例：技术文档翻译流水线 chain = SequentialChain() chain.add_step("extract_keypoints", "分析以下技术文档，提取关键技术概念和术语（JSON数组格式）：\n{prev_output}") chain.add_step("translate", "将以下内容翻译为中文，保持技术术语准确：\n{prev_output}") chain.add_step("review", "审校以下翻译，修正不准确之处，输出最终版本：\n{prev_output}") 2. 并行链 (Parallel Chain) 多个独立子任务并行执行后汇总： ...

Meta-Prompting 原理 Meta-Prompting 的核心思想：用 LLM 来优化 LLM 的 prompt。这是一种"元"层面的优化——不在 prompt 内部做文章，而是让模型自己去寻找更好的 prompt。 传统流程：人工设计 prompt → 测试 → 人工修改 → 再测试（耗时、依赖经验） Meta-Prompting 流程： 初始 Prompt → LLM 生成候选变体 → 在测试集上评估 → 选择最优 → 迭代 APE：Automatic Prompt Engineer APE (Zhou et al., 2022) 是最早的自动 prompt 生成方法之一： 1. 给 LLM 少量输入-输出示例 2. 让 LLM 推断可能的 instruction (prompt) 3. 在验证集上评估每个候选 prompt 4. 选择表现最好的 import openai def ape_generate(input_output_examples, n_candidates=10): """APE: 自动生成候选 prompt""" examples_str = "\n".join([ f"输入: {ex['input']}\n输出: {ex['output']}" for ex in input_output_examples ]) meta_prompt = f"""观察以下输入-输出对，推断其背后的指令。 {examples_str} 请生成 {n_candidates} 条可能的指令，使模型能根据该指令从输入得到输出。 每条指令独占一行，以数字编号。指令应简洁、准确、可泛化。""" resp = openai.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": meta_prompt}], temperature=0.9, ) return resp.choices[0].message.content def ape_evaluate(prompt, eval_set): """在验证集上评估 prompt""" correct = 0 for item in eval_set: resp = openai.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": f"{prompt}\n\n输入: {item['input']}"}], temperature=0.0, ) if resp.choices[0].message.content.strip() == item['output'].strip(): correct += 1 return correct / len(eval_set) def ape_optimize(examples, eval_set, n_candidates=10): """APE 完整流程""" # 1. 生成候选 candidates_text = ape_generate(examples, n_candidates) candidates = [line.split('. ', 1)[1] for line in candidates_text.strip().split('\n') if '. ' in line] # 2. 评估 scores = [(c, ape_evaluate(c, eval_set)) for c in candidates] scores.sort(key=lambda x: x[1], reverse=True) return scores[0] # (best_prompt, score) OPRO：优化 by PROgression OPRO (Yang et al., 2023) 将 prompt 优化建模为优化问题，用 LLM 作为优化器： ...

Few-shot 的本质：上下文学习 大语言模型具备一种独特能力——上下文学习（In-Context Learning, ICL）。模型不需要更新参数，仅通过 Prompt 中提供的少量示例就能学会新任务。Few-shot Prompt 就是利用这一能力的技术。 核心公式： 模型输出 = f(Prompt 中的示例分布 × 输入查询) 模型并非"理解"了任务规则，而是从示例中提取了统计模式（输入-输出的映射关系、格式规范、语言风格），并在推理时应用这些模式。这意味着：示例的选择、排列、格式直接决定了模型的表现。 示例选择策略 随机选择 从示例池中随机抽取 K 个样本。最简单但效果最不稳定。 import random examples = load_examples("data/training_set.jsonl") selected = random.sample(examples, k=4) 问题： 随机选取可能选到不代表任务分布的示例，导致模型行为偏移。 相似性选择 基于语义相似度，选取与当前输入最接近的示例。这是最常用的策略： from sentence_transformers import SentenceTransformer import numpy as np class SimilarExampleSelector: def __init__(self, examples, model_name="all-MiniLM-L6-v2"): self.encoder = SentenceTransformer(model_name) self.examples = examples self.embeddings = self.encoder.encode( [ex["input"] for ex in examples], normalize_embeddings=True ) def select(self, query, k=4): query_emb = self.encoder.encode([query], normalize_embeddings=True) # 余弦相似度 scores = self.embeddings @ query_emb.T top_k_idx = np.argsort(scores.flatten())[-k:][::-1] return [self.examples[i] for i in top_k_idx] 效果： 在分类任务上，相似性选择通常比随机选择提升 3-8% 准确率。 注意： 相似性度量很关键。文本嵌入的相似性不等于任务语义相似性。在情感分析中，与查询情感极性相同的示例可能比主题相同的示例更有用。 多样性选择 确保所选示例覆盖不同类别/模式，避免同质化： def diverse_select(query, examples, k=4, n_clusters=4): """先聚类，再从每个簇中选最相似的""" from sklearn.cluster import KMeans embeddings = encode_all(examples) kmeans = KMeans(n_clusters=n_clusters, random_state=42).fit(embeddings) selected = [] for cluster_id in range(n_clusters): cluster_indices = np.where(kmeans.labels_ == cluster_id)[0] cluster_examples = [examples[i] for i in cluster_indices] # 从该簇中选与 query 最相似的 best = select_most_similar(query, cluster_examples) selected.append(best) return selected[:k] 选择策略对比 策略 实现难度 计算成本 效果稳定性 最佳场景 随机 极低 O(1) 低 基线对比、快速原型 相似性 中 O(N) 检索 中-高 分类、抽取、QA 多样性 高 O(N) + 聚类 高 多类别任务、生成任务 混合 高 中-高 最高 生产环境 示例顺序的影响 一个被广泛验证的现象：示例顺序对模型输出有显著影响。 ...

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 管理 ...