数字旗手

数字化、智能化、智能体化

0%

多智能体开发框架Agno教程0——快速上手和相关概念

Posted on Edited on In Disqus:

Agno是一个用于构建AI智能体（包括多模态智能体和多智能体）的开源Python框架，支持工具调用、记忆、知识检索、可观测性等特性，可用于生产环境，它一个特点是非常快，官网有它与其他框架的一个速度对比。
这里将对Agno进行下研究。

安装

创建虚拟环境

1
2
uv venv --python 3.12
source .venv/bin/activate

安装依赖

1
uv pip install -U agno openai anthropic mcp "fastapi[standard]" sqlalchemy

配置key

1
export OPENAI_API_KEY=sk-***

快速开始

创建一个hackernews_agent.py文件，然后粘帖：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools

agent = Agent(
    model=OpenAIChat(
        id="gpt-4o",
        base_url="https://api.xxx.com/v1" # 不配置的话就默认使用OpenAI的url
        ),
    tools=[HackerNewsTools()],
    markdown=True,
)

agent.print_response("Write a report on trending startups and products.", stream=True)

输出结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
┃                                                                                                         ┃
┃                                                1. Ventoy                                                ┃
┃                                                                                                         ┃
┃  • Created by: wilsonfiifi                                                                              ┃
┃  • Description: Ventoy is a tool to create bootable USB drives for ISO/WIM/IMG/VHD(x)/EFI files. It's   ┃
┃    popular for its ease of use and flexibility allowing multiple bootable files on a single USB device. ┃
┃  • Score: 159                                                                                           ┃
┃  • Discussion: Hacker News Thread                                                                       ┃
┃  • URL: github.com/ventoy/Ventoy                                                                        ┃
┃                                                                                                         ┃
┃                                          2. Affinity by Canva                                           ┃
┃                                                                                                         ┃
┃  • Created by: microflash                                                                               ┃
┃  • Description: Affinity is a product from Canva aimed at enhancing design productivity, particularly   ┃
┃    focusing on affinity-based features and UI/UX design elements.                                       ┃
┃  • Score: 8                                                                                             ┃
┃  • Discussion: Hacker News Thread                                                                       ┃
┃  • URL: affinity.studio                                                                                 ┃
┃------------------------------------------ 还有更多，略去 ------------------                             |

第一个智能体

上面的示例算是一个toy demo，下面是一个完整体的智能体。

创建智能体OS

创建agno_agent.py，然后

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.os import AgentOS
from agno.tools.mcp import MCPTools

# Create the Agent
agno_agent = Agent(
    name="Agno Agent",
    model=OpenAIChat(
        id="gpt-4o",
        base_url="https://api.zhizengzeng.com/v1"),
    # Add a database to the Agent
    db=SqliteDb(db_file="agno.db"),
    # Add the Agno MCP server to the Agent
    tools=[MCPTools(transport="streamable-http", url="https://docs.agno.com/mcp")],
    # Add the previous session history to the context
    add_history_to_context=True,
    markdown=True,
)


# Create the AgentOS
agent_os = AgentOS(agents=[agno_agent])
# Get the FastAPI app for the AgentOS
app = agent_os.get_app()

启动这个AgentOS：

1
fastapi dev agno_agent.py

该智能体OS将会运行在http://localhost:8000/。

连接该智能体OS

Agno提供了一个连接到AgentOS的网页界面，可用于监控、管理和测试智能体系统。
打开 os.agno.com 并登录账户。

  • 点击顶部导航栏中的“Add new OS”（添加新 OS）。
  • 选择“Local”（本地），以连接运行在电脑上的本地AgentOS。
  • 输入上面AgentOS的端点URL。默认是 http://localhost:8000。
  • 给AgentOS起一个容易辨识的名称，比如“Development OS”或“Local 8000”。
  • 点击“Connect”（连接）。

可以通过这个页面来与AgentOS进行对话、查看对话历史、进行评估等等。

使用API

上面的AgentOS提供了API，用来调用，api文档见：http://localhost:8000/docs。

智能体

一个 Agent（智能体）是一个由大语言模型（LLM）作为“大脑”的自治程序，它不仅能够对话，还能决策、调用工具、访问知识、记忆状态，从而执行更复杂的任务。
换句话说，它不同于传统只是“接收问题——返回答案”的聊天机器人，而是能够在运行时决定：我需要先思考／调用工具／查知识／记忆下来／然后回应。
一个 Agno Agent 一般包含以下几个关键组成部分：

组成部分 作用
模型（Model） 驱动智能体“思考”的语言模型，例如 GPT‑4、Claude 等。Agent 的决策逻辑、是否调用工具、如何表达答案，都是由模型判断。
指令/提示（Instructions / Prompts） 设定 Agent 的行为规范、风格、工具使用规则、输出格式。告诉模型“你是这个样子”“你要这么做”。
工具（Tools） Agent 可调用的外部能力，比如网络搜索、金融数据接口、数据库查询、上传下载文件等。通过工具，Agent 能“出屋子”去获取或操作外部信息。
记忆（Memory） 用于让 Agent 在会话中或跨会话保存上下文、用户偏好、历史操作，以便更个性化或长期追踪。
知识库／检索（Knowledge / Retrieval） Agent 可以访问专门的知识库（例如向量数据库、PDF 文件、文档集合等）来做检索增强（RAG: Retrieval Augmented Generation）。
存储／状态（Storage / Persistence） 因为模型 API 本身通常是无状态的，Agent 需要持久化机制来保留会话数据、历史、工具调用记录、知识检索缓存等。
执行与控制（Execution & Control） 管理 Agent 的生命周期、工具调用时机、校验机制、守卫（guardrails）、日志监控等，以便真实环境中稳定运行。

构建智能体

在构建高效的智能体时，建议 从简单开始 —— 只包含模型（model）、工具（tools）和指令（instructions）。
当这些基础功能运作良好后，再逐步叠加更多高级特性。
🧩 最简单的报告生成 Agent 示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
# hackernews_agent.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[HackerNewsTools()],
    instructions="撰写一份关于该主题的报告，只输出报告内容。",
    markdown=True,
)

agent.print_response("当前热门的初创公司和产品。", stream=True)

🚀 运行Agent
在开发阶段，可以使用 Agent.print_response() 方法在终端中直接输出结果。
⚠️ 注意：该方法仅适用于开发调试，在生产环境中请使用 Agent.run() 或 Agent.arun()。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from typing import Iterator
from agno.agent import Agent, RunOutput, RunOutputEvent, RunEvent
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools
from agno.utils.pprint import pprint_run_response

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[HackerNewsTools()],
    instructions="撰写一份关于该主题的报告，只输出报告内容。",
    markdown=True,
)

# -------- 普通运行 --------
response: RunOutput = agent.run("当前热门的初创公司和产品。")
print(response.content)

# -------- 流式输出 --------
stream: Iterator[RunOutputEvent] = agent.run("热门产品趋势", stream=True)
for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(chunk.content)

# -------- 流式输出 + 美化打印 --------
stream: Iterator[RunOutputEvent] = agent.run("热门产品趋势", stream=True)
pprint_run_response(stream, markdown=True)

运行智能体

通过调用 Agent.run() 或 Agent.arun() 来运行智能体。运行流程如下：

  1. 智能体构建要发送给模型的上下文（包括系统消息、用户消息、聊天历史、用户记忆、会话状态及其他相关输入）。
  2. 智能体将该上下文发送给模型。
  3. 模型处理输入，返回 一个消息 或 一个工具调用（tool call）。
  4. 如果模型做了工具调用，智能体会执行该工具，并将结果返回给模型。
  5. 模型处理更新后的上下文，重复步骤 3–4，直到它生成一个 无需再调用工具 的最终消息。
  6. 智能体将此最终响应返回给调用方。

基本执行

Agent.run() 方法可运行智能体，并返回一个 RunOutput 对象（非流式）或当 stream=True 时返回 RunOutputEvent 对象的迭代器。示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from agno.agent import Agent, RunOutput
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools
from agno.utils.pprint import pprint_run_response

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[HackerNewsTools()],
    instructions="撰写一份关于该主题的报告，只输出报告内容。",
    markdown=True,
)

# 运行智能体，并将响应存为变量
response: RunOutput = agent.run("热门初创公司和产品趋势。")
# 以 markdown 格式打印响应
pprint_run_response(response, markdown=True)

也可以使用异步方式 Agent.arun() 来运行。参见异步示例。

运行输入

input 参数为发送给智能体的输入。它可以是字符串、列表、字典、消息对象、pydantic 模型或消息列表。例如：

1
response: RunOutput = agent.run(input="热门初创公司和产品趋势。")

若要了解如何使用结构化输入／输出，请参见 “输入 & 输出” 文档。

运行输出

Agent.run()（非流式）返回一个 RunOutput 对象，包含以下核心属性：

  • run_id: 本次运行的 ID。
  • agent_id: 智能体的 ID。
  • agent_name: 智能体名称。
  • session_id: 会话 ID。
  • user_id: 用户 ID。
  • content: 响应的内容。
  • content_type: 内容类型；若输出为结构化模型，则为该模型的类名。
  • reasoning_content: 推理内容。
  • messages: 发送给模型的消息列表。
  • metrics: 本次运行的指标。

更多细节请参见 RunOutput 的文档。

流式（Streaming）

若设定 stream=True，run() 将返回一个 RunOutputEvent 对象的迭代器，用于逐步接收响应。例如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Iterator
from agno.agent import Agent, RunOutputEvent, RunEvent
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[HackerNewsTools()],
    instructions="撰写一份关于该主题的报告，只输出报告内容。",
    markdown=True,
)

stream: Iterator[RunOutputEvent] = agent.run("热门产品趋势", stream=True)
for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(chunk.content)

若要异步流式运行，参见异步示例。

流式所有事件

默认情况下，流模式仅返回 RunContent 事件。
也可以通过设定 stream_events=True 来流式接收 所有事件，包括工具调用、推理步骤等。例如：

1
2
3
4
5
response_stream: Iterator[RunOutputEvent] = agent.run(
    "热门产品趋势",
    stream=True,
    stream_events=True
)

处理事件

可以在收到事件时逐一处理，例如：

1
2
3
4
5
6
7
8
9
stream = agent.run("热门产品趋势", stream=True, stream_events=True)

for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(f"内容: {chunk.content}")
    elif chunk.event == RunEvent.tool_call_started:
        print(f"工具调用启动: {chunk.tool.tool_name}")
    elif chunk.event == RunEvent.reasoning_step:
        print(f"推理步骤: {chunk.content}")

事件类型

下面是 Agent.run() 和 Agent.arun() 在不同配置下可能产生的事件类别：

核心事件

事件类型 描述
RunStarted 表示运行开始
RunContent 模型响应文本按块返回
RunContentCompleted 内容流式输出完成
RunCompleted 表示运行成功结束
RunError 表示运行过程中发生错误
RunCancelled 表示运行被取消

控制流程事件

事件类型 描述
RunPaused 运行被暂停
RunContinued 暂停后的运行被继续

工具相关事件

事件类型 描述
ToolCallStarted 表示工具调用开始
ToolCallCompleted 表示工具调用结束并返回结果

推理相关事件

事件类型 描述
ReasoningStarted 推理过程开始
ReasoningStep 推理过程中的一个步骤
ReasoningCompleted 推理过程完成

记忆相关事件

事件类型 描述
MemoryUpdateStarted 智能体开始更新记忆
MemoryUpdateCompleted 智能体完成记忆更新

会话摘要相关事件

事件类型 描述
SessionSummaryStarted 会话摘要生成开始
SessionSummaryCompleted 会话摘要生成完成

前置钩子（Pre-Hook）事件

事件类型 描述
PreHookStarted 前置运行钩子开始
PreHookCompleted 前置钩子执行完成

后置钩子（Post-Hook）事件

事件类型 描述
PostHookStarted 后置运行钩子开始
PostHookCompleted 后置钩子执行完成

解析器模型事件（Parser Model）

事件类型 描述
ParserModelResponseStarted 解析器模型响应开始
ParserModelResponseCompleted 解析器模型响应完成

输出模型事件（Output Model）

事件类型 描述
OutputModelResponseStarted 输出模型响应开始
OutputModelResponseCompleted 输出模型响应完成

自定义事件

如果使用自定义工具，也可以定义自定义事件。方法如下：

1
2
3
4
5
6
7
8
9
from dataclasses import dataclass
from agno.run.agent import CustomEvent

@dataclass
class CustomerProfileEvent(CustomEvent):
    """客户档案的自定义事件。"""
    customer_name: Optional[str] = None
    customer_email: Optional[str] = None
    customer_phone: Optional[str] = None

然后工具中可 yield 这个事件。该事件会被 Agno 内部作为普通事件处理。例如：

1
2
3
4
5
6
7
8
9
10
11
from agno.tools import tool

@tool()
async def get_customer_profile():
    """示例：仅 yield 一个自定义事件的自定义工具。"""

    yield CustomerProfileEvent(
        customer_name="John Doe",
        customer_email="[email protected]",
        customer_phone="1234567890",
    )

更多细节请见完整示例文档。

指定运行用户与会话

可以通过传入 user_id 与 session_id 参数，指定当前运行关联的用户和会话。例如：

1
2
3
4
5
agent.run(
    "讲一个关于机器人 5 秒钟的短故事",
    user_id="[email protected]",
    session_id="session_123"
)

更多信息请参见 “Agent 会话” 文档。

传递图像／音频／视频／文件

可以通过 images、audio、video 或 files 参数向智能体传递图像、音频、视频或文件。例如：

1
2
3
4
agent.run(
    "基于这张图片讲一个 5 秒钟短故事",
    images=[Image(url="https://example.com/image.jpg")]
)

更多详情请参见 “多模态 Agent” 文档。

暂停与继续运行

如果运行过程中触发了 “人类在环（Human-in-the-Loop）” 的流程，智能体运行可能被暂停。这时可以调用 Agent.continue_run() 方法继续执行。

更多细节请参见 “Human-in-the-Loop” 文档。

取消运行

可以调用 Agent.cancel_run() 方法来取消当前运行。

更多详情请参见 “取消运行” 文档。

调试智能体

Agno 提供了一个非常完善的 调试模式（Debug Mode），它能显著提升你的开发体验，帮助你理解代理（Agent）的执行流程和中间步骤。例如：

  1. 检查发送给模型的消息及其返回的响应。
  2. 跟踪中间步骤并监控指标（如 token 使用量、执行时间等）。
  3. 检查工具调用、错误及其结果。

启用调试模式

有三种方式可以启用调试模式：

  1. 在创建Agent时设置 debug_mode=True，对所有运行生效。
  2. 在调用 run() 方法时设置 debug_mode=True，仅对当前运行生效。
  3. 设置环境变量 AGNO_DEBUG=True，启用全局调试模式。

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from agno.agent import Agent
from agno.models.openai import OpenAI
from agno.tools.hackernews import HackerNewsTools

agent = Agent(
    model=OpenAI(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    instructions="撰写关于该主题的报告，仅输出报告内容。",
    markdown=True,
    debug_mode=True,
    # debug_level=2,  # 取消注释可获得更详细的日志
)

# 运行代理并在终端中打印结果
agent.print_response("热门初创公司和产品趋势。")

💡 可以设置 debug_level=2 来输出更详细的调试日志。

交互式 CLI

Agno 还提供了一个内置的 交互式命令行界面（CLI），可以直接在终端中与代理进行对话式测试，非常适合调试多轮交互。

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAI
from agno.tools.hackernews import HackerNewsTools

agent = Agent(
    model=OpenAI(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    db=SqliteDb(db_file="tmp/data.db"),
    add_history_to_context=True,  # 将对话历史添加到上下文
    num_history_runs=3,           # 仅保留最近3轮对话
    markdown=True,
)

# 以交互式 CLI 方式运行代理
agent.cli_app(stream=True)

智能体会话

当我们调用 Agent.run() 时，它会创建一个无状态（stateless）的单次运行（run）。
但如果我们希望继续对话、实现多轮交互（multi-turn conversation），就需要用到 “会话（Session）”。
一个会话是由多次连续运行组成的集合。

基本概念

  • Session（会话）：表示与 Agent 的一次多轮对话，包含多个连续的 run。每个会话由 session_id 标识，内部保存所有运行记录、指标和状态信息。
  • Run（运行）：每次与 Agent 的交互（即一次用户输入与模型响应）称为一次运行，由 run_id 标识。
  • Messages（消息）：表示模型与 Agent 之间传递的单条消息，是双方的通信单位。

更多细节请参考 Session Storage（会话存储）。

单轮会话示例

在下例中，Agno 自动为我们生成 run_id 和 session_id：

1
2
3
4
5
6
7
8
9
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

response = agent.run("讲一个关于机器人的5秒短故事")
print(response.content)
print(response.run_id)
print(response.session_id)

多轮会话（Multi-turn Sessions）

每个用户都可以拥有自己的会话集，多个用户可同时与同一个 Agent 交互。

可以使用 user_id 和 session_id 来区分不同用户与会话：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="tmp/data.db")

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    db=db,
    add_history_to_context=True,
    num_history_runs=3,
)

user_1_id = "user_101"
user_2_id = "user_102"

user_1_session_id = "session_101"
user_2_session_id = "session_102"

agent.print_response("讲一个关于机器人的短故事", user_id=user_1_id, session_id=user_1_session_id)
agent.print_response("再讲一个笑话", user_id=user_1_id, session_id=user_1_session_id)

agent.print_response("告诉我关于量子物理的事情", user_id=user_2_id, session_id=user_2_session_id)
agent.print_response("光速是多少？", user_id=user_2_id, session_id=user_2_session_id)

agent.print_response("总结一下我们的对话", user_id=user_1_id, session_id=user_1_session_id)

在上下文中加入历史记录

可以让 Agent 自动将对话历史加入上下文，这样模型就能记住前面的信息：

1
2
3
4
5
6
7
8
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.in_memory import InMemoryDb

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), db=InMemoryDb())

agent.print_response("嗨，我叫小明。很高兴认识你！")
agent.print_response("我叫什么名字？", add_history_to_context=True)

会话摘要（Session Summaries）

当会话内容过长时，Agent 可以生成简短摘要来概括整个对话。
设置 enable_session_summaries=True 即可启用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="tmp/data.db")
agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    db=db,
    enable_session_summaries=True,
)

session_id = "1001"
agent.print_response("什么是量子计算？", user_id="user_1", session_id=session_id)
agent.print_response("那大语言模型（LLM）呢？", user_id="user_1", session_id=session_id)

summary = agent.get_session_summary(session_id=session_id)
print(f"会话摘要: {summary.summary}")

可以通过 SessionSummaryManager 自定义摘要的生成方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from agno.agent import Agent
from agno.session import SessionSummaryManager
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="agno.db")

summary_manager = SessionSummaryManager(
    model=OpenAIChat(id="gpt-4o-mini"),
    session_summary_prompt="请为以下对话创建一个简短的总结：",
)

agent = Agent(
    db=db,
    session_summary_manager=summary_manager,
    enable_session_summaries=True,
)

访问会话历史

启用存储（Storage）后，可以随时访问某个会话的历史记录：

1
2
agent.get_messages_for_session(session_id)
agent.get_chat_history(session_id)

也可以自动让 Agent 在上下文中加载最近几次对话：

1
2
3
4
5
6
7
8
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    db=SqliteDb(db_file="tmp/data.db"),
    add_history_to_context=True,
    num_history_runs=3,
    read_chat_history=True,
    description="你是一位友好积极的智能助手。",
)

搜索历史会话

可以设置 Agent 搜索过去的多次会话：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
import os

os.remove("tmp/data.db")

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    user_id="user_1",
    db=SqliteDb(db_file="tmp/data.db"),
    search_session_history=True,
    num_history_sessions=2,
)

这将允许模型搜索最近两次会话内容。

控制会话中存储的内容

为了节省数据库空间，可设置以下参数：

  • store_media: 是否存储图片、音频、视频、文件等
  • store_tool_messages: 是否存储工具调用请求与结果
  • store_history_messages: 是否存储历史消息

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.db.sqlite import SqliteDb

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[DuckDuckGoTools()],
    db=SqliteDb(db_file="tmp/agents.db"),
    add_history_to_context=True,
    num_history_runs=5,
    store_media=False,
    store_tool_messages=False,
    store_history_messages=False,
)

agent.print_response("搜索最新的AI新闻并总结")

输入与输出

Agno 的 Agent 支持多种输入与输出形式，从最基础的字符串交互，到基于 Pydantic 模型 的结构化数据验证。

🧩 基础模式：字符串输入输出

最常见的用法是以 str 输入、str 输出：

1
2
3
4
5
6
7
8
9
10
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    description="You write movie scripts.",
)

response = agent.run("Write movie script about a girl living in New York")
print(response.content)

💡 高级模式请参考：

🏗️ 结构化输出（Structured Output）

Agno 的一个强大特性是：可以让 Agent 生成 结构化数据（Pydantic 模型）。
这让 Agent 能够输出固定格式的数据，适合：

  • 特征提取
  • 数据分类
  • 模拟数据生成
  • 需要确定输出结构的生产系统。

例如，我们创建一个 “电影脚本” Agent：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import List
from rich.pretty import pprint
from pydantic import BaseModel, Field
from agno.agent import Agent
from agno.models.openai import OpenAIChat

class MovieScript(BaseModel):
    setting: str = Field(..., description="电影背景设置")
    ending: str = Field(..., description="电影结尾，没有就写快乐结局")
    genre: str = Field(..., description="电影类型")
    name: str = Field(..., description="电影名")
    characters: List[str] = Field(..., description="角色名称")
    storyline: str = Field(..., description="三句话总结剧情")

structured_output_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    description="You write movie scripts.",
    output_schema=MovieScript,
)

structured_output_agent.print_response("New York")

输出结果是一个 MovieScript 对象：

1
2
3
4
5
6
7
8
MovieScript(
    setting='在繁华的纽约街头与天际线中…',
    ending='主角在帝国大厦顶端拥吻...',
    genre='Action Thriller',
    name='The NYC Chronicles',
    characters=['Isabella Grant', 'Alex Chen', ...],
    storyline='一名记者揭露巨大阴谋...'
)

🧠 JSON 模式（use_json_mode）

部分模型无法直接生成结构化输出。
此时可以让 Agno 指示模型以 JSON 形式返回：

1
2
3
4
5
6
agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    description="You write movie scripts.",
    output_schema=MovieScript,
    use_json_mode=True,
)

⚠️ JSON 模式比结构化模式精度稍差，但在部分模型上更稳定。

⚡ 流式结构化输出（Streaming Structured Output）

结构化输出也可以流式返回，Agno 会在事件流中生成一个结构化结果：

1
2
3
structured_output_agent.print_response(
    "New York", stream=True, stream_events=True
)

📥 结构化输入（Structured Input）

Agent 的输入也可以是结构化数据（Pydantic 模型或 TypedDict）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from typing import List
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.hackernews import HackerNewsTools
from pydantic import BaseModel, Field

class ResearchTopic(BaseModel):
    topic: str
    focus_areas: List[str]
    target_audience: str
    sources_required: int = 5

hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
)

hackernews_agent.print_response(
    input=ResearchTopic(
        topic="AI",
        focus_areas=["AI", "Machine Learning"],
        target_audience="Developers",
        sources_required=5,
    )
)

✅ 输入验证（input_schema）

可通过 input_schema 参数验证传入的输入字典是否合法：

1
2
3
4
5
6
7
8
9
10
11
12
13
hackernews_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    input_schema=ResearchTopic,
)

hackernews_agent.print_response(
    input={
        "topic": "AI",
        "focus_areas": ["AI", "Machine Learning"],
        "target_audience": "Developers",
        "sources_required": "5",
    }
)

Agno 会自动将输入校验并转化为 Pydantic 模型对象。

🔒 类型安全 Agent（Typesafe Agents）

同时设置 input_schema 和 output_schema，可以构建端到端类型安全 Agent。

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from typing import List
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.hackernews import HackerNewsTools
from pydantic import BaseModel, Field
from rich.pretty import pprint

class ResearchTopic(BaseModel):
    topic: str
    sources_required: int = 5

class ResearchOutput(BaseModel):
    summary: str
    insights: List[str]
    top_stories: List[str]
    technologies: List[str]
    sources: List[str]

hn_researcher_agent = Agent(
    model=Claude(id="claude-sonnet-4-0"),
    tools=[HackerNewsTools()],
    input_schema=ResearchTopic,
    output_schema=ResearchOutput,
    instructions="Research hackernews posts for a given topic",
)

response = hn_researcher_agent.run(
    input=ResearchTopic(topic="AI", sources_required=5)
)

pprint(response.content)

输出：

1
2
3
4
5
6
7
ResearchOutput(
    summary='AI development is accelerating...',
    insights=['LLMs 更高效', '开源模型崛起'],
    top_stories=['GPT-5 消息', 'Claude 新版本发布'],
    technologies=['GPT-4', 'Claude', 'Transformers'],
    sources=['https://news.ycombinator.com/item?id=123', ...]
)

🧩 使用解析模型（Parser Model）

可以使用一个单独的模型来解析主模型的输出：

1
2
3
4
5
6
agent = Agent(
    model=Claude(id="claude-sonnet-4-20250514"),
    description="You write movie scripts.",
    output_schema=MovieScript,
    parser_model=OpenAIChat(id="gpt-5-mini"),
)

💡 优点：

  • 主模型负责推理；
  • 小模型负责结构化解析；
  • 提高可靠性、降低成本。

还可以通过 parser_model_prompt 自定义解析模型的提示词。

🔄 使用输出模型（Output Model）

当主模型擅长多模态任务（如图像分析）时，可用另一模型专门生成结构化输出：

1
2
3
4
5
6
agent = Agent(
    model=Claude(id="claude-sonnet-4-20250514"),
    description="You write movie scripts.",
    output_schema=MovieScript,
    output_model=OpenAIChat(id="gpt-5-mini"),
)

✨ 一些 Gemini 模型无法同时使用工具与结构化输出，此法是一个有效解决方案。

上下文工程

上下文工程是指设计和控制发送给语言模型的信息（上下文）的过程，以此来引导模型的行为和输出。
在实践中，构建上下文可以归结为一个问题：“哪些信息最有可能实现期望的结果？”
在 Agno 中，这意味着要仔细构建系统消息（system message），其中包含Agent的描述、指令以及其他相关设定。通过精心设计这些上下文，你可以：

  • 引导Agent表现出特定行为或角色；
  • 限制或扩展Agent的能力；
  • 确保输出结果一致、相关，并符合应用需求；
  • 启用更高级的用例，例如多步推理、工具使用或结构化输出。

有效的上下文工程是一个迭代过程：反复优化系统消息，尝试不同的描述和指令，并利用诸如 schemas、delegation、tool integrations 等特性。

Agno智能体的上下文由以下部分组成：

  • System message（系统消息）：发送给智能体的主要上下文信息，包括所有附加内容。
  • User message（用户消息）：发送给智能体的用户输入。
  • Chat history（聊天记录）：智能体与用户的对话历史。
  • Additional input（附加输入）：添加到上下文中的 few-shot 示例或其他额外内容。

系统消息上下文（System message context）

以下是用于创建系统消息的一些关键参数：

  1. Description（描述）：指导代理总体行为的描述。
  2. Instructions（指令）：一组具体、任务导向的操作指令，用于实现目标。
  3. Expected Output（期望输出）：描述代理预期生成的输出形式。

系统消息由代理的 description、instructions 和其他设置构建而成。

1
2
3
4
5
6
7
8
9
10
11
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    description="You are a famous short story writer asked to write for a magazine",
    instructions=["Always write 2 sentence stories."],
    markdown=True,
    debug_mode=True,  # 设置为 True 以查看详细日志及系统消息内容
)
agent.print_response("Tell me a horror story.", stream=True)

该代码将生成以下系统消息：

1
2
3
4
5
6
7
8
You are a famous short story writer asked to write for a magazine                                                                          
<instructions>                                                                                                                             
- Always write 2 sentence stories.                                                                                                         
</instructions>                                                                                                                            
                                                                                                                                            
<additional_information>                                                                                                                   
- Use markdown to format your answer
</additional_information>

系统消息参数说明

Agent 会创建一个默认的系统消息，可通过以下参数进行自定义：

参数名 类型 默认值 说明
description str None 添加到系统消息开头的代理描述。
instructions List[str] None 添加到系统提示中 <instructions> 标签内的指令列表。默认指令会根据 markdown、expected_output 等自动生成。
additional_context str None 添加到系统消息结尾的附加上下文。
expected_output str None 期望输出描述，添加到系统消息末尾。
markdown bool False 若为 True，则添加“使用 markdown 格式化输出”的指令。
add_datetime_to_context bool False 若为 True，则在提示中添加当前日期时间，让代理具备时间感知能力。
add_name_to_context bool False 若为 True，则将代理名称添加到上下文。
add_location_to_context bool False 若为 True，则添加代理的地理位置，用于生成与地点相关的回复。
add_session_summary_to_context bool False 若为 True，则将会话摘要加入上下文。详见 sessions。
add_memories_to_context bool False 若为 True，则添加用户记忆。详见 memory。
add_session_state_to_context bool False 若为 True，则添加会话状态。详见 state。
enable_agentic_knowledge_filters bool False 若为 True，则允许代理选择知识过滤器。详见 knowledge filters。
system_message str None 直接覆盖默认系统消息。
build_context bool True 若为 False，可禁用自动构建上下文。

详见 Agent 参考文档。

系统消息的构建方式

来看以下示例代理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from agno.agent import Agent

agent = Agent(
    name="Helpful Assistant",
    role="Assistant",
    description="You are a helpful assistant",
    instructions=["Help the user with their question"],
    additional_context="""
    Here is an example of how to answer the user's question: 
        Request: What is the capital of France?
        Response: The capital of France is Paris.
    """,
    expected_output="You should format your response with `Response: <response>`",
    markdown=True,
    add_datetime_to_context=True,
    add_location_to_context=True,
    add_name_to_context=True,
    add_session_summary_to_context=True,
    add_memories_to_context=True,
    add_session_state_to_context=True,
)

生成的系统消息如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
You are a helpful assistant
<your_role>
Assistant
</your_role>

<instructions>
  Help the user with their question
</instructions>

<additional_information>
Use markdown to format your answers.
The current time is 2025-09-30 12:00:00.
Your approximate location is: New York, NY, USA.
Your name is: Helpful Assistant.
</additional_information>

<expected_output>
  You should format your response with `Response: <response>`
</expected_output>

Here is an example of how to answer the user's question: 
    Request: What is the capital of France?
    Response: The capital of France is Paris.

You have access to memories from previous interactions with the user that you can use:

<memories_from_previous_interactions>
- User really likes Digimon and Japan.
- User really likes Japan.
- User likes coffee.
</memories_from_previous_interactions>

Note: this information is from previous interactions and may be updated in this conversation. You should always prefer information from this conversation over the past memories.

Here is a brief summary of your previous interactions:

<summary_of_previous_interactions>
The user asked about information about Digimon and Japan.
</summary_of_previous_interactions>

Note: this information is from previous interactions and may be outdated. You should ALWAYS prefer information from this conversation over the past summary.

<session_state> ... </session_state>

💡 提示：
这个示例展示了系统消息的完整结构，以说明它的可定制性。但在实际应用中，你通常只会启用其中的一部分配置。

附加上下文（Additional Context）

你可以通过 additional_context 参数在系统消息的末尾添加额外说明。

例如，下面的 additional_context 参数为代理添加了一条说明，告诉它可以访问特定数据库表。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from textwrap import dedent
from agno.agent import Agent
from agno.models.langdb import LangDB
from agno.tools.duckdb import DuckDbTools

duckdb_tools = DuckDbTools(
    create_tables=False, export_tables=False, summarize_tables=False
)
duckdb_tools.create_table_from_path(
    path="https://phidata-public.s3.amazonaws.com/demo_data/IMDB-Movie-Data.csv",
    table="movies",
)

agent = Agent(
    model=LangDB(id="llama3-1-70b-instruct-v1.0"),
    tools=[duckdb_tools],
    markdown=True,
    additional_context=dedent("""\
    You have access to the following tables:
    - movies: contains information about movies from IMDB.
    """),
)
agent.print_response("What is the average rating of movies?", stream=True)
工具指令（Tool Instructions）

当智能体使用某个 Toolkit 时，可以通过 instructions 参数将工具说明加入系统消息：

1
2
3
4
5
6
7
8
9
10
from agno.agent import Agent
from agno.tools.slack import SlackTools

slack_tools = SlackTools(
    instructions=["Use `send_message` to send a message to the user.  If the user specifies a thread, use `send_message_thread` to send a message to the thread."],
    add_instructions=True,
)
agent = Agent(
    tools=[slack_tools],
)

这些指令会被注入到系统消息的 <additional_information> 标签之后。

智能体记忆

当智能体设置了 enable_agentic_memory=True 时，它将具备创建或更新用户记忆的能力。
此时系统消息中会新增如下内容：

1
2
3
4
5
6
7
8
<updating_user_memories>
- You have access to the `update_user_memory` tool that you can use to add new memories, update existing memories, delete memories, or clear all memories.
- If the user's message includes information that should be captured as a memory, use the `update_user_memory` tool to update your memory database.
- Memories should include details that could personalize ongoing interactions with the user.
- Use this tool to add new memories or update existing memories that you identify in the conversation.
- Use this tool if the user asks to update their memory, delete a memory, or clear all memories.
- If you use the `update_user_memory` tool, remember to pass on the response to the user.
</updating_user_memories>
知识过滤器

若启用了知识功能并设置了 enable_agentic_knowledge_filters=True，则它能自动选择合适的知识过滤器。
系统消息会新增以下说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
The knowledge base contains documents with these metadata filters: [filter1, filter2, filter3].
Always use filters when the user query indicates specific metadata.

Examples:
1. If the user asks about a specific person like "Jordan Mitchell", you MUST use the search_knowledge_base tool with the filters parameter set to {{'<valid key like user_id>': '<valid value based on the user query>'}}.
2. If the user asks about a specific document type like "contracts", you MUST use the search_knowledge_base tool with the filters parameter set to {{'document_type': 'contract'}}.
4. If the user asks about a specific location like "documents from New York", you MUST use the search_knowledge_base tool with the filters parameter set to {{'<valid key like location>': 'New York'}}.

General Guidelines:
- Always analyze the user query to identify relevant metadata.
- Use the most specific filter(s) possible to narrow down results.
- If multiple filters are relevant, combine them in the filters parameter (e.g., {{'name': 'Jordan Mitchell', 'document_type': 'contract'}}).
- Ensure the filter keys match the valid metadata filters: [filter1, filter2, filter3].

You can use the search_knowledge_base tool to search the knowledge base and get the most relevant documents. Make sure to pass the filters as [Dict[str: Any]] to the tool. FOLLOW THIS STRUCTURE STRICTLY.

详细内容可参见 知识过滤器。

直接设置系统消息

可以通过 system_message 参数手动定义系统消息。
此时，所有其他设置将被忽略，仅使用你提供的内容。

1
2
3
4
5
from agno.agent import Agent
agent.print_response("What is the capital of France?")

agent = Agent(system_message="Share a 2 sentence story about")
agent.print_response("Love in the year 12000.")

💡 提示：
某些模型（例如 Groq 平台上的 llama-3.2-11b-vision-preview）要求不包含系统消息。
若要移除系统消息，请设置 build_context=False 且 system_message=None。
注意：若设置了 markdown=True，仍会自动添加系统消息，因此需关闭或显式禁用。

用户消息上下文

传递给 Agent.run() 或 Agent.print_response() 的 input 即为用户消息。

附加用户上下文

可以使用以下参数为用户消息添加额外上下文：

  • add_knowledge_to_context
  • add_dependencies_to_context
1
2
3
from agno.agent import Agent
agent = Agent(add_knowledge_to_context=True, add_dependencies_to_context=True)
agent.print_response("What is the capital of France?", dependencies={"name": "John Doe"})

发送给模型的用户消息如下：

1
2
3
4
5
6
7
8
9
10
11
What is the capital of France?

Use the following references from the knowledge base if it helps:
<references>
- Reference 1
- Reference 2
</references>

<additional context>
{"name": "John Doe"}
</additional context>

详见 依赖注入。

聊天记录

当智能体启用数据库存储后，会自动保存会话历史（参见 sessions）。
可以通过 add_history_to_context=True 将对话历史添加到上下文中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from agno.agent.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIChat

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    db=db,
    session_id="chat_history",
    instructions="You are a helpful assistant that can answer questions about space and oceans.",
    add_history_to_context=True,
    num_history_runs=2,  # 可选：限制添加到上下文中的历史轮数
)

agent.print_response("Where is the sea of tranquility?")
agent.print_response("What was my first question?")

这会将之前的对话添加到上下文中，使智能体能利用先前的信息生成更连贯的回答。
详见 sessions#session-history。

工具调用管理

参数 max_tool_calls_from_history 用于限制上下文中保留的最近 n 次工具调用，
以控制上下文大小并降低 token 成本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
import random

def get_weather_for_city(city: str) -> str:
    conditions = ["Sunny", "Cloudy", "Rainy", "Snowy", "Foggy", "Windy"]
    temperature = random.randint(-10, 35)
    condition = random.choice(conditions)
    return f"{city}: {temperature}°C, {condition}"

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[get_weather_for_city],
    db=SqliteDb(db_file="tmp/agent.db"),
    add_history_to_context=True,
    max_tool_calls_from_history=3,  # 仅保留最近 3 次工具调用
)
agent.print_response("What's the weather in Tokyo?")
agent.print_response("What's the weather in Paris?")  
agent.print_response("What's the weather in London?")
agent.print_response("What's the weather in Berlin?")
agent.print_response("What's the weather in Mumbai?")
agent.print_response("What's the weather in Miami?")
agent.print_response("What's the weather in New York?")
agent.print_response("What's the weather in above cities?")

此时模型仅会看到最近 3 个城市（Mumbai、Miami、New York）的工具调用结果。

🔎 说明：
max_tool_calls_from_history 仅过滤由 num_history_runs 加载的历史记录。
数据库中仍会保留完整历史。

少样本学习（Few-shot learning）与附加输入

通过 additional_input 参数可以在上下文中添加额外的消息（如 few-shot 示例），
这些消息会像对话历史一样参与上下文构建。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from agno.agent import Agent
from agno.models.message import Message
from agno.models.openai.chat import OpenAIChat

# Few-shot 示例
support_examples = [
    Message(role="user", content="I forgot my password and can't log in"),
    Message(role="assistant", content="""I'll help you reset your password right away...
"""),
    ...
]

agent = Agent(
    name="Customer Support Specialist",
    model=OpenAIChat(id="gpt-5-mini"),
    add_name_to_context=True,
    additional_input=support_examples,
    instructions=[
        "You are an expert customer support specialist.",
        "Always be empathetic, professional, and solution-oriented.",
        "Provide clear, actionable steps to resolve customer issues.",
        "Follow the established patterns for consistent, high-quality support.",
    ],
    markdown=True,
)

这让智能体能够根据少量示例学习回应风格与格式。

上下文缓存

多数模型提供商支持系统与用户消息的缓存机制，但实现方式各不相同。
通用思路是缓存重复或静态内容，在后续请求中重用，以减少 token 消耗。

Agno 的上下文构建逻辑天然会将最可能缓存的静态内容放在系统消息的开头。
如需进一步优化，可手动设置 system_message。

示例：

依赖注入

依赖项（Dependencies） 是一种向智能体上下文（Agent Context）注入变量的方式。
dependencies 是一个字典，包含一组函数（或静态变量），这些依赖项会在智能体运行前被解析。

你可以使用依赖项来注入记忆、动态 few-shot 示例、检索得到的文档等。

基本用法

可以在智能体的 instructions（指令）或用户消息中引用依赖项。

1
2
3
4
5
6
7
8
9
10
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    dependencies={"name": "John Doe"},
    instructions="You are a story writer. The current user is {name}."
)

agent.print_response("Write a 5 second short story about {name}")
你既可以在 `Agent` 初始化时设置 `dependencies`， 也可以在运行时通过 `run()` 或 `arun()` 方法传入。

使用函数作为依赖项

你可以将一个可调用函数指定为依赖项。
当代理运行时，该依赖项会被自动解析并执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import json
from textwrap import dedent
import httpx
from agno.agent import Agent
from agno.models.openai import OpenAIChat


def get_top_hackernews_stories() -> str:
    """获取并返回 HackerNews 上的热门新闻。

    Args:
        num_stories: 要获取的热门新闻数量（默认：5）
    Returns:
        JSON 字符串，包含新闻的标题、链接、评分等信息。
    """
    # 获取热门新闻
    stories = [
        {
            k: v
            for k, v in httpx.get(
                f"https://hacker-news.firebaseio.com/v0/item/{id}.json"
            )
            .json()
            .items()
            if k != "kids"  # 排除评论部分
        }
        for id in httpx.get(
            "https://hacker-news.firebaseio.com/v0/topstories.json"
        ).json()[:num_stories]
    ]
    return json.dumps(stories, indent=4)


agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    # 每个依赖项函数会在代理运行时自动求值
    # 可以将其理解为 Agent 的“依赖注入”
    dependencies={"top_hackernews_stories": get_top_hackernews_stories},
    # 也可以手动将依赖项添加到指令中
    instructions=dedent("""\
        You are an insightful tech trend observer! 📰

        Here are the top stories on HackerNews:
        {top_hackernews_stories}\
    """),
    markdown=True,
)

# 示例使用
agent.print_response(
    "Summarize the top stories on HackerNews and identify any interesting trends.",
    stream=True,
)
依赖项会在代理运行时自动解析。

将依赖项添加到上下文

设置 add_dependencies_to_context=True，
可以将整个依赖项字典添加到用户消息中。
这样你就不必手动把依赖项插入到指令中了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import json
from textwrap import dedent
import httpx
from agno.agent import Agent
from agno.models.openai import OpenAIChat


def get_user_profile() -> str:
    """获取并返回指定用户 ID 的用户档案。

    Args:
        user_id: 要检索的用户 ID
    """

    # 从数据库中获取用户信息（此处为示例）
    user_profile = {
      "name": "John Doe",
      "experience_level": "senior",
    }

    return json.dumps(user_profile, indent=4)

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    dependencies={"user_profile": get_user_profile},
    # 将整个依赖项字典添加到用户消息中
    add_dependencies_to_context=True,
    markdown=True,
)

agent.print_response(
    "Get the user profile for the user with ID 123 and tell me about their experience level.",
    stream=True,
)
# 也可以在调用 print_response 时传入依赖项
# agent.print_response(
#     "Get the user profile for the user with ID 123 and tell me about their experience level.",
#     dependencies={"user_profile": get_user_profile},
#     stream=True,
# )
这会将整个依赖项字典插入到用户消息中，位于 `` 标签之间。 新的用户消息看起来如下： 
1
2
3
4
5
6
7
Get the user profile for the user with ID 123 and tell me about their experience level.                                                       
                                                                                                                                                 
<additional context>                                                                                                                     
{                                                                                                                                        
"user_profile": "{\n    \"name\": \"John Doe\",\n    \"experience_level\": \"senior\"\n}"                                              
}                                                                                                                                        
</additional context>
你可以在以下方法中传入 `dependencies` 和 `add_dependencies_to_context` 参数： `run()`、`arun()`、`print_response()`、`aprint_response()`。

Agno 的 “Dependencies” 机制本质上是一种 轻量级依赖注入（Dependency Injection），
它允许在代理运行前动态加载变量、函数结果或外部数据（如 API 响应、数据库记录、用户信息等）。
这样可以让 LLM 代理在执行时拥有实时、个性化的上下文，而无需手动拼接 prompt。

智能体状态

状态（State） 是指在一个会话（session）的多次运行之间需要保持的数据。

一个常见的 Agent 用例是帮助用户管理各种列表或信息，例如：购物清单、待办事项清单、心愿单等。
这些内容都可以通过 session_state 来轻松管理。

Agent 可以在工具调用（tool calls）或指令中访问或更新 session_state，并在系统消息中将其暴露给模型。

然后，该会话状态会被持久化到配置的数据库中，并在同一会话的多次运行之间保持一致。

**理解 Agno 中的“无状态性”**： Agno 中的Agent本身在不同的会话或运行之间并不直接维护工作状态，但它提供了完善的状态管理机制： * `Agent` 的 `session_state` 参数为新会话提供默认状态模板。 * `get_session_state()` 方法可从数据库中检索特定会话的状态。 * 工作状态按运行（run）进行管理，并在每个会话中持久化保存。 * 智能体实例（或其实例属性）本身在运行过程中不会被修改。

状态管理

Agno 提供了一个功能强大且优雅的状态管理系统。其工作机制如下：

  • 你可以通过 session_state 参数为 Agent 设置默认状态变量（字典格式）。
  • 你可以在工具（tools）或其他函数中更新 session_state。
  • 你可以在系统消息的 description 和 instructions 中引用状态变量，从而让 LLM 了解当前状态。
  • 你可以在调用 agent.run() 时传入 session_state，它的优先级高于 Agent 默认状态。
  • session_state 会在数据库中按会话存储，并在该会话的多次运行间保持。
  • 当你在 agent.run() 中指定 session_id 时，会自动从数据库加载对应会话的状态。

以下是一个管理购物清单（shopping list）的代理示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat

# 定义一个工具，用于向购物清单添加物品
def add_item(session_state, item: str) -> str:
    """添加物品到购物清单"""
    session_state["shopping_list"].append(item)
    return f"The shopping list is now {session_state['shopping_list']}"

# 创建一个维护状态的智能体
agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    # 存储会话和状态的数据库
    db=SqliteDb(db_file="tmp/agents.db"),
    # 初始化会话状态（空购物清单）
    session_state={"shopping_list": []},
    tools=[add_item],
    # 可以在指令中引用状态变量
    instructions="Current state (shopping list) is: {shopping_list}",
    markdown=True,
)

# 示例使用
agent.print_response("Add milk, eggs, and bread to the shopping list", stream=True)
print(f"Final session state: {agent.get_session_state()}")
`session_state` 变量会自动作为参数传递给工具函数。 任何对它的修改都会自动同步到共享状态中。 在团队模式下（Team），会话状态会在团队成员之间共享。 详情参见 [Teams](https://docs.agno.com/concepts/teams/state)。

在同一会话中保持状态

使用 sessions 的最大优势之一，就是能在同一会话的多次运行中保持状态。
例如，一个帮助用户维护购物清单的 Agent。

要让状态在多次运行之间持久化，你必须通过 `db` 参数配置存储。 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from textwrap import dedent
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat

# 定义管理购物清单的工具
def add_item(session_state, item: str) -> str:
    """添加物品"""
    if item.lower() not in [i.lower() for i in session_state["shopping_list"]]:
        session_state["shopping_list"].append(item)
        return f"Added '{item}' to the shopping list"
    else:
        return f"'{item}' is already in the shopping list"

def remove_item(session_state, item: str) -> str:
    """删除物品"""
    for i, list_item in enumerate(session_state["shopping_list"]):
        if list_item.lower() == item.lower():
            session_state["shopping_list"].pop(i)
            return f"Removed '{list_item}' from the shopping list"
    return f"'{item}' was not found in the shopping list"

def list_items(session_state) -> str:
    """列出购物清单中的所有物品"""
    shopping_list = session_state["shopping_list"]
    if not shopping_list:
        return "The shopping list is empty."
    items_text = "\n".join([f"- {item}" for item in shopping_list])
    return f"Current shopping list:\n{items_text}"

# 创建一个带状态的购物清单管理 Agent
agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    session_state={"shopping_list": []},
    db=SqliteDb(db_file="tmp/example.db"),
    tools=[add_item, remove_item, list_items],
    instructions=dedent("""\
        Your job is to manage a shopping list.

        The shopping list starts empty. You can add items, remove items by name, and list all items.

        Current shopping list: {shopping_list}
    """),
    markdown=True,
)

# 示例使用
agent.print_response("Add milk, eggs, and bread to the shopping list", stream=True)
print(f"Session state: {agent.get_session_state()}")
agent.print_response("I got bread", stream=True)
agent.print_response("I need apples and oranges", stream=True)
agent.print_response("whats on my list?", stream=True)
agent.print_response("Clear everything and start with bananas and yogurt", stream=True)

会话状态

Agno 还支持让智能体自动更新会话状态。
只需设置参数 enable_agentic_state=True。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

agent = Agent(
    db=SqliteDb(db_file="tmp/agents.db"),
    model=OpenAIChat(id="gpt-5-mini"),
    session_state={"shopping_list": []},
    add_session_state_to_context=True,  # 让代理能够感知会话状态
    enable_agentic_state=True,          # 自动添加管理状态的工具
)

agent.print_response("Add milk, eggs, and bread to the shopping list", stream=True)
print(f"Session state: {agent.get_session_state()}")
请务必设置 `add_session_state_to_context=True`， 以便代理能够访问并感知当前会话状态。

在指令中使用状态

你可以在指令中引用会话状态中的变量。

不要使用 Python 的 f-string 语法（`f"{}"`）， 直接写 `{key}` 即可，Agno 会自动替换。 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from textwrap import dedent
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

agent = Agent(
    db=SqliteDb(db_file="tmp/agents.db"),
    model=OpenAIChat(id="gpt-5-mini"),
    session_state={"user_name": "John"},
    instructions="Users name is {user_name}",
    markdown=True,
)

agent.print_response("What is my name?", stream=True)

在运行时切换状态

当你在 agent.run() 时传入 session_id，
代理会自动切换到对应的会话，并加载该会话的状态。
这在为不同用户维持独立会话时非常有用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

agent = Agent(
    db=SqliteDb(db_file="tmp/agents.db"),
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Users name is {user_name} and age is {age}",
)

# 用户1
agent.print_response("What is my name?", session_id="user_1_session_1", user_id="user_1", session_state={"user_name": "John", "age": 30})
agent.print_response("How old am I?", session_id="user_1_session_1", user_id="user_1")

# 用户2
agent.print_response("What is my name?", session_id="user_2_session_1", user_id="user_2", session_state={"user_name": "Jane", "age": 25})
agent.print_response("How old am I?", session_id="user_2_session_1", user_id="user_2")

覆盖数据库中的状态

默认情况下，如果在运行时传入了 session_state，
Agno 会将新的状态与数据库中的状态合并。

如果你希望覆盖数据库中的状态，可以启用以下参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    db=SqliteDb(db_file="tmp/agents.db"),
    markdown=True,
    session_state={},
    add_session_state_to_context=True,
    overwrite_db_session_state=True,  # 允许覆盖数据库状态
)

agent.print_response(
    "Can you tell me what's in your session_state?",
    session_state={"shopping_list": ["Potatoes"]},
    stream=True,
)
print(f"Stored session state: {agent.get_session_state()}")

agent.print_response(
    "Can you tell me what is in your session_state?",
    session_state={"secret_number": 43},
    stream=True,
)
print(f"Stored session state: {agent.get_session_state()}")

✅ 总结：
Agno 的状态系统让每个 Agent 都能像“有记忆的助手”一样，
在多轮对话中保存并使用上下文信息（如购物清单、用户资料等）。
它通过 session_state 与数据库配合，实现了持久化、可控、可覆盖的状态管理。

存储

为什么我们需要会话存储？

智能体是临时且无状态（stateless）的。当你运行一个智能体时，它的状态不会被自动保存。
在生产环境中，我们通常通过 API 来调用（或触发）智能体，并希望能在多次请求之间延续同一个会话。

Storage（存储）的作用是将会话历史与状态保存到数据库中，使我们能够从上次中断的地方继续运行。

除此之外，存储还可以让我们：

  • 检查与评估智能体的会话；
  • 提取 few-shot 示例；
  • 构建内部监控工具。

换句话说，它让我们能够观察数据本身，从而构建更优秀的智能体。

只需为Agent、Team或Workflow提供一个数据库驱动（DB driver），Agno 就会自动完成剩下的工作。
你可以使用 SQLite、Postgres、MongoDB 或任何其他数据库。

下面是一个展示智能体在多次执行间持久化状态的简单示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
from rich.pretty import pprint

agent = Agent(
     model=OpenAIChat(id="gpt-5-mini"),
    # 固定会话 ID，以便在多次运行间保持同一会话
    session_id="fixed_id_for_demo",
    db=SqliteDb(db_file="tmp/data.db"),
    # 让智能体能够访问会话历史
    add_history_to_context=True,
    num_history_runs=3,
)

agent.print_response("What was my last question?")
agent.print_response("What is the capital of France?")
agent.print_response("What was my last question?")
pprint(agent.get_messages_for_session())

首次运行时，智能体无法回答 “What was my last question?”，
但再次运行后，它就能回答正确。
因为我们为其设置了固定的 session_id，智能体会在每次运行时延续相同的会话上下文。

存储的优势

存储通常是「智能体工程」中被低估但极其关键的部分。
在生产系统中，存储的重要性体现在：

  • 延续会话：恢复历史记录，从上次中断处继续；
  • 维护会话列表：便于用户选择继续先前的对话；
  • 保存会话状态：将智能体的运行状态存入数据库或文件，以便后续分析。

但它的价值远不止于此：

  • 存储让我们可以检查智能体的会话数据与评估指标；
  • 通过存储的数据，我们可以提取 few-shot 示例 来优化模型；
  • 存储是构建 内部监控面板与运营工具 的基础。

⚠️ 警告：

存储是智能体基础架构中至关重要的一环，绝不应完全交给第三方服务。
在生产系统中，你几乎总是应该自建存储层来管理智能体的状态与会话。

会话表结构（Session Table Schema）

当为智能体配置了 db 参数后，Agno 会在数据库中创建一个 sessions 表，
用于保存每个会话的信息。

表结构如下：

字段名 类型 说明
session_id str 会话的唯一标识符
session_type str 会话类型
agent_id str 所属智能体的 ID
team_id str 所属团队的 ID
workflow_id str 所属工作流的 ID
user_id str 所属用户的 ID
session_data dict 会话相关数据
agent_data dict 智能体相关数据
team_data dict 团队相关数据
workflow_data dict 工作流相关数据
metadata dict 会话元数据
runs list 会话中的执行记录
summary dict 会话摘要信息
created_at int 会话创建时间戳
updated_at int 会话最后更新时间戳

这些数据最直观地展示在
👉 AgentOS 的会话页面。

记忆

Memory 赋予智能体记住用户信息的能力。

记忆 是智能体上下文的一部分，用于帮助其生成更好、更个性化的回答。

例如：如果用户告诉智能体他们喜欢滑雪，那么智能体在之后的对话中可以引用这一信息，为用户提供更个性化的体验。

用户记忆

下面是一个使用记忆功能的简单示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.postgres import PostgresDb
from rich.pretty import pprint

user_id = "ava"

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"

db = PostgresDb(
  db_url=db_url,
  memory_table="user_memories",  # 可选：指定存储记忆的表名
)

# 初始化智能体
memory_agent = Agent(
    model=OpenAIChat(id="gpt-4.1"),
    db=db,
    # 让智能体具备更新用户记忆的能力
    enable_agentic_memory=True,
    # 或者：在每次回复后自动运行 MemoryManager 来更新记忆
    enable_user_memories=True,
    markdown=True,
)

db.clear_memories()

# 第一次交互
memory_agent.print_response(
    "My name is Ava and I like to ski.",
    user_id=user_id,
    stream=True,
    stream_events=True,
)
print("关于 Ava 的记忆：")
pprint(memory_agent.get_user_memories(user_id=user_id))

# 第二次交互
memory_agent.print_response(
    "I live in san francisco, where should i move within a 4 hour drive?",
    user_id=user_id,
    stream=True,
    stream_events=True,
)
print("关于 Ava 的记忆：")
pprint(memory_agent.get_user_memories(user_id=user_id))
设置 `enable_agentic_memory=True` 后，智能体会获得一个「管理用户记忆」的工具， 该工具的实际处理逻辑由 `MemoryManager` 类负责。 你也可以使用 `enable_user_memories=True`，这样智能体会在**每次用户发消息后自动运行 MemoryManager**， 从而自动更新用户记忆。 想了解更多关于记忆机制的细节，请参阅 [Memory 概览](/concepts/memory/overview)。

知识

知识（Knowledge） 用于存储特定领域的内容，这些内容可以被添加到智能体（Agent）的上下文中，以帮助其做出更好的决策。

Agno 提供了一个通用的知识系统，支持多种形式的内容。 详细信息请参阅 [知识文档](https://docs.agno.com/concepts/knowledge/overview)。

智能体可以在运行时搜索这些知识，以便更好地决策并提供更准确的回答。
这种“按需搜索”的模式被称为 Agentic RAG（智能体检索增强生成）。

**示例：** 如果我们在构建一个 Text2SQL 智能体，我们需要向其提供表结构、列名、数据类型、示例查询等信息，以帮助它生成最优的 SQL 查询。 这些信息不适合全部放入系统提示（system message）中。 因此，我们将这些信息存储为“知识”，让智能体在运行时查询。 智能体借助这些信息，能够生成最优 SQL 查询。这种机制称为 **动态少样本学习（dynamic few-shot learning）**。

智能体的知识系统

Agno 智能体默认使用 Agentic RAG。
这意味着当我们为智能体提供 knowledge（知识）时，它会在运行时搜索该知识库，以检索完成任务所需的特定信息。

示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import asyncio
from agno.agent import Agent
from agno.db.postgres.postgres import PostgresDb
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.pgvector import PgVector

db = PostgresDb(
    db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
    knowledge_table="knowledge_contents",
)

# 创建知识实例
knowledge = Knowledge(
    name="Basic SDK Knowledge Base",
    description="Agno 2.0 知识库实现",
    contents_db=db,
    vector_db=PgVector(
        table_name="vectors",
        db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
        embedder=OpenAIEmbedder(),
    ),
)

# 从 URL 添加内容到知识库
asyncio.run(
    knowledge.add_content_async(
        name="Recipes",
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
        metadata={"user_tag": "Recipes from website"},
    )
)

agent = Agent(
    name="My Agent",
    description="Agno 2.0 智能体实现",
    knowledge=knowledge,
    search_knowledge=True,
)

agent.print_response(
    "How do I make chicken and galangal in coconut milk soup?",
    markdown=True,
)

我们可以通过以下方式为智能体提供知识库访问能力：

  • 设置 search_knowledge=True，为智能体添加 search_knowledge_base() 工具。
    如果你为智能体提供了 knowledge，则默认 search_knowledge=True。
  • 设置 add_knowledge_to_context=True，自动将与用户消息相关的知识引用添加到智能体上下文中（即传统 RAG 模式）。

自定义知识检索

如果你希望完全控制知识库的搜索方式，可以定义自定义的 knowledge_retriever 函数：

1
2
def knowledge_retriever(agent: Agent, query: str, num_documents: Optional[int], **kwargs) -> Optional[list[dict]]:
    ...

示例配置：

1
2
3
4
5
6
7
def knowledge_retriever(agent: Agent, query: str, num_documents: Optional[int], **kwargs) -> Optional[list[dict]]:
    ...

agent = Agent(
    knowledge_retriever=knowledge_retriever,
    search_knowledge=True,
)

该函数会在执行 search_knowledge_base() 时被调用，用于从知识库检索相关内容。

支持异步（async）检索函数。 只需将其定义为异步函数并传递给 `knowledge_retriever` 参数即可。

知识存储

知识内容会被分别存储在 内容数据库（Contents DB） 和 向量数据库（Vector DB） 中。

📘 内容数据库（Contents Database）

内容数据库存储知识条目的名称、描述、元数据及其他信息。
其表结构如下：

字段名 类型 描述
id str 知识内容的唯一标识符
name str 知识内容名称
description str 知识内容描述
metadata dict 知识内容的元数据
type str 内容类型
size int 内容大小（仅适用于文件类型）
linked_to str 链接到的知识内容 ID
access_count int 内容被访问的次数
status str 内容状态
status_message str 状态相关信息
created_at int 创建时间戳
updated_at int 最后更新时间戳
external_id str 外部 ID，用于外部向量库（如 LightRAG）

这些数据可在 AgentOS UI 的知识页面 中查看。

🧭 向量数据库（Vector Databases）

向量数据库用于高效地从海量密集信息中检索相关结果。
它通过嵌入向量搜索的方式快速找到与查询最相关的知识片段。

📥 添加内容（Adding Contents）

将内容添加到知识库的典型流程如下：

根据内容类型使用相应的读取器（Reader）解析内容。 将内容分割成较小的片段（Chunk），以提高检索相关性。 将每个片段转换为向量并存储到向量数据库中。

例如，将一个 PDF 文件添加到知识库：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
knowledge = Knowledge(
    name="Basic SDK Knowledge Base",
    description="Agno 2.0 知识库实现",
    vector_db=vector_db,
    contents_db=contents_db,
)

asyncio.run(
    knowledge.add_content_async(
        name="CV",
        path="cookbook/knowledge/testing_resources/cv_1.pdf",
        metadata={"user_tag": "Engineering Candidates"},
    )
)
详细步骤请参阅 [加载知识库（Loading the Knowledge Base）](/concepts/knowledge/overview#loading-the-knowledge)。 当前知识过滤器（Knowledge Filters）支持以下类型： **PDF**、**PDF_URL**、**Text**、**JSON**、**DOCX**。 详情请见 [知识过滤文档](/concepts/knowledge/filters/overview)。

工具

智能体通过工具来执行操作并与外部系统交互。

工具（Tools）是智能体可以调用的函数，用来完成特定任务。例如：搜索网页、运行 SQL、发送邮件或调用外部 API。
你可以使用任意 Python 函数作为工具，也可以使用 Agno 提供的 预构建工具包（Toolkits）。

基本语法如下：

1
2
3
4
5
6
from agno.agent import Agent

agent = Agent(
    # 添加函数或工具包
    tools=[...],
)

使用工具包

Agno 提供了许多预构建的 工具包（toolkits），可直接添加到你的智能体中。
例如，下面展示了如何使用 DuckDuckGo 工具包来进行网页搜索。

更多工具包请参见 [Toolkits 指南](https://docs.agno.com/concepts/tools/toolkits)。

创建文件 web_search.py

1
2
3
4
5
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(tools=[DuckDuckGoTools()], markdown=True)
agent.print_response("法国现在发生了什么？", stream=True)

安装依赖库：

1
pip install openai ddgs agno

运行智能体：
1
python web_search.py

编写自定义工具

如果你想要更强的控制能力，可以自己编写 Python 函数并作为工具添加给智能体。
例如，下面展示了如何添加一个 get_top_hackernews_stories 工具，用于获取 Hacker News 的热门文章。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# hn_agent.py
import json
import httpx
from agno.agent import Agent

def get_top_hackernews_stories(num_stories: int = 10) -> str:
    """获取 Hacker News 的热门文章。
    参数:
        num_stories (int): 返回的文章数量，默认 10。
    """
    # 获取热门文章 ID
    response = httpx.get('https://hacker-news.firebaseio.com/v0/topstories.json')
    story_ids = response.json()

    # 获取每篇文章的详细信息
    stories = []
    for story_id in story_ids[:num_stories]:
        story_response = httpx.get(f'https://hacker-news.firebaseio.com/v0/item/{story_id}.json')
        story = story_response.json()
        if "text" in story:
            story.pop("text", None)
        stories.append(story)
    return json.dumps(stories)

agent = Agent(tools=[get_top_hackernews_stories], markdown=True)
agent.print_response("请总结 Hacker News 上最热门的 5 篇文章。", stream=True)

📚 延伸阅读：

在工具中访问内置参数（Accessing built-in parameters in tools）

在工具函数中，你可以访问智能体的一些内置属性，如：

  • session_state（会话状态）
  • dependencies（依赖项）
  • agent（当前智能体实例）
  • team（团队信息）

例如：

1
2
3
4
5
6
7
8
9
10
11
12
from agno.agent import Agent

def get_shopping_list(session_state: dict) -> str:
    """获取购物清单"""
    return session_state["shopping_list"]

agent = Agent(
    tools=[get_shopping_list],
    session_state={"shopping_list": ["牛奶", "面包", "鸡蛋"]},
    markdown=True
)
agent.print_response("我的购物清单上有什么？", stream=True)

更多信息请参见 工具内置参数。

MCP工具

Agno 支持 Model Context Protocol (MCP) 工具。

基本语法如下：

1
2
3
4
5
6
7
8
9
10
11
12
from agno.agent import Agent
from agno.tools.mcp import MCPTools

async def run_mcp_agent():
    # 初始化 MCP 工具
    mcp_tools = MCPTools(command=f"uvx mcp-server-git")

    # 连接 MCP 服务端
    await mcp_tools.connect()

    agent = Agent(tools=[mcp_tools], markdown=True)
    await agent.aprint_response("这个项目的许可证是什么？", stream=True)
想了解更多 MCP 工具，请参见 [MCP 工具指南](/concepts/tools/mcp)。

多模态智能体

Agno 智能体支持 文本、图像、音频、视频和文件 作为输入，并且也可以生成这些类型的输出。
这意味着你可以构建能“看图”“听音”“读文档”的智能体。

想了解多模态功能的完整概述，请查看 multimodal 官方文档。

并非所有模型都支持多模态输入和输出。
查看兼容性矩阵 以了解哪些模型具备该能力。

向智能体提供多模态输入（Multimodal Inputs）

下面我们创建一个能理解图像的智能体，它会根据需要自动调用工具。

图像智能体

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# image_agent.py
from agno.agent import Agent
from agno.media import Image
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[DuckDuckGoTools()],
    markdown=True,
)

agent.print_response(
    "请描述这张图片，并告诉我有关它的最新新闻。",
    images=[
        Image(
            url="https://upload.wikimedia.org/wikipedia/commons/0/0c/GoldenGateBridge-001.jpg"
        )
    ],
    stream=True,
)

运行智能体：

1
python image_agent.py

更多详情请见 Image as input。

音频智能体（Audio Agent）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# audio_agent.py
import requests
from agno.agent import Agent
from agno.media import Audio
from agno.models.openai import OpenAIChat

# 下载音频文件
url = "https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav"
response = requests.get(url)
response.raise_for_status()
wav_data = response.content

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini-audio-preview", modalities=["text"]),
    markdown=True,
)

agent.print_response("这段音频中说了什么？", audio=[Audio(content=wav_data, format="wav")])

视频智能体（Video Agent）

目前，Agno 仅在 Gemini 模型中支持视频输入。 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# video_agent.py
from pathlib import Path
from agno.agent import Agent
from agno.media import Video
from agno.models.google import Gemini

agent = Agent(
    model=Gemini(id="gemini-2.0-flash-exp"),
    markdown=True,
)

# 请先下载视频
# wget https://storage.googleapis.com/generativeai-downloads/images/GreatRedSpot.mp4
video_path = Path(__file__).parent.joinpath("GreatRedSpot.mp4")

agent.print_response("请描述这段视频内容。", videos=[Video(filepath=video_path)])

智能体的多模态输出

除了接收多模态输入外，Agno 智能体还可以生成图像、音频等多模态输出。

图像生成（Image Generation）

以下示例展示了如何使用 DALL·E 生成图像：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# image_agent.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.dalle import DalleTools

image_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[DalleTools()],
    description="你是一个可以用 DALL-E 生成图像的智能体。",
    instructions="当用户请求创建图像时，请使用 `create_image` 工具。",
    markdown=True,
)

image_agent.print_response("生成一只白色的暹罗猫的图像。")

images = image_agent.get_images()
if images and isinstance(images, list):
    for image_response in images:
        print(image_response.url)

音频响应（Audio Response）

以下示例展示了如何让智能体同时返回文本与音频响应：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# audio_agent.py
from agno.agent import Agent, RunOutput
from agno.models.openai import OpenAIChat
from agno.utils.audio import write_audio_to_file

agent = Agent(
    model=OpenAIChat(
        id="gpt-5-mini-audio-preview",
        modalities=["text", "audio"],
        audio={"voice": "alloy", "format": "wav"},
    ),
    markdown=True,
)

response: RunOutput = agent.run("请讲一个 5 秒钟的恐怖故事。")

# 保存生成的音频
if response.response_audio is not None:
    write_audio_to_file(
        audio=agent.run_response.response_audio.content,
        filename="tmp/scary_story.wav"
    )

多模态输入 + 多模态输出（Inputs and Outputs Together）

你还可以创建既接收多模态输入、又输出多模态内容的智能体。

音频输入 + 音频输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# audio_agent.py
import requests
from agno.agent import Agent
from agno.media import Audio
from agno.models.openai import OpenAIChat
from agno.utils.audio import write_audio_to_file

# 下载音频
url = "https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav"
response = requests.get(url)
response.raise_for_status()
wav_data = response.content

agent = Agent(
    model=OpenAIChat(
        id="gpt-5-mini-audio-preview",
        modalities=["text", "audio"],
        audio={"voice": "alloy", "format": "wav"},
    ),
    markdown=True,
)

agent.run("这段录音中说了什么？", audio=[Audio(content=wav_data, format="wav")])

# 保存智能体的语音回应
if agent.run_response.response_audio is not None:
    write_audio_to_file(
        audio=agent.run_response.response_audio.content,
        filename="tmp/result.wav"
    )

✅ 总结：

  • Agno 智能体支持多种输入与输出模态；
  • 可实现“看图说话”“语音交互”“视频理解”等应用；
  • 通过模型选择与工具组合，可构建复杂的多模态交互系统。

前置钩子与后置钩子

Agno 提供了一种简单的机制，让你在智能体执行前后插入自定义逻辑，从而实现 输入验证、输出过滤、合规控制、数据增强 等功能。

🧩 Pre-hooks（前置钩子）

前置钩子在智能体处理输入之前执行，让你完全掌控送入 LLM 的数据。

它非常适合：

  • 验证输入是否合法；
  • 移除或脱敏敏感信息；
  • 对输入数据进行格式化或预处理。

✅ 常见使用场景

🔒 输入验证
  • 检查格式、长度或内容；
  • 屏蔽敏感信息；
  • 拒绝违规内容。
⚙️ 数据预处理
  • 转换输入格式或结构；
  • 增加上下文信息；
  • 在输入前应用业务逻辑。

💡 示例：验证输入长度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from agno.agent import Agent
from agno.exceptions import CheckTrigger, InputCheckError
from agno.models.openai import OpenAIChat

# 定义一个前置钩子函数
def validate_input_length(
    run_input,
    session,
    user_id=None,
    debug_mode=None,
):
    """前置钩子：校验输入长度"""
    max_length = 1000
    if len(run_input.input_content) > max_length:
        raise InputCheckError(
            f"输入过长，最大允许 {max_length} 个字符。",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

# 创建智能体并注册 pre-hook
agent = Agent(
    name="My Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    pre_hooks=[validate_input_length],
)

如果输入超过 1000 字符，智能体会在处理前抛出错误，避免无效或危险请求继续执行。

📚 完整示例可见：Pre-hooks and Post-hooks Examples

🧠 Pre-hooks 参数说明

前置钩子会在每次运行时自动执行，Agno 会自动注入以下参数：

  • run_input：本次运行的输入；
  • session：当前会话；
  • user_id：可选的用户标识；
  • debug_mode：调试模式标记。

详细参数请见：Pre-hooks 参考文档

🧩 Post-hooks（后置钩子）

后置钩子在智能体生成响应之后执行，允许你对输出结果进行验证、清理或增强。

它非常适合：

  • 过滤输出中的违规内容；
  • 添加元数据；
  • 进行输出格式转换。

✅ 常见使用场景

🧾 输出验证
  • 检查响应长度、格式、质量；
  • 移除敏感或不合规内容；
  • 确保响应符合业务要求。
🔄 输出转换
  • 为响应添加上下文或元信息；
  • 转换格式以适配不同客户端；
  • 进行二次加工（如摘要、翻译等）。

💡 示例：验证输出长度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.exceptions import CheckTrigger, OutputCheckError
from agno.run.agent import RunOutput

# 定义一个后置钩子函数
def validate_output_length(run_output: RunOutput) -> None:
    """后置钩子：校验输出长度"""
    max_length = 1000
    if len(run_output.content) > max_length:
        raise OutputCheckError(
            f"输出过长，最大允许 {max_length} 个字符。",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

# 创建智能体并注册 post-hook
agent = Agent(
    name="My Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    post_hooks=[validate_output_length],
)

📚 完整示例可见：Post-hooks Examples

🧠 Post-hooks 参数说明

与前置钩子类似，Agno 会在执行后自动注入：

  • run_output：本次执行的输出对象；
  • 其他上下文参数（如会话、配置等）。

详细说明请见：Post-hooks 参考文档

🛡️ Guardrails（防护栏）

Guardrails 是基于 Pre-hooks 的一种常见应用，用于为智能体添加内置安全防护机制。

例如：

  • 防止提示注入；
  • 限制请求内容；
  • 阻止敏感词输入。

详细文档请见：Guardrails

✅ 总结

  • Pre-hooks：在输入前执行，做验证、过滤或增强。
  • Post-hooks：在输出后执行，做清理、格式化或合规检查。
  • 二者结合可构建更安全、可控、可扩展的智能体系统。

防护栏

🛡️ 什么是 Guardrails？

Guardrails 是智能体内置的安全防护机制，用于确保发送给大语言模型（LLM）的输入是安全的，不包含敏感或恶意内容。
它可以在模型运行前自动检测和阻止潜在风险。

🔍 常见用途

Guardrails 通常用于以下几种安全场景：

  • 🧩 PII 检测与脱敏（检测个人隐私信息）
  • 🧱 Prompt 注入防御（防止提示词攻击）
  • 🔓 越狱防御（阻止模型被诱导绕过安全限制）
  • 🧾 数据泄露防护
  • 🚫 NSFW 内容过滤（过滤不适宜内容）

🧰 Agno 内置 Guardrails

Agno 提供了多种开箱即用的防护模块，可直接集成到智能体中：

Guardrail 名称 功能说明
PII Detection Guardrail 检测输入中是否包含个人身份信息（如邮箱、手机号等）。
Prompt Injection Guardrail 检测并阻止 Prompt 注入攻击。
OpenAI Moderation Guardrail 检测违反 OpenAI 内容政策的输入。

✅ 示例：使用 PII 检测防护

1
2
3
4
5
6
7
8
9
10
11
12
13
from agno.guardrails import PIIDetectionGuardrail
from agno.agent import Agent
from agno.models.openai import OpenAIChat

# 初始化内置 Guardrail
pii_guardrail = PIIDetectionGuardrail()

# 创建带防护功能的 Agent
agent = Agent(
    name="Privacy-Protected Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    pre_hooks=[pii_guardrail],  # 在输入前执行
)

📘 查看完整示例：Guardrails Examples

⚙️ 自定义 Guardrails

如果内置防护无法满足需求，你可以自定义 Guardrail，通过继承 BaseGuardrail 类实现自己的检测逻辑。

实现步骤

  1. 继承 BaseGuardrail；
  2. 实现 check() 和（可选）async_check() 方法；
  3. 当检测到违规内容时，抛出 InputCheckError。
Agno 会根据你使用的运行方式自动调用同步或异步版本： - `.run()` → 调用 `check()` - `.arun()` → 调用 `async_check()`

💡 示例：检测输入中是否包含 URL

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import re
from agno.exceptions import CheckTrigger, InputCheckError
from agno.guardrails import BaseGuardrail
from agno.run.agent import RunInput

class URLGuardrail(BaseGuardrail):
    """检测输入中是否包含 URL 的防护。"""

    def check(self, run_input: RunInput) -> None:
        """如果输入包含 URL，则抛出异常。"""
        if isinstance(run_input.input_content, str):
            url_pattern = r'https?://[^\s]+|www\.[^\s]+|[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}[^\s]*'
            if re.search(url_pattern, run_input.input_content):
                raise InputCheckError(
                    "输入内容中包含 URL，不被允许。",
                    check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
                )

    async def async_check(self, run_input: RunInput) -> None:
        """异步版本的检测逻辑。"""
        if isinstance(run_input.input_content, str):
            url_pattern = r'https?://[^\s]+|www\.[^\s]+|[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}[^\s]*'
            if re.search(url_pattern, run_input.input_content):
                raise InputCheckError(
                    "输入内容中包含 URL，不被允许。",
                    check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
                )

🚀 使用自定义 Guardrail

1
2
3
4
5
6
7
8
9
10
11
12
from agno.agent import Agent
from agno.models.openai import OpenAIChat

# 使用自定义 URL 检测防护
agent = Agent(
    name="URL-Protected Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    pre_hooks=[URLGuardrail()],
)

# 会触发 InputCheckError
agent.run("Can you check what's in https://fake.com?")

✅ 总结

类型 作用 示例
内置 Guardrail 开箱即用的标准防护（PII、注入检测等） PIIDetectionGuardrail
自定义 Guardrail 按需编写检测逻辑（如 URL、敏感词等） URLGuardrail
运行机制 作为 pre_hooks 执行 Agent(pre_hooks=[...])

🔐 Guardrails 是智能体安全体系的第一道防线，确保所有传入 LLM 的数据符合你的安全与合规要求。

指标

📊 什么是 Metrics？

当你在 Agno 中运行一个智能体时，系统返回的结果对象 RunOutput 会包含详细的运行指标（metrics），用于帮助你分析：

  • 资源使用情况（如 token 消耗）
  • 性能数据（如响应时间）
  • 工具调用的成本与耗时

这些指标在多个层级上可用：

层级 描述
每条消息（Per Message） 每个消息（如模型回复、工具调用）都有独立指标。
每次运行（Per Run） 每个 RunOutput 对象都有整体运行指标。
每个会话（Per Session） AgentSession 汇总了本次会话中所有运行的总指标。

🧠 示例：如何访问指标

下面的示例展示了如何运行一个智能体，并打印出消息级、运行级、会话级的指标。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.db.sqlite import SqliteDb
from rich.pretty import pprint

agent = Agent(
    model=Gemini(id="gemini-2.5-flash"),
    tools=[DuckDuckGoTools()],
    db=SqliteDb(db_file="tmp/agents.db"),
    markdown=True,
)

# 执行一次智能体任务
run_response = agent.run("What is current news in the world?")

# 输出每条消息的指标
if run_response.messages:
    for message in run_response.messages:
        if message.role == "assistant":
            if message.content:
                print(f"Message: {message.content}")
            elif message.tool_calls:
                print(f"Tool calls: {message.tool_calls}")
            print("---" * 5, "Message Metrics", "---" * 5)
            pprint(message.metrics.to_dict())
            print("---" * 20)

# 输出本次运行的总体指标
print("---" * 5, "Run Metrics", "---" * 5)
pprint(run_response.metrics.to_dict())

# 输出整个会话的累计指标
print("---" * 5, "Session Metrics", "---" * 5)
pprint(agent.get_session_metrics().to_dict())

📈 典型指标字段说明

指标名称 含义
input_tokens 发送给模型的 token 数。
output_tokens 模型输出的 token 数。
total_tokens input_tokens + output_tokens。
audio_input_tokens 音频输入的 token 数（针对多模态模型）。
audio_output_tokens 音频输出的 token 数。
audio_total_tokens 音频输入输出 token 总和。
cache_read_tokens 从缓存中读取的 token 数。
cache_write_tokens 写入缓存的 token 数。
reasoning_tokens 模型推理过程中使用的 token 数（适用于 reasoning 模型）。
duration 整个运行的持续时间（秒）。
time_to_first_token 从请求到第一个 token 生成的耗时（秒）。
provider_metrics 各模型提供商特定的附加指标（如延迟、API 调用状态等）。

🧭 数据层级示意

1
2
3
4
5
6
7
8
9
10
AgentSession
 ├── session_metrics (所有 run 的累计)
 │
 ├── RunOutput #1
 │    ├── run_metrics
 │    ├── messages[assistant/tool/...].metrics
 │
 ├── RunOutput #2
      ├── run_metrics
      ├── messages[assistant/tool/...].metrics

🧩 典型用途

使用场景 说明
💰 成本监控 统计 token 使用量，控制 API 成本。
🧪 性能分析 分析响应耗时、延迟、缓存命中率。
📊 用户分析 跟踪会话中每个用户的模型使用趋势。
⚙️ 调优模型参数 比较不同模型或温度参数下的性能表现。

✅ 总结
Agno 的 Metrics 系统提供了精细化的运行监控机制，让你能够清晰了解智能体的：

  • token 使用与成本；
  • 执行性能；
  • 缓存效率；
  • 模型与工具的具体表现。

这使得你能够基于数据，持续优化智能体的性能与成本效率。

取消运行

在 Agno 中，你可以通过调用 agent.cancel_run() 来取消正在执行的智能体任务。

这在以下场景非常有用：

  • 当任务运行时间过长；
  • 当用户主动中断操作；
  • 当通过 API 或 WebSocket 管理智能体运行时；
  • 当希望限制模型计算成本或时间。

Agno 的取消机制也与 AgentOS 集成，可通过 API 实现远程取消。

🧩 示例：多线程取消运行

下面的完整示例展示了如何：

  1. 在一个线程中启动智能体；
  2. 在另一个线程中延迟几秒后取消它；
  3. 并检测取消状态（RunEvent.run_cancelled）。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import threading
import time

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.run.agent import RunEvent
from agno.run.base import RunStatus


def long_running_task(agent: Agent, run_id_container: dict):
    """
    模拟一个可以被取消的长时间运行任务。
    """
    final_response = None
    content_pieces = []

    for chunk in agent.run(
        "Write a very long story about a dragon who learns to code. "
        "Make it at least 2000 words with detailed descriptions and dialogue. "
        "Take your time and be very thorough.",
        stream=True,  # 启用流式输出以便实时取消
    ):
        if "run_id" not in run_id_container and chunk.run_id:
            run_id_container["run_id"] = chunk.run_id

        if chunk.event == RunEvent.run_content:
            print(chunk.content, end="", flush=True)
            content_pieces.append(chunk.content)
        elif chunk.event == RunEvent.run_cancelled:
            print(f"\n🚫 Run was cancelled: {chunk.run_id}")
            run_id_container["result"] = {
                "status": "cancelled",
                "run_id": chunk.run_id,
                "cancelled": True,
                "content": "".join(content_pieces)[:200] + "..." if content_pieces else "No content before cancellation",
            }
            return
        elif hasattr(chunk, "status") and chunk.status == RunStatus.completed:
            final_response = chunk

    # 正常结束
    if final_response:
        run_id_container["result"] = {
            "status": final_response.status.value if final_response.status else "completed",
            "run_id": final_response.run_id,
            "cancelled": final_response.status == RunStatus.cancelled,
            "content": ("".join(content_pieces)[:200] + "...") if content_pieces else "No content",
        }


def cancel_after_delay(agent: Agent, run_id_container: dict, delay_seconds: int = 3):
    """
    在延迟一段时间后取消运行。
    """
    print(f"⏰ Will cancel run in {delay_seconds} seconds...")
    time.sleep(delay_seconds)

    run_id = run_id_container.get("run_id")
    if run_id:
        print(f"🚫 Cancelling run: {run_id}")
        success = agent.cancel_run(run_id)
        if success:
            print(f"✅ Run {run_id} marked for cancellation")
        else:
            print(f"❌ Failed to cancel run {run_id} (may not exist or already completed)")
    else:
        print("⚠️  No run_id found to cancel")


def main():
    agent = Agent(
        name="StorytellerAgent",
        model=OpenAIChat(id="gpt-5-mini"),
        description="An agent that writes detailed stories",
    )

    print("🚀 Starting agent run cancellation example...")
    print("=" * 50)

    run_id_container = {}

    agent_thread = threading.Thread(
        target=lambda: long_running_task(agent, run_id_container),
        name="AgentRunThread",
    )

    cancel_thread = threading.Thread(
        target=cancel_after_delay,
        args=(agent, run_id_container, 8),  # 8 秒后取消
        name="CancelThread",
    )

    print("🏃 Starting agent run thread...")
    agent_thread.start()

    print("🏃 Starting cancellation thread...")
    cancel_thread.start()

    print("⌛ Waiting for threads to complete...")
    agent_thread.join()
    cancel_thread.join()

    print("\n" + "=" * 50)
    print("📊 RESULTS:")
    print("=" * 50)

    result = run_id_container.get("result")
    if result:
        print(f"Status: {result['status']}")
        print(f"Run ID: {result['run_id']}")
        print(f"Was Cancelled: {result['cancelled']}")
        print(f"Content Preview: {result['content']}")
        if result["cancelled"]:
            print("\n✅ SUCCESS: Run was successfully cancelled!")
        else:
            print("\n⚠️  WARNING: Run completed before cancellation")
    else:
        print("❌ No result obtained - check if cancellation happened during streaming")

    print("\n🏁 Example completed!")


if __name__ == "__main__":
    main()

🔍 工作原理

概念 描述
agent.run(..., stream=True) 启动流式运行，允许边生成边取消。
RunEvent.run_cancelled 表示运行被取消的事件类型。
agent.cancel_run(run_id) 主动请求取消指定运行（需提供 run_id）。
RunStatus.cancelled 运行状态字段，标记已取消。

🧠 注意事项

  • 取消机制通过 事件流（RunEvent） 实现，适合异步或多线程任务。
  • 一旦运行完成或出错，调用 cancel_run() 将不再生效。
  • 若你使用 AgentOS 部署智能体，可直接通过 REST API 或 SDK 调用取消接口。
  • 对于批量任务或流式生成类应用，强烈推荐实现“中断控制”以防模型过度计算。

✅ 总结

Agno 的取消机制让你可以：

  • 在任何阶段中止长时间任务；
  • 在前端或 API 层实现“停止生成”；
  • 节省模型计算成本；
  • 保持智能体响应灵活与安全。

这为构建交互式或多用户 LLM 应用提供了重要的可控性。

自定义日志记录器

Agno 默认自带一个标准化的日志系统，但你也可以用自己定义的日志记录器（logger）替换它。

这在以下场景非常有用：

  • 想要与现有系统的日志标准（如 JSON 格式、结构化日志）集成；
  • 想将日志输出到特定目标（如文件、数据库、Elastic、Grafana 等）；
  • 想为不同类型的组件（Agent / Team / Workflow）使用不同的日志格式。

🛠️ 指定自定义日志记录器

下面示例展示如何替换 Agno 的默认日志系统为自定义日志器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import logging

from agno.agent import Agent
from agno.utils.log import configure_agno_logging, log_info


# 1️⃣ 设置自定义日志器
custom_logger = logging.getLogger("custom_logger")
handler = logging.StreamHandler()  # 控制台输出
formatter = logging.Formatter("[CUSTOM_LOGGER] %(levelname)s: %(message)s")
handler.setFormatter(formatter)
custom_logger.addHandler(handler)
custom_logger.setLevel(logging.INFO)
custom_logger.propagate = False  # 不向上层传播日志


# 2️⃣ 告诉 Agno 使用我们的自定义日志器
configure_agno_logging(custom_default_logger=custom_logger)

# 所有来自 agno.utils.log 的日志都会使用我们的 custom_logger
log_info("This is using our custom logger!")


# 3️⃣ 创建并运行 Agent
agent = Agent()
agent.print_response("What can I do to improve my sleep?")

运行时输出类似：

1
2
3
[CUSTOM_LOGGER] INFO: This is using our custom logger!
[CUSTOM_LOGGER] INFO: Starting Agent run...
[CUSTOM_LOGGER] INFO: Agent responded: “To improve your sleep, try keeping a consistent bedtime...”

🔧 多日志器配置（Multiple Loggers）

你也可以分别为不同组件（Agent、Team、Workflow）定义独立的日志器：

1
2
3
4
5
6
configure_agno_logging(
    custom_default_logger=custom_agent_logger,
    custom_agent_logger=custom_agent_logger,
    custom_team_logger=custom_team_logger,
    custom_workflow_logger=custom_workflow_logger,
)

📘 用法建议：

  • Agent 日志：用于记录智能体的推理、模型调用、输入输出。
  • Team 日志：用于多智能体协作或团队任务追踪。
  • Workflow 日志：用于长流程任务、管道式执行日志。

🧱 使用命名日志器（Named Loggers）

Agno 遵循 Python 的标准日志命名约定。

如果你在应用的日志配置文件中预先定义了这些命名日志器，它们会被 自动识别和使用：

名称 用途
agno.agent 智能体（Agent）日志
agno.team 团队（Team）日志
agno.workflow 工作流（Workflow）日志

例如在你的 logging.conf 中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
[loggers]
keys=root,agno.agent,agno.team,agno.workflow

[logger_agno.agent]
level=INFO
handlers=consoleHandler
qualname=agno.agent
propagate=0

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[formatter_simpleFormatter]
format=[%(name)s] %(levelname)s - %(message)s

Agno 启动后会自动检测并使用这些命名日志器。

💡 提示与最佳实践

  • ✅ 若你使用 FastAPI / Flask 构建后端，可以统一使用应用的 logging 设置。
  • ✅ 可配合 Loguru、structlog 或 JSONFormatter 输出结构化日志。
  • ✅ 在生产环境建议设置日志级别为 INFO 或 WARNING，避免模型调用日志过多。
  • ✅ 支持同时写入多个目标（如控制台 + 文件 + 远程）。

✅ 总结

Agno 的自定义日志系统让你：

  • 灵活接入企业级日志平台；
  • 对智能体运行进行可观测性分析；
  • 保持统一日志格式；
  • 精确控制不同组件的日志输出。

这为构建生产级智能体系统提供了强大的调试与监控支持。