企业官网建设流程全解析

大模型输出格式总不对？这套结构化解析方案一劳永逸

前言

"老王，大模型又输出格式错误了！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 约束

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="年龄", ge=0, le=150) 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:end+1] 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 结构化输出技术速查

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 defaults

3.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_ascii=False)} 只输出 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_retries=3): 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_ascii=False)} 信息：{context} 字段说明：{json.dumps(descriptions, ensure_ascii=False)} 输出要求：严格的 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:end+1] # 修复常见错误 json_str = re.sub(r"'", '"', json_str) json_str = re.sub(r",(\s*[}\]])", r"\1", json_str) json_str = re.sub(r"True", "true", json_str) json_str = re.sub(r"False", "false", json_str) json_str = re.sub(r"None", "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 指导模型输出

这样搞，大模型输出的格式问题就不是问题了。