Agno Agent 核心循环与执行管线

Agno Agent 核心循环与执行管线

学习目标

本章将深入分析 Agno 的 _run.py 中实现的 16 步确定性执行管线，理解：

• 为什么将执行拆分为 16 个明确步骤而非单体循环
• 后台并行线程如何与主执行管线协同
• 取消检查点的插入策略
• 重试逻辑为什么选择”完整重走管线”而非”局部重试”

前置知识

建议先阅读：

• Agent 核心循环设计 — Session/Turn 两层循环模型与事件驱动架构
• 事件驱动的 Agent 运行时 — 发布/订阅消息路由

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

---

项目实践

16 步执行管线

Agno 的核心执行函数 _run() 是一个确定性的 16 步管线：

每一步有明确的输入和输出：

步骤	函数	输入	输出
1	read_or_create_session	session_id, user_id	AgentSession
2	update_metadata	agent, session	更新后的 metadata
3	resolve_run_dependencies	run_context.dependencies	解析后的依赖值
4	execute_pre_hooks	agent.pre_hooks, run_input	可能修改的 run_input
5a	agent.get_tools	run_response, run_context	处理后的工具列表
5b	determine_tools_for_model	处理后的工具, model	发送给模型的工具定义
6	get_run_messages	所有上下文源	RunMessages（消息列表）
7	start_memory/learning/culture_future	run_messages	后台任务 Future
8	handle_reasoning	run_messages, run_context	推理结果注入消息
9	call_model_with_fallback	messages, tools	ModelResponse
10	convert_response_to_structured_format	model_response	结构化输出
11	execute_post_hooks	agent.post_hooks, run_output	可能修改的 run_output
12	wait_for_open_threads	memory/learning/culture future	等待完成
13	cleanup_and_store	run_response, session	持久化

后台并行线程模式

管线步骤 7 是关键创新：在模型调用之前启动三个后台线程：

# 伪代码：后台线程启动
memory_future = start_memory_future(agent, run_messages, user_id)
learning_future = start_learning_future(agent, run_messages, session, user_id)
cultural_future = start_cultural_knowledge_future(agent, run_messages)

# 这三个线程与主管线并行运行，不阻塞模型调用
model_response = call_model_with_fallback(...)  # ← 立即执行，不等后台

在管线末尾（步骤 12）等待后台完成：

# 伪代码：等待后台
wait_for_open_threads(memory_future, cultural_future, learning_future)

为什么在模型调用前启动？ 记忆提取和学习提取需要完整的 run_messages（包含用户输入和系统提示），但不需要模型响应。因此可以在等待模型调用的同时，并行执行记忆和学习提取。

收益：假设模型调用耗时 3s，记忆提取耗时 2s。如果串行执行总耗时 5s。并行执行总耗时 3s（取最大值）。

取消检查点

管线在 5 个位置插入 raise_if_cancelled(run_id)：

# 伪代码：取消检查点
# 步骤 3 后
raise_if_cancelled(run_id)

# 步骤 6 后
raise_if_cancelled(run_id)

# 步骤 7 后
raise_if_cancelled(run_id)

# 步骤 9 前
raise_if_cancelled(run_id)

# 步骤 9 后
raise_if_cancelled(run_id)

这确保了：

• 用户取消后，最多再执行一个步骤就退出
• 不在取消后继续存储或执行钩子（避免资源浪费）
• 模型调用前后的检查确保不会”调用完后发现被取消”

问题与规避

1. 后台线程异常未被传播

如果记忆提取线程抛出异常，主管线不会立即感知。异常被 Future 捕获，在步骤 12 等待时才会暴露。如果后台线程失败，用户响应已生成，但记忆未更新——这是一个”静默失败”场景。

规避：Agno 通过 _background_tasks: set[asyncio.Task[None]] 保持强引用，防止后台任务被 GC。异常在 wait_for_open_threads 中被处理，记录日志但不中断主响应。

2. 重试重走管线的性能开销

每次重试完整重走 16 步（包括重新创建 session、重新解析依赖）。对于依赖解析耗时长的场景（如数据库查询），重试成本高。

设计取舍：一致性优于性能——每次重试都从干净的起点开始，避免”部分状态已修改”的脏重试。如果性能是关键指标，建议在步骤 3 之前缓存依赖解析结果。

3. 指数退避与重试次数的平衡

默认 retries=0（不重试）。开启后：

• delay_between_retries=1（秒）
• exponential_backoff=False（线性延迟）

陷阱：如果 exponential_backoff=True 且 retries=5，最大延迟为 16s，总重试时间可达 31s。用户可能早已离开。

设计取舍

为什么是 16 步而不是更少？

Agno 的选择是显式步骤优于隐式耦合。每一步可以独立替换（通过继承/钩子），且每一步前后的取消检查提供精确的控制粒度。

替代方案：单体循环（如简单的 while 循环 + if/else 分支）。优点是代码简洁，缺点是每个功能点都需要修改循环核心，难以扩展。

为什么后台线程使用 Future 而非 await？

因为记忆提取和学习提取是”非关键的后台工作”——它们的结果不影响当前响应的正确性。使用 Future 允许它们在后台线程池中运行，不阻塞事件循环。

替代方案：完全串行（等记忆提取完成再调用模型）。这会导致用户感知延迟增加，但实现更简单、错误处理更直接。

参考来源

• Agno Agent _run.py 源码