Microsoft Agent Framework 完全ガイド - これから始める人のために

Last updated at 2025-10-25Posted at 2025-10-25

はじめに
Microsoft Agent Frameworkとは
なぜMicrosoft Agent Frameworkなのか
環境構築とセットアップ
基本的な使い方
ツールの使い方
ワークフローとオーケストレーション
開発アプローチ：コードファースト vs 宣言的ワークフロー
DevUI：開発・テスト用のWebインターフェース
実践的なパターンとベストプラクティス
既存フレームワークからの移行
よくある質問とトラブルシューティング

はじめに

AIエージェント開発の世界へようこそ！この記事では、Microsoftが提供するMicrosoft Agent Frameworkについて、ゼロから丁寧に解説していきます。

この記事で学べること

Microsoft Agent Frameworkの基本概念と設計思想
環境構築から最初のエージェント作成まで
ツールとワークフローの実装方法
本番運用を見据えたベストプラクティス
Semantic KernelやAutoGenからの移行方法

Microsoft Agent Frameworkとは

概要

Microsoft Agent Frameworkは、Semantic KernelとAutoGenという2つの先行フレームワークの強みを統合した、エンタープライズグレードのAIエージェント開発フレームワークです。2025/10/1に発表されたこのフレームワークは、Microsoftのエンタープライズ向けAI戦略の中核を担います。

誕生の背景

Microsoftは、既存のAIエージェントフレームワークにおいて、開発者が直面している根本的な課題を認識していました：

「プロトタイプから本番環境への道が多くの障害で満ちている」

「イノベーションの追求と本番運用の安定性の間で、開発者が選択を迫られる状況」

この課題を解決するために、Microsoft Agent Frameworkは研究から本番環境への橋渡しを実現することを目的として設計されました。

対応言語

Python: 3.10以降
C#/.NET: .NET 8.0以降

両言語で完全に一貫したAPIを提供しており、同じ概念と設計パターンを共有します。

なぜMicrosoft Agent Frameworkなのか

Microsoft Agent Frameworkの設計は、公式に発表された4つの基本柱に基づいています。この4つの柱が、他のフレームワークとの差別化ポイントになっています。

1. オープン標準と相互運用性

ベンダーロックインを回避し、エコシステム全体との統合を実現します。

MCP (Model Context Protocol)

Anthropicが提唱する標準プロトコルで、外部ツールやデータサーバーの動的な発見と呼び出しが可能です。

from agent_framework.mcp import MCPStdioTool

# ファイルシステムMCPサーバーと接続
filesystem_tool = MCPStdioTool(
    command="uvx",
    args=["mcp-server-filesystem", "--allowed-directory", "/data"]
)

agent = ChatAgent(
    chat_client=client,
    tools=[filesystem_tool]
)

利点:

コミュニティが開発した多様なMCPサーバーをそのまま利用可能
ファイルシステム、データベース、外部APIなど、幅広い統合に対応

Agent-to-Agent (A2A) プロトコル

異なるフレームワーク間でのエージェント協働を実現します。

利点:

LangChainなど他のフレームワークのエージェントとも通信可能
既存のエージェントエコシステムを活用できる

OpenAPI-first 設計

REST APIを直接ツール化できます。

利点:

既存のAPIドキュメントをそのまま活用
カスタムツール実装が不要

クラウド非依存

Azure OpenAI、OpenAI、その他任意のLLM SDKで稼働可能です。

利点:

特定のクラウドに縛られない
複数のLLMプロバイダーを使い分け可能

2. 研究から本番環境へのパイプライン

Microsoft Researchの最先端オーケストレーションパターンを、本番運用可能な形で実装しています。

先進的な多エージェント編成パターン

研究レベルの高度なパターンが、耐久性、ガバナンス、パフォーマンスを損なわずに実世界のシステムで利用可能です。

豊富なサンプルパターン

並列処理（ファンアウト・ファンイン）
チェックポイントによる耐久性
Human-in-the-loop（人間による承認フロー）
長期実行タスクの中断と再開

プロトタイプから本番へのスムーズな移行

同じコードベースで、プロトタイプ検証から本番スケーリングまで実現できます。

具体例:

# プロトタイプ段階
workflow = Sequential()
    .add_executor(agent1)
    .add_executor(agent2)
    .build()

# 本番環境用に拡張（チェックポイント追加）
workflow = Sequential()
    .add_executor(agent1)
    .add_executor(agent2)
    .enable_checkpointing(  # 本番運用機能を追加
        strategy=FileCheckpointStrategy(directory="./checkpoints")
    )
    .build()

3. コミュニティ主導の拡張性

100%オープンソースで、モジュール設計により柔軟なカスタマイズが可能です。

完全オープンソース

GitHubで開発され、コミュニティからの貢献を歓迎しています。

リポジトリ: https://github.com/microsoft/agent-framework
ライセンス: MIT

エンタープライズコネクタ

すぐに使える統合が提供されています：

Azure AI Foundry
Microsoft Graph（Office 365データへのアクセス）
SharePoint
その他Microsoft 365サービス

プラガブルメモリモジュール

多様なメモリバックエンドに対応：

Redis（高速キャッシュ）
Pinecone（ベクトルデータベース）
mem0（長期記憶管理）

宣言的エージェント定義

YAML/JSON形式でエージェントを定義でき、コード不要で設定を管理できます。

# agent-config.yaml
name: CustomerSupportAgent
instructions: |
  あなたは親切なカスタマーサポートエージェントです。
  顧客の質問に丁寧に答えてください。
tools:
  - search_knowledge_base
  - create_ticket
model:
  provider: azure_openai
  deployment: gpt-4

4. 本番運用対応（Production-Ready）

エンタープライズ要件を最初から満たす設計です。

観測可能性（Observability）

OpenTelemetryによる自動計測で、エージェントアクション、ツール呼び出し、オーケストレーションステップのすべてを可視化できます。

from agent_framework.observability import configure_observability

# 設定するだけで自動計測開始
configure_observability(
    service_name="my-agent-app",
    tracing_enabled=True,
    metrics_enabled=True
)

# 以降のすべてのエージェント実行が自動的にトレースされる
agent = ChatAgent(...)
response = await agent.run("タスクを実行")  # 自動的に計測

収集されるメトリクス:

トークン使用量（入力/出力/合計）
レイテンシー（エージェント/チャット/関数）
エラー率
ツール呼び出し頻度

セキュリティ

企業レベルのセキュリティを実現：

Azure AI Content Safety統合: 有害なコンテンツのフィルタリング
Azure Entra ID認証: 企業の認証基盤と統合
秘密情報の安全な管理: Azure Key Vault統合

耐久性（Durability）

チェックポイントと再開機能により、長時間実行タスクを中断から確実に復旧できます。

# 5時間かかるワークフローでも安心
workflow = Sequential()
    .add_executor(step1)  # 1時間
    .add_executor(step2)  # 2時間
    .add_executor(step3)  # 2時間
    .enable_checkpointing()
    .build()

# 途中で失敗しても、失敗した場所から再開可能

Human-in-the-loop

人間の承認が必要なツール実行を設定でき、重要な操作の制御を維持できます。

from agent_framework import FunctionMiddleware

class ApprovalMiddleware(FunctionMiddleware):
    async def on_invoke(self, context, next):
        if context.function.name == "delete_database":
            approval = input("データベース削除を承認しますか? (y/n): ")
            if approval != "y":
                raise ValueError("承認されませんでした")
        return await next(context)

環境構築とセットアップ

実際に手を動かして、Microsoft Agent Frameworkを使い始めましょう。

前提条件

Python環境の場合

Python: 3.10以降
パッケージマネージャー: pip または uv（推奨）
Azure CLI: Azure環境を使う場合

.NET環境の場合

.NET SDK: 8.0以降
NuGet: パッケージマネージャー

Azure環境（推奨）

Azure OpenAI リソース または Azure AI プロジェクト
適切なロールベースアクセス権限
- Cognitive Services OpenAI User
- または Cognitive Services OpenAI Contributor

インストール手順

Python

# 全機能をインストール（初心者向け）
pip install agent-framework --pre

# または、uvを使用（推奨）
uv pip install agent-framework --pre

# 特定の統合のみをインストール（本番環境向け）
pip install agent-framework-core
pip install agent-framework-azure-ai

.NET

# NuGetパッケージをインストール
dotnet add package Microsoft.Agents.AI --prerelease

# Azure OpenAI統合を使う場合
dotnet add package Microsoft.Agents.AI.OpenAI --prerelease

# ワークフローを使う場合
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

認証設定

Azure CLIによる認証（推奨）

# Azure CLIにログイン
az login

# サブスクリプションを確認
az account show

# 必要に応じてサブスクリプションを切り替え
az account set --subscription "Your Subscription Name"

APIキーによる認証

環境変数で設定：

# Linux/Mac
export AZURE_OPENAI_API_KEY="your-api-key"
export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4"

# Windows (PowerShell)
$env:AZURE_OPENAI_API_KEY="your-api-key"
$env:AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
$env:AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4"

または、.envファイルで管理：

# .env
AZURE_OPENAI_API_KEY=your-api-key
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4

# Pythonで読み込み
from dotenv import load_dotenv
load_dotenv()

動作確認

最小限のコードで動作を確認しましょう。

Python

import asyncio
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

async def main():
    # エージェント作成
    agent = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    ).create_agent(
        name="TestAgent",
        instructions="あなたは親切なアシスタントです。"
    )

    # 実行
    response = await agent.run("こんにちは！")
    print(response)

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

C#/.NET

using Azure.Identity;
using OpenAI;

var agent = new OpenAIClient(
    new BearerTokenPolicy(
        new AzureCliCredential(),
        "https://ai.azure.com/.default"
    ),
    new OpenAIClientOptions() {
        Endpoint = new Uri("https://your-resource.openai.azure.com/openai/v1")
    }
)
.GetOpenAIResponseClient("gpt-4")
.CreateAIAgent(
    name: "TestAgent",
    instructions: "あなたは親切なアシスタントです。"
);

var response = await agent.RunAsync("こんにちは！");
Console.WriteLine(response);

基本的な使い方

ここからは、Microsoft Agent Frameworkの基本的な使い方を、実践的なサンプルとともに学んでいきます。

エージェントの基本構造

Microsoft Agent Frameworkのエージェントは、以下の要素で構成されます：

Chat Client: LLMとの通信を担当
Agent: エージェントの振る舞いを定義
Instructions: エージェントの役割と指示
Tools: エージェントが使用できるツール（オプション）

シンプルなチャットエージェント

import asyncio
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

async def main():
    # クライアント作成
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # エージェント作成
    agent = client.create_agent(
        name="AssistantAgent",
        instructions="""
        あなたは親切で知識豊富なアシスタントです。
        ユーザーの質問に対して、正確で分かりやすい回答を提供してください。
        不確実な情報については、推測せずにその旨を伝えてください。
        """
    )

    # 会話
    response = await agent.run("Pythonの基本的な使い方を教えてください")
    print(response)

asyncio.run(main())

ストリーミング応答

長い応答をリアルタイムで受け取ることができます。

async def streaming_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="StreamingAgent",
        instructions="あなたは詳しい説明をするアシスタントです。"
    )

    print("エージェントの応答:")
    async for update in agent.run_streaming("機械学習について詳しく説明してください"):
        # 更新があるたびに表示
        print(update.content, end="", flush=True)
    print()  # 改行

asyncio.run(streaming_example())

会話履歴の管理

エージェントは会話の文脈を保持できます。

from agent_framework import AgentThread

async def conversation_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="ConversationAgent",
        instructions="あなたは会話の文脈を理解するアシスタントです。"
    )

    # スレッド（会話セッション）を作成
    thread = AgentThread()

    # 最初のメッセージ
    response1 = await agent.run(
        "私の名前は太郎です",
        thread=thread
    )
    print(f"応答1: {response1}")

    # 文脈を保持した2番目のメッセージ
    response2 = await agent.run(
        "私の名前は何でしたか？",
        thread=thread
    )
    print(f"応答2: {response2}")  # "太郎"と答えるはず

asyncio.run(conversation_example())

マルチモーダル対応（画像処理）

画像を含むメッセージを処理できます。

from agent_framework import ChatMessage, TextContent, DataContent
from pathlib import Path
import base64

async def image_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="VisionAgent",
        instructions="あなたは画像を分析するアシスタントです。"
    )

    # 画像を読み込み
    image_path = Path("sample.jpg")
    image_data = base64.b64encode(image_path.read_bytes()).decode()

    # マルチモーダルメッセージを作成
    message = ChatMessage(
        content=[
            TextContent(text="この画像には何が写っていますか？"),
            DataContent(
                data=image_data,
                mime_type="image/jpeg"
            )
        ]
    )

    response = await agent.run(message)
    print(response)

asyncio.run(image_example())

ツールの使い方

エージェントにツール（関数）を提供することで、外部システムとの連携や特定のタスクの実行が可能になります。Microsoft Agent Frameworkでは、3つの方式でツールを統合できます。

1. カスタム関数ツール

Pythonの関数を直接ツールとして使用できます。

基本的なツール定義

from agent_framework import AIFunction
from typing import Annotated

@AIFunction
async def get_weather(
    city: Annotated[str, "都市名（例: 東京、大阪）"],
    unit: Annotated[str, "温度の単位（celsius または fahrenheit）"] = "celsius"
) -> str:
    """指定された都市の現在の天気を取得します。"""
    # 実際のAPI呼び出しをここに実装
    # この例ではダミーデータを返す
    return f"{city}の天気は晴れ、気温は25度です（{unit}）"

@AIFunction
async def search_web(
    query: Annotated[str, "検索キーワード"]
) -> str:
    """Web検索を実行します。"""
    # 実際の検索API呼び出しをここに実装
    return f"'{query}'の検索結果: ..."

ツールを使うエージェント

async def tool_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="ToolAgent",
        instructions="""
        あなたは天気情報とWeb検索を使えるアシスタントです。
        必要に応じてツールを使用して、正確な情報を提供してください。
        """,
        tools=[get_weather, search_web]  # ツールを登録
    )

    # エージェントが自動的にツールを呼び出す
    response = await agent.run("東京の天気を教えてください")
    print(response)

    # 複数のツールを組み合わせる
    response2 = await agent.run("大阪の天気を調べて、雨が降りそうなら傘の情報を検索してください")
    print(response2)

asyncio.run(tool_example())

Pydanticによる型安全なツール

より複雑なツールはPydanticモデルで定義できます。

from pydantic import BaseModel, Field
from typing import List

class SearchResult(BaseModel):
    title: str = Field(description="検索結果のタイトル")
    url: str = Field(description="URL")
    snippet: str = Field(description="抜粋")

class SearchResults(BaseModel):
    results: List[SearchResult]
    total_count: int

@AIFunction
async def advanced_search(
    query: Annotated[str, "検索クエリ"],
    max_results: Annotated[int, "最大結果数"] = 10
) -> SearchResults:
    """高度なWeb検索を実行し、構造化された結果を返します。"""
    # 実装
    return SearchResults(
        results=[
            SearchResult(
                title="サンプル結果",
                url="https://example.com",
                snippet="これはサンプルです..."
            )
        ],
        total_count=1
    )

2. MCPツール

Model Context Protocol（MCP）を使用して、外部ツールサーバーと統合できます。

ファイルシステムMCPツール

from agent_framework.mcp import MCPStdioTool

async def mcp_filesystem_example():
    # ファイルシステムMCPサーバーと接続
    filesystem_tools = await MCPStdioTool.from_server(
        command="uvx",
        args=["mcp-server-filesystem", "--allowed-directory", "./documents"]
    )

    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="FileAgent",
        instructions="""
        あなたはファイルシステムにアクセスできるアシスタントです。
        ユーザーの要求に応じて、ファイルの読み書きを行ってください。
        """,
        tools=filesystem_tools  # MCPツールを登録
    )

    # エージェントがファイルシステムにアクセス
    response = await agent.run("documentsフォルダ内のREADME.mdを読んで要約してください")
    print(response)

asyncio.run(mcp_filesystem_example())

データベースMCPツール

from agent_framework.mcp import MCPWebsocketTool

async def mcp_database_example():
    # データベースMCPサーバーと接続
    db_tools = await MCPWebsocketTool.from_server(
        url="ws://localhost:8080/mcp"
    )

    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="DatabaseAgent",
        instructions="""
        あなたはデータベースにアクセスできるアシスタントです。
        ユーザーの質問に答えるために、データベースにクエリを実行してください。
        """,
        tools=db_tools
    )

    response = await agent.run("2024年の売上トップ10の商品を教えてください")
    print(response)

asyncio.run(mcp_database_example())

3. OpenAPIツール

既存のREST APIをOpenAPI仕様から自動的にツール化できます。

from agent_framework.tools import OpenAPITool

async def openapi_example():
    # OpenAPI仕様からツールを生成
    api_tools = await OpenAPITool.from_url(
        "https://api.example.com/openapi.json"
    )

    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="APIAgent",
        instructions="""
        あなたは外部APIにアクセスできるアシスタントです。
        ユーザーの要求に応じて、APIを呼び出してください。
        """,
        tools=api_tools
    )

    response = await agent.run("ユーザーID 123の情報を取得してください")
    print(response)

asyncio.run(openapi_example())

ツール実行の制御

Human-in-the-loop（人間による承認）

重要な操作には人間の承認を必要とすることができます。

from agent_framework import FunctionMiddleware

class HumanApprovalMiddleware(FunctionMiddleware):
    async def on_invoke(self, context, next):
        # 危険な操作には承認を要求
        dangerous_functions = ["delete_file", "drop_database", "send_email"]

        if context.function.name in dangerous_functions:
            print(f"\n⚠️  警告: {context.function.name}を実行しようとしています")
            print(f"引数: {context.arguments}")

            approval = input("実行を承認しますか？ (yes/no): ")

            if approval.lower() != "yes":
                raise ValueError("ユーザーによって拒否されました")

        # 承認された、または安全な操作の場合は続行
        return await next(context)

# 使用例
async def approval_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="ControlledAgent",
        instructions="あなたはファイル操作ができるアシスタントです。",
        tools=[delete_file, read_file],
        function_middleware=[HumanApprovalMiddleware()]  # ミドルウェアを追加
    )

    # delete_fileが呼ばれる場合、承認が必要
    response = await agent.run("不要なファイルを削除してください")
    print(response)

asyncio.run(approval_example())

ワークフローとオーケストレーション

複数のエージェントやタスクを組み合わせて、複雑な処理を実現できます。Microsoft Agent Frameworkは、イベント駆動のワークフローシステムを提供しています。

基本的なワークフローパターン

1. 順次実行（Sequential）

エージェントを順番に実行します。

from agent_framework.workflows import Sequential

async def sequential_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 各ステップのエージェント
    researcher = client.create_agent(
        name="Researcher",
        instructions="与えられたトピックについて調査してください。",
        tools=[search_web]
    )

    analyzer = client.create_agent(
        name="Analyzer",
        instructions="調査結果を分析し、重要なポイントを抽出してください。"
    )

    writer = client.create_agent(
        name="Writer",
        instructions="分析結果を基に、分かりやすいレポートを作成してください。"
    )

    # ワークフロー構築
    workflow = Sequential()
        .add_executor(researcher)  # 1. 調査
        .add_executor(analyzer)    # 2. 分析
        .add_executor(writer)      # 3. レポート作成
        .build()

    # 実行
    result = await workflow.run("AIエージェントの最新動向についてレポートを作成してください")
    print(result)

asyncio.run(sequential_example())

2. 並行実行（Concurrent）

複数のエージェントを同時に実行します。

from agent_framework.workflows import Concurrent

async def concurrent_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 異なる観点で調査するエージェント
    tech_researcher = client.create_agent(
        name="TechResearcher",
        instructions="技術的な観点から調査してください。"
    )

    business_researcher = client.create_agent(
        name="BusinessResearcher",
        instructions="ビジネス的な観点から調査してください。"
    )

    market_researcher = client.create_agent(
        name="MarketResearcher",
        instructions="市場動向の観点から調査してください。"
    )

    # ワークフロー構築（並行実行）
    workflow = Concurrent()
        .add_executor(tech_researcher)
        .add_executor(business_researcher)
        .add_executor(market_researcher)
        .build()

    # 実行（3つのエージェントが同時に動作）
    results = await workflow.run("AIエージェント市場について調査してください")

    # 結果は配列で返される
    print("技術的観点:", results[0])
    print("ビジネス観点:", results[1])
    print("市場動向:", results[2])

asyncio.run(concurrent_example())

3. グループチャット（GroupChat）

複数のエージェントが協力して問題を解決します。

from agent_framework.workflows import GroupChat

async def group_chat_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 異なる専門性を持つエージェント
    researcher = client.create_agent(
        name="Researcher",
        instructions="事実に基づいた情報を提供する研究者です。"
    )

    critic = client.create_agent(
        name="Critic",
        instructions="提案に対して批判的に検討する評論家です。"
    )

    implementer = client.create_agent(
        name="Implementer",
        instructions="具体的な実装方法を提案する実装者です。"
    )

    # グループチャット構築
    workflow = GroupChat()
        .add_agent(researcher)
        .add_agent(critic)
        .add_agent(implementer)
        .set_selection_strategy("round_robin")  # 順番に発言
        .set_max_turns(10)  # 最大10ターン
        .build()

    # 実行（エージェント同士が議論）
    result = await workflow.run("効果的なAIエージェントシステムの設計方法を議論してください")
    print(result)

asyncio.run(group_chat_example())

高度なワークフロー機能

コンディショナルエッジ（条件分岐）

実行結果に基づいて、次に実行するエージェントを動的に決定できます。

from agent_framework.workflows import WorkflowBuilder
from agent_framework.workflows.edges import ConditionalEdge

async def conditional_workflow_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # エージェント定義
    validator = client.create_agent(
        name="Validator",
        instructions="入力データの妥当性を検証し、問題があれば指摘してください。"
    )

    processor = client.create_agent(
        name="Processor",
        instructions="検証済みデータを処理してください。"
    )

    error_handler = client.create_agent(
        name="ErrorHandler",
        instructions="エラーを分析し、修正方法を提案してください。"
    )

    # ワークフロー構築
    workflow = WorkflowBuilder()

    workflow.add_executor("validator", validator)
    workflow.add_executor("processor", processor)
    workflow.add_executor("error_handler", error_handler)

    # 条件分岐の定義
    def route_based_on_validation(context):
        # validatorの出力を確認
        if "エラー" in context.output or "問題" in context.output:
            return "error_handler"  # エラーがあればエラーハンドラーへ
        else:
            return "processor"  # 問題なければ処理へ

    # エッジを追加
    workflow.add_edge(
        from_executor="validator",
        edge=ConditionalEdge(route_based_on_validation)
    )
    workflow.add_edge("processor", END)
    workflow.add_edge("error_handler", END)

    final_workflow = workflow.build()

    # 実行
    result = await final_workflow.run("データ: ...")
    print(result)

asyncio.run(conditional_workflow_example())

チェックポイント（中断と再開）

長時間実行されるワークフローの途中状態を保存し、障害発生時に再開できます。

from agent_framework.workflows import Sequential
from agent_framework.workflows.checkpoint import FileCheckpointStrategy

async def checkpoint_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 時間がかかるステップ
    step1 = client.create_agent(
        name="Step1",
        instructions="大量のデータを収集してください。"  # 1時間かかる
    )

    step2 = client.create_agent(
        name="Step2",
        instructions="収集したデータを分析してください。"  # 2時間かかる
    )

    step3 = client.create_agent(
        name="Step3",
        instructions="分析結果をレポートにまとめてください。"  # 1時間かかる
    )

    # チェックポイント機能を有効化
    workflow = Sequential()
        .add_executor(step1)
        .add_executor(step2)
        .add_executor(step3)
        .enable_checkpointing(
            strategy=FileCheckpointStrategy(directory="./checkpoints")
        )
        .build()

    try:
        # 実行
        result = await workflow.run("大規模データ分析を実行してください")
        print(result)
    except Exception as e:
        print(f"エラー発生: {e}")
        print("チェックポイントから再開できます")

        # 失敗した場所から再開
        # checkpoint_idは自動的に保存されている
        resumed_workflow = workflow.resume_from_last_checkpoint()
        result = await resumed_workflow.run()
        print(result)

asyncio.run(checkpoint_example())

ストリーミングイベント

ワークフローの進捗をリアルタイムで監視できます。

async def streaming_workflow_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 複数ステップのワークフロー
    workflow = Sequential()
        .add_executor(agent1)
        .add_executor(agent2)
        .add_executor(agent3)
        .build()

    print("ワークフロー実行中...")

    # ストリーミングで進捗を監視
    async for event in workflow.run_streaming("タスクを実行してください"):
        if event.type == "executor_start":
            print(f"🔄 {event.executor_name} を開始...")

        elif event.type == "executor_progress":
            print(f"⏳ {event.executor_name}: {event.progress}")

        elif event.type == "executor_complete":
            print(f"✅ {event.executor_name} が完了")
            print(f"   結果: {event.result[:100]}...")  # 最初の100文字

        elif event.type == "workflow_complete":
            print(f"\n🎉 ワークフロー完了!")
            print(f"最終結果:\n{event.result}")

asyncio.run(streaming_workflow_example())

開発アプローチ：コードファースト vs 宣言的ワークフロー

Microsoft Agent Frameworkでは、エージェントとワークフローを構築する際に2つの主要なアプローチが用意されています。プロジェクトの性質やチーム構成に応じて、最適な方法を選択できます。

フレームワークが推奨する基本アプローチ

Microsoft Agent Frameworkの公式ドキュメントとクイックスタートガイドは、コードファースト（Code-First）アプローチを基本として紹介しています。

# コードファーストアプローチの例
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

# エージェントをプログラムで定義
agent = AzureOpenAIResponsesClient(
    credential=AzureCliCredential()
).create_agent(
    name="AssistantAgent",
    instructions="あなたは親切なアシスタントです。"
)

# ワークフローもコードで構築
from agent_framework.workflows import Sequential

workflow = Sequential()
    .add_executor(researcher)
    .add_executor(analyzer)
    .add_executor(writer)
    .build()

このアプローチが推奨される理由：

柔軟性と拡張性: Pythonや.NETの全機能を活用でき、複雑なロジックや条件分岐を自由に実装できます
型安全性: IDEの補完機能やコンパイル時の型チェックが効きます
バージョン管理: コードベースとして管理でき、Gitなどでの変更追跡が容易です
デバッグのしやすさ: ブレークポイントやログを使った詳細なデバッグが可能です

宣言的ワークフロー（Declarative Workflows）

一方、宣言的ワークフローは、YAML/JSON形式の設定ファイルでワークフローを定義するノーコード/ローコードアプローチです。

宣言的ワークフローの例

# HelloWorld.yaml
name: HelloWorldWorkflow
description: シンプルな挨拶ワークフロー

steps:
  - id: greeter
    type: InvokeAzureAgent
    agentId: "greeter-agent-id"
    inputs:
      message: "こんにちは！"

  - id: responder
    type: InvokeAzureAgent
    agentId: "responder-agent-id"
    dependsOn: greeter

.NETでの実行例

// YAMLファイルからワークフローを読み込み
Workflow<string> workflow = DeclarativeWorkflowBuilder.Build<string>("HelloWorld.yaml", options);

// 実行はコード定義のワークフローと同じ
var result = await workflow.RunAsync("タスクを実行");

宣言的ワークフローの特徴

ノーコードで構築: プログラミング知識がなくても、設定ファイルでワークフローを定義できます
Azure Foundry統合: Azure AI Foundryプロジェクトで定義済みのエージェントを簡単に組み合わせられます
視覚的な理解: YAML構造により、ワークフローの流れが視覚的に分かりやすくなります
実行時の柔軟性: コードの再コンパイルなしに、設定ファイルを変更するだけでワークフローを更新できます

注意: 現時点（2025年1月）では、宣言的ワークフローは主に.NETでサポートされており、Python対応は開発中です。

どちらを選ぶべきか？使い分けガイド

コードファーストを選ぶべきケース

以下の場合は、コードファーストが最適です：

高度なカスタマイズが必要
- 複雑な条件分岐やループ処理
- カスタムミドルウェアの実装
- 独自のツールやエグゼキューターの開発
既存コードベースとの統合
- PythonやC#の既存プロジェクトに組み込む
- 既存のライブラリやフレームワークと連携
開発チームがエンジニア中心
- 型安全性や自動補完を活用したい
- ユニットテストや統合テストを重視
迅速なプロトタイピング
- Jupyter NotebookやREPL環境で試行錯誤
- SDK機能をフルに活用した実験

実例: 研究論文を分析するワークフロー

# 細かい制御が必要なケース
from agent_framework.workflows import WorkflowBuilder

workflow = WorkflowBuilder()
    .add_executor("pdf_parser", pdf_parser_agent)
    .add_executor("summarizer", summarizer_agent)
    .add_executor("analyzer", analyzer_agent)

# カスタム条件分岐
workflow.add_edge(
    from_executor="pdf_parser",
    edge=ConditionalEdge(lambda state:
        "summarizer" if state.page_count < 10 else "analyzer"
    )
)

# チェックポイント機能を追加
workflow.enable_checkpointing(
    strategy=FileCheckpointStrategy(directory="./checkpoints")
)

final_workflow = workflow.build()

宣言的ワークフローを選ぶべきケース

以下の場合は、宣言的ワークフローが適しています：

非エンジニアとの協業
- ビジネスアナリストやプロダクトマネージャーがワークフローを設計
- プログラミング知識がないメンバーも編集可能
Azure Foundry中心の運用
- Azure AI Foundryで管理されているエージェントを再利用
- クラウドベースのエージェントオーケストレーション
GUI/ビジュアルツールでの管理
- 将来的にビジュアルエディターで編集する予定
- 設定ファイルベースの運用が望ましい
シンプルなワークフロー
- 基本的な順次実行や並行実行のみ
- 複雑なロジックが不要

実例: カスタマーサポートワークフロー

# customer_support.yaml - 非エンジニアでも編集可能
name: CustomerSupportWorkflow
description: お客様からの問い合わせを処理

steps:
  - id: classifier
    type: InvokeAzureAgent
    agentId: "inquiry-classifier"
    description: 問い合わせを分類

  - id: technical_support
    type: InvokeAzureAgent
    agentId: "tech-support-agent"
    condition: "classifier.category == 'technical'"

  - id: billing_support
    type: InvokeAzureAgent
    agentId: "billing-agent"
    condition: "classifier.category == 'billing'"

  - id: escalation
    type: Question
    message: "担当者にエスカレートしますか？"
    condition: "not resolved"

ハイブリッドアプローチ

多くの実際のプロジェクトでは、両方のアプローチを組み合わせるのが効果的です：

# 1. コアロジックはコードで定義
class CustomDataProcessor:
    async def process(self, data):
        # 複雑な前処理ロジック
        return processed_data

# 2. 標準的な部分は宣言的に定義
workflow = DeclarativeWorkflowBuilder.Build<string>("standard_flow.yaml", options)

# 3. カスタムエグゼキューターを追加
workflow.add_custom_executor("preprocessor", CustomDataProcessor())

# 4. 実行
result = await workflow.run(input_data)

公式ガイドラインのまとめ

Microsoft Agent Frameworkのドキュメントとサンプルコードから得られる推奨事項：

観点	推奨アプローチ	理由
スタート地点	コードファースト	公式クイックスタートはコードベースのサンプルから始まる
Python開発者	コードファースト	現時点でPythonの宣言的ワークフロー対応は開発中
細かい制御	コードファースト	Pythonサンプルは「.pyファイルを順番に読み進める」ことを推奨
Azure Foundry活用	宣言的ワークフロー	クラウド上の既存エージェントを組み合わせる場合に最適
ビジュアル管理	宣言的ワークフロー	設定ファイルベースで運用したい場合

実際の開発ワークフロー例

典型的なプロジェクトの進め方：

プロトタイプ段階（コードファースト）

# Jupyter Notebookで素早く試行錯誤
agent = client.create_agent(...)
response = await agent.run("テストタスク")

開発段階（コードファースト）

# 詳細な実装とテスト
class ProductionWorkflow:
    def __init__(self):
        self.workflow = Sequential()...

    async def run(self, input):
        return await self.workflow.run(input)

運用段階（ハイブリッド）

# 環境ごとに設定を切り替え
production:
  agents:
    - id: prod-agent-1
      endpoint: ${PROD_ENDPOINT}

staging:
  agents:
    - id: staging-agent-1
      endpoint: ${STAGING_ENDPOINT}

まとめ

基本的にはコードファーストで始めることを推奨
Azure Foundryの既存エージェントを活用する場合や、非エンジニアとの協業が必要な場合は宣言的ワークフローを検討
実行パターン（workflow.run()など）は両アプローチで共通なので、後から切り替えることも可能
Python開発者は現時点でコードファーストが主流（宣言的サポートは今後強化予定）

DevUI：開発・テスト用のWebインターフェース

Microsoft Agent Frameworkには、エージェントとワークフローをブラウザ上でテスト・デバッグするためのDevUIというサンプルアプリケーションが付属しています。

DevUIとは

DevUIは、既存のエージェント/ワークフローを発見・実行・デバッグするための軽量なWebインターフェースです。

主な機能

ディレクトリベースの自動検出: 指定したディレクトリ内のエージェント/ワークフローを自動的に発見
インメモリ登録: プログラムからエージェントを直接登録して即座にテスト
サンプルギャラリー: 公式サンプルをブラウザからダウンロード・実行
OpenAI互換API: OpenAI SDKからアクセス可能なREST API
会話履歴管理: マルチターン会話のテストが可能
OpenTelemetryトレース表示: エージェント実行のトレース情報を可視化

重要な注意点

DevUIはサンプルアプリケーションです

開発・テスト環境専用：本番環境での使用は想定されていません

ワークフロー作成機能なし：AutoGenのようなGUIでワークフローを組み立てる機能はありません

既存エンティティの実行に特化：コードまたはYAMLで事前定義されたエージェント/ワークフローを読み込んで実行・デバッグする用途

インストール

# DevUIパッケージをインストール
pip install agent-framework-devui --pre

使い方

パターン1：インメモリモード（最もシンプル）

コードから直接エージェントを登録して起動できます。

from agent_framework import ChatAgent
from agent_framework.openai import OpenAIChatClient
from agent_framework.devui import serve

def get_weather(location: str) -> str:
    """天気情報を取得します。"""
    return f"{location}の天気: 晴れ、気温25度"

# エージェントを作成
agent = ChatAgent(
    name="WeatherAgent",
    description="天気情報を提供するエージェント",
    chat_client=OpenAIChatClient(),
    tools=[get_weather]
)

# DevUIを起動（ブラウザが自動的に開く）
serve(entities=[agent], auto_open=True)
# → http://localhost:8080 でアクセス可能

実行:

python my_agent.py

ブラウザが自動的に開き、作成したエージェントをすぐにテストできます。

パターン2：ディレクトリ検出モード

エージェント/ワークフローをディレクトリ構造で管理し、DevUIが自動検出します。

ディレクトリ構造:

my_agents/
├── weather_agent/
│   ├── __init__.py      # agent = ChatAgent(...) をエクスポート
│   ├── agent.py         # 実装コード
│   └── .env             # 環境変数（オプション）
├── translator_agent/
│   ├── __init__.py      # agent = ChatAgent(...) をエクスポート
│   └── agent.py
└── my_workflow/
    ├── __init__.py      # workflow = Sequential()... をエクスポート
    └── workflow.py

weather_agent/__init__.py の例:

from agent_framework import ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

def get_weather(location: str) -> str:
    """天気情報を取得"""
    return f"{location}の天気: 晴れ、25度"

# このagent変数が自動的に検出される
agent = ChatAgent(
    name="WeatherAgent",
    description="天気情報を提供するエージェント",
    chat_client=AzureOpenAIChatClient(
        credential=AzureCliCredential()
    ),
    tools=[get_weather]
)

DevUIの起動:

# ディレクトリを指定して起動
devui ./my_agents --port 8080

# または、現在のディレクトリから起動
cd my_agents
devui

# トレース表示を有効化
devui ./my_agents --tracing framework

ブラウザで http://localhost:8080 にアクセスすると、すべてのエージェントとワークフローが自動的にリストアップされます。

パターン3：複数エンティティの一括登録

1つのPythonスクリプトから複数のエージェントとワークフローを登録できます。

from agent_framework import ChatAgent
from agent_framework.workflows import Sequential
from agent_framework.openai import OpenAIChatClient
from agent_framework.devui import serve

# エージェント1
weather_agent = ChatAgent(
    name="WeatherAgent",
    chat_client=OpenAIChatClient(),
    tools=[get_weather]
)

# エージェント2
translator_agent = ChatAgent(
    name="TranslatorAgent",
    chat_client=OpenAIChatClient(),
    tools=[translate_text]
)

# ワークフロー
workflow = Sequential()
    .add_executor(weather_agent)
    .add_executor(translator_agent)
    .build()

# すべてをDevUIに登録
serve(
    entities=[weather_agent, translator_agent, workflow],
    port=8090,
    auto_open=True
)

OpenAI互換APIの使用

DevUIはOpenAI Responses API互換のエンドポイントを提供しており、OpenAI SDKから直接アクセスできます。

cURLでの実行

curl -X POST http://localhost:8080/v1/responses \
  -H "Content-Type: application/json" \
  -d '{
    "model": "WeatherAgent",
    "input": "東京の天気を教えてください"
  }'

Python（OpenAI SDK）での実行

from openai import OpenAI

# DevUIのAPIエンドポイントに接続
client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="not-needed"  # ローカル実行時はAPIキー不要
)

# エージェント名をmodelとして指定
response = client.responses.create(
    model="WeatherAgent",  # DevUIで登録したエージェント名
    input="大阪の天気を教えてください"
)

print(response.output[0].content[0].text)

# ストリーミングも対応
for chunk in client.responses.create(
    model="WeatherAgent",
    input="福岡の天気は？",
    stream=True
):
    if chunk.output:
        print(chunk.output[0].content[0].text, end="", flush=True)

マルチターン会話

会話の文脈を保持した連続的なやり取りが可能です。

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="not-needed"
)

# 会話セッションを作成
conversation = client.conversations.create(
    metadata={"agent_id": "WeatherAgent"}
)

# 1回目の質問
response1 = client.responses.create(
    model="WeatherAgent",
    input="東京の天気を教えてください",
    conversation=conversation.id
)
print(response1.output[0].content[0].text)

# 2回目の質問（文脈を保持）
response2 = client.responses.create(
    model="WeatherAgent",
    input="明日はどうですか？",  # "東京"を覚えている
    conversation=conversation.id
)
print(response2.output[0].content[0].text)

DevUIの主要機能

1. エンティティ一覧表示

登録されているすべてのエージェントとワークフローを確認できます。

# APIでエンティティ一覧を取得
curl http://localhost:8080/v1/entities

2. ホットリロード

開発中にエージェントのコードを変更した場合、サーバーを再起動せずに更新できます。

# 開発モードで起動（ファイル変更を自動検出）
devui ./my_agents --reload

# または、手動でリロード
curl -X POST http://localhost:8080/v1/entities/{entity_id}/reload

3. OpenTelemetryトレース表示

エージェントの実行フローを可視化できます。

# トレース表示を有効化
devui ./my_agents --tracing framework

# すべてのトレースを表示
devui ./my_agents --tracing all

ブラウザのDevUIで、各エージェントの実行ステップ、ツール呼び出し、エラーなどが時系列で表示されます。

4. サンプルギャラリー

エンティティが1つも検出されない場合、DevUIは公式サンプルギャラリーを表示します。

ブラウザからサンプルコードをダウンロード
レビュー後、ローカルで実行
学習用のテンプレートとして活用

DevUIのCLIオプション

devui [ディレクトリ] [オプション]

オプション:
  --port, -p      ポート番号（デフォルト: 8080）
  --host          ホスト（デフォルト: 127.0.0.1）
  --headless      UIなし、APIのみ起動
  --config        YAML設定ファイル
  --tracing       トレースレベル（none|framework|workflow|all）
  --reload        ファイル変更の自動リロード

DevUIでできないこと（重要）

DevUIは既存のエージェント/ワークフローの実行・デバッグツールであり、以下のことはできません：

GUIでのワークフロー作成
- AutoGenのようなビジュアルエディターは提供されていません
- ワークフローは必ずコードまたはYAMLで事前定義する必要があります
ブラウザ上でのエージェント定義
- エージェントの新規作成はコードで行います
- DevUIはコードで定義されたエージェントを「読み込んで実行する」ツールです
本番環境での使用
- DevUIはローカル開発・テスト専用です
- セキュリティ機能が限定的なため、インターネットに公開しないでください

推奨される開発フロー

DevUIを活用した典型的な開発フローは以下の通りです：

エージェント作成（コード）

# my_agents/weather_agent/__init__.py
agent = ChatAgent(...)

DevUIで即座にテスト

devui ./my_agents
# ブラウザで動作確認

コードを修正

# エージェントの指示を調整
agent = ChatAgent(
    instructions="より詳しい天気情報を提供してください"
)

ホットリロードで即反映

# --reloadオプションで起動していれば自動更新
# または、ブラウザのリロードボタンをクリック

満足したら本番コードに統合

# アプリケーション本体に組み込む
from my_agents.weather_agent import agent

まとめ

DevUIは以下のような場面で非常に便利です：

迅速なプロトタイピング: エージェントをコードで書いて、すぐにブラウザでテスト
デバッグ: トレース機能で実行フローを可視化
チーム共有: ディレクトリベースで複数のエージェントを管理・共有
API統合テスト: OpenAI互換APIで外部システムからの呼び出しをテスト

ただし、以下の点に注意してください：

ワークフロー作成はコードで行う：GUIエディターはありません
開発環境専用：本番環境では使用しないでください
既存エンティティの実行に特化：新規作成はコードエディターで行います

実践的なパターンとベストプラクティス

実際のプロジェクトで役立つパターンとベストプラクティスを紹介します。

1. 環境変数とシークレット管理

ベストプラクティス:

APIキーなどの機密情報はコードにハードコードしない
.envファイルと環境変数を活用
本番環境ではAzure Key Vaultを使用

# .env ファイル
"""
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4
AZURE_KEY_VAULT_URL=https://your-vault.vault.azure.net/
"""

# Pythonコード
from dotenv import load_dotenv
import os

# 開発環境: .envから読み込み
load_dotenv()

# 本番環境: Azure Key Vaultから読み込み
if os.getenv("ENVIRONMENT") == "production":
    from azure.keyvault.secrets import SecretClient
    from azure.identity import DefaultAzureCredential

    credential = DefaultAzureCredential()
    vault_url = os.getenv("AZURE_KEY_VAULT_URL")
    client = SecretClient(vault_url=vault_url, credential=credential)

    api_key = client.get_secret("OPENAI-API-KEY").value
else:
    api_key = os.getenv("AZURE_OPENAI_API_KEY")

2. エラーハンドリング

ベストプラクティス:

適切な例外処理
リトライロジック
ユーザーフレンドリーなエラーメッセージ

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),  # 最大3回リトライ
    wait=wait_exponential(multiplier=1, min=4, max=10)  # 指数バックオフ
)
async def robust_agent_call(agent, message):
    """リトライ機能付きのエージェント呼び出し"""
    try:
        response = await agent.run(message)
        return response
    except Exception as e:
        print(f"エラー発生: {e}")
        raise  # リトライのために例外を再スロー

# 使用例
async def error_handling_example():
    try:
        response = await robust_agent_call(agent, "タスクを実行")
        print(response)
    except Exception as e:
        print(f"すべてのリトライが失敗しました: {e}")
        # フォールバック処理
        print("デフォルトの応答を返します")

asyncio.run(error_handling_example())

3. ロギングと監視

ベストプラクティス:

構造化ログの使用
適切なログレベル
OpenTelemetryによる自動計測

import logging
from agent_framework.observability import configure_observability

# ロギング設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# OpenTelemetry設定
configure_observability(
    service_name="my-agent-app",
    tracing_enabled=True,
    metrics_enabled=True
)

async def monitored_example():
    logger.info("エージェント実行を開始します")

    try:
        agent = client.create_agent(...)
        response = await agent.run("タスク")

        logger.info("エージェント実行が成功しました", extra={
            "response_length": len(response),
            "agent_name": agent.name
        })

        return response
    except Exception as e:
        logger.error("エージェント実行が失敗しました", exc_info=True, extra={
            "error_type": type(e).__name__
        })
        raise

asyncio.run(monitored_example())

4. テスト戦略

ベストプラクティス:

ユニットテスト
統合テスト
モック��使用

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_agent_with_mock():
    """エージェントのユニットテスト"""

    # モックのChat Clientを作成
    mock_client = AsyncMock()
    mock_client.create_agent.return_value = AsyncMock()
    mock_client.create_agent.return_value.run.return_value = "モックの応答"

    # テスト対象の関数
    async def process_query(query):
        agent = mock_client.create_agent(
            name="TestAgent",
            instructions="テスト用"
        )
        return await agent.run(query)

    # テスト実行
    result = await process_query("テストクエリ")

    # 検証
    assert result == "モックの応答"
    mock_client.create_agent.assert_called_once()

# 統合テスト
@pytest.mark.asyncio
@pytest.mark.integration
async def test_agent_integration():
    """エージェントの統合テスト（実際のLLMを使用）"""

    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    agent = client.create_agent(
        name="IntegrationTestAgent",
        instructions="簡潔に答えてください。"
    )

    response = await agent.run("1+1は？")

    # 応答に数字の2が含まれることを確認
    assert "2" in response

5. パフォーマンス最適化

ベストプラクティス:

並行処理の活用
キャッシング
トークン使用量の最適化

import asyncio
from functools import lru_cache

# キャッシング
@lru_cache(maxsize=100)
def get_agent_instructions(agent_type: str) -> str:
    """エージェントの指示をキャッシュ"""
    instructions = {
        "researcher": "あなたは調査を行う研究者です...",
        "analyst": "あなたは分析を行うアナリストです...",
    }
    return instructions.get(agent_type, "デフォルトの指示")

# 並行処理
async def parallel_processing_example():
    client = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    )

    # 複数のタスクを並行実行
    tasks = [
        client.create_agent(
            name=f"Agent{i}",
            instructions=get_agent_instructions("researcher")
        ).run(f"タスク{i}")
        for i in range(5)
    ]

    # 並行実行（5つ同時）
    results = await asyncio.gather(*tasks)

    return results

asyncio.run(parallel_processing_example())

既存フレームワークからの移行

Semantic KernelやAutoGenを使っている場合、Microsoft Agent Frameworkへの移行は比較的スムーズです。

Semantic Kernelからの移行

主な変更点

名前空間の変更
- Microsoft.SemanticKernel.* → Microsoft.Extensions.AI.*
ボイラープレートの削減
- より簡潔なAPI
メモリ管理の簡素化
- 組み込みのメモリシステム

移行例

Semantic Kernel:

var kernel = Kernel.CreateBuilder()
    .AddAzureOpenAIChatCompletion(
        deploymentName: "gpt-4",
        endpoint: endpoint,
        apiKey: apiKey)
    .Build();

var chatCompletionService = kernel.GetRequiredService<IChatCompletionService>();
var history = new ChatHistory();
history.AddUserMessage("こんにちは");

var result = await chatCompletionService.GetChatMessageContentAsync(
    history,
    executionSettings: new OpenAIPromptExecutionSettings(),
    kernel: kernel);

Microsoft Agent Framework:

var agent = new OpenAIClient(apiKey)
    .GetOpenAIResponseClient("gpt-4")
    .CreateAIAgent(
        name: "Assistant",
        instructions: "あなたは親切なアシスタントです。"
    );

var result = await agent.RunAsync("こんにちは");

AutoGenからの移行

主な変更点

エージェントクラスの変更
- AssistantAgent → ChatAgent
Workflow APIによる構成性・耐久性強化
- より柔軟なワークフロー定義
- チェックポイント機能の組み込み
型安全性の向上
- Pydanticによる厳格な型チェック

移行例

AutoGen:

from autogen import AssistantAgent, UserProxyAgent

assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4"}
)

user_proxy = UserProxyAgent(
    name="user",
    human_input_mode="NEVER"
)

user_proxy.initiate_chat(
    assistant,
    message="こんにちは"
)

Microsoft Agent Framework:

from agent_framework.azure import AzureOpenAIResponsesClient

client = AzureOpenAIResponsesClient(...)

agent = client.create_agent(
    name="Assistant",
    instructions="あなたは親切なアシスタントです。"
)

response = await agent.run("こんにちは")

移行のステップ

評価フェーズ
- 現在のコードベースを分析
- 使用している機能をリストアップ
- Microsoft Agent Frameworkでの対応方法を確認
パイロットプロジェクト
- 小規模な機能で試験的に移行
- パフォーマンスと互換性を検証
段階的移行
- モジュールごとに少しずつ移行
- 既存システムと並行稼働
- A2Aプロトコルで相互運用
完全移行
- すべてのコードを移行
- レガシーコードの削除
- ドキュメント更新

よくある質問とトラブルシューティング

Q1: どのLLMプロバイダーが使えますか？

A: 以下のプロバイダーに対応しています：

Azure OpenAI Service（推奨）
OpenAI
Azure AI Foundry
その他のOpenAI互換API

将来的にはさらに多くのプロバイダーに対応予定です。

Q2: 料金はかかりますか？

A: Microsoft Agent Framework自体は無料でオープンソースです。ただし、使用するLLMサービス（Azure OpenAI、OpenAIなど）には料金がかかります。

Q3: 本番環境で使えますか？

A: はい。Microsoft Agent Frameworkは本番運用を想定して設計されています。

OpenTelemetryによる観測可能性
チェックポイント機能による耐久性
エンタープライズグレードのセキュリティ
Azure環境との深い統合

ただし、現在はプレビュー版（--pre）のため、本番環境に導入する前に十分なテストを行うことを推奨します。

Q4: LangChainとの違いは？

A: 主な違い：

項目	Microsoft Agent Framework	LangChain
焦点	エンタープライズ、本番運用	柔軟性、エコシステム
言語	Python + C#/.NET	Python + JavaScript
観測可能性	OpenTelemetry組み込み	LangSmith（有料）
ワークフロー	DAG、イベント駆動	任意のグラフ（サイクル可）
セキュリティ	Azure統合、エンタープライズグレード	基本的な機能

詳細はMicrosoft Agent Framework vs LangChain - 徹底比較を参照してください。

Q5: トークン使用量を最適化する方法は？

A: 以下の方法があります：

適切なモデル選択

# 簡単なタスクにはgpt-3.5-turboを使用
simple_agent = client.create_agent(
    model="gpt-3.5-turbo",  # より安価
    ...
)

# 複雑なタスクにはgpt-4を使用
complex_agent = client.create_agent(
    model="gpt-4",
    ...
)

指示の簡潔化

# ❌ 冗長な指示
instructions = """
あなたは非常に親切で、知識豊富で、ユーザーフレンドリーな
アシスタントです。常にユーザーの質問に対して、詳細で、
分かりやすく、正確な回答を提供してください...
"""

# ✅ 簡潔な指示
instructions = "簡潔に答えてください。"

会話履歴の管理

# 古いメッセージを削除
thread = AgentThread()
thread.trim_messages(max_messages=10)  # 最新10件のみ保持

トラブルシューティング

問題: 認証エラーが発生する

症状:

azure.core.exceptions.ClientAuthenticationError: Authentication failed

解決方法:

# Azure CLIの再認証
az login
az account show

# 正しいサブスクリプションを確認
az account list
az account set --subscription "Your Subscription"

問題: インポートエラー

症状:

ModuleNotFoundError: No module named 'agent_framework'

解決方法:

# 最新版を再インストール
pip uninstall agent-framework
pip install agent-framework --pre

# または、uvを使用
uv pip install agent-framework --pre

問題: ワークフローが途中で止まる

症状:
ワークフローの実行が途中で応答しなくなる

解決方法:

# タイムアウトを設定
import asyncio

async def run_with_timeout():
    try:
        result = await asyncio.wait_for(
            workflow.run("タスク"),
            timeout=300  # 5分でタイムアウト
        )
        return result
    except asyncio.TimeoutError:
        print("タイムアウトしました")
        # チェックポイントから再開
        return await workflow.resume_from_last_checkpoint()

問題: メモリ不足

症状:
長時間実行するとメモリ使用量が増え続ける

解決方法:

# 会話履歴を定期的にクリア
thread = AgentThread()

for i in range(100):
    response = await agent.run(f"タスク{i}", thread=thread)

    # 10回ごとに履歴をクリア
    if i % 10 == 0:
        thread.clear_old_messages(keep_last=5)

次のステップ

おめでとうございます！Microsoft Agent Frameworkの基礎を学びました。

さらに学ぶために

公式ドキュメント
- Microsoft Learn - Agent Framework
- GitHub リポジトリ
サンプルコード
- python/samples/ GitHub リポジトリディレクトリ - dotnet/samples/ GitHub リポジトリディレクトリ
コミュニティ
- GitHub Discussions
- Discord
関連技術
- Azure AI Foundry
- Azure OpenAI Service
- Model Context Protocol (MCP)

実践プロジェクトのアイデア

カスタマーサポートボット
- FAQ検索ツール
- チケット作成ツール
- エスカレーションワークフロー
ドキュメント分析システム
- PDF抽出
- 要約生成
- Q&A機能
データ分析アシスタント
- データベースクエリツール
- グラフ生成
- レポート作成
開発者支援ツール
- コードレビュー
- テスト生成
- ドキュメント生成

最後に

Microsoft Agent Frameworkは、**「研究から本番環境への橋渡し」**を実現するフレームワークです。

プロトタイプ段階: シンプルで始めやすい
開発段階: 豊富なツールとワークフロー
本番段階: エンタープライズグレードの機能

この記事が何かのお役に立てれば幸いです！

参考記事

本記事の作成にあたり、以下の公式ドキュメントと技術記事を参考にしました：

Microsoft Agent Framework - クイックスタート（Python）
- Microsoft公式のクイックスタートガイド。Pythonでの基本的な使い方、環境構築、サンプルコードが網羅されています。
Introducing Microsoft Agent Framework: The Open-Source Engine for Agentic AI Apps
- Microsoft公式ブログによるAgent Frameworkの発表記事。フレームワークの誕生背景、4つの柱、ビジョンが詳しく解説されています。
Microsoft Agent Frameworkの概要
- Microsoft公式による日本語技術記事。フレームワークの全体像、主要機能、実装パターンが分かりやすく説明されています。
Microsoft Agent Frameworkの実践ガイド
- コミュニティによる実践的な技術記事。具体的な実装例、ベストプラクティス、トラブルシューティングが紹介されています。

最新情報は公式ドキュメントを参照してください。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

Microsoft Agent Framework 完全ガイド - これから始める人のために

目次

はじめに

この記事で学べること

Microsoft Agent Frameworkとは

概要

誕生の背景

対応言語

なぜMicrosoft Agent Frameworkなのか

1. オープン標準と相互運用性

MCP (Model Context Protocol)

Agent-to-Agent (A2A) プロトコル

OpenAPI-first 設計

クラウド非依存

2. 研究から本番環境へのパイプライン

先進的な多エージェント編成パターン

豊富なサンプルパターン

プロトタイプから本番へのスムーズな移行

3. コミュニティ主導の拡張性

完全オープンソース

エンタープライズコネクタ

プラガブルメモリモジュール

宣言的エージェント定義

4. 本番運用対応（Production-Ready）

観測可能性（Observability）

セキュリティ

耐久性（Durability）

Human-in-the-loop

環境構築とセットアップ

前提条件

Python環境の場合

.NET環境の場合

Azure環境（推奨）

インストール手順

Python

.NET

認証設定

Azure CLIによる認証（推奨）

APIキーによる認証

動作確認

Python

C#/.NET

基本的な使い方

エージェントの基本構造

シンプルなチャットエージェント

ストリーミング応答

会話履歴の管理

マルチモーダル対応（画像処理）

ツールの使い方

1. カスタム関数ツール

基本的なツール定義

ツールを使うエージェント

Pydanticによる型安全なツール

2. MCPツール

ファイルシステムMCPツール

データベースMCPツール

3. OpenAPIツール

ツール実行の制御

Human-in-the-loop（人間による承認）

ワークフローとオーケストレーション

基本的なワークフローパターン

1. 順次実行（Sequential）

2. 並行実行（Concurrent）

3. グループチャット（GroupChat）

高度なワークフロー機能

コンディショナルエッジ（条件分岐）

チェックポイント（中断と再開）

ストリーミングイベント

開発アプローチ：コードファースト vs 宣言的ワークフロー

フレームワークが推奨する基本アプローチ

宣言的ワークフロー（Declarative Workflows）

宣言的ワークフローの例

.NETでの実行例

宣言的ワークフローの特徴

どちらを選ぶべきか？使い分けガイド

コードファーストを選ぶべきケース

宣言的ワークフローを選ぶべきケース

ハイブリッドアプローチ

公式ガイドラインのまとめ

実際の開発ワークフロー例