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.OpenAI;
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 AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

Výstraha

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 = new(client,
    "You provide assistance with historical queries. Explain important events and context clearly. Only respond about history.",
    "history_tutor",
    "Specialist agent for historical questions");

ChatClientAgent mathTutor = new(client,
    "You provide help with math problems. Explain your reasoning at each step and include examples. Only respond about math.",
    "math_tutor",
    "Specialist agent for math questions");

ChatClientAgent triageAgent = new(client,
    "You determine which agent to use based on the user's homework question. ALWAYS handoff to another agent.",
    "triage_agent",
    "Routes messages to the appropriate specialist agent");

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.StartHandoffWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor]) // Triage can route to either specialist
    .WithHandoff(mathTutor, triageAgent)                  // Math tutor can return to triage
    .WithHandoff(historyTutor, triageAgent)               // History tutor 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
    StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages);
    await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

    List<ChatMessage> newMessages = new();
    await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
    {
        if (evt is AgentResponseUpdateEvent e)
        {
            Console.WriteLine($"{e.ExecutorId}: {e.Data}");
        }
        else if (evt is WorkflowOutputEvent outputEvt)
        {
            newMessages = (List<ChatMessage>)outputEvt.Data!;
            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.

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

from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Initialize the Azure OpenAI chat client
chat_client = AzureOpenAIChatClient(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_stream("I need help with my order")]

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

    # 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.

Návod

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í

from agent_framework import Agent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

client = AzureOpenAIChatClient(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 (
    FunctionApprovalRequestContent,
    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_stream("My order 12345 arrived damaged. I need a refund."):
    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, FunctionApprovalRequestContent):
            # 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.create_response(approved=approval)

    # Send all responses and collect new requests
    pending_requests = []
    async for event in workflow.run(responses=responses):
        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_stream("I need a refund for order 12345"):
    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()
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_stream(checkpoint_id=latest.checkpoint_id):
    if event.type == "request_info":
        restored_requests.append(event)

# Step 2: Send responses
responses = {}
for req in restored_requests:
    if isinstance(req.data, FunctionApprovalRequestContent):
        responses[req.request_id] = req.data.create_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):
    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ů.

Návod

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.StartHandoffWith(): 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í.

• 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.
• 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.
• FunctionApprovalRequestContent: Vygeneruje se, když agent volá nástroj vyžadující schválení; použít create_response(approved=...) k reagování
• Vytváření kontrolních bodů: Použijte with_checkpointing() pro přetrvávající pracovní postupy, které lze pozastavit a obnovit při restartování procesu.
• Specializované odborné znalosti: Každý agent se zaměřuje na svoji doménu a spolupracuje prostřednictvím předávání.

Další kroky

Orchestrace skupinového chatu