引言
每一个在生产环境中用过LLM的工程师都遇到过这个问题：模型返回了"差不多"符合格式的JSON，但多了一对引号，少了一个逗号，或者字段名大小写不对……然后你的json.loads()抛出异常，整个流程崩掉。传统做法是写一堆后处理代码来"修复"LLM的输出。但这既不优雅，也不可靠。Structured Outputs（结构化输出）技术的出现，从根本上解决了这个问题——通过约束模型的解码过程，保证输出符合指定格式。—## 一、结构化输出的技术原理### 1.1 为什么普通提示词不够可靠？传统方式：python# 这种方式有一定概率失败response = openai.chat.completions.create( model="gpt-4o", messages=[{ "role": "user", "content": "请提取以下文本中的人名和联系方式，以JSON格式返回：\n\n张三，电话：13800138000" }])# 可能返回：# {"name": "张三", "phone": "13800138000"} # 正确# {'name': '张三', 'phone': '13800138000'} # 单引号，解析失败# 我提取到的信息如下：\njson\n{…} # 包裹在代码块中# 张三，电话13800138000 # 完全没有JSON### 1.2 约束解码（Constrained Decoding）结构化输出的底层原理是**约束解码**：在模型生成每个token时，根据当前JSON解析状态，屏蔽掉所有不符合语法的候选token。当前状态：{“name”: “允许的下一个token：任意字符（因为在字符串中）当前状态：{“name”: “张三"允许的下一个token：, 或 }（因为字符串已结束，需要分隔符或关闭括号）这样，无论模型的"本意"是什么，输出的token序列**在物理上不可能**违反JSON语法。---## 二、OpenAI Structured Outputs实战### 2.1 基础用法：response_formatpythonfrom openai import OpenAIfrom pydantic import BaseModel, Fieldfrom typing import List, Optionalclient = OpenAI()class ContactInfo(BaseModel): name: str = Field(description=“人名”) phone: Optional[str] = Field(default=None, description=“电话号码”) email: Optional[str] = Field(default=None, description=“邮箱地址”) company: Optional[str] = Field(default=None, description=“所在公司”)class ExtractedContacts(BaseModel): contacts: List[ContactInfo] extraction_confidence: float = Field( description=“提取置信度，0-1之间” )def extract_contacts(text: str) -> ExtractedContacts: “”“从文本中提取联系人信息””” completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[ { “role”: “system”, “content”: “从用户提供的文本中提取所有联系人信息。” }, { “role”: “user”, “content”: text } ], response_format=ExtractedContacts, ) # 直接获取类型化结果，无需JSON解析 return completion.choices[0].message.parsed# 使用示例result = extract_contacts(“”“参会人员：- 张三，技术总监，13800138000，zhang@company.com- 李四，产品经理，010-12345678- 王五，来自 Acme Corp”“”)print(f"提取了 {len(result.contacts)} 个联系人")for contact in result.contacts: print(f" {contact.name}: {contact.phone or ‘无电话’} / {contact.email or ‘无邮箱’}“)### 2.2 复杂嵌套结构pythonfrom enum import Enumfrom typing import Unionfrom datetime import dateclass Priority(str, Enum): LOW = “low” MEDIUM = “medium” HIGH = “high” CRITICAL = “critical"class TaskStatus(str, Enum): PENDING = “pending” IN_PROGRESS = “in_progress” COMPLETED = “completed” BLOCKED = “blocked"class SubTask(BaseModel): title: str assignee: Optional[str] estimated_hours: Optional[float] class Task(BaseModel): title: str description: str priority: Priority status: TaskStatus due_date: Optional[str] = Field(description=“格式：YYYY-MM-DD”) tags: List[str] = Field(default_factory=list) sub_tasks: List[SubTask] = Field(default_factory=list) blockers: List[str] = Field( default_factory=list, description=“阻塞原因列表” )class ProjectPlan(BaseModel): project_name: str total_tasks: int tasks: List[Task] summary: strdef parse_project_requirements(requirements: str) -> ProjectPlan: “”“将需求文档解析为结构化项目计划””” completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[ { “role”: “system”, “content”: “”“你是项目管理专家。将用户提供的需求描述转化为结构化的项目任务列表。合理拆分任务，评估优先级和工作量。””" }, {“role”: “user”, “content”: requirements} ], response_format=ProjectPlan, ) return completion.choices[0].message.parsed# 测试plan = parse_project_requirements(“”“我们需要在下个月上线一个用户管理系统，包括：1. 用户注册和登录（高优先级）2. 密码重置功能（高优先级，依赖邮件服务） 3. 用户档案编辑（中优先级）4. 管理员后台（低优先级，当前被其他项目阻塞）”“”)print(f"项目：{plan.project_name}“)for task in plan.tasks: print(f”\n[{task.priority.upper()}] {task.title}“) print(f” 状态：{task.status}“) if task.blockers: print(f” 阻塞：{'; ‘.join(task.blockers)}“)---## 三、Anthropic的结构化输出方案### 3.1 工具调用方式Anthropic通过工具调用实现结构化输出：pythonimport anthropicimport jsonfrom pydantic import BaseModelclient = anthropic.Anthropic()class SentimentAnalysis(BaseModel): sentiment: str # positive, negative, neutral score: float # -1.0 to 1.0 key_emotions: list[str] reasoning: strdef analyze_sentiment(text: str) -> SentimentAnalysis: “”“使用Claude分析文本情感””" # 将Pydantic模型转为JSON Schema schema = SentimentAnalysis.model_json_schema() response = client.messages.create( model=“claude-3-5-sonnet-20241022”, max_tokens=1000, tools=[{ “name”: “output_analysis”, “description”: “输出情感分析结果”, “input_schema”: schema }], tool_choice={“type”: “tool”, “name”: “output_analysis”}, messages=[{ “role”: “user”, “content”: f"请分析以下文本的情感：\n\n{text}" }] ) # 提取工具调用参数（即结构化输出） for block in response.content: if block.type == “tool_use” and block.name == “output_analysis”: return SentimentAnalysis(**block.input) raise ValueError(“未获得结构化输出”)result = analyze_sentiment(“这个产品真的超出了我的预期！界面美观，功能强大，客服响应也很快。”)print(f"情感：{result.sentiment}（分数：{result.score:.2f}）")print(f"关键情绪：{’, '.join(result.key_emotions)}“)### 3.2 使用instructor库简化`instructor`库提供了统一的结构化输出接口：pythonimport instructorimport anthropicfrom openai import OpenAIfrom pydantic import BaseModel# OpenAI + instructoroai_client = instructor.from_openai(OpenAI())# Anthropic + instructorant_client = instructor.from_anthropic(anthropic.Anthropic())class NewsArticle(BaseModel): title: str summary: str key_points: list[str] category: str estimated_read_time_minutes: intdef extract_article_info(text: str, provider: str = “openai”) -> NewsArticle: “”“提取文章信息（支持多provider）””" kwargs = { “response_model”: NewsArticle, “messages”: [{ “role”: “user”, “content”: f"请提取以下文章的关键信息：\n\n{text}" }] } if provider == “openai”: return oai_client.chat.completions.create( model=“gpt-4o-mini”, **kwargs ) elif provider == “anthropic”: return ant_client.messages.create( model=“claude-3-5-haiku-20241022”, max_tokens=1000, **kwargs )---## 四、高级应用：动态Schema生成### 4.1 根据业务需求动态生成Schemapythonfrom pydantic import create_modelfrom typing import get_type_hintsdef create_extraction_schema( fields: dict[str, tuple[type, str]]) -> type[BaseModel]: “”" 动态创建提取Schema Args: fields: {字段名: (类型, 描述)} Returns: 动态创建的Pydantic模型类 “”" field_definitions = {} for field_name, (field_type, description) in fields.items(): field_definitions[field_name] = ( Optional[field_type], Field(default=None, description=description) ) return create_model(“DynamicExtraction”, **field_definitions)# 业务配置驱动的提取extraction_config = { “product_name”: (str, “产品名称”), “price”: (float, “价格，单位：元”), “brand”: (str, “品牌名称”), “rating”: (float, “用户评分，1-5分”), “availability”: (bool, “是否有货”),}DynamicSchema = create_extraction_schema(extraction_config)completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[{ “role”: “user”, “content”: “iPhone 16 Pro，苹果品牌，售价9999元，综合评分4.8，现货” }], response_format=DynamicSchema,)result = completion.choices[0].message.parsedprint(result.model_dump())---## 五、错误处理与边界情况### 5.1 处理解析失败pythonfrom openai import LengthFinishReasonErrordef safe_structured_parse( text: str, schema: type[BaseModel], fallback: Optional[BaseModel] = None) -> Optional[BaseModel]: “”“带完整错误处理的结构化解析”“” try: completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[{“role”: “user”, “content”: text}], response_format=schema, max_tokens=2048 ) message = completion.choices[0].message # 检查是否因token限制被截断 if completion.choices[0].finish_reason == “length”: raise LengthFinishReasonError( “输出被截断，结构化数据可能不完整” ) # 检查是否有内容过滤 if message.refusal: print(f"内容被拒绝: {message.refusal}“) return fallback return message.parsed except LengthFinishReasonError: # 尝试增加max_tokens重试 print(“首次输出被截断，增加token重试…”) completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[{“role”: “user”, “content”: text}], response_format=schema, max_tokens=4096 # 翻倍 ) return completion.choices[0].message.parsed except Exception as e: print(f"结构化解析失败: {e}”) return fallback---## 六、性能与成本优化### 6.1 Schema缓存pythonimport hashlibfrom functools import lru_cache@lru_cache(maxsize=100)def get_cached_schema_json(schema_class: type) -> str: “”“缓存Schema的JSON序列化（避免重复计算）”“” return json.dumps(schema_class.model_json_schema())# 批量处理时复用Schemadef batch_extract( texts: list[str], schema: type[BaseModel], batch_size: int = 5) -> list[BaseModel]: “”“批量结构化提取”“” results = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] # 并发处理 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor: futures = [ executor.submit(safe_structured_parse, text, schema) for text in batch ] batch_results = [f.result() for f in futures] results.extend(batch_results) return [r for r in results if r is not None]---## 七、实际应用场景### 7.1 简历解析系统pythonclass WorkExperience(BaseModel): company: str title: str start_date: str end_date: Optional[str] = None # None表示在职 responsibilities: List[str]class Resume(BaseModel): name: str email: Optional[str] phone: Optional[str] years_of_experience: float skills: List[str] education: List[str] work_experience: List[WorkExperience] summary: strdef parse_resume(resume_text: str) -> Resume: completion = client.beta.chat.completions.parse( model=“gpt-4o-2024-08-06”, messages=[ {“role”: “system”, “content”: “请解析以下简历内容”}, {“role”: “user”, “content”: resume_text} ], response_format=Resume, ) return completion.choices[0].message.parsed### 7.2 合同信息提取pythonclass ContractClause(BaseModel): clause_type: str content: str is_standard: bool risk_level: str # low, medium, highclass ContractAnalysis(BaseModel): parties: List[str] effective_date: str expiry_date: Optional[str] total_value: Optional[float] currency: str = “CNY” key_clauses: List[ContractClause] risk_summary: str recommendation: str # approve, review, reject```—## 八、总结Structured Outputs技术已经成为生产级LLM应用的必备工具。核心要点：1.约束解码保证输出100%符合语法，不再需要繁琐的后处理2.Pydantic模型是定义Schema的最佳方式，类型安全+文档合一3.instructor库提供跨provider的统一接口，降低迁移成本4.动态Schema支持业务配置驱动的灵活提取需求5.错误处理不可忽视：token截断、内容拒绝需要有明确的fallback策略在信息提取、数据解析、表单处理等场景中，Structured Outputs技术可以将"成功率从95%提升到99.9%"，是值得优先投入工程化的能力。
