エージェントは作業の最小単位で、狭いスコープと少ない数のツールを持つ場合に最善に機能します。ツールの数がモデルの処理能力を超えたり、複数のコンセプトを処理する必要がある場合、負荷を分散するためにエージェントのチームを使用します。

Agno : ユーザガイド : イントロダクション : マルチエージェント・システム

作成 : クラスキャット・セールスインフォメーション
作成日時 : 07/13/2025
バージョン : Agno 1.7.1

* 本記事は docs.agno.com の以下のページを独自に翻訳した上で、Google Colab 上での動作確認を加えてまとめ直しています :

• User Guide : Introduction : Multi Agent Systems

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

 

Agno ユーザガイド : イントロダクション : マルチエージェント・システム

共通の目標に向かって協力して作業するエージェントのチーム。

 

レベル 4 : 推論してコラボレーション可能なエージェント・チーム

エージェントは作業の最小単位で、狭いスコープと少ない数のツールを持つ場合に最善に機能します。ツールの数がモデルの処理能力を超えたり、複数のコンセプトを処理する必要がある場合、負荷を分散するためにエージェントのチームを使用します。
​
Agno は業界をリードするマルチエージェント・アーキテクチャを提供しており、これは推論、コラボレーション、coordinate 可能なエージェント・チームの構築を可能にします。この例では、半導体マーケットのパフォーマンスを段階的に推論しながら分析するために 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="Handle web search requests and general research",
    model=OpenAIChat(id="gpt-4.1"),
    tools=[DuckDuckGoTools()],
    instructions="Always include sources",
    add_datetime_to_instructions=True,
)

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,
)

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,
    )

依存関係をインストールしてエージェント・チームを実行します。

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 : 状態と決定性を持つエージェント型ワークフロー

ワークフローは、production 環境用に構築された決定論的、ステートフルなマルチエージェント・プログラムです。私たちは pure Python でワークフローを記述し、実行フローを非常に細かく制御できます。

数百のエージェント型システムを構築してきましたが、pure-python の柔軟性や信頼性を提供できるフレームワークやステップベースのアプローチはありません。ループが必要なら while/for を、条件分岐が必要なら if/else を、そして例外処理が必要なら try/except を使用してください。

Note : Because the workflow logic is a python function, AI code editors can vibe code workflows for you. Add https://docs.agno.com as a document source and vibe away.

以前の出力をキャッシュする単純なワークフローが以下です、キャッシュされるもの、ストリーミングされるもの、ログに記録されるもの、そして返されるものなど、すべてのステップを制御できます。

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):
    # Add agents or teams as attributes on the workflow
    agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

    # Write the logic in the `run()` method
    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
        yield from self.agent.run(message, stream=True)

        # Cache the output after response is yielded
        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)

Run the workflow

python level_5_workflow.py

以上