探索 Microsoft Agent Framework (MAF)

本課概覽

本課將涵蓋：

• 認識 MAF：核心功能與價值
• MAF 關鍵概念：Agent、Tool、Thread、Middleware、Memory、Observability
• 進階模式：Workflow、Checkpointing、Human-in-the-loop
• 在 Microsoft Foundry 上託管 LangChain / LangGraph Agent

學習目標

完成本課後，你將能夠：

• 使用 MAF 建立可上線的 AI Agent
• 將 MAF 核心功能應用到你的 Agentic 使用場景
• 使用進階模式，包括 Workflow、Middleware 和 Observability

認識 Microsoft Agent Framework

Microsoft Agent Framework (MAF) 是微軟統一的 AI Agent 開發框架。它提供足夠的靈活性來應對各種 Agentic 場景，包括：

五種編排模式

上線必備四大支柱

互通性設計

• 雲端中立：Agent 可在容器、本地部署、跨多雲環境中執行
• 服務提供者中立：支援 Azure OpenAI、OpenAI 等多種 SDK
• 開放標準整合：支援 A2A (Agent-to-Agent) 和 MCP (Model Context Protocol) 協定
• 外掛與連接器：可連接 Microsoft Fabric、SharePoint、Pinecone、Qdrant 等資料與記憶體服務

MAF 關鍵概念

1. Agent 建立

建立 Agent 需要定義推論服務（LLM 提供者）、一組指令以及一個 name：

使用 Azure OpenAI

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions="You are good at recommending trips to customers based on their preferences.",
    name="TripRecommender"
)

使用 Microsoft Foundry Agent Service

AzureAIAgentClient(async_credential=credential).create_agent(
    name="HelperAgent",
    instructions="You are a helpful assistant."
) as agent

使用 OpenAI Responses / ChatCompletion API

agent = OpenAIResponsesClient().create_agent(
    name="WeatherBot",
    instructions="You are a helpful weather assistant.",
)

agent = OpenAIChatClient().create_agent(
    name="HelpfulAssistant",
    instructions="You are a helpful assistant.",
)

使用 MiniMax（OpenAI 相容 API，支援 204K tokens 上下文）

agent = OpenAIChatClient(
    base_url="https://api.minimax.io/v1",
    api_key=os.environ["MINIMAX_API_KEY"],
    model_id="MiniMax-M3"
).create_agent(
    name="HelpfulAssistant",
    instructions="You are a helpful assistant.",
)

使用 A2A 協定的遠端 Agent

agent = A2AAgent(
    name=agent_card.name,
    description=agent_card.description,
    agent_card=agent_card,
    url="https://your-a2a-agent-host"
)

2. Agent 執行

使用 .run（非串流）或 .run_stream（串流）方法：

# 非串流
result = await agent.run("What are good places to visit in Amsterdam?")
print(result.text)

# 串流
async for update in agent.run_stream("What are the good places to visit in Amsterdam?"):
    if update.text:
        print(update.text, end="", flush=True)

每次執行可自訂 max_tokens、可用 tools 甚至更換 model。這在特定任務需要特定模型或工具時非常有用。

3. 工具 (Tools)

工具可以在定義 Agent 時指定，也可以在執行時才提供：

def get_attractions(
    location: Annotated[str, Field(description="The location to get the top tourist attractions for")],
) -> str:
    """Get the top tourist attractions for a given location."""
    return f"The top attractions for {location} are."

# 建立 Agent 時定義工具
agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a helpful assistant",
    tools=[get_attractions]
)

# 也可以在執行時才提供工具
result = await agent.run(
    "What's the best place to visit in Seattle?",
    tools=[get_attractions]  # 僅本次執行可用
)

4. Agent 執行緒 (Threads)

Thread 用來處理多輪對話。可以持久化儲存以便後續使用：

# 建立新執行緒
thread = agent.get_new_thread()

# 使用執行緒執行
response = await agent.run(
    "Hello, I am here to help you book travel. Where would you like to go?",
    thread=thread
)

# 序列化以儲存
serialized_thread = await thread.serialize()

# 從儲存中恢復
resumed_thread = await agent.deserialize_thread(serialized_thread)

5. Middleware（中介層）

MAF 提供兩種 Middleware，讓你在 Agent 與工具/LLM 之間插入自訂邏輯：

Function Middleware

在 Agent 呼叫函式/工具前後執行：

async def logging_function_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    """記錄函式執行的 Function Middleware"""
    print(f"[Function] Calling {context.function.name}")
    await next(context)
    print(f"[Function] {context.function.name} completed")

Chat Middleware

在 Agent 與 LLM 之間攔截請求：

async def logging_chat_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    """記錄 AI 互動的 Chat Middleware"""
    print(f"[Chat] Sending {len(context.messages)} messages to AI")
    await next(context)
    print("[Chat] AI response received")

6. Agent 記憶體 (Memory)

# Dynamic Memory with Mem0
from agent_framework.mem0 import Mem0Provider

memory_provider = Mem0Provider(
    api_key="your-mem0-api-key",
    user_id="user_123",
    application_id="my_app"
)

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a helpful assistant with memory.",
    context_providers=memory_provider
)

7. 可觀測性 (Observability)

MAF 整合 OpenTelemetry 提供 Tracing 和 Metrics：

from agent_framework.observability import get_tracer, get_meter

tracer = get_tracer()
meter = get_meter()

with tracer.start_as_current_span("my_custom_span"):
    # 你的自訂邏輯
    pass

counter = meter.create_counter("my_custom_counter")
counter.add(1, {"key": "value"})

Workflow（工作流程）

MAF 的 Workflow 是預先定義的步驟來完成任務，Agent 作為這些步驟中的元件。Workflow 支援多 Agent 編排和檢查點 (Checkpointing) 來保存狀態。

核心元件

Executor（執行器）

接收輸入訊息、執行指派任務、產生輸出訊息，推動工作流程。可以是 AI Agent 或自訂邏輯。

Edge（邊）

定義訊息在 Workflow 中的流向：

事件 (Events)

提供對 Workflow 執行的觀測：

• WorkflowStartedEvent — 工作流程開始
• WorkflowOutputEvent — 工作流程產出結果
• WorkflowErrorEvent — 工作流程遇到錯誤
• ExecutorInvokeEvent — Executor 開始處理
• ExecutorCompleteEvent — Executor 完成處理
• RequestInfoEvent — 發出請求

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder()
builder.add_edge(source_executor, target_executor)
builder.set_start_executor(source_executor)
workflow = builder.build()

進階 MAF 模式

• Middleware 組合：鏈結多個 Middleware（日誌、認證、速率限制），使用 Function 和 Chat Middleware 精細控制 Agent 行為
• Workflow Checkpointing：使用 Workflow 事件和序列化來保存和恢復長時間執行的 Agent 流程
• 動態工具選擇：結合 RAG 與工具描述，每次查詢只呈現相關工具
• 多 Agent Handoff：使用 Workflow Edge 和條件路由來編排專業 Agent 之間的交接

在 Microsoft Foundry 上託管 LangChain / LangGraph Agent

步驟

1. 安裝託管套件

pip install -U "langchain-azure-ai[hosting]>=1.2.4" azure-identity

2. 選擇託管協定

3. 設定環境變數

export FOUNDRY_PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL_NAME="gpt-5-mini"

4. 透過 Responses 協定暴露 LangGraph Agent

import os
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_azure_ai.agents.hosting import ResponsesHostServer

_AZURE_AI_SCOPE = "https://ai.azure.com/.default"

def build_chat_model() -> ChatOpenAI:
    project_endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"].rstrip("/")
    deployment = os.environ.get("FOUNDRY_MODEL_NAME", "gpt-5-mini")
    credential = DefaultAzureCredential()
    project = AIProjectClient(endpoint=project_endpoint, credential=credential)
    openai_client = project.get_openai_client()
    token_provider = get_bearer_token_provider(credential, _AZURE_AI_SCOPE)
    return ChatOpenAI(
        model=deployment,
        base_url=str(openai_client.base_url),
        api_key=token_provider,
    )

def main() -> None:
    graph = create_agent(build_chat_model(), tools=[])
    port = int(os.environ.get("PORT", "8088"))
    ResponsesHostServer(graph).run(port=port)

if __name__ == "__main__":
    main()

關鍵行為

• 對話持續性：客戶端透過 previous_response_id 或 conversation ID 繼續對話。若圖使用 LangGraph Checkpointer 編譯，Foundry 會將對話狀態對應到檢查點。
• Human-in-the-loop：若圖使用 LangGraph interrupt()，ResponsesHostServer 會將其以 function_call / mcp_approval_request 形式呈現，客戶端可用對應的回應來恢復執行。
• 部署到 Foundry：使用 Azure Developer CLI — azd ai agent init → azd ai agent run（本地，需 Docker）→ azd provision → azd deploy。需要 Foundry Project Manager 角色。

本課重點回顧

1. MAF 是微軟統一的 Agent 框架，支援五種編排模式（循序、並行、群組對話、交接、磁性）
2. 核心概念：Agent → Tool → Thread → Middleware → Memory → Observability，六大模組環環相扣
3. Workflow 提供 Executor + Edge + Event 的結構化編排，支援 Checkpoint 和人機迴圈
4. 框架互通：可以在 Foundry 上託管 LangChain / LangGraph Agent，不限定使用 MAF 原生 Agent
5. 上線就緒：內建 OpenTelemetry 觀測、RBAC 安全、持久性執行緒和 Human-in-the-loop 控制

原始資源

• 原始 README（英文）
• Microsoft Agent Framework 官方文件
• Microsoft Foundry Discord 社群
• Host LangGraph agents as Foundry hosted agents