Microsoft Agent Framework - coreモジュールとかの実装を見る

概要

Microsoft Agent Framework (以下MAF) の Github に公開されている python/packages/core/agentframework の中身を見て、どんな機能があるか確認します。

agent_framework

_agents.py

• AgentProtocol: 型チェックのためのプロトコル
• BaseAgent: エージェントの具体的な関数のベース
• ChatAgent: ChatClientを使ったエージェント

MAFを使ったチャットボットやワークフローを構成するエージェントはここを確認して作成する

_clients.py

• ChatClientProtocol: 型チェックのためのプロトコル
• BaseChatClient: クライアントに必要な具体的な関数のベース

クライアントはメッセージを送ってそのレスポンスを受ける、ツールを処理、設定を辞書化、クライアントをエージェント化する関数などを持つ

_logging.py

• loggingモジュールを使って、MAFを利用する上で決まった書式のロギング機能とロガー取得方法を定義している

_mcp.py

MCPクライアントのフレームワーク

• MCPTool: ベースクラス
• MCPStdioTool: 標準入出力でMCPサーバーをやり取りする
• MCPStreamableHTTPTool: HTTP（SSE/ストリーミング含む）経由でMCPサーバと接続する
• MCPWebsocketTool: WebSocket経由でMCPサーバに接続する

が定義されている。サーバー側の提供はないため、自作のMCPサーバーを作りたい場合はFastMCP等を使って自作する必要がある。

_memory.py

• AIモデルに渡すコンテキストを管理するためのクラスContext、ContextProvider、AggregateContextProviderを実装
• エージェントのメモリ機能、文脈情報の提供の基盤モジュールとなっている

_middleware.py

正直まだ実際の使い道にあまりピンと来ていないので個別に詳しくやりたい

• AgentMiddleware, FunctionMiddleware, ChatMiddleware
  • それぞれ「エージェント呼び出し」「関数実行」「チャットクライアント実行」に割り込むためのミドルウェアの抽象クラスが定義されている。processメソッドを実装することで、好きな処理（ロギング、バリデーション、リトライ等）を挿入できる
• AgentRunContext, FunctionInvocationContext, ChatContext
  • 各実行時の状態・情報（メッセージ列、引数、結果、終了フラグ、メタデータ等）を保持・共有するためのコンテキストオブジェクトが定義されている
• AgentMiddlewarePipeline, FunctionMiddlewarePipeline, ChatMiddlewarePipeline
  • 複数のミドルウェアをチェーン形式で順番に実行し、必要なら途中で終了 (terminate)、結果を上書き (result) できる仕組みを持つ
• 自作エージェントやクライアントへのミドルウェア拡張のためのクラスデコレータ等

_pydantic.py

• Pydantic の BaseSettingsを拡張してAFBaseSettingsクラスを作成し、環境変数や.envファイルから設定を読み込む方法を用意
• AnyUrlを使ってHTTPsUrlエイリアス、HTTPSで最大長2083のみを許可したAnnotated型を定義

_serialization.py

• SerializationProtocol: インスタンスがto_dict, from_dictを持つかどうか（シリアライズ可能か）を判定するためのプロトコル
• SerializationMixinサブクラス: 継承させることによってシリアライズ可能なクラスを簡単に作れるようにしている。(to_jsonとfrom_jsonも合わせて実装されている)

SerializationMixinによってMAF内で扱う各インスタンスはto_dict, from_dictを使ってその詳細を取得・保存・復元できるようになっている。実際各モジュールを確認するとほとんどのクラスはSerializationMixinを継承して作成されている。

_telemetry.py

• prepend_agent_framework_to_user_agent(headers)関数: HTTPリクエスト送信時のヘッダーにMAFのフレームワーク名とそのバージョンを"User-Agent": "agent-framework-python/{version}"という形で設定するだけの実装。

他の実装で使われているところを確認する限り、おそらくOpenAIやAzureOpenAIの呼び出しクライアントのヘッダーに追加することで、公式サポートされているテレメトリーで確認した時にこのフレームワークを使って呼び出したことがわかるようにすることが目的

_threads.py

• ChatMessageStoreProtocol, ChatMessageStore: チャットメッセージの保存機構の抽象化・実装
• ChatMessageStore: ChatMessageのリストを持つ
• AgentThread: エージェントの会話スレッド自体の状態・履歴管理
  • service_thread_id, message_store（上記のChatMessageStoreProtocol）, context_provider を持つ

_tools.py

• エージェントが使うツールの管理をするための色んなものが定義されているが2000行以上あるモジュールのため詳しくは個別に解説したい
• エージェントが利用するツール（Web検索、ファイル検索、コードインタプリター、MCP、カスタム関数 など）をクラスベースで抽象化し一元管理
• Python関数をLLMから呼び出し可能な関数にするデコレータの定義
• Approval（承認）が必要な関数や一度しか実行できない関数など、制御の定義

_types.py

• AIエージェント・チャット・ツール呼び出し・構造化応答などのデータ型・定数・ヘルパー機能をまとめており、MAFの中核となっている
• こちらも3500行程度あるので個別に機能を使いながら細かいところを確認する必要がある
• 全てのデータ型が_serialization.pyで出てきたSerializationMixinを継承しているため、シリアライズ・デシリアライズに対応している
• 主要な型
• 基本メッセージ・応答: ChatMessage, ChatResponse, AgentRunResponse, ChatResponseUpdate, AgentRunResponseUpdate
• コンテンツ型: TextContent, TextReasoningContent, DataContent, UriContent, FunctionCallContent, FunctionResultContent, ErrorContent, UsageContent, HostedFileContent, HostedVectorStoreContent, FunctionApprovalRequestContent, FunctionApprovalResponseContent
  • エージェントの各処理結果コンテンツの種類
• アノテーション型: BaseAnnotation, CitationAnnotation, TextSpanRegion
  • テキストの一部分に引用元をを付与したりできる
• 補助的な型やUtil関数
  • Role, FinishReason, ToolModeなど一般的なLLMクライアントで使われるものの型定義
  • 辞書やテキストからChatMessageなどの主要な型への変換を行うヘルパー関数の定義

exceptions.py

• 各クラスで発生したエラーの型定義

observability.py

• OpenTelemetryベースの監視・可観測機能を提供するライブラリ
• クラウドならAzure Monitorなど、ローカルならVSCodeの拡張機能(AI Toolkit もしくは Azure AI Foundry) や OpenTelemetryの可視化ライブラリであるJaeger等を使って使用状況を確認できる
• PoC・開発・運用等でどの方法で可視化するのが楽・便利かなどをまた個別に検証したい

_workflows

• エージェントワークフロー構築のためのモジュール群
• 役割: 複数のエージェントや実行単位(Executor)を有向グラフとして接続し、複数の処理を段階的に同期させながら協調動作させる仕組みを構築している
  • 複雑なマルチエージェントワークフローの定義と実行
  • グラフベースの実行エンジン(Executorをノード、メッセージパッシングをエッジとする)
  • チェックポイント機能による状態の永続化と再開
  • Microsoftが提言している標準的なマルチエージェントワークフロービルダーの実装(Sequential, Concurrent, GroupChat, Handoff, Magentic)
• 単一エージェント処理(coreモジュール群)とは異なり、複雑なオーケストレーションを実現するためのコンポーネントとするため、独立した実装になっている
• 主な実装
• コア実行エンジン (_workflow.py, _executor.py, _runner.py)
  • エージェント統合 (_agent.py, _agent_executor.py)
  • グラフ構造 (_edge.py, _edge_runner.py, _workflow_builder.py)
  • マルチエージェントワークフロービルダー
    • WorkflowBuilder (_workflow_builder.py): 基本のワークフロー。一番カスタマイズ性が高く、下記の個別ワークフローは全てこのビルダーを使っている(継承はしていない点に注意)
    • SequentialBuilder (_sequential.py): 順次実行パイプライン
    • ConcurrentBuilder (_concurrent.py): 並列実行(fan-out/fan-in)
    • GroupChatBuilder (_group_chat.py): エージェント同士のマルチターン会話フロー
    • HandoffBuilder (_handoff.py): 各エージェントが次に実行されるべきエージェントを指名していくフロー
    • MagenticBuilder (_magentic.py): タスクの作成と進捗管理をするマネージャーと専門家エージェント間の協業フロー
  • 状態管理とチェックポイント (_checkpoint.py, _shared_state.py, _orchestration_state.py)
  • イベントシステム (_events.py)
  • ユーティリティ
    • 会話管理 (_conversation_history.py, _conversation_state.py, _message_utils.py)
    • 型システム (_typing_utils.py, _model_utils.py)
    • バリデーション (_validation.py)
    • 可視化 (_viz.py)
• その他特筆すべき点
  • 簡単な設定でHuman-in-the-Loopに対応できる
  • エージェントと同じくOpenTelemetry統合の実装がある

外部機能統合

外部機能統合により、各LLMプロバイダの認証を使ってクライアントを使えるようにしたり、可視化・履歴の管理等の拡張機能を利用できるようにしたり等を可能にしている。
詳しくは気力があれば個別に解説する。

• a2a
• ag_ui
• anthropic
• azure
  • Azureの各サービスとの接続・連携のための機能。
  • 記事執筆時点ではAzureOpenAI・Azure AI(Microsoft Foundryエージェントなど)・Azure Function / Durable Function・Azure AI Search関連の実装
• chatkit
• declarative
• devui
  • 個人的に開発で便利だと思っている拡張機能
  • 簡単にエージェント・ワークフローの可視化やテストが出来る
• lab
• mem0
  • ChatGPTのメモリ機能みたいなものを実装したければ楽そう
  • と思ったが、Mem0のAPIをMem0社経由の有料APIで使うか、OSSのセルフホスティングが必要なので普通に運用・保守の学習コストが高そう
  • Mem0にはLLMによる自動要約機能やセマンティックな検索などが組み込まれており、LLMの記憶管理としてはかなり完成されていそう
• microsoft
  • Microsoftの各サービスとの接続のための機能
  • 記事執筆時点ではCopilot Studio・Purview関連の実装がされていた
• ollama
• openai
  • OpenAIのAPI利用、モニタリング等の機能統合。シンプル
• redis
  • Redisでコンテキストと会話履歴の管理を実現する機能
  • チャット単位で履歴を管理する用途

おわり

外部機能統合が歯抜けになっていることからわかるように途中で気力が尽きました。
自分の理解のために書いたのですが、読み返してみて他の人が読んでも何のことやらよくわからない記事になっている自覚があります。次回以降は、実際に色んな機能を実装・テストして利用方法をより具体的にイメージできるような内容を書きたいと思います。