工作流中的代理

2025-10-23

本教程演示如何使用 Agent Framework 将 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 AzureCliCredential());

步骤 3：创建专用 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);

步骤 4：创建代理工厂方法

实现辅助方法，并根据特定指令创建 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);
    }
}

步骤 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 AgentRunUpdateEvent 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 上创建持久性代理
顺序处理：法语代理先翻译输入，然后是西班牙语代理，然后是英语代理
轮次令牌模式：代理缓存消息，仅在收到消息时处理 TurnToken
流式更新：AgentRunUpdateEvent提供实时令牌更新，随着代理生成响应而进行
资源管理：使用管理 API 正确清理 Azure Foundry 代理

关键概念

Azure Foundry 代理服务：具有高级推理功能的基于云的 AI 代理
PersistentAgentsClient：用于在 Azure Foundry 上创建和管理代理的客户端
AgentRunUpdateEvent：代理执行期间实时流更新
TurnToken：在消息缓存后触发代理处理的信号
顺序工作流：在管道中连接的代理，其中输出从一个流向下一个

完成实现

有关此 Azure Foundry 代理工作流的完整工作实现，请参阅 Agent Framework 存储库中的 FoundryAgent Program.cs 示例。

你将构建的内容

你将创建一个工作流，该工作流：

使用 Azure AI 代理服务创建智能代理
实现基于提示创建内容的编写器代理
实现提供内容反馈的审阅者代理
连接顺序工作流管道中的代理
在代理处理请求时流式传输实时更新
演示 Azure AI 客户端的正确异步上下文管理

先决条件

Python 3.10 或更高版本
已安装代理框架： pip install agent-framework-azure-ai
使用适当的环境变量配置的 Azure AI 代理服务
Azure CLI 身份验证： az login

步骤 1：导入所需的依赖项

首先导入 Azure AI 代理和工作流所需的组件：

import asyncio
from collections.abc import Awaitable, Callable
from contextlib import AsyncExitStack
from typing import Any

from agent_framework import AgentRunUpdateEvent, WorkflowBuilder, WorkflowOutputEvent
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

步骤 2：创建 Azure AI 代理工厂

使用适当的异步上下文处理来创建一个帮助函数，以管理 Azure AI 代理的创建。

async def create_azure_ai_agent() -> tuple[Callable[..., Awaitable[Any]], Callable[[], Awaitable[None]]]:
    """Helper method to create an Azure AI agent factory and a close function.

    This makes sure the async context managers are properly handled.
    """
    stack = AsyncExitStack()
    cred = await stack.enter_async_context(AzureCliCredential())

    client = await stack.enter_async_context(AzureAIAgentClient(async_credential=cred))

    async def agent(**kwargs: Any) -> Any:
        return await stack.enter_async_context(client.create_agent(**kwargs))

    async def close() -> None:
        await stack.aclose()

    return agent, close

步骤 3：创建专用 Azure AI 代理

创建两个专用代理以创建和审阅内容：

async def main() -> None:
    agent, close = await create_azure_ai_agent()
    try:
        # Create a Writer agent that generates content
        writer = await 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 = await 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：生成工作流

使用 Fluent Builder 在顺序工作流中连接代理：

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

步骤 5：使用流式处理执行

使用流式处理运行工作流，观察两个代理的实时更新：

        last_executor_id: str | None = None

        events = workflow.run_stream("Create a slogan for a new electric SUV that is affordable and fun to drive.")
        async for event in events:
            if isinstance(event, AgentRunUpdateEvent):
                # 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 isinstance(event, WorkflowOutputEvent):
                print("\n===== Final output =====")
                print(event.data)
    finally:
        await close()

步骤 6：完成主函数

使用适当的异步执行将所有内容包装在主函数中：

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

工作原理

Azure AI 客户端设置：与 Azure CLI 凭据一起使用 AzureAIAgentClient 进行身份验证
代理工厂模式：创建一个工厂函数，用于管理多个代理的异步上下文生命周期
顺序处理：编写器代理首先生成内容，然后将其传递给审阅者代理
流式更新：AgentRunUpdateEvent提供实时令牌更新，随着代理生成响应而进行
上下文管理：正确清理 Azure AI 资源使用 AsyncExitStack

关键概念

Azure AI 代理服务：具有高级推理功能的基于云的 AI 代理
AgentRunUpdateEvent：代理执行期间实时流更新
AsyncExitStack：多个资源的正确异步上下文管理
代理工厂模式：使用共享客户端配置创建可重用代理
顺序工作流：在管道中连接的代理，其中输出从一个流向下一个

完成实现

有关此 Azure AI 代理工作流的完整工作实现，请参阅 Agent Framework 存储库中的 azure_ai_agents_streaming.py 示例。

后续步骤

了解工作流中的分支

反馈

此页面是否有帮助？

通过

工作流中的代理

你将构建的内容

先决条件

步骤 1：安装 NuGet 包

步骤 2：设置 Azure Foundry 客户端

步骤 3：创建专用 Azure Foundry 代理

步骤 4：创建代理工厂方法

步骤 5：生成工作流

步骤 6：使用流式处理执行

步骤 7：资源清理

工作原理

关键概念

完成实现

你将构建的内容

先决条件

步骤 1：导入所需的依赖项

步骤 2：创建 Azure AI 代理工厂

步骤 3：创建专用 Azure AI 代理

步骤 4：生成工作流

步骤 5：使用流式处理执行

步骤 6：完成主函数

工作原理

关键概念

完成实现

后续步骤

反馈

其他资源