​
4 级：能够推理和协作的代理团队

代理是工作的原子单位，当它们的作用域窄且工具数量少时，效果最佳。当工具数量超出模型处理能力或需要处理多个概念时，请使用代理团队来分摊负载。

Agno 提供行业领先的多代理架构，使您能够构建能够推理、协作和协调的代理团队。在此示例中，我们将构建一个由 2 个代理组成的团队，以逐步分析半导体市场的表现。

level_4_team.py
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,
    )

安装依赖项并运行代理团队

1

安装依赖项

uv pip install -U agno anthropic openai duckduckgo-search yfinance
2

导出您的 API 密钥

export ANTHROPIC_API_KEY=sk-***
export OPENAI_API_KEY=sk-***
3

运行代理团队

python level_4_team.py

​
5 级：具有状态和确定性的代理工作流

工作流是为生产应用程序构建的确定性、有状态的多代理程序。我们在纯 Python 中编写工作流，从而让我们对执行流程拥有极大的控制权。

在构建了数百个代理系统之后，没有哪个框架或基于步骤的方法能像纯 Python 那样提供灵活性和可靠性。想要循环 - 使用 while/for，想要条件 - 使用 if/else，想要异常处理 - 使用 try/except。

因为工作流逻辑是一个 Python 函数，AI 代码编辑器可以为您生成代码工作流。

将 https://docs.agno.com 添加为文档源，即可开始畅写。

这是一个缓存了先前输出的简单工作流，您可以控制每个步骤：缓存什么内容、流式传输什么内容、记录什么内容以及返回什么内容。

level_5_workflow.py
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

​
下一步

  • 查看 Agent Playground 与您的代理、团队和工作流进行交互。
  • 了解如何 监控 您的代理、团队和工作流。
  • 在 社区 中获得帮助。