Agent架构实战：如何高效将工具调用结果返回给大模型

1次阅读

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

在构建 AI Agent 时，工具调用结果的返回处理是一个看似简单却暗藏玄机的环节。最近我在开发一个电商客服 Agent 时，就遇到了这样的问题：当调用商品查询接口返回的 JSON 数据直接拼接到 prompt 中时，大模型经常无法正确理解嵌套的规格参数，甚至有时会因为特殊字符导致整个对话崩溃。更糟的是，某次接口返回了包含用户手机号的调试信息，差点引发隐私泄露事故。

这些经历让我意识到，工具调用结果的返回绝非简单的字符串拼接。经过多次迭代，我总结了三种主流方案，下面结合 Python 示例详细解析。

这是新手最常采用的方案，看似简单却隐患重重：

# 危险示例：直接拼接 API 返回的 JSON
try:
    product_data = requests.get('api/products/123').json()
    prompt = f"用户询问的商品信息如下：{str(product_data)}"
except Exception as e:
    prompt = "查询失败：" + str(e)

优点：零学习成本，快速实现
致命缺点：
无法控制 JSON 嵌套结构对大模型的影响
可能泄露敏感字段
特殊字符（如 Markdown 符号）可能破坏 prompt 结构

使用 Jinja2 等模板引擎可以解决格式问题：

from jinja2 import Template

product_template = Template(""" 商品信息：名称：{{name}}
价格：¥{{price}}
规格：{% for spec in specs %}
- {{spec.name}}: {{spec.value}}{% endfor %}
""")

def safe_render(data):
    # 数据过滤层
    filtered = {'name': data.get('name', ''),'price': data.get('price', 0),'specs': [s for s in data.get('specs', []) 
                 if s['name'] not in ['internal_code']]
    }
    return product_template.render(filtered)

核心改进：
显式控制输出字段和格式
内置敏感字段过滤
支持条件渲染和循环
注意事项：
需要为每类工具编写模板
仍然需要处理原始数据解析

对于企业级系统，建议采用中间件设计：

class ResultMiddleware:
    def __init__(self):
        self.filters = {
            'products': self._filter_product,
            'users': self._filter_user
        }

    def process(self, tool_name, raw_data):
        handler = self.filters.get(tool_name, self._default_filter)
        return handler(raw_data)

    def _filter_product(self, data):
        return {'summary': f"{data['name']} (库存: {data['stock']})",
            'details': [f"{k}: {v}" for k,v in data.items() 
                       if k not in ['cost_price', 'supplier']]
        }

    def _default_filter(self, data):
        if isinstance(data, dict):
            return {k:v for k,v in data.items() 
                   if not k.startswith('_')}
        return str(data)[:500]  # 截断长文本

大模型的上下文窗口是宝贵资源，需精确计算：

import tiktoken

def count_tokens(text):
    encoder = tiktoken.get_encoding("cl100k_base")
    return len(encoder.encode(text))

# 在中间件中添加检查
def process_with_token_check(data, max_tokens=1000):
    result = self.process(data)
    token_count = count_tokens(json.dumps(result))
    if token_count > max_tokens:
        return self._truncate_result(result, max_tokens)
    return result

对于耗时工具调用，应采用异步模式：

import asyncio

async def call_tool(tool_func, callback):
    result = await tool_func()
    processed = middleware.process(result)
    await callback(processed)

# 使用示例
async def handle_user_query():
    tasks = [call_tool(fetch_product_info, update_ui),
        call_tool(check_inventory, update_ui)
    ]
    await asyncio.gather(*tasks)