Agno 工具调用协议与中间件链
Agno 工具调用协议与中间件链
学习目标
本章将分析 Agno 的工具系统架构：
- Toolkit + Function 两层封装的设计动机
@tooldecorator 如何自动从 Python 函数生成 JSON Schema- Callable Factory 模式实现运行时动态工具
- 工具中间件 Hook 的环绕执行
前置知识
建议先阅读：
- 工具调用协议与执行模型 — OpenAI Function Calling 协议
- 工具调用中间件链 — 中间件模式在工具调用中的应用
下文假设你已理解上述概念，直接聚焦 Agno 的具体实现。
项目实践
Toolkit + Function 两层封装
Agno 的工具系统采用两层抽象：
# 伪代码：两层封装class Function: """单点工具定义""" name: str # 工具名称，对应 LLM tool_call.name description: str # 工具描述，LLM 用于判断是否调用 parameters: dict # JSON Schema，LLM 用于生成参数 callable: Callable # 实际执行的 Python 函数 hooks: Optional[List] # 工具级 Hook
class Toolkit: """一组 Function 的容器""" name: str # 命名空间 functions: List[Function] # 包含的工具为什么两层？ Toolkit 提供命名空间和批量操作（一次性注册多个工具），Function 是 LLM 工具调用协议的最小单位。这种设计使得：
- 新增工具只需定义一个
Toolkit子类 - LLM 看到的是扁平的工具列表，不感知 Toolkit 结构
- 不同 Toolkit 的工具可以混合使用
@tool 装饰器自动提取 Schema
Agno 的 @tool decorator 将普通 Python 函数转为 Function：
# 伪代码：@tool decorator@tooldef get_weather(city: str, unit: Literal["celsius", "fahrenheit"] = "celsius") -> str: """获取指定城市的天气信息。""" ...
# 自动生成的 Function# name: "get_weather"# description: "获取指定城市的天气信息。"# parameters: {# "type": "object",# "properties": {# "city": {"type": "string"},# "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}# },# "required": ["city"]# }参数 Schema 从函数签名和类型注解自动提取。Literal 类型转为 enum，Optional 字段不加入 required。
Callable Factory 模式
tools 字段接受三种类型：
| 类型 | 行为 | 示例 |
|---|---|---|
List[Toolkit/Function] | 静态工具列表 | tools=[WeatherToolkit(), CalculatorToolkit()] |
Callable[..., List] | 运行时动态生成 | tools=lambda ctx: get_tools_for_user(ctx.user_id) |
Dict | 直接传 JSON Schema | 手动定义工具定义 |
Callable Factory 在每次 run 时调用，注入 agent 和 run_context：
# 伪代码：动态工具工厂def get_db_tools(run_context): db_name = run_context.session_state.get("db_name", "default") return [QueryTool(db=db_name), InsertTool(db=db_name)]
agent = Agent(tools=get_db_tools, cache_callables=True)cache_callables=True 缓存 factory 结果，避免同一 run 内重复调用。
工具中间件 Hook
tool_hooks 是环绕每个工具调用的函数列表：
# 伪代码：工具中间件def log_tool_hook(function_name, args, result, **kwargs): log.info(f"Calling {function_name} with {args}") # 可以修改 result return result
agent = Agent(tool_hooks=[log_tool_hook, pii_mask_hook])每个 Hook 接收工具名称、参数和结果。与 LangChain 的责任链中间件不同，Agno 的工具 Hook 是简单的函数列表——每个 Hook 独立执行，不形成嵌套链。这种设计更轻量，适合不需要复杂状态流转的场景。
工具确定（Dynamic Tool Filtering）
在管线步骤 5，Agno 执行 determine_tools_for_model——不是所有声明的工具都发送给模型：
# 伪代码：工具确定_tools = determine_tools_for_model( agent, model=agent.model, processed_tools=processed_tools, run_response=run_response, session=agent_session, run_context=run_context,)过滤逻辑考虑：
- 模型的能力（某些模型不支持并行工具调用）
- 当前 session 状态的权限
- 工具之间的互斥关系
问题与规避
1. Callable Factory 的缓存陷阱
cache_callables=True 时，factory 结果在 Agent 生命周期内缓存。如果 factory 依赖的外部状态变化（如数据库切换），缓存的旧工具列表不会更新。
规避：在 factory 函数中使用 cache_callables=False，或使用自定义 callable_tools_cache_key 函数，在状态变化时返回不同的缓存键。
2. 工具数量过多导致上下文溢出
100+ 工具集成意味着单个 Agent 可能声明大量工具。每个工具的 JSON Schema 定义占用数百 token。
规避：使用 determine_tools_for_model 动态过滤，或在 Agent 配置中按场景拆分多个小 Agent，每个只携带相关工具。
3. @tool 装饰器的类型推断限制
@tool 从 Python 类型注解提取 JSON Schema，但不支持所有 Python 类型（如自定义 dataclass、泛型）。
规避：对复杂参数类型，使用 parameters 参数手动传 JSON Schema。
设计取舍
为什么工具 Hook 是函数列表而非责任链？
责任链中间件（如 LangChain）提供更强的控制力（可以拦截、跳过、重试），但实现复杂度高。Agno 的工具 Hook 选择简单函数列表：
- 优势：实现简单（几十行代码），用户容易理解
- 代价：无法在 Hook 中跳过工具执行或修改执行顺序
- 适用场景：日志、审计、简单的结果转换
替代方案：如果需要在 Hook 中做复杂控制流（如权限检查失败时跳过工具执行），使用 pre-hooks 的 guardrail 模式，在工具调用前拦截。