Оркестрации рабочих процессов Microsoft Agent Framework — последовательные

В последовательной оркестрации агенты организованы в цепочке. Каждый агент обрабатывает задачу в свою очередь, передавая выходные данные следующему агенту в последовательности. Это идеально подходит для рабочих процессов, в которых каждый шаг строится на основе предыдущего, например проверки документов, потоков обработки данных или многоэтапного рассуждения.

Это важно

По умолчанию каждый агент в последовательности использует полный диалог предыдущего агента — как входные сообщения, предоставленные предыдущему агенту, так и его ответные сообщения. Агентов можно настроить так, чтобы они отвечали только на сообщения предыдущего агента. Дополнительные сведения см. в разделе "Управление контекстом между агентами ".

Цели обучения

• Создание последовательного конвейера агентов
• Как связать агентов, где каждый из них строится на основе предыдущих результатов
• Добавление одобрения с участием человека для вызовов конфиденциальных инструментов
• Как комбинировать агентов с пользовательскими исполнителями для специализированных задач
• Отслеживание потока беседы через конвейер

Определение агентов

В последовательной оркестрации агенты организованы в конвейере, где каждый агент обрабатывает задачу в свою очередь, передав выходные данные следующему агенту в последовательности.

Настройка клиента Azure OpenAI

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// 1) Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ??
    throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var client = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetProjectOpenAIClient()
    .GetProjectResponsesClient()
    .AsIChatClient(deploymentName);

Предупреждение

DefaultAzureCredential удобно для разработки, но требует тщательного рассмотрения в рабочей среде. В рабочей среде рекомендуется использовать определенные учетные данные (например, ManagedIdentityCredential), чтобы избежать проблем с задержкой, непреднамеренной проверки данных аутентификации и потенциальных рисков безопасности из-за резервных механизмов.

Создайте специализированные агенты, которые будут работать в последовательности:

// 2) Helper method to create translation agents
static ChatClientAgent GetTranslationAgent(string targetLanguage, IChatClient chatClient) =>
    new(chatClient,
        $"You are a translation assistant who only responds in {targetLanguage}. Respond to any " +
        $"input by outputting the name of the input language and then translating the input to {targetLanguage}.");

// Create translation agents for sequential processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Настройка последовательной оркестрации

Создание рабочего процесса с помощью AgentWorkflowBuilder:

// 3) Build sequential workflow
var workflow = AgentWorkflowBuilder.BuildSequential(translationAgents);

Запуск последовательного рабочего процесса

Выполните рабочий процесс и обработайте события:

// 4) Run the workflow
var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, world!") };

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

string? lastExecutorId = null;
List<ChatMessage> result = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
    {
        if (e.ExecutorId != lastExecutorId)
        {
            lastExecutorId = e.ExecutorId;
            Console.WriteLine();
            Console.Write($"{e.ExecutorId}: ");
        }

        Console.Write(e.Update.Text);
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = outputEvt.As<List<ChatMessage>>()!;
        break;
    }
}

// Display final result
Console.WriteLine();
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

Пример выходных данных

French_Translation: User: Hello, world!
French_Translation: Assistant: English detected. Bonjour, le monde !
Spanish_Translation: Assistant: French detected. ¡Hola, mundo!
English_Translation: Assistant: Spanish detected. Hello, world!

Последовательная оркестрация с человеком в процессе

Последовательные оркестрации поддерживают взаимодействие с человеком в цикле с помощью утверждения средства. Когда агенты используют инструменты, завернутые в ApprovalRequiredAIFunction оболочку, рабочий процесс приостанавливается и излучает объект RequestInfoEvent, содержащий ToolApprovalRequestContent. Внешние системы (например, человеческий оператор) могут проверять вызов инструмента, одобрить или отклонить его, и рабочий процесс возобновляется в соответствии с решением.

Подсказка

Дополнительные сведения о модели запроса и ответа см. в разделе "Человек в цикле".

Определение агентов с помощью инструментов, требующих одобрения

Создайте агентов, где конфиденциальные инструменты обернуты в ApprovalRequiredAIFunction:

ChatClientAgent deployAgent = new(
    client,
    "You are a DevOps engineer. Check staging status first, then deploy to production.",
    "DeployAgent",
    "Handles deployments",
    [
        AIFunctionFactory.Create(CheckStagingStatus),
        new ApprovalRequiredAIFunction(AIFunctionFactory.Create(DeployToProduction))
    ]);

ChatClientAgent verifyAgent = new(
    client,
    "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
    "VerifyAgent",
    "Verifies deployments");

Сборка и запуск с обработкой утверждений/одобрений

Обычно создайте последовательный рабочий процесс. Процесс согласования осуществляется через поток событий.

var workflow = AgentWorkflowBuilder.BuildSequential([deployAgent, verifyAgent]);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is RequestInfoEvent e &&
        e.Request.TryGetDataAs(out ToolApprovalRequestContent? approvalRequest))
    {
        await run.SendResponseAsync(
            e.Request.CreateResponse(approvalRequest.CreateResponse(approved: true)));
    }
}

Замечание

AgentWorkflowBuilder.BuildSequential() поддерживает утверждение инструмента из коробки — дополнительная конфигурация не требуется. Когда агент вызывает инструмент, упакованный в оболочку ApprovalRequiredAIFunction, рабочий процесс автоматически приостанавливается и эмитирует RequestInfoEvent.

Подсказка

Полный исполняемый пример этого процесса утверждения см. в GroupChatToolApproval примере. Тот же RequestInfoEvent шаблон обработки применяется к другим оркестрациям.

Не только одобрение инструмента: интерактивная обратная связь

Подтверждение вызова инструмента позволяет человеку принять или отклонить конкретный вызов инструмента, но последовательная оркестрация не включает встроенный шаг для приостановки и получения произвольной обратной связи от пользователя между агентами, а также не может вернуть управление предыдущему агенту. Когда агенту необходимо интерактивно попросить пользователя получить дополнительные сведения и выполнить итерацию перед продолжением; например, сбор сведений о резервировании перед вызовом средства резервирования; Вместо этого используйте один из следующих подходов:

• Оркестрация передачи управления по умолчанию интерактивна: когда агент отвечает без передачи управления, управление возвращается пользователю в ожидании следующего ввода, что позволяет вести многошаговый диалог в рамках оркестрации. Ограничьте каждого агента одним адресатом передачи, чтобы имитировать последовательный поток, который всё равно приостанавливается для ввода данных пользователем.
• Настраиваемый рабочий процесс, созданный с помощью WorkflowBuilder и RequestPort, позволяет отправлять пользователю типизированный запрос в любой момент и направлять ответ обратно в исполняющий компонент, который можно разместить до или после ваших агентов в конвейере обработки.

Основные понятия

• Последовательная обработка: каждый агент обрабатывает выходные данные предыдущего агента в порядке
• AgentWorkflowBuilder.BuildSequential(): создает рабочий процесс конвейера из коллекции агентов
• ChatClientAgent: представляет собой агента, привязанного к клиенту чата и работающего по конкретным инструкциям
• InProcessExecution.RunStreamingAsync(): выполняет рабочий процесс и возвращает потоковую передачу StreamingRun событий в режиме реального времени.
• Обработка событий: мониторинг хода выполнения агента с помощью AgentResponseUpdateEvent и завершения с помощью WorkflowOutputEvent
• Утверждение инструмента: оберните конфиденциальные инструменты с ApprovalRequiredAIFunction, чтобы требовать утверждение человека перед выполнением
• RequestInfoEvent: создается, когда инструменту требуется утверждение; содержит ToolApprovalRequestContent сведения о вызове средства
• Интерактивный HITL: последовательная оркестрация охватывает подтверждение использования инструмента; для интерактивного обмена сообщениями, при котором агент получает от пользователя дополнительную информацию, используйте оркестрацию передачи управления или пользовательский RequestPort рабочий процесс.

В последовательной оркестрации каждый агент обрабатывает задачу в свою очередь, при этом выходные данные будут поступать от одного до следующего. Начните с определения агентов для двухэтапного процесса:

import os
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

# 1) Create agents using FoundryChatClient
chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

writer = chat_client.as_agent(
    instructions=(
        "You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."
    ),
    name="writer",
)

reviewer = chat_client.as_agent(
    instructions=(
        "You are a thoughtful reviewer. Give brief feedback on the previous assistant message."
    ),
    name="reviewer",
)

Настройка последовательной оркестрации

Класс SequentialBuilder создает конвейер, в котором агенты обрабатывают задачи по порядку. Каждый агент видит полный журнал бесед и добавляет свой ответ:

from agent_framework.orchestrations import SequentialBuilder

# 2) Build sequential workflow: writer -> reviewer
workflow = SequentialBuilder(participants=[writer, reviewer]).build()

Запуск последовательного рабочего процесса

Выполните рабочий процесс и соберите окончательные выходные данные. Выходные данные терминала — это AgentResponse содержащие ответные сообщения последнего агента:

from agent_framework import AgentResponse

# 3) Run and print the last agent's response
events = await workflow.run("Write a tagline for a budget-friendly eBike.")
outputs = events.get_outputs()

if outputs:
    print("===== Final Response =====")
    final: AgentResponse = outputs[0]
    for msg in final.messages:
        name = msg.author_name or "assistant"
        print(f"[{name}]\n{msg.text}")

Пример выходных данных

===== Final Response =====
[reviewer]
This tagline clearly communicates affordability and the benefit of extended travel, making it
appealing to budget-conscious consumers. It has a friendly and motivating tone, though it could
be slightly shorter for more punch. Overall, a strong and effective suggestion!

Продвинутый: Сочетание агентов с пользовательскими исполнителями

Последовательная оркестрация позволяет использовать агентов в сочетании с пользовательскими исполнителями для специализированной обработки. Это полезно, если требуется пользовательская логика, которая не требует LLM:

Определение пользовательского исполнителя

Замечание

Когда пользовательский исполнитель следует за агентом в последовательности, его обработчик получает AgentExecutorResponse (так как агенты внутренне обёрнуты AgentExecutor). Используйте agent_response.full_conversation для доступа к полному журналу бесед. Пользовательский исполнитель, используемый в качестве последнего участника (терминатора), должен вызывать ctx.yield_output(AgentResponse(...)) , чтобы его выходные данные становятся выходными данными терминала рабочего процесса.

from agent_framework import AgentExecutorResponse, AgentResponse, Executor, WorkflowContext, handler
from agent_framework import Message
from typing_extensions import Never

class Summarizer(Executor):
    """Terminator custom executor: consumes full conversation and yields a summary as the workflow's final answer."""

    @handler
    async def summarize(
        self,
        agent_response: AgentExecutorResponse,
        ctx: WorkflowContext[Never, AgentResponse]
    ) -> None:
        if not agent_response.full_conversation:
            await ctx.yield_output(AgentResponse(messages=[Message("assistant", ["No conversation to summarize."])]))
            return

        users = sum(1 for m in agent_response.full_conversation if m.role == "user")
        assistants = sum(1 for m in agent_response.full_conversation if m.role == "assistant")
        summary = Message("assistant", [f"Summary -> users:{users} assistants:{assistants}"])
        await ctx.yield_output(AgentResponse(messages=[summary]))

Создание смешанного последовательного рабочего процесса

# Create a content agent
content = chat_client.as_agent(
    instructions="Produce a concise paragraph answering the user's request.",
    name="content",
)

# Build sequential workflow: content -> summarizer
summarizer = Summarizer(id="summarizer")
workflow = SequentialBuilder(participants=[content, summarizer]).build()

Пример выходных данных с пользовательским экзекутором

===== Final Summary =====
Summary -> users:1 assistants:1

Управление контекстом между агентами

По умолчанию каждый агент в рабочем процессе использует полный SequentialBuilder диалог предыдущего агента (входные и ответные сообщения). Параметр chain_only_agent_responses=True настраивает всех агентов в цепочке на обработку только ответных сообщений от предыдущего агента.

workflow = SequentialBuilder(
    participants=[writer, translator, reviewer],
    chain_only_agent_responses=True,
).build()

Это полезно для конвейеров перевода, прогрессивного уточнения и других сценариев, в которых каждый агент должен сосредоточиться исключительно на преобразовании выходных данных предыдущего агента, не влияя на предыдущие повороты диалога.

Полный пример можно найти в репозитории Agent Framework в sequential_chain_only_agent_responses.py.

Подсказка

Более точное управление потоком контекста, включая пользовательские функции фильтра, см. в разделе "Режимы контекста " в справочнике исполнителя агента.

Промежуточные выходные данные

По умолчанию SequentialBuilder определяет последнего участника в качестве источника выходных данных терминала (output_from). Только результат этого участника появляется как событие "output".

Чтобы также отображались результаты более ранних участников, передайте intermediate_output_from вместе с участниками, которых нужно обозначить как промежуточные источники. Это неявно исключает этих участников из финального набора по умолчанию — они генерируют события "intermediate", а не события "output":

workflow = SequentialBuilder(
    participants=[writer, reviewer, editor],
    intermediate_output_from=[writer, reviewer],
).build()

Вы можете обрабатывать как события "intermediate", так и "output" в реальном времени в потоковом режиме:

from agent_framework import AgentResponseUpdate

# Track the last author to format streaming output.
last_author: str | None = None

async for event in workflow.run("Write a tagline for a budget-friendly eBike.", stream=True):
    if event.type in ("output", "intermediate") and isinstance(event.data, AgentResponseUpdate):
        update = event.data
        author = update.author_name
        if author != last_author:
            if last_author is not None:
                print()  # Newline between different authors
            label = "FINAL" if event.type == "output" else "intermediate"
            print(f"[{label}] {author}: {update.text}", end="", flush=True)
            last_author = author
        else:
            print(update.text, end="", flush=True)

Последовательная оркестрация с человеком в процессе

Последовательные оркестрации поддерживают взаимодействие с участием человека на каждом этапе двумя способами: утверждение для управления вызовами инструментов с повышенной чувствительностью и запрос сведений на паузу после каждого ответа агента для получения отзывов.

Подсказка

Дополнительные сведения о модели запроса и ответа см. в разделе "Человек в цикле".

Утверждение инструмента в последовательных рабочих процессах

Используйте @tool(approval_mode="always_require") для пометки инструментов, которым требуется утверждение человеком перед выполнением. Рабочий процесс приостанавливается и выдает событие, когда агент пытается вызвать инструмент.

@tool(approval_mode="always_require")
def execute_database_query(query: str) -> str:
    return f"Query executed successfully: {query}"


database_agent = Agent(
    client=chat_client,
    name="DatabaseAgent",
    instructions="You are a database assistant.",
    tools=[execute_database_query],
)

workflow = SequentialBuilder(participants=[database_agent]).build()

Обработайте поток событий и обработайте запросы на утверждение:

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info" and event.data.type == "function_approval_request":
            responses[event.request_id] = event.data.to_function_approval_response(approved=True)
    return responses if responses else None

stream = workflow.run("Check the schema and update all pending orders", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Подсказка

Полный пример исполняемого кода см. в разделе sequential_builder_tool_approval.py. Утверждение инструмента работает с SequentialBuilder без какой-либо дополнительной конфигурации билдера.

Запрос сведений об отзыве агента

Используйте .with_request_info() для приостановки после ответа определенных агентов, позволяя внешнему вводу (например, человеческой проверке) до начала следующего агента.

drafter = Agent(
    client=chat_client,
    name="drafter",
    instructions="You are a document drafter. Create a brief draft on the given topic.",
)

editor = Agent(
    client=chat_client,
    name="editor",
    instructions="You are an editor. Review and improve the draft. Incorporate any human feedback.",
)

finalizer = Agent(
    client=chat_client,
    name="finalizer",
    instructions="You are a finalizer. Create a polished final version.",
)

# Enable request info for the editor agent only
workflow = (
    SequentialBuilder(participants=[drafter, editor, finalizer])
    .with_request_info(agents=["editor"])
    .build()
)

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info":
            responses[event.request_id] = AgentRequestInfoResponse.approve()
    return responses if responses else None

stream = workflow.run("Write a brief introduction to artificial intelligence.", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Подсказка

См. полные примеры: последовательное утверждение инструмента и последовательная информация о запросе.

Основные понятия

• Общий контекст. По умолчанию каждый агент использует полный диалог предыдущего агента, включая входные и ответные сообщения.
• Элемент управления контекстом: используется chain_only_agent_responses=True для настройки агентов для использования только сообщений ответа предыдущего агента
• Выходные данные AgentResponse: выходные данные терминала рабочего процесса — это AgentResponse ответ последнего агента (не полный диалог).
• Порядок исполнения имеет значение: агенты выполняются строго в порядке, указанном в списке participants
• Гибкие участники: вы можете смешивать агенты и пользовательских исполнителей в любом порядке
• Пользовательский контракт терминатора: настраиваемый исполнитель, используемый в качестве последнего участника, должен вызывать ctx.yield_output(AgentResponse(...)) для создания выходных данных на терминал.
• Промежуточные результаты: Используйте intermediate_output_from=[...] или intermediate_output_from="all_other", чтобы отображать ход выполнения участника как промежуточные события рабочего процесса, а не только итоговый результат последнего участника
• Утверждение инструмента: использование @tool(approval_mode="always_require") для конфиденциальных операций, требующих проверки человека
• Запрос информации: используйте .with_request_info(agents=[...]) для приостановки после конкретных агентов для внешней обратной связи

Go позволяет создавать последовательные агентные рабочие процессы с помощью workflow/agentworkflow. NewSequentialWorkflowBuilder размещает каждый агент в качестве исполнителя рабочего процесса, подключает их по порядку и выдает окончательный пакет сообщений в виде выходных данных рабочего процесса.

Настройка конфигурации Foundry

endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

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

Предупреждение

azidentity.NewDefaultAzureCredential удобно для разработки, но требует тщательного рассмотрения в рабочей среде. В рабочей среде рекомендуется использовать определенные учетные данные, например azidentity.NewManagedIdentityCredential, чтобы избежать проблем с задержкой, непреднамеренного проверки учетных данных и потенциальных рисков безопасности от резервных механизмов.

Определите агентов Go

Создайте специализированные агенты, которые будут работать в последовательности:

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(
        endpoint,
        token,
        foundryprovider.ModelDeployment(model),
        foundryprovider.AgentConfig{
            Instructions: fmt.Sprintf(
                "You are a translation assistant who only responds in %s. Respond to any input by outputting the name of the input language and then translating the input to %s.",
                language,
                language,
            ),
            Config: agent.Config{Name: language},
        },
    )
}

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

Настройка последовательной оркестрации

wf, err := agentworkflow.NewSequentialWorkflowBuilder(
    frenchAgent,
    spanishAgent,
    englishAgent,
).
    WithName("translation-pipeline").
    Build()
if err != nil {
    return err
}

Запуск последовательного рабочего процесса

Выполните рабочий процесс и обработайте выходные события:

run, err := inproc.Default.RunStreaming(ctx, wf, []*message.Message{message.NewText("Hello, world!")})
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

lastExecutorID := ""
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch e := evt.(type) {
    case workflow.OutputEvent:
        switch value := e.Output.(type) {
        case *agent.ResponseUpdate:
            if e.ExecutorID != lastExecutorID {
                lastExecutorID = e.ExecutorID
                fmt.Printf("\n%s: ", e.ExecutorID)
            }
            fmt.Print(value.String())
        case []*message.Message:
            fmt.Println("\n===== Final Response =====")
            for _, msg := range value {
                fmt.Printf("%s: %s\n", msg.Role, msg.String())
            }
        }
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Пример выходных данных

French: English detected. Bonjour, le monde !
Spanish: French detected. ¡Hola, mundo!
English: Spanish detected. Hello, world!

===== Final Response =====
assistant: Spanish detected. Hello, world!

Последовательная оркестрация с человеком в процессе

Последовательные рабочие процессы могут приостанавливаться в ожидании утверждения инструмента, если размещённый агент использует инструмент, требующий утверждения. Оберните инструмент в tool.ApprovalRequiredFunc, затем отслеживайте workflow.RequestInfoEvent и отвечайте с помощью ToolApprovalResponseContent.

Определение агентов с помощью инструментов, требующих одобрения

deployAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a DevOps engineer. Check staging status first, then deploy to production.",
        Config: agent.Config{
            Name:  "DeployAgent",
            Tools: []tool.Tool{tool.ApprovalRequiredFunc(deployTool)},
        },
    },
)

verifyAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
        Config:      agent.Config{Name: "VerifyAgent"},
    },
)

wf, err := agentworkflow.NewSequentialWorkflowBuilder(deployAgent, verifyAgent).
    WithName("deployment-pipeline").
    Build()
if err != nil {
    return err
}

Сборка и запуск с обработкой утверждений/одобрений

Обработка запросов на утверждение в потоке событий:

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    requestEvent, ok := evt.(workflow.RequestInfoEvent)
    if !ok {
        continue
    }

    requestContent, ok := requestEvent.Request.Data.As(reflect.TypeFor[*message.ToolApprovalRequestContent]())
    if !ok {
        continue
    }

    approvalRequest := requestContent.(*message.ToolApprovalRequestContent)
    response, err := requestEvent.Request.CreateResponse(approvalRequest.CreateResponse(true, "approved"))
    if err != nil {
        return err
    }

    if err := run.SendResponse(ctx, response); err != nil {
        return err
    }
}

Продвинутый: Сочетание агентов с пользовательскими исполнителями

Для смешанных конвейеров размещайте агенты с помощью agentworkflow.New и подключайте их к пользовательским исполнителям с помощью workflow.NewBuilder:

writer := agentworkflow.New(writerAgent, agentworkflow.Config{})

summarizer := workflow.NewExecutor("Summarizer", func(messages []*message.Message) string {
    return summarizeMessages(messages)
}).Bind()

wf, err := workflow.NewBuilder(writer).
    AddEdge(writer, summarizer).
    WithOutputFrom(summarizer).
    Build()
if err != nil {
    return err
}

Управление контекстом между агентами

NewSequentialWorkflowBuilder использует конфигурацию размещённого агента, используемую по умолчанию, при которой каждый последующий агент получает входящие сообщения и сообщения-ответы предыдущего агента. Чтобы создать цепочку только предыдущих ответов агента, задайте:WithChainOnlyAgentResponses(true)

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithChainOnlyAgentResponses(true).
    Build()
if err != nil {
    return err
}

Промежуточные выходные данные

По умолчанию NewSequentialWorkflowBuilder выходные данные каждого участника выдаются в виде выходных данных промежуточного рабочего процесса и выдают окончательный пакет сообщений в качестве выходных данных терминала. Чтобы явно выбрать нужные выходные данные участника, объединить WithIntermediateOutputFrom и WithOutputFrom:

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithIntermediateOutputFrom(frenchAgent, spanishAgent).
    WithOutputFrom(englishAgent).
    Build()
if err != nil {
    return err
}

Используется OutputEvent.IsIntermediate() для отличия выходных данных промежуточного участника от выходных данных терминала.

Основные понятия

• Последовательная обработка: каждый агент или исполнитель обрабатывает выходные данные предыдущего шага в порядке.
• agentworkflow. NewSequentialWorkflowBuilder(): создает рабочий процесс конвейера из коллекции агентов.
• Размещенные агенты: agentworkflow.New предоставляет параметры конфигурации агента для пересылки сообщений, переназначения ролей, событий обновления и перехвата запросов.
• Пользовательские исполнители: workflow.NewBuilderконвейеры, запускаемые вручную, могут сочетать размещённые агенты и детерминированные исполнители.
• Утверждение инструмента: Инструменты, требующие утверждения, приостанавливают рабочий процесс и возвращают значения RequestInfoEvent, содержащие ToolApprovalRequestContent.
• Промежуточные выходные данные: WithIntermediateOutputFrom помечает выбранные выходные данные участников значком workflow.OutputTagIntermediate.

Подсказка

См. пример шаблонов рабочих процессов агентов и пример агентов в рабочих процессах, чтобы ознакомиться с полными исполняемыми последовательными рабочими процессами.

Дальнейшие шаги

Параллельная оркестрация