跳转到内容
AI Agent 教程 AI Agent 教程

Agno 取消机制与生命周期管理

Agno 取消机制与生命周期管理

学习目标

本章将分析 Agno 的运行取消和生命周期管理：

  • 全局运行注册表（register_run / raise_if_cancelled / cancel_run）
  • 管线中 5 个取消检查点的插入策略
  • 工作流级别的成员 run 清理
  • 后台任务清理与资源回收

项目实践

全局运行注册表

Agno 使用全局集合管理所有运行中的 run：

# 伪代码：全局注册表
_running_runs: Dict[str, RunStatus] = {}


def register_run(run_id: str) -> None:
    """注册一个新 run 到全局集合"""
    _running_runs[run_id] = RunStatus.RUNNING


def cancel_run(run_id: str) -> None:
    """标记一个 run 为取消状态"""
    _running_runs[run_id] = RunStatus.CANCELLED


def raise_if_cancelled(run_id: str) -> None:
    """如果 run 被取消，抛出 RunCancelledException"""
    if _running_runs.get(run_id) == RunStatus.CANCELLED:
        raise RunCancelledException(f"Run {run_id} was cancelled")


def cleanup_run(run_id: str) -> None:
    """从注册表中移除 run"""
    _running_runs.pop(run_id, None)

管线中的取消检查点

取消检查点插入在管线的关键节点之间：

# 伪代码：_run.py 中的取消检查点
def _run(agent, run_response, run_context, ...):
    register_run(run_context.run_id)


    try:
        # 步骤 1-2: Session 创建与 State 更新
        agent_session = read_or_create_session(...)
        update_metadata(...)


        # 步骤 3: 依赖解析
        resolve_run_dependencies(...)
        raise_if_cancelled(run_response.run_id)  # 检查点 1


        # 步骤 4-5: 预钩子与工具确定
        execute_pre_hooks(...)
        _tools = determine_tools_for_model(...)


        # 步骤 6: 消息准备
        run_messages = get_run_messages(...)
        raise_if_cancelled(run_response.run_id)  # 检查点 2


        # 步骤 7: 后台线程启动
        memory_future = start_memory_future(...)
        learning_future = start_learning_future(...)
        raise_if_cancelled(run_response.run_id)  # 检查点 3


        # 步骤 8-9: 推理与模型调用
        handle_reasoning(...)
        raise_if_cancelled(run_response.run_id)  # 检查点 4
        model_response = call_model_with_fallback(...)
        raise_if_cancelled(run_response.run_id)  # 检查点 5


        # 步骤 10-13: 输出转换、后钩子、存储
        convert_response_to_structured_format(...)
        execute_post_hooks(...)
        cleanup_and_store(...)


    finally:
        cleanup_run(run_context.run_id)

工作流级别的成员 run 清理

工作流包含多个 Agent/Team 的 run，取消工作流需要清理所有成员 run：

# 伪代码：工作流成员清理
def cleanup_member_runs(workflow_id: str) -> None:
    """清理工作流下所有成员 run"""
    member_ids = get_member_run_ids(workflow_id)
    for run_id in member_ids:
        cleanup_run(run_id)


def cancel_member_runs(workflow_id: str) -> None:
    """取消工作流下所有成员 run"""
    member_ids = get_member_run_ids(workflow_id)
    for run_id in member_ids:
        cancel_run(run_id)

Async 版本

所有取消操作都有 async 版本：

# 伪代码
async def acancel_run(run_id: str) -> None: ...
async def acleanup_run(run_id: str) -> None: ...
async def araise_if_cancelled(run_id: str) -> None: ...
async def aregister_run(run_id: str) -> None: ...

取消事件的传播

取消时，通过事件系统通知调用方：

# 伪代码：取消事件
event = create_run_cancelled_event(
    from_run_response=run_response,
    reason="User requested cancellation",
)
yield event

问题与规避

1. 检查点之间的操作不可中断

取消检查点之间的操作（如模型调用）无法中断。如果模型调用耗时 30s，取消请求可能要等 30s 才生效。

规避：

  • 使用模型提供商的超时设置
  • 在模型调用前后都设置检查点，确保调用完成后立即退出
  • 对于长时间运行的工具调用，工具内部应该有取消检查

2. 全局注册表的线程安全

_running_runs 是全局字典，在多线程/异步环境中可能不安全。

规避：

  • Python 的 dict 操作是原子的（GIL 保护）
  • 但在高并发场景中，建议使用 asyncio.Lock 或 threading.Lock

3. 取消后资源泄漏

如果取消发生在资源分配之后、释放之前，可能导致泄漏。

规避：

  • finally 块中的 cleanup_run 确保注册表清理
  • 后台任务通过 _background_tasks 强引用管理
  • 数据库 session 在 cleanup_and_store 中统一清理

设计取舍

为什么用全局注册表而非实例级状态？

优势：可以从任何地方取消任何 run（包括其他线程/进程） 代价：全局状态，测试时需要清理 替代方案：Agent 实例级别的取消状态——更隔离但不能跨实例取消

为什么取消不立即停止执行？

优势：在检查点之间完成当前操作，确保不留下不一致的状态 代价：取消有延迟 替代方案：立即抛出异常——更快但可能留下脏状态

参考来源

版本信息

📅 2026-05-30 🏷️ 1444c9b 🔀 1444c9b

标签

cancellationlifecyclerun-registrycleanup