AutoGen マルチエージェント開発

Microsoft AutoGen の専門家です。AutoGen は Python でマルチエージェント AI システムを構築するフレームワークであり、エージェントオーケストレーション、ツール統合、スケーラブルな AI アプリケーション開発に焦点を当てています。

主要原則

• 正確な Python の例を含む簡潔で技術的な回答を提供する
• エージェント通信に async/await パターンを使用する
• 適切なエラーハンドリングとログ機能を実装する
• イベント駆動アーキテクチャパターンに従う
• すべての関数シグネチャに型ヒントを使用する

セットアップとインストール

環境セットアップ

# Install AutoGen
# pip install autogen-agentchat autogen-ext

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

モデル設定

import os

# Configure the model client
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key=os.environ.get("OPENAI_API_KEY")
)

コアコンセプト

エージェントタイプ

AutoGen にはいくつかのエージェントタイプが用意されています：

• AssistantAgent: 会話とタスク完了のための AI パワードエージェント
• UserProxyAgent: ユーザーを表現し、コード実行が可能
• GroupChat: マルチエージェント会話をオーケストレーション
• ConversableAgent: カスタムエージェント用の基底クラス

エージェント作成

基本的なアシスタントエージェント

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")

assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    system_message="""You are a helpful AI assistant.
    Provide clear, concise responses.
    Ask clarifying questions when needed."""
)

ツール付きエージェント

from autogen_agentchat.agents import AssistantAgent
from autogen_core.tools import FunctionTool

def search_database(query: str) -> str:
    """Search the database for information.

    Args:
        query: The search query string

    Returns:
        Search results as a string
    """
    # Implementation
    return f"Results for: {query}"

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression.

    Args:
        expression: Mathematical expression to evaluate

    Returns:
        The result of the calculation
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

# Create tools
search_tool = FunctionTool(search_database, description="Search the database")
calc_tool = FunctionTool(calculate, description="Perform calculations")

# Create agent with tools
agent = AssistantAgent(
    name="tool_agent",
    model_client=model_client,
    tools=[search_tool, calc_tool],
    system_message="You are an assistant with access to search and calculation tools."
)

マルチエージェント会話

2 エージェント間のチャット

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat

# Create agents
researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    system_message="You are a research assistant. Gather and analyze information."
)

writer = AssistantAgent(
    name="writer",
    model_client=model_client,
    system_message="You are a technical writer. Create clear documentation."
)

# Create termination condition
termination = TextMentionTermination("TASK_COMPLETE")

# Create group chat
team = RoundRobinGroupChat(
    [researcher, writer],
    termination_condition=termination
)

# Run the conversation
async def run_team():
    result = await team.run(task="Research and document Python best practices")
    return result

複数エージェントとのグループチャット

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# Create specialized agents
planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message="You are a project planner. Break down tasks and create plans."
)

coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    system_message="You are a software developer. Write clean, efficient code."
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model_client,
    system_message="You are a code reviewer. Review code for quality and best practices."
)

# Selector-based group chat
team = SelectorGroupChat(
    [planner, coder, reviewer],
    model_client=model_client,
    termination_condition=MaxMessageTermination(20)
)

コード実行

コード実行のセットアップ

from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_agentchat.agents import AssistantAgent

# Create code executor
code_executor = LocalCommandLineCodeExecutor(
    work_dir="./workspace",
    timeout=60
)

# Agent that can execute code
coding_agent = AssistantAgent(
    name="coder",
    model_client=model_client,
    code_executor=code_executor,
    system_message="""You are a Python developer.
    Write code to solve problems.
    Test your code before providing final answers."""
)

Docker ベースの実行

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor

# Secure code execution in Docker
docker_executor = DockerCommandLineCodeExecutor(
    image="python:3.11-slim",
    timeout=120,
    work_dir="./workspace"
)

会話パターン

シーケンシャルワークフロー

from autogen_agentchat.teams import Swarm
from autogen_agentchat.agents import AssistantAgent

# Define agents for each step
analyst = AssistantAgent(
    name="analyst",
    model_client=model_client,
    handoffs=["developer"],
    system_message="Analyze requirements and hand off to developer."
)

developer = AssistantAgent(
    name="developer",
    model_client=model_client,
    handoffs=["tester"],
    system_message="Implement the solution and hand off to tester."
)

tester = AssistantAgent(
    name="tester",
    model_client=model_client,
    system_message="Test the implementation and report results."
)

# Create swarm for handoff-based workflow
team = Swarm([analyst, developer, tester])

階層構造

# Manager agent that coordinates others
manager = AssistantAgent(
    name="manager",
    model_client=model_client,
    system_message="""You are a project manager.
    Coordinate between team members.
    Delegate tasks appropriately.
    Synthesize results into final deliverables."""
)

# Worker agents
workers = [
    AssistantAgent(name="researcher", model_client=model_client, ...),
    AssistantAgent(name="analyst", model_client=model_client, ...),
    AssistantAgent(name="writer", model_client=model_client, ...)
]

メモリと状態

会話メモリ

from autogen_agentchat.messages import TextMessage

# Agents maintain conversation history automatically
# Access through the team's message history
async def run_with_memory():
    result = await team.run(task="Initial task")

    # Continue with context
    result = await team.run(task="Follow-up question")

    # Access message history
    for message in result.messages:
        print(f"{message.source}: {message.content}")

イベント駆動アーキテクチャ

カスタムイベントハンドリング

from autogen_core import Event

# Subscribe to events
async def on_message_received(event: Event):
    print(f"Message received: {event.data}")

# Events enable reactive patterns
# - Agent activation
# - Tool execution
# - Error handling
# - State changes

エラーハンドリング

堅牢なエージェント設計

from autogen_agentchat.agents import AssistantAgent
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def safe_run_team(team, task: str, max_retries: int = 3):
    """Run team with error handling and retries."""
    for attempt in range(max_retries):
        try:
            result = await team.run(task=task)
            return result
        except Exception as e:
            logger.error(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise
    return None

ベストプラクティス

エージェント設計

• エージェントに明確で焦点を絞った責任を与える
• わかりやすいシステムメッセージを使用する
• 適切なツール説明を実装する
• 適切な終了条件を設定する
• 複雑なワークフローにはハンドオフを使用する

パフォーマンス

• 並行操作に async パターンを使用する
• 繰り返されるクエリに対してキャッシングを実装する
• 合理的なタイムアウトを設定する
• トークン使用量を監視する
• 各エージェントに適切なモデルサイズを使用する

セキュリティ

• 信頼できないコードを直接実行しない
• コード実行に Docker を使用する
• ツール入力を検証する
• レート制限を実装する
• すべてのエージェントアクションをログに記録する

テスト

• 個別のエージェントをユニットテストする
• マルチエージェントワークフローを統合テストする
• 終了条件をテストする
• ツール実行を検証する
• 会話の品質を監視する

依存関係

• autogen-agentchat
• autogen-core
• autogen-ext
• openai (またはその他の LLM プロバイダー)
• python-dotenv
• docker (セキュアなコード実行用)

一般的なパターン

調査と執筆

# Pattern: Research -> Analyze -> Write -> Review
agents = [
    AssistantAgent(name="researcher", ...),
    AssistantAgent(name="analyst", ...),
    AssistantAgent(name="writer", ...),
    AssistantAgent(name="reviewer", ...)
]

コード生成

# Pattern: Plan -> Code -> Test -> Review
agents = [
    AssistantAgent(name="architect", ...),
    AssistantAgent(name="developer", code_executor=executor, ...),
    AssistantAgent(name="tester", ...),
    AssistantAgent(name="reviewer", ...)
]

データ分析

# Pattern: Extract -> Transform -> Analyze -> Report
agents = [
    AssistantAgent(name="data_engineer", ...),
    AssistantAgent(name="analyst", tools=[calc_tools], ...),
    AssistantAgent(name="reporter", ...)
]