![]()
本課程將涵蓋:
完成本課程後,您將能夠:
Microsoft Agent Framework (MAF) 的程式碼範例可在本儲存庫中的 xx-python-agent-framework 和 xx-dotnet-agent-framework 檔案找到。

Microsoft Agent Framework (MAF) 是微軟用來建構 AI 代理的統一框架。它靈活地解決生產及研究環境中廣泛的代理使用案例,包括:
為了在生產中交付 AI 代理,MAF 還包括以下功能:
Microsoft Agent Framework 也聚焦於互通性,具備:
現在讓我們看看這些功能如何應用於 Microsoft Agent Framework 的核心概念。

建立代理
代理的建立是透過定義推理服務(LLM 提供者)、
供 AI 代理遵循的一組指令,及指定的 name:
agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent( instructions="You are good at recommending trips to customers based on their preferences.", name="TripRecommender" )
上述範例使用 Azure OpenAI,但代理亦可透過多種服務建立,包括 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,它提供具有大型上下文窗口(最多 204K 令牌)的 OpenAI 相容 API:
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 = A2AAgent( name=agent_card.name, description=agent_card.description, agent_card=agent_card, url="https://your-a2a-agent-host" )
執行代理
代理透過 .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。
此功能在需要特定模型或工具來完成使用者任務時特別有用。
工具
工具可以在定義代理時設定:
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."
# 直接創建一個ChatAgent時
agent = ChatAgent( chat_client=OpenAIChatClient(), instructions="You are a helpful assistant", tools=[get_attractions]
也可以在執行代理時設定:
result1 = await agent.run( "What's the best place to visit in Seattle?", tools=[get_attractions] # 僅供此運行使用的工具 )
代理執行緒
代理執行緒用於處理多輪對話。執行緒可通過以下方式建立:
get_new_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)
隨後您可以序列化該執行緒以便日後使用:
# 建立一個新執行緒。
thread = agent.get_new_thread()
# 用該執行緒執行代理。
response = await agent.run("Hello, how are you?", thread=thread)
# 將執行緒序列化以作儲存。
serialized_thread = await thread.serialize()
# 從儲存中載入後反序列化執行緒狀態。
resumed_thread = await agent.deserialize_thread(serialized_thread)
代理中介軟體
代理與工具和 LLM 互動來完成使用者任務。在某些情況下,我們想在這些互動間執行或追蹤動作。代理中介軟體使我們能夠透過以下方式做到這點:
功能中介軟體
此中介軟體允許我們在代理與其將呼叫的功能/工具之間執行動作。舉例來說,這可用於在呼叫功能時記錄日誌。
下方程式碼中的 next 定義是否呼叫下一個中介軟體或實際功能。
async def logging_function_middleware(
context: FunctionInvocationContext,
next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
"""Function middleware that logs function execution."""
# 預處理:函數執行前記錄日誌
print(f"[Function] Calling {context.function.name}")
# 繼續至下一個中介軟體或函數執行
await next(context)
# 後處理:函數執行後記錄日誌
print(f"[Function] {context.function.name} completed")
聊天中介軟體
此中介軟體允許我們在代理與 LLM 之間的請求中執行或記錄動作。
其包含重要資訊,如傳送給 AI 服務的 messages。
async def logging_chat_middleware(
context: ChatContext,
next: Callable[[ChatContext], Awaitable[None]],
) -> None:
"""Chat middleware that logs AI interactions."""
# 預處理:在 AI 調用之前記錄日誌
print(f"[Chat] Sending {len(context.messages)} messages to AI")
# 繼續至下一中介軟體或 AI 服務
await next(context)
# 後處理:在 AI 回應後記錄日誌
print("[Chat] AI response received")
代理記憶
如在 Agentic Memory 課程中所述,記憶是使代理能跨不同上下文運作的關鍵元素。MAF 提供多種記憶型態:
記憶體內儲存
這是應用程式執行期間執行緒中儲存的記憶。
# 建立一個新執行緒。
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)
持久訊息
這用於跨不同會話儲存對話歷史。它透過 chat_message_store_factory 定義:
from agent_framework import ChatMessageStore
# 建立一個自訂訊息儲存區
def create_message_store():
return ChatMessageStore()
agent = ChatAgent(
chat_client=OpenAIChatClient(),
instructions="You are a Travel assistant.",
chat_message_store_factory=create_message_store
)
動態記憶
這段記憶會在代理運行前被加入到上下文中。這些記憶可以存儲在外部服務中,例如 mem0:
from agent_framework.mem0 import Mem0Provider
# 使用 Mem0 進行高級記憶體功能
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
)
代理可觀察性
可觀察性對於構建可靠且可維護的代理系統非常重要。MAF 與 OpenTelemetry 集成,提供追蹤和度量,以提升可觀察性。
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"})
MAF 提供了預定義步驟的工作流來完成任務,並在這些步驟中包含 AI 代理作為組件。
工作流由不同的組件構成,以實現更好的控制流程。工作流還支持多代理協同和檢查點以保存工作流狀態。
工作流的核心組件包括:
執行者
執行者接收輸入訊息,執行指定任務,然後產生輸出訊息。這推動工作流向完成更大任務的方向進展。執行者可以是 AI 代理或自定義邏輯。
連接邊
連接邊用於定義工作流中訊息的流向。這些可以是:
直接連接邊 - 執行者之間的一對一簡單連接:
from agent_framework import WorkflowBuilder
builder = WorkflowBuilder()
builder.add_edge(source_executor, target_executor)
builder.set_start_executor(source_executor)
workflow = builder.build()
條件連接邊 - 在特定條件達成後啟動。例如,當酒店房間不可用時,執行者可以建議其他選項。
分支選擇連接邊 - 根據定義條件將訊息路由至不同執行者。例如,若旅客享有優先通道,其任務將通過另一個工作流處理。
擴散連接邊 - 一條訊息發送至多個目標。
匯聚連接邊 - 收集來自不同執行者的多條訊息並發送至單一目標。
事件
為了更好地觀察工作流,MAF 提供了內建的執行事件,包括:
WorkflowStartedEvent - 工作流執行開始WorkflowOutputEvent - 工作流產生輸出WorkflowErrorEvent - 工作流遇到錯誤ExecutorInvokeEvent - 執行者開始處理ExecutorCompleteEvent - 執行者完成處理RequestInfoEvent - 發出請求上述章節介紹了 Microsoft 代理框架的關鍵概念。在構建更複雜的代理時,可考慮以下進階模式:
Microsoft 代理框架是框架互通的——您不必限於使用 MAF 寫的代理。如果您已有使用 LangChain 或 LangGraph 構建的代理,可以將其作為 Microsoft Foundry 托管代理 運行,讓 Foundry 管理運行時、會話、擴展、身份及協議端點,您的代理邏輯依然保持在 LangGraph。
這是通過 langchain_azure_ai.agents.hosting 套件實現,該套件以與 Foundry 托管代理相同的協議暴露已編譯的 LangGraph 圖。
1. 安裝 hosting 額外套件:
pip install -U "langchain-azure-ai[hosting]>=1.2.4" azure-identity
hosting 額外套件安裝 Foundry 協議庫:azure-ai-agentserver-responses(與 OpenAI 兼容的 /responses 端點)和 azure-ai-agentserver-invocations(通用的 /invocations 端點)。
2. 選擇託管協議:
| 協議 | 主機類別 | 端點 | 適用情況 |
|---|---|---|---|
| Responses | ResponsesHostServer |
/responses |
您需要 OpenAI 兼容的聊天、串流、響應歷史和對話線程——這是對話代理的推薦預設。 |
| Invocations | InvocationsHostServer |
/invocations |
您需自訂 JSON 格式、Webhook 風格端點或非對話式處理。 |
因為Responses API 是 Foundry 中代理式開發的主要 API,大部分代理建議先用 ResponsesHostServer。
3. 配置環境變數(先執行 az login 以讓 DefaultAzureCredential 認證):
export FOUNDRY_PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL_NAME="gpt-5-mini"
代理稍後作為 Foundry 托管代理運行時,平台會自動注入 FOUNDRY_PROJECT_ENDPOINT。
4. 透過 Responses 協議暴露 LangGraph 代理:
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)
# ChatOpenAI 係針對 Foundry 計劃嘅 OpenAI 相容(回應)端點。
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()
在本地用 python main.py 運行,之後可向 http://localhost:8088/responses 發送 Responses 請求。
主要行為:
previous_response_id 或 conversation ID 來繼續對話。如果您的圖編譯時使用 LangGraph 檢查點功能,Foundry 將會將對話狀態鍵控至該檢查點(在生產環境使用持久檢查點;本地測試 MemorySaver 即可)。interrupt(),ResponsesHostServer 會將待處理中斷作為 Responses function_call / mcp_approval_request 項目暴露,客戶端以相應的 function_call_output / mcp_approval_response 繼續。azd ext install azure.ai.agents、azd ai agent init -m <manifest>、azd ai agent run(本地需 Docker),然後執行 azd provision 和 azd deploy。托管代理部署需具備Foundry 專案管理員權限。本範例的可執行版本位於 code-samples/14-langchain-hosted-agent.py。完整教學(Invocations 協議、自訂請求結構及故障排除)請參考 將 LangGraph 代理作為 Foundry 托管代理。
Microsoft 代理框架的範例程式碼可在此存儲庫中,位於 xx-python-agent-framework 和 xx-dotnet-agent-framework 文件夾。
加入 Microsoft Foundry Discord 與其他學習者交流,參加辦公時間並獲得 AI 代理問題解答。
免責聲明: 本文件使用 AI 翻譯服務 Co-op Translator 進行翻譯。雖然我們力求準確,但請注意,自動翻譯可能包含錯誤或不準確之處。原始文件的母語版本應被視為權威來源。對於重要資訊,建議尋求專業人工翻譯。我們不對因使用本翻譯而引起的任何誤解或曲解承擔責任。