このガイドと次のセクションでは、AutoGen の基本概念: エージェント、エージェント・ランタイム、メッセージと通信 – マルチエージェント・アプリケーション用の基礎的なビルディングブロック – にフォーカスします。

AutoGen Core : フレームワークガイド : エージェントとエージェント・ランタイム

作成 : クラスキャット・セールスインフォメーション
作成日時 : 04/12/2025

* 本記事は github microsoft/autogen の以下のページを独自に翻訳した上でまとめ直し、補足説明を加えています :

• AutoGen Core : Framework Guide : Agent and Agent Runtime

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

 

 

 

AutoGen Core : フレームワークガイド : エージェントとエージェント・ランタイム

このガイドと次のセクションでは、AutoGen の基本概念: エージェント、エージェント・ランタイム、メッセージと通信 – マルチエージェント・アプリケーション用の基礎的なビルディングブロック – にフォーカスします。

Note : Core API は非主張的で (unopinionated) 柔軟に設計されています。そのため挑戦的と感じる場合が時々あるかもしれません。インタラクティブでスケーラブルな分散型マルチエージェント・システムを構築していてすべてのワークフローの完全な制御を望む場合には、このまま続けてください。何かを迅速に動かしたいだけならば、AgentChat API を見てみると良いかもしれません。

 
AutoGen のエージェントは基底インターフェイス Agent により定義されるエンティティです。
それは AgentId 型の一意な識別子、AgentMetadata 型のメタデータ辞書を持ちます。

殆どの場合、上位クラス RoutedAgent からエージェントをサブクラス化できます、これはメッセージを message_handler() デコレータと message 変数用の適切な型ヒントを使用して指定された、対応するメッセージハンドラにルーティングすることを可能にします。エージェント・ランタイムは AutoGen のエージェントの実行環境です。

プログラミング言語のランタイム環境と同様に、エージェント・ランタイムは必要なインフラを提供してエージェント間の通信を容易にし、エージェントのライフサイクルを管理し、セキュリティ境界を強化し、そして監視とデバッグをサポートします。

ローカル開発のために、開発者は SingleThreadedAgentRuntime を使用できます、これは Python アプリケーションに埋め込むことができます。

Note : エージェントはアプリケーションコードにより直接にはインスタンス化されたり管理されません。代わりに、必要に応じてランタイムにより作成され、ランタイムにより管理されます。

既に AgentChat に馴染みがあるならば、AssistantAgent のような AgentChat のエージェントはアプリケーションにより作成され、従ってランタイムにより直接管理されないことに注意することは重要です。Core で AgentChat を使用するには、メッセージを AgentChat エージェントに委任するラッパー Core エージェントを作成して、ランタイムにラッパー・エージェントを管理させる必要があります。

 

エージェントの実装

エージェントを実装するには、開発者は RoutedAgent クラスをサブクラス化して、エージェントが処理すると想定する各メッセージタイプ用に message_handler() デコレータを使用してメッセージハンドラを実装する必要があります。例えば、次のエージェントは単純なメッセージタイプ MyMessageType を処理して、受信したメッセージを出力します :

from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler


@dataclass
class MyMessageType:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")

このエージェントは MyMessageType を処理するだけで、メッセージは handle_my_message_type メソッドに配信されます。開発者は message_handler() デコレータを使用して、ハンドラ関数内の message 変数の型ヒントを設定することで、異なるメッセージタイプ用に複数のメッセージハンドラを持つことができます。エージェントのロジックにより良く適す場合には、1 つのメッセージハンドラ関数で message 変数用に python typing union を使用することもできます。See the next section on message and communication.

 

AgentChat エージェントの使用

AgentChat エージェントを持ち、それを Core API で使用したい場合、メッセージを AgentChat エージェントに委任するラッパー RoutedAgent を作成することができます。次の例は、AgentChat で AssistantAgent のラッパーエージェントを作成する方法を示します。

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


class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self._delegate = AssistantAgent(name, model_client=model_client)

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} responded: {response.chat_message}")

モデルクライアントの使用方法については、Model Client セクションをご覧ください。

Core API は unopinionated ですので、Core API を使用するために AgentChat API を使用することを要求されません。独自のエージェントを実装したり別のエージェント・フレームワークを使用できます。

 

エージェントタイプの登録

エージェントをランタイムで利用可能にするため、開発者は BaseAgent クラスの register() クラスメソッドを使用できます。登録プロセスは、文字列により一意に識別されるエージェントタイプと、指定のクラスのエージェントタイプのインスタンスを作成するファクトリ関数を関連付けします。ファクトリ関数は必要に応じてエージェント・インスタンスの自動作成を可能にするために使用されます。

エージェントタイプ (AgentType) はエージェントクラスと同じではありません。この例では、エージェントタイプは AgentType(“my_agent”) か AgentType(“my_assistant”) で、エージェントクラスは Python クラス MyAgent か MyAssistantAgent です。ファクトリ関数は、register() クラスメソッドが呼び出された、エージェントクラスのインスタンスを返すことが想定されます。エージェントタイプと ID について更に学習するには Agent Identity and Lifecycles を読んでください。

Note : 同じエージェントクラスを返すファクトリ関数で、異なるエージェントタイプを登録できます。例えば、ファクトリ関数では、コンストラクタ・パラメータのバリエーションを使用して同じエージェントクラスの異なるインスタンスを作成できます。

SingleThreadedAgentRuntime でエージェントタイプを登録するために、以下のコードが使用できます :

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())
await MyAssistant.register(runtime, "my_assistant", lambda: MyAssistant("my_assistant"))

AgentType(type='my_assistant')

エージェントタイプが登録されれば、AgentId を使用してエージェントインスタンスに直接メッセージを送信できます。ランタイムは、このインスタンスに初めてメッセージを送信する際にインスタンスを作成します。

runtime.start()  # Start processing messages in the background.
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_agent", "default"))
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_assistant", "default"))
await runtime.stop()  # Stop processing messages in the background.

my_agent received message: Hello, World!
my_assistant received message: Hello, World!
my_assistant responded: Hello! How can I assist you today?

Note : ランタイムはエージェントのライフサイクルを管理するので、AgentId はエージェントと通信したりメタデータ (e.g., 説明) を取得するためだけに使用されます。

 

シングルスレッドのエージェント・ランタイムの実行

上記のコードスニペットは、start() を使用してバックグラウンド・タスクの処理を開始し、メッセージを受信者のメッセージハンドラに配信します。これはローカルの組み込み (embedded) ランタイム SingleThreadedAgentRuntime の機能です。

バックグラウンド・タスクを直ちに停止するには、stop() メソッドを使用します :

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop()  # This will return immediately but will not cancel
# any in-progress message handling.

start() を再度呼び出すことで、バックグラウンド・タスクを再開できます。

エージェントを評価するためにベンチマークを実行するようなバッチ・シナリオについては、未処理のメッセージがなくメッセージを処理しているエージェントがない時 – バッチは完了したと考えられます -、バックグラウンド・タスクが自動的に停止するまで待機したいかもしれません。stop_when_idle() メソッドを使用してこれを実現できます :

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop_when_idle()  # This will block until the runtime is idle.

ランタイムをクローズしてリソースを解放するには、close() メソッドを使用します :

await runtime.close()

Other runtime implementations will have their own ways of running the runtime.

以上