跳到主要内容

4. 输出侧:工具、MCP 桥接、Token

这章讲 agent 怎么和外部世界连:本地工具的标准形态、把远程 MCP 服务器“编译”成本地工具、以及 token 计数/裁剪与 hook。

4.1 BaseTool:和 agent 同构的工具接口​

工具就是另一种“吃 schema、吐 schema”的东西。BaseTool[InputSchema, OutputSchema] 用的是和 AtomicAgent 一模一样的泛型截获 + 三级回退机制(base/base_tool.py:50-105),只多一个抽象方法 run(params) -> OutputSchema(base/base_tool.py:127-141)。

一个真实工具长这样(计算器,atomic-forge/tools/calculator/tool/calculator.py:54-91):输入 schema 描述“一个数学表达式”,run 用 sympy.sympify 求值,返回输出 schema。tool_name / tool_description 默认从输入 schema 的 JSON schema title/description 取,可被 BaseToolConfig 覆盖(base/base_tool.py:107-125)。

这种同构是“链式拼接”能成立的根因:agent 和工具的接口形状相同,所以一个的输出能当另一个的输入(回看 02-schema-and-chaining.md)。

4.2 MCPFactory:把远程 MCP 工具变成本地 BaseTool 子类​

MCP(Model Context Protocol,模型上下文协议)服务器对外暴露工具/资源/提示,但都是 JSON schema 描述的“远程能力”。MCPFactory 的工作是:连上去、读定义、在运行时动态合成一批 BaseTool 子类,让远程工具用起来和本地工具毫无差别。

流程(以工具为例,connectors/mcp/mcp_factory.py:124-388):

MCP 服务器                MCPDefinitionService          MCPFactory
    │  list tools  ──►  fetch_*_definitions()  ──►  _create_tool_classes()
    │                   (name/description/             │
    │                    input_schema/output_schema)   │
    │                                                  ├─ SchemaTransformer 把 JSON schema
    │                                                  │   → 真正的 Pydantic 输入/输出模型
    │                                                  ├─ 合成 async run + sync run 闭包
    │                                                  └─ types.new_class() 造出 BaseTool 子类
    ▼
生成的工具类:tool.run(params) 时才真正连服务器、call_tool、把结果塞回输出 schema

三个值得看的设计:

(1) JSON schema → Pydantic 模型的翻译器。 SchemaTransformer.create_model_from_schema 递归处理 $ref / oneOf / anyOf / 数组 / 嵌套对象,把远程 JSON schema 变成本地 Pydantic 类(connectors/mcp/schema_transformer.py:147-206)。输入 schema 会额外加一个 tool_name 字面量字段(Literal["该工具名"]),输出 schema 不加——因为输出时工具已经选定了(connectors/mcp/schema_transformer.py:189-195)。这个字面量字段是下面编排 Union 能区分的关键。

(2) 三种传输 + 两种会话模式。 工具被调用时,按 transport_type 选 STDIO / HTTP_STREAM / SSE 三种传输(connectors/mcp/mcp_factory.py:187-217);并支持“每次调用开新连接”(legacy)或复用构造时传入的常驻 client_session(connectors/mcp/mcp_factory.py:234-241)。HTTP 用尾斜杠 /mcp/ 规避重定向,代码里还附了对应的 python-sdk issue 链接(connectors/mcp/mcp_factory.py:200-206)。

(3) 结果抽取有优先级。 当工具带 typed output schema 时,从结果里抽结构化数据有一条明确的降级链(connectors/mcp/mcp_factory.py:242-327):

① tool_result.structuredContent(MCP 规范主路径)
② content[0].text 当 JSON 解析
③ content[0].data 字典
④ dict 里的 structuredContent / content 键
⑤ 直接当 dict 用
都失败 ⇒ 报错(typed schema 没有通用 result 字段,不能兜底)

没有 typed schema 的工具走老路:把结果塞进通用的 MCPToolOutputSchema(result=...)(connectors/mcp/mcp_factory.py:329-337)。

4.3 编排 schema:让一个 agent 从多个工具里选​

有了一堆工具类,怎么让 agent “选一个工具并填参数”?create_orchestrator_schema 把所有工具的输入 schema 组成一个 Union,做成一个输出字段 tool_parameters(connectors/mcp/mcp_factory.py:390-466):

# 摘自 create_orchestrator_schema,connectors/mcp/mcp_factory.py:424-432(节选)
ToolUnion = Union[tuple(tool_schemas)]      # 所有工具输入 schema 的联合
field_defs["tool_parameters"] = (ToolUnion, Field(..., description="..."))

模型被要求输出一个匹配某个工具 schema 的对象;因为每个工具输入 schema 带 tool_name 字面量字段(见 4.2),这个 Union 实际上是一个判别联合(discriminated union)——tool_name 的值就告诉你模型选了哪个工具。资源和提示同理,各占 resource_parameters / prompt_parameters 字段。

这就是“orchestration agent”的骨架:agent 的输出 schema = 工具集合的 Union,一次 LLM 调用同时完成“选工具 + 出参数”。

4.4 Token 计数与历史裁剪​

max_context_tokens 设了之后,每次 run 前会 _trim_context()。它的 token 数从哪来?get_context_token_count() 把上下文按 Instructor 的真实方式序列化后交给 TokenCounter(agents/atomic_agent.py:474-549)。

精妙处:它连 schema/tools 的开销也算进去。TOOLS 模式下用 Instructor 真实生成的 tools 定义算 token;JSON 模式下把 schema 文本追加进系统消息再算(agents/atomic_agent.py:524-537)。TokenCounter 底层走 LiteLLM 的 token_counter,因此与厂商无关——OpenAI/Anthropic/Gemini 等 100+ 模型通吃(utils/token_counter.py:81-113)。

算 tools 开销的小技巧:用“空消息 + tools”减去“空消息无 tools”,差值就是 tools 净开销(utils/token_counter.py:189-195)。裁剪本身按 turn 整轮删最老的,直到总数达标;若单个 turn 就超限则直接报错让你调大预算(agents/atomic_agent.py:317-347)。

4.5 Hooks:监控与重试​

agent 集成了 Instructor 的 hook 系统。register_hook(event, handler) 既存在本地 dict,也转发给 instructor 客户端的 .on()(agents/atomic_agent.py:740-754)。支持 parse:error(Pydantic 校验失败)、completion:kwargs/response/error/last_attempt 等事件(类 docstring,agents/atomic_agent.py:131-146)。本地派发 _dispatch_hook 对 handler 异常做隔离——一个 hook 抛错只记日志,不打断主流程(agents/atomic_agent.py:790-808)。

4.6 巧妙之处(可借鉴)​

  • docstring 即 description,且强制非空——把“写文档”变成编译期约束,文档直接成为 prompt(base/base_io_schema.py:22-39)。
  • 泛型参数当 schema,用 __init_subclass__ 偷出来——绕开不可靠的 __orig_class__,agent 与 tool 共用同一招(agents/atomic_agent.py:165-181、base/base_tool.py:50-65)。
  • 远程 MCP 工具运行时编译成本地类——types.new_class + 闭包,让远程能力和本地工具同构(connectors/mcp/mcp_factory.py:358-382)。
  • judge/编排靠 Union + 字面量判别字段——一次调用选工具又填参(connectors/mcp/mcp_factory.py:424-432、schema_transformer.py:189-195)。
  • token 计数厂商无关,且把 schema/tools 开销也纳入(agents/atomic_agent.py:474-549、utils/token_counter.py)。

4.7 边界与局限(诚实)​

  • 不是编排/状态机框架。 它给你“原子积木”,但多步流程的控制流要你自己写(循环、分支、并行都在你的代码里,框架不管)。
  • 强依赖 Instructor + Pydantic v2 + Python ≥3.12(pyproject.toml:18-30)。底层结构化输出的能力边界 = Instructor 的能力边界。
  • MCP 动态类在异常时静默跳过。 某个工具生成失败只记 error 然后 continue(connectors/mcp/mcp_factory.py:384-387),不会整体失败——好处是健壮,坏处是你可能没注意到少了个工具。
  • 历史 load() 要求 schema 类全局可 import,局部定义的 schema 无法复活(context/chat_history.py:298-314)。
  • token 裁剪是“尽力”:get_max_tokens 拿不到模型信息时 utilization 为 None(utils/token_counter.py:154-161),裁剪仍按你给的绝对值进行。

4.8 横向对比(同 shelf 兄弟)​

  • 与 Letta 这类“有状态、自管理记忆”的 agent 不同,Atomic Agents 无内建长期记忆/自治循环——它选择当“积木盒”而非“大脑”,把状态与编排留给使用者。
  • 与重编排框架(图/DAG 式)相比,它的“编排”是类型对齐而非显式图;代价是复杂流程要手写控制流,收益是几乎零学习曲线、易测易调。
  • 与直接裸用 Instructor 相比,它多给了:提示分节模板、turn 化历史、MCP 桥接、token 裁剪、hook——即“把 Instructor 调用组织成 agent”的那层。

4.9 代码地图(导航索引)​

主题文件路径符号名
核心循环 / 四个 run 入口atomic-agents/atomic_agents/agents/atomic_agent.pyAtomicAgent, run, run_stream, run_async, run_async_stream
消息组装 / 跨厂商角色atomic-agents/atomic_agents/agents/atomic_agent.py_prepare_messages, _build_system_messages, tool_result_role
配置atomic-agents/atomic_agents/agents/atomic_agent.pyAgentConfig
泛型 schema 截获atomic-agents/atomic_agents/agents/atomic_agent.py__init_subclass__, input_schema, output_schema
流式增量解析补丁atomic-agents/atomic_agents/agents/atomic_agent.pymodel_from_chunks_patched
token 计数 / 裁剪atomic-agents/atomic_agents/agents/atomic_agent.pyget_context_token_count, _trim_context
schema 基类 / docstring 强制atomic-agents/atomic_agents/base/base_io_schema.pyBaseIOSchema, _validate_description, model_json_schema
工具基类atomic-agents/atomic_agents/base/base_tool.pyBaseTool, BaseToolConfig, run
系统提示生成atomic-agents/atomic_agents/context/system_prompt_generator.pySystemPromptGenerator, generate_prompt
动态上下文atomic-agents/atomic_agents/context/system_prompt_generator.pyBaseDynamicContextProvider, get_info
历史 / turn / 序列化atomic-agents/atomic_agents/context/chat_history.pyChatHistory, initialize_turn, delete_turn_id, dump, load
多模态抽取atomic-agents/atomic_agents/context/chat_history.py_extract_multimodal_info, _process_multimodal_paths
MCP 工厂atomic-agents/atomic_agents/connectors/mcp/mcp_factory.pyMCPFactory, create_tools, _create_tool_classes, create_orchestrator_schema
MCP 公共 APIatomic-agents/atomic_agents/connectors/mcp/mcp_factory.pyfetch_mcp_tools, fetch_mcp_attributes_with_schema
JSON→Pydantic 翻译atomic-agents/atomic_agents/connectors/mcp/schema_transformer.pySchemaTransformer, create_model_from_schema, json_to_pydantic_field
MCP 定义抓取 / 传输atomic-agents/atomic_agents/connectors/mcp/mcp_definition_service.pyMCPDefinitionService, MCPTransportType, MCPToolDefinition
token 计数底层atomic-agents/atomic_agents/utils/token_counter.pyTokenCounter, count_context, get_max_tokens
真实工具范例atomic-forge/tools/calculator/tool/calculator.pyCalculatorTool, CalculatorToolInputSchema