模型有时候会在JSON里夹杂解释文字有时候JSON格式不对有时候字段名大小写不一致……这些问题困扰着每一个在生产环境使用LLM的工程师。这篇文章给你系统性的解法。结构化输出为什么难表面上看让LLM输出JSON很简单——在Prompt里加一句以JSON格式输出就好了。但在生产环境运行一段时间后你就会发现各种意想不到的问题问题1格式不一致有时候输出的JSON外面包着Markdown代码块json … 有时候直接输出解析逻辑需要处理两种情况。问题2字段缺失或多余用户问了一个简单问题模型可能懒得输出所有字段或者模型发挥创意加了一些你没要求的字段。问题3类型错误你要求的是数字模型给了字符串你要求的是数组模型给了对象。问题4Markdown干扰模型输出了{summary: 这是**重要**的信息}值里面夹了Markdown格式。解决这些问题需要多层防御策略。## 第一层原生JSON模式最可靠现代主流模型都支持原生的JSON输出模式这是最可靠的方案pythonfrom openai import AsyncOpenAIfrom anthropic import AsyncAnthropicimport json# OpenAI使用 response_format 参数async def openai_json_output(prompt: str, schema: dict) - dict: client AsyncOpenAI() response await client.chat.completions.create( modelgpt-4o, messages[ { role: system, content: You are a data extraction assistant. Always respond with valid JSON. }, {role: user, content: prompt} ], response_format{ type: json_schema, json_schema: { name: extraction_result, schema: schema, strict: True # 严格模式模型输出必须匹配schema } } ) return json.loads(response.choices[0].message.content)# Anthropic Claude在系统提示中强制JSONasync def claude_json_output(prompt: str) - dict: client AsyncAnthropic() response await client.messages.create( modelclaude-opus-4-5, max_tokens2048, systemAlways respond with valid JSON only. No explanations, no markdown., messages[{role: user, content: prompt}] ) # Claude有时会在JSON前后添加说明文字需要提取 content response.content[0].text return extract_json(content)def extract_json(text: str) - dict: 从文本中提取JSON处理Markdown代码块等情况 import re # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 提取json …中的内容 json_match re.search(r(?:json)?\s*({.?})\s, text, re.DOTALL) if json_match: try: return json.loads(json_match.group(1)) except json.JSONDecodeError: pass # 提取第一个完整的 { ... } 块 brace_match re.search(r\{.*\}, text, re.DOTALL) if brace_match: try: return json.loads(brace_match.group()) except json.JSONDecodeError: pass raise ValueError(f无法从文本中提取JSON: {text[:200]}...)## 第二层Pydantic验证类型安全保障用Pydantic对输出进行结构化验证将类型错误在边界处捕获pythonfrom pydantic import BaseModel, Field, validatorfrom typing import List, Optionalfrom datetime import datetimeimport jsonclass ProductInfo(BaseModel): 产品信息提取结果 product_name: str Field(..., min_length1, max_length200) price: float Field(..., gt0) currency: str Field(defaultCNY, patternr^[A-Z]{3}$) features: List[str] Field(default_factorylist, max_items20) availability: bool True category: Optional[str] None validator(price, preTrue) def parse_price(cls, v): 处理价格字段的各种输入格式 if isinstance(v, str): # 移除货币符号和逗号 cleaned v.replace(¥, ).replace(,, ).replace($, ).strip() return float(cleaned) return v validator(features, preTrue) def ensure_list(cls, v): 确保features是列表 if isinstance(v, str): return [v] # 单个字符串转为列表 return vclass LLMStructuredExtractor: 带验证的结构化信息提取器 def __init__(self, llm_client): self.llm llm_client EXTRACTION_PROMPT 从以下文本中提取产品信息以JSON格式输出 必填字段 - product_name: 字符串产品名称 - price: 数字价格不含货币符号 - currency: 字符串货币代码如 CNY, USD - features: 数组产品特性列表 - availability: 布尔值是否有货 可选字段 - category: 字符串产品类别 文本 {text} async def extract_product_info(self, text: str) - ProductInfo: 提取并验证产品信息 max_retries 3 last_error None for attempt in range(max_retries): try: raw_output await self.llm.complete( self.EXTRACTION_PROMPT.format(texttext) ) json_data extract_json(raw_output) # Pydantic验证会自动做类型转换和验证 return ProductInfo(**json_data) except (ValueError, json.JSONDecodeError) as e: last_error e if attempt max_retries - 1: # 在重试提示中包含错误信息让模型自我修正 self.EXTRACTION_PROMPT f\n\n上次输出有错误{e}请修正后重新输出。 raise RuntimeError(f提取失败最后错误{last_error})## 第三层Instructor库最优雅的方案instructor库是目前最优雅的结构化输出解决方案它无缝整合了Pydantic和各大模型APIpythonimport instructorfrom openai import OpenAIfrom pydantic import BaseModelfrom typing import List# 一行代码让OpenAI客户端支持结构化输出client instructor.patch(OpenAI())class TechArticleMetadata(BaseModel): title: str summary: str Field(..., max_length200) topics: List[str] Field(..., min_items1, max_items5) difficulty_level: str Field(..., patternr^(beginner|intermediate|advanced)$) estimated_read_time_minutes: int Field(..., gt0, lt60)def extract_article_metadata(article_text: str) - TechArticleMetadata: 自动重试直到输出符合Pydantic schema # instructor会自动处理 # 1. 将Pydantic model转为JSON Schema # 2. 注入到API调用中 # 3. 验证输出验证失败自动重试 return client.chat.completions.create( modelgpt-4o, response_modelTechArticleMetadata, # 关键指定response_model max_retries3, # 自动重试次数 messages[ { role: user, content: f提取以下文章的元数据\n\n{article_text} } ] )# 使用示例metadata extract_article_metadata(这篇文章介绍了RAG技术...)print(metadata.title) # 自动类型转换print(metadata.topics) # 保证是列表print(metadata.difficulty_level) # 保证是指定枚举值之一## 第四层修复机制模型自我纠错当验证失败时让模型自我修正而不是简单重试pythonclass SelfCorrectingExtractor: 带自我修正能力的结构化提取器 CORRECTION_PROMPT 你之前的输出无法通过验证 原始输出 {original_output} 验证错误 {validation_errors} 请根据错误信息修正你的输出确保 1. 输出是合法的JSON格式 2. 所有必填字段都存在 3. 字段类型正确 只输出修正后的JSON不要包含任何解释。 async def extract_with_correction( self, prompt: str, response_model: type[BaseModel], max_attempts: int 3 ) - BaseModel: conversation_history [ {role: user, content: prompt} ] for attempt in range(max_attempts): response await self.llm.complete(conversation_history) try: json_data extract_json(response) return response_model(**json_data) except Exception as e: if attempt max_attempts - 1: raise # 构建自我修正提示 correction_prompt self.CORRECTION_PROMPT.format( original_outputresponse, validation_errorsstr(e) ) # 将错误反馈加入对话历史 conversation_history.extend([ {role: assistant, content: response}, {role: user, content: correction_prompt} ])## 复杂场景嵌套结构和动态Schemapythonfrom pydantic import BaseModel, create_modelfrom typing import Any, Dict# 场景根据不同的业务类型动态生成Schemadef create_dynamic_schema(entity_type: str, fields: Dict[str, Any]) - type[BaseModel]: 动态创建Pydantic模型 field_definitions {} for field_name, field_config in fields.items(): python_type { string: str, number: float, integer: int, boolean: bool, array: list }.get(field_config[type], str) default field_config.get(default, ...) field_definitions[field_name] (python_type, default) return create_model(f{entity_type}Schema, **field_definitions)# 使用示例OrderSchema create_dynamic_schema(Order, { order_id: {type: string}, amount: {type: number}, items_count: {type: integer}, is_paid: {type: boolean, default: False}})# 嵌套结构处理class Address(BaseModel): street: str city: str country: str CNclass UserProfile(BaseModel): user_id: str name: str email: str address: Optional[Address] None # 可选的嵌套对象 tags: List[str] []## 性能优化批量结构化提取对于大量文本的批量处理并发是关键pythonimport asynciofrom typing import TypeVar, GenericT TypeVar(T, boundBaseModel)async def batch_extract( texts: List[str], prompt_template: str, response_model: type[T], concurrency: int 5) - List[T | Exception]: 并发批量提取带错误收集 semaphore asyncio.Semaphore(concurrency) async def extract_one(text: str) - T | Exception: async with semaphore: try: return await extract_with_schema( prompt_template.format(texttext), response_model ) except Exception as e: return e # 不中断批量处理收集错误 tasks [extract_one(text) for text in texts] results await asyncio.gather(*tasks) # 统计成功率 successes [r for r in results if not isinstance(r, Exception)] errors [r for r in results if isinstance(r, Exception)] print(f成功: {len(successes)}/{len(texts)}, 失败: {len(errors)}) return results## 监控与告警pythonimport structlogfrom dataclasses import dataclasslogger structlog.get_logger()dataclassclass ParseMetrics: total_attempts: int 0 success_count: int 0 retry_count: int 0 final_failure_count: int 0class MonitoredExtractor: 带监控的结构化提取器 def __init__(self): self.metrics ParseMetrics() async def extract(self, *args, **kwargs) - BaseModel: self.metrics.total_attempts 1 try: result await self._do_extract(*args, **kwargs) self.metrics.success_count 1 logger.info( structured_extraction_success, success_rateself.metrics.success_count / self.metrics.total_attempts ) return result except Exception as e: self.metrics.final_failure_count 1 logger.error( structured_extraction_failed, errorstr(e), failure_rateself.metrics.final_failure_count / self.metrics.total_attempts ) raise property def health_report(self) - dict: total self.metrics.total_attempts if total 0: return {status: no_data} return { success_rate: f{self.metrics.success_count/total:.1%}, retry_rate: f{self.metrics.retry_count/total:.1%}, failure_rate: f{self.metrics.final_failure_count/total:.1%}, total_requests: total }## 方案选型总结| 场景 | 推荐方案 ||------|---------|| 简单字段提取 | OpenAI JSON Schema模式 || 复杂嵌套结构 | instructor Pydantic || 需要自我修正 | SelfCorrectingExtractor || 动态Schema | create_model 自定义解析 || 批量处理 | 并发批量提取 || 多模型兼容 | LiteLLM instructor |结构化输出工程是LLM应用基础设施的核心部分。投入精力做好这一层能显著提升整个应用的稳定性和可维护性。