大模型应用落地中基于 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 组件

Agent 类型 职责 关键能力
输入处理 Agent 理解用户需求 意图识别、上下文分析
格式定义 Agent 定义输出结构 Schema 设计、类型约束
提示词构建 Agent 生成结构化提示 模板管理、动态填充
输出解析 Agent 解析模型输出 格式提取、错误修正
验证 Agent 校验结果 格式验证、语义检查
错误处理 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. 可扩展性：支持多种输出格式和解析策略

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

# 后端 # golang
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

用音频处理项目串联介绍：大模型技术演进路线

avatar AtomGit开源社区
cover

免费AI编程工具每日3000万Token，注册即领专业版会员

avatar AtomGit开源社区
cover

专业干货！AI写专著工具推荐，助力20万字专著快速生成！

avatar AtomGit开源社区