Orchestrace pracovních postupů rozhraní Microsoft Agent Framework – Předání

Orchestrace předání umožňuje agentům převést řízení mezi sebou na základě kontextu nebo požadavku uživatele. Každý agent může "předat" konverzaci jinému agentu s odpovídajícími odbornými znalostmi a zajistit, aby správný agent zvládl každou část úlohy. To je zvlášť užitečné pro zákaznickou podporu, odborné systémy nebo jakýkoli scénář vyžadující dynamické delegování.

Interně se orchestrace předání implementuje pomocí síťové topologie, kde jsou agenti připojeni přímo bez orchestrátoru. Každý agent se může rozhodnout, kdy konverzaci předat na základě předdefinovaných pravidel nebo obsahu zpráv.

Poznámka:

Orchestrace předání podporuje Agent pouze a agenti musí podporovat spouštění místních nástrojů.

Rozdíly mezi předáním a agenti jako nástroje

I když se agent-jako-nástroj běžně považuje za multiprocesový vzor a může na první pohled vypadat jako předání, mezi těmito dvěma jsou základní rozdíly:

• Tok řízení: Při orchestraci předání se řízení explicitně předává mezi agenty na základě definovaných pravidel. Každý agent se může rozhodnout předat celou úlohu jinému agentu. Pracovní postup nespravuje žádná centrální autorita. Naproti tomu agent jako nástroje zahrnuje primárního agenta, který deleguje dílčí úlohy na jiné agenty a jakmile agent dokončí dílčí úlohu, řízení se vrátí do primárního agenta.
• Vlastnictví úkolu: Při předání převezme agent, který přijímá předání, úplné vlastnictví úkolu. V agent-as-tools primární agent zachovává celkovou odpovědnost za úlohu, zatímco ostatní agenti jsou považováni za nástroje, které pomáhají v konkrétních dílčích úkolech.
• Správa kontextu: Při orchestraci předání je konverzace zcela předána jinému agentovi. Přijímající agent má úplný kontext toho, co se zatím provedlo. V nástrojích jako agent spravuje primární agent celkový kontext a podle potřeby může agentům nástrojů poskytnout jenom relevantní informace.

Co se naučíte

• Jak vytvořit specializované agenty pro různé domény
• Konfigurace pravidel předání mezi agenty
• Vytváření interaktivních pracovních postupů pomocí dynamického směrování agentů
• Řízení víceotáčkových konverzací pomocí přepínání agentů
• Implementace schvalování nástrojů pro citlivé operace (HITL)
• Použití kontrolních bodů pro trvalé pracovní postupy předávání

Při orchestraci předání můžou agenti předávat kontrolu jeden druhému na základě kontextu, což umožňuje dynamické směrování a specializované zpracování dovedností.

Nastavení klienta 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);

Warning

DefaultAzureCredential je vhodný pro vývoj, ale vyžaduje pečlivé zvážení v produkčním prostředí. V produkčním prostředí zvažte použití konkrétních přihlašovacích údajů (např ManagedIdentityCredential. ) k zabránění problémům s latencí, neúmyslnému testování přihlašovacích údajů a potenciálním bezpečnostním rizikům z náhradních mechanismů.

Definování specializovaných agentů

Vytvořte agenty specifické pro doménu a agenta pro třídění pro směrování:

// 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."
    }
});

Poznámka:

Pokud je workflow Handoff uložen s kontrolním bodem a později znovu vytvořen, použijte pro každého zúčastněného agenta stejný jedinečný Id (a pokud je nastaven, i stejný Name). Stabilní ID jsou obzvláště důležitá, pokud jsou agenti pro jednotlivé požadavky vymezeni nebo rekonstruovány, protože směrování předání a kompatibilita kontrolních bodů závisí na vnitřních identitách agenta. Další informace naleznete v tématu Obnovení z kontrolních bodů.

Konfigurace pravidel předání

Definujte, kteří agenti můžou předat ostatním agentům:

// 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();

Spuštění interaktivního pracovního postupu předání

Řízení vícekrokových konverzací pomocí dynamického přepínání agentů.

// 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));
}

Ukázková interakce

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.

Autonomní režim

Ve výchozím nastavení je orchestrace předávání interaktivní: když agent odpoví, aniž by předal řízení, pracovní postup vrátí řízení vám pro další vstup od uživatele. Povolte autonomní režim , aby agent mohl dál pracovat, aniž by čekal na vstup uživatele. Když agent nepředá řízení dál, pracovní postup mu poskytne výzvu k pokračování a znovu ho vyvolá, dokud agent nepředá řízení dál, není splněna podmínka ukončení nebo není dosažen limit počtu tahů na agenta.

Povolíte to voláním WithAutonomousMode() u handoff builderu:

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

Ve výchozím nastavení každý agent provede až 50 autonomních kroků a každé pokračování používá výzvu "User did not respond. Continue assisting autonomously.". Podle potřeby přepište limit otáčení a výzvu:

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

Předejte do parametru agents seznam agentů, aby byl povolen autonomní režim pouze pro podmnožinu účastníků. Agenti, kteří nejsou v seznamu, vždy vrátí řízení po jedné odpovědi:

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

Zkombinujte autonomní režim s podmínkou ukončení a zastavte smyčku, když konverzace dosáhne určitého stavu:

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

Pokročilé: Schválení nástrojů v předávacích pracovních postupech

Agenti v pracovním postupu předání mohou používat nástroje, které před spuštěním vyžadují lidské schválení; užitečné pro citlivé operace, jako je zpracování refundací, provádění nákupů nebo provádění nevratných akcí. Zabalte citlivou funkci pomocí ApprovalRequiredAIFunctionfunkce . Když se jej agent pokusí zavolat, pracovní postup se pozastaví a vydá RequestInfoEvent, který obsahuje ToolApprovalRequestContent.

Definování agentů pomocí nástrojů vyžadujících schválení

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))]);

Zpracování uživatelských vstupů a žádostí o schválení nástrojů

Dvě věci mohou pozastavit workflow předání: agent dokončí svůj tah a čeká na další zprávu od uživatele a volání nástroje vyžadující schválení. Zpracujte obojí v rámci stejné smyčky událostí; odpovězte na schválení RequestInfoEvent pomocí SendResponseAsync a dodejte další zprávu uživatele, jakmile pracovní postup vrátí řízení:

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));
}

Poznámka:

Schválení nástroje funguje s CreateHandoffBuilderWith() hned po instalaci; není potřeba žádná dodatečná konfigurace nástroje pro sestavení. Když agent volá nástroj zabalený ApprovalRequiredAIFunction, pracovní postup se automaticky pozastaví a vygeneruje RequestInfoEvent. Stejný RequestInfoEvent vzor zpracování se používá napříč orchestracemi. Podívejte se na GroupChatToolApproval ukázku kompletního spustitelného projektu.

Definování několika nástrojů pro ukázku

@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."

Nastavení chatovacího klienta

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(),
)

Definování specializovaných agentů

Vytvořte agenty specifické pro doménu s koordinátorem pro směrování:

# 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],
)

Konfigurace pravidel předání

Sestavení pracovního postupu předání pomocí 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()
)

Ve výchozím nastavení si všichni agenti mohou vzájemně předávat úkoly. Pro pokročilejší směrování můžete nakonfigurovat předání řízení:

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()
)

Poznámka:

I s vlastními pravidly předání jsou všichni agenti stále připojeni v topologii sítě. Důvodem je to, že agenti musí vzájemně sdílet kontext, aby zachovali historii konverzací (další podrobnosti najdete v tématu Synchronizace kontextu ). Pravidla předání řídí jenom to, kteří agenti můžou převzít konverzaci v dalším kroku.

Interakce agenta spuštění handoffu

Na rozdíl od jiných orchestrací je předání interaktivní, protože agent se nemusí rozhodnout předat po každém turnu. Pokud se agent nedoručí, je pro pokračování v konverzaci nutný vstup člověka. Pokud chcete tento požadavek obejít, podívejte se na autonomní režim . V jiných orchestracích po reakci agenta ovládací prvek buď přejde do orchestrátoru nebo dalšího agenta.

Když se agent v pracovním postupu předání rozhodne nepředat (předání se aktivuje speciálním voláním nástroje), pracovní postup vygeneruje WorkflowEvent s type="request_info" a datovou část HandoffAgentUserRequest obsahující nejnovější zprávy agenta. Uživatel musí na tento požadavek odpovědět, aby pokračoval v pracovním postupu.

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

Autonomní režim

Orchestrace předání je určená pro interaktivní scénáře, kde je vyžadován vstup člověka, když se agent rozhodne nepředávat. Jako experimentální funkci však můžete povolit "autonomní režim", který pracovnímu postupu umožní pokračovat bez lidského zásahu. Když se v tomto režimu agent rozhodne nepředávat, pracovní postup automaticky odešle výchozí odpověď (např.User did not respond. Continue assisting autonomously.), což mu umožňuje pokračovat v konverzaci.

Tip

Proč je orchestrace Handoffu svou povahou interaktivní? Na rozdíl od jiných orchestrací, kde po reakci agenta existuje pouze jedna možná cesta (např. návrat k orchestrátorovi nebo k dalšímu agentovi), v orchestraci Handoff má agent možnost buď předat úkol jinému agentovi, nebo pokračovat v asistenci uživateli sám. A protože předání se dosahuje voláním nástrojů, pokud agent nevolá nástroj pro předání, ale místo toho vygeneruje odpověď, pracovní postup nebude vědět, co dělat dál, ale delegovat zpět na uživatele pro další vstup. Agenta také není možné vynutit, aby vždy předal úkol, pokud by bylo vyžadováno, aby používal nástroj pro předání, protože by jinak nemohl generovat smysluplné odpovědi.

Autonomní režim je povolen voláním funkce with_autonomous_mode() na HandoffBuilder. Tím se pracovní postup nakonfiguruje tak, aby automaticky reagoval na vstupní požadavky s výchozí zprávou, což agentu umožní pokračovat bez čekání na vstup člověka.

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

Můžete také povolit autonomní režim pouze pro podmnožinu agentů předáním seznamu instancí agentů do 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()
)

Výchozí zprávu odpovědi můžete přizpůsobit.

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()
)

Počet otáček, které agent může autonomně vykonat, než bude vyžadovat lidský vstup, lze přizpůsobit. To může zabránit, aby pracovní postup běžel neomezeně bez zásahu uživatele.

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()
)

Pokročilé: Schválení nástrojů v předávacích pracovních postupech

Pracovní postupy předání můžou zahrnovat agenty používajícími nástroje, které vyžadují schválení člověkem před provedením. To je užitečné pro citlivé operace, jako je zpracování refundací, nákupy nebo provádění nevratných akcí.

Definování nástrojů s požadovaným schválením

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}."

Vytvořte agenty s nástroji vyžadujícími schválení

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],
)

Zpracování žádostí o uživatelský vstup i schválení nástrojů

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!")

Kontrolní body pro trvalé pracovní postupy

V případě dlouhotrvajících pracovních postupů, kdy může dojít ke schválení nástrojů hodiny nebo dny později, použijte kontrolní body:

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!")

Ukázková interakce

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?

Synchronizace kontextu

Agenti v rozhraní Agent Framework spoléhají na relace agenta (AgentSession) ke správě kontextu. V orchestraci předávání agenti nesdílí stejnou instanci relace; zajištění konzistence kontextu je na odpovědnosti účastníků. Aby to bylo dosaženo, účastníci vysílají své odpovědi nebo uživatelské vstupy přijaté všemi ostatními v pracovním postupu kdykoliv vygenerují odpověď, a tím zajišťují, že všichni účastníci mají pro svůj další krok nejnovější kontext.

Poznámka:

Obsah související s nástroji, včetně volání nástrojů pro předání, se nevysílá jiným agentům. Mezi všemi účastníky se synchronizují jenom zprávy uživatelů a agentů.

Tip

Agenti nesdílejí stejnou instanci relace, protože různé typy agentů mohou mít různé implementace AgentSession abstrakce. Sdílení stejné instance relace může vést ke nekonzistencím v tom, jak každý agent zpracovává a udržuje kontext.

Po vysílání odpovědi účastník zkontroluje, jestli má předat konverzaci jinému agentovi. Pokud ano, odešle žádost vybranému agentu, aby převzal konverzaci. V opačném případě požaduje vstup uživatele nebo pokračuje samostatně na základě konfigurace pracovního postupu.

Klíčové koncepty

• Dynamické směrování: Agenti můžou rozhodnout, který agent by měl zpracovat další interakci na základě kontextu.
• AgentWorkflowBuilder.CreateHandoffBuilderWith(): Definuje počátečního agenta, který spouští pracovní postup.
• WithHandoff() a WithHandoffs(): Konfiguruje pravidla předání mezi konkrétními agenty.
• Zachování kontextu: Úplná historie konverzací se udržuje napříč všemi předáními
• Podpora vícenásobného otáčení: Podporuje probíhající konverzace s bezproblémovým přepínáním agentů.
• Specializované odborné znalosti: Každý agent se zaměřuje na svoji doménu a spolupracuje prostřednictvím předávání.
• WithAutonomousMode(): Umožňuje agentům pokračovat bez čekání na vstup uživatele, a to až do limitu počtu tahů na agenta nebo dokud není splněna podmínka ukončení.
• Schválení nástroje (HITL): Obalte citlivé nástroje pomocí ApprovalRequiredAIFunction; pracovní postup se pozastaví a vyšle RequestInfoEvent s ToolApprovalRequestContent, na který odpovíte prostřednictvím SendResponseAsync

• Dynamické směrování: Agenti můžou rozhodnout, který agent by měl zpracovat další interakci na základě kontextu.
• HandoffBuilder: Vytvoří pracovní postupy s automatickou registrací nástroje pro předání
• with_start_agent(): Definuje, který agent nejprve přijímá uživatelský vstup.
• add_handoff(): Konfiguruje konkrétní vztahy předání mezi agenty.
• Výstup: Ve výchozím nastavení je output_from nastavena na všechny účastníky, takže se odpověď každého agenta objeví jako "output" koncová událost (AgentResponse v režimu bez streamování, AgentResponseUpdate v režimu streamování). Pokud chcete místo toho určit konkrétní agenty jako mezilehlé zdroje, předejte intermediate_output_from=[agent_a, agent_b] do HandoffBuilder — tím tyto agenty implicitně vyřadíte z výchozí výstupní sady, takže se jejich odpovědi stanou událostmi typu "intermediate". Nejde o chybu překrytí; degradace proběhne bez upozornění a je záměrná.
• Zachování kontextu: Úplná historie konverzací se udržuje napříč všemi předáními
• Cyklus požadavků/odpovědí: Pracovní postup vyžaduje vstup uživatele, zpracovává odpovědi a pokračuje, dokud není splněna podmínka ukončení.
• Schválení nástroje: Slouží @tool(approval_mode="always_require") k citlivým operacím, které potřebují schválení člověkem.
• Zpracování schválení funkce: Když agent volá nástroj vyžadující schválení, Content vygeneruje se objekt s typem "function_approval_request" ; slouží to_function_approval_response(approved=...) k odpovídání.
• Vytváření kontrolních bodů: Předání checkpoint_storage= do HandoffBuilder pro trvalé pracovní postupy, které lze pozastavit a obnovit napříč restarty procesů
• Specializované odborné znalosti: Každý agent se zaměřuje na svoji doménu a spolupracuje prostřednictvím předávání.

Poznámka:

Podpora této funkce v Go již brzy. Nejnovější stav najdete v úložišti Agent Framework Go .

Exekutor pro předání agenta

Na rozdíl od standardních pracovních postupů, ve kterých jsou agenti zabaleni do exekutoru pro obecné účely, používá orchestrace předání speciální HandoffAgentExecutor. Tento exekutor rozšiřuje exekutor základního agenta o funkce specifické pro předání:

• Vložení nástroje pro předání – automaticky registruje nástroje pro předání na každém agentu na základě nakonfigurovaných pravidel předání, aby agent mohl spustit nástroj pro přenos kontroly.
• Detekce funkce předání – kontroluje odpověď agenta na volání nástroje předání a směruje řízení do cílového agenta.
• Filtrování volání nástrojů – filtruje volání funkcí souvisejících s předáním úkolů a výsledky nástrojů z historie konverzací, než se přesměrují k dalšímu agentovi, aby model nebyl zmaten vnitřní mechanikou pracovních postupů.

Další kroky

Orchestrace skupinového chatu