Agno 异步双版本方法模式

Agno 异步双版本方法模式

学习目标

本章将分析 Agno 的异步双版本模式：

• 为什么每个公共方法都同时提供 sync 和 async 变体
• 后台任务管理防止 GC 的模式
• 依赖解析的双版本实现
• 流式响应的双版本设计

项目实践

双版本方法列表

Agno 的所有公共方法都有 sync/async 双版本：

Sync 方法	Async 方法	功能
run()	arun()	执行 Agent
print_response()	aprint_response()	执行并打印响应
stream()	astream()	流式执行

核心实现文件 _run.py 中：

• Sync 路径使用 call_model_with_fallback
• Async 路径使用 acall_model_with_fallback
• 依赖解析有双版本：resolve_run_dependencies / aresolve_run_dependencies
• 暂停处理有四个变体：handle_agent_run_paused / handle_agent_run_paused_stream / ahandle_agent_run_paused / ahandle_agent_run_paused_stream

依赖解析的双版本

# 伪代码：Sync 版本
def resolve_run_dependencies(agent: Agent, run_context: RunContext) -> None:
    for key, value in run_context.dependencies.items():
        if callable(value):
            result = value(**kwargs)  # 直接调用
            run_context.dependencies[key] = result

# 伪代码：Async 版本
async def aresolve_run_dependencies(agent: Agent, run_context: RunContext) -> None:
    for key, value in run_context.dependencies.items():
        if callable(value):
            result = value(**kwargs)
            if iscoroutine(result):
                result = await result  # await 协程
            run_context.dependencies[key] = result

关键区别：async 版本检查返回值是否为协程（iscoroutine），如果是则 await。

后台任务管理

Agno 使用全局集合保持后台任务的强引用，防止被垃圾回收：

# _run.py 顶层
_background_tasks: set[asyncio.Task[None]] = set()

# 创建后台任务
def _create_background_task(coro):
    task = asyncio.create_task(coro)
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)  # 完成后自动移除

为什么需要强引用？ Python 的 asyncio 不会保持对 create_task 返回的 Task 对象的引用。如果 Task 在运行前被 GC，任务不会执行。

流式响应的双版本

# 伪代码：Sync 流式
def stream(...) -> Iterator[RunOutputEvent]:
    for event in events:
        yield event

# 伪代码：Async 流式
async def astream(...) -> AsyncIterator[RunOutputEvent]:
    async for event in events:
        yield event

问题与规避

1. 双版本代码同步

sync 和 async 版本的逻辑几乎相同，但手动维护两份代码容易不同步。

规避：

• 将公共逻辑提取到独立函数（两个版本都调用）
• 使用类型注解和签名检查确保一致性
• Code review 时对比两个版本

2. 混合 sync/async 调用

在 async 上下文中调用 sync 方法，或在 sync 上下文中调用 async 方法。

规避：

• sync 方法中检测到协程依赖时，发 warning 并跳过
• log_warning(f"Dependency {key} is a coroutine. Use agent.arun() instead.")

3. 后台任务的异常处理

后台任务的异常如果不被捕获，会导致 “Task exception was never retrieved” 警告。

规避：

• 后台任务使用 try/except 包裹
• 异常结果在 wait_for_open_threads 中合并到主响应的 metrics
• _background_tasks.discard 回调确保任务完成后清理

设计取舍

为什么维护双版本而非只提供 async？

优势：兼容非 async 用户——不是所有项目都使用 asyncio 代价：维护成本翻倍 替代方案：只提供 async，sync 用户用 asyncio.run() 包装——但体验不如原生 sync

为什么后台任务用全局集合？

优势：简单有效，确保任务不被 GC 代价：全局状态，多实例可能冲突 替代方案：Agent 实例级别的集合——更隔离但可能漏掉跨 Agent 的任务

参考来源

• Agno agent/_run.py 源码