如何规范化处理agent工具调用的返回结果：从混乱到标准化的实践指南

1次阅读

没有评论

共计 2229 个字符，预计需要花费 6 分钟才能阅读完成。

在开发基于 agent 的工具调用系统时，返回结果的不规范性常常导致后续处理逻辑复杂、错误难以追踪。以下是几个典型的场景：

字段不一致 ：同一工具在不同情况下返回的字段名或类型不一致
错误处理混乱 ：错误信息格式不统一，有的返回字符串，有的返回字典
嵌套结构随意 ：有些结果过度嵌套，有些则过于扁平
缺少元信息 ：如请求 ID、耗时等关键信息缺失

这些问题会导致：

下游业务逻辑需要编写大量防御性代码
错误排查困难，日志分析效率低下
系统监控指标难以统一采集
API 文档与实际行为脱节

优点：从源头解决问题，最彻底的方案
缺点：需要协调所有 agent 开发者，改动成本高
适用场景：新系统或对 agent 有完全控制权的场景

优点：可以针对每个工具单独适配，灵活度高
缺点：适配器数量会随着工具增加而膨胀
适用场景：工具数量较少且差异大的情况

优点：集中处理，不影响现有 agent 逻辑
缺点：需要额外的性能开销
适用场景：大多数生产环境，特别是已有大量 agent 的场景

from typing import Any, Dict
import json
import logging
from datetime import datetime

class NormalizationMiddleware:
    """
    Agent 结果规范化中间件
    主要功能：1. 统一错误格式
    2. 标准化成功响应
    3. 添加元数据
    4. 日志记录
    """

    def __init__(self):
        self.logger = logging.getLogger(__name__)

    async def process_response(self, 
                             agent_name: str, 
                             raw_response: Any) -> Dict[str, Any]:
        """处理 agent 原始响应"""
        start_time = datetime.now()

        try:
            # 第一步：统一错误处理
            normalized = self._handle_errors(raw_response)

            # 第二步：标准化结构
            if not normalized.get('error'):
                normalized['data'] = self._standardize_data(
                    agent_name, 
                    normalized.get('data', {})
                )

            # 第三步：添加元信息
            normalized['metadata'] = {
                'agent': agent_name,
                'processed_at': datetime.utcnow().isoformat(),
                'duration_ms': (datetime.now() - start_time).total_seconds() * 1000}

            return normalized

        except Exception as e:
            self.logger.exception(f"Normalization failed for {agent_name}")
            return {
                'error': True,
                'code': 'NORMALIZATION_FAILED',
                'message': str(e),
                'metadata': {
                    'agent': agent_name,
                    'processed_at': datetime.utcnow().isoformat()
                }
            }

    def _handle_errors(self, response: Any) -> Dict[str, Any]:
        """统一错误格式"""
        # 实现细节省略...
        pass

    def _standardize_data(self, agent_name: str, data: Any) -> Dict[str, Any]:
        """根据 agent 类型标准化数据结构"""
        # 实现细节省略...
        pass