数据截至 2026-08-06(上游 commit 460c729002dc)

第 5 章 · 工具、记忆、工作流与对外托管

本章讲什么: 前四章讲的是 agent 怎么想。这一章讲它周围的四类基础设施:工具怎么定义和执行、记忆有哪几种策略、多步/多 agent 怎么编排、跑好的 agent 怎么对外提供服务。当作模块速查读。

5.1 工具:一条固定的执行流水线

定义一个工具需要三样东西

# python/beeai_framework/tools/tool.py:62-75 抽象成员
name: str            # 工具名
description: str     # 给模型看的说明
input_schema: type[BaseModel]   # pydantic 入参模型

加上一个 _run(input, options, context) 就完事了。

每次调用固定走这条流水线

Tool.run(tools/tool.py:117-190)不可改写,顺序是:

  入参(dict 或模型)
        ▼
  ① pydantic 校验 ── 失败 → ToolInputValidationError
        ▼
  ② 进 Retryable 包装
        ▼
  ③ 发 start 事件  ←── 中间件可在这里短路(第 4 章)
        ▼
  ④ 查缓存(命中直接返回)
        ▼
  ⑤ 真正执行 _run
        ▼
  ⑥ 写缓存 → 发 success → finally 发 finish

错误路径也定义得很细:on_error 先发 error 事件,若是致命错误就立刻上抛不再重试(tool.py:142-150);on_retry 发 retry 事件。缓存键由入参 + 选项算出,并刻意剔除 signal 和 retry_options(tool.py:102-106)——这两个不影响结果,不该参与键计算。

用装饰器写工具最省事

# 示意,非源码
@tool
def get_stock_price(ticker: str, currency: str = "USD") -> str:
    """Return the latest stock price for a ticker."""   # 这句 docstring 就是给模型看的 description
    return fetch(ticker, currency)

tool 装饰器(tools/tool.py:254-322)做三件事:名字取函数名、描述取 docstring(没有就报错)、入参 schema 从函数签名反推。

反推逻辑在 get_input_schema(tool.py:202-227):用 inspect.getfullargspec 拿到参数名、类型标注、默认值,喂给 pydantic 的 create_model。有 **kwargs 就设 extra="allow",否则 "ignore"。函数注释还挂了来源:inspired by pydantic issue #1391。

返回值不是 ToolOutput 时自动包成 StringToolOutput;with_context=True 时会把 RunContext 作为额外入参注入。

内置工具清单

类别	工具	位置
思考	`ThinkTool`	`tools/think.py`
搜索	DuckDuckGo、Wikipedia、xquik、retrieval	`tools/search/`
天气	`OpenMeteoTool`	`tools/weather/openmeteo.py`
代码	沙箱执行、本地 Python、Shell	`tools/code/`
文件	读、编辑、glob、grep	`tools/filesystem/`
协议	OpenAPI、MCP	`tools/openapi.py`、`tools/mcp/mcp.py`
多 agent	`HandoffTool`	`tools/handoff.py`

ThinkTool 值得单独说。 它对外界没有任何副作用,_run 只是把构造时传进来的 tool_output 原样回吐:

# python/beeai_framework/tools/think.py:28 与 :42-44
tool_output: str | Callable[[ThinkSchema], str] = "OK",     # 构造参数,默认就是这个字符串
...
async def _run(self, input, options, context) -> StringToolOutput:
    output: str = self._tool_output(input) if isinstance(self._tool_output, Callable) else self._tool_output
    return StringToolOutput(output)

也就是说默认情况下它回一句 "OK",但你可以传一个 (ThinkSchema) -> str 的函数,把模型写下的 thoughts 加工后再喂回去。

它存在的唯一意义是:给「思考」这个动作一个可被约束系统引用的把手。有了它,ConditionalRequirement(ThinkTool, force_at_step=1) 才能表达「第一步必须先想」;force_after=Tool 配 consecutive_allowed=False 就重建出了 ReAct 的「想—做—想—做」节奏(examples/agents/requirement/complex.py)。

源码里还留了一条负面经验:schema 原本有个 next_step 字段,被删了,注释写着 removed as it did not perform well(think.py:16-17)。

MCP 工具

MCPTool.from_client(tools/mcp/mcp.py:110)从一个 MCP 会话拉取工具列表,每个包成 MCPTool。包完之后它和本地工具毫无区别 —— 一样能被需求约束、被审批拦截、被缓存。

5.2 记忆:四种策略

所有实现都继承 BaseMemory(memory/base_memory.py:14),接口只有 messages / add / delete / reset 几个。

实现	策略	何时用
`UnconstrainedMemory`	全存,不淘汰	默认;单次运行的临时记忆
`SlidingMemory`	按条数滑窗	简单粗暴省 token
`TokenMemory`	按 token 预算淘汰	长对话
`SummarizeMemory`	老消息压成摘要	要保留远期信息
`ReadOnlyMemory`	只读视图	传给子 agent 时防写

`TokenMemory` 的淘汰逻辑

# python/beeai_framework/memory/token_memory.py:115-120
capacity = self._max_tokens * self._threshold          # 默认 128000 * 0.75
while self._messages and self.tokens_used > capacity - estimated_tokens:
    message_to_delete = self.handlers["removal_selector"](self._messages)   # 默认删最老的
    deleted = await self.delete(message_to_delete) if message_to_delete is not None else False
    if not deleted:
        raise ResourceFatalError('The "removal_selector" handler must return a valid message!')

removal_selector 返回 None 时不会去 delete,而是让 deleted 保持 False,直接落到下面那句报错 —— 自定义淘汰策略选不出消息,会被当成配置错误而不是静默死循环。

三个可替换的 handler:tokenize(精确计数)、estimate(快速估算,默认 len/4)、removal_selector(淘汰谁)。

两个巧思:

懒同步。 新消息先用估算值,标记 dirty;脏消息比例超过 sync_threshold(默认 25%)才批量精确计数(token_memory.py:131-133)。
按对象身份做键。 _get_message_key 返回 str(id(message)),注释解释得很清楚:如果按「角色+文本」做键,两条一模一样的 "yes" 会互相覆盖(token_memory.py:56-64)。

5.3 `Workflow`:显式状态机

它是什么

一个泛型状态机:状态是你给的 pydantic 模型,每一步是 (state) -> 下一步名字。

# 示意,非源码
class State(BaseModel):
    query: str
    draft: str = ""

workflow = Workflow(schema=State)
workflow.add_step("research", research_step)   # 返回 None → 顺序走到下一步
workflow.add_step("write", write_step)         # 返回 "research" → 跳回去重查

五个保留字

返回值	跳到哪
`None` 或 `Workflow.NEXT`	声明顺序的下一步
`Workflow.SELF`	自己(循环)
`Workflow.PREV`	上一步
`Workflow.START`	本次运行的第一步
`Workflow.END`	结束
其它字符串	那个名字的步骤

路由逻辑在 workflow.py:143-155,_find_step(:190-196)靠步骤的声明顺序算 prev/next —— 所以 add_step 的顺序有语义。

每步深拷贝状态

# python/beeai_framework/workflows/workflow.py:134-140 摘要
step_res = WorkflowStepRes(name=next, state=run.state.model_copy(deep=True))
run.steps.append(step_res)
step_next = await ensure_async(step.handler)(step_res.state)
check_model(step_res.state)          # 步骤改完状态后重新校验
run.state = step_res.state

每一步操作的是深拷贝的副本,并全部留在 run.steps 里 —— 跑完能拿到完整的状态演化史。代价是每步一次深拷贝。

`AgentWorkflow`:把 agent 串成流水线

AgentWorkflow(workflows/agent/agent.py:59)在 Workflow 上加一层语义:每个步骤是一个 agent,状态里存着待处理输入队列和累积消息。

两个细节:

每步克隆 agent 并换掉记忆(agent.py:124-128),再传只读记忆视图进去 —— 子 agent 看得到历史但改不了。
只把最后两条消息(工具调用 + 结果)传给下一个 agent(agent.py:171-176),避免上下文爆炸。

5.4 `HandoffTool`:把交接做成工具

多 agent 的另一种玩法:不预定义流水线,而是让主 agent 自己决定把任务交给谁。做法是把目标 agent 包成一个工具:

# python/beeai_framework/tools/handoff.py:26 及构造
HandoffTool(target=expert_agent)   # name/description 默认取 target.meta

_run(handoff.py:69-98)干四件事:

从 context.context["state"]["memory"] 取出主 agent 的记忆 —— 这个上下文是 Runner 调工具时塞进去的(_runner.py:208)。
克隆目标 agent(可克隆的话),避免并发污染。
裁掉消息尾巴:反向找到最后一条「不是带工具调用的助手消息」,只传到那为止(handoff.py:81-86)。理由很实际——把一条悬空的工具调用消息传给子 agent,大多数 provider 会直接报错。
可选地把 task 作为新的用户消息追加,跑目标 agent,返回文本。

因为它是个普通工具,所以第 2 章的所有约束都能用在交接上:限制交接次数、规定交接顺序、交接前必须审批。

5.5 `Serve`:一个类型工厂注册表

抽象

Server(serve/server.py:19)只有一个核心机制:类级别的「成员类型 → 工厂函数」注册表。

# python/beeai_framework/serve/server.py:74-80
@classmethod
def _get_factory(cls, input: TInput) -> Callable[[TInput], TInternal]:
    for obj_type in type(input).__mro__:            # 沿继承链找
        if factory := cls._factories.get(obj_type):
            return factory
    raise ValueError(f"No factory registered for {type(input)}.")

沿 MRO 查找意味着:给 BaseAgent 注册一个工厂,你自定义的子类 agent 就自动被支持。

__init_subclass__(server.py:29-38)保证每个 Server 子类拥有自己的注册表,不会共享父类的 dict —— 这是 Python 类属性继承的经典坑,这里显式规避了。

注册在 register() 时就查一次工厂(server.py:56-62),不支持的类型在注册那一刻就报错,而不是等到起服务。

支持的协议

协议	适配器目录
A2A(Agent2Agent)	`adapters/a2a/serve/`
MCP	`adapters/mcp/serve/`
ACP / ACP-Zed	`adapters/acp/serve/`、`adapters/acp_zed/serve/`
OpenAI 兼容(Chat Completions + Responses)	`adapters/openai/serve/`
AgentStack	`adapters/agentstack/serve/`
watsonx Orchestrate	`adapters/watsonx_orchestrate/serve/`

可选依赖是硬性的:导入 A2A 服务时缺包会给出明确指引(adapters/a2a/serve/server.py:40-42):Run 'pip install "beeai-framework[a2a]"'。A2A 服务支持 jsonrpc / grpc / http_json 三种传输,默认 jsonrpc(adapters/a2a/serve/server.py:63)。

会话记忆

服务端是多会话的,记忆不能只有一份。MemoryManager 协议(serve/utils.py:15-20)给了两个实现:

实现	行为
`UnlimitedMemoryManager`	全存,不淘汰(默认)
`LRUMemoryManager(maxsize)`	LRU 淘汰,可自定义大小计算

5.6 提示模板

PromptTemplate(template.py:26)= Mustache(chevron)+ pydantic schema。

三个能力:

方法	干什么
`render(input)`	校验入参 → 填默认值 → 执行 `functions` → 渲染
`update(...)`	原地改模板 / 默认值(`template.py:121-122` 直接 `dict.update`)
`fork(customizer)`	拷一份再改,不影响原模板

functions 是「渲染期计算的变量」,系统提示里的当前日期就是这么来的:functions={"formatDate": lambda data: datetime.now(tz=UTC).strftime("%Y-%m-%d")}(agents/requirement/prompts.py:44)。函数名和入参字段重名会直接报错(template.py:94-95)。

`fork` 的副本靠构造函数保证,不靠 `fork` 自己

RequirementAgent 的四个模板字段都用 fork(None) 当默认值(types.py:35-46,例如 default_factory=lambda: RequirementAgentSystemPrompt.fork(None)),而 RequirementAgentSystemPrompt 是 prompts.py:42 上的一个进程级模块变量。只看 fork 会以为副本没做出来:

# python/beeai_framework/template.py:103-107 与 :71-81 节选
new_config = customizer(self._config) if customizer else self._config   # 无 customizer → 就是原 config 本身
return PromptTemplate(new_config)
# —— 副本在 __init__ 的最后一行:
self._config = (config if config is not None else PromptTemplateInput(...)).model_copy(deep=True)

customizer=None 时 new_config 确实就是原模板的那个 PromptTemplateInput 实例,但 PromptTemplate.__init__ 结尾那句 .model_copy(deep=True)(template.py:81)把它连同 defaults 字典一起深拷贝了。所以每个 agent 实例拿到的是独立副本。

这条链是承重的,因为 RequirementAgent.__init__ 紧接着就会原地改默认值:

# python/beeai_framework/agents/requirement/agent.py:140-151 节选
if role or instructions or notes:
    self._templates.system.update(defaults=exclude_none({"role": role, ...}))

update 走的是 self._config.defaults.update(...)(template.py:122),是原地写。要不是 __init__ 那句深拷贝,这里就会写穿到进程级的 RequirementAgentSystemPrompt —— 一个 agent 传了 role,同进程里所有后建的 agent 都跟着变。

读这段代码的教训:fork 的语义不在 fork 函数体里,而在它调用的构造函数最后一行。 只读 fork(...) 的四行会得出相反结论。

5.7 代码地图

主题	文件路径	符号名
工具基类与流水线	`python/beeai_framework/tools/tool.py`	`Tool`、`Tool.run`
函数转工具	`python/beeai_framework/tools/tool.py`	`tool`、`get_input_schema`
思考工具	`python/beeai_framework/tools/think.py`	`ThinkTool`、`ThinkSchema`
交接工具	`python/beeai_framework/tools/handoff.py`	`HandoffTool`
MCP 工具接入	`python/beeai_framework/tools/mcp/mcp.py`	`MCPTool.from_client`
记忆接口	`python/beeai_framework/memory/base_memory.py`	`BaseMemory`
token 预算记忆	`python/beeai_framework/memory/token_memory.py`	`TokenMemory.add`、`sync`
状态机	`python/beeai_framework/workflows/workflow.py`	`Workflow`、`_find_step`
多 agent 流水线	`python/beeai_framework/workflows/agent/agent.py`	`AgentWorkflow.add_agent`
服务注册表	`python/beeai_framework/serve/server.py`	`Server.register_factory`、`_get_factory`
会话记忆管理	`python/beeai_framework/serve/utils.py`	`MemoryManager`、`LRUMemoryManager`
A2A 服务	`python/beeai_framework/adapters/a2a/serve/server.py`	`A2AServer`、`A2AServerConfig`
提示模板	`python/beeai_framework/template.py`	`PromptTemplate`、`fork`、`update`、`PromptTemplateInput`
模板默认副本	`python/beeai_framework/agents/requirement/types.py`	`RequirementAgentTemplates`

5.1 工具:一条固定的执行流水线​

定义一个工具需要三样东西​

每次调用固定走这条流水线​

用装饰器写工具最省事​

内置工具清单​

MCP 工具​

5.2 记忆:四种策略​

TokenMemory 的淘汰逻辑​

5.3 Workflow:显式状态机​

它是什么​

五个保留字​

每步深拷贝状态​

AgentWorkflow:把 agent 串成流水线​

5.4 HandoffTool:把交接做成工具​

5.5 Serve:一个类型工厂注册表​

抽象​

支持的协议​

会话记忆​

5.6 提示模板​

fork 的副本靠构造函数保证,不靠 fork 自己​

5.7 代码地图​