工作流程中的代理

本教學課程示範如何使用代理程式架構將 AI 代理程式整合到工作流程中。您將學習創建工作流程，利用專門的 AI 代理的強大功能來進行內容創建、審查和其他協作任務。

您將構建什麼

您將建立工作流程，以：

使用 Azure Foundry 代理程式服務來建立智慧型代理程式
實作法語翻譯代理，將輸入翻譯成法語
實施將法語翻譯成西班牙語的西班牙語翻譯代理
實施一個英語翻譯程式，將西班牙語翻譯回英語。
在循序工作流程管線中連接代理人
在客服人員處理請求時串流即時更新
示範如何為 Azure Foundry 代理程式正確地釋放資源

涵蓋概念

先決條件

.NET 8.0 SDK 或更新版本
已配置 Azure Foundry 服務端點和部署
已安裝並驗證 Azure CLI （適用於 Azure 認證驗證）
新的主控台應用程式

步驟 1：安裝 NuGet 套件

首先，安裝 .NET 專案所需的套件：

dotnet add package Azure.AI.Agents.Persistent --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.AzureAI --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

步驟 2：設定 Azure Foundry 用戶端

使用環境變數和驗證來設定 Azure Foundry 用戶端：

using System;
using System.Threading.Tasks;
using Azure.AI.Agents.Persistent;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure Foundry client
        var endpoint = Environment.GetEnvironmentVariable("AZURE_FOUNDRY_PROJECT_ENDPOINT") ?? throw new Exception("AZURE_FOUNDRY_PROJECT_ENDPOINT is not set.");
        var model = Environment.GetEnvironmentVariable("AZURE_FOUNDRY_PROJECT_MODEL_ID") ?? "gpt-4o-mini";
        var persistentAgentsClient = new PersistentAgentsClient(endpoint, new DefaultAzureCredential());

警告

DefaultAzureCredential 開發方便，但在生產過程中需謹慎考量。在生產環境中，建議使用特定的憑證（例如 ManagedIdentityCredential），以避免延遲問題、意外的憑證探測，以及備援機制帶來的安全風險。

步驟三：建立代理工廠方法

實作協助程式方法，以使用特定指示建立 Azure Foundry 代理程式：

    /// <summary>
    /// Creates a translation agent for the specified target language.
    /// </summary>
    /// <param name="targetLanguage">The target language for translation</param>
    /// <param name="persistentAgentsClient">The PersistentAgentsClient to create the agent</param>
    /// <param name="model">The model to use for the agent</param>
    /// <returns>A ChatClientAgent configured for the specified language</returns>
    private static async Task<ChatClientAgent> GetTranslationAgentAsync(
        string targetLanguage,
        PersistentAgentsClient persistentAgentsClient,
        string model)
    {
        var agentMetadata = await persistentAgentsClient.Administration.CreateAgentAsync(
            model: model,
            name: $"{targetLanguage} Translator",
            instructions: $"You are a translation assistant that translates the provided text to {targetLanguage}.");

        return await persistentAgentsClient.GetAIAgentAsync(agentMetadata.Value.Id);
    }
}

步驟 4：建立專用的 Azure Foundry 代理

使用協助程式方法建立三個翻譯代理程式：

        // Create agents
        AIAgent frenchAgent = await GetTranslationAgentAsync("French", persistentAgentsClient, model);
        AIAgent spanishAgent = await GetTranslationAgentAsync("Spanish", persistentAgentsClient, model);
        AIAgent englishAgent = await GetTranslationAgentAsync("English", persistentAgentsClient, model);

步驟 5：建立工作流程

使用 WorkflowBuilder 在循序工作流程中連接代理程式：

        // Build the workflow by adding executors and connecting them
        var workflow = new WorkflowBuilder(frenchAgent)
            .AddEdge(frenchAgent, spanishAgent)
            .AddEdge(spanishAgent, englishAgent)
            .Build();

第 6 步：使用串流執行

使用串流執行工作流程，以觀察所有代理程式的即時更新：

        // Execute the workflow
        await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));

        // Must send the turn token to trigger the agents.
        // The agents are wrapped as executors. When they receive messages,
        // they will cache the messages and only start processing when they receive a TurnToken.
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is AgentResponseUpdateEvent executorComplete)
            {
                Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            }
        }

第 7 步：資源清理

使用後正確清除 Azure Foundry 代理程式：

        // Cleanup the agents created for the sample.
        await persistentAgentsClient.Administration.DeleteAgentAsync(frenchAgent.Id);
        await persistentAgentsClient.Administration.DeleteAgentAsync(spanishAgent.Id);
        await persistentAgentsClient.Administration.DeleteAgentAsync(englishAgent.Id);
    }

運作方式

Azure Foundry 用戶端設定： PersistentAgentsClient 搭配 Azure CLI 認證進行驗證
代理程式建立：在 Azure Foundry 上建立持續性代理程式，並提供特定的翻譯指示
順序處理：法語代理先翻譯輸入，然後翻譯西班牙語代理，然後翻譯英語代理
轉權杖模式：代理程式會快取訊息，並且僅在收到權杖後才處理訊息。
流式更新：AgentResponseUpdateEvent 在代理生成響應時提供實時標記更新
資源管理：使用系統管理 API 正確清除 Azure Foundry 代理程式

關鍵概念

Azure Foundry 代理服務：具有進階推理功能的雲端式 AI 代理程式
PersistentAgentsClient：用於在 Azure Foundry 上建立和管理代理程式的用戶端
WorkflowEvent：輸出事件（type="output"）包含代理輸出資料（AgentResponseUpdate 串流、 AgentResponse 非串流）
TurnToken：訊息快取後觸發代理程式處理的訊號
循序工作流程：在管道中連接的代理程式，其中輸出從一個流向下一個

完成實施

如需此 Azure Foundry 代理程式工作流程的完整工作實作，請參閱代理程式架構存放庫中的 FoundryAgent Program.cs 範例。

您將構建什麼

您將建立工作流程，以：

AzureOpenAIResponsesClient用於創造智慧代理
實作根據提示建立內容的 Writer 代理程式
實作檢閱者代理程式，提供內容的意見反應
在循序工作流程管線中連接代理人
在客服人員處理請求時串流即時更新
展示由同一客戶端建立的代理程式的可選共享會話模式

涵蓋概念

先決條件

Python 3.10 或更新版本
已安裝代理程式架構： pip install agent-framework --pre
Azure OpenAI 回應已設定為具備合適的環境變數
Azure CLI 驗證： az login

步驟 1：匯入所需的依賴項

首先從匯入工作流程與 Azure OpenAI 回應代理所需的元件開始：

import asyncio
import os

from agent_framework import AgentResponseUpdate, WorkflowBuilder
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

步驟 2：建立 Azure OpenAI 回應客戶端

建立一個共用客戶端，用來構建多個代理：

async def main() -> None:
    client = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        credential=AzureCliCredential(),
    )

步驟三：建立專業代理人

建立兩個專門的代理程式以進行內容建立和檢閱：

    # Create a Writer agent that generates content
    writer = client.as_agent(
        name="Writer",
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
    )

    # Create a Reviewer agent that provides feedback
    reviewer = client.as_agent(
        name="Reviewer",
        instructions=(
            "You are an excellent content reviewer. "
            "Provide actionable feedback to the writer about the provided content. "
            "Provide the feedback in the most concise manner possible."
        ),
    )

步驟 4：建立工作流程

使用建構工具以順序工作流程連接代理：

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder(start_executor=writer).add_edge(writer, reviewer).build()

可選：跨代理共享一次會話`AzureOpenAIResponsesClient`

預設情況下，每個代理執行者使用自己的會話狀態。對於由相同 AzureOpenAIResponsesClient建立的代理，你可以明確地布線共享會話：

- Basic familiarity with agents, workflows, and executors in the agent framework.
"""


@executor(id="intercept_agent_response")
async def intercept_agent_response(
    agent_response: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorRequest]
) -> None:
    """This executor intercepts the agent response and sends a request without messages.

    This essentially prevents duplication of messages in the shared thread. Without this
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        credential=AzureCliCredential(),
    )

    # set the same context provider (same default source_id) for both agents to share the thread
    writer = client.as_agent(
        instructions=("You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."),
        name="writer",
        context_providers=[InMemoryHistoryProvider()],
    )

    reviewer = client.as_agent(
        instructions=("You are a thoughtful reviewer. Give brief feedback on the previous assistant message."),
        name="reviewer",
        context_providers=[InMemoryHistoryProvider()],
    )

    # Create the shared session
    shared_session = writer.create_session()
    writer_executor = AgentExecutor(writer, session=shared_session)
    reviewer_executor = AgentExecutor(reviewer, session=shared_session)

步驟 5：使用串流執行

使用串流執行工作流程，以觀察兩個代理程式的即時更新：

    last_executor_id: str | None = None

    events = workflow.run("Create a slogan for a new electric SUV that is affordable and fun to drive.", stream=True)
    async for event in events:
        if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
            # Handle streaming updates from agents
            eid = event.executor_id
            if eid != last_executor_id:
                if last_executor_id is not None:
                    print()
                print(f"{eid}:", end=" ", flush=True)
                last_executor_id = eid
            print(event.data, end="", flush=True)
        elif event.type == "output":
            print("\n===== Final output =====")
            print(event.data)

第 6 步：完成主要功能

使用適當的非同步執行將所有內容包裝在 main 函數中：

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

運作方式

用戶端設定：使用帶有 Azure CLI 憑證的裝置 AzureOpenAIResponsesClient 來進行驗證。
代理建立：從同一客戶端設定中建立寫入者與審核者代理。
序列處理：作者代理先產生內容，然後再交給審稿代理。
串流更新：包含AgentResponseUpdate資料的輸出事件type="output"提供代理生成回應時的即時令牌更新。
共享會話（可選）：當兩個代理從同一 AzureOpenAIResponsesClient節點建立時，可以有線連接共享會話。

關鍵概念

AzureOpenAIResponsesClient：用於建立具有一致性設定的工作流程代理的共享客戶端。
工作流程事件：輸出事件（type="output"）包含代理的輸出資料（AgentResponseUpdate 串流、 AgentResponse 非串流）。
序列式工作流程：代理在管線中連接，輸出從一個流向下一個。
共享會話模式：在工作流中選定代理間的共享記憶體/執行緒可選配置。

完成實施

完整的工作實作，請參閱代理框架倉庫中的 azure_ai_agents_streaming.py 與 azure_ai_agents_with_shared_session.py 。

後續步驟

人類參與循環

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-02-20

共用方式為

工作流程中的代理

您將構建什麼

涵蓋概念

先決條件

步驟 1：安裝 NuGet 套件

步驟 2：設定 Azure Foundry 用戶端

步驟三：建立代理工廠方法

步驟 4：建立專用的 Azure Foundry 代理

步驟 5：建立工作流程

第 6 步：使用串流執行

第 7 步：資源清理

運作方式

關鍵概念

完成實施

您將構建什麼

涵蓋概念

先決條件

步驟 1：匯入所需的依賴項

步驟 2：建立 Azure OpenAI 回應客戶端

步驟三：建立專業代理人

步驟 4：建立工作流程

可選：跨代理共享一次會話AzureOpenAIResponsesClient

步驟 5：使用串流執行

第 6 步：完成主要功能

運作方式

關鍵概念

完成實施

後續步驟

意見反應

其他資源

可選：跨代理共享一次會話`AzureOpenAIResponsesClient`