← 返回教程

Agno 框架架构全景图

深入理解 Agno 多层次、高度模块化的 AI Agent 框架设计

总体架构鸟瞰图

Agno 采用分层架构,从编排层到基础设施层,每层之间通过清晰的抽象接口连接。

编排层
执行层
能力层
存储层
接口层
接口层 (Interface) CLI / print Playground REST API Streaming HITL (Human) RunEvent (28+) 编排层 (Orchestration) Workflow Step | Steps | Loop | Parallel | Router Team Coordinate | Route | Broadcast | Tasks Agent run() | arun() | stream | session 执行层 (Execution Engine) Model (LLM) 30+ providers RunContext state | deps | tools RunMessages system | user | history Reasoning CoT | Native Hooks pre | post 能力层 (Capabilities) Tools 70+ toolkits Knowledge RAG / VectorDB Memory user memories Learning 6 store types Guardrails pre/post hooks Embedder 15+ providers 存储层 (Storage) BaseDb / AsyncBaseDb PG | SQLite | Mongo | Redis... VectorDb pgvector | Qdrant | Pinecone... Session Agent | Team | Workflow Tables (10+) sessions | memories | traces... 输出层 (Output) RunOutput content | tools | metrics ModelResponse parsed_response ToolExecution call | result | confirm output_schema Pydantic BaseModel Media image | audio | video

核心编排三剑客:Workflow / Team / Agent

Agno 的三个顶层抽象,从宏观编排到微观执行的层次关系。

Workflow 声明式流水线编排器 Step Parallel Loop Router Condition Steps executor Team 多 Agent 协作管理器 Coordinate Route Broadcast Tasks members: List[Agent | Team] Agent 单体智能执行器 (核心基本单元) Model Tools Knowledge Memory Learning Reasoning Guardrails Hooks members nest 共享基础设施 BaseDb Session RunContext RunOutput RunEvent Workflow / Team / Agent 共享相同的存储、会话和运行时数据结构

Agent 内部架构:模块委托模式

Agent 将核心功能拆分到多个子模块中,通过委托模式实现关注点分离。

Agent agent.py (dataclass, init=False) _run.py 核心 run 循环 (14 步) _messages.py 消息构建 + 知识注入 _tools.py 工具解析与执行循环 _session.py 会话 CRUD + 历史 _storage.py 序列化 / 版本化存储 _managers.py 后台异步任务管理 _hooks.py pre/post hook 执行 _response.py 结构化输出转换 _init.py 初始化 + 工具注册 _cli.py CLI + print_response _telemetry.py 遥测日志 _reasoning.py CoT / 原生推理 外部依赖注入 Model Tools Knowledge Memory Learning BaseDb Guardrails

Agent.run() 完整数据流

从用户输入到最终输出,一次完整调用经历的 14 个步骤。

  1. 加载 / 创建 Session
    根据 session_idBaseDb 加载 AgentSession,不存在则创建新会话。
  2. 合并 Metadata 和 State
    合并 session_state(用户数据)和 extra_data,更新元信息。
  3. 解析 Dependencies
    执行 callable 工厂函数,动态注入 tools、knowledge、members 等依赖。支持 cache_callables=True
  4. 执行 Pre-Hooks
    运行 Guardrails、Evals 和自定义前置钩子。可修改输入或阻止执行。
  5. 确定工具集
    合并用户工具、默认工具(search_knowledgeread_chat_history)和 Knowledge 暴露的工具。
  6. 构建 RunMessages
    组装完整的消息链:system_message(含指令 + 工具说明 + 知识上下文) → history_messages(从历史 runs 重建) → user_message(用户输入 + RAG 检索结果 + 多模态媒体)
  7. 启动后台任务
    通过 ThreadPoolExecutor 并行启动:Memory 更新、Learning 提取、Cultural Knowledge 更新。
  8. 推理阶段 (可选)
    如果 reasoning=True,执行多步 Chain-of-Thought 推理,或调用原生推理模型(DeepSeek / Claude)。生成 ReasoningStep 序列。
  9. Model.response() 核心循环
    LLM 调用Model._invoke_with_retry() → HTTP API
    工具循环:发现 tool_callsFunction.execute() → 追加 tool 消息 → 再次调用模型
    循环直到模型不再请求工具调用。
  10. 输出后处理
    可选的 output_model(二次 LLM 结构化)或 parser_model(解析模型),最后通过 Pydantic 验证转为 output_schema
  11. 执行 Post-Hooks
    运行后置 Guardrails、Evals 和自定义钩子。可检查输出质量、安全性。
  12. 等待后台任务
    等待 Memory、Learning、Culture 后台任务完成。
  13. 创建会话摘要
    SessionSummaryManager 生成会话摘要,用于长对话的上下文压缩。
  14. 持久化 + 返回
    AgentSession.upsert_run()BaseDb.upsert_session() 持久化会话。返回 RunOutput(或流式 Iterator[RunOutputEvent])。

存储与基础设施层

Agno 的可插拔存储架构:每个抽象接口都有多种实现可供选择。

BaseDb / AsyncBaseDb 统一关系型数据库接口 (ABC) 管理的数据表 (10+): agno_sessions agno_memories agno_traces agno_learnings agno_knowledge agno_approvals 实现: Postgres SQLite MySQL Mongo Redis DynamoDB Firestore InMemory VectorDb 向量数据库接口 (ABC) 核心方法: insert() / upsert() / search() async_insert() / async_search() delete_by_id() / delete_by_name() 18+ 实现: pgvector Qdrant Pinecone Weaviate Milvus Chroma LanceDB MongoDB Redis ClickHouse Model Providers (30+) OpenAI Claude Gemini DeepSeek Groq Mistral Ollama AWS Bedrock Azure Cohere... 抽象基类: Model(ABC) → invoke() / ainvoke() / invoke_stream() / ainvoke_stream() Embedder Providers (15+) get_embedding(text) → List[float] OpenAI | Azure | Cohere | HuggingFace | Ollama | SentenceTransformer... Session System 会话管理 + 历史重建 + 摘要压缩 AgentSession | TeamSession | WorkflowSession | SessionSummaryManager

各模块详解

Agno 框架中的关键模块,点击卡片查看更多细节。

A
Agent
libs/agno/agno/agent/
核心基本单元。通过 @dataclass(init=False) 装饰,使用「模块委托」模式将功能拆分到 12 个子模块。支持同步/异步、流式/非流式四种调用方式。一次 run() 经历 14 个步骤完成。
run() arun() stream print_response
T
Team
libs/agno/agno/team/
多 Agent 协作管理器。支持四种协作模式:Coordinate(领导者分配)、Route(路由到专家)、Broadcast(广播)、Tasks(自主任务分解)。支持嵌套 Team。
Coordinate Route Broadcast Tasks
W
Workflow
libs/agno/agno/workflow/
声明式流水线系统。将 Agent/Team 组合为有序或并行执行步骤。支持 6 种步骤类型:StepStepsLoopParallelConditionRouter
Step Parallel Loop Router
M
Model
libs/agno/agno/models/
LLM 抽象基类 (ABC)。定义 invoke / ainvoke / invoke_stream / ainvoke_stream 四种核心方法。内置自动重试。已有 30+ 提供者实现。
OpenAI Claude Gemini 30+
F
Tools
libs/agno/agno/tools/
工具系统。Toolkit 是工具集容器,Function 是单个工具的元数据和执行入口。支持 HITL(需确认/需用户输入/外部执行)。内置 70+ 工具覆盖搜索、代码执行、数据库、媒体等。
Toolkit Function HITL MCP
K
Knowledge / RAG
libs/agno/agno/knowledge/
知识库系统。通过 KnowledgeProtocol 协议接口实现可插拔。Knowledge 类集成 VectorDb + Embedder + Document + Reader。支持远程加载(S3、GCS、SharePoint、GitHub)。
VectorDb Embedder Document RAG
Mem
Memory
libs/agno/agno/memory/
用户记忆管理。在每次 run 结束后用 LLM 从对话中提取 UserMemory,存入数据库,下次 run 时注入上下文。支持可插拔优化策略。
MemoryManager UserMemory Strategy
L
Learning
libs/agno/agno/learn/
统一学习系统。LearningMachine 协调 6 种学习存储:UserProfile、UserMemory、SessionContext、EntityMemory、LearnedKnowledge、DecisionLog。支持命名空间隔离和自定义存储。
Profile Entity Knowledge Curator
R
Reasoning
libs/agno/agno/reasoning/
推理管理器。支持两种模式:多步 Chain-of-Thought(通过 ReasoningStep 逐步推理)和原生推理(委托给 DeepSeek/Claude 等原生推理模型)。可配置最小/最大推理步数。
CoT Native ReasoningStep
DB
Database
libs/agno/agno/db/
统一数据库抽象层。BaseDb / AsyncBaseDb 管理 10+ 数据表。推荐生产使用 PostgreSQL,开发用 SQLite。已有 10+ 数据库实现。
Postgres SQLite MongoDB Redis
Run
Run Layer
libs/agno/agno/run/
运行时状态层。RunContext 贯穿整个运行过程,携带 session_state、依赖和工具。RunOutput 封装完整输出。RunEvent 定义 28 种事件类型用于流式传输。
RunContext RunOutput 28 Events
G
Guardrails & Hooks
libs/agno/agno/guardrails/
安全护栏和钩子系统。BaseGuardrail 基类实现 pre/post 两个阶段的安全检查。支持输入验证、输出审查和自定义逻辑。与 Evals 评估系统集成。
PreHook PostHook Guardrail Eval