工作流中的代理

本教程演示如何使用 Agent Framework 将 AI 代理集成到工作流中。你将学习如何创建工作流，以利用专用 AI 代理来执行内容创作、审阅和其他协作任务。

您将构建什么

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

使用 Azure Foundry 代理服务创建智能代理
实现将输入转换为法语的法语翻译代理
实现一个将法语翻译成西班牙语的西班牙语翻译代理程序
实现一个将西班牙语翻译回英语的英语翻译智能体
连接顺序工作流管道中的代理
在代理处理请求时流式传输实时更新
演示 Azure Foundry 智能体的正确资源清理

涵盖的概念

Prerequisites

.NET 8.0 SDK 或更高版本
已配置的 Azure Foundry 项目终结点和模型
已安装并经过身份验证的 Azure CLI（用于 Azure 凭据身份验证）
新的控制台应用程序

步骤 1：安装 NuGet 包

首先，安装 .NET 项目所需的包：

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

步骤 2：设置 Azure Foundry 客户端

使用环境变量和身份验证配置 Azure Foundry 客户端：

using Azure.AI.Projects;
using Azure.AI.Projects.Agents;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure AI Project client
        var endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
            ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-4o-mini";
        var aiProjectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential());

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

实现辅助方法，并根据特定指令创建 Azure Foundry 代理：

    /// <summary>
    /// Creates a translation agent for the specified target language.
    /// </summary>
    /// <param name="targetLanguage">The target language for translation</param>
    /// <param name="aiProjectClient">The AIProjectClient to create the agent</param>
    /// <param name="model">The model to use for the agent</param>
    /// <returns>A FoundryAgent configured for the specified language</returns>
    private static async Task<FoundryAgent> GetTranslationAgentAsync(
        string targetLanguage,
        AIProjectClient aiProjectClient,
        string model)
    {
        string agentName = $"{targetLanguage}Translator";
        var version = await aiProjectClient.AgentAdministrationClient.CreateAgentVersionAsync(
            agentName,
            new ProjectsAgentVersionCreationOptions(
                new DeclarativeAgentDefinition(model)
                {
                    Instructions = $"You are a translation assistant that translates the provided text to {targetLanguage}."
                }));

        return aiProjectClient.AsAIAgent(version);
    }
}

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

使用帮助程序方法创建三个翻译代理：

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

步骤 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.RunStreamingAsync(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())
        {
            if (evt is AgentResponseUpdateEvent executorComplete)
            {
                Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            }
        }

步骤 7：资源清理

使用后正确清理 Azure Foundry 智能体：

        // Cleanup the agents created for the sample.
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(frenchAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(spanishAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(englishAgent.Name);
    }

工作原理

Azure Foundry 客户端设置：使用 AIProjectClient 和 Azure CLI 凭据进行身份验证
代理创建：在 Azure Foundry 上使用特定翻译说明创建版本化代理
顺序处理：法语代理先翻译输入，然后是西班牙语代理，然后是英语代理
轮次令牌模式：代理缓存消息，仅在收到消息时处理 TurnToken
流式更新：AgentResponseUpdateEvent 在智能体生成响应时提供实时令牌更新
资源管理：使用管理 API 正确清理 Azure Foundry 代理

关键概念

Azure Foundry 代理服务：具有高级推理功能的基于云的 AI 代理
AIProjectClient：用于在 Azure Foundry 上创建和管理代理的客户端
WorkflowEvent：输出事件（type="output"）包含代理输出数据（AgentResponseUpdate 用于流式处理、 AgentResponse 非流式处理）
TurnToken：在消息缓存后触发代理处理的信号
顺序工作流：在管道中连接的代理，其中输出从一个流向下一个

完成实现

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

您将构建什么

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

使用 FoundryChatClient 来创建智能代理
实现基于提示创建内容的编写器代理
实现提供内容反馈的审阅者代理
连接顺序工作流管道中的代理
在代理处理请求时流式传输实时更新

涵盖的概念

Prerequisites

Python 3.11 或更高版本
已安装代理框架： pip install agent-framework
使用适当的环境变量配置的 Azure OpenAI 响应
Azure CLI 身份验证： az login

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

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

import asyncio
import os

from agent_framework import AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

步骤 2：创建 Azure OpenAI 响应客户端

创建一个可用于构造多个代理的共享客户端：

async def main() -> None:
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

步骤 3：创建专用代理

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

    # Create a Writer agent that generates content
    writer_agent = 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_agent = 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_agent).add_edge(writer_agent, reviewer_agent).build()

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

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

    last_author: 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):
            update = event.data
            author = update.author_name
            if author != last_author:
                if last_author is not None:
                    print()
                print(f"{author}: {update.text}", end="", flush=True)
                last_author = author
            else:
                print(update.text, end="", flush=True)

步骤 6：完成主函数

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

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

工作原理

客户端设置：使用一个 FoundryChatClient 和 Azure CLI 凭据进行身份验证。
代理创建：从同一客户端配置创建编写器和审阅者代理。
顺序处理：编写器代理首先生成内容，然后将其传递给审阅者代理。
流式更新：带有 AgentResponseUpdate 数据的输出事件 (type="output") 在智能体生成响应时提供实时令牌更新。

关键概念

FoundryChatClient：用于创建具有一致配置的工作流代理的共享客户端。
WorkflowEvent：输出事件（type="output"）包含代理输出数据（AgentResponseUpdate 对于流式处理， AgentResponse 对于非流式处理）。
顺序工作流：在管道中连接的代理，其中输出从一个流向下一个。

完成实现

有关完整的工作实现，请参阅 Agent Framework 存储库中的 azure_ai_agents_streaming.py 。

您将构建什么

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

使用 Microsoft Foundry 模型部署代理作为工作流执行程序
实现一个法语翻译智能体
实现了一个西班牙语翻译代理
实现英文翻译代理
连接顺序工作流管道中的代理
在代理处理请求时流式传输实时更新

涵盖的概念

Prerequisites

Go 1.26 或更高版本
已配置 Microsoft Foundry 项目终结点和模型部署
Azure CLI身份验证或其他Azure凭据源

在新目录中，初始化 Go 模块并安装 Agent Framework：

go mod init example.com/agent-workflow
go get github.com/microsoft/agent-framework-go

设置项目终结点和已部署模型的名称：

$env:FOUNDRY_PROJECT_ENDPOINT = "https://<resource>.services.ai.azure.com/api/projects/<project>"
$env:FOUNDRY_MODEL = "<deployment-name>"

步骤 1：设置 Foundry 配置

以下各节将构建完整实现中的 runExample 函数。从导入和配置开始：

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
if endpoint == "" {
    return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
}
model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
if model == "" {
    return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
}

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

Warning

azidentity.NewDefaultAzureCredential 对于开发来说很方便，但在生产中需要仔细考虑。在生产环境中，请考虑使用特定的凭据，例如 azidentity.NewManagedIdentityCredential，避免延迟问题、意外凭据探测以及回退机制的潜在安全风险。

步骤 2：创建代理工厂函数

使用特定翻译说明创建代理：

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: fmt.Sprintf(
            "Translate the user's text to %s. Return only the translation.",
            language,
        ),
        Config: agent.Config{Name: language + "Agent"},
    })
}

步骤 3：创建专用 Foundry 智能体

frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")

步骤 4：生成工作流

代理可用作工作流执行程序，支持 AI 支持的工作流步骤。

将每个代理绑定为工作流执行程序，然后将执行程序与边缘连接：

cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
french := agentworkflow.New(frenchAgent, cfg)
spanish := agentworkflow.New(spanishAgent, cfg)
english := agentworkflow.New(englishAgent, cfg)

wf, err := workflow.NewBuilder(french).
    AddEdge(french, spanish).
    AddEdge(spanish, english).
    WithIntermediateOutputFrom(french, spanish).
    WithOutputFrom(english).
    Build()
if err != nil {
    return err
}

ForwardIncomingMessages: new(false) 在下游传递每个代理的响应，而无需转发该代理的输入。每个托管的执行程序管理其自己的代理会话。

将法语和西班牙语注册为中间输出源，以便调用方可以观察所有三个代理。英语仍然是终端输出源。输出指定控制调用方看到的内容，而不是控制哪些消息沿这些边流动。

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

运行工作流，并使用 workflow.TurnToken 启用更新事件。处理工作流失败事件以及来自事件迭代器的错误：

run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
if err != nil {
    return err
}
defer func() {
    err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()

emitEvents := true
if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
} else if !accepted {
    return fmt.Errorf("workflow doesn't accept turn tokens")
}

lastExecutorID := ""
for evt, streamErr := range run.WatchStream(ctx) {
    if streamErr != nil {
        return streamErr
    }
    switch event := evt.(type) {
    case workflow.OutputEvent:
        update, ok := event.Output.(*agent.ResponseUpdate)
        if !ok || update.String() == "" {
            continue
        }
        if event.ExecutorID != lastExecutorID {
            if lastExecutorID != "" {
                fmt.Println()
            }
            fmt.Printf("%s: ", event.ExecutorID)
            lastExecutorID = event.ExecutorID
        }
        fmt.Print(update.String())
    case workflow.ErrorEvent:
        return event.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w",
            event.ExecutorID, event.Error)
    }
}
if lastExecutorID != "" {
    fmt.Println()
}

输出包含法语、西班牙语和英语翻译，按执行者分组。完整示例在记录致命错误之前会从 runExample 返回，因此延迟的工作流清理操作仍会执行。

与 C# 示例不同，本 Go 示例不会创建需要在 Foundry 中删除的存储代理版本。

工作原理

客户端设置：使用 Foundry 项目终结点、模型部署和Azure凭据。
代理创建：使用特定于语言的说明创建专用代理。
代理托管：用于 agentworkflow.New 将每个代理绑定为工作流执行程序。
顺序处理：法国执行程序首先运行，然后是西班牙语，然后是英语。
轮次令牌模式：托管代理会缓冲消息，并在收到 workflow.TurnToken 后运行。
流式处理更新：中间和终端输出注册项显示来自三个智能体的更新。
故障处理：主机在报告失败之前检查 ErrorEvent 和 ExecutorFailedEvent，并关闭工作流运行。

关键概念

Foundry 智能体：一种由 Foundry 模型部署提供支持的 agent.Agent。
agentworkflow.New：使代理适配为工作流执行器。
workflow.TurnToken：触发托管代理处理已缓冲消息的信号。
工作流 OutputEvent：传递代理响应更新和最终工作流输出。
顺序工作流：在管道中连接的代理，其中输出从一个流向下一个。

完成实现

请将此示例保存为 main.go，并将其放入你的 Go 模块中：

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func main() {
    if err := runExample(context.Background()); err != nil {
        log.Fatal(err)
    }
}

func runExample(ctx context.Context) (err error) {
    endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
    if endpoint == "" {
        return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
    }
    model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
    if model == "" {
        return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
    }

    token, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        return err
    }

    newTranslationAgent := func(language string) *agent.Agent {
        return foundryprovider.NewAgent(
            endpoint,
            token,
            foundryprovider.ModelDeployment(model),
            foundryprovider.AgentConfig{
                Instructions: fmt.Sprintf(
                    "Translate the user's text to %s. Return only the translation.",
                    language,
                ),
                Config: agent.Config{Name: language + "Agent"},
            },
        )
    }

    cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
    french := agentworkflow.New(newTranslationAgent("French"), cfg)
    spanish := agentworkflow.New(newTranslationAgent("Spanish"), cfg)
    english := agentworkflow.New(newTranslationAgent("English"), cfg)

    wf, err := workflow.NewBuilder(french).
        AddEdge(french, spanish).
        AddEdge(spanish, english).
        WithIntermediateOutputFrom(french, spanish).
        WithOutputFrom(english).
        Build()
    if err != nil {
        return err
    }

    run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    emitEvents := true
    if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
        return err
    } else if !accepted {
        return fmt.Errorf("workflow doesn't accept turn tokens")
    }

    lastExecutorID := ""
    for evt, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            update, ok := event.Output.(*agent.ResponseUpdate)
            if !ok || update.String() == "" {
                continue
            }
            if event.ExecutorID != lastExecutorID {
                if lastExecutorID != "" {
                    fmt.Println()
                }
                fmt.Printf("%s: ", event.ExecutorID)
                lastExecutorID = event.ExecutorID
            }
            fmt.Print(update.String())
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    if lastExecutorID != "" {
        fmt.Println()
    }
    return nil
}

运行示例：

go run .

Warning

Tip

有关另一个可运行的翻译管道，请参阅工作流中的代理示例。

后续步骤

人工干预

反馈

此页面是否有帮助？

上次更新时间为 2026-10-29

工作流中的代理

您将构建什么

涵盖的概念

Prerequisites

步骤 1：安装 NuGet 包

步骤 2：设置 Azure Foundry 客户端

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

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

步骤 5：生成工作流

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

步骤 7：资源清理

工作原理

关键概念

完成实现

您将构建什么

涵盖的概念

Prerequisites

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

步骤 2：创建 Azure OpenAI 响应客户端

步骤 3：创建专用代理

步骤 4：生成工作流

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

步骤 6：完成主函数

工作原理

关键概念

完成实现

您将构建什么

涵盖的概念

Prerequisites

步骤 1：设置 Foundry 配置

步骤 2：创建代理工厂函数

步骤 3：创建专用 Foundry 智能体

步骤 4：生成工作流

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

工作原理

关键概念

完成实现

后续步骤

反馈

其他资源