Orkestreringar för Microsoft Agent Framework-arbetsflöden – sekventiella

I sekventiell orkestrering organiseras agenter till en pipeline. Varje agent bearbetar uppgiften i sin tur och skickar utdata till nästa agent i sekvensen. Detta är idealiskt för arbetsflöden där varje steg bygger på det föregående, till exempel dokumentgranskning, databearbetningspipelines eller resonemang i flera steg.

Sekventiell orkestrering

Viktigt!

Som standard förbrukar varje agent i sekvensen den föregående agentens fullständiga konversation – både indatameddelandena som angavs till den tidigare agenten och dess svarsmeddelanden. Du kan konfigurera agenter att endast använda den tidigare agentens svarsmeddelanden i stället. Mer information finns i Kontrollera kontext mellan agenter .

Vad du ska lära dig

  • Så här skapar du en sekventiell pipeline med agenter
  • Så här kedjar du agenter där var och en bygger på tidigare utdata
  • Lägga till human-in-the-loop-godkännande för känsliga verktygsanrop
  • Så här blandar du agenter med anpassade utförare för specialiserade uppgifter
  • Så här spårar du konversationsflödet via pipelinen

Definiera dina agenter

I sekventiell orkestrering organiseras agenter i en pipeline där varje agent bearbetar uppgiften i sin tur och skickar utdata till nästa agent i sekvensen.

Konfigurera Azure OpenAI-klienten

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

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

Varning

DefaultAzureCredential är praktiskt för utveckling men kräver noggrant övervägande i produktion. I produktion bör du överväga att använda en specifik autentiseringsuppgift (t.ex. ManagedIdentityCredential) för att undvika problem med svarstid, oavsiktlig avsökning av autentiseringsuppgifter och potentiella säkerhetsrisker från reservmekanismer.

Skapa specialiserade agenter som fungerar i ordning:

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

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

Konfigurera Sekventiell Orkestrering

Skapa arbetsflödet med :AgentWorkflowBuilder

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

Kör sekventiellt arbetsflöde

Kör arbetsflödet och bearbeta händelserna:

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

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

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

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

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

Exempelutdata

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

Sekventiell orkestrering med Human-in-the-Loop

Sekventiella orkestreringar stöder interaktioner mellan människor i loopen via verktygsgodkännande. När agenter använder verktyg omslutna med ApprovalRequiredAIFunction, pausar och sänder ut arbetsflödet en RequestInfoEvent som innehåller en ToolApprovalRequestContent. Externa system (till exempel en mänsklig operatör) kan inspektera verktygets anrop, godkänna eller avvisa det och arbetsflödet återupptas därefter.

Sekventiell orkestrering med Human-in-the-Loop

Tips/Råd

Mer information om begäran och svarsmodellen finns i Human-in-the-Loop.

Definiera agenter med godkännande-krävande verktyg

Skapa agenter där känsliga verktyg omsluts med ApprovalRequiredAIFunction:

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

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

Skapa och köra med godkännandehantering

Skapa det sekventiella arbetsflödet normalt. Godkännandeflödet hanteras via händelseströmmen:

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

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

Anmärkning

AgentWorkflowBuilder.BuildSequential() stöder verktygsgodkännande direkt – ingen ytterligare konfiguration krävs. När en agent anropar ett verktyg omslutet med ApprovalRequiredAIFunctionpausar arbetsflödet automatiskt och genererar en RequestInfoEvent.

Tips/Råd

Ett fullständigt körbart exempel på det här godkännandeflödet finns i GroupChatToolApproval exemplet. Samma RequestInfoEvent hanteringsmönster gäller för andra orkestreringar.

Viktiga begrepp

  • Sekventiell bearbetning: Varje agent bearbetar utdata från den tidigare agenten i ordning
  • AgentWorkflowBuilder.BuildSequential(): Skapar ett pipelinearbetsflöde från en samling agenter
  • ChatClientAgent: Representerar en agent som backas upp av en chattklient med specifika instruktioner
  • InProcessExecution.RunStreamingAsync(): Kör arbetsflödet och returnerar en StreamingRun för händelseströmning i realtid
  • Händelsehantering: Övervaka agentens förlopp genom AgentResponseUpdateEvent och slutförande via WorkflowOutputEvent
  • Verktygsgodkännande: Omge känsliga verktyg med ApprovalRequiredAIFunction som kräver godkännande före utförande
  • RequestInfoEvent: Genereras när ett verktyg kräver godkännande; innehåller information om verktygets ToolApprovalRequestContent anrop

I sekventiell orkestrering bearbetar varje agent uppgiften i sin tur, med utdata som flödar från en till en annan. Börja med att definiera agenter för en process i två steg:

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

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

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

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

Konfigurera Sekventiell Orkestrering

Klassen SequentialBuilder skapar en pipeline där agenter bearbetar uppgifter i ordning. Varje agent ser hela konversationshistoriken och lägger till sitt svar:

from agent_framework.orchestrations import SequentialBuilder

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

Kör sekventiellt arbetsflöde

Kör arbetsflödet och samla in de slutliga utdata. Terminalutdata är en AgentResponse som innehåller den senaste agentens svarsmeddelanden:

from agent_framework import AgentResponse

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

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

Exempelutdata

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

Avancerat: Blanda agenter med anpassade exekutorer

Sekventiell orkestrering stöder att kombinera agenter med skräddarsydda utförare för specialiserad bearbetning. Detta är användbart när du behöver anpassad logik som inte kräver en LLM:

Definiera en anpassad exekverare

Anmärkning

När en anpassad exekverare följer en agent i sekvensen, tar dess hanterare emot en AgentExecutorResponse (eftersom agenter omsluts internt av AgentExecutor). Använd agent_response.full_conversation för att komma åt hela konversationshistoriken. En anpassad exekutor som används som den sista aktören (terminator) måste anropa ctx.yield_output(AgentResponse(...)) så att dess resultat blir arbetsflödets slutgiltiga resultat.

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

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

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

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

Skapa ett blandat sekventiellt arbetsflöde

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

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

Exempelutdata med anpassad exekverare

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

Kontrollera kontexten mellan agenter

Som standard använder varje agent i ett SequentialBuilder arbetsflöde den föregående agentens fullständiga konversation (indata + svarsmeddelanden). Inställningen chain_only_agent_responses=True konfigurerar alla agenter i sekvensen så att de endast använder den tidigare agentens svarsmeddelanden i stället:

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

Detta är användbart för översättningspipelines, progressiv förfining och andra scenarier där varje agent enbart bör fokusera på att transformera den tidigare agentens utdata utan att påverkas av tidigare konversationssvängningar.

För ett komplett exempel, se sequential_chain_only_agent_responses.py i Agent Framework-lagringsplatsen.

Tips/Råd

Mer detaljerad kontroll över kontextflödet – inklusive anpassade filterfunktioner – finns i Kontextlägen i referensen agentexekutor.

Mellanliggande utdata

Som standard visas endast den sista deltagarens utdata som en arbetsflödeshändelse output . Ställ in intermediate_outputs=True för att visa varje deltagares utdata, utöver de slutliga utdata:

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

Du kan hantera dessa händelser i realtid i strömningsläge:

from agent_framework import AgentResponseUpdate

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

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

Sekventiell orkestrering med Human-in-the-Loop

Sekventiella orkestreringar stöder interaktioner mellan människor i loopen på två sätt: verktygsgodkännande för att kontrollera känsliga verktygsanrop och begära information om att pausa efter varje agentsvar för att samla in feedback.

Sekventiell orkestrering med Human-in-the-Loop

Tips/Råd

Mer information om begäran och svarsmodellen finns i Human-in-the-Loop.

Verktygsgodkännande i sekventiella arbetsflöden

Använd @tool(approval_mode="always_require") för att markera verktyg som behöver mänskligt godkännande före körning. Arbetsflödet pausar och genererar en request_info händelse när agenten försöker anropa verktyget.

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


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

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

Bearbeta händelseströmmen och hantera begäranden om godkännande:

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

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

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

Tips/Råd

Ett fullständigt körbart exempel finns i sequential_builder_tool_approval.py. Verktygsgodkännande fungerar med SequentialBuilder utan någon extra builder-konfiguration.

Begär information om agentfeedback

Använd .with_request_info() för att pausa när specifika agenter svarar och tillåta externa indata (till exempel mänsklig granskning) innan nästa agent börjar:

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

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

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

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

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

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

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

Tips/Råd

Se de fullständiga exemplen: godkännande av sekventiella verktyg och information om sekventiella begäranden.

Viktiga begrepp

  • Delad kontext: Som standard använder varje agent den föregående agentens fullständiga konversation, inklusive indata- och svarsmeddelanden
  • Kontextkontroll: Använd chain_only_agent_responses=True för att konfigurera agenter att endast använda den tidigare agentens svarsmeddelanden
  • AgentResponse-utdata: Arbetsflödets terminalutdata innehåller AgentResponse den senaste agentens svar (inte hela konversationen)
  • Orderärenden: Agenter körs strikt i den ordning som anges i participants listan
  • Flexibla deltagare: Du kan blanda agenter och anpassade utförare i valfri ordning
  • Anpassat Terminator-kontrakt: En anpassad exekverare som används som den sista deltagaren måste anropa ctx.yield_output(AgentResponse(...)) för att generera terminal-utdata
  • Mellanliggande utdata: Ställ in intermediate_outputs=True för att visa varje deltagares utdata som en arbetsflödeshändelse output , inte bara den sista deltagarens
  • Godkännande av verktyg: Används @tool(approval_mode="always_require") för känsliga åtgärder som behöver mänsklig granskning
  • Begärandeinformation: Använd .with_request_info(agents=[...]) för att pausa efter specifika agenter för extern feedback

Nästa steg

Samtidig orkestrering

  • Last updated on