autogen-development

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

Python を使用したマルチエージェント AI システムを構築するためのフレームワークである Microsoft AutoGen のエキスパートとして、エージェントのオーケストレーション、ツール統合、スケーラブルな 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

ベストプラクティス

エージェント設計

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

パフォーマンス

• 並行処理には非同期パターンを使用します
• 繰り返し行われるクエリにはキャッシュを実装します
• 妥当なタイムアウトを設定します
• トークンの使用量を監視します
• 各エージェントに適切なモデルサイズを使用します

セキュリティ

• 信頼できないコードを直接実行しないでください
• コードの実行には 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", ...)
]