Closed5ヶ月前にクローズ5

AWS製のマルチエージェントフレームワーク「Multi-Agent Orchestrator」を試す

AWS

LLM

Agent

kun432

AWSLabsなので、一応試験的なものなのかな？
https://github.com/awslabs/multi-agent-orchestrator

 Multi-Agent Orchestrator柔軟かつ強力なフレームワークで、複数のAIエージェントを管理し、複雑な会話を処理します。

 🔖 主な特徴🧠インテリジェントな意図分類

文脈と内容に基づいて、最適なエージェントに動的にクエリをルーティング。
🔤二言語対応

PythonとTypeScriptの両方で完全実装。
🌊柔軟なエージェントレスポンス

ストリーミングレスポンスと非ストリーミングレスポンスの両方をサポート。
📚コンテキスト管理

会話の文脈を複数のエージェント間で維持し、統一されたインタラクションを実現。
🔧拡張可能なアーキテクチャ

新しいエージェントを簡単に統合したり、既存のものをカスタマイズして特定のニーズに適合。
🌐ユニバーサルデプロイメント

AWS Lambdaからローカル環境、または任意のクラウドプラットフォームまで、どこでも動作可能。
📦事前構築済みエージェントと分類器

すぐに使えるエージェントと複数の分類器の実装が利用可能。

 Multi-Agent Orchestratorとは❓Multi-Agent Orchestratorは、複数のAIエージェントを管理し、複雑な会話を処理するための柔軟なフレームワークです。クエリを賢くルーティングし、インタラクション全体でコンテキストを維持します。
このシステムは、迅速なデプロイを可能にする事前構築済みコンポーネントを提供する一方で、カスタムエージェントや会話メッセージのストレージソリューションを簡単に統合できる柔軟性も備えています。
その適応性により、単純なチャットボットから高度なAIシステムに至るまで、さまざまな用途に適しており、多様な要件に対応しながら効率的にスケーリング可能です。

 🏗️ハイレベルなアーキテクチャフローダイアグラムユーザー入力を受け取り、Classifierがそれを分析します。
Classifierは、エージェントの特性と会話履歴の両方を活用して、最も適切なエージェントを選択します。
選択されたエージェントがユーザー入力を処理します。
オーケストレーターは会話を保存し、エージェントの会話履歴を更新し、最後にレスポンスをユーザーに返します。
とてもシンプルな設計に思える。

kun432

では実際に動かしてみる。自分はPythonで。

Python-3.12以上が要件のようなので、Colaboratoryは使えず。ローカルのMac上のDockerでJupyterLabを使うこととする。

作業ディレクトリ作成

mkdir multi-agent-orchestrator-test && cd multi-agent-orchestrator-test

JupyterLabのDockerコンテナを起動。上に書いた通り、今回はBedrockのAnthropicではなく、Anthropic本家を使うので、AWSのクレデンシャル等は不要なはずなのだけど、現状の作りだとBedrockを使わなくてもAWSのクレデンシャルが必要になるようなので、併せてマウントしている。

docker run --rm \
    -p 8888:8888 \
    -u root \
    -e GRANT_SUDO=yes \
    -v .:/home/jovyan/work \
    -v ~/.aws:/home/jovyan/.aws \
    quay.io/jupyter/minimal-notebook:latest

以降はJupyterLab上で。

パッケージインストール

!pip install multi-agent-orchestrator

!pip freeze | grep -i "multi"

出力

multi_agent_orchestrator==0.0.21

AnthropicのAPIキーをセット

import getpass
import os

os.environ["ANTHROPIC_API_KEY"] = getpass.getpass('ANTHROPIC_API_KEY')

ではQuickstart。まず最初に概念を整理。以下に記載されている。

リクエストの開始
- ユーザがオーケストレータにリクエストを送信。
分類
- 分類器（Classifier）がLLMを使用して、現在のユーザーIDとセッションIDにおける、ユーザのリクエスト・各エージェントの説明・全エージェントの会話履歴を分析し、最適なエージェントを選択。
エージェント選択
- 選択されたエージェントの名前で分類器が応答。
リクエストルーティング
- ユーザの入力を選択されたエージェントに渡す。
エージェント処理
- 選択されたエージェントがリクエストを処理。
- 現在のユーザIDとセッションIDに基づいて、選択されたエージェントとの会話履歴を自動的に取得
- 他のエージェントとの会話にはアクセスしない
会話ストレージ
- オーケストレータが、ユーザーの入力とエージェントの応答を、特定のユーザーIDとセッションIDニモも付けてストレージに保存
レスポンス配信
- オーケストレータが、エージェントのレスポンスをユーザに返す

ということでコンポーネントは以下となる

Orchestrator
Classifier
Agents
Conversation Storage

上記以外にもRetrieverもあるが、一旦置いておく。

まずはオーケストレータを定義する。

import os
import asyncio
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
from multi_agent_orchestrator.classifiers import AnthropicClassifier, AnthropicClassifierOptions
from multi_agent_orchestrator.storage import InMemoryChatStorage
import nest_asyncio

nest_asyncio.apply()

memory_storage = InMemoryChatStorage()

anthropic_classifier = AnthropicClassifier(
    AnthropicClassifierOptions(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        model_id="claude-3-haiku-20240307"
    )
)

orchestrator = MultiAgentOrchestrator(
    classifier=anthropic_classifier,
    storage=memory_storage,
    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             # エージェントごとにメッセージのペアを最大何件保持するか
    )
)

オーケストレータには、オーケストレータ自身の振る舞い、メモリ、分類器などをセットする。ビルトインの分類器は

Bedrock（デフォルト）
Anthropic
OpenAI（ただしTypeScriptのみ、PythonはPRがある模様）

が用意されており、また自分でカスタムな分類器を実装することもできる。

また、メモリは

インメモリ（デフォルト）
DynamoDB

がビルトインで用意されており、こちらも自分でカスタムなメモリを実装することができるみたい。

次に各エージェントを定義して、オーケストレータに追加する。

from multi_agent_orchestrator.agents import AnthropicAgent, AnthropicAgentOptions, AgentCallbacks

class LLMAgentCallbacks(AgentCallbacks):
    def __init__(self):
        super().__init__()
        self.first_token = True

    def on_llm_new_token(self, token: str) -> None:
        if self.first_token:
            print("Assistant: ", end='', flush=True)
            self.first_token = False
        print(token, end='', flush=True)
        
japanese_agent = AnthropicAgent(AnthropicAgentOptions(
  name="Japanese Agent",
  streaming=True,
  description="あなたの名前は太郎です。日本語に堪能で、日本の話題に詳しいです。",
  model_id="claude-3-haiku-20240307",
  api_key=os.environ["ANTHROPIC_API_KEY"],
  callbacks=LLMAgentCallbacks()
))
orchestrator.add_agent(japanese_agent)

english_agent = AnthropicAgent(AnthropicAgentOptions(
  name="English Agent",
  streaming=True,
  description="あなたの名前はジョンです。英語に堪能で、アメリカの話題に詳しいです。",
  model_id="claude-3-haiku-20240307",
  api_key=os.environ["ANTHROPIC_API_KEY"],
  callbacks=LLMAgentCallbacks()
))
orchestrator.add_agent(english_agent)

spanish_agent = AnthropicAgent(AnthropicAgentOptions(
  name="Spanish Agent",
  streaming=True,
  description="あなたの名前はラウルです。スペイン語に堪能で、スペインの話題に詳しいです。",
  model_id="claude-3-haiku-20240307",
  api_key=os.environ["ANTHROPIC_API_KEY"],
  callbacks=LLMAgentCallbacks()
))
orchestrator.add_agent(spanish_agent)

ここでは日本・アメリカ・スペインにそれぞれ精通したLLMエージェントを用意した。エージェントも複数の種類がある。

Bedrock（Converse APIを使用するものと恐らくbedrock-agent-runtimeを使用するものの２種と翻訳に特化したものがある）
OpenAI（これもTypeScriptのみ）
Anthropic

などのLLMを使用したエージェントもあれば、

Amazon Lex
AWS Lambda
Amazon Comprehend

など、AWSのサービスに紐づいたものもある。これらはいわゆる「ツール」的なイメージのように思えるが、ドキュメントを見る限りツールは別に定義できるところもあるようで、まだよくわからない。

上記以外にも複数エージェントをチェーン化するChain Agentや、自分でカスタムなエージェントを実装することもできる様子。

では実行。orchestrator.route_request()に、ユーザからの入力・ユーザID・セッションIDを渡すことでオーケストレータが処理を開始する。

async def main():
    while True:
        user_input = input("User: ")
        if user_input.lower() == "quit":
            print("チャットを終了します。さようなら。")
            break
        
        response = await orchestrator.route_request(
            user_input,
            "user_001",     # ユーザID
            "session_001",  # セッションID
        )

        if response.streaming:
            pass
        else:
            print('Assitant:', response.output.content[0]['text'])

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

こんな感じで出力される。

ピンクの箇所はオーケストレータのログ出力設定によるもの。初回のクエリに対しては、日本に詳しいエージェントが選択されていることがわかる。画像だとわからないが、レスポンスはストリーミング出力されている。

続けて会話してみる。

アメリカに関する話題になったら、別のエージェントが選択されていることがわかる。回答は日本語になっているけども。

さらに続けてみる。

スペインに関する話題には更に別のエージェントが選択され、解答もスペイン語になっている。

今回の例はエージェントの定義が曖昧すぎてあまり良い例ではないが、ただ分類器によりエージェントが選択されているということはわかる。

kun432

 まとめ現時点での所感
とりあえず触りだけしかやっていないけども、ドキュメントを見た限りでも、最近のエージェント系フレームワークとしては非常にシンプル。
学習コストは低いというメリットはある。
逆にいうと、商用で使うとするならば、機能が足りないかもしれない。AWS Labs、というのはあるにしても、LangGraphなどに比べるとかなりできることは限られている。

AWS謹製ならば当然AWSサービスとの連携はしやすいはず。
Lambda/Lexなどのエージェントがあったり
retrieverはBedrock Knowledge Baseと連携できたり
メモリはDynamoDBに保存できたり

TypeScriptのほうがややサポート強めな印象を持った
OpenAI ClassifierやOpenAI AgentなどはPythonにはまだ含まれていない
用意されているデモもTypeScriptのものが多い

個人的にはとりあえず今後の展開次第かなーという感じ。

kun432

https://weel.co.jp/media/tech/multi-agent-orchestrator/
私が実装した時には、「An error occurred (ThrottlingException) when calling the Converse operation (reached max retries: 4): Too many requests, please wait before trying again.」というエラーが出てしまい、レート・リクエスト間隔の調整を行わないと実行不可でした。
これが出たのでBedrock使わずにAnthropicにしたんだよなー。Bedrockのクォータ周りは以前よりも面倒になっているのがなぁ・・・

kun432

2024/11/27補足
現状の作りだとBedrockを使わなくてもAWSのクレデンシャルが必要になるようなので、併せてマウントしている。
Bedrockを使わずにAnthropicやOpenAIを使用する場合のインストールオプションが追加された模様。
multi-agent-orchestratorにはdefault agentという概念がある。
classifierが分類できない場合に処理をするエージェント
orchestratorでデフォルトのエージェントを指定することができる。
というもので、指定がない場合のデフォルトはBedrock LLM Agentになっている。で素直にBedrockをLLMで使う場合には困らないのだが、他のLLM、AnthropicやOpenAIを使う場合にややこしいことになる。
Bedrockではないclassifierを使ったorchestratorを使う場合は、classifierをまず定義する必要がある
この時点でBedrock LLM Agentがデフォルトエージェントとして定義される。
Bedrock LLM Agentの定義にAWSのクレデンシャルは必須

作成したclassifierを使って、orchestratorを定義する
デフォルトのエージェントをagentとして定義
作成したagentをorchestratorに追加、およびデフォルトのエージェントに設定
デフォルトのBedrock LLM Agentを上書きする

つまり、コンポーネント構成上、Bedrockを使わなないとしても、classifier設定時点でBedrock Agentが初期化され、その際に必ずAWSのクレデンシャルが必要になる。例えばこう。
!pip install multi-agent-orchestrator
from multi_agent_orchestrator.classifiers import AnthropicClassifier, AnthropicClassifierOptions
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator

anthropic_classifier = AnthropicClassifier(AnthropicClassifierOptions(
    api_key=os.environ["ANTHROPIC_API_KEY"]
))
とするだけで、以下となる
(snip)
File /opt/conda/lib/python3.12/site-packages/multi_agent_orchestrator/classifiers/classifier.py:16, in Classifier.__init__(self)
     15 def __init__(self):
---> 16     self.default_agent = BedrockLLMAgent(
     17         options=BedrockLLMAgentOptions(
     18         name=AgentTypes.DEFAULT.value,
     19         streaming=True,
     20         description="A knowledgeable generalist capable of addressing a wide range of topics.\
     21         This agent should be selected if no other specialized agent is a better fit."
     22     ))
     24     self.agent_descriptions = ""
     25     self.history = ""
(snip)
NoRegionError: You must specify a region.
コードを読んだところ、classifierの定義ではデフォルトエージェントを外から定義することができなかったので、AWSの認証情報がなければ詰む、という感じだったので、Issueを上げておいたところ、
https://github.com/awslabs/multi-agent-orchestrator/issues/106
以下のPRがすでにマージされていて（ただし2024/11/27時点ではまだパッケージは更新されていない）、ざっと見た感じ、classifierからBedrock LLM Agentへの依存が取り除かれて、orchestratorでデフォルトエージェントを設定できるようになっているみたい。あと、anthropic・openaiのパッケージもextrasで分かれるようになった。
https://github.com/awslabs/multi-agent-orchestrator/pull/108

このスクラップは5ヶ月前にクローズされました

ログインするとコメントできます