Deep Agents 11. Human-in-the-loop 与 Rubric 质量控制

1. HITL、Rubric 与确定性校验的分工

Agent 能生成报告、调用发布 API，并不代表这些动作应该完全自动执行。真实业务至少要回答两个不同问题：

1. 这份结果是否达到质量标准；
2. 这个副作用动作是否允许在此时、由此人执行。

Rubric 和 Human-in-the-loop 分别处理这两个问题，但它们都不能替代确定性校验与服务端授权。

1.1 四层控制各自检查什么

控制层	回答的问题	适合检查	不适合单独承担
确定性程序	数据是否满足明确约束	金额、Schema、Hash、测试结果	文风和语义完整性
Rubric	输出是否满足语义标准	证据说明、风险、表达质量	最终权限和副作用
HITL	人是否同意这次动作	发布、发送、删除、付款	自动判断内容绝对正确
Tool/业务后端	调用者是否真正有权限	租户、角色、额度、幂等	代替人工业务判断

数字一致性、文件存在性和 JSON Schema 应优先用程序断言。让 LLM Judge 重新计算一个可以由脚本确定的金额，不仅成本更高，也更不可靠。

1.2 推荐的严格发布链路

推荐顺序是：

Draft Agent
  -> Deterministic Validation
  -> Rubric Gate（仅 satisfied 放行）
  -> Publish Action Proposal
  -> HITL（仅 approve/edit 放行）
  -> Side-effect Tool
  -> Audit Log

这里的顺序必须由显式 LangGraph Workflow 或服务编排层保证。简单地在一个自由 Agent 上同时配置 RubricMiddleware 和 interrupt_on，并不能保证模型一定在 Rubric 成功之后才提出发布 Tool Call。

1.3 为什么 Middleware 并列不等于串行质量门

RubricMiddleware 在 Agent 自然准备结束、模型不再发出 Tool Call 时评分；HITL 则在受保护 Tool 即将执行时暂停。如果模型在第一次生成中直接发出 publish_report，HITL 会先看到发布提案，此时 Rubric 还没有对最终答案形成成功终态。

因此：

• Rubric 是输出质量循环，不是发布事务管理器；
• HITL 是 Tool Call 审批，不是内容质量评分器；
• 需要严格先后顺序时，把“写草稿”“评分”“提出发布”“执行发布”建成不同节点；
• 发布节点必须只接受明确的 rubric_status == satisfied Workflow State。

这是全文最重要的架构边界。

1.4 哪些动作需要人工审批

优先审批不可逆、外部可见或高成本动作：

• 对外发布报告、邮件和公告；
• 删除、覆盖、批量移动文件；
• 写入生产数据库；
• 付款、下单、退款；
• 修改权限、密钥和基础设施；
• 向外部系统上传敏感数据；
• 高成本 Sandbox 或云资源创建。

只读查询通常无需逐次审批，但仍要做后端授权、租户隔离和审计。审批粒度过细会产生疲劳，最终让审核人机械点击“同意”。

1.5 版本与 Beta 边界

本文使用的真实环境是：

Python 3.12.11
deepagents==0.6.12
langchain==1.3.13
langchain-core==1.4.9
langgraph==1.2.9
pydantic==2.13.4
langchain-openai==1.3.5

RubricMiddleware 在本文版本中仍属于 Beta。其下划线状态字段、自定义事件名、评分输入窗口、回调异常语义和循环实现都必须按版本看待，升级后重新执行契约测试。条件审批 when 需要 langchain>=1.3.3。

官方入口包括 Deep Agents Human-in-the-loop、Grading rubrics 和 LangGraph Interrupts。

2. 配置 Human-in-the-loop

2.1 HITL 在 Tool 执行前暂停

create_deep_agent(…, interrupt_on=…) 会把 HumanInTheLoopMiddleware 加入 Agent 栈。当模型生成受保护 Tool Call 时，Middleware 在执行器真正产生副作用前调用 LangGraph Interrupt：

Model -> Tool Call -> HITL Middleware -> Interrupt
                                      X Tool 尚未执行

暂停不是抛出一个普通异常后从头再跑，而是生成可恢复 Checkpoint。审核服务之后使用相同 Thread 和 Command(resume=…) 继续。

2.2 定义副作用 Tool

教学示例不连接外部发布系统，而是返回确定性字符串：

@tool
def publish_report(path: str, channel: str = "internal") -> str:
    """Publish an already reviewed report to a named channel."""

    return f"published:path={path};channel={channel}"

真实 Tool 仍要检查调用身份、租户、目标频道、报告 Hash、审批记录和幂等 Operation ID。HITL 只表示 Graph 收到一份 Decision，不等于后端自动获得授权。

2.3 interrupt_on 的三种配置

interrupt_on = {
    "read_public_report": False,
    "ask_user": True,
    "publish_report": {
        "allowed_decisions": ["approve", "edit", "reject"],
        "description": "发布会产生外部副作用，请核对路径和频道。",
    },
}

在本文锁定版本中：

配置	含义
False	该 Tool 自动通过 HITL 层
True	允许 approve、edit、reject、respond
Config 字典	显式定义 Decision、说明、条件等

False 只是关闭这层人工暂停，不会跳过 Tool 后端授权。

2.4 四种 Decision 的语义

Decision	Tool 是否执行	ToolMessage 语义
approve	执行原动作	真实 Tool 结果
edit	校验修改后执行	修改后 Tool 的真实结果
reject	不执行	status=”error” 的拒绝反馈
respond	不执行	status=”success” 的人工合成结果

respond 不是“更温和的 reject”。它适合 ask_user 一类由人充当 Tool 的交互；用它拒绝付款或发布，会让模型把人工消息理解为 Tool 已成功完成。

2.5 显式审批策略

示例只允许前三种 Decision：

PUBLISH_POLICY = {
    "publish_report": {
        "allowed_decisions": ["approve", "edit", "reject"],
        "description": "发布报告会产生外部副作用，请核对路径和发布频道。",
    }
}

按风险设置最小 Decision 集：高风险操作可能只允许 approve/reject；允许 edit 时，审批 API 必须重新验证修改后的完整 Action。

2.6 条件审批 when

当内部频道可以自动发布、公共频道必须人工确认时：

from langchain.agents.middleware import ToolCallRequest


def requires_public_review(request: ToolCallRequest) -> bool:
    return request.tool_call["args"].get("channel") == "public"


policy = {
    "publish_report": {
        "allowed_decisions": ["approve", "reject"],
        "when": requires_public_review,
    }
}

版本要求是 langchain>=1.3.3。Predicate 收到 ToolCallRequest，应从服务端可信 Runtime Context 或策略服务取得角色、租户和额度，不能把用户消息中的自我声明当成权限事实。

当 when 返回 False 时，该调用不会进入待审批批次，而是直接越过 HITL 层；Tool 后端授权仍照常执行。

2.7 动态审批描述的真实签名

动态 description 与 when 的签名不同。在 langchain==1.3.13 中，它接收三个参数：

from langchain.agents.middleware import AgentState
from langchain_core.messages.tool import ToolCall
from langgraph.runtime import Runtime


def describe_publish(
    tool_call: ToolCall,
    state: AgentState,
    runtime: Runtime,
) -> str:
    return (
        f"审批 {tool_call['name']}，目标频道="
        f"{tool_call['args'].get('channel')}，消息数={len(state['messages'])}"
    )

测试通过真实 Interrupt Payload 验证了 Tool Call、State 和 Runtime 三个参数均传入。这里是当前 LangChain API 契约，不应与 when(ToolCallRequest) 混写。

2.8 Filesystem Permission Interrupt

Deep Agents 的 Filesystem Permission 可以把某些路径操作设为 ask。写权限规则不仅影响 write_file 和 edit_file，也覆盖删除类操作。文件审批仍要考虑：

• 虚拟路径映射到哪个 Backend；
• 编辑后路径是否越界；
• 符号链接和路径规范化；
• SubAgent 是否继承权限；
• Sandbox 内外是否指向同一文件；
• 删除和覆盖能否恢复。

2.9 SubAgent 的审批边界

声明式 SubAgent 可以继承主 Agent 的 Tool-call HITL 配置，也可以提供自己的 interrupt_on。但这只覆盖通过该 SubAgent Agent Loop 发起、且真正暴露同名 Tool 的调用。

以下场景必须分别处理：

• CompiledSubAgent 已经拥有自己的编译图；
• Async SubAgent 在远程 Graph 中运行；
• Tool 内部绕过 Agent Graph 直接产生副作用；
• MCP/HTTP 服务端单独执行真实操作。

不能因为主 Agent 配置了 interrupt_on，就推断所有下游副作用都自动受保护。

3. Interrupt、Decision 与恢复

3.1 Checkpointer 和 Thread ID 都是必需条件

HITL 跨请求恢复明确需要 Checkpointer。示例 Builder 使用 is not None，避免 falsy 自定义 Saver 被静默替换：

def build_hitl_agent(model, *, checkpointer=None, interrupt_on=None):
    resolved_checkpointer = (
        checkpointer if checkpointer is not None else InMemorySaver()
    )
    resolved_policy = interrupt_on if interrupt_on is not None else PUBLISH_POLICY
    return create_deep_agent(
        model=model,
        tools=[publish_report],
        interrupt_on=resolved_policy,
        checkpointer=resolved_checkpointer,
    )

测试注入了 bool 返回 False 的合法 Saver，确认 HITL 与 Rubric Builder 都没有替换它。

每次业务流程还要分配稳定 Thread ID：

config = {"configurable": {"thread_id": "publish-20260706-001"}}

第一次暂停和后续 Resume 必须使用同一个 thread_id 和同一个持久存储域。

3.2 Checkpointer 为什么能恢复

Checkpointer 保存恢复 Graph 所需的 State、下一步任务、Interrupt 和相关执行元数据。具体物理字段、Channel Values、Task 结构与序列化格式取决于 LangGraph 版本和 Saver 实现。

不要把概念清单误当作固定数据库 Schema：消息可能已经摘要，Interrupt 可能位于 Task 中，Config 与 Metadata 的物理存储也随 Saver 不同。

InMemorySaver 适合单进程测试，不适合生产审批：进程重启、滚动部署和多实例路由都会让恢复失败。生产应使用持久 Saver，并加密敏感 State、设置保留期和访问控制。

3.3 第一次调用和 v2 GraphOutput

paused = agent.invoke(
    {"messages": [{"role": "user", "content": "发布报告"}]},
    config=config,
    version="v2",
)

在本文 langgraph==1.2.9 的 v2 调用中，通过 paused.interrupts 读取暂停信息，通过 paused.value 读取当前 Graph 输出。不要把 v1/v2 返回结构混在同一段代码中。

典型 Interrupt Payload 包含：

{
    "action_requests": [
        {
            "name": "publish_report",
            "args": {"path": "/reports/draft.md", "channel": "internal"},
            "description": "发布报告会产生外部副作用，请核对路径和发布频道。",
        }
    ],
    "review_configs": [
        {
            "action_name": "publish_report",
            "allowed_decisions": ["approve", "edit", "reject"],
        }
    ],
}

暂停时还没有出现 published: Tool 结果，这是副作用未执行的直接断言。

3.4 Approve：执行原动作

resumed = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    version="v2",
)

Approve 后，原 Tool Call 的名称、参数和 Tool Call ID 继续进入执行器。

3.5 Edit：服务端重新校验后执行

decision = {
    "type": "edit",
    "edited_action": {
        "name": "publish_report",
        "args": {"path": "/reports/final.md", "channel": "public"},
    },
}

前端禁用 Tool 名编辑只是用户体验，不是安全控制。审批 API 必须：

1. 验证 Tool 名与原 Action 一致；
2. 使用 Tool Schema 校验完整参数；
3. 重新执行角色、租户、目标和额度检查；
4. 判断修改是否扩大风险；
5. 对修改后的 Action 重新计算 Hash。

示例提供了真实校验函数：

def validate_publish_decision(action_request, decision) -> None:
    if decision.get("type") != "edit":
        return
    edited = decision.get("edited_action")
    if not isinstance(edited, Mapping):
        raise ValueError("edit decision requires edited_action")
    if edited.get("name") != action_request.get("name"):
        raise ValueError("edited tool name must match the reviewed action")
    args = edited.get("args")
    if not isinstance(args, Mapping):
        raise ValueError("edited action requires an args mapping")
    publish_report.get_input_schema().model_validate(dict(args))

3.6 Reject：不执行并返回错误语义

{"type": "reject", "message": "报告尚未经过法务审核，不要重试发布。"}

Reject 会跳过 Tool，生成 status=”error” 的 ToolMessage。若省略 message，当前实现会生成“未执行且除非用户再次要求，不要重试同一调用”的默认反馈。生产环境建议显式给出可操作、但不泄露内部安全策略的消息。

3.7 Respond：不执行但返回成功语义

{"type": "respond", "message": "由审核人直接提供结果。"}

测试确认 Tool 不执行，但合成 ToolMessage 的 status 是 success。这正是不能用 respond 表达拒绝的原因。

3.8 多动作批量审批不是事务

模型一次产生多个受保护 Tool Call 时，当前 HITL 会把它们批量放入 Interrupt，并要求 Decision 数量与顺序一一对应。

action_requests[0] <-> decisions[0]
action_requests[1] <-> decisions[1]

这不是数据库事务。第一个动作成功、第二个失败时，框架不会自动回滚第一个动作。相互依赖的高风险动作应该拆成显式 Workflow 节点，并设计补偿、重试和幂等语义。

3.9 自定义 interrupt() 的重放语义

Tool 或节点也可以直接调用 interrupt(payload)。恢复时节点会从头重放到 Interrupt 位置，因此 Interrupt 之前的代码必须可重放，副作用应放到恢复值之后。

同一节点内有多个 Interrupt 时，调用顺序必须保持一致；Resume 值按顺序匹配。不要根据不稳定条件改变 Interrupt 数量或顺序，否则恢复值可能对应到错误位置。

4. 审批安全与生产治理

4.1 审批 UI 最少展示什么

审核人至少要看到：

• Tool 名和业务动作名称；
• 原参数与修改后参数 Diff；
• Tool Schema 校验错误；
• 目标租户、环境和频道；
• 影响范围、成本与不可逆性；
• 报告摘要、证据链接和内容 Hash；
• 允许的 Decision；
• Thread、Tool Call、Operation ID；
• 审批截止时间和策略版本。

前端校验用于及时反馈，服务端必须重复校验。

4.2 审批过期与参数漂移

审批对象不是一句“允许发布”，而是某个确定 Action。记录至少包括：

tenant_id
operation_id
reviewed_action_hash
content_hash
environment / deployment
tool_version
backend_revision
policy_engine_version
schema_version
expires_at

Resume 前重新计算 Hash；内容、参数、环境、Tool 版本或策略变化时，旧审批失效并重新申请。

4.3 幂等与重复 Resume

thread_id + tool_call_id 可以关联 Graph 中的具体尝试，但不足以标识稳定业务操作。模型重新规划、浏览器重试或新 Thread 都可能产生新 Tool Call ID。

业务幂等键应以稳定 Operation ID 为核心：

tenant_id + operation_id + reviewed_action_hash

thread_id 和 tool_call_id 作为 Trace 关联字段保存。后端对相同幂等键返回既有结果，而不是再次执行副作用。

4.4 审批授权与职责分离

审批 API 必须独立验证：

• 当前登录者是否可审批该租户和动作；
• 提案人能否审批自己的高风险操作；
• 是否需要双人复核；
• Decision 是否在 allowed_decisions 中；
• 审批是否过期；
• 修改是否超出审批人权限。

双人复核应由 Workflow 和持久审批表实现。第二位审核人需要看到原 Action、第一 Decision、Hash、风险依据和当前状态，而不只是模型摘要。

4.5 审计与流事件不是同一种数据

Streaming Event 可能丢失、重复或因重连只看到后续部分，不能作为唯一审计源。持久审计表至少记录：

• 提案与 Decision；
• 审批人、时间和授权结果；
• 原参数与修改参数；
• Hash、版本与 Operation ID；
• Tool 执行结果和错误；
• Trace ID 与 Thread ID。

LangSmith Trace 适合可观测性，业务审计数据库负责合规和追责，两者职责不同。

4.6 Checkpointer 数据治理

Checkpoint 可能包含消息、Tool 参数、评分反馈和业务数据。生产 Saver 需要：

• 静态与传输加密；
• 租户隔离；
• 最小读取权限；
• 数据保留和删除策略；
• 备份与恢复验证；
• 多实例一致性；
• 版本升级迁移测试。

5. RubricMiddleware 质量循环（Beta）

RubricMiddleware 在本文版本中是 Beta。以下状态字段、事件名称和输入窗口都属于 deepagents==0.6.12 的版本实现，不应直接固化为长期公共协议。

5.1 Rubric 解决什么问题

Rubric 描述“什么才算完成”。示例将标准分成证据与表达两类：

REPORT_RUBRIC = """Every criterion must pass:
Evidence criteria:
1. State the data source, date range, and currency unit.
2. Explain the deterministic calculation or evidence path.
3. Keep every reported number consistent with verified evidence.

Communication criteria:
4. State at least one limitation or risk.
5. Present the conclusion clearly and concisely.
"""

生产项目还应让确定性校验先处理数字，Grader 主要检查证据说明与表达；分类 Rubric 只是降低混淆，不能替代脚本断言。

5.2 何时激活

在 0.6.12 Beta 实现中，RubricMiddleware 为 Agent State 增加 Rubric 相关字段，仅在调用 State 中存在非空 rubric 时启动评分。没有 Rubric 时，before/after Hook 保持 No-op。

result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "编写报告"}],
        "rubric": REPORT_RUBRIC,
    },
    config=config,
)

只把 Middleware 放进 Agent、却不在调用 State 传 rubric，不会发生评分。

5.3 主模型与内部 Grader Agent

RubricMiddleware 首次评分时懒加载一个专用 Grader Agent。它不是主 Agent 通过 task 委派的普通 Deep Agents SubAgent，不自动获得主 Agent 的：

• Backend；
• Skills；
• 业务 Tools；
• System Prompt；
• Permissions；
• 发布能力。

它只获得显式配置的 Grader Model、System Prompt 和可选验证 Tools，并通过结构化输出返回 GraderResponse。

使用不同模型可以降低部分相关错误，但不代表真正独立：模型可能共享训练数据、证据缺口和错误模式。关键事实仍需要确定性程序与外部证据。

5.4 创建 Rubric Agent 与 Checkpointer 边界

def build_rubric_agent(
    model,
    grader_model,
    *,
    max_iterations=2,
    on_evaluation=None,
    checkpointer=None,
):
    rubric = RubricMiddleware(
        model=grader_model,
        max_iterations=max_iterations,
        on_evaluation=on_evaluation,
    )
    resolved_checkpointer = (
        checkpointer if checkpointer is not None else InMemorySaver()
    )
    return create_deep_agent(
        model=model,
        middleware=[rubric],
        checkpointer=resolved_checkpointer,
    )

Rubric 在单次 Agent Run 内完成评分循环时不一定强制需要 Checkpointer。本文配置它，是为了读取 Beta 状态、跨调用观察和生产恢复；HITL 跨请求 Resume 则明确要求 Checkpointer。

功能	Checkpointer
HITL 跨请求恢复	必需
Rubric 单次循环	不一定
Rubric 跨调用观察	推荐
通过 Snapshot 读取 Private State	需要可访问状态

5.5 当前 GraderResponse Schema

在 0.6.12 中，结构化结果包含：

字段	类型	作用
result	satisfied / needs_revision / failed	Grader Verdict
explanation	string	简短结论
criteria	list	每条标准的 passed 与 gap

当前 Pydantic Validator 会拒绝明显矛盾的输出，例如 result=”satisfied”，但 criteria 中仍有 passed=False。这只能保证结构自洽，不能保证事实正确。

5.6 五种状态与产生方

状态	产生方	是否继续主模型
satisfied	Grader	否，成功结束
needs_revision	Grader	在预算内继续
failed	Grader	否，Rubric 不可评估
max_iterations_reached	Middleware	否
grader_error	Middleware	否

只有 needs_revision 且未达到次数上限时会注入修订反馈并跳回主模型。

5.7 Grader Feedback 会进入主模型上下文

当前实现把修订反馈作为带来源标记的 HumanMessage 注入：

name = rubric_grader
additional_kwargs["lc_source"] = rubric_grader

它会进入下一轮主模型上下文，因此反馈不能包含 Secret、内部评分 Prompt、审核人专用信息或冗长 Trace。来源标记用于观察和过滤，但不改变模型看到 HumanMessage 的事实。

5.8 max_iterations 是评分次数上限

在 0.6.12 中，iteration 以 Grader 调用为单位，从第一次评分开始计数，不是“额外重写次数”。

例如 max_iterations=1：

1. 主模型生成初稿；
2. Grader 第一次评分；
3. 若结果为 needs_revision，预算已经耗尽；
4. 状态变成 max_iterations_reached，不再进入主模型修订。

当前构造函数只接受 1 到 20 的整数，并拒绝 bool。Rubric 仍是 Beta，未来命名或循环策略可能调整。

5.9 非成功终态不会自动替换最后答案

当状态是 failed、max_iterations_reached 或 grader_error 时，当前 Middleware 保留主模型最后一个 AIMessage，不会自动改成“质量检查失败”。

如果发布层只读取最后回答，不检查 Rubric 状态，不合格草稿仍可能被发布。因此 Workflow Gate 必须显式要求 status 为 satisfied。

5.10 用适配层读取 Private State

_rubric_status、_rubric_iterations、_rubric_evaluations、_current_grading_run_id 等下划线字段是 Beta 内部状态。业务代码不应散落硬编码，示例集中封装：

@dataclass(frozen=True)
class RubricStatusView:
    status: str | None
    iterations: int
    evaluation_count: int
    grading_run_id: str | None


def read_rubric_status(agent, config) -> RubricStatusView:
    values = agent.get_state(config).values
    evaluations = values.get("_rubric_evaluations", [])
    return RubricStatusView(
        status=values.get("_rubric_status"),
        iterations=int(values.get("_rubric_iterations", 0) or 0),
        evaluation_count=len(evaluations),
        grading_run_id=values.get("_current_grading_run_id"),
    )

升级时只需重点修改适配层并运行契约测试。强制发布控制更适合使用自己的稳定 Workflow State，而不是让外部系统直接依赖下划线字段。

6. Grader 的安全、证据、观察与成本

6.1 Custom Event 是 Beta 观察接口

当前 0.6.12 会通过 stream_mode=”custom” 发送：

rubric_evaluation_start
rubric_evaluation_end

使用 v2 流格式时，外层事件还有 type、namespace 和 data 结构。事件名与 Payload 都是 Beta 版本契约，后端应转换成自己的稳定事件模型，前端不要永久硬编码内部名称。

Streaming 不是持久审计：断线、重连、重复消费和回放行为取决于传输与 Checkpointer。

6.2 Callback 只能观察

on_evaluation 可以接收每次结构化 RubricEvaluation。本文源码和测试确认：在 0.6.12 中 Callback 抛出的普通 Exception 会记录日志并被抑制，不改变 Agent 控制流。

这是一条 Beta 实现语义，不是安全保证。Callback 不得承担“失败就禁止发布”的强制职责；它适合写 Metrics、Trace 或异步观察，发布 Gate 由 Workflow State 控制。

6.3 当前评分输入窗口

0.6.12 内部实现对 Grader 输入做了成本限制：

• 主要保留最近 30 条消息；
• 若最初真实 HumanMessage 不在窗口中，会额外保留；
• 单条文本最多保留 4000 字符，并追加截断标记；
• Middleware 自己注入的评分反馈不会被误认成原始用户请求。

这些数值和选择算法是 Beta 实现，不是 Rubric 概念的长期保证。大 Tool 结果仍应写入 Backend，向 Grader 提供摘要、路径和 Hash，而不是依赖自动截断。

6.4 Prompt Injection 缓解不是安全证明

当前实现使用随机边界标记、转义部分伪造关闭标签，并通过结构化输出约束格式。这些措施能降低边界混淆和输出漂移，但不能证明 Grader 不受恶意消息语义影响。

需要额外措施：

• 把所有消息和 Tool 输出视为不可信证据；
• Rubric 来自服务端可信配置；
• 对抗测试覆盖间接 Prompt Injection；
• 关键事实由程序或只读证据 Tool 验证；
• Grader 不持有发布、删除和付款 Tool；
• 高风险结论仍由人审核。

6.5 Grader Tool 必须最小权限

允许 Grader读取文件或运行测试时，不仅要“只读”，还要限制：

• 可访问路径；
• Tenant Namespace；
• 查询范围和返回行数；
• 网络域名和协议；
• 执行时间与资源；
• 返回证据 Hash；
• Tool Call 审计。

Grader 不应拥有 publish_report，否则质量评估器本身也可能产生副作用。

6.6 成本模型

一次“初稿不合格、修订后通过”的最短路径通常包含：

主模型初稿
Grader 1
主模型修订
Grader 2

实际成本还可能增加：

• 主模型 Tool Calling；
• Grader 验证 Tool；
• 结构化输出修复或重试；
• Provider 网络重试；
• 长消息历史；
• 多模态输入。

生产指标要按模型调用记录输入/输出 Token、延迟、错误和 Grading Run ID，不能只记录整个流程总数。

6.7 评分历史的保留策略

_rubric_evaluations 会累积评分记录。长期 Thread 如果无限评分，会导致 Checkpoint 增大、隐私数据长期保留和读取变慢。

可以只在 State 保留最近 N 次摘要，把完整 Evaluation 写入受控外部存储；外部记录按租户、用途和法规设置保留期，并避免保存不必要的完整消息内容。

7. 测试、排错与上线验收

7.1 为什么使用确定性模型

HITL 与 Rubric 的回归测试关注结构行为：是否暂停、Tool 是否执行、Decision 如何转成 ToolMessage、评分循环走几次、终态是什么。随机在线模型不适合稳定断言这些契约。

ScriptedChatModel 返回预先构造的真实 AIMessage 和 Tool Call，仍经过完整 Deep Agents/LangGraph 执行栈。在线测试则补充真实 Provider 的 Tool Calling、结构化输出、延迟和 Token 验证。

7.2 24 个 pytest Test Item

当前测试覆盖：

• Interrupt Payload 与审批策略；
• approve 原样执行；
• edit 参数执行与 Tool Call ID 保留；
• 服务端拒绝修改 Tool 名和非法 Schema；
• reject 生成 error ToolMessage；
• respond 生成 success ToolMessage 且无副作用；
• Decision 数量与批次顺序；
• 条件审批自动放行内部频道；
• 动态 description 三参数签名；
• True 的四种 Decision；
• falsy Checkpointer 依赖注入；
• Rubric needs_revision 到 satisfied；
• 没有 rubric 时 No-op；
• max_iterations_reached；
• grader_error；
• Callback 异常抑制的 Beta 契约；
• 消息截断和边界转义；
• start/end Custom Event；
• GraderResponse 自洽校验；
• 0、21、True 三个非法迭代参数；
• 发布 Tool 确定性契约。

参数化用例使上述验收点形成 24 个实际 Test Item。

运行：

cd source/_posts/deepagent/examples
.venv_deepagent/bin/ruff check p11_hitl_rubric
.venv_deepagent/bin/python -m compileall -q p11_hitl_rubric
.venv_deepagent/bin/pytest -o addopts='' -q p11_hitl_rubric/test_quality.py

结果：

All checks passed!
........................                                                 [100%]
24 passed, 1 warning

Warning 是 RubricMiddleware 的官方 Beta 警告，文章保留它作为版本证据，而不是隐藏。

7.3 确定性演示结果

approve: action=publish_report args={'path': '/reports/draft.md', 'channel': 'internal'} tool_status=success result=published:path=/reports/draft.md;channel=internal
edit: action=publish_report args={'path': '/reports/draft.md', 'channel': 'internal'} tool_status=success result=published:path=/reports/final.md;channel=public
reject: action=publish_report args={'path': '/reports/draft.md', 'channel': 'internal'} tool_status=error result=报告尚未经过法务审核。
rubric_verdicts=['needs_revision', 'satisfied']
final_report=数据来源：sales.csv；范围：2026-06-01 至 2026-06-30；单位：人民币元。确定性脚本 sum(amount) 得到收入 100。限制：样本只覆盖已入账订单，未包含退款延迟。

7.4 常见问题

没有出现 Interrupt

检查 Tool 名是否与 interrupt_on 一致、是否配置 Checkpointer、when 是否返回 False、模型是否真的生成目标 Tool Call，以及调用路径是否绕过该 Agent Graph。

无法 Resume

检查是否使用同一 Thread ID、同一 Saver 和同一部署数据域；Decision 数量、顺序是否匹配 Action；Checkpoint 是否已过期或在滚动部署中使用了不兼容 Graph。

Rubric 看似启用却没评分

检查调用 State 是否传入非空 rubric、主模型是否自然停止、Grader 是否支持结构化输出，以及是否把 Beta Warning 当成普通日志忽略。

非成功报告仍被发布

检查是否只读取最后 AIMessage，是否没有经过 read_rubric_status/稳定 Workflow State，或者把 Rubric 与 HITL 并列后错误假设存在先后顺序。

7.5 在线验收及边界

保存了一次 OpenAI-compatible 第三方网关验收结果：

项目	值
模型 ID	gpt-5.5，网关别名
Grader ID	gpt-5.5，网关别名
deepagents	0.6.12
langchain	1.3.13
langgraph	1.2.9
langchain-openai	1.3.5
Rubric	Beta

保存的脱敏结果为：

{
  "scene": "p11_hitl_and_rubric",
  "status": "passed",
  "model": "gpt-5.5",
  "grader_model": "gpt-5.5",
  "latency_ms": 48764,
  "interrupt_tool": "publish_report",
  "side_effect_before_approval": false,
  "resume_tool_result": "published:path=/reports/final.md;channel=public",
  "rubric_verdicts": ["needs_revision", "satisfied"],
  "rubric_status": "satisfied",
  "usage": {"input_tokens": 30405, "output_tokens": 907, "total_tokens": 31312}
}

该运行只验证了锁定端点、模型别名和版本组合下出现预期 Tool Call、Interrupt、Resume 与 Rubric 循环，不代表所有输入、网络重试和 Provider 版本都具有同样行为。名称来自测试网关，不证明存在同名原厂公开模型或固定底层权重。

保存的结果只包含总延迟与总 Token，无法可靠拆分为每次模型调用，不能补造分阶段数字。live_demo.py 支持逐调用 Trace，分别记录 HITL 提案、Resume 和 Rubric Loop 中每次模型调用的模型 ID、延迟与 Usage。

7.6 生产检查表

HITL

• Checkpointer 持久化且经过重启恢复验证；
• Resume 使用同一 Thread；
• Decision 与 Action 数量、顺序严格匹配；
• respond 不用于拒绝副作用；
• Edit 由服务端重新校验 Schema、名称和权限。

Rubric

• Beta 版本固定并有契约测试；
• rubric 来自可信配置；
• 只有 satisfied 才进入发布节点；
• 非成功终态不依赖最后 AIMessage；
• Private State 通过适配层读取；
• Grader Tool 最小权限且无副作用。

Workflow 与安全

• 确定性校验、Rubric、发布提案、HITL 和 Tool 是显式节点；
• Operation ID 和 Action Hash 实现业务幂等；
• 审批授权、过期和职责分离在服务端执行；
• Checkpoint 与评分历史有加密和保留策略。

Observability

• 每次模型调用记录 Token、延迟和模型 ID；
• Streaming Event 不作为唯一审计；
• Callback 不承担发布安全门；
• Trace、业务审计和 Metrics 分层存储。

7.7 示例文件

source/_posts/deepagent/examples/p11_hitl_rubric/
├── quality_agent.py   # HITL、Rubric Builder、Edit 校验和状态适配层
├── scripted_model.py  # 确定性 Tool Calling 与 Grader 模型
├── run_demo.py        # approve/edit/reject 和修订循环
├── live_demo.py       # 在线 HITL、Rubric 与逐模型调用 Trace
├── live_result.json   # 已保存的历史脱敏结果
└── test_quality.py    # 24 个 pytest Test Item

8. 总结

Human-in-the-loop 和 RubricMiddleware 解决的是两种不同风险：HITL 决定“这次副作用是否获准执行”，Rubric 判断“当前输出是否满足语义质量标准”。确定性程序负责可计算事实，Tool 后端负责真正授权和幂等。

HITL 的可靠恢复依赖 Checkpointer、稳定 Thread ID 和正确的 Decision 顺序。Approve、Edit、Reject、Respond 有不同 ToolMessage 语义，其中 Respond 表示成功的人工结果，不能拿来拒绝副作用。

RubricMiddleware 在本文版本仍是 Beta。内部状态、事件、输入窗口、Callback 行为和迭代语义必须经过版本契约测试。非成功终态不会自动替换最后回答，所以发布 Gate 必须显式检查 satisfied。

最后，Rubric 与 HITL 并列配置不自动产生严格顺序。真正可靠的发布系统应使用显式 Workflow：确定性验证通过后评分，评分成功后才允许提出发布，人工批准后才执行 Tool，并把 Operation ID、Hash、Decision 和执行结果写入持久审计。这样，质量控制才从 Prompt 中的约定变成可恢复、可测试、可审计的工程流程。