数据截至 (上游 commit 460c729002dc)
第 6 章 · 巧妙之处、边界与横向对比
本章讲什么: 前五章拆完了机制。这一章做三件事:挑出可以直接搬进你自己项目的设计;诚实列出这个框架的边界和我在源码里查证到的缺陷;最后和兄弟框架比一比取舍。
6.1 巧妙之处(可借鉴的技术)
一、把「结束」做成一次工具调用
妙在哪: 用同一套机制解决了三个问题——循环终止条件、结构化输出、以及「结束」本身也能被约束(可以规定「没查过天气不准交卷」)。
FinalAnswerTool._run 直接写共享 state 的 answer 字段,掐断主循环条件(python/beeai_framework/agents/requirement/utils/_tool.py:52-59)。它的入参 schema 就是你的输出格式:传 Pydantic 模型进 expected_output,拿回来的就是校验过的对象(_tool.py:36-50)。
二、不允许的工具照样出现在提示里,还附上理由
妙在哪: 大多数框架的做法是「不给的工具就不列」。BeeAI 反过来:列出来,标 Allowed: False,再写一句 Reason(utils/_llm.py:176-184 + prompts.py:71-79)。这等于持续告诉模型「你现在该干别的,而且我告诉你为什么」,比让模型对着缩水的工具表瞎猜有效。
真要藏,另有 hidden=True 这一档。两档语义分开,是这个设计的关键。
三、能力探测 + 结构化输出降级
妙在哪: 「强制模型调某个工具」这件事在 API 层面各家支持度天差地别。BeeAI 用一个 tool_choice_support 集合声明能力,撑不住时把工具表编译成 JSON schema 联合类型,用结构化输出把工具调用伪造出来(backend/chat.py:797-820 + backend/utils.py:125-201)。
上层完全不知情——伪造出来的和原生的都是同一个 MessageToolCallContent。
四、fallback_tool:给漏字段的模型补壳
妙在哪: 重写生成 schema 的 model_validate,模型只吐 {"response": "..."} 时自动补上 {"name": "final_answer", "parameters": {...}}(backend/utils.py:169-188)。
更妙的是它是条件性的:final_answer 只在这一轮允许收尾时才被当作兜底目标(_runner.py:150 的 fallback_tool=request.final_answer if request.can_stop else None)。容错不能损害约束。
这里有一处容易读漏的补丁,值得说清楚:
# python/beeai_framework/backend/chat.py:459-461
fallback_tool = options.get("fallback_tool")
if fallback_tool is None and isinstance(tool_choice, Tool):
fallback_tool = tool_choice
也就是说 fallback_tool 不总是 None。prevent_stop 恰好会把 final_answer 摘出允许集(utils/_llm.py:144-146),只剩一个工具时 tool_choice 就被设成那个 Tool 实例(:156-158)——于是兜底目标变成那个被强制的工具。
| 这一轮的状态 | fallback_tool 实际是谁 |
|---|---|
| 允许收尾 | final_answer |
禁止收尾,且 tool_choice 是某个具体工具 | 那个工具(chat.py:460-461 自动填上) |
禁止收尾,且 tool_choice 还是 "required" | None,格式错误就老实报错 |
核心语义没变:禁止收尾时绝不会兜底到 final_answer。补壳只在「已经被锁死的那一个工具」上生效,约束没被削弱。
五、start 事件可写回:一个钩子换来短路 + 改写输入
妙在哪: RunContext.enter 发 start 事件后读回 start_event.input 和 start_event.output(context.py:209-220)。填了 output 就跳过执行,改了 input 就用新的。
因为 agent / 工具 / 模型调用共用这一个入口,一个约定同时给了整个框架审批、缓存、mock、输入重写四种能力。人工审批需求就是这么实现的(ask_permission.py:82-90)。
六、临时消息:自愈引导用完即删
妙在哪: 模型产出坏工具调用时,框架追加一条「你写错了,可用工具是 X/Y/Z」再重试(backend/chat.py:555-572),消息打上 tempMessage 标记,每轮末尾统一清掉(_runner.py:322)。
纠错痕迹不进历史,下一轮的上下文是干净的。
七、双深度的缩进日志
妙在哪: GlobalTrajectoryMiddleware 给每个 run_id 记两个深度:absolute(真实层级)和 relative(只数被过滤器放行的层)(middleware/trajectory.py:34-38、:144-159)。所以 included=[Tool] 时缩进依然连续,不会因为跳过中间层出现断层。
八、报错信息直接给可复制的修法
妙在哪: tool_choice 校验失败时,错误日志列出三种修法的完整代码(backend/chat.py:940-949)。框架承认自己的 provider 能力表可能过时,并把纠正手段交到用户手上。
6.2 边界与局限
A. 一个可查证的缺陷:backstory 在 RequirementAgent 里被丢弃
RequirementAgent.run 支持 backstory 参数,_process_input 本应把它连同 expected_output 一起渲染进用户消息。但那个分支的条件永远为假:
# python/beeai_framework/agents/requirement/agent.py:220-221
*msgs, last_message = [UserMessage(input)] if isinstance(input, str) else input
if last_message is None and isinstance(last_message, UserMessage) and last_message.text:
last_message is None 和 isinstance(last_message, UserMessage) 不可能同时成立。所以 self._templates.task.render(...)(:222-231)在本次提交下是死代码,函数总是走 else 分支原样返回消息(:236-238)。
实际影响:
| 参数 | 是否仍然生效 | 为什么 |
|---|---|---|
expected_output(Pydantic 模型) | ✅ 生效 | 走 FinalAnswerTool.input_schema 这条独立路径 |
expected_output(字符串) | ✅ 生效 | 走 FinalAnswerTool.instructions,渲染进系统提示 |
backstory | ❌ 静默丢弃 | 只有 task 模板这一条路,而那条路进不去 |
RequirementAgentTaskPrompt 这个模板(prompts.py:104-116)因此在 RequirementAgent 上完全没被使用。以上是对 commit 460c729 的源码判读;修复与否请以上游后续提交为准。
B. 优先级不能覆盖禁令
规则折叠是布尔 OR:任何一条 allowed=False 都是终局,priority 只决定「多个工具同时被强制时选谁」(utils/_llm.py:112-138,完整算法见 第 2 章 2.4 节)。
这里要指出的边界是命名与预期不符: priority 这个词很容易让人以为是通用的冲突消解,但它压不住禁令。想做「特殊情况下放行」,只能让那条禁令自己在 run 里判断后不吐规则。
C. Python 与 TypeScript 不对等
| 项 | Python | TypeScript |
|---|---|---|
| 版本号 | 0.1.82 | 0.1.30 |
serializer 模块 | 没有 | 有(typescript/src/serializer) |
parsers / middleware / serve | 有 | 有 |
| RequirementAgent | 有 | 有(typescript/src/agents/requirement) |
README 的功能表里列了 Serialization,但 Python 包里没有 serializer 目录。要在 Python 侧做 agent 状态落盘,得自己实现(Requirement.to_json_safe、Tool.to_json_safe、BaseMemory.to_json_safe 提供了一些原料,但没有统一的序列化框架)。
D. 其它需要留意的地方
_run用递归而非迭代重试。 死循环检测、格式纠错、强制交卷都是return await self._run(...)自调用(_runner.py:296、:303、:317)。有重试计数器兜底,但栈深度随重试次数增长。- RequirementAgent 只能通过
final_answer退出。run末尾是assert final_state.answer is not None(agent.py:207)。所有异常路径最终要么交卷要么抛错,没有「中途返回部分结果」这个档位。 - 需求实例带跨运行状态。
Requirement.state和子类的私有字段在多次run()之间共享。remember_choices靠这个工作,但自定义需求里存计数器要自己复位。 - 运行时底座的三条坑照单全收:
Emitter.root()是进程级单例、同一个中间件实例不能并发用于两次 run、事件回调抛异常会打断主流程。多租户或并发部署前请逐条对照 第 4 章 4.9 节。 - 不做 token 级预算控制。
TokenMemory管的是记忆容量,框架不会在发请求前估算总 token 并主动裁剪。 - 可选依赖很多。 A2A / MCP / ACP / transformers / langchain 都是 extras,缺包时报错清楚但确实要装(
adapters/a2a/serve/server.py:40-42)。
6.3 横向对比
定位差异
同在 agent-frameworks 这一格,几个项目回答的其实不是同一个问题:
| 项目 | 核心抽象 | 它认为「难点」在哪 | 子库文档 |
|---|---|---|---|
| BeeAI Framework | Requirement / Rule | 模型不听话 —— 用声明式约束把自由度夹住 | 本文档 |
| Apache Burr | State + Action + 转移边 | 流程不可见、不可续跑 —— 显式状态机 + 持久化 | ../burr/index.md |
| PocketFlow | Node + Flow(极小内核) | 框架太重 —— 百行内核,一切自己拼 | ../pocketflow/index.md |
| Strands Agents | Agent + Tool(模型驱动循环) | 上手成本 —— 让模型自己规划,框架尽量不挡路 | ../strands-agents/index.md |
| Atomic Agents | schema 化的原子组件 | 可组合性 —— 输入输出都是显式 schema | ../atomic-agents/index.md |
「怎么控制 agent 的行为」这一维的对比
| 手段 | 谁这么做 | 强度 | 代价 |
|---|---|---|---|
| 提示词里写规则 | 几乎所有框架的默认 | 弱(模型可以不听) | 零成本 |
| 约束 API 参数(tools + tool_choice) | BeeAI | 强(API 层面锁死) | 需要处理 provider 能力差异 |
| 画死状态机 | Burr、Workflow 类框架 | 最强(根本没有别的路) | 失去模型的灵活规划 |
| 完全交给模型 | Strands 等 | 无 | 依赖强模型 |
BeeAI 站在中间那一档:模型仍然在每一步自主决策,但决策空间由框架每轮重新划定。这是它区别于兄弟项目最本质的一点,也是「同一份代码在 GPT-4 和本地 7B 上都能跑」这个卖点的技术来源。
什么时候选它
| 场景 | 建议 |
|---|---|
| 要在弱模型 / 本地模型上跑多步流程 | 强烈推荐,这是它的主场 |
| 需要「工具调用顺序 / 次数」的硬保证 | 推荐,Requirement 就是为此设计 |
| 要对外暴露成 A2A / MCP 服务 | 推荐,Serve 模块覆盖面广 |
| 流程完全固定、没有模型决策 | 用 Burr 这类状态机更直接 |
| 需要 agent 状态 持久化 / 断点续跑(Python) | 谨慎,Python 侧没有序列化模块 |
| 想要极小依赖 | 谨慎,LiteLLM + pydantic + 一堆 extras |
6.4 总代码地图
按「想搞清楚什么」组织,可直接 grep 符号名定位。
主线
| 想搞清楚 | 文件路径 | 符号名 |
|---|---|---|
| agent 怎么配置的 | python/beeai_framework/agents/requirement/agent.py | RequirementAgent.__init__ |
| 循环怎么转的 | python/beeai_framework/agents/requirement/_runner.py | RequirementAgentRunner.run |
| 单轮做了什么 | python/beeai_framework/agents/requirement/_runner.py | RequirementAgentRunner._run |
| 循环怎么停的 | python/beeai_framework/agents/requirement/utils/_tool.py | FinalAnswerTool._run |
| 模型不调工具怎么办 | python/beeai_framework/agents/requirement/_runner.py | _create_final_answer_tool_call |
约束系统
| 想搞清楚 | 文件路径 | 符号名 |
|---|---|---|
| 约束的最小单位 | python/beeai_framework/agents/requirement/requirements/requirement.py | Rule |
| 怎么写自定义约束 | python/beeai_framework/agents/requirement/requirements/requirement.py | requirement、run_with_context |
| 内置的时序约束 | python/beeai_framework/agents/requirement/requirements/conditional.py | ConditionalRequirement.run |
| 规则怎么合并 | python/beeai_framework/agents/requirement/utils/_llm.py | RequirementsReasoner.create_request |
| 人工审批 | python/beeai_framework/agents/requirement/requirements/ask_permission.py | AskPermissionRequirement |
模型层
| 想搞清楚 | 文件路径 | 符号名 |
|---|---|---|
| 统一 LLM 抽象 | python/beeai_framework/backend/chat.py | ChatModel |
| 能力降级判定 | python/beeai_framework/backend/chat.py | _force_tool_call_via_response_format |
入参归一与 fallback_tool 回填 | python/beeai_framework/backend/chat.py | _prepare_model_input |
| 伪造工具调用的 schema | python/beeai_framework/backend/utils.py | generate_tool_union_schema |
| 响应合法性校验 | python/beeai_framework/backend/chat.py | _assert_tool_response |
| provider 适配 | python/beeai_framework/adapters/litellm/chat.py | LiteLLMChatModel |
运行时
| 想搞清楚 | 文件路径 | 符号名 |
|---|---|---|
run() 返回的是什么 | python/beeai_framework/context.py | Run |
| 执行上下文树 | python/beeai_framework/context.py | RunContext.enter |
| 中间件怎么短路 | python/beeai_framework/context.py | RunContextStartEvent |
| 事件总线 | python/beeai_framework/emitter/emitter.py | Emitter |
| 轨迹日志 | python/beeai_framework/middleware/trajectory.py | GlobalTrajectoryMiddleware |
| 流式最终答案 | python/beeai_framework/middleware/stream_tool_call.py | StreamToolCallMiddleware |
周边
| 想搞清楚 | 文件路径 | 符号名 |
|---|---|---|
| 工具执行流水线 | python/beeai_framework/tools/tool.py | Tool.run、tool |
| 记忆策略 | python/beeai_framework/memory/token_memory.py | TokenMemory.add |
| 状态机 | python/beeai_framework/workflows/workflow.py | Workflow |
| 多 agent 交接 | python/beeai_framework/tools/handoff.py | HandoffTool._run |
| 对外托管 | python/beeai_framework/serve/server.py | Server._get_factory |
可运行的参考示例
| 场景 | 路径 |
|---|---|
| 最小可跑的约束 agent | python/examples/agents/requirement/quickstart_requirement.py |
| 全参数 + 自定义需求 | python/examples/agents/requirement/complex.py |
| 自定义需求单独示例 | python/examples/agents/requirement/custom_requirement.py |
| 结构化输出 | python/examples/agents/requirement/structured_output.py |
| 多 agent / 交接 | python/examples/agents/requirement/multi_agent.py、handoff.py |
| 折叠算法单测 | python/tests/agents/test_requirements_reasoner.py |