LLM的输出天然是自由形式的文本但AI应用需要的往往是结构化数据。这中间的gap就是结构化输出工程要解决的问题。2026年这个问题有了更成熟的解法。本文系统梳理从基础到进阶的结构化输出技术栈以及常见坑和对应的解决策略。## 为什么结构化输出难直觉上很简单告诉模型用JSON格式输出就行了。实际上没那么简单模型不总是遵守。在复杂推理场景下模型倾向于先用自然语言思考然后顺手输出了文字而不是JSON。格式正确但语义错误。JSON格式对了但字段值是模型编的——比如要求输出置信度0-1之间的数字模型输出了0.99999看起来没问题其实是没有根据的。边界情况导致解析失败。字符串里包含未转义的特殊字符数组元素数量不对嵌套层级缺失这些都会让你的json.loads()抛异常。流式输出的挑战。流式响应下无法在输出完成前验证JSON完整性。## 方法一原生结构化输出首选OpenAI、Anthropic等主流模型现在都支持原生的结构化输出通过在API层面约束输出格式可靠性比提示词方法高得多。pythonfrom openai import AsyncOpenAIfrom pydantic import BaseModel, Fieldfrom typing import Literalclient AsyncOpenAI()# 定义输出结构class ArticleAnalysis(BaseModel): title: str Field(description文章标题) category: Literal[技术, 产品, 市场, 其他] Field(description文章分类) key_points: list[str] Field( description文章核心观点3-5条, min_length3, max_length5 ) sentiment: Literal[positive, neutral, negative] difficulty_level: int Field(ge1, le5, description阅读难度1-5) tags: list[str] Field(max_length5, description相关标签最多5个)async def analyze_article(article_text: str) - ArticleAnalysis: response await client.beta.chat.completions.parse( modelgpt-4o, messages[ { role: system, content: 你是一个内容分析专家请对文章进行结构化分析。 }, { role: user, content: f请分析以下文章\n\n{article_text} } ], response_formatArticleAnalysis, ) return response.choices[0].message.parsedresponse.choices[0].message.parsed直接返回一个Pydantic对象不需要手动解析JSONPydantic的验证在模型输出阶段就已经完成。### 处理refused情况模型有时会因为内容安全原因拒绝输出结构化结果pythonasync def safe_analyze(article_text: str) - ArticleAnalysis | None: response await client.beta.chat.completions.parse( model“gpt-4o”, messages[…], response_formatArticleAnalysis, ) message response.choices[0].message if message.refusal: logging.warning(fModel refused: {message.refusal}“) return None return message.parsed## 方法二Instructor库跨模型兼容如果你需要同时支持多个模型提供商Instructor是目前最好的抽象层pythonimport instructorfrom anthropic import AsyncAnthropicfrom openai import AsyncOpenAIfrom pydantic import BaseModel# 支持OpenAIopenai_client instructor.from_openai(AsyncOpenAI())# 支持Anthropicanthropic_client instructor.from_anthropic(AsyncAnthropic())# 支持本地模型通过Ollamaollama_client instructor.from_openai( AsyncOpenAI(base_url“http://localhost:11434/v1”, api_key“ollama”), modeinstructor.Mode.JSON # 本地模型用JSON mode)class UserProfile(BaseModel): name: str email: str role: Literal[“admin”, “user”, “viewer”] permissions: list[str]async def extract_user_profile(text: str, use_model: str “openai”) - UserProfile: client openai_client if use_model “openai” else anthropic_client return await client.chat.completions.create( model“gpt-4o” if use_model “openai” else “claude-3-5-sonnet-20241022”, response_modelUserProfile, messages[ {“role”: “user”, “content”: f从以下文本中提取用户信息\n{text}”} ], max_retries3 # Instructor自动处理格式错误重试 )Instructor的自动重试机制是个杀手锏——当模型输出不符合Pydantic Schema时它会自动把错误信息反馈给模型让模型修正最多重试N次。### 复杂嵌套结构pythonfrom pydantic import BaseModel, model_validatorclass Address(BaseModel): street: str city: str country: str “中国” postal_code: str | None Noneclass ContactInfo(BaseModel): phone: str | None None email: str | None None address: Address | None None model_validator(mode‘after’) def at_least_one_contact(self): if not self.phone and not self.email: raise ValueError(“至少需要提供手机号或邮箱中的一种”) return selfclass CustomerRecord(BaseModel): customer_id: str name: str contact: ContactInfo tags: list[str] [] created_at: str # ISO format model_validator(mode‘after’) def validate_customer_id(self): if not self.customer_id.startswith(“CUS-”): self.customer_id fCUS-{self.customer_id} return selfmodel_validator让你可以在Pydantic层面做跨字段的业务验证模型如果输出不符合的内容Instructor会自动触发重试。## 方法三输出解析器后处理方式有时候你无法控制模型的API参数比如用第三方代理只能对原始文本做后处理pythonimport jsonimport refrom typing import TypeVar, TypeT TypeVar(T, boundBaseModel)class RobustOutputParser: 鲁棒的输出解析器处理各种格式问题 def parse(self, text: str, model_class: Type[T]) - T | None: # 尝试方法1直接解析 try: return model_class.model_validate_json(text) except Exception: pass # 尝试方法2提取JSON块 json_block self._extract_json_block(text) if json_block: try: return model_class.model_validate_json(json_block) except Exception: pass # 尝试方法3修复常见的JSON错误 fixed_json self._fix_json(text) if fixed_json: try: return model_class.model_validate_json(fixed_json) except Exception: pass return None def _extract_json_block(self, text: str) - str | None: 提取markdown代码块中的JSON patterns [ rjson\s*(.?)\s, #json …r\s*([[{].?[]}])\s, #{ … }r([\[{].*[\]}]), # 直接找JSON结构贪婪 ] for pattern in patterns: match re.search(pattern, text, re.DOTALL) if match: return match.group(1) return None def _fix_json(self, text: str) - str | None: 修复常见JSON格式错误 # 提取JSON部分 json_start text.find({) json_end text.rfind(}) 1 if json_start -1 or json_end 0: json_start text.find([) json_end text.rfind(]) 1 if json_start -1: return None json_text text[json_start:json_end] # 常见修复尾部逗号 json_text re.sub(r,\s*}, }, json_text) json_text re.sub(r,\s*], ], json_text) # 单引号改双引号Python习惯 # 注意这个替换不完美只做简单处理 # json_text json_text.replace(, ) return json_text## 流式结构化输出流式场景下你需要在输出完整之前就开始处理数据pythonimport instructorfrom openai import AsyncOpenAIclient instructor.from_openai(AsyncOpenAI())async def stream_analysis(text: str): 流式获取结构化输出适合前端实时展示 async with client.chat.completions.stream( modelgpt-4o, response_modelArticleAnalysis, messages[ {role: user, content: f分析文章\n{text}} ] ) as stream: # 实时获取部分完成的对象 async for partial in stream.partial: # partial是一个部分填充的ArticleAnalysis对象 # 字段值可能是None还没生成到 if partial.title: yield {field: title, value: partial.title} if partial.key_points: yield {field: key_points, value: partial.key_points} # 获取最终完整对象 final await stream.get_final_completion() yield {field: complete, value: final.model_dump()}## 结构设计的最佳实践### 让Schema对LLM友好python# 不友好的Schemaclass BadSchema(BaseModel): d: str # 缩写字段名模型不知道是什么 v: int f: list # 太模糊# 友好的Schemaclass GoodSchema(BaseModel): description: str Field(description对象的详细描述100字以内) confidence_score: int Field( ge0, le100, description置信度分数0-100整数100表示完全确定 ) features: list[str] Field( description对象的特征列表每条特征用一句话描述, max_length10 )字段名要语义清晰description要明确说明格式要求这些信息会被注入到模型的上下文中直接影响输出质量。### 用枚举约束取值范围pythonfrom enum import Enumclass Priority(str, Enum): LOW low MEDIUM medium HIGH high CRITICAL criticalclass TaskItem(BaseModel): title: str priority: Priority # 模型只能输出这四个值之一 status: Literal[todo, in_progress, done, blocked]### 分阶段输出复杂结构对于非常复杂的输出可以分两步先输出骨架再填充细节pythonclass DocumentOutline(BaseModel): title: str sections: list[str] # 先只要章节标题class DocumentSection(BaseModel): title: str content: str # 再逐节生成内容 word_count: intasync def generate_long_document(topic: str) - list[DocumentSection]: # 第一步生成大纲 outline await client.chat.completions.create( modelgpt-4o, response_modelDocumentOutline, messages[{role: user, content: f为「{topic}」生成一个5节的文档大纲}] ) # 第二步逐节生成内容 sections [] for section_title in outline.sections: section await client.chat.completions.create( modelgpt-4o, response_modelDocumentSection, messages[{ role: user, content: f为文章「{outline.title}」写第「{section_title}」节约300字 }] ) sections.append(section) return sections## 生产环境监控结构化输出在生产环境需要监控两个关键指标解析成功率你的结构化输出有多少次真正返回了有效的对象。低于95%就需要排查。字段填充完整率哪些字段经常是None或空值这可能说明Schema设计有问题或者模型对某类信息的理解不够。pythonclass StructuredOutputMonitor: def record(self, schema_name: str, result, success: bool): metrics.increment(fstructured_output.{schema_name}.total) if success: metrics.increment(fstructured_output.{schema_name}.success) # 记录字段填充情况 if result: for field, value in result.model_dump().items(): if value is not None and value ! [] and value ! : metrics.increment( fstructured_output.{schema_name}.field.{field}.filled ) else: metrics.increment(fstructured_output.{schema_name}.failed)—本文关键词结构化输出、Pydantic、Instructor、JSON Schema、LLM工程化
