大模型应用落地中基于 Agent 拓扑设计模式实现大模型结构化输出解析的工程路径

大模型应用落地中基于 Agent 拓扑设计模式实现大模型结构化输出解析的工程路径

一、结构化输出解析概述

在大模型应用中，结构化输出解析是将模型生成的非结构化文本转换为计算机可理解的数据格式的关键环节。Agent 拓扑设计模式为实现这一目标提供了系统化的解决方案。

flowchart TD A[用户请求] --> B[输入处理 Agent] B --> C[解析策略选择] C --> D[格式定义 Agent] C --> E[提示词构建 Agent] D --> F[Schema 定义] E --> G[结构化提示词] F --> H[LLM 生成] G --> H H --> I[输出解析 Agent] I --> J[格式验证] J --> K{验证通过?} K -->|是| L[结构化结果] K -->|否| M[错误处理 Agent] M --> N[重新生成] N --> H

二、Agent 拓扑设计模式

2.1 核心 Agent 组件

2.2 Agent 协作流程

class AgentCoordinator: def __init__(self): self.agents = { 'input_processor': InputProcessorAgent(), 'format_definer': FormatDefinerAgent(), 'prompt_builder': PromptBuilderAgent(), 'output_parser': OutputParserAgent(), 'validator': ValidatorAgent(), 'error_handler': ErrorHandlerAgent() } def execute(self, user_request): # 1. 输入处理 intent = self.agents['input_processor'].process(user_request) # 2. 格式定义 schema = self.agents['format_definer'].define(intent) # 3. 提示词构建 prompt = self.agents['prompt_builder'].build(schema) # 4. LLM 生成 raw_output = llm.generate(prompt) # 5. 输出解析 parsed = self.agents['output_parser'].parse(raw_output, schema) # 6. 验证 if self.agents['validator'].validate(parsed, schema): return parsed else: return self.agents['error_handler'].handle(parsed, schema)

三、结构化输出定义

3.1 Schema 设计

class SchemaDefinition: def __init__(self): self.types = { 'string': {'type': 'str', 'max_length': 1000}, 'integer': {'type': 'int', 'min': None, 'max': None}, 'float': {'type': 'float', 'min': None, 'max': None}, 'boolean': {'type': 'bool'}, 'array': {'type': 'list', 'items': None, 'max_items': 100}, 'object': {'type': 'dict', 'properties': None} } def define(self, structure): schema = {'type': 'object', 'properties': {}} for field_name, field_spec in structure.items(): field_type = field_spec.get('type', 'string') if field_type not in self.types: raise ValueError(f"未知类型: {field_type}") schema['properties'][field_name] = { 'type': field_type, **field_spec.get('constraints', {}) } return schema

3.2 JSON Schema 生成

def generate_json_schema(schema_def): json_schema = { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": {} } for field, spec in schema_def['properties'].items(): json_schema['properties'][field] = translate_to_json_schema(spec) return json_schema

四、提示词构建策略

4.1 结构化提示词模板

class PromptTemplateManager: def __init__(self): self.templates = {} def register_template(self, name, template): self.templates[name] = template def render(self, name, **kwargs): if name not in self.templates: raise ValueError(f"模板不存在: {name}") return self.templates[name].format(**kwargs)

4.2 动态提示词生成

def build_structured_prompt(user_query, schema): schema_str = json.dumps(schema, indent=2, ensure_ascii=False) prompt = f""" 请根据以下规则回答用户问题： 用户问题：{user_query} 输出格式要求： {schema_str} 注意事项： 1. 必须严格按照 JSON 格式输出 2. 确保所有字段都符合类型要求 3. 如果无法回答，返回包含 error 字段的 JSON 4. 不要输出任何额外文字 输出： """ return prompt

五、输出解析机制

5.1 多格式解析

class MultiFormatParser: def __init__(self): self.parsers = { 'json': self._parse_json, 'xml': self._parse_xml, 'yaml': self._parse_yaml, 'csv': self._parse_csv } def parse(self, raw_output, format_type='json'): if format_type not in self.parsers: raise ValueError(f"不支持的格式: {format_type}") return self.parsers[format_type](raw_output) def _parse_json(self, raw_output): try: return json.loads(raw_output) except json.JSONDecodeError: # 尝试修复常见格式问题 cleaned = self._clean_json(raw_output) return json.loads(cleaned) def _clean_json(self, text): # 移除前后的 markdown 标记 text = re.sub(r'^```json\s*', '', text) text = re.sub(r'\s*```$', '', text) return text.strip()

5.2 错误修正

class ErrorCorrector: def __init__(self): self.max_retries = 3 def correct(self, raw_output, schema, attempt=1): if attempt > self.max_retries: raise ValueError("解析失败，已达到最大重试次数") try: parsed = self._parse(raw_output) if self._validate(parsed, schema): return parsed # 生成修正提示 correction_prompt = self._build_correction_prompt(raw_output, schema) corrected_output = llm.generate(correction_prompt) return self.correct(corrected_output, schema, attempt + 1) except Exception as e: return self.correct(self._generate_retry_prompt(schema), schema, attempt + 1)

六、验证与质量保障

6.1 格式验证

class OutputValidator: def __init__(self): self.validator = jsonschema.Draft7Validator def validate(self, output, schema): try: self.validator(schema).validate(output) return True, None except jsonschema.ValidationError as e: return False, str(e)

6.2 语义验证

class SemanticValidator: def __init__(self): self.llm = ValidationModel() def validate(self, output, user_query): prompt = f""" 请验证以下输出是否准确回答了用户问题： 用户问题：{user_query} 输出：{json.dumps(output, indent=2)} 请判断输出是否： 1. 与问题相关 2. 内容准确 3. 格式正确 输出 YES 或 NO，并简要说明原因。 """ response = self.llm.generate(prompt) return response.strip().upper() == 'YES'

七、工程化实施建议

7.1 架构部署

flowchart TD A[API Gateway] --> B[Agent 调度层] B --> C[输入处理 Agent] B --> D[格式定义 Agent] B --> E[提示词构建 Agent] B --> F[输出解析 Agent] B --> G[验证 Agent] E --> H[LLM 服务] H --> F G --> I[(验证结果存储)] F --> J[结果缓存] J --> K[客户端]

7.2 性能优化

class PerformanceOptimizer: def __init__(self): self.cache = LRUCache(maxsize=1000) def optimize(self, user_query, schema): cache_key = hash((user_query, json.dumps(schema))) if cache_key in self.cache: return self.cache[cache_key] result = self._execute_pipeline(user_query, schema) self.cache[cache_key] = result return result

八、总结

基于 Agent 拓扑设计模式的结构化输出解析方案具有以下优势：

1. 模块化：每个 Agent 职责清晰，易于维护和扩展
2. 可复用：Schema 定义和提示词模板可重复使用
3. 鲁棒性：多层验证和错误处理确保输出质量
4. 可扩展性：支持多种输出格式和解析策略

通过系统化的工程实施，可以构建高质量、高可靠性的大模型结构化输出解析系统。