数据截至 (上游 commit 460c729002dc)
第 2 章 · 需求与规则
本章讲什么: BeeAI 最值得学的一块。你声明「什么时候允许 / 强制哪个工具」,框架每一轮把这些声明折叠成一张「通行证」,直接变成发给模型的
tools和tool_choice。本章拆开这个折叠算法,并指出它的优先级语义里一个容易误解的地方。
2.1 最小单位:Rule 的五个开关
一条规则说的是「某个工具在这一轮的处境」,只有五个字段(外加一个说明理由的字符串):
# python/beeai_framework/agents/requirement/requirements/requirement.py:27-33
class Rule(BaseModel):
target: str # 作用于哪个工具(按名字)
allowed: bool = True # 能不能用
reason: str | None = None # 为什么(会渲染进系统提示)
prevent_stop: bool = False # 禁止 agent 这一轮收尾
forced: bool = False # 必须用它
hidden: bool = False # 连提都不提(不出现在系统提示里)
五个开关的语义差别值得摆成表:
| 开关 | 模型能在工具列表里看到吗 | 模型能调吗 | 典型用途 |
|---|---|---|---|
| 默认 | 能 | 能 | 正常 |
allowed=False | 能(标注 Allowed: False + reason) | 不能 | 「现在还不能搜,得先查天气」 |
hidden=True | 不能 | 不能 | 权限不足的工具,索性不让模型知道 |
forced=True | 能 | 必须 | 「第 1 步必须先思考」 |
prevent_stop=True | — | — | 「至少调一次天气才准交卷」 |
allowed=False 和 hidden=True 的区别是本框架的一个小巧思: 不允许但可见,会把工具连同「为什么现在不给用」一起写进系统提示——等于教模型「你现在该干别的」。相比之下直接藏掉,模型可能反复瞎猜。
2.2 需求:一个每轮被调用的纯函数
Requirement 是抽象基类,核心只有一个方法:
# python/beeai_framework/agents/requirement/requirements/requirement.py:75-79 节选
@abstractmethod
def run(self, state: T) -> Run[list[Rule]]: ...
async def init(self, *, tools: list[AnyTool], ctx: RunContext) -> None: ...
| 方法 | 何时被调 | 干什么 |
|---|---|---|
init | 每次 run() 开始时一次(utils/_llm.py:42-56) | 校验它引用的工具真的存在;可以在这里挂事件监听 |
run | 每一轮迭代都调一次 | 看着当前 state 决定这一轮吐哪些 Rule |
还有两个属性:priority(默认 10,必须为正,requirement.py:64-73)和 enabled(关掉的需求直接跳过)。
写一个自定义需求只要几行
最轻的写法是装饰器(requirement.py:125-156):
# 示意,非源码
@requirement(name="no-search-before-think", targets=["duckduckgo"])
def gate_search(state, ctx) -> list[Rule]:
thought_done = any(step.tool and step.tool.name == "think" for step in state.steps)
# 没思考过就先别搜,并告诉模型原因
return [Rule(target="duckduckgo", allowed=thought_done, reason="先用 think 梳理再搜")]
targets 参数不是装饰用的——它在 init 阶段做存在性断言(_assert_all_rules_found,requirement.py:148-152),你写错工具名会在启动时炸,而不是跑到第 5 轮才炸。
仓库里有一个更实用的真实例子(python/examples/agents/requirement/complex.py)叫 RepeatIfEmptyRequirement:如果上一步某个工具输出为空,就 Rule(target=..., forced=True) 逼模型再试一次。「工具返回空就重试」这条策略,在这个框架里是十几行的声明,而不是循环里的一个 if 分支。
2.3 主力内置需求:ConditionalRequirement
90% 的场景不用自己写需求,ConditionalRequirement 已经把常见的时序约束参数化了。
| 参数 | 含义 |
|---|---|
force_at_step=N | 第 N 步必须调它(算的是成功步数 + 1) |
only_after=[A, B] | A 和 B 都调过之后才允许 |
only_before=[C] | 一旦 C 调过就不再允许 |
force_after=[A] | 上一步是 A 时,这一步强制调它 |
min_invocations=N | 不到 N 次不准交卷(靠 prevent_stop) |
max_invocations=N | 超过 N 次就禁用 |
consecutive_allowed=False | 不许和上一步是同一个工具 |
only_success_invocations | 计数时是否忽略失败的步骤(默认 True) |
custom_checks=[fn] | 任意 (state) -> bool 断言,一个不过就禁用 |
priority / enabled / reason | 优先级 / 开关 / 给模型的解释 |
目标可以写成工具名字符串、工具类、或工具实例三种形式(_utils.py:46-64 的 _target_seen_in 分别按名字相等、isinstance 、is 匹配)。写类名最省事:ConditionalRequirement(ThinkTool, force_at_step=1)。
判定顺序(命中即停)
ConditionalRequirement.run(conditional.py:133-199)按这个顺序短路:
┌──────────────────────────────────┐
state ──▶│ 过滤步骤:only_success 则丢掉失败步 │
└───────────────┬──────────────────┘
▼
① 不许连着调,且上一步就是它? ──▶ 禁用
▼
② 调用次数 ≥ max_invocations? ──▶ 禁用
▼
③ only_after 里还有没调过的? ──▶ 禁用
(遍历途中撞见 only_before 的工具也禁用)
▼
④ 任一 custom_check 返回 False? ──▶ 禁用
▼
允许
最后统一交给内部函数 resolve(allowed) 生成规则,它顺带算两件事:
# python/beeai_framework/agents/requirement/requirements/conditional.py:156-172
forced = bool(
_target_seen_in(last_step_tool, self._force_after) or self._force_at_step == current_step
if allowed
else False
)
return [
Rule(
# pyrefly: ignore [missing-attribute]
target=source_tool.name,
allowed=allowed,
forced=forced,
hidden=False,
prevent_stop=(self._min_invocations > invocations) or (forced and self._force_prevent_stop),
reason=self._reason if not allowed else None,
)
]
注意 prevent_stop 的两个来源:还没达到最小调用次数,或者这一轮本来就强制它(强制却允许交卷是自相矛盾的)。
启动期的自洽性检查
_check_invariant(conditional.py:72-99)和 init(:101-125)在跑之前就拦下一堆写错的组合,比如:
min_invocations > max_invocations- 同一个工具既在
only_before又在only_after - 自己引用自己的
only_before/force_after consecutive_allowed=True时把自己写进force_after—— 报错信息直接点破后果:会造成无限循环
把配置错误挡在启动期而不是运行期,是这类声明式 API 能好用的前提。
2.4 折叠算法:从一堆规则到一张通行证
这是本章的核心。全部逻辑在 RequirementsReasoner.create_request(utils/_llm.py:64-168),分四段。
┌───────────── ─────────────────────────────────────────┐
│ 第一段 · 收集 │
│ 遍历所有 enabled 的需求 → requirement.run(state) │
│ 产出的每条 Rule 按目标工具塞进 rules_by_tool[工具名] │
│ 同时记下产出它的需求的 priority │
└──────────────────────┬───────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 第二段 · 追加临时规则(extra_rules) │
│ 运行时的补丁(死循环检测、强制交卷) │
│ 优先级 = 该工具现有最高优先级 + 1(永远压过静态需求) │
└──────────────────────┬───────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 第三段 · 逐工具折叠 5 个布尔开关 │
└──────────────────────┬───────────────────────────────┘
▼
┌──────────────────────────────────────────────────────┐
│ 第四段 · 全局收敛 → RequirementAgentRequest │
└──────────────────────────────────────────────────────┘
一个决定后面所有推理的前提: rules_by_tool 在第一段开始前就用全部工具初始化成空列表(_llm.py:81:{t.name: [] for t in self._tools},其中 self._tools 已含 final_answer,见 _llm.py:37)。所以第三段的循环会遍历到每一个工具,包括一条规则都没收到的那些——它们走的是「四个布尔量全取初值」的路径,结论是允许。
第三段的真实规则:布尔 OR,不是优先级覆盖
这里有个很容易误解的点,值得逐字看源码——所以下面两块都不压行,原样贴:
# python/beeai_framework/agents/requirement/utils/_llm.py:112-127
for rule_entry in rules:
rule = rule_entry.rule
if not rule.allowed:
is_allowed = False
if rule.hidden:
is_hidden = True
if rule.forced:
is_forced = True
if rule.prevent_stop:
is_prevent_stop = True
prevent_step_refs.append(rule_entry)
if rule.reason:
reason_by_tool[tool] = rule.reason
if is_allowed and is_hidden:
is_allowed = False
# python/beeai_framework/agents/requirement/utils/_llm.py:129-133
if is_allowed:
_append_if_not_exists(allowed, tool)
if is_forced and (not forced or forced_level < max_priority):
forced = tool
forced_level = max_priority
三个结论,按重要性排:
- 禁止永远压过允许。 循环把所有规则都走一遍,任何一条
allowed=False就把这个工具判死——哪怕它来自优先级最低的需求。想「用高优先级需求覆盖低优先级的禁令」是做不到的;要实现「特殊情况下放行」,只能让那条禁令自己在run里判断后不吐 这条规则。 priority只在一处起作用:多个工具同时被强制时,选优先级最高的那个。 看forced_level < max_priority这一行——比较的是工具的最高规则优先级(max_priority取自该工具排序后的首条规则,_llm.py:104-106),不是逐条规则的优先级。- 隐藏蕴含禁止(
is_allowed and is_hidden → is_allowed = False),但反之不然。仓库有专门的单测守着「隐藏一个工具不会连累其它工具」(python/tests/agents/test_requirements_reasoner.py,用例名test_hidden_tool_does_not_disable_other_tools)。
第四段:三步全局收敛
# python/beeai_framework/agents/requirement/utils/_llm.py:139-158 节选
if forced is not None:
allowed.clear()
_append_if_not_exists(allowed, forced)
_append_if_not_exists(allowed, self.final_answer) # 强制时仍保留交卷通道
if prevent_stop and not isinstance(forced, FinalAnswerTool):
remove_by_reference(allowed, self.final_answer) # 除非强制的就是交卷
if not allowed:
raise AgentError("One of the generated rules is preventing the agent from continuing. ...")
tool_choice = forced if forced is not None else "required"
if len(allowed) == 1:
tool_choice = allowed[0]
逐条读:
| 步骤 | 效果 |
|---|---|
| 有强制工具 | 允许集清空重建,只剩 [强制工具, final_answer] |
有 prevent_stop | 把 final_answer 从允许集里摘掉——模型这一轮物理上无法交卷 |
| 允许集空了 | 抛 AgentError,并把导致 prevent_stop 的规则连同它们来自哪个需求一起 JSON 打印出来 |
| 只剩一个工具 | tool_choice 直接设成那个工具实例(比 "required" 更强的约束) |
最后一行还有一个降级:
# python/beeai_framework/agents/requirement/utils/_llm.py:164
tool_choice=tool_choice if isinstance(tool_choice, Tool) or force_tool_call or prevent_stop else "auto"
也就是说:没有强制、也不禁止收尾时,如果你把 final_answer_as_tool=False(它经 Runner 传成 force_tool_call,_runner.py:196),tool_choice 会退回 "auto",模型可以直接吐文本(靠上一章讲的兜底转成 final_answer)。这是给那些一被 required 就输出质量下降的模型留的开关。
用这套算法反推第 1 章的第三级兜底
第 1 章讲过,模型既不调工具、又榨不出文本时,Runner 走最后一档:_reasoner.update(requirements=[]) + extra_rules=[Rule(target=final_answer, allowed=True)] + 把 _force_final_answer_as_tool 置为 True(_runner.py:298-303)。把它代进上面的算法,结果是这样:
| 环节 | 代入后的结论 |
|---|---|
| 需求清空 | rules_by_tool 每个键都是空列表,没有任何 allowed=False |
| 逐工具折叠 | 每个工具 is_allowed 保持初值 True → 全部工具重新进入允许集 |
| extra_rules | 只是给 final_answer 补一条显式的 allowed=True,不排他 |
force_tool_call=True | tool_choice 被钉在 "required",不会退回 "auto" |
所以这一档的准确表述是「弃掉全部约束 + 兜底允许 final_answer + 强制这一轮必须调工具」,而不是「只放行 final_answer」。它逼模型交卷靠的是「必须调一个工具、而此刻交卷是最顺的那条路」,不是靠把别的工具拿走。
2.5 通行证怎么变成提示词
通行证 RequirementAgentRequest(types.py:87-96)有七个字段:全部工具(tools)、允许集(allowed_tools)、每个工具的理由(reason_by_tool)、隐藏集(hidden_tools)、tool_choice、final_answer 工具实例、can_stop。它在两个地方被消费:
其一,渲染系统提示(utils/_llm.py:176-184 节选):
tools=[RequirementAgentToolTemplateDefinition.from_tool(
tool, allowed=tool in request.allowed_tools,
reason=request.reason_by_tool.get(tool, None))
for tool in request.tools if tool not in request.hidden_tools]
模板里每个工具渲染成四行(prompts.py:71-79):Name / Description / Allowed: True|False / 可选的 Reason。不允许的工具也在列表里,只是被标了 False ——这就是 2.1 说的那个巧思落地的地方。同一次渲染还会带上 final_answer 的名字、自定义 schema 和 instructions(_llm.py:185-191),第 1 章讲的结构化输出就是从这里进提示词的。
其二,变成 API 参数(_runner.py:147-153):
options = ChatModelOptions(
max_retries=self._run_config.max_retries_per_step,
tools=request.allowed_tools,
tool_choice=request.tool_choice,
stream_partial_tool_calls=True,
fallback_tool=request.final_answer if request.can_stop else None,
)
双保险:提示词里说清楚(软约束) + API 参数里锁死(硬约束)。 支持 tool_choice 的模型吃硬约束,不支持的靠软约束加第 3 章的降级路径。fallback_tool 这个参数的语义(以及 Backend 侧还会对它做什么补充)见 3.4。
系统提示每轮重建
因为允许集每轮都变,系统提示也每轮重新渲染(_runner.py:137-143 每次都 _create_system_message)。这带来一个副作用,框架顺手处理了:
# python/beeai_framework/agents/requirement/_runner.py:155-158 节选
cache_control_injection_points = [
{"location": "message",
"index": 1 if self._requirements else 0}, # 有需求时系统提示是动态的,缓存断点后移一条
...
]
提示缓存(prompt caching)要求被缓存的前缀逐字节稳定。有需求 = 系统提示会变 = 缓存断点必须跳过它。 这一行注释写得很直白:system prompt might be dynamic when requirements are set。另一个断点的算法和 provider 特判见 3.7。
2.6 一个双机制的需求:AskPermissionRequirement
人工审批这个需求(requirements/ask_permission.py)很特别——它同时用了两条路:
路一:出规则。 run 把已记住的选择转成规则(ask_permission.py:94-104),被拒的工具在下一轮直接不允许(可选地隐藏)。
路二:在工具真正执行前拦截。 在 init 阶段给每个受管工具挂一个事件监听:
# python/beeai_framework/agents/requirement/requirements/ask_permission.py:69-73
ctx.emitter.on(
create_internal_event_matcher("start", tool, parent_run_id=ctx.run_id),
handler,
EmitterOptions(is_blocking=True, persistent=True, match_nested=True),
)
回调里如果用户拒绝,它不抛异常,而是直接改写事件对象:
# python/beeai_framework/agents/requirement/requirements/ask_permission.py:89-90
if not allowed:
data.output = StringToolOutput("This tool is not allowed to be used.")
为什么这一行就能阻止工具执行?因为 RunContext.enter 在真正调用 handler 之前会看一眼 start_event.output:
# python/beeai_framework/context.py:217-220
if start_event.output is not None:
return start_event.output
else:
return await fn(context)
任何一次执行都可以被 start 事件的监听者短路。 这是第 4 章要展开的框架级能力,审批需求只是它最漂亮的一个用例。FinalAnswerTool 被默认排除在审批之外(ask_permission.py:50),不然 agent 永远交不了卷。
2.7 关键细节与坑
- 需求间冲突的报错信息很好。 允许集被清空时,
AgentError会把prevent_step_refs(规则 + 来源需求 + 优先级)整个序列化出来(utils/_llm.py:148-154),定位很快。 force_at_step数的是成功步数。current_step = len(steps) + 1(conditional.py:147),而steps默认过滤掉了失败步(only_success_invocations=True)。工具失败重试不会消耗「第几步」的名额。force_at_step到点却不满足条件会抛错。resolve里显式抛RequirementError,消息是「工具 X 无法在第 N 步执行,因为它未满足全部要求」(conditional.py:148-154)——这是有意让配置矛盾早暴露,而不是静默跳过。- 需求实例带状态。
Requirement.state是个 dict,AskPermissionRequirement._state记住用户选择(ask_permission.py:52)。同一个需求实例在多次run()之间是共享的,这既是remember_choices能工作的原因,也意味着自定义需求里存计数器要小心跨运行残留(RepeatIfEmptyRequirement就是靠每次不命中时把_remaining复位来自保)。
2.8 代码地图
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 规则定义 | python/beeai_framework/agents/requirement/requirements/requirement.py | Rule |
| 需求基类 / 装饰器 | python/beeai_framework/agents/requirement/requirements/requirement.py | Requirement、requirement、run_with_context |
| 条件需求 | python/beeai_framework/agents/requirement/requirements/conditional.py | ConditionalRequirement、_check_invariant |
| 审批需求 | python/beeai_framework/agents/requirement/requirements/ask_permission.py | AskPermissionRequirement |
| 折叠算法 | python/beeai_framework/agents/requirement/utils/_llm.py | RequirementsReasoner.create_request、RuleEntry |
| 需求初始化 | python/beeai_framework/agents/requirement/utils/_llm.py | RequirementsReasoner.update |
| 系统提示渲染 | python/beeai_framework/agents/requirement/utils/_llm.py | _create_system_message |
| 目标匹配(名/类/实例) | python/beeai_framework/agents/requirement/requirements/_utils.py | _target_seen_in、_assert_all_rules_found |
| 通行证数据结构 | python/beeai_framework/agents/requirement/types.py | RequirementAgentRequest |
| 系统提示模板 | python/beeai_framework/agents/requirement/prompts.py | RequirementAgentSystemPrompt |
| 折叠算法单测 | python/tests/agents/test_requirements_reasoner.py | test_hidden_tool_does_not_disable_other_tools |
| 自定义需求实例 | python/examples/agents/requirement/complex.py | RepeatIfEmptyRequirement |