マルチエージェント並行処理のストリーミングを Microsoft Agent Framework で実装する

Microsoft Agent Framework でエージェントを並行で実行しつつその出力結果をフロントエンドに Streaming するサンプルです。

実行イメージ

_{これどんどん横に増やしていったら面白いかも…}

マルチエージェントパターンの利用シナリオ

結構利用シーンはあるのではないでしょうか。簡単なタスクであればマルチエージェントにする必要はないかと思いますが、それぞれ独自のモデルやナレッジ、ツールを持った専門家エージェントを用意するシーンなどで利用できると思います。

1. 意思決定支援システム - 新規事業評価、M&A判断、投資判断
2. 多角的レビューシステム - プロジェクト評価、製品企画レビュー
3. 学術研究支援 - 論文レビュー、文献調査、仮説検証
4. 市場調査・競合分析 - SWOT 分析、競合比較、市場トレンド分析
5. コード/設計レビュー - セキュリティ監査、レビュー、品質保証
6. カスタマーサポート - 問い合わせ対応、クレーム処理、FAQ 生成
7. コンテンツ制作支援 - 記事執筆、マーケティングコピー、プレゼン資料作成
8. ブレインストーミング - 新製品アイデア創出、戦略立案
9. 医療診断支援 - 症状分析、方針検討、リスク評価
10. 法務・コンプライアンス - 契約書レビュー、法的判断支援、規制対応
11. 学習支援システム - 課題フィードバック、学習計画立案、個別指導支援

最終出力だけでなく、各エージェントの中間出力も可視化したいシーンに最適です。

フロー

Microsoft Agent Framework 実装

エージェント定義

    # 批判的エージェント
    critical_agent = chat_client.create_agent(
        name="CriticalAnalyst",
        instructions="""
あなたは批判的思考の専門家です。
ユーザーの質問やアイデアに対して、潜在的な問題点、リスク、改善が必要な点を指摘します。
徹底的な批判を行い、具体的な懸念事項を簡潔に述べてください。
"""
    )
    
    # 肯定的エージェント
    positive_agent = chat_client.create_agent(
        name="PositiveAdvocate",
        instructions="""
あなたは肯定的思考の専門家です。
ユーザーの質問やアイデアに対して、利点、機会、成功の可能性を強調します。
前向きな視点から価値を見出し、具体的なメリットを簡潔に述べてください。
"""
    )
    
    # 統合エージェント
    synthesizer_agent = chat_client.create_agent(
        name="Synthesizer",
        instructions="""
あなたは統合の専門家です。
批判的な視点と肯定的な視点の両方を考慮し、バランスの取れた総合的な分析を提供します。
両方の視点を統合し、実用的な結論を導き出してください。
"""
    )

並行ワークフローの作成とイベントストリーミング

実装はかなりシンプルになりました。

# ConcurrentBuilderを使用して並行ワークフローを作成
agents = [critical_agent, positive_agent]
workflow = ConcurrentBuilder().participants(agents).build()

# エージェントの結果を格納
agent_results = {}

# ワークフローをストリーミング実行
async for event in workflow.run_stream(prompt):
    # AgentRunUpdateEvent: エージェントからのストリーミング更新
    if isinstance(event, AgentRunUpdateEvent):
        # executor_idからエージェント名を取得
        agent_name = event.executor_id
        
        # AgentRunResponseUpdateからテキストを取得
        if event.data and hasattr(event.data, 'text') and event.data.text:
            data = {
                "agent": agent_name,
                "content": event.data.text,
                "is_final": False
            }
            yield json.dumps(data, ensure_ascii=False) + "\n"

Microsoft Agent Framework では AgentRunUpdateEvent イベントの executor_id プロパティを参照することで、並行実行中のどのエージェントからの出力であるかを自動的に識別でき、統一されたイベントストリームによる一貫した処理が可能です。

並行処理は ConcurrentBuilder で実現できます。

注意

現時点では Azure OpenAI Service の非同期変更フィルターを使用して Streaming すると Microsoft Agent Framework 側で Stream がエラーになる事象が発生しています。（ソースの修正で対応）

GitHub

参考