大模型输出格式总不对这套结构化解析方案一劳永逸前言老王大模型又输出格式错误了JSON 缺逗号字段类型不对 后端工程师小李烦躁地说。本文看了看日志发现这是第100次格式错误了。你这是缺少系统化的结构化输出方案啊那该怎么办每次都要人工处理吗看来得分享本文们团队的经验了。今天聊聊如何用 Agent 拓扑设计模式解决结构化输出问题。一、底层原理1.1 大模型结构化输出的难点大模型本质是文本生成不是 JSON 生成器graph TD A[大模型生成] -- B[文本输出] B -- C{期望结构化} C --|JSON| D[可能格式错误] C --|XML| E[可能标签不匹配] D -- F[解析失败] F -- G[重试] G -- H[业务卡顿] I[解决方案] -- J[约束解码] I -- K[后处理验证] I -- L[分步生成]核心问题大模型没有类型系统输出是概率采样不稳定复杂嵌套结构容易出错频繁出错影响业务1.2 结构化方案对比方案稳定性灵活性实现难度Prompt 约束低高低后处理正则中中中Pydantic 验证高中低约束解码高低高Agent 拓扑高高中二、快速上手基础版Prompt 约束prompt 请输出 JSON 格式包含以下字段 - name: 字符串 - age: 整数 - hobbies: 字符串数组 只输出 JSON不要其他内容。 不稳定经常多输出内容或格式不对。改进版Pydantic 验证from pydantic import BaseModel, Field from typing import List import json class UserInfo(BaseModel): name: str Field(description姓名) age: int Field(description年龄, ge0, le150) hobbies: List[str] Field(description爱好列表) def extract_json(text: str) - dict: # 提取 JSON 部分 start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(未找到 JSON) json_str text[start:end1] data json.loads(json_str) # 验证 user UserInfo(**data) return user.model_dump() # 使用 text 本文叫张三30岁喜欢打篮球和编程 result extract_json(f{{name: 张三, age: 30, hobbies: [打篮球, 编程]}}) print(result)三、核心 API / 深水区3.1 结构化输出技术速查技术解决的问题使用时机格式示例指导输出格式每次 PromptPydantic 校验类型验证解析后容错解析格式不正确时兜底解析失败多步生成复杂结构分步嵌套数据3.2 容错解析器import json import re class LenientParser: def parse_json(self, text: str) - dict: # 1. 直接解析 try: return json.loads(text) except: pass # 2. 提取 JSON 块 json_match re.search(r\{.*\}, text, re.DOTALL) if json_match: try: return json.loads(json_match.group()) except: pass # 3. 修复常见错误 fixed text.strip() fixed re.sub(r, , fixed) fixed re.sub(r,(\s*[}\]]), r\1, fixed) try: return json.loads(fixed) except: pass raise ValueError(无法解析 JSON) def parse_with_defaults(self, text: str, defaults: dict) - dict: try: data self.parse_json(text) return {**defaults, **data} except: return defaults3.3 分步生成复杂结构class StepwiseStructGenerator: def __init__(self, llm): self.llm llm def generate(self, schema: dict) - dict: result {} for field, field_type in schema.items(): prompt f 生成字段 {field} 的值类型为 {field_type}。 要求只输出值不要字段名不要其他内容。 raw_value self.llm(prompt).strip() try: if field_type int: result[field] int(raw_value) elif field_type float: result[field] float(raw_value) elif field_type list: result[field] eval(raw_value) else: result[field] raw_value except: result[field] None return result四、实战演练完整的 Agent 拓扑结构化输出系统from typing import Dict, Any, List, Optional from pydantic import BaseModel, ValidationError import json import re class StructConfig(BaseModel): format: str json strict: bool True max_retries: int 3 class FormatNode: def process(self, input_text: str) - str: return input_text class ValidateNode: def process(self, schema: type, data: dict) - bool: try: schema(**data) return True except ValidationError: return False class FixNode: def process(self, data: dict, errors: str) - dict: # 修正错误的字段 return data class StructuredOutputPipeline: def __init__(self, llm, config: StructConfig): self.llm llm self.config config self.format_node FormatNode() self.validate_node ValidateNode() self.fix_node FixNode() def generate(self, schema: type, context: str) - Optional[dict]: schema_json schema.model_json_schema() for attempt in range(self.config.max_retries): # 1. 生成 prompt f根据以下内容生成符合 Schema 的 JSON 内容{context} Schema{json.dumps(schema_json, ensure_asciiFalse)} 只输出 JSON 格式不要其他任何内容 raw_output self.llm(prompt) formatted self.format_node.process(raw_output) # 2. 解析 data self._safe_parse(formatted) if not data: continue # 3. 校验 if self.validate_node.process(schema, data): return data # 4. 修正循环继续 if self.config.strict: continue return None def _safe_parse(self, text: str) - Optional[dict]: json_match re.search(r\{.*\}, text, re.DOTALL) if not json_match: return None try: return json.loads(json_match.group()) except: return None class OrderInfo(BaseModel): order_id: str amount: float status: str items: List[str] pipeline StructuredOutputPipeline(llm, StructConfig()) result pipeline.generate(OrderInfo, 订单 12345金额 99.9已发货包含书和笔) print(result)五、避坑指南与最佳实践 **技巧Pydantic Schema 作为 Prompt把 Schema 放到 Prompt 里模型的输出会规范很多。⚠️ **警告不要期望一次成功要有多步验证和重试机制。✅ **推荐容错解析 Pydantic 校验双重保障解析失败有兜底。六、综合实战演示生产级结构化输出系统from typing import Any, Dict, Optional, Type from pydantic import BaseModel, Field import json import re import time class RobustStructGenerator: def __init__(self, llm, max_retries3): self.llm llm self.max_retries max_retries self.stats {attempts: 0, success: 0, failures: 0} def generate(self, schema: Type[BaseModel], context: str) - Optional[BaseModel]: schema_json schema.model_json_schema() field_descriptions self._get_descriptions(schema) for attempt in range(self.max_retries): self.stats[attempts] 1 prompt self._build_prompt(schema_json, field_descriptions, context) response self.llm(prompt) parsed self._parse_robust(response) if not parsed: continue try: instance schema(**parsed) self.stats[success] 1 return instance except Exception as e: if attempt self.max_retries - 1: self.stats[failures] 1 return self._create_default(schema) continue return None def _build_prompt(self, schema, descriptions, context): return f生成 JSON 数据 Schema{json.dumps(schema, ensure_asciiFalse)} 信息{context} 字段说明{json.dumps(descriptions, ensure_asciiFalse)} 输出要求严格的 JSON 格式不要多余内容。 def _get_descriptions(self, schema): descriptions {} for name, field in schema.model_fields.items(): if field.description: descriptions[name] field.description return descriptions def _parse_robust(self, text: str) - Optional[dict]: text text.strip() start text.find({) end text.rfind(}) if start -1 or end -1: return None json_str text[start:end1] # 修复常见错误 json_str re.sub(r, , json_str) json_str re.sub(r,(\s*[}\]]), r\1, json_str) json_str re.sub(rTrue, true, json_str) json_str re.sub(rFalse, false, json_str) json_str re.sub(rNone, null, json_str) try: return json.loads(json_str) except: return None def _create_default(self, schema): try: return schema() except: return None def get_stats(self): return self.stats class ProductInfo(BaseModel): name: str Field(description商品名称) price: float Field(description价格) category: str Field(description分类) gen RobustStructGenerator(llm) product gen.generate(ProductInfo, 一个苹果手机价格 5999属于电子产品类) if product: print(f名称: {product.name}, 价格: {product.price}) print(f统计: {gen.get_stats()})七、总结Agent 拓扑设计模式解决结构化输出Prompt 约束 Pydantic 校验双重保障确保输出格式正确容错解析器兜底处理各种格式错误和边缘情况多步重试机制一次失败自动重试提高成功率Schema 驱动生成用 Pydantic Schema 指导模型输出这样搞大模型输出的格式问题就不是问题了。
大模型输出格式总不对？这套结构化解析方案一劳永逸
大模型输出格式总不对这套结构化解析方案一劳永逸前言老王大模型又输出格式错误了JSON 缺逗号字段类型不对 后端工程师小李烦躁地说。本文看了看日志发现这是第100次格式错误了。你这是缺少系统化的结构化输出方案啊那该怎么办每次都要人工处理吗看来得分享本文们团队的经验了。今天聊聊如何用 Agent 拓扑设计模式解决结构化输出问题。一、底层原理1.1 大模型结构化输出的难点大模型本质是文本生成不是 JSON 生成器graph TD A[大模型生成] -- B[文本输出] B -- C{期望结构化} C --|JSON| D[可能格式错误] C --|XML| E[可能标签不匹配] D -- F[解析失败] F -- G[重试] G -- H[业务卡顿] I[解决方案] -- J[约束解码] I -- K[后处理验证] I -- L[分步生成]核心问题大模型没有类型系统输出是概率采样不稳定复杂嵌套结构容易出错频繁出错影响业务1.2 结构化方案对比方案稳定性灵活性实现难度Prompt 约束低高低后处理正则中中中Pydantic 验证高中低约束解码高低高Agent 拓扑高高中二、快速上手基础版Prompt 约束prompt 请输出 JSON 格式包含以下字段 - name: 字符串 - age: 整数 - hobbies: 字符串数组 只输出 JSON不要其他内容。 不稳定经常多输出内容或格式不对。改进版Pydantic 验证from pydantic import BaseModel, Field from typing import List import json class UserInfo(BaseModel): name: str Field(description姓名) age: int Field(description年龄, ge0, le150) hobbies: List[str] Field(description爱好列表) def extract_json(text: str) - dict: # 提取 JSON 部分 start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(未找到 JSON) json_str text[start:end1] data json.loads(json_str) # 验证 user UserInfo(**data) return user.model_dump() # 使用 text 本文叫张三30岁喜欢打篮球和编程 result extract_json(f{{name: 张三, age: 30, hobbies: [打篮球, 编程]}}) print(result)三、核心 API / 深水区3.1 结构化输出技术速查技术解决的问题使用时机格式示例指导输出格式每次 PromptPydantic 校验类型验证解析后容错解析格式不正确时兜底解析失败多步生成复杂结构分步嵌套数据3.2 容错解析器import json import re class LenientParser: def parse_json(self, text: str) - dict: # 1. 直接解析 try: return json.loads(text) except: pass # 2. 提取 JSON 块 json_match re.search(r\{.*\}, text, re.DOTALL) if json_match: try: return json.loads(json_match.group()) except: pass # 3. 修复常见错误 fixed text.strip() fixed re.sub(r, , fixed) fixed re.sub(r,(\s*[}\]]), r\1, fixed) try: return json.loads(fixed) except: pass raise ValueError(无法解析 JSON) def parse_with_defaults(self, text: str, defaults: dict) - dict: try: data self.parse_json(text) return {**defaults, **data} except: return defaults3.3 分步生成复杂结构class StepwiseStructGenerator: def __init__(self, llm): self.llm llm def generate(self, schema: dict) - dict: result {} for field, field_type in schema.items(): prompt f 生成字段 {field} 的值类型为 {field_type}。 要求只输出值不要字段名不要其他内容。 raw_value self.llm(prompt).strip() try: if field_type int: result[field] int(raw_value) elif field_type float: result[field] float(raw_value) elif field_type list: result[field] eval(raw_value) else: result[field] raw_value except: result[field] None return result四、实战演练完整的 Agent 拓扑结构化输出系统from typing import Dict, Any, List, Optional from pydantic import BaseModel, ValidationError import json import re class StructConfig(BaseModel): format: str json strict: bool True max_retries: int 3 class FormatNode: def process(self, input_text: str) - str: return input_text class ValidateNode: def process(self, schema: type, data: dict) - bool: try: schema(**data) return True except ValidationError: return False class FixNode: def process(self, data: dict, errors: str) - dict: # 修正错误的字段 return data class StructuredOutputPipeline: def __init__(self, llm, config: StructConfig): self.llm llm self.config config self.format_node FormatNode() self.validate_node ValidateNode() self.fix_node FixNode() def generate(self, schema: type, context: str) - Optional[dict]: schema_json schema.model_json_schema() for attempt in range(self.config.max_retries): # 1. 生成 prompt f根据以下内容生成符合 Schema 的 JSON 内容{context} Schema{json.dumps(schema_json, ensure_asciiFalse)} 只输出 JSON 格式不要其他任何内容 raw_output self.llm(prompt) formatted self.format_node.process(raw_output) # 2. 解析 data self._safe_parse(formatted) if not data: continue # 3. 校验 if self.validate_node.process(schema, data): return data # 4. 修正循环继续 if self.config.strict: continue return None def _safe_parse(self, text: str) - Optional[dict]: json_match re.search(r\{.*\}, text, re.DOTALL) if not json_match: return None try: return json.loads(json_match.group()) except: return None class OrderInfo(BaseModel): order_id: str amount: float status: str items: List[str] pipeline StructuredOutputPipeline(llm, StructConfig()) result pipeline.generate(OrderInfo, 订单 12345金额 99.9已发货包含书和笔) print(result)五、避坑指南与最佳实践 **技巧Pydantic Schema 作为 Prompt把 Schema 放到 Prompt 里模型的输出会规范很多。⚠️ **警告不要期望一次成功要有多步验证和重试机制。✅ **推荐容错解析 Pydantic 校验双重保障解析失败有兜底。六、综合实战演示生产级结构化输出系统from typing import Any, Dict, Optional, Type from pydantic import BaseModel, Field import json import re import time class RobustStructGenerator: def __init__(self, llm, max_retries3): self.llm llm self.max_retries max_retries self.stats {attempts: 0, success: 0, failures: 0} def generate(self, schema: Type[BaseModel], context: str) - Optional[BaseModel]: schema_json schema.model_json_schema() field_descriptions self._get_descriptions(schema) for attempt in range(self.max_retries): self.stats[attempts] 1 prompt self._build_prompt(schema_json, field_descriptions, context) response self.llm(prompt) parsed self._parse_robust(response) if not parsed: continue try: instance schema(**parsed) self.stats[success] 1 return instance except Exception as e: if attempt self.max_retries - 1: self.stats[failures] 1 return self._create_default(schema) continue return None def _build_prompt(self, schema, descriptions, context): return f生成 JSON 数据 Schema{json.dumps(schema, ensure_asciiFalse)} 信息{context} 字段说明{json.dumps(descriptions, ensure_asciiFalse)} 输出要求严格的 JSON 格式不要多余内容。 def _get_descriptions(self, schema): descriptions {} for name, field in schema.model_fields.items(): if field.description: descriptions[name] field.description return descriptions def _parse_robust(self, text: str) - Optional[dict]: text text.strip() start text.find({) end text.rfind(}) if start -1 or end -1: return None json_str text[start:end1] # 修复常见错误 json_str re.sub(r, , json_str) json_str re.sub(r,(\s*[}\]]), r\1, json_str) json_str re.sub(rTrue, true, json_str) json_str re.sub(rFalse, false, json_str) json_str re.sub(rNone, null, json_str) try: return json.loads(json_str) except: return None def _create_default(self, schema): try: return schema() except: return None def get_stats(self): return self.stats class ProductInfo(BaseModel): name: str Field(description商品名称) price: float Field(description价格) category: str Field(description分类) gen RobustStructGenerator(llm) product gen.generate(ProductInfo, 一个苹果手机价格 5999属于电子产品类) if product: print(f名称: {product.name}, 价格: {product.price}) print(f统计: {gen.get_stats()})七、总结Agent 拓扑设计模式解决结构化输出Prompt 约束 Pydantic 校验双重保障确保输出格式正确容错解析器兜底处理各种格式错误和边缘情况多步重试机制一次失败自动重试提高成功率Schema 驱动生成用 Pydantic Schema 指导模型输出这样搞大模型输出的格式问题就不是问题了。
