執行代理程式

基本代理程式抽象概念會公開執行代理程式的各種選項。 呼叫者可選擇提供零、一或多個輸入訊息。 來電者還可以在流媒體和非流媒體之間進行選擇。 讓我們深入研究不同的使用場景。

串流和非串流

Microsoft Agent Framework 支援串流與非串流兩種方式來執行代理程式。

對於非串流,請使用RunAsync

Console.WriteLine(await agent.RunAsync("What is the weather like in Amsterdam?"));

對於串流,請使用RunStreamingAsync

await foreach (var update in agent.RunStreamingAsync("What is the weather like in Amsterdam?"))
{
    Console.Write(update);
}

對於非串流,請使用run

result = await agent.run("What is the weather like in Amsterdam?")
print(result.text)

串流時,請使用 run 帶有 stream=True的 方法 。 此時回傳一個 ResponseStream 可非同步迭代的物件:

async for update in agent.run("What is the weather like in Amsterdam?", stream=True):
    if update.text:
        print(update.text, end="", flush=True)

回應串流

回傳的ResponseStreamrun(..., stream=True)物件支援兩種消費模式:

模式一:非同步迭代 ——即時處理更新以實現即時顯示:

response_stream = agent.run("Tell me a story", stream=True)
async for update in response_stream:
    if update.text:
        print(update.text, end="", flush=True)

模式二:直接定稿 ——跳過迭代,獲得完整回應:

response_stream = agent.run("Tell me a story", stream=True)
final = await response_stream.get_final_response()
print(final.text)

模式三:結合 — 重複以實現即時顯示,然後取得彙總結果:

response_stream = agent.run("Tell me a story", stream=True)

# First, iterate to display streaming output
async for update in response_stream:
    if update.text:
        print(update.text, end="", flush=True)

# Then get the complete response (uses already-collected updates, does not re-iterate)
final = await response_stream.get_final_response()
print(f"\n\nFull response: {final.text}")
print(f"Messages: {len(final.messages)}")

代理程式執行選項

基本代理程式抽象化允許為每個代理程式執行傳遞選項物件,但在抽象層級自訂執行的能力非常有限。 代理可能會有很大差異,因此沒有真正常見的自定義選項。

對於呼叫者知道他們正在使用的代理的類型,可以傳遞特定於類型的選項以允許自訂運行。

例如,這裡的代理程式是 , ChatClientAgent 並且可以傳遞繼 ChatClientAgentRunOptions 承自 AgentRunOptions的物件。 這允許呼叫者提供與任何代理級別選項合併的自定義ChatOptions,然後再傳遞給構建的IChatClientChatClientAgent

var chatOptions = new ChatOptions() { Tools = [AIFunctionFactory.Create(GetWeather)] };
Console.WriteLine(await agent.RunAsync("What is the weather like in Amsterdam?", options: new ChatClientAgentRunOptions(chatOptions)));

Python代理程式支援透過 options 參數自訂每次執行。 選項以 TypedDict 形式傳遞,並可在建構時(透過 default_options)或每次執行(透過 options)設定。 每個提供者都有自己的 TypedDict 類別,提供完整的 IDE 自動補全功能,並為提供者特定設定進行類型檢查。

常用選項包含:

  • max_tokens:要產生的權杖數目上限
  • temperature:控制響應生成的隨機性
  • model:覆寫此特定執行的模型
  • top_p: 核取樣參數
  • response_format: 指定回應格式(例如,結構化輸出)

備註

toolsinstructions參數仍作為直接關鍵字參數,不會透過options字典傳遞。

from agent_framework.openai import OpenAIChatClient, OpenAIChatOptions

# Set default options at construction time
agent = OpenAIChatClient().as_agent(
    instructions="You are a helpful assistant",
    default_options={
        "temperature": 0.7,
        "max_tokens": 500
    }
)

# Run with custom options (overrides defaults)
# OpenAIChatOptions provides IDE autocomplete for all OpenAI-specific settings
options: OpenAIChatOptions = {
    "temperature": 0.3,
    "max_tokens": 150,
    "model": "gpt-4o",
    "presence_penalty": 0.5,
    "frequency_penalty": 0.3
}

result = await agent.run(
    "What is the weather like in Amsterdam?",
    options=options
)

# Streaming with custom options
async for update in agent.run(
    "Tell me a detailed weather forecast",
    stream=True,
    options={"temperature": 0.7, "top_p": 0.9},
    tools=[additional_weather_tool]  # tools is still a keyword argument
):
    if update.text:
        print(update.text, end="", flush=True)

每個提供者都有自己的 TypedDict 類別(例如 OpenAIChatOptions, , AnthropicChatOptions), OllamaChatOptions該類別會揭露該提供者所支援的完整選項集合。

default_options 同時提供 和 每次運行 options 時,每次運行選項優先,並與預設值合併。

回應類型

來自客服人員的串流和非串流回應都包含客服人員產生的所有內容。 內容可能包含與代理人非結果(即用戶問題的答案)有關的資料。 傳回的其他資料範例包括函式工具呼叫、函式工具呼叫的結果、推理文字、狀態更新等等。

由於並非所有傳回的內容都是結果,因此在嘗試將結果與其他內容隔離時,請務必尋找特定的內容類型。

若要從回應擷取文字結果, TextContent 必須彙總所有 ChatMessages 專案中的所有專案。 為了簡化說明,所有 Text 回應類型都有一個屬性,可以聚合所有 TextContent

針對非串流案例,所有專案都會傳回在一個 AgentResponse 物件中。 AgentResponse 允許透過屬性存取 Messages 產生的訊息。

var response = await agent.RunAsync("What is the weather like in Amsterdam?");
Console.WriteLine(response.Text);
Console.WriteLine(response.Messages.Count);

針對串流案例, AgentResponseUpdate 物件會在產生時串流。 每次更新可能包含代理的部分結果,以及其他各種內容項目。 與非串流案例類似,可以使用 Text 屬性來取得更新中包含的結果部分,並透過屬性切 Contents 入詳細數據。

await foreach (var update in agent.RunStreamingAsync("What is the weather like in Amsterdam?"))
{
    Console.WriteLine(update.Text);
    Console.WriteLine(update.Contents.Count);
}

針對非串流案例,所有專案都會傳回在一個 AgentResponse 物件中。 AgentResponse 允許透過屬性存取 messages 產生的訊息。

若要從回應擷取文字結果, TextContent 必須彙總所有 Message 專案中的所有專案。 為了簡化說明,所有 Text 回應類型都有一個屬性,可以聚合所有 TextContent

response = await agent.run("What is the weather like in Amsterdam?")
print(response.text)
print(len(response.messages))

# Access individual messages
for message in response.messages:
    print(f"Role: {message.role}, Text: {message.text}")

在串流的情況下, AgentResponseUpdate 物件會隨著產生而串流,透過 ResponseStream 返回 run(..., stream=True)。 每次更新可能包含代理的部分結果,以及其他各種內容項目。 與非串流案例類似,可以使用 text 屬性來取得更新中包含的結果部分,並透過屬性切 contents 入詳細數據。

response_stream = agent.run("What is the weather like in Amsterdam?", stream=True)
async for update in response_stream:
    print(f"Update text: {update.text}")
    print(f"Content count: {len(update.contents)}")

    # Access individual content items
    for content in update.contents:
        if hasattr(content, 'text'):
            print(f"Content: {content.text}")

# Get the aggregated final response after streaming
final = await response_stream.get_final_response()
print(f"Complete text: {final.text}")

訊息類型

代理程式的輸入和輸出會表示為訊息。 訊息會細分為內容項目。

Microsoft 代理框架使用Microsoft.Extensions.AI抽象所提供的訊息與內容類型。 訊息由類別表示 ChatMessage ,而所有內容類別都繼承自基類 AIContent

存在各種 AIContent 子類別,可用來表示不同類型的內容。 部分作為基礎 Microsoft.Extensions.AI 抽象的一部分提供,但提供者也可以在需要時加入自己的類型。

以下是Microsoft.Extensions.AI中一些常見的類型:

類型 Description
TextContent 文字內容可以是輸入的,例如來自使用者或開發者,也可以是代理人的輸出。 通常包含來自客服專員的文字結果。
DataContent 既可以是輸入,也可以是輸出的二進位內容。 可用於將影像、音訊或視訊資料傳遞至代理程式或從代理程式傳遞(如果支援)。
UriContent 一個通常指向託管內容的網址,例如圖片、音訊或影片。
FunctionCallContent 推論服務叫用函式工具的要求。
FunctionResultContent 函數工具呼叫的結果。

Python 代理框架使用來自 agent_framework 套件的訊息與內容類型。 訊息由 Message 類別表示,所有內容項目則由 Content 屬性 type 區分的類別表示。

所有內容皆由統一 Content 類別表示,並針對每種內容類型提供工廠化方法。 使用該 type 屬性來檢查內容類型。 以下內容類型可供使用:

內容類型 工廠方法 Description
"text" Content.from_text() 輸入與輸出的文字內容。 通常包含來自客服專員的文字結果。
"text_reasoning" Content.from_text_reasoning() 來自支持思考鏈推理的模型推理文本。 可能包含受保護的資料。
"data" Content.from_data()Content.from_uri() 二進位內容以資料 URI 編碼。 用於影像、音訊、影片及文件。
"uri" Content.from_uri() 指向託管內容的網址,例如圖片、音訊或影片。
"error" Content.from_error() 處理失敗時的錯誤資訊。 包含可選錯誤代碼及詳細資訊。
"function_call" Content.from_function_call() AI 服務叫用函式工具的要求。
"function_result" Content.from_function_result() 函數工具呼叫的結果。
"usage" Content.from_usage() 來自 AI 服務的權杖使用情況和計費資訊。
"hosted_file" Content.from_hosted_file() 參考由提供者託管的檔案(例如上傳至 OpenAI)。
"hosted_vector_store" Content.from_hosted_vector_store() 這是對供應商所主機向量儲存的參考。
"code_interpreter_tool_call" Content.from_code_interpreter_tool_call() AI 服務透過程式碼直譯器執行程式碼的請求。
"code_interpreter_tool_result" Content.from_code_interpreter_tool_result() 程式碼直譯器執行的結果。
"image_generation_tool_call" Content.from_image_generation_tool_call() AI 服務要求生成一張圖片。
"image_generation_tool_result" Content.from_image_generation_tool_result() 這是影像生成請求的結果。
"mcp_server_tool_call" Content.from_mcp_server_tool_call() 一個在 MCP 伺服器上呼叫工具的請求。
"mcp_server_tool_result" Content.from_mcp_server_tool_result() 這是 MCP 伺服器工具調用的結果。
"shell_tool_call" Content.from_shell_tool_call() AI 服務要求執行 shell 指令。
"shell_tool_result" Content.from_shell_tool_result() 這是 shell 工具呼叫的總體結果。
"shell_command_output" Content.from_shell_command_output() 單一 shell 指令執行的輸出。
"function_approval_request" Content.from_function_approval_request() 在執行函式呼叫前請求使用者批准。
"function_approval_response" Content.from_function_approval_response() 使用者對函式核准請求的回應。
"oauth_consent_request" Content.from_oauth_consent_request() 請求使用者透過提供的連結完成 OAuth 同意。

以下是如何使用不同的內容類型:

from agent_framework import Message, Content

# Create a text message
text_message = Message(role="user", contents=["Hello!"])

# Create a message with multiple content types
image_data = b"..."  # your image bytes
mixed_message = Message(
    role="user",
    contents=[
        Content.from_text("Analyze this image:"),
        Content.from_data(data=image_data, media_type="image/png"),
    ]
)

# Access content from responses
response = await agent.run("Describe the image")
for message in response.messages:
    for content in message.contents:
        if content.type == "text":
            print(f"Text: {content.text}")
        elif content.type == "data":
            print(f"Data URI: {content.uri}")
        elif content.type == "uri":
            print(f"External URI: {content.uri}")

後續步驟