需求规则引擎:把约束算成「本步通行证」

这是整个框架最值得带走的设计。本章自底向上讲:Rule 是什么 → Requirement 怎么每步产出规则 → RequirementsReasoner 怎么把规则聚合成一次「请求」→ 两个内置需求(条件 / 询问许可)怎么实现。

3.1 最小单元:Rule

Rule 是一个对某一个工具的裁决(agents/requirement/requirements/requirement.py:27-36)。它只有几个布尔旗标:

字段	含义
target	这条规则管哪个工具(工具名)
allowed	这一步能不能用它(默认 True)
forced	必须用它(把它设成 tool_choice)
hidden	完全不给模型看见(连 prompt 里都不出现)
prevent_stop	阻止 agent 这一步停下(不许调 final_answer)
reason	给模型的解释(写进系统提示)

关键直觉:规则不是「一次性配置」,而是「每一步重新生成」。 同一个需求在第 1 步可能产出 forced=True,第 5 步产出 allowed=False——因为它读的是当前状态。

3.2 每步产出规则:Requirement

Requirement 是抽象基类,核心就一个抽象方法 run(state) -> list[Rule](requirement.py:75-76)。它每一步被调用一次,拿到当前运行状态(里面有 state.steps,即到目前为止所有工具调用的轨迹),据此决定要发哪些规则。

# 示意,演示一个自定义需求的最小形态
class AfterSearchOnly(Requirement):
    name = "answer_after_search"

    def run(self, state):
        searched = any(s.tool and s.tool.name == "wikipedia" for s in state.steps)
        # 没查过资料 → 禁止 final_answer 这一步停下
        return [Rule(target="final_answer", allowed=searched, prevent_stop=not searched)]

框架还提供 @requirement 装饰器(requirement.py:125-156),让你不用写类、直接用函数定义一个需求。每个需求带一个 priority(默认 10,requirement.py:48),聚合时用来定胜负。

3.3 聚合:RequirementsReasoner.create_request

一个 agent 可能挂多个需求,它们可能对同一个工具发出互相矛盾的规则(A 说 allowed,B 说 forbidden)。RequirementsReasoner.create_request(agents/requirement/utils/_llm.py:64-168)负责把这些规则聚合成一致的本步决策。

聚合分四步:

① 跑所有 enabled 的需求 → 收集全部 Rule,按 target 工具分组
      rules_by_tool = { "wikipedia": [RuleEntry, ...], "think": [...], ... }
② 注入「额外规则」(runner 临时加的,如「禁掉死循环工具」)
③ 对每个工具,把它的规则按 priority 降序排,逐条「叠加」:
      任一条 not allowed → 该工具最终 not allowed
      任一条 forced      → 候选「强制工具」
      任一条 hidden      → 隐藏(且 hidden ⇒ not allowed)
      任一条 prevent_stop→ 本步不许停
④ 收尾裁决:
      有 forced 工具 → allowed 清空,只留它 + final_answer
      prevent_stop 且 forced 不是 final_answer → 从 allowed 里摘掉 final_answer
      allowed 为空 → 抛错(需求互相冲突,卡死了)

真实实现里这段逻辑的核心循环在 _llm.py:102-137:

# 示意,浓缩自 agents/requirement/utils/_llm.py:102-137
for tool_name, rules in rules_by_tool.items():
    rules.sort(key=lambda x: x.priority, reverse=True)   # 高优先级先看
    is_allowed, is_forced, is_hidden, is_prevent_stop = True, False, False, False
    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
        if rule.reason:          reason_by_tool[tool] = rule.reason
    if is_allowed and is_hidden: is_allowed = False     # 隐藏 ⇒ 不可用
    # ...据此填 allowed / hidden / forced / prevent_stop...

注意一个反直觉的细节: 优先级排序后,代码并不是「高优先级一票否决」,而是「只要有任意一条 not allowed,就 not allowed」(逐条叠加,禁止优先)。priority 真正起决定作用的地方在选哪个 forced 工具——多个工具都被强制时,取 max_priority 那个胜出(_llm.py:131-133)。

聚合的产物是一个 RequirementAgentRequest(agents/requirement/types.py:87-96):allowed_tools / hidden_tools / tool_choice / can_stop / reason_by_tool。这正是 01 章 第 6 节里喂给模型的那张「本步通行证」。

3.4 final_answer 也是一个工具

聚合时,final_answer 被当成和别的工具一样的 target参与裁决。这带来一个优雅的统一:

• 想让 agent 必须先做某事再答?发 Rule(target="final_answer", prevent_stop=True) 即可——它会被从 allowed 里摘掉,模型这一步就停不下来(_llm.py:144-146)。
• 想强制 agent 立刻交答案?把 final_answer 设成 forced。

Reasoner 在构造时就把 final_answer 塞进工具列表(_llm.py:37,self._tools = [*tools, final_answer]),所以它从头到尾都是「一等公民」。

3.5 内置需求一:ConditionalRequirement

ConditionalRequirement(agents/requirement/requirements/conditional.py)是最常用的需求,把「在什么条件下允许 / 强制某工具」做成一组声明式参数:

参数	作用
force_at_step=N	第 N 步强制调这个工具(README 例子用的就是它)
only_before / only_after	只在某些工具调用之前 / 之后才允许
force_after	某工具调用后,下一步强制调这个
min_invocations / max_invocations	至少 / 至多调几次
consecutive_allowed	是否允许连续两步调同一个工具
custom_checks	任意 state -> bool 的自定义条件

它的 run(conditional.py:133-199)就是把这些参数翻译成 Rule:数当前 steps 里这个工具被调了几次、上一步是谁、after 依赖是否都满足,然后 resolve(allowed=True/False) 产出规则。force_at_step 命中时还会把 prevent_stop 一起打开(conditional.py:169),确保「该这一步做却没做」时 agent 停不下来。

一个安全细节:如果某工具被 force_at_step 强制、却因别的条件不满足而 not allowed,resolve 会直接抛 RequirementError(conditional.py:147-154)——宁可显式报「规则冲突」,也不静默放过。

3.6 内置需求二:AskPermissionRequirement

AskPermissionRequirement(agents/requirement/requirements/ask_permission.py)实现「调危险工具前先问我」。它的实现方式很能体现框架的 emitter 设计:它不是在 run 里拦截,而是在 init 时给目标工具的 start 事件挂一个阻塞监听器。

# 示意,浓缩自 ask_permission.py:65-90
def setup_tool(tool):
    async def handler(data, _):
        # data 是工具的 start 事件;若用户拒绝,直接把 output 改写成「不允许使用」
        allowed = await self._handler(tool, data.input)
        if not allowed:
            data.output = StringToolOutput("This tool is not allowed to be used.")
    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),
    )

重点看: 监听器是 is_blocking=True,所以它能在工具真正执行之前插一个问询(默认走 io_confirm 命令行确认,ask_permission.py:107-108),用户说不就把工具输出替换掉。这依赖 03 章 讲的 emitter 阻塞监听机制——规则系统与事件系统在这里漂亮地合流了。

---

下一章把支撑这一切的横切骨架(执行 + 可观测)讲清楚。→ 03-runtime-emitter-context.md