Agno 人工审核与暂停/恢复机制

Agno 人工审核与暂停/恢复机制

学习目标

本章将分析 Agno 的人工审核系统：

• @approval decorator 标记需要审批的工具
• ApprovalType 审批类型系统
• Run 暂停/恢复的完整流程
• 流式和非流式两种暂停处理

前置知识

建议先阅读：

• Human-in-the-loop 设计 — 人工介入的三种模式
• Intervention Gateway — 安全干预网关

下文假设你已理解上述概念，直接聚焦 Agno 的具体实现。

---

项目实践

@approval 装饰器

Agno 使用 @approval decorator 标记需要人工审批的工具：

# 伪代码：@approval decorator
from agno.approval import approval, ApprovalType

@approval(type=ApprovalType.REQUIRED)
def delete_production_record(db: str, record_id: str) -> str:
    """删除生产数据库中的记录。"""
    ...

ApprovalType 定义审批的严格程度：

类型	行为	适用场景
REQUIRED	必须人工确认才能执行	数据删除、配置修改
OPTIONAL	可以跳过，但建议审批	敏感数据读取
BLOCKED	完全禁止，除非管理员解除	高风险操作

暂停/恢复完整流程

当 Agent 执行到带 @approval 标记的工具时，触发暂停流程：

关键步骤：

1. 检测暂停：if any(tool_call.is_paused for tool_call in run_response.tools or [])
2. 存储审批 ID：create_approval_from_pause(db=agent.db, run_response=run_response, ...)
3. 返回暂停状态：run_response.status = RunStatus.paused
4. 等待确认：AgentOS 或调用方提供审批界面
5. 恢复执行：用户确认后，从暂停点继续

流式暂停处理

流式响应的暂停需要额外处理：

# 伪代码：流式暂停
def handle_agent_run_paused_stream(agent, run_response, session, ...):
    run_response.status = RunStatus.paused
    create_approval_from_pause(...)

    # 创建暂停事件
    pause_event = handle_event(
        create_run_paused_event(
            from_run_response=run_response,
            tools=run_response.tools,
            requirements=run_response.requirements,
        ),
        run_response,
        events_to_skip=agent.events_to_skip,
        store_events=agent.store_events,
    )

    yield pause_event
    if yield_run_output:
        yield run_response

流式暂停会：

• 创建 RunPausedEvent 并加入事件流
• 如果 yield_run_output=True，同时产出 RunOutput 供调用方捕获
• 清理并存储当前状态

审批 ID 的持久化

审批 ID 存储在数据库中，与 run_id 和 tool_call 关联：

# 伪代码：审批记录
class ApprovalRecord:
    approval_id: str          # 唯一审批标识
    run_id: str               # 关联的 run
    tool_call_id: str         # 触发审批的工具调用
    status: str               # "pending" / "approved" / "rejected"
    user_id: str              # 审批操作人
    created_at: datetime      # 创建时间
    completed_at: datetime    # 审批完成时间

这使得：

• 审批状态在服务器重启后不丢失
• 可以审计历史审批记录
• 支持多用户审批流程（如需要多人确认）

问题与规避

1. 暂停后工具状态不一致

暂停时，部分工具可能已经执行了一半（如写入了数据库但未完成）。

规避：is_paused 在工具执行之前检测。如果工具已经开始执行，不会在中间暂停。审批只阻止尚未开始的工具调用。

2. 长时间等待的资源泄漏

用户可能在暂停后很长时间才审批，期间 session 和其他资源保持占用。

规避：

• 暂停时清理不需要的资源（cleanup_and_store）
• 审批超时设置（在 AgentOS 层实现）
• 使用 store_events 持久化事件，重启后可恢复

3. 多人审批流程

当前实现是单人审批。如果需要多人确认（如关键操作需要两人审批），需要自定义审批界面。

规避：使用 ApprovalRecord 的扩展字段记录多人的审批状态，在恢复执行前检查所有必要条件。

设计取舍

为什么使用装饰器而非配置？

Agno 选择 @approval decorator 而非 Agent 配置字典：

• 优势：在工具定义处声明审批需求，代码即文档
• 代价：审批策略与工具代码耦合，不能在不修改代码的情况下改变
• 替代方案：Agent 配置中的审批列表——可以运行时修改，但与工具定义分离，容易遗漏

为什么暂停时存储审批 ID？

优势：审批状态持久化，支持服务器重启、多实例部署 代价：需要数据库支持 替代方案：内存中存储审批状态——实现简单但不可靠

参考来源

• Agno approval/ 源码
• Agno agent/_run.py 暂停处理