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.

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.

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.

Bortom godkännande av verktyg: Interaktiv feedback

Med godkännande av verktyg kan en människa acceptera eller avvisa ett specifikt verktygsanrop, men en sekventiell orkestrering innehåller inte ett inbyggt steg för att pausa för fri form av användarfeedback mellan agenter, och den kan inte returnera kontroll till en tidigare agent. När en agent behöver be användaren om mer information i flera steg innan den fortsätter, till exempel genom att samla in bokningsuppgifter innan den anropar ett bokningsverktyg, ska du i stället använda någon av följande metoder:

• Orkestrering av överlämning är interaktiv som standard: när en agent svarar utan att lämna över återgår kontrollen till användaren för nästa inmatning, vilket möjliggör flera växlingar fram och tillbaka inom orkestreringen. Begränsa varje agent till ett enda mål för överlämning för att efterlikna ett sekventiellt flöde som fortfarande pausar för användarinmatning.
• Ett anpassat arbetsflöde som byggts med WorkflowBuilder och RequestPort låter dig skicka en typad förfrågan till användaren vid valfri tidpunkt och skicka svaret vidare till en exekverare, som du kan placera före eller efter dina agenter i pipelinen.

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
• Interaktiv HITL: Sekventiell orkestrering omfattar godkännande av verktyg; för interaktiv fram och tillbaka där en agent samlar in mer information från användaren, använder överlämningsorkestrering eller ett anpassat RequestPort arbetsflöde

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 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 SequentialBuilder anger den sista deltagaren som terminalutdatakälla (output_from). Endast deltagarens utdata visas som en "output" händelse.

Om du även vill visa tidigare deltagares utdata skickar intermediate_output_from du med de deltagare som du vill ange som mellanliggande källor. Detta nedgraderar implicit dessa deltagare från standarduppsättningen för final – de avger "intermediate" händelser i stället för "output" händelser:

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

Du kan hantera både "intermediate" och "output" 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 in ("output", "intermediate") 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
            label = "FINAL" if event.type == "output" else "intermediate"
            print(f"[{label}] {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.

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: Använd intermediate_output_from=[...] eller intermediate_output_from="all_other" för att visa deltagarnas förlopp som mellanliggande arbetsflödeshändelser, inte bara den sista deltagarens terminalutdata
• 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

Go kan skapa sekventiella agentarbetsflöden med workflow/agentworkflow. NewSequentialWorkflowBuilder är värd för varje agent som en arbetsflödesexekutor, ansluter dem i ordning och ger den slutliga meddelandebatchen som arbetsflödesutdata.

Konfigurera Foundry

endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

Varning

azidentity.NewDefaultAzureCredential ä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, till exempel azidentity.NewManagedIdentityCredential, för att undvika problem med svarstid, oavsiktlig avsökning av autentiseringsuppgifter och potentiella säkerhetsrisker från reservmekanismer.

Definiera dina Go-agenter

Skapa specialiserade agenter som fungerar i ordning:

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(
        endpoint,
        token,
        foundryprovider.ModelDeployment(model),
        foundryprovider.AgentConfig{
            Instructions: fmt.Sprintf(
                "You are a translation assistant who only responds in %s. Respond to any input by outputting the name of the input language and then translating the input to %s.",
                language,
                language,
            ),
            Config: agent.Config{Name: language},
        },
    )
}

frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")

Konfigurera Sekventiell Orkestrering

wf, err := agentworkflow.NewSequentialWorkflowBuilder(
    frenchAgent,
    spanishAgent,
    englishAgent,
).
    WithName("translation-pipeline").
    Build()
if err != nil {
    return err
}

Kör sekventiellt arbetsflöde

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

run, err := inproc.Default.RunStreaming(ctx, wf, []*message.Message{message.NewText("Hello, world!")})
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

lastExecutorID := ""
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch e := evt.(type) {
    case workflow.OutputEvent:
        switch value := e.Output.(type) {
        case *agent.ResponseUpdate:
            if e.ExecutorID != lastExecutorID {
                lastExecutorID = e.ExecutorID
                fmt.Printf("\n%s: ", e.ExecutorID)
            }
            fmt.Print(value.String())
        case []*message.Message:
            fmt.Println("\n===== Final Response =====")
            for _, msg := range value {
                fmt.Printf("%s: %s\n", msg.Role, msg.String())
            }
        }
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Exempelutdata

French: English detected. Bonjour, le monde !
Spanish: French detected. ¡Hola, mundo!
English: Spanish detected. Hello, world!

===== Final Response =====
assistant: Spanish detected. Hello, world!

Sekventiell orkestrering med Human-in-the-Loop

Sekventiella arbetsflöden kan pausas för godkännande av verktyg när en värdbaserad agent använder ett verktyg som krävs för godkännande. Omslut verktyget med tool.ApprovalRequiredFunc, lyssna sedan efter workflow.RequestInfoEvent och svara med ett ToolApprovalResponseContent.

Definiera agenter med godkännande-krävande verktyg

deployAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a DevOps engineer. Check staging status first, then deploy to production.",
        Config: agent.Config{
            Name:  "DeployAgent",
            Tools: []tool.Tool{tool.ApprovalRequiredFunc(deployTool)},
        },
    },
)

verifyAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
        Config:      agent.Config{Name: "VerifyAgent"},
    },
)

wf, err := agentworkflow.NewSequentialWorkflowBuilder(deployAgent, verifyAgent).
    WithName("deployment-pipeline").
    Build()
if err != nil {
    return err
}

Skapa och köra med godkännandehantering

Hantera begäranden om godkännande i händelseströmmen:

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    requestEvent, ok := evt.(workflow.RequestInfoEvent)
    if !ok {
        continue
    }

    requestContent, ok := requestEvent.Request.Data.As(reflect.TypeFor[*message.ToolApprovalRequestContent]())
    if !ok {
        continue
    }

    approvalRequest := requestContent.(*message.ToolApprovalRequestContent)
    response, err := requestEvent.Request.CreateResponse(approvalRequest.CreateResponse(true, "approved"))
    if err != nil {
        return err
    }

    if err := run.SendResponse(ctx, response); err != nil {
        return err
    }
}

Avancerat: Blanda agenter med anpassade exekutorer

För blandade pipelines kan du köra agenter med agentworkflow.New och ansluta dem till anpassade exekverare med workflow.NewBuilder:

writer := agentworkflow.New(writerAgent, agentworkflow.Config{})

summarizer := workflow.NewExecutor("Summarizer", func(messages []*message.Message) string {
    return summarizeMessages(messages)
}).Bind()

wf, err := workflow.NewBuilder(writer).
    AddEdge(writer, summarizer).
    WithOutputFrom(summarizer).
    Build()
if err != nil {
    return err
}

Kontrollera kontexten mellan agenter

NewSequentialWorkflowBuilder använder standardkonfigurationen för värdbaserad agent, där varje nedströmsagent tar emot den tidigare agentens inkommande meddelanden och svarsmeddelanden. Om du endast vill kedja föregående agentsvar anger du WithChainOnlyAgentResponses(true):

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithChainOnlyAgentResponses(true).
    Build()
if err != nil {
    return err
}

Mellanliggande utdata

Som standard NewSequentialWorkflowBuilder genererar varje deltagares utdata som ett mellanliggande arbetsflöde och genererar den slutliga meddelandebatchen som terminalutdata. Om du explicit vill välja de deltagarutdata du vill ha, kombinera WithIntermediateOutputFrom och WithOutputFrom:

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithIntermediateOutputFrom(frenchAgent, spanishAgent).
    WithOutputFrom(englishAgent).
    Build()
if err != nil {
    return err
}

Använd OutputEvent.IsIntermediate() för att skilja mellanliggande deltagares utdata från terminalutdata.

Viktiga begrepp

• Sekventiell bearbetning: Varje agent eller utförare bearbetar utdata från föregående steg i ordning.
• agentworkflow. NewSequentialWorkflowBuilder(): Skapar ett pipelinearbetsflöde från en samling agenter.
• Värdbaserade agenter: agentworkflow.New exponerar agentkonfigurationsalternativ för vidarebefordran av meddelanden, omtilldelning av roller, uppdateringshändelser och avlyssning av begäranden.
• Anpassade utförare: Manuella workflow.NewBuilder pipelines kan blanda värdbaserade agenter och deterministiska utförare.
• Verktygsgodkännande: Verktyg som krävs för godkännande pausar arbetsflödet och genererar RequestInfoEvent värden som innehåller ToolApprovalRequestContent.
• Mellanliggande utdata: WithIntermediateOutputFrom markerar valda deltagares utdata med workflow.OutputTagIntermediate.

Tips/Råd

Se exemplet på agentarbetsflödesmönster och agenter i arbetsflödesexemplet för fullständiga körbara sekventiella arbetsflöden.

Nästa steg

Samtidig orkestrering