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

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

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

Замечание

Оркестрация передачи поддерживает только Agent, и агенты должны поддерживать выполнение локальных инструментов.

Различия между передачей управления и агентами как инструментами

Хотя агенты в роли инструментов обычно рассматриваются как многоагентный шаблон и на первый взгляд могут показаться схожими с передачей, между ними существуют фундаментальные различия:

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

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

• Создание специализированных агентов для разных доменов
• Как настроить правила передачи между агентами
• Создание интерактивных рабочих процессов с помощью динамической маршрутизации агентов
• Как обрабатывать многоходовые диалоги с переключением агентов
• Внедрение утверждения инструмента для чувствительных операций (HITL)
• Использование контрольных точек для устойчивых рабочих процессов передачи

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

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

using System;
using System.Collections.Generic;
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) Create specialized agents
ChatClientAgent historyTutor = client.AsAIAgent(new ChatClientAgentOptions
{
    Id = "history-tutor",
    Name = "history_tutor",
    Description = "Specialist agent for historical questions",
    ChatOptions = new()
    {
        Instructions = "You provide assistance with historical queries. Explain important events and context clearly. Only respond about history."
    }
});

ChatClientAgent mathTutor = client.AsAIAgent(new ChatClientAgentOptions
{
    Id = "math-tutor",
    Name = "math_tutor",
    Description = "Specialist agent for math questions",
    ChatOptions = new()
    {
        Instructions = "You provide help with math problems. Explain your reasoning at each step and include examples. Only respond about math."
    }
});

ChatClientAgent triageAgent = client.AsAIAgent(new ChatClientAgentOptions
{
    Id = "triage-agent",
    Name = "triage_agent",
    Description = "Routes messages to the appropriate specialist agent",
    ChatOptions = new()
    {
        Instructions = "You determine which agent to use based on the user's homework question. ALWAYS handoff to another agent."
    }
});

Замечание

Если рабочий процесс Handoff сохранён в контрольной точке, а затем перестроен, повторно используйте один и тот же уникальный Id (и, если он задан, тот же Name) для каждого участвующего агента. Стабильные идентификаторы особенно важны, когда агенты создаются в пределах каждого запроса или пересоздаются для каждого запроса, поскольку маршрутизация передачи и совместимость контрольных точек зависят от идентификаторов внутренних агентов. Дополнительные сведения см. в разделе Восстановление из контрольных точек.

Настройка правил передачи

Определите, какие агенты могут передавать задачи другим агентам.

// 3) Build handoff workflow with routing rules
var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor]) // Triage can route to either specialist
    .WithHandoffs([mathTutor, historyTutor], triageAgent) // Both specialists can return to triage
    .Build();

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

Динамическое переключение агента для обработки многократных диалогов:

// 4) Process multi-turn conversations
List<ChatMessage> messages = new();

while (true)
{
    Console.Write("Q: ");
    string userInput = Console.ReadLine()!;
    messages.Add(new(ChatRole.User, userInput));

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

    string? lastExecutorId = null;
    List<ChatMessage> newMessages = new();
    await foreach (WorkflowEvent evt in run.WatchStreamAsync())
    {
        if (evt is AgentResponseUpdateEvent e)
        {
            if (e.ExecutorId != lastExecutorId)
            {
                lastExecutorId = e.ExecutorId;
                Console.WriteLine();
                Console.WriteLine(e.ExecutorId);
            }

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

    // Add new messages to conversation history
    messages.AddRange(newMessages.Skip(messages.Count));
}

Пример взаимодействия

Q: What is the derivative of x^2?
triage_agent: This is a math question. I'll hand this off to the math tutor.
math_tutor: The derivative of x^2 is 2x. Using the power rule, we bring down the exponent (2) and multiply it by the coefficient (1), then reduce the exponent by 1: d/dx(x^2) = 2x^(2-1) = 2x.

Q: Tell me about World War 2
triage_agent: This is a history question. I'll hand this off to the history tutor.
history_tutor: World War 2 was a global conflict from 1939 to 1945. It began when Germany invaded Poland and involved most of the world's nations. Key events included the Holocaust, Pearl Harbor attack, D-Day invasion, and ended with atomic bombs on Japan.

Q: Can you help me with calculus integration?
triage_agent: This is another math question. I'll route this to the math tutor.
math_tutor: I'd be happy to help with calculus integration! Integration is the reverse of differentiation. The basic power rule for integration is: ∫x^n dx = x^(n+1)/(n+1) + C, where C is the constant of integration.

Автономный режим

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

Включите это, вызвав WithAutonomousMode() в конструкторе handoff:

var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithHandoffs([mathTutor, historyTutor], triageAgent)
    .WithAutonomousMode()
    .Build();

По умолчанию каждый агент выполняет до 50 автономных поворотов, и каждое продолжение использует запрос "User did not respond. Continue assisting autonomously.". Переопределите ограничение поворота и запрос по мере необходимости:

var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithAutonomousMode(turnLimit: 10, continuationPrompt: "Continue assisting the user.")
    .Build();

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

var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithAutonomousMode(agents: [triageAgent]) // Only triageAgent runs autonomously
    .Build();

Объедините автономный режим с условием завершения, чтобы остановить цикл, когда беседа достигает определенного состояния:

var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithAutonomousMode(turnLimit: 10)
    .WithTerminationCondition(conversation => conversation.Any(m => m.Text?.Contains("RESOLVED") == true))
    .Build();

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

Агенты в процессе передачи управления могут использовать инструменты, требующие одобрения со стороны человека перед запуском; это полезно для чувствительных операций, таких как обработка возвратов средств, совершение покупок или выполнение необратимых действий. Оберните чувствительную функцию с помощью ApprovalRequiredAIFunction. Когда агент пытается вызвать его, рабочий процесс приостанавливается и выдает RequestInfoEvent, содержащий ToolApprovalRequestContent.

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

ChatClientAgent triageAgent = new(client,
    "You are frontline support. Route the customer to the right specialist.",
    "triage_agent",
    "Routes customers to specialists");

ChatClientAgent refundAgent = new(client,
    "You process refund requests.",
    "refund_agent",
    "Handles refund requests",
    [new ApprovalRequiredAIFunction(AIFunctionFactory.Create(ProcessRefund))]);

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

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

var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [refundAgent])
    .WithHandoffs([refundAgent], triageAgent)
    .Build();

List<ChatMessage> messages = [];

while (true)
{
    Console.Write("You: ");
    string userInput = Console.ReadLine()!;
    if (userInput.Equals("exit", StringComparison.OrdinalIgnoreCase))
    {
        break;
    }

    messages.Add(new(ChatRole.User, userInput));

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

    List<ChatMessage> newMessages = [];
    await foreach (WorkflowEvent evt in run.WatchStreamAsync())
    {
        // An approval-required tool call pauses the workflow and emits a RequestInfoEvent.
        if (evt is RequestInfoEvent requestEvt &&
            requestEvt.Request.TryGetDataAs(out ToolApprovalRequestContent? approval))
        {
            var toolCall = (FunctionCallContent)approval.ToolCall;
            Console.Write($"Approve {toolCall.Name}? (y/n): ");
            bool approved = (Console.ReadLine() ?? "n").Trim().Equals("y", StringComparison.OrdinalIgnoreCase);
            await run.SendResponseAsync(requestEvt.Request.CreateResponse(approval.CreateResponse(approved)));
        }
        else if (evt is AgentResponseUpdateEvent update)
        {
            Console.Write(update.Update.Text);
        }
        else if (evt is WorkflowOutputEvent outputEvt)
        {
            newMessages = outputEvt.As<List<ChatMessage>>()!;
            break;
        }
    }

    // Control returns here after the agent responds without handing off. Merge the new
    // messages into the conversation and loop to collect the next user input.
    messages.AddRange(newMessages.Skip(messages.Count));
}

Замечание

Подтверждение инструмента работает с CreateHandoffBuilderWith() сразу, без дополнительной настройки сборщика. Когда агент вызывает инструмент, упакованный в оболочку ApprovalRequiredAIFunction, рабочий процесс автоматически приостанавливается и эмитирует RequestInfoEvent. Одна и та же RequestInfoEvent схема обработки используется во всех оркестрациях; см. GroupChatToolApproval пример для полного готового к запуску проекта.

Определение нескольких инструментов для демонстрации

@tool
def process_refund(order_number: Annotated[str, "Order number to process refund for"]) -> str:
    """Simulated function to process a refund for a given order number."""
    return f"Refund processed successfully for order {order_number}."

@tool
def check_order_status(order_number: Annotated[str, "Order number to check status for"]) -> str:
    """Simulated function to check the status of a given order number."""
    return f"Order {order_number} is currently being processed and will ship in 2 business days."

@tool
def process_return(order_number: Annotated[str, "Order number to process return for"]) -> str:
    """Simulated function to process a return for a given order number."""
    return f"Return initiated successfully for order {order_number}. You will receive return instructions via email."

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

import os

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

chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

Определите своих специализированных агентов

Создайте агенты, относящиеся к домену, с координатором для маршрутизации:

# Create triage/coordinator agent
triage_agent = chat_client.as_agent(
    instructions=(
        "You are frontline support triage. Route customer issues to the appropriate specialist agents "
        "based on the problem described."
    ),
    description="Triage agent that handles general inquiries.",
    name="triage_agent",
)

# Refund specialist: Handles refund requests
refund_agent = chat_client.as_agent(
    instructions="You process refund requests.",
    description="Agent that handles refund requests.",
    name="refund_agent",
    # In a real application, an agent can have multiple tools; here we keep it simple
    tools=[process_refund],
)

# Order/shipping specialist: Resolves delivery issues
order_agent = chat_client.as_agent(
    instructions="You handle order and shipping inquiries.",
    description="Agent that handles order tracking and shipping issues.",
    name="order_agent",
    # In a real application, an agent can have multiple tools; here we keep it simple
    tools=[check_order_status],
)

# Return specialist: Handles return requests
return_agent = chat_client.as_agent(
    instructions="You manage product return requests.",
    description="Agent that handles return processing.",
    name="return_agent",
    # In a real application, an agent can have multiple tools; here we keep it simple
    tools=[process_return],
)

Настройка правил передачи

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

from agent_framework.orchestrations import HandoffBuilder

# Build the handoff workflow
workflow = (
    HandoffBuilder(
        name="customer_support_handoff",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
        termination_condition=lambda conversation: len(conversation) > 0 and "welcome" in conversation[-1].text.lower(),
    )
    .with_start_agent(triage_agent) # Triage receives initial user input
    .build()
)

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

workflow = (
    HandoffBuilder(
        name="customer_support_handoff",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
        termination_condition=lambda conversation: len(conversation) > 0 and "welcome" in conversation[-1].text.lower(),
    )
    .with_start_agent(triage_agent) # Triage receives initial user input
    # Triage cannot route directly to refund agent
    .add_handoff(triage_agent, [order_agent, return_agent])
    # Only the return agent can handoff to refund agent - users wanting refunds after returns
    .add_handoff(return_agent, [refund_agent])
    # All specialists can handoff back to triage for further routing
    .add_handoff(order_agent, [triage_agent])
    .add_handoff(return_agent, [triage_agent])
    .add_handoff(refund_agent, [triage_agent])
    .build()
)

Замечание

Даже с пользовательскими правилами передачи все агенты по-прежнему подключены в топологии сетки. Это связано с тем, что агенты должны совместно использовать контекст друг с другом для поддержания журнала бесед (дополнительные сведения см. в разделе "Синхронизация контекста "). Правила передачи управляют только тем, какие агенты могут продолжить беседу в дальнейшем.

Запуск взаимодействия агента передачи

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

Когда агент в рабочем процессе передачи решает не передавать (передача активируется специальным вызовом средства), рабочий процесс генерирует WorkflowEvent с type="request_info" и HandoffAgentUserRequest, содержащие самые последние сообщения агента. Пользователь должен ответить на этот запрос, чтобы продолжить рабочий процесс.

from agent_framework import WorkflowEvent
from agent_framework.orchestrations import HandoffAgentUserRequest

# Start workflow with initial user message
events = [event async for event in workflow.run("I need help with my order", stream=True)]

# Process events and collect pending input requests
pending_requests = []
for event in events:
    if event.type == "request_info" and isinstance(event.data, HandoffAgentUserRequest):
        pending_requests.append(event)
        request_data = event.data
        print(f"Agent {event.executor_id} is awaiting your input")
        # The request contains the most recent messages generated by the
        # agent requesting input
        for msg in request_data.agent_response.messages[-3:]:
            print(f"{msg.author_name}: {msg.text}")

# Interactive loop: respond to requests
while pending_requests:
    user_input = input("You: ")

    # Send responses to all pending requests
    responses = {req.request_id: HandoffAgentUserRequest.create_response(user_input) for req in pending_requests}
    # You can also send a `HandoffAgentUserRequest.terminate()` to end the workflow early
    events = [event async for event in workflow.run(responses=responses, stream=True)]

    # Process new events
    pending_requests = []
    for event in events:
        # Check for new input requests

Автономный режим

Оркестрация Handoff предназначена для интерактивных сценариев, где требуется участие человека в случае, если агент принимает решение не передавать управление. Однако в качестве экспериментальной функции можно включить "автономный режим", чтобы рабочий процесс продолжался без вмешательства человека. В этом режиме, когда агент решает не передавать управление, рабочий процесс автоматически отправляет агенту ответ по умолчанию (например, User did not respond. Continue assisting autonomously.), что позволяет ему продолжить беседу.

Tip

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

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

workflow = (
    HandoffBuilder(
        name="autonomous_customer_support",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent)
    .with_autonomous_mode()
    .build()
)

Вы также можете включить автономный режим только в подмножестве агентов, передав список экземпляров агента с помощью with_autonomous_mode().

workflow = (
    HandoffBuilder(
        name="partially_autonomous_support",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent)
    .with_autonomous_mode(agents=[triage_agent])  # Only triage_agent runs autonomously
    .build()
)

Вы можете настроить ответное сообщение по умолчанию.

workflow = (
    HandoffBuilder(
        name="custom_autonomous_support",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent)
    .with_autonomous_mode(
        agents=[triage_agent],
        prompts={triage_agent.name: "Continue with your best judgment as the user is unavailable."},
    )
    .build()
)

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

workflow = (
    HandoffBuilder(
        name="limited_autonomous_support",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent)
    .with_autonomous_mode(
        agents=[triage_agent],
        turn_limits={triage_agent.name: 3},  # Max 3 autonomous turns
    )
    .build()
)

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

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

Определение инструментов с обязательным утверждением

from typing import Annotated
from agent_framework import tool

@tool(approval_mode="always_require")
def process_refund(order_number: Annotated[str, "Order number to process refund for"]) -> str:
    """Simulated function to process a refund for a given order number."""
    return f"Refund processed successfully for order {order_number}."

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

import os
from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import AzureCliCredential

chat_client = OpenAIChatCompletionClient(
    model=os.environ["AZURE_OPENAI_CHAT_COMPLETION_MODEL"],
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
    credential=AzureCliCredential(),
)

triage_agent = chat_client.as_agent(
    instructions=(
        "You are frontline support triage. Route customer issues to the appropriate specialist agents "
        "based on the problem described."
    ),
    description="Triage agent that handles general inquiries.",
    name="triage_agent",
)

refund_agent = chat_client.as_agent(
    instructions="You process refund requests.",
    description="Agent that handles refund requests.",
    name="refund_agent",
    tools=[process_refund],
)

order_agent = chat_client.as_agent(
    instructions="You handle order and shipping inquiries.",
    description="Agent that handles order tracking and shipping issues.",
    name="order_agent",
    tools=[check_order_status],
)

Обработка пользовательского ввода и запросов на утверждение инструментов

from agent_framework import (
    Content,
    WorkflowEvent,
)
from agent_framework.orchestrations import HandoffBuilder, HandoffAgentUserRequest

workflow = (
    HandoffBuilder(
        name="support_with_approvals",
        participants=[triage_agent, refund_agent, order_agent],
    )
    .with_start_agent(triage_agent)
    .build()
)

pending_requests: list[WorkflowEvent] = []

# Start workflow
async for event in workflow.run("My order 12345 arrived damaged. I need a refund.", stream=True):
    if event.type == "request_info":
        pending_requests.append(event)

# Process pending requests - could be user input OR tool approval
while pending_requests:
    responses: dict[str, object] = {}

    for request in pending_requests:
        if isinstance(request.data, HandoffAgentUserRequest):
            # Agent needs user input
            print(f"Agent {request.executor_id} asks:")
            for msg in request.data.agent_response.messages[-2:]:
                print(f"  {msg.author_name}: {msg.text}")

            user_input = input("You: ")
            responses[request.request_id] = HandoffAgentUserRequest.create_response(user_input)

        elif isinstance(request.data, Content) and request.data.type == "function_approval_request":
            # Agent wants to call a tool that requires approval
            func_call = request.data.function_call
            args = func_call.parse_arguments() or {}

            print(f"\nTool approval requested: {func_call.name}")
            print(f"Arguments: {args}")

            approval = input("Approve? (y/n): ").strip().lower() == "y"
            responses[request.request_id] = request.data.to_function_approval_response(approved=approval)

    # Send all responses and collect new requests
    pending_requests = []
    async for event in workflow.run(responses=responses, stream=True):
        if event.type == "request_info":
            pending_requests.append(event)
        elif event.type == "output":
            print("\nWorkflow completed!")

Использование контрольных точек для долговечных рабочих процессов

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

from agent_framework import FileCheckpointStorage

storage = FileCheckpointStorage(storage_path="./checkpoints")

workflow = (
    HandoffBuilder(
        name="durable_support",
        participants=[triage_agent, refund_agent, order_agent],
        checkpoint_storage=storage,
    )
    .with_start_agent(triage_agent)
    .build()
)

# Initial run - workflow pauses when approval is needed
pending_requests = []
async for event in workflow.run("I need a refund for order 12345", stream=True):
    if event.type == "request_info":
        pending_requests.append(event)

# Process can exit here - checkpoint is saved automatically

# Later: Resume from checkpoint and provide approval
checkpoints = await storage.list_checkpoints(workflow_name="durable_support")
latest = sorted(checkpoints, key=lambda c: c.timestamp, reverse=True)[0]

# Step 1: Restore checkpoint to reload pending requests
restored_requests = []
async for event in workflow.run(checkpoint_id=latest.checkpoint_id, stream=True):
    if event.type == "request_info":
        restored_requests.append(event)

# Step 2: Send responses
responses = {}
for req in restored_requests:
    if isinstance(req.data, Content) and req.data.type == "function_approval_request":
        responses[req.request_id] = req.data.to_function_approval_response(approved=True)
    elif isinstance(req.data, HandoffAgentUserRequest):
        responses[req.request_id] = HandoffAgentUserRequest.create_response("Yes, please process the refund.")

async for event in workflow.run(responses=responses, stream=True):
    if event.type == "output":
        print("Refund workflow completed!")

Пример взаимодействия

User: I need help with my order

triage_agent: I'd be happy to help you with your order. Could you please provide more details about the issue?

User: My order 1234 arrived damaged

triage_agent: I'm sorry to hear that your order arrived damaged. I will connect you with a specialist.

support_agent: I'm sorry about the damaged order. To assist you better, could you please:
- Describe the damage
- Would you prefer a replacement or refund?

User: I'd like a refund

triage_agent: I'll connect you with the refund specialist.

refund_agent: I'll process your refund for order 1234. Here's what will happen next:
1. Verification of the damaged items
2. Refund request submission
3. Return instructions if needed
4. Refund processing within 5-10 business days

Could you provide photos of the damage to expedite the process?

Синхронизация контекста

Агенты в Agent Framework используют сеансы агента (AgentSession) для управления контекстом. В сценарии оркестрации передачи агенты не используют один и тот же экземпляр сеанса; участники несут ответственность за обеспечение согласованности контекста. Для этого участники транслируют свои ответы или полученные входные данные пользователей всем участникам рабочего процесса при создании ответа, обеспечивая актуальность контекста для всех участников к их следующему шагу.

Замечание

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

Tip

Агенты не используют один экземпляр сеанса, так как разные типы агентовAgentSession могут иметь разные реализации абстракции. Совместное использование одного экземпляра сеанса может привести к непоследовательностям в том, как каждый агент обрабатывает и поддерживает контекст.

После трансляции ответа участник затем проверяет, нужно ли передать беседу другому агенту. Если да, он отправляет запрос выбранному агенту, чтобы взять на себя беседу. В противном случае он запрашивает входные данные пользователя или продолжает автономно на основе конфигурации рабочего процесса.

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

• Динамическая маршрутизация: агенты могут решить, какой агент должен обрабатывать следующее взаимодействие на основе контекста
• AgentWorkflowBuilder.CreateHandoffBuilderWith(): определяет начальный агент, запускающий рабочий процесс.
• WithHandoff() и WithHandoffs(): настраивает правила передачи между определенными агентами
• Сохранение контекста: полная история беседы сохраняется во всех переходах.
• Поддержка с несколькими шагами: поддерживает текущие беседы с простым переключением агента
• Специализированный опыт: каждый агент фокусируется на своем домене в процессе совместной работы и передачи полномочий
• WithAutonomousMode(): позволяет агентам продолжать работу без ожидания ввода данных пользователем до достижения лимита числа шагов для каждого агента или до выполнения условия завершения.
• Подтверждение инструмента (HITL): оберните чувствительные инструменты с помощью ApprovalRequiredAIFunction; рабочий процесс приостанавливается и создаёт RequestInfoEvent с ToolApprovalRequestContent, на который вы отвечаете через SendResponseAsync

• Динамическая маршрутизация: агенты могут решить, какой агент должен обрабатывать следующее взаимодействие на основе контекста
• HandoffBuilder: создание рабочих процессов с автоматической регистрацией средства передачи
• with_start_agent(): определяет, какой агент получает входные данные пользователя в первую очередь
• add_handoff(): конфигурирует конкретные связи передачи между агентами
• Вывод: По умолчанию для output_from задано значение все участники, поэтому ответ каждого агента отображается как событие "output" (terminal) (AgentResponse в непотоковом режиме, AgentResponseUpdate в потоковом режиме). Чтобы вместо этого назначить определённых агентов промежуточными источниками, передайте intermediate_output_from=[agent_a, agent_b] в HandoffBuilder — это неявно исключит этих агентов из набора выходных данных по умолчанию, и их ответы станут событиями "intermediate". Ошибка перекрытия отсутствует; понижение безмолвно и намеренно.
• Сохранение контекста: полная история беседы сохраняется во всех переходах.
• Цикл запроса и ответа: рабочий процесс запрашивает входные данные пользователя, обрабатывает ответы и продолжается до тех пор, пока не будет выполнено условие завершения.
• Утверждение инструмента: использование @tool(approval_mode="always_require") для конфиденциальных операций, которым требуется утверждение человека
• Обработка утверждений функций: когда агент запускает инструмент, требующий утверждения, объект с типом Content создается; с помощью "function_approval_request" можно ответить
• Контрольная точка: передача checkpoint_storage=HandoffBuilder для устойчивых рабочих процессов, которые могут приостановить и возобновить работу в процессе перезапуска.
• Специализированный опыт: каждый агент фокусируется на своем домене в процессе совместной работы и передачи полномочий

Замечание

Поддержка Go для этой функции скоро появится. Сведения о последнем состоянии см. в репозитории Agent Framework Go .

Исполнитель агента передачи функций

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

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

Дальнейшие действия

Оркестрация группового чата