Guard cn

Guard 体系是 Jianmu 在工具调用与模型调用路径上插入的一道策略执行层，负责在 Agent 执行高风险操作之前进行多维度的安全检查。它提供四类内建检查器——工具策略 (Tool Policy)、调用频率限制 (Rate Limit)、Token 预算控制 (Budget)、确认拦截 (Confirm)——并通过 GuardEnforcer 将它们组合为一条有序的检查链。当检查链判定某操作需要人工审批时，系统会转入挂起等待状态，由外部回调或 TUI 交互完成裁决后再恢复执行。本章聚焦 Guard 体系自身的检查机制；关于挂起恢复的完整生命周期，参见人机交互挂起恢复。

架构概览：检查链与上下文传播¶

Guard 体系围绕一条有序检查器链展开工作。每次工具调用或 LLM 调用被拦截时，GuardEnforcer 将上下文依次传递给链上的每个 CheckerProtocol 实现，直到某个检查器返回 DENY（硬拒绝）或 PENDING（挂起等待），或者遍历完成返回 ALLOW（放行）或 CONFIRM（需要确认）。

flowchart TD
    A[ToolExecutor.update_async] --> B[_preflight_actions]
    B --> C{GuardEnforcer 已构建?}
    C -->|否| D[from_guard_constraints<br/>构建检查链]
    C -->|是| E[遍历 actions]
    D --> E
    E --> F[ToolPolicyChecker.check]
    F -->|DENY| G[拒绝, 写入错误消息]
    F -->|CONFIRM| H[标记 _needs_confirm]
    F -->|ALLOW| I[RateLimitChecker.check]
    H --> I
    I -->|超限 DENY| G
    I -->|ALLOW| J[BudgetChecker.check]
    J -->|超预算 DENY| G
    J -->|ALLOW| K[ConfirmChecker.check]
    K -->|需要确认| L[调用 ApprovalManager<br/>挂起或批准]
    K -->|ALLOW| M[执行工具]
    L -->|PENDING| N[挂起等待外部裁决]
    L -->|APPROVED| M
    L -->|REJECTED| G

检查器之间的通信通过 GuardCheckContext 承载。该上下文是一个 dataclass 实例，同时兼容旧式 dict 访问语法——CheckerProtocol.check() 的类型签名接受 CheckContextLike = GuardCheckContext | dict[str, Any]，这使得既有检查器无需迁移即可继续工作。

核心类型系统：Decision、BudgetConstraints 与 GuardConstraints¶

Guard 体系的类型层定义了三组核心数据结构，分别对应检查决策、预算限制和整体约束配置。

Decision 枚举¶

Decision 是检查器返回的四种标准结果：

值	含义	后续行为
`ALLOW`	无条件放行	继续下一个检查器
`DENY`	硬拒绝	立即终止检查链，操作被阻止
`CONFIRM`	需要确认（软拦截）	继续后续检查器，但最终需用户批准
`PENDING`	挂起等待外部裁决	立即终止检查链，等待恢复

GuardEnforcer.check() 的遍历策略为：DENY 和 PENDING 立即短路返回；CONFIRM 继续遍历（后续检查器可能升级为 DENY）；全部通过后若存在 CONFIRM 标记则返回 CONFIRM，否则返回 ALLOW。

BudgetConstraints¶

BudgetConstraints 是一个简单的 dataclass，包含三个可选字段，任一为 None 表示该维度不设限：

total_tokens: Optional[int]     # Token 总量上限
prompt_tokens: Optional[int]    # Prompt token 上限
completion_tokens: Optional[int] # Completion token 上限

BudgetChecker 在检查 llm_call 和 tool_call 动作时，从上下文 usage 字段中读取 {prompt_tokens, completion_tokens, total_tokens} 并逐一比对，任一维度超限即返回 DENY。

GuardConstraints¶

GuardConstraints 是面向用户的顶层配置入口，将 Guard 体系的各项策略收敛到一个 dataclass 中：

tool_policy: dict[str, str]     # 工具名 → "allow" | "deny" | "confirm"
default_policy: str = "allow"   # 未在 tool_policy 中出现的工具的默认策略
max_tool_calls: Optional[int]   # 单次运行最大工具调用次数
budget: BudgetConstraints       # Token 预算限制

它提供 from_dict() 工厂方法，使 YAML / JSON 配置可以直接反序列化为该结构。在 Constraints 聚合类中，guard 字段承载的就是 GuardConstraints 实例。

GuardCheckContext：类型化上下文与字典兼容¶

GuardCheckContext 是 Guard 体系的核心上下文载体，它将工具调用信息、审批状态、用量数据和附加元数据整合为一个结构化的 dataclass。它的设计要点在于同时支持类型化访问和字典式访问——通过实现 __getitem__、__setitem__ 和 get() 方法，它可以被既有代码当作 dict 使用，同时为新代码提供编译期类型安全。

其内部子结构如下：

字段	类型	用途
`action`	`str`	当前动作标识（`"tool_call"` / `"llm_call"`）
`usage`	`dict`	累积的 Token 用量统计数据
`metadata`	`dict`	自由扩展的附加键值对
`tool`	`ToolCheckContext \\| None`	工具名称、参数、线程 ID、节点定位符
`approval`	`ApprovalCheckContext \\| None`	审批请求 ID、进度状态、待裁决结果

ApprovalCheckEffect 是关键的可变副作用载体：confirm 检查器通过它写入 progress_state、request_id 和 approval_result，ToolExecutor 通过读取这些字段判断是否需要挂起。to_legacy_context() 方法将整个类型化上下文扁平化为一个 dict，供 ConfirmChecker 的旧式路径使用。

GuardEnforcer：检查链编排与工厂构建¶

GuardEnforcer 是 GuardProtocol 的具体实现，它持有一组有序的 CheckerProtocol 检查器，并提供 check() 方法对每个动作执行整条链。其工厂方法 from_guard_constraints() 是从配置到运行时检查链的桥梁，构建逻辑如下：

flowchart LR
    A[GuardConstraints] --> B{有 tool_policy<br/>或 default_policy?}
    B -->|是| C[ToolPolicyChecker]
    B -->|否| D{max_tool_calls<br/>不为 None?}
    C --> D
    D -->|是| E[RateLimitChecker]
    D -->|否| F{budget<br/>不为空?}
    E --> F
    F -->|是| G[BudgetChecker]
    F -->|否| H[ConfirmChecker]
    G --> H
    H --> I[GuardEnforcer<br/>checkers 列表]

工厂方法接受 tool_aliases 参数用于工具名别名解析（默认从 BuiltinToolProvider.aliases() 获取），以及 approval_manager 参数（未提供时创建默认实例）。最终构建的检查链顺序固定为：ToolPolicy → RateLimit → Budget → Confirm。

在 ToolExecutor.update_async() 中，GuardEnforcer 在首次执行时延迟构建——它从节点的 constraints 或运行上下文的 ctx.constraints 中提取 GuardConstraints，再调用 from_guard_constraints() 完成装配。reset_for_run() 方法在每次 Agent 运行前重置所有检查器状态（清零调用计数等）。

四类内建检查器详解¶

ToolPolicyChecker：工具级策略执行¶

ToolPolicyChecker 维护一个 工具名 → 策略 映射，策略值可以是 "allow"、"deny" 或 "confirm"。未在映射中出现的工具使用 default 回退策略。

检查流程如下： 1. 仅处理 action == "tool_call" 的动作，其余直接放行 2. 从上下文中提取 tool_name，依次尝试原始名和别名（通过 aliases 映射） 3. 匹配到 "deny" → 返回 DENY 并给出拒绝原因 4. 匹配到 "confirm" → 在上下文中标记 _needs_confirm = True，返回 CONFIRM 5. 匹配到 "allow" 或默认 → 返回 ALLOW

标记 _needs_confirm 是关键步骤：它让下游的 ConfirmChecker 能够识别需要确认的工具调用。

RateLimitChecker：工具调用频率限制¶

RateLimitChecker 实现了一个简单的计数器逻辑：它在每次 tool_call 检查时递增内部 call_count，当计数超过 max_calls 阈值时返回 DENY。

初始化: call_count = 0, max_calls = N
每次 tool_call 检查: call_count += 1; 若 call_count > max_calls → DENY
reset(): call_count = 0

注意此限制是单次运行内的累计计数——每次 Agent.run() 调用 reset_for_run() 会通过 GuardEnforcer.reset() 清零所有检查器状态。

BudgetChecker：Token 预算控制¶

BudgetChecker 在 llm_call 和 tool_call 两种动作上均执行检查。它从上下文的 usage 字段中读取累计用量，与 BudgetConstraints 中配置的三个维度逐项比对：

usage = {prompt_tokens: int, completion_tokens: int, total_tokens: int}
对每个维度 key:
   若 budget.key 不为 None 且 usage[key] > budget.key → DENY
全部通过 → ALLOW

这意味着预算检查是跨 LLM 调用累积的——上下文中的 usage 字段需要由上游（通常是 LLM 调用路径）持续更新，才能正确反映累计消耗。

ConfirmChecker：人机确认拦截¶

ConfirmChecker 是四类检查器中最复杂的一个，因为它涉及到与 ApprovalManager 的交互、挂起/恢复状态的管理、以及类型化/旧式两条代码路径的兼容。

检查流程（类型化路径）： 1. 若上下文中 approval.needs_confirm 为 False → 直接 ALLOW 2. 若 approval.progress_state == "approved" → 标记放行并返回 ALLOW 3. 尝试通过 pop_resolved 回调查找已裁决结果 4. 调用 approval_manager.confirm_tool()，执行配置的确认回调 5. 回调返回 APPROVED → 标记放行 6. 回调返回 PENDING → 标记挂起并返回 PENDING 7. 回调返回 REJECTED → 标记拒绝并返回 DENY

ConfirmChecker 同时维护一条旧式 dict 兼容路径（_check_legacy），通过 ToolApprovalCheckerContext 适配器操作旧式上下文映射。

ApprovalManager：统一审批入口与回调解析¶

ApprovalManager 是整个 Guard 体系中审批流的中央调度器。它提供三层回调解析机制：

优先级	回调源	说明
1	显式传入的 `callback`	直接传给 `ApprovalManager.__init__`
2	`tool_callback` / `node_callback`	分类型的专用回调
3	`default_confirm_callback` / `default_node_approval_callback`	内置 CLI 交互回退

核心方法 request(req: ApprovalRequest) → ApprovalResult 执行如下流程： 1. 发出 approval.requested 运行时事件 2. 根据 req.kind 选择回调（通用 → 工具 → 节点 → 回退默认） 3. 规范化回调返回值（支持 bool、str、ApprovalResult） 4. 若结果为 PENDING，将请求存入 pending_store 5. 发出 approval.resolved 运行时事件 6. 调用审计钩子（若配置）

confirm_tool() 和 approve_node() 是两个便捷方法，它们内部构建 ApprovalRequest 并委托给 request()。resolve_request() 提供外部裁决入口——外部系统（如 TUI 或 API）通过它将审批决定写入 pending_store，触发订阅回调。

审批存储生命周期¶

ApprovalStoreProtocol 定义了审批持久化的完整契约：

put_request → get_request → resolve_request → get_result → ack_result → prune_result

InMemoryApprovalStore 提供了线程安全的内存实现。ack_result 标记结果已被消费，prune_result 在 Checkpoint 持久化完成后物理删除。这种分阶段生命周期确保了在挂起恢复场景中审批结果不会丢失或重复消费。

默认确认回调¶

当未配置自定义回调时，default_confirm_callback 按以下优先级决策： 1. 读取 execution.confirm_default 配置 → 若设为 allow/deny，直接返回 2. 若无 TTY（非交互环境）→ 返回 False（拒绝）或 PENDING（当 non_tty_pending=True） 3. 若有 TTY → 通过 input() 提示用户输入 [y/N]

这确保了在 CI/CD 等非交互环境中默认安全拒绝，而在开发终端中提供即时确认能力。

Approval 装饰器：节点级人工审批门¶

Approval 是 py_trees Decorator 的子类，为任意行为树节点添加执行前/执行后的人工审批门。它支持三种 phase 配置：

Phase	行为
`"before"`	子节点执行前等待审批
`"after"`	子节点执行完成后等待审批
`"both"`	执行前后均需审批

其核心 tick 流程为： 1. 若需预审批且未通过 → 调用 _consume_approval("before")，可能返回 RUNNING（等待中） 2. 驱动子节点 tick 3. 子节点完成后，若需后审批且未通过 → 调用 _consume_approval("after")

_consume_approval() 是状态机核心：它首先从运行时元数据中同步审批进度缓存，检查是否已有已批准的进度记录或已解析的审批结果；若无，则启动异步审批任务；若任务返回 PENDING，则调用 interaction.suspend() 挂起整个运行等待外部裁决。

审批进度通过 InteractionController 持久化到运行时元数据中，支持跨 tick 恢复。on_reject 参数控制拒绝后的行为——"failure"（默认）返回 Status.FAILURE，"success_skip" 返回 Status.SUCCESS 跳过子节点。

ToolExecutor 中的 Guard 集成全流程¶

ToolExecutor 是 Guard 体系的主要消费方。其 update_async() 方法中 Guard 的参与分为三个阶段：

阶段一：延迟构建。首次执行时，ToolExecutor 检查 _enforcer 和 _tool_runner 是否已初始化。若否，从 self.constraints 或 self.ctx.constraints 中提取 GuardConstraints，调用 GuardEnforcer.from_guard_constraints() 构建检查链，同时构建 ToolRunner。此设计允许用户在 ToolExecutor 构造时通过 guard_enforcer 参数注入自定义实现，或完全依赖配置驱动。

阶段二：预检拦截。_preflight_actions() 对每个待执行的工具调用构建 ToolApprovalContext，转换为 GuardCheckContext，调用 enforcer.check()。返回决策的处理逻辑如下：

DENY 或 CONFIRM：为所有 actions 写入错误消息，返回 SUCCESS（工具未执行）
PENDING：写入审批进度，调用 interaction.suspend() 挂起，返回 RUNNING
ALLOW：继续执行，同时将已批准的 request_id 写入进度记录

阶段三：执行与清理。工具执行完成后，ToolExecutor 将工具结果写入状态消息，清除对应审批进度记录。

配置组装全景¶

从顶层配置到运行时检查链的完整通路如下：

flowchart TD
    A[jianmu.yaml / .env] --> B[Config 加载]
    B --> C[Constraints 聚合]
    C --> D[Constraints.guard<br/>→ GuardConstraints]
    D --> E[GuardEnforcer.from_guard_constraints]
    E --> F1[ToolPolicyChecker]
    E --> F2[RateLimitChecker]
    E --> F3[BudgetChecker]
    E --> F4[ConfirmChecker]
    F4 --> G[ApprovalManager]
    G --> H{回调配置}
    H --> I1[自定义 callback]
    H --> I2[tool_callback]
    H --> I3[default_confirm_callback]

GuardConstraints 的 YAML 配置结构示例：

constraints:
  guard:
    tool_policy:
      calculator: confirm
      python_repl: deny
      http: allow
    default_policy: allow
    max_tool_calls: 20
    budget:
      total_tokens: 100000
      prompt_tokens: null
      completion_tokens: null

扩展指南：自定义检查器¶

CheckerProtocol 是一个极简协议，实现自定义检查器只需完成两个方法：

class MyCustomChecker:
    async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
        # 检查逻辑，返回 (Decision, reason)
        ...

    def reset(self) -> None:
        # 重置内部状态
        ...

自定义检查器可以通过两种方式集成： 1. 直接注入：构造 GuardEnforcer(checkers=[MyCustomChecker(), ...])，传给 ToolExecutor(guard_enforcer=...) 2. 继承工厂：重写 GuardEnforcer.from_guard_constraints() 的检查链构建逻辑

CheckContextLike 的联合类型设计意味着检查器可以自由选择访问方式——使用 isinstance(context, GuardCheckContext) 走类型化路径，或直接使用 context.get("key") 走字典路径。

阅读推进¶

Guard 体系通过检查链实现工具调用的安全护栏，而挂起后的恢复机制涉及运行时与外部系统的深度协作。建议继续阅读以下页面：

人机交互挂起恢复：审批流、用户输入请求与外部回调编排 — 理解 suspend() / resume() 的完整生命周期
工具与技能节点：ToolExecutor、SkillNode 与约束联动 — 了解 ToolExecutor 的完整执行流程
执行环境：LocalSandbox 与 DockerSandbox 的工具隔离运行 — 了解工具执行层的内置沙箱隔离
项目配置体系：jianmu.yaml 与 .env 协同工作 — 了解 Guard 策略的配置入口