Agno — 架构与原理

30 秒导读: Agno 是一个 Python 的 agent 框架 + 运行平台。你用一个 Agent 对象声明「用哪个模型、给哪些工具、记什么、查什么知识库」，调 agent.run("...") 它就替你跑完整个「问模型 → 模型要调工具 → 执行工具 → 把结果喂回模型 → 直到模型给出最终答案」的循环。再把若干 Agent 交给 AgentOS，它就生成一个带 50+ REST 端点、SSE 流、RBAC、定时任务的 FastAPI 服务——本地写的 agent 直接变成生产服务。

1. 这是什么(零基础也能懂)

一句话定义: Agno 是一个「构建 + 运行 + 管理」AI agent 的 SDK。它把「让大模型自己调工具、记笔记、查资料、完成任务」这件事封装成几行配置,再把这些 agent 一键挂成生产级 Web 服务。

解决什么问题 / 给谁用: 假设你想做一个「会查资料、会调你公司 API、记得用户偏好、出错能让人审批」的 AI 助手。手写的话,你得自己实现:把工具描述喂给模型、解析模型要调哪个工具、执行它、把结果再喂回去、循环到模型满意;还得自己接数据库存会话、接向量库做检索、写一套 HTTP 接口给前端。Agno 把这一整套都做好了——你只写「模型是谁、工具有哪些、规则是什么」,执行循环和服务化它包办。

给两类人用:

应用开发者 —— 想快速搭一个能用工具、有记忆、能联网的 agent。
平台团队 —— 想把一堆 agent 统一托管成带权限、可观测、能调度的服务,跑在自己的云里。

它能做什么:

单 Agent:模型 + 工具循环 + 记忆 + 知识检索(RAG) + 会话持久化。
多 Agent:Team(一个领队把任务委派给成员)、Workflow(确定性的步骤编排,带循环/分支/并行)。
人审(HITL):某个工具调用前暂停,等人确认/补输入/外部执行。
服务化:AgentOS 把上面这些变成 FastAPI 应用(SSE、WebSocket、RBAC、cron 调度)。

用起来什么样: 一个最小的真实 agent 就是一个 dataclass。

# 示意,非源码 —— 体现 Agno 的最小用法
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),     # 用哪个模型
    tools=[DuckDuckGoTools()],          # 给它一个搜索工具
    instructions="用中文回答,先搜索再总结",
)
agent.print_response("今天 AI 圈有什么大新闻?")  # 它会自己决定要不要搜、搜完总结

挂成服务也只是再包一层:

# 示意,非源码
from agno.os import AgentOS
os_app = AgentOS(agents=[agent])
app = os_app.get_app()    # 一个完整的 FastAPI app,带 /runs、/sessions 等 50+ 端点

一句话直觉/类比: 把 Agent 想成「一个配置好的员工」——你告诉他用什么脑子(模型)、能用哪些工具、记什么、查什么资料;agent.run() 就是「派他干一件活」,他会自己反复「想—动手—看结果—再想」直到交活。AgentOS 则是「把一群这样的员工编进一个公司前台系统」,外界通过 HTTP 找他们办事。

2. 顶层全景(它大概怎么转)

Agno 的代码都在 libs/agno/agno/ 下。心脏是 Agent,它本身只是个大 dataclass(配置容器);真正的执行逻辑拆在一组 _xxx.py 私有模块里,而「反复调工具」的循环其实下沉到了 Model 层。

2.1 主要部件

下面这张图是「一次 agent.run() 从上到下经过的层」。从上往下读:上层管「准备上下文、收尾存储」,下层管「真正反复调模型 + 执行工具」。

  你的代码: agent.run("问题")
         │
         ▼
  ┌─────────────────────────────────────────────┐
  │ Agent (dataclass 配置)   agent/agent.py        │
  │  run() → 转发给 _run.run_dispatch              │
  └─────────────────────────────────────────────┘
         │
         ▼
  ┌─────────────────────────────────────────────┐
  │ 编排层  agent/_run.py  (_run / _arun / *_stream)│
  │  读会话→跑pre-hook→定工具→拼消息→推理→调模型→收尾 │
  └─────────────────────────────────────────────┘
         │  call_model_with_fallback(messages, tools, ...)
         ▼
  ┌─────────────────────────────────────────────┐
  │ Model 层  models/base.py  response()           │
  │  while True: 问模型 → 有工具调用?              │
  │     ├─是→执行工具→结果喂回→continue            │
  │     └─否→break,返回最终回答                    │
  └─────────────────────────────────────────────┘
         │ 旁路依赖(都在编排层注入/收尾)
         ▼
  Memory(记笔记) · Knowledge(RAG检索) · Session(会话存DB) · Reasoning(推理)

两层循环是关键直觉: 编排层(_run.py)管「一次 run 的全生命周期」——它只调一次 model.response(...);而「反复调工具」的真正 while 循环在 Model.response() 里。换句话说,「agentic loop」住在 Model 层,不在 Agent 层。

2.2 部件一句话职责

部件	干什么	在哪个文件
`Agent`	配置容器 + 对外 `run/arun/print_response` 入口	`agent/agent.py:69`
编排函数	准备上下文、收尾存储,串起整条流水线	`agent/_run.py:339`(`_run`)
`Model.response`	真正的「问模型 ↔ 调工具」while 循环	`models/base.py:650`
`Function`	把一个 Python 函数变成模型可调的工具(含 JSON schema)	`tools/function.py:132`
`Team`	一个领队 agent 把任务委派给成员 agent	`team/team.py:73`
`Workflow`	确定性步骤编排(Step/Loop/Parallel/Condition/Router)	`workflow/workflow.py`
`MemoryManager`	从对话里抽取「用户记忆」存库,并在上下文里回放	`memory/manager.py:46`
`Knowledge`	文档分块 + 向量检索(RAG)	`knowledge/knowledge.py`
`BaseDb`	会话/记忆/知识的存储抽象(Postgres/SQLite/Mongo/…)	`db/base.py:30`
`AgentOS`	把 agent/team/workflow 挂成 FastAPI 服务	`os/app.py:221`

2.3 主线走一遍(高层,不进代码)

以 agent.run("问题") 为例,端到端是这样:

Agent.run 把活转发给 _run.run_dispatch,生成 run_id 和一个 RunOutput(本次运行的「答卷」对象)。
进 _run:读/建会话 → 跑 pre-hooks(校验、护栏) → 决定这次给模型哪些工具 → 拼好要发给模型的消息(system + 历史 + 记忆 + 知识 + 用户输入)。
后台线程同时开始抽记忆、抽学习信号(不阻塞主流程)。
若开了推理,先让模型「想一轮」。
调 model.response(...) 进入 Model 层的 while 循环:模型回答 → 若要调工具就执行、把结果喂回去再问一次 → 直到模型不再调工具。
回到 _run:把模型结果写进 RunOutput、存媒体、跑 post-hooks、等后台记忆写完、生成会话摘要、把会话和本次运行存库。
返回 RunOutput(或在 streaming 模式下沿途 yield 事件)。

想看这条主线的每一步对应的真实代码,读 01-agent-run-loop.md。

3. 阅读地图(按这个顺序读)

建议顺序——前 3 章是「单个 agent 怎么跑」,后 3 章是「组合与服务化」:

01-agent-run-loop.md — 一次 run 的完整流水线;Agent 编排层与 Model 工具循环的分工。先读这章。
02-tools-and-hitl.md — 普通函数→工具的 schema 自动生成;requires_confirmation 等 4 种暂停如何打断循环做人审。
03-context-memory-knowledge.md — system message 怎么拼;记忆/知识(RAG)/会话状态/历史如何进上下文。
04-teams-and-workflows.md — Team(委派)与 Workflow(步骤图)两种组合方式。
05-models-and-reasoning.md — 模型 provider 的统一抽象;native vs 默认 CoT 推理。
06-agentos-platform.md — AgentOS 如何生成 FastAPI 服务、RBAC、调度。

4. 巧妙之处(先记住这几条)

agentic loop 下沉到 Model 层。 Agent 只调一次 model.response,工具反复执行的 while 在 models/base.py:703。好处:每个 provider 的差异(消息格式、工具格式)都被 Model 子类吸收,而循环逻辑只写一份。
工具 = 普通函数 + docstring。 Function.from_callable 用 inspect + docstring 解析自动生成 JSON schema(tools/function.py:278),还会自动剔除 agent/run_context/images 这类「框架注入参数」,不暴露给模型。
HITL 靠「在循环里 break」实现。 工具标了 requires_confirmation 时,Model 不执行它,而是产出一个「暂停」标记并 break 出 while(models/base.py:849),把控制权交回给人。
配置即 dataclass。 Agent 是 @dataclass(init=False),几十个字段全是声明式开关(agent/agent.py:68)——读这个类就读懂了「一个 agent 能配什么」。

5. 边界与局限(诚实)

同步/异步要选对路。 若给的是异步 DB,调同步 run() 会直接抛错,必须用 arun()(agent/_run.py:1322)。
add_history_to_context 需要 DB。 没配数据库时开历史会被警告并静默忽略(agent/_run.py:1328)。
checkpoint="tools" 还没实现。 注释明说是「reserved for 3.0,2.x 抛 NotImplementedError」(agent/agent.py:138)。
本系列文档只读了核心层(agent / model base / tools / os / 部分 memory·knowledge·reasoning·team·workflow)。每个具体 provider 适配器、每个 DB 后端的实现细节未逐行读;凡推断处文中标 (inferred)。

6. 横向对比(同 shelf 兄弟)

Agno 在 ai-frontier-reference/agent-frameworks 区。和兄弟框架的取舍差异(对比维度,具体兄弟 doc 见各自页面):

维度	Agno 的取舍
Agent 表示	纯 dataclass 配置容器,声明式;不是图(graph)节点
循环位置	agentic loop 在 Model 层,Agent 只编排一次
编排模型	两套并存:Team(LLM 自主委派) + Workflow(确定性步骤图)
服务化	内建 `AgentOS`,一行变 FastAPI + RBAC + SSE + cron
多 provider	30+ provider 收敛到一个 `Model` 抽象

7. 代码地图(导航索引)

主题	文件	符号
Agent 配置与入口	`libs/agno/agno/agent/agent.py`	`Agent`、`Agent.run`、`Agent.arun`
一次 run 的编排	`libs/agno/agno/agent/_run.py`	`run_dispatch`、`_run`、`_arun`、`_run_stream`
Model 工具循环	`libs/agno/agno/models/base.py`	`Model.response`、`Model.run_function_calls`
函数→工具 schema	`libs/agno/agno/tools/function.py`	`Function.from_callable`、`Function`
system message 拼装	`libs/agno/agno/agent/_messages.py`	`get_system_message`、`get_run_messages`
多 Agent 委派	`libs/agno/agno/team/team.py`	`Team`、`_get_delegate_task_function`
工作流编排	`libs/agno/agno/workflow/workflow.py`	`Workflow`、`Step`、`Loop`、`Parallel`
记忆	`libs/agno/agno/memory/manager.py`	`MemoryManager`、`create_user_memories`
存储抽象	`libs/agno/agno/db/base.py`	`BaseDb`、`upsert_session`、`upsert_user_memory`
服务化	`libs/agno/agno/os/app.py`	`AgentOS`、`AgentOS.get_app`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

2.1 主要部件​

2.2 部件一句话职责​

2.3 主线走一遍(高层,不进代码)​

3. 阅读地图(按这个顺序读)​

4. 巧妙之处(先记住这几条)​

5. 边界与局限(诚实)​

6. 横向对比(同 shelf 兄弟)​

7. 代码地图(导航索引)​