深入理解 Agno 多层次、高度模块化的 AI Agent 框架设计
Agno 采用分层架构,从编排层到基础设施层,每层之间通过清晰的抽象接口连接。
Agno 的三个顶层抽象,从宏观编排到微观执行的层次关系。
Agent 将核心功能拆分到多个子模块中,通过委托模式实现关注点分离。
从用户输入到最终输出,一次完整调用经历的 14 个步骤。
session_id 从 BaseDb 加载 AgentSession,不存在则创建新会话。session_state(用户数据)和 extra_data,更新元信息。cache_callables=True。search_knowledge、read_chat_history)和 Knowledge 暴露的工具。system_message(含指令 + 工具说明 + 知识上下文) → history_messages(从历史 runs 重建) → user_message(用户输入 + RAG 检索结果 + 多模态媒体)ThreadPoolExecutor 并行启动:Memory 更新、Learning 提取、Cultural Knowledge 更新。reasoning=True,执行多步 Chain-of-Thought 推理,或调用原生推理模型(DeepSeek / Claude)。生成 ReasoningStep 序列。Model._invoke_with_retry() → HTTP APItool_calls → Function.execute() → 追加 tool 消息 → 再次调用模型output_model(二次 LLM 结构化)或 parser_model(解析模型),最后通过 Pydantic 验证转为 output_schema。SessionSummaryManager 生成会话摘要,用于长对话的上下文压缩。AgentSession.upsert_run() → BaseDb.upsert_session() 持久化会话。返回 RunOutput(或流式 Iterator[RunOutputEvent])。Agno 的可插拔存储架构:每个抽象接口都有多种实现可供选择。
Agno 框架中的关键模块,点击卡片查看更多细节。
@dataclass(init=False) 装饰,使用「模块委托」模式将功能拆分到 12 个子模块。支持同步/异步、流式/非流式四种调用方式。一次 run() 经历 14 个步骤完成。
Coordinate(领导者分配)、Route(路由到专家)、Broadcast(广播)、Tasks(自主任务分解)。支持嵌套 Team。
Step、Steps、Loop、Parallel、Condition、Router。
invoke / ainvoke / invoke_stream / ainvoke_stream 四种核心方法。内置自动重试。已有 30+ 提供者实现。
Toolkit 是工具集容器,Function 是单个工具的元数据和执行入口。支持 HITL(需确认/需用户输入/外部执行)。内置 70+ 工具覆盖搜索、代码执行、数据库、媒体等。
KnowledgeProtocol 协议接口实现可插拔。Knowledge 类集成 VectorDb + Embedder + Document + Reader。支持远程加载(S3、GCS、SharePoint、GitHub)。
UserMemory,存入数据库,下次 run 时注入上下文。支持可插拔优化策略。
LearningMachine 协调 6 种学习存储:UserProfile、UserMemory、SessionContext、EntityMemory、LearnedKnowledge、DecisionLog。支持命名空间隔离和自定义存储。
ReasoningStep 逐步推理)和原生推理(委托给 DeepSeek/Claude 等原生推理模型)。可配置最小/最大推理步数。
BaseDb / AsyncBaseDb 管理 10+ 数据表。推荐生产使用 PostgreSQL,开发用 SQLite。已有 10+ 数据库实现。
RunContext 贯穿整个运行过程,携带 session_state、依赖和工具。RunOutput 封装完整输出。RunEvent 定义 28 种事件类型用于流式传输。
BaseGuardrail 基类实现 pre/post 两个阶段的安全检查。支持输入验证、输出审查和自定义逻辑。与 Evals 评估系统集成。