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

第 1 章 · RequirementAgent 主线

本章讲什么: 把一次 agent.run("...") 从入参拆到最终答案,逐层看清楚谁在推进循环、循环凭什么停下、中途出岔子怎么救。约束机制本身(需求怎么变成通行证)留到第 2 章。

---

1.1 先看清一件事:门面和跑腿是两个对象

RequirementAgent 本身不跑循环。它是个配置容器:持有 LLM、工具表、需求表、提示模板,以及一份跨多次对话存活的持久记忆。

真正跑循环的是 RequirementAgentRunner,它在每次 run() 里新建:

# python/beeai_framework/agents/requirement/agent.py:175-189 节选
runner = self.runner_cls(
    llm=self._llm,
    config=AgentExecutionConfig(...),
    tools=self._tools,
    requirements=self._requirements,
    tool_call_cycle_checker=self._create_tool_call_checker(),
    run_context=RunContext.get(),
    ...
)

这行 self.runner_cls 是个可替换的类属性(agent.py:155),你可以继承 RequirementAgentRunner 换掉整个循环体而不动门面。

双层记忆是理解这个设计的钥匙

层	谁持有	生命周期	作用
持久记忆	RequirementAgent._memory	跨多次 run()	多轮对话的历史
本轮记忆	RequirementAgentRunState.memory	单次 run()	装本次的推理草稿:助手消息、工具调用、工具结果、临时消息

本轮记忆在 Runner 构造时是一个全新的 UnconstrainedMemory(_runner.py:67-69),然后把持久记忆和新消息倒进去:

# python/beeai_framework/agents/requirement/agent.py:195-196
await runner.add_messages(self.memory.messages)
await runner.add_messages(new_messages)

这样做的好处很直接:本次运行可以随便往记忆里塞临时消息、再删掉,不会污染真正的对话历史。

---

1.2 循环的骨架

RequirementAgentRunner.run 只有二十来行(_runner.py:250-273),骨架极其干净:

# python/beeai_framework/agents/requirement/_runner.py:257-272 节选
await self._reasoner.update(self._requirements)      # 初始化需求(校验目标工具存在)

while self._state.answer is None:                    # 唯一的终止条件
    self._increment_iteration()                      # 超过 max_iterations 就抛 AgentError
    request = await self._create_request()           # 领通行证
    await self._ctx.emitter.emit("start", ...)
    self._iteration_error_counter.reset()
    response = await self._run(request)              # 跑一轮
    await self._ctx.emitter.emit("success", ...)

终止条件只有一个:state.answer 被填上。 谁填?下一节。

每一轮内部做四件事

单轮逻辑在 _run(_runner.py:275-324),顺序如下:

  ┌─────────────────────────────────────────────┐
  │ ① 调模型                                     │
  │    _run_llm(request)                         │
  └───────────────┬─────────────────────────────┘
                  │ 没产出工具调用?
                  ▼
  ┌─────────────────────────────────────────────┐
  │ ② 兜底:把纯文本硬转成 final_answer 调用      │
  │    _create_final_answer_tool_call            │
  └───────────────┬─────────────────────────────┘
                  ▼
  ┌─────────────────────────────────────────────┐
  │ ③ 死循环检测                                 │
  │    ToolCallChecker.register → 命中就重签证    │
  └───────────────┬─────────────────────────────┘
                  ▼
  ┌─────────────────────────────────────────────┐
  │ ④ 并发跑工具 → 结果写回本轮记忆 → 清临时消息  │
  │    _invoke_tool_calls                        │
  └─────────────────────────────────────────────┘

---

1.3 机制一:final_answer 是个假工具

它要解决的小问题

循环怎么知道该停?常见做法是「模型这一轮没调工具 = 结束」。这个判据在弱模型上很脆:模型可能只是忘了调工具,或者中途吐了一段闲聊。

思路

BeeAI 的做法是把「结束」也做成一次工具调用。框架偷偷往工具表里塞一个 final_answer 工具(utils/_llm.py:37:self._tools = [*tools, final_answer]),模型必须显式调用它才算交卷。

这个工具的 _run 不干别的,只把答案写进共享的 state:

# python/beeai_framework/agents/requirement/utils/_tool.py:52-59
async def _run(self, input, options, context) -> StringToolOutput:
    self._state.result = input
    if self.input_schema is self._expected_output:
        self._state.answer = AssistantMessage(input.model_dump_json())
    else:
        self._state.answer = AssistantMessage(input.response)
    return StringToolOutput("Message has been sent")

它持有的 self._state 就是 Runner 那个 RequirementAgentRunState 实例(_runner.py:73 传进去的)。工具执行的副作用直接掐断了 while 循环的条件——这是整个设计里最巧的一处耦合。

顺带解决了结构化输出

因为「结束」是一次工具调用,它的入参 schema 就天然是「最终答案的格式」:

你传的 expected_output	final_answer 的入参 schema
None	FinalAnswerToolSchema(单字段 response: str)
一个字符串	同上,但把你的字符串塞进 response 字段的 description
一个 Pydantic 模型类	直接用你的模型当 schema,答案就是校验过的结构化对象

逻辑在 utils/_tool.py:36-50。所以「让 agent 返回结构化结果」这件事在 BeeAI 里不需要额外的解析层——它复用了模型自己的工具入参校验。

---

1.4 机制二:模型不调工具时的三级兜底

问题

弱模型经常这样:你要求它调 final_answer,它直接吐一段普通文本。这轮就白跑了。

三级降级

_run 里这段(_runner.py:280-303)按顺序试:

模型没产出工具调用
   │
   ├─ 允许结束(can_stop)且有文本?
   │     └→ ① 从文本里抠出第一对 {...},当作 final_answer 的入参
   │        抠不到、且没有自定义 schema → 直接把整段文本塞进 response 字段
   │        成功 → 手工造一条 AssistantMessage(tool-call),当作模型调过了
   │
   │  ① 没兜住 → 先记一次错误(两个重试计数器各消耗一次)
   │
   ├─ 不允许结束(某个需求 prevent_stop)?
   │     └→ ② 原样重来一轮(递归 self._run(request))
   │
   └─ 允许结束但连文本都榨不出?
         └→ ③ 清空全部需求 + 兜底允许 final_answer + 把 tool_choice 钉在 required

第一级的实现值得看一眼:

# python/beeai_framework/agents/requirement/_runner.py:177-180
json_object_pair = find_first_pair(full_text, ("{", "}"))
final_answer_input = parse_broken_json(json_object_pair.outer) if json_object_pair else None
if not final_answer_input and not self._reasoner.final_answer.custom_schema:
    final_answer_input = FinalAnswerToolSchema(response=full_text).model_dump()

parse_broken_json 背后是 json_repair 库(backend/utils.py:105-111),能救回缺引号、多逗号这类残缺 JSON。

错误计数发生在分岔之前。 ② 和 ③ 都要先走这两行(_runner.py:291-293):

# python/beeai_framework/agents/requirement/_runner.py:291-293
err = AgentError("Model produced an invalid final answer tool call.")
self._iteration_error_counter.use(err)
self._global_error_counter.use(err)

也就是说:只要第一级没兜住,不管接下来走哪条路都会记账,计数器耗尽即抛 AgentError,递归不会无限深下去。

第三级到底做了什么:弃约束,不是「只放行 final_answer」

三步,顺序有讲究(_runner.py:298-303):

# python/beeai_framework/agents/requirement/_runner.py:298-303
await self._reasoner.update(requirements=[])
updated_request = await self._create_request(
    extra_rules=[Rule(target=self._reasoner.final_answer.name, allowed=True, hidden=False)],
)
self._force_final_answer_as_tool = True
return await self._run(updated_request)

对照第 2 章的折叠算法逐条看效果:

这一步	实际效果
update(requirements=[])	需求表清空(utils/_llm.py:42-43 的 self._entries.clear()),于是 rules_by_tool 每一项都是空列表,每个工具默认 is_allowed=True —— 全部工具重新回到允许集,而不是只剩 final_answer
extra_rules=[Rule(final_answer, allowed=True)]	显式给交卷通道兜个底,保证它一定在允许集里
_force_final_answer_as_tool = True	它就是 create_request 的 force_tool_call 入参(_runner.py:196)。置 True 后,tool_choice 不会再退回 "auto",而是钉在 "required";允许集只剩一个工具时直接钉成那个工具实例

净效果一句话:工具全放开,但这一轮必须调某个工具。这是一个有意识的权衡——宁可拿到一个不完美的答案,也不要卡死。

两个容易看漏的细节:

• _force_final_answer_as_tool = True 写在 _create_request 之后。 所以 updated_request 本身用的还是旧值(默认来自 RequirementAgent(final_answer_as_tool=True),agent.py:70/187);这个赋值真正管的是从下一次签证开始的所有轮次。
• 没有需求了,prevent_stop 自然也没了,can_stop 恢复为 True,第一级兜底在下一轮重新可用。

---

1.5 机制三:死循环检测

问题

模型用同样的参数反复调同一个工具,是 agent 最常见的死法。

检测器

ToolCallChecker(agents/tool_calling/utils.py:16-45)维护两个计数器:

计数器	默认阈值	抓什么
_strike_counter	max_strike_length=1	连续重复:同样的 (工具名, 参数) 连着出现超过 1 次
_occurrences_counter	max_total_occurrences=5,window_size=10	窗口内重复:最近 10 次里同一调用出现超过 5 次

相等判据是「工具名 + 参数字符串 + 类型」三者全同(utils.py:48-49 的 _is_same_tool_call)。底层是 OccurrencesCounter(utils/counter.py),它给每个条目记一个 distance,超过窗口 n 就淘汰——用很小的代码实现了滑动窗口去重计数。

抓到之后怎么办

不是抛错,而是临时禁掉那个工具再重来一轮:

# python/beeai_framework/agents/requirement/_runner.py:311-317
self._tool_call_cycle_checker.register(tool_call_msg)
if self._tool_call_cycle_checker.cycle_found:
    self._tool_call_cycle_checker.reset()
    updated_request = await self._create_request(
        extra_rules=[Rule(target=tool_call_msg.tool_name, allowed=False, hidden=False, forced=True)],
    )
    return await self._run(updated_request)

注意 extra_rules 这个口子——它让运行时的临时判断能和用户声明的静态需求走同一条聚合管道(第 2 章会讲 extra_rules 的优先级是怎么算的)。

---

1.6 机制四:工具执行与错误转述

所有工具调用是并发执行的(agents/_utils.py:50-56,asyncio.gather),然后按顺序整理结果。

关键在于错误不会往上抛,而是被翻译成给模型看的话:

# python/beeai_framework/agents/requirement/_runner.py:221-230 节选
if tool_call.error is not None:
    result = self._templates.tool_error.render(
        RequirementAgentToolErrorPromptInput(reason=tool_call.error.explain()))
else:
    result = (tool_call.output.get_text_content()
              if not tool_call.output.is_empty()
              else self._templates.tool_no_result.render(tool_call=tool_call))

对应的两个模板都很短(prompts.py:123-138),大意是「这个工具失败了,换个工具或说明为什么用不了」「没查到结果,换个查询词」。

每次调用还会记一条 step:

# python/beeai_framework/agents/requirement/_runner.py:211-218 结构
RequirementAgentRunStateStep(id=..., iteration=..., input=..., output=..., tool=..., error=...)

state.steps 是需求判断的唯一事实来源——ConditionalRequirement 数「这个工具调过几次」「上一步是哪个工具」全靠它。

工具出错也会消耗同样那两个重试计数器(_runner.py:241-243):_iteration_error_counter(每轮开头重置)和 _global_error_counter(整次运行累计)。RetryCounter.use 在耗尽时直接抛出一个把原始错误挂成 cause 的 AgentError(utils/counter.py)。

---

1.7 收尾:落账策略

循环结束后,本轮记忆怎么并回持久记忆,由 save_intermediate_steps 决定:

# python/beeai_framework/agents/requirement/agent.py:200-205
if self._save_intermediate_steps:
    self.memory.reset()
    await self.memory.add_many(final_state.memory.messages)   # 全量搬,含所有工具调用
else:
    await self.memory.add_many(new_messages)
    await self.memory.add_many(extract_last_tool_call_pair(final_state.memory) or [])

两种策略的取舍:

模式	保留什么	代价
True(默认)	全部中间步骤	下一轮对话 token 消耗大,但模型能复用已有的工具结果
False	只保留新消息 + 最后一对(工具调用, 工具结果)	省 token,但下一轮要重新查

extract_last_tool_call_pair(memory/utils.py:10-36)的做法很细致:先反向找到最后一条带工具调用的助手消息,取它最后一个 tool_call_id,再反向找到引用这个 id 的 ToolMessage——保证配对不错位(很多 provider 会因为孤立的 tool 消息直接报错)。

临时消息机制

每轮末尾还有一句清理:

# python/beeai_framework/agents/requirement/_runner.py:322
await delete_messages_by_meta_key(self._state.memory, key=TEMP_MESSAGE_META_KEY, value=True)

TEMP_MESSAGE_META_KEY = "tempMessage"(memory/utils.py:51)。谁会打上这个标记?ChatModel 的重试逻辑——模型产出坏工具调用时,框架会临时追加一条「你刚才写错了,可用工具有 X/Y/Z」的用户消息来引导重试(backend/chat.py:555-572)。这些引导消息用完即删,不留进记忆。

---

1.8 其它 agent:一张对比表

同一个仓库里还有几个 agent,理解它们的差异有助于看清 RequirementAgent 的定位。

Agent	交互协议	约束能力	适合
RequirementAgent	原生工具调用(或降级伪造)	声明式需求	生产,尤其跨强弱模型
ToolCallingAgent	原生工具调用	只有循环检测	简单工具循环
ReActAgent	文本协议 Thought/Tool Name/Tool Input/Final Answer	靠解析器纠错	完全不支持工具调用的老模型
LiteAgent	原生工具调用,不注入任何系统提示	无	裸测模型自身能力,不被框架提示词污染
RAGAgent	检索 + 生成	无	向量库问答

两个值得注意的实现细节:

• ReActAgent 会按模型名挑解析器:模型 id 里含 granite 就用 GraniteRunner,否则 DefaultRunner(agents/react/agent.py:72-77)。文本协议时代遗留的模型特判。
• LiteAgent 的循环用的是 Run 的异步迭代器形态 async for data, _ in self._llm.run(...)(agents/lite/agent.py:112-117),边消费事件边推进——是第 4 章 Run.__aiter__ 的最佳示例。

命名迁移

RequirementAgent 原来住在 beeai_framework.agents.experimental。现在那个路径是一个自动生成的弃用垫片:

# python/beeai_framework/agents/experimental/agent.py 节选
# This file is auto-generated by scripts/generate_shims.py
warnings.warn("beeai_framework.agents.experimental.agent is deprecated ...", DeprecationWarning)
sys.modules[__name__] = _new_module

sys.modules[__name__] = _new_module 这招让旧路径的 import 完全等价于新路径,连 isinstance 都不会出问题。

---

1.9 代码地图

主题	文件路径	符号名
门面与配置	python/beeai_framework/agents/requirement/agent.py	RequirementAgent
入参处理	python/beeai_framework/agents/requirement/agent.py	_process_input
落账策略	python/beeai_framework/agents/requirement/agent.py	save_intermediate_steps 分支
主循环	python/beeai_framework/agents/requirement/_runner.py	RequirementAgentRunner.run
单轮逻辑	python/beeai_framework/agents/requirement/_runner.py	RequirementAgentRunner._run
文本转工具调用兜底	python/beeai_framework/agents/requirement/_runner.py	_create_final_answer_tool_call
签发本轮通行证	python/beeai_framework/agents/requirement/_runner.py	_create_request、_force_final_answer_as_tool
工具执行与错误转述	python/beeai_framework/agents/requirement/_runner.py	_invoke_tool_calls
假工具 / 结构化输出	python/beeai_framework/agents/requirement/utils/_tool.py	FinalAnswerTool
运行状态	python/beeai_framework/agents/requirement/types.py	RequirementAgentRunState、RequirementAgentRunStateStep
死循环检测	python/beeai_framework/agents/tool_calling/utils.py	ToolCallChecker、_is_same_tool_call
滑动窗口计数	python/beeai_framework/utils/counter.py	OccurrencesCounter、RetryCounter
工具并发执行	python/beeai_framework/agents/_utils.py	run_tools、ToolInvocationResult
记忆配对提取	python/beeai_framework/memory/utils.py	extract_last_tool_call_pair、TEMP_MESSAGE_META_KEY
系统 / 任务 / 错误提示模板	python/beeai_framework/agents/requirement/prompts.py	RequirementAgentSystemPrompt 等