一、基本功能介绍
Agno 是一个 python 框架，用于构建具有共享内存、知识和推理的多代理系统。
工程师和研究人员使用 Agno 构建：
- 1 级： 带有工具和说明的代理
- 2 级： 具有知识和存储的代理
- 3 级： 具有记忆和推理能力的代理
- 4 级： 可以推理和协作的代理团队
- 5 级： 具有状态和确定性的代理工作流
例： 使用 YFinance API 回答问题的 1 级推理代理：
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.reasoning import ReasoningTools
from agno.tools.yfinance import YFinanceTools
reasoning_agent = Agent(
model=Claude(id="claude-sonnet-4-20250514"),
tools=[
# 工具一： ReasoningTools对于返回内容进行美化
ReasoningTools(add_instructions=True),
# 工具二： 财经新闻获取
YFinanceTools(stock_price=True, analyst_recommendations=True, company_info=True, company_news=True),
],
instructions="Use tables to display data.",
markdown=True,
)
二、Getting Started
为什么选择 Agno？
Agno 将帮助您构建一流的高性能代理系统，为您节省数小时的研究和样板工作。以下是使 Agno 与众不同的一些关键功能：
- 与模型无关 ：Agno 为 23+ 模型提供商提供统一接口，无锁定。
- 高性能 ：代理在 ~3μs 内实例化，平均使用 ~6.5Kib 内存。
- 推理是一等公民 ：推理提高了可靠性，是复杂自主代理的必备条件。Agno 支持 3 种推理方法：Reasoning Models、ReasoningTools 或我们的自定义思维链方法。
- 原生多模态 ：Agno Agent 原生是多模态的，它们接受文本、图像、音频和视频作为输入，并生成文本、图像、音频和视频作为输出。
- 高级多代理架构 ：Agno 提供行业领先的多代理架构（ 代理团队 ），具有推理、内存和共享上下文。
- 内置代理搜索 ：代理可以在运行时使用 20+ 矢量数据库搜索信息。Agno 提供最先进的 Agentic RAG， 完全异步且高性能。
- 内置内存和会话存储 ：代理附带内置的存储和内存驱动程序，为您的代理提供长期内存和会话存储。
- 结构化输出 ：Agno 代理可以使用模型提供的结构化输出或 json_mode 返回全类型响应。
- 预构建的 FastAPI 路由 ：构建代理后，使用预构建的 FastAPI 路由为它们提供服务。0 到 3 分钟即可投入生产。
- 监控 ：在 agno.com 上实时监控代理会话和性能。
三、什么是代理？
代理是自主运行的 AI 程序。
传统软件遵循预编程的步骤序列。代理使用机器学习模型动态确定其行动方案，其核心组件包括：
- Model： 控制执行流程。它决定是推理、行动还是回应
- 工具： 使 Agent 能够执行作并与外部系统交互
- 说明： 是我们如何对 Agent 进行编程，教它如何使用工具和响应
代理还具有内存 、 知识 、 存储和推理能力：
- 推理： 使代理能够在响应之前“思考”并“分析”工具调用的结果，从而提高响应的可靠性和质量。
- 知识： 是特定于域的信息，代理可以在运行时搜索这些信息，以做出更好的决策并提供准确的响应 （RAG）。知识存储在向量数据库中，这种运行时搜索模式称为 Agentic RAG/Agentic Search。
- 存储： 代理 （Agent） 使用该数据库将会话历史记录和状态保存在数据库中。模型 API 是无状态的，存储使我们能够从上次中断的地方继续对话。这使得 Agent 有状态，从而支持多轮次的长期对话。
- 内存： 使代理能够存储和调用以前交互的信息，使他们能够了解用户偏好并个性化他们的响应。
第 1 级：带有工具和说明的代理
最简单的 Agent 有一个模型、一个工具和一个指令。让我们构建一个可以使用 yfinance 库获取数据的 Agent，以及在表中显示结果的说明。
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.yfinance import YFinanceTools
agent = Agent(
model=Claude(id="claude-sonnet-4-20250514"),
tools=[YFinanceTools(stock_price=True)],
instructions="Use tables to display data. Don't include any other text.",
markdown=True,
)
agent.print_response("What is the stock price of Apple?", stream=True)
第 2 级：具有知识和存储能力的代理
知识： 虽然模型有大量的训练数据，但我们几乎总是需要为它们提供特定于领域的信息，以做出更好的决策并提供准确的响应 （RAG）。我们将此信息存储在矢量数据库中，并让 Agent 在运行时搜索它。
存储： 模型 API 是无状态的， 存储驱动程序将聊天历史记录和状态保存到数据库中。当 Agent 运行时，它会从数据库中读取聊天历史记录和状态，并将其添加到消息列表中，从而恢复对话并使 Agent 有状态。
在此示例中，我们将使用：
- UrlKnowledge 将 Agno 文档加载到 LanceDB，使用 OpenAI 进行嵌入。
- SqliteStorage 将 Agent 的会话历史记录和状态保存在数据库中。
from agno.agent import Agent
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.url import UrlKnowledge
from agno.models.anthropic import Claude
from agno.storage.sqlite import SqliteStorage
from agno.vectordb.lancedb import LanceDb, SearchType
# Load Agno documentation in a knowledge base
# You can also use `https://docs.agno.com/llms-full.txt` for the full documentation
# 将公网的文档 切分，向量化存储到LanceDb
knowledge = UrlKnowledge(
urls=["https://docs.agno.com/introduction.md"],
vector_db=LanceDb(
uri="tmp/lancedb",
table_name="agno_docs",
search_type=SearchType.hybrid,
# Use OpenAI for embeddings
embedder=OpenAIEmbedder(id="text-embedding-3-small", dimensions=1536),
),
)
# Store agent sessions in a SQLite database
# 将 agent的历史会话 存储到 sqlite数据库中
storage = SqliteStorage(table_name="agent_sessions", db_file="tmp/agent.db")
# 创建agent
agent = Agent(
name="Agno Assist",
model=Claude(id="claude-sonnet-4-20250514"),
instructions=[
"Search your knowledge before answering the question.",
"Only include the output in your response. No other text.",
],
knowledge=knowledge,
storage=storage,
add_datetime_to_instructions=True,
# Add the chat history to the messages
add_history_to_messages=True,
# Number of history runs
num_history_runs=3,
markdown=True,
)
if __name__ == "__main__":
# Load the knowledge base, comment out after first run
# Set recreate to True to recreate the knowledge base if needed
agent.knowledge.load(recreate=False)
agent.print_response("What is Agno?", stream=True)
第 3 级：具有记忆力和推理能力的代理
推理： 使代理人能够 “思考”和“分析”，提高可靠性和质量。ReasoningTools 是提高 Agent 响应质量的最佳方法之一。
内存： 使 Agent 能够对用户首选项进行分类、存储和调用，从而个性化他们的响应。记忆有助于 Agent 构建角色并从以前的交互中学习。
from agno.agent import Agent
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory
from agno.models.anthropic import Claude
from agno.tools.reasoning import ReasoningTools
from agno.tools.yfinance import YFinanceTools
# 创建 Memory组件
memory = Memory(
# Use any model for creating and managing memories
# 使用大模型创建和管理上下文记忆
model=Claude(id="claude-sonnet-4-20250514"),
# Store memories in a SQLite database
# 使用sqllite作为上下文存储数据库
db=SqliteMemoryDb(table_name="user_memories", db_file="tmp/agent.db"),
# We disable deletion by default, enable it if needed
# 删除上下文记忆
delete_memories=True,
clear_memories=True,
)
agent = Agent(
model=Claude(id="claude-sonnet-4-20250514"),
tools=[
# 推理工具
ReasoningTools(add_instructions=True),
# 查询金融数据工具
YFinanceTools(stock_price=True, analyst_recommendations=True, company_info=True, company_news=True),
],
# User ID for storing memories, `default` if not provided
user_id="ava",
instructions=[
"Use tables to display data.",
"Include sources in your response.",
"Only include the report in your response. No other text.",
],
memory=memory,
# Let the Agent manage its memories
enable_agentic_memory=True,
markdown=True,
)
if __name__ == "__main__":
# This will create a memory that "ava's" favorite stocks are NVIDIA and TSLA
agent.print_response(
"My favorite stocks are NVIDIA and TSLA",
stream=True,
show_full_reasoning=True,
stream_intermediate_steps=True,
)
# This will use the memory to answer the question
agent.print_response(
"Can you compare my favorite stocks?",
stream=True,
show_full_reasoning=True,
stream_intermediate_steps=True,
)
四、多代理系统
代理团队为实现共同目标而共同努力。
第 4 级：能够推理和协作的代理团队
代理是原子工作单元，当它们的范围狭窄且工具数量较少时，效果最佳。当工具数量超出模型可以处理的范围，或者您需要处理多个概念时，请使用代理团队来分散负载。
Agno 提供行业领先的多代理架构，允许您构建可以推理、协作和协调的代理团队。在这个例子中，我们将建立一个由 2 个代理组成的团队来分析半导体市场的表现，一步一步地推理。
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.reasoning import ReasoningTools
from agno.tools.yfinance import YFinanceTools
# 创建第一个agent， 主要作用是处理 网络搜索请求，并且生成 研究结果
web_agent = Agent(
name="Web Search Agent",
role="Handle web search requests and general research",
model=OpenAIChat(id="gpt-4.1"),
tools=[DuckDuckGoTools()],
instructions="Always include sources",
add_datetime_to_instructions=True,
)
# 创建第二个agent, 主要作用是处理财经金融数据请求，和市场分析， 主要用的tool是 YFinanceTools
finance_agent = Agent(
name="Finance Agent",
role="Handle financial data requests and market analysis",
model=OpenAIChat(id="gpt-4.1"),
tools=[YFinanceTools(stock_price=True, stock_fundamentals=True,analyst_recommendations=True, company_info=True)],
instructions=[
"Use tables to display stock prices, fundamentals (P/E, Market Cap), and recommendations.",
"Clearly state the company name and ticker symbol.",
"Focus on delivering actionable financial insights.",
],
add_datetime_to_instructions=True,
)
# 创建team， 成员包括刚才定义的web_agent, finance_agent
reasoning_finance_team = Team(
name="Reasoning Finance Team",
mode="coordinate",
model=Claude(id="claude-sonnet-4-20250514"),
members=[web_agent, finance_agent],
tools=[ReasoningTools(add_instructions=True)],
instructions=[
"Collaborate to provide comprehensive financial and investment insights",
"Consider both fundamental analysis and market sentiment",
"Use tables and charts to display data clearly and professionally",
"Present findings in a structured, easy-to-follow format",
"Only output the final consolidated analysis, not individual agent responses",
],
markdown=True,
show_members_responses=True,
enable_agentic_context=True,
add_datetime_to_instructions=True,
success_criteria="The team has provided a complete financial analysis with data, visualizations, risk assessment, and actionable investment recommendations supported by quantitative analysis and market research.",
)
if __name__ == "__main__":
reasoning_finance_team.print_response("""Compare the tech sector giants (AAPL, GOOGL, MSFT) performance:
1. Get financial data for all three companies
2. Analyze recent news affecting the tech sector
3. Calculate comparative metrics and correlations
4. Recommend portfolio allocation weights""",
stream=True,
show_full_reasoning=True,
stream_intermediate_steps=True,
)
第 5 级：具有状态和确定性的代理工作流
工作流是为生产应用程序构建的确定性、有状态的多代理程序。我们用纯 python 编写工作流，从而对执行流程进行极端控制。我们已经构建了 100 多个代理系统， 没有任何框架或基于步骤的方法会为您提供纯 Python 的灵活性和可靠性 。想要循环 - 使用 while/for，想要条件 - 使用 if/else，想要异常处理 - 使用 try/except。
这是一个缓存先前输出的简单工作流，您可以控制每个步骤：缓存哪些内容、流式传输哪些内容、记录哪些内容以及返回哪些内容。
from typing import Iterator
from agno.agent import Agent, RunResponse
from agno.models.openai import OpenAIChat
from agno.utils.log import logger
from agno.utils.pprint import pprint_run_response
from agno.workflow import Workflow
class CacheWorkflow(Workflow):
# Add agents or teams as attributes on the workflow
# 在工作流中增加 agent
agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))
# Write the logic in the `run()` method
# 在run()中实现具体逻辑
def run(self, message: str) -> Iterator[RunResponse]:
logger.info(f"Checking cache for '{message}'")
# Check if the output is already cached
if self.session_state.get(message):
logger.info(f"Cache hit for '{message}'")
yield RunResponse(
run_id=self.run_id, content=self.session_state.get(message)
)
return
logger.info(f"Cache miss for '{message}'")
# Run the agent and yield the response
# 运行agent生成 response
yield from self.agent.run(message, stream=True)
# Cache the output after response is yielded
# response生成后， 将生成内容缓存
self.session_state[message] = self.agent.run_response.content
if __name__ == "__main__":
workflow = CacheWorkflow()
# Run workflow (this is takes ~1s)
response: Iterator[RunResponse] = workflow.run(message="Tell me a joke.")
# Print the response
pprint_run_response(response, markdown=True, show_time=True)
# Run workflow again (this is immediate because of caching)
response: Iterator[RunResponse] = workflow.run(message="Tell me a joke.")
# Print the response
pprint_run_response(response, markdown=True, show_time=True)
