複数エージェントを管理するMulti-Agent Orchestratorの使い方からクイックスタートまで

pip install multi-agent-orchestrator

# エージェントの設定
tech_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Tech Agent",
    streaming=True,
    description="Specializes in technology areas including software development, hardware, AI, \
    cybersecurity, blockchain, cloud computing, emerging tech innovations.",
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(tech_agent)

data_analysis_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Data Analyst",
    streaming=True,
    description="Expert in data analysis, statistics, and visualization. Can help with interpreting data, \
                creating visualizations, and providing analytical insights.",
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    callbacks=BedrockLLMAgentCallbacks(),
    temperature=0.3 
))

# オーケストレーターの設定
orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
    LOG_AGENT_CHAT=True,
    LOG_CLASSIFIER_CHAT=True,
    LOG_CLASSIFIER_RAW_OUTPUT=True,
    LOG_CLASSIFIER_OUTPUT=True,
    LOG_EXECUTION_TIMES=True,
    MAX_RETRIES=3,
    USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
    MAX_MESSAGE_PAIRS_PER_AGENT=10
))

import uuid
import asyncio
import os
from typing import Optional, List, Dict, Any
import json
import sys
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
from multi_agent_orchestrator.agents import (BedrockLLMAgent,
    BedrockLLMAgentOptions,
    AgentResponse,
    AgentCallbacks)
from multi_agent_orchestrator.types import ConversationMessage, ParticipantRole
import time

# AWS認証情報の直接設定（ここに自分の認証情報を入力）
os.environ['AWS_ACCESS_KEY_ID'] = ''  # ここにアクセスキーを入力
os.environ['AWS_SECRET_ACCESS_KEY'] = ''  # ここにシークレットキーを入力
os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'  # 使用するリージョン
# レート制限用の変数
RETRY_DELAYS = [1, 2, 4, 8]  # 指数バックオフの待機時間（秒）
last_request_time = 0
MIN_REQUEST_INTERVAL = 1.0  # 最小リクエスト間隔（秒）

class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        print(token, end='', flush=True)

# オーケストレーターの設定
orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
    LOG_AGENT_CHAT=True,
    LOG_CLASSIFIER_CHAT=True,
    LOG_CLASSIFIER_RAW_OUTPUT=True,
    LOG_CLASSIFIER_OUTPUT=True,
    LOG_EXECUTION_TIMES=True,
    MAX_RETRIES=3,
    USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
    MAX_MESSAGE_PAIRS_PER_AGENT=10
))

# エージェントの設定
tech_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Tech Agent",
    streaming=True,
    description="Specializes in technology areas including software development, hardware, AI, \
    cybersecurity, blockchain, cloud computing, emerging tech innovations.",
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(tech_agent)

async def handle_request(_orchestrator: MultiAgentOrchestrator, _user_input: str, _user_id: str, _session_id: str):
    global last_request_time
    
    # リクエスト間隔の制御
    current_time = time.time()
    time_since_last_request = current_time - last_request_time
    if time_since_last_request < MIN_REQUEST_INTERVAL:
        await asyncio.sleep(MIN_REQUEST_INTERVAL - time_since_last_request)
    
    for retry_count, delay in enumerate(RETRY_DELAYS):
        try:
            response: AgentResponse = await _orchestrator.route_request(_user_input, _user_id, _session_id)
            print("\n選択されたエージェント:", response.metadata.agent_name)
            
            if response.streaming:
                pass  # ストリーミング応答はコールバックで処理済み
            else:
                print("応答:", response.output)
            
            last_request_time = time.time()
            break
            
        except Exception as e:
            if "ThrottlingException" in str(e):
                if retry_count < len(RETRY_DELAYS) - 1:
                    print(f"\nレート制限に達しました。{delay}秒待機します... (リトライ {retry_count + 1}/{len(RETRY_DELAYS)})")
                    await asyncio.sleep(delay)
                    continue
                else:
                    print("\nリトライ回数の上限に達しました。しばらく時間をおいて再度お試しください。")
            else:
                print(f"\nエラーが発生しました: {str(e)}")
                print("エラーの詳細:", type(e).__name__)
            break

async def main():
    USER_ID = str(uuid.uuid4())  # ユニークなユーザーID
    SESSION_ID = str(uuid.uuid4())  # ユニークなセッションID
    
    print("\nMulti-Agentシステムへようこそ。'quit'と入力すると終了します。")
    print("エージェントの準備が完了しました。質問を入力してください。")
    
    while True:
        try:
            user_input = input("\nあなた: ").strip()
            if user_input.lower() == 'quit':
                print("プログラムを終了します。さようなら！")
                break
            
            await handle_request(orchestrator, user_input, USER_ID, SESSION_ID)
            
        except KeyboardInterrupt:
            print("\nプログラムを終了します。")
            break
        except Exception as e:
            print(f"エラーが発生しました: {str(e)}")

if __name__ == "__main__":
    asyncio.run(main())

git clone https://github.com/awslabs/multi-agent-orchestrator.git
cd multi-agent-orchestrator

pip install -r requirements.txt

Metadata:
Selected Agent: Weather Agent
東京の現在の天気は晴れで、気温は8.4°C、風速は5.0km/hです。晴れ渡った空が広がっています。 🌞