Agent架构深度解析：如何高效将工具调用数据返回给大模型

1次阅读

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

在构建基于大模型的 Agent 系统时，工具调用数据的返回机制是系统设计中的关键环节。开发者常常会遇到数据格式不一致、上下文丢失、并发竞争等问题，直接影响 Agent 的响应质量和用户体验。本文将深入探讨三种主流的数据返回方案，并通过代码示例和性能分析，帮助你构建高效可靠的数据传输通道。

在 Agent 系统中，工具调用后返回的数据需要经过处理后才能供大模型使用。这个过程中常见的问题包括：

数据格式错乱：不同工具返回的数据格式差异大，直接拼接可能导致大模型无法正确解析
上下文丢失：在多轮对话场景中，工具返回数据与当前对话上下文的关联性可能丢失
并发竞争：高并发场景下，多个工具同时返回数据可能导致数据覆盖或顺序错乱
token 超限：工具返回数据量过大时可能超出大模型的输入 token 限制

这是最简单的实现方式，直接将工具返回的文本拼接到大模型的输入中。

优点：

实现简单，无需额外处理
延迟最低

缺点：

容易造成格式污染
无法处理结构化数据
难以维护数据一致性

将工具返回的数据封装为结构化 JSON 格式，再传递给大模型。

优点：

保持数据结构一致性
便于后续处理和分析

缺点：

需要处理 Schema 兼容性问题
增加了序列化 / 反序列化开销

在工具和大模型之间增加一个中间件层，专门处理数据转换和传输。

优点：

解耦工具和大模型
可以实现复杂的数据处理逻辑
支持数据校验和转换

缺点：

增加了系统复杂性
带来额外的延迟

下面我们通过 Python 代码示例展示如何实现结构化数据返回。

from pydantic import BaseModel
from typing import Dict, Any

class ToolResponse(BaseModel):
    tool_name: str
    status: str
    data: Dict[str, Any]
    metadata: Dict[str, Any] = {}

import asyncio
from typing import Optional

class DataChannel:
    def __init__(self):
        self.queue = asyncio.Queue()

    async def put(self, response: ToolResponse):
        await self.queue.put(response)

    async def get(self) -> Optional[ToolResponse]:
        try:
            return await asyncio.wait_for(self.queue.get(), timeout=1.0)
        except asyncio.TimeoutError:
            return None

from pydantic import ValidationError

class ValidationMiddleware:
    def __init__(self, next_handler):
        self.next_handler = next_handler

    async def handle(self, raw_data: dict):
        try:
            response = ToolResponse(**raw_data)
            return await self.next_handler.handle(response)
        except ValidationError as e:
            print(f"Validation failed: {e}")
            raise