Orchestrazioni dei flussi di lavoro di Microsoft Agent Framework - Handoff

Configurare il client OpenAI di Azure

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 AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

Definire gli agenti specializzati

Creare agenti specifici del dominio e un agente di valutazione per il routing:

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

Configurare le regole di trasferimento

Definire quali agenti possono trasferire ad altri agenti.

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

Eseguire il flusso di lavoro interattivo di consegna

Gestire le conversazioni a più turni con il cambio di agente dinamico:

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

Interazione di esempio

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.

Definire alcuni strumenti per la dimostrazione

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

Configurare il client chat

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

# Initialize the Azure OpenAI chat client
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

Definire gli agenti specializzati

Creare agenti specifici del dominio con un coordinatore per il routing:

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

Configurare le regole di trasferimento

Creare il flusso di lavoro di consegna usando HandoffBuilder:

from agent_framework import HandoffBuilder

# Build the handoff workflow
workflow = (
    HandoffBuilder(
        name="customer_support_handoff",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent) # Triage receives initial user input
    .with_termination_condition(
        # Custom termination: Check if one of the agents has provided a closing message.
        # This looks for the last message containing "welcome", which indicates the
        # conversation has concluded naturally.
        lambda conversation: len(conversation) > 0 and "welcome" in conversation[-1].text.lower()
    )
    .build()
)

Per impostazione predefinita, tutti gli agenti possono passare l'uno all'altro. Per il routing più avanzato, è possibile configurare gli handoff:

workflow = (
    HandoffBuilder(
        name="customer_support_handoff",
        participants=[triage_agent, refund_agent, order_agent, return_agent],
    )
    .with_start_agent(triage_agent) # Triage receives initial user input
    .with_termination_condition(
        # Custom termination: Check if one of the agents has provided a closing message.
        # This looks for the last message containing "welcome", which indicates the
        # conversation has concluded naturally.
        lambda conversation: len(conversation) > 0 and "welcome" in conversation[-1].text.lower()
    )
    # 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()
)

Eseguire l'interazione agente Handoff

A differenza di altre orchestrazioni, il handoff è interattivo perché un agente potrebbe non decidere di consegnare dopo ogni turno. Se un agente non viene consegnato, l'input umano è necessario per continuare la conversazione. Per ignorare questo requisito, vedere Modalità autonoma . In altre orchestrazioni, dopo che un agente risponde, il controllo passa all'agente di orchestrazione o all'agente successivo.

Quando un agente in un flusso di lavoro di handoff decide di non effettuare l'handoff (che viene attivato da una chiamata speciale dello strumento), il flusso di lavoro emette un RequestInfoEvent con un HandoffAgentUserRequest payload contenente i messaggi più recenti dell'agente. L'utente deve rispondere a questa richiesta per continuare il flusso di lavoro.

from agent_framework import RequestInfoEvent, HandoffAgentUserRequest, WorkflowOutputEvent

# 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 isinstance(event, RequestInfoEvent) and isinstance(event.data, HandoffAgentUserRequest):
        pending_requests.append(event)
        request_data = event.data
        print(f"Agent {event.source_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.send_responses_streaming(responses)]

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

Modalità autonoma

L'orchestrazione Handoff è progettata per scenari interattivi in cui è richiesto l'intervento umano nel caso un agente decida di non effettuare il passaggio. Tuttavia, come funzionalità sperimentale, è possibile abilitare la "modalità autonoma" per consentire al flusso di lavoro di continuare senza l'intervento umano. In questa modalità, quando un agente decide di non consegnare, il flusso di lavoro invia automaticamente una risposta predefinita (ad esempioUser did not respond. Continue assisting autonomously.) all'agente, consentendogli di continuare la conversazione.

La modalità autonoma è abilitata chiamando with_autonomous_mode() su HandoffBuilder. In questo modo il flusso di lavoro viene configurato per rispondere automaticamente alle richieste di input con un messaggio predefinito, consentendo all'agente di continuare senza attendere l'input umano.

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

È anche possibile abilitare la modalità autonoma solo in un subset di agenti passando un elenco di istanze dell'agente a 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()
)

È possibile personalizzare il messaggio di risposta predefinito.

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

È possibile personalizzare il numero di turni che un agente può operare autonomamente prima di richiedere l'input umano. Ciò può impedire l'esecuzione illimitata del flusso di lavoro senza coinvolgimento dell'utente.

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

Avanzato: Approvazione degli strumenti nei flussi di lavoro di trasferimento

I flussi di lavoro di trasferimento possono includere agenti con strumenti che richiedono l'approvazione umana prima dell'esecuzione. Ciò è utile per operazioni sensibili come l'elaborazione dei rimborsi, l'esecuzione di acquisti o l'esecuzione di azioni irreversibili.

Definire gli strumenti con approvazione richiesta

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

Creare agenti con strumenti con approvazione richiesta

from agent_framework import ChatAgent
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],
)

Gestire sia l'input dell'utente che le richieste di approvazione degli strumenti

from agent_framework import (
    FunctionApprovalRequestContent,
    HandoffBuilder,
    HandoffAgentUserRequest,
    RequestInfoEvent,
    WorkflowOutputEvent,
)

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

pending_requests: list[RequestInfoEvent] = []

# Start workflow
async for event in workflow.run_stream("My order 12345 arrived damaged. I need a refund."):
    if isinstance(event, RequestInfoEvent):
        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.source_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.send_responses_streaming(responses):
        if isinstance(event, RequestInfoEvent):
            pending_requests.append(event)
        elif isinstance(event, WorkflowOutputEvent):
            print("\nWorkflow completed!")

Con checkpoint per flussi di lavoro durevoli

Per i flussi di lavoro di lunga durata in cui le approvazioni degli strumenti possono avvenire ore o giorni più tardi, usare il checkpointing:

from agent_framework import FileCheckpointStorage

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

workflow = (
    HandoffBuilder(
        name="durable_support",
        participants=[triage_agent, refund_agent, order_agent],
    )
    .with_start_agent(triage_agent)
    .with_checkpointing(storage)
    .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 isinstance(event, RequestInfoEvent):
        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 isinstance(event, RequestInfoEvent):
        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.send_responses_streaming(responses):
    if isinstance(event, WorkflowOutputEvent):
        print("Refund workflow completed!")

Interazione di esempio

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?

• Routing dinamico: gli agenti possono decidere quale agente deve gestire l'interazione successiva in base al contesto
• AgentWorkflowBuilder.StartHandoffWith(): definisce l'agente iniziale che avvia il flusso di lavoro
• WithHandoff() e WithHandoffs():Configura le regole di handoff tra agenti specifici
• Conservazione del contesto: la cronologia completa delle conversazioni viene mantenuta in tutti gli handoff
• Supporto a più turni: supporta le conversazioni in corso con il cambio di agente facile
• Esperienza specializzata: ogni agente si concentra sul proprio dominio mentre collabora tramite trasferimenti

• Routing dinamico: gli agenti possono decidere quale agente deve gestire l'interazione successiva in base al contesto
• HandoffBuilder: crea flussi di lavoro con registrazione automatica dello strumento handoff
• with_start_agent(): definisce l'agente che riceve prima l'input dell'utente
• add_handoff(): configura relazioni di handoff specifiche tra gli agenti
• Conservazione del contesto: la cronologia completa delle conversazioni viene mantenuta in tutti gli handoff
• Ciclo di richiesta/risposta: il flusso di lavoro richiede l'input dell'utente, elabora le risposte e continua fino a quando non viene soddisfatta la condizione di terminazione
• Approvazione degli strumenti: usare @tool(approval_mode="always_require") per operazioni sensibili che richiedono l'approvazione umana
• FunctionApprovalRequestContent: generato quando un agente chiama uno strumento che richiede l'approvazione; usare create_response(approved=...) per rispondere
• Checkpointing: usare with_checkpointing() per flussi di lavoro durevoli che possono essere sospesi e ripresi tra i riavvii del processo
• Esperienza specializzata: ogni agente si concentra sul proprio dominio mentre collabora tramite trasferimenti