Documentation Index
Fetch the complete documentation index at: https://ikun.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
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
web_agent = Agent(
name="Web Search Agent",
role="处理网络搜索请求和一般研究",
model=OpenAIChat(id="gpt-4.1"),
tools=[DuckDuckGoTools()],
instructions="始终包含来源",
add_datetime_to_instructions=True,
)
finance_agent = Agent(
name="Finance Agent",
role="处理财务数据请求和市场分析",
model=OpenAIChat(id="gpt-4.1"),
tools=[YFinanceTools(stock_price=True, stock_fundamentals=True,analyst_recommendations=True, company_info=True)],
instructions=[
"使用表格显示股价、基本面(市盈率、市值)和建议。",
"清楚注明公司名称和股票代码。",
"专注于提供可行的财务见解。",
],
add_datetime_to_instructions=True,
)
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=[
"协作提供全面的财务和投资见解。",
"考虑基本面分析和市场情绪。",
"使用表格和图表清晰专业地展示数据。",
"以结构化、易于理解的格式呈现调查结果。",
"仅输出最终的综合分析,不输出单独的代理响应。",
],
markdown=True,
show_members_responses=True,
enable_agentic_context=True,
add_datetime_to_instructions=True,
success_criteria="该团队提供了完整的财务分析,其中包含数据、可视化、风险评估和可通过量化分析及市场研究支持的可行投资建议。",
)
if __name__ == "__main__":
reasoning_finance_team.print_response("""比较科技行业巨头 (AAPL, GOOGL, MSFT) 的表现:
1. 获取所有三家公司的财务数据
2. 分析影响科技行业的近期新闻
3. 计算比较指标和相关性
4. 推荐投资组合配置权重""",
stream=True,
show_full_reasoning=True,
stream_intermediate_steps=True,
)
安装依赖项并运行代理团队
安装依赖项
uv pip install -U agno anthropic openai duckduckgo-search yfinance
导出您的 API 密钥
export ANTHROPIC_API_KEY=sk-***
export OPENAI_API_KEY=sk-***
5 级:具有状态和确定性的代理工作流
工作流是为生产应用程序构建的确定性、有状态的多代理程序。我们在纯 Python 中编写工作流,从而让我们对执行流程拥有极大的控制权。
在构建了数百个代理系统之后,没有哪个框架或基于步骤的方法能像纯 Python 那样提供灵活性和可靠性。想要循环 - 使用 while/for,想要条件 - 使用 if/else,想要异常处理 - 使用 try/except。
因为工作流逻辑是一个 Python 函数,AI 代码编辑器可以为您生成代码工作流。将 https://docs.agno.com 添加为文档源,即可开始畅写。
这是一个缓存了先前输出的简单工作流,您可以控制每个步骤:缓存什么内容、流式传输什么内容、记录什么内容以及返回什么内容。
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):
# 将代理或团队添加为工作流的属性
agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))
# 在 `run()` 方法中编写逻辑
def run(self, message: str) -> Iterator[RunResponse]:
logger.info(f"正在为 '{message}' 检查缓存")
# 检查输出是否已缓存
if self.session_state.get(message):
logger.info(f"'{message}' 缓存命中")
yield RunResponse(
run_id=self.run_id, content=self.session_state.get(message)
)
return
logger.info(f"'{message}' 缓存未命中")
# 运行代理并生成响应
yield from self.agent.run(message, stream=True)
# 在生成响应后缓存输出
self.session_state[message] = self.agent.run_response.content
if __name__ == "__main__":
workflow = CacheWorkflow()
# 运行工作流(耗时约 1 秒)
response: Iterator[RunResponse] = workflow.run(message="给我讲个笑话。")
# 打印响应
pprint_run_response(response, markdown=True, show_time=True)
# 再次运行工作流(因为已缓存,所以立即完成)
response: Iterator[RunResponse] = workflow.run(message="给我讲个笑话。")
# 打印响应
pprint_run_response(response, markdown=True, show_time=True)
运行工作流
python level_5_workflow.py
下一步