输出控制

一、输出控制的重要性 大语言模型的自由生成特性是一把双刃剑——带来创造力但也带来不可预测性。在工程化应用中，输出控制是决定 AI 能否可靠交付的关键能力。 1.1 失控输出的代价 场景 失控表现 影响 客服自动化 格式混乱、缺少关键信息 客户投诉 数据批处理 输出结构不一致，无法解析 下游管道崩溃 代码生成 格式不对、缺少注释 CI/CD 失败 内容生成 多版本风格不统一 品牌一致性受损 二、格式约束技术 2.1 Markdown 格式控制 def markdown_format_control(): prompt = """请严格按照以下 Markdown 格式输出： # 标题 ## 摘要 <200 字以内的摘要> ## 核心要点 - 要点 1：<具体内容> - 要点 2：<具体内容> - 要点 3：<具体内容> ## 数据表格 | 指标 | 值 | 说明 | |------|----|------| | <名称> | <数值> | <说明> | ## 结论 <150 字以内> --- 注意：每个部分都必须输出，缺失则不合格。""" return prompt # 格式验证函数 import re def validate_markdown_structure(text: str) -> dict: """验证 Markdown 结构完整性""" checks = { "has_title": bool(re.search(r"^# ", text, re.MULTILINE)), "has_summary": bool(re.search(r"^## 摘要", text, re.MULTILINE)), "has_keypoints": bool(re.search(r"^## 核心要点", text, re.MULTILINE)), "has_table": bool(re.search(r"\|.*\|.*\|", text)), "has_conclusion": bool(re.search(r"^## 结论", text, re.MULTILINE)), "summary_length_ok": None, } # 检查摘要长度 summary_match = re.search( r"## 摘要\s*\n(.+?)(?=\n##)", text, re.DOTALL ) if summary_match: checks["summary_length_ok"] = len(summary_match.group(1)) < 200 return checks 2.2 JSON 输出控制 通用的 JSON 约束 Prompt 输出必须是一个合法的 JSON 对象，格式如下： { "status": "success | error", "data": { // 按需填充 }, "metadata": { "timestamp": "<当前时间戳>", "confidence": 0.0-1.0 } } 要求： 1. 所有 key 必须使用双引号 2. 字符串值必须使用双引号 3. 不能包含注释或 Markdown 代码块标记 4. 不能有其他文字描述 5. 一定输出纯 JSON JSON 修复技术 import json import re def extract_and_fix_json(text: str) -> dict: """从模型输出中提取并修复 JSON""" # Step 1: 尝试提取 JSON 块 # 移除代码块标记 text = re.sub(r'```json\s*|\s*```', '', text) try: return json.loads(text) except json.JSONDecodeError: pass # Step 2: 常见修复 fixes = [ # 修复单引号 lambda s: s.replace("'", '"'), # 修复末尾逗号 lambda s: re.sub(r',(\s*[}\]])', r'\1', s), # 修复注释 lambda s: re.sub(r'//.*?\n', '\n', s), # 修复布尔值 lambda s: s.replace('True', 'true').replace('False', 'false'), ] for fix in fixes: try: return json.loads(fix(text)) except json.JSONDecodeError: continue # Step 3: 最后手段 - 重新生成 raise ValueError("无法修复 JSON 输出") 2.3 XML 输出控制 <response> <metadata> <model>gpt-4</model> <timestamp>2026-06-25T12:00:00Z</timestamp> <type>analysis</type> </metadata> <content> <section id="overview"> <title>概述</title> <paragraph>{overview_text}</paragraph> </section> <section id="details"> <title>详细信息</title> <item key="point_1">{detail_1}</item> <item key="point_2">{detail_2}</item> </section> </content> <validation> <passed>true</passed> <score>{quality_score}</score> </validation> </response> 三、结构化输出技术 3.1 Pydantic + Function Calling from pydantic import BaseModel, Field, validator from typing import List, Optional from openai import OpenAI class ProductReview(BaseModel): """结构化的产品评价输出""" product_name: str = Field(description="产品名称") rating: int = Field(ge=1, le=5, description="评分 1-5") pros: List[str] = Field(min_length=1, max_length=5, description="优点列表") cons: List[str] = Field(min_length=1, max_length=5, description="缺点列表") summary: str = Field(max_length=200, description="总结摘要") recommend: bool = Field(description="是否推荐") @validator('rating') def rating_matches_sentiment(cls, v, values): if 'cons' in values and len(values['cons']) > 3 and v > 4: raise ValueError('评分与缺点数量不符') return v def structured_review_analysis(review_text: str) -> ProductReview: client = OpenAI() response = client.chat.completions.create( model="gpt-4", messages=[{ "role": "user", "content": f"分析以下产品评价并提取结构化信息：\n{review_text}" }], functions=[{ "name": "extract_review", "description": "提取产品评价的结构化信息", "parameters": ProductReview.schema() }], function_call={"name": "extract_review"} ) return ProductReview.parse_raw( response.choices[0].message.function_call.arguments ) 3.2 枚举约束 from enum import Enum class Sentiment(Enum): POSITIVE = "positive" NEGATIVE = "negative" NEUTRAL = "neutral" MIXED = "mixed" class Category(Enum): COMPLAINT = "complaint" INQUIRY = "inquiry" FEEDBACK = "feedback" SUGGESTION = "suggestion" OTHER = "other" def constrained_classification(text: str) -> tuple: """强制输出限定在枚举范围内""" prompt = f"""从以下选项中选择情感倾向（仅输出一个词）： {', '.join([s.value for s in Sentiment])} 同时从以下选项选择类别（仅输出一个词）： {', '.join([c.value for c in Category])} 文本：{text} 情感： 类别：""" # ... 调用 LLM 并严格验证输出 # 如果输出不在枚举范围内，重试或降级处理 pass 四、长度与粒度控制 4.1 精确长度控制 控制需求 Prompt 写法 效果 字/词精确 “用 100-150 字回答” 近似控制，±20% 段落数 “写 3 段，每段 2-3 句” 较好控制 列表项 “列出恰好 5 个要点” 较好控制 代码行数 “不超过 20 行代码” 中等控制 4.2 渐进式扩展技术 def progressive_expansion(topic: str, max_length: int) -> str: """从核心内容开始，逐步扩展文字到目标长度""" # Phase 1: 生成核心骨架 skeleton_prompt = f"为 '{topic}' 写一个三行的核心概要。" # ... 获取 skeleton # Phase 2: 扩展到段落 expand_prompt = f"""基于以下概要，将内容扩展到约 {max_length//2} 字： {skeleton} 要求：添加具体例子和数据佐证。""" # ... 获取 expanded # Phase 3: 精细化调整 final_prompt = f"""将以下内容调整到恰好 {max_length} 字： {expanded} 如果超出则精简，不足则补充细节。""" # ... 返回最终结果 pass 五、输出验证与后处理 5.1 验证管道 class OutputValidator: """多层次输出验证""" def __init__(self, rules: dict): self.rules = rules def validate(self, output: str) -> dict: results = { "passed": True, "checks": [], "errors": [] } # Layer 1: 结构完整性 if self.rules.get("require_structure"): for section in self.rules["required_sections"]: if section not in output: results["passed"] = False results["errors"].append(f"缺少必要段落: {section}") # Layer 2: 格式正确性 if self.rules.get("require_json"): try: json.loads(output) except json.JSONDecodeError as e: results["passed"] = False results["errors"].append(f"JSON 格式错误: {e}") # Layer 3: 内容合理性 if self.rules.get("min_length"): if len(output) < self.rules["min_length"]: results["passed"] = False results["errors"].append("内容长度不足") # Layer 4: 安全过滤 for pattern in self.rules.get("block_patterns", []): if re.search(pattern, output, re.IGNORECASE): results["passed"] = False results["errors"].append(f"检测到违禁内容: {pattern}") results["checks"] = len(results["errors"]) return results # 使用 validator = OutputValidator({ "require_structure": True, "required_sections": ["摘要", "正文", "结论"], "require_json": False, "min_length": 100, "block_patterns": [r"(密码|passwd|secret)"] }) 5.2 后处理管道 class OutputPostProcessor: """输出后处理管道""" def __init__(self): self.pipeline = [] def add_step(self, name: str, fn): self.pipeline.append((name, fn)) def process(self, output: str) -> str: for name, fn in self.pipeline: try: output = fn(output) except Exception as e: print(f"后处理步骤 '{name}' 失败: {e}") return output # 示例管道 processor = OutputPostProcessor() processor.add_step("trim_whitespace", lambda x: x.strip()) processor.add_step("fix_encoding", lambda x: x.encode('utf-8', errors='replace').decode('utf-8')) processor.add_step("remove_markdown_code_block", lambda x: re.sub(r'```\w*\n?|```', '', x)) processor.add_step("ensure_trailing_newline", lambda x: x if x.endswith('\n') else x + '\n') 六、对比总结 技术 可靠性 灵活性 实现复杂度 适用场景 Markdown 格式约束 ⭐⭐⭐ ⭐⭐⭐⭐ ⭐ 文档、报告 JSON 输出 ⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐ API 集成、数据管道 Function Calling ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐⭐ 生产级应用 XML 输出 ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐ 结构化文档 枚举约束 ⭐⭐⭐⭐⭐ ⭐ ⭐ 分类任务 长度控制 ⭐⭐ ⭐⭐⭐ ⭐ 内容生成 七、总结 精确的输出控制是将大模型从"玩具"变为"工具"的关键一步： ...