Orchestrace pracovních postupů architektury Microsoft Agent – sekvenční

V sekvenční orchestraci jsou agenti uspořádáni do řetězce. Každý agent zpracovává úlohu postupně a předává výstup dalšímu agentu v pořadí. To je ideální pro pracovní postupy, ve kterých každý krok vychází z předchozího kroku, jako je kontrola dokumentů, kanály zpracování dat nebo vícefázové odůvodnění.

Důležité

Ve výchozím nastavení každý agent v sekvenci využívá úplnou konverzaci předchozího agenta – jak vstupní zprávy zadané předchozímu agentovi, tak i zprávy s odpovědí. Agenty můžete nakonfigurovat tak, aby místo toho spotřebovávají pouze zprávy odpovědí předchozího agenta. Podrobnosti najdete v tématu Řízení kontextu mezi agenty .

Co se naučíte

• Jak vytvořit sekvenční potrubí agentů
• Jak propojit agenty, kde každý navazuje na předchozí výstup
• Jak přidat schválení s lidským zapojením pro citlivá volání nástrojů
• Jak kombinovat agenty s vlastními exekutory pro specializované úlohy
• Jak sledovat tok konverzace prostřednictvím pipeline

Definování agentů

V sekvenční orchestraci jsou agenti uspořádáni do kanálu, kde každý agent zpracovává úlohu postupně a předává výstup dalšímu agentu v pořadí.

Nastavení klienta Azure OpenAI

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

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

Vytvořte specializované agenty, které budou fungovat postupně:

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

Nastavení sekvenční orchestrace

Sestavení pracovního postupu pomocí AgentWorkflowBuilder:

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

Spuštění sekvenčního pracovního postupu

Spusťte pracovní postup a zpracujte události:

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

Ukázkový výstup

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!

Sekvenční orchestrace se zapojením člověka

Postupné orchestrace podporují interakce řízené lidmi prostřednictvím schvalování nástrojů. Když agenti používají nástroje zabalené ApprovalRequiredAIFunction, pracovní postup se pozastaví a vygeneruje RequestInfoEvent obsahující ToolApprovalRequestContent. Externí systémy (například lidské operátory) můžou zkontrolovat volání nástroje, schválit nebo odmítnout a pracovní postup se odpovídajícím způsobem obnoví.

Návod

Další podrobnosti o modelu požadavků a odpovědí najdete v tématu Human-in-the-Loop.

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

Vytvořte agenty, ve kterých jsou citlivé nástroje zabalené: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");

Sestavení a spuštění pomocí zpracování schválení

Normálně sestavte sekvenční pracovní postup. Tok schválení se zpracovává prostřednictvím streamu událostí:

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

Poznámka:

AgentWorkflowBuilder.BuildSequential() podporuje schválení nástroje bez nutnosti další konfigurace. Když agent volá nástroj zabalený ApprovalRequiredAIFunction, pracovní postup se automaticky pozastaví a vygeneruje RequestInfoEvent.

Návod

Kompletní spustitelný příklad procesu schvalování najdete v ukázceGroupChatToolApproval. Stejný RequestInfoEvent vzor zpracování platí i pro jiné orchestrace.

Nad rámec schválení nástrojů: Interaktivní zpětná vazba

Schválení nástroje umožňuje člověku přijmout nebo odmítnout konkrétní volání nástroje, ale sekvenční orchestrace neobsahuje integrovaný krok, který by se pozastavil pro bezplatnou zpětnou vazbu uživatelů mezi agenty a nemůže vrátit řízení do předchozího agenta. Když agent potřebuje interaktivně požádat uživatele o další informace a iterovat před pokračováním; Například shromažďování podrobností o rezervacích před voláním nástroje pro rezervaci; místo toho použijte jeden z následujících přístupů:

• Orchestrace předání je ve výchozím nastavení interaktivní: když agent reaguje bez předání, vrátí se ovládací prvek uživateli pro další vstup a povolí v rámci orchestrace vícenásobný návrat zpět a zpět. Omezte každého agenta na jediný cíl předání, aby se napodobil sekvenční tok, který se stále pozastaví kvůli vstupu uživatele.
• Vlastní pracovní postup vytvořený pomocí WorkflowBuilder a RequestPort vám umožňuje v kterémkoli bodě odeslat uživateli typovaný požadavek a směrovat odpověď zpět exekutoru, který můžete v pipeline umístit před své agenty nebo za ně.

Klíčové koncepty

• Sekvenční zpracování: Každý agent zpracovává výstup předchozího agenta v pořadí.
• AgentWorkflowBuilder.BuildSequential(): Vytvoří sekvenční pracovní postup z kolekce agentů.
• ChatClientAgent: Představuje agenta zálohovaného chatovacím klientem s konkrétními pokyny.
• InProcessExecution.RunStreamingAsync(): Spustí pracovní postup a pro streamování událostí v reálném čase vrátí StreamingRun.
• Zpracování událostí: Monitorování průběhu agenta prostřednictvím AgentResponseUpdateEvent a dokončení prostřednictvím WorkflowOutputEvent
• Schválení nástroje: Zabalte citlivé nástroje tak, aby před provedením ApprovalRequiredAIFunction vyžadovaly lidské schválení.
• RequestInfoEvent: Vygenerováno, když nástroj vyžaduje schválení; obsahuje ToolApprovalRequestContent podrobnosti o volání nástroje.
• Interaktivní HITL: Sekvenční orchestrace zahrnuje schvalování nástrojů; pro interaktivní obousměrnou komunikaci, kdy agent získává od uživatele další informace, použijte orchestrace předání nebo vlastní RequestPort pracovní postup.

V sekvenční orchestraci každý agent zpracovává úlohu postupně, přičemž výstup plynule přechází z jednoho agenta na druhého. Začněte definováním agentů pro dvoufázový proces:

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

Nastavení sekvenční orchestrace

Třída SequentialBuilder vytvoří kanál, ve kterém agenti zpracovávají úlohy v pořadí. Každý agent vidí celou historii konverzací a přidá odpověď:

from agent_framework.orchestrations import SequentialBuilder

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

Spuštění sekvenčního pracovního postupu

Spusťte pracovní postup a shromážděte konečný výstup. Výstup terminálu je AgentResponse obsahující zprávy odpovědí posledního agenta:

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

Ukázkový výstup

===== 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!

Pokročilé: Kombinování agentů s vlastními exekutory

Sekvenční orchestrace podporuje kombinování agentů s vlastními exekutory pro specializované zpracování. To je užitečné, když potřebujete vlastní logiku, která nevyžaduje LLM:

Definování vlastního exekutoru

Poznámka:

Když vlastní exekutor následuje agenta v sekvenci, jeho obslužná rutina obdrží AgentExecutorResponse (protože agenti jsou interně zabalení AgentExecutor). Umožňuje agent_response.full_conversation přístup k celé historii konverzací. Vlastní exekutor, který se používá jako poslední účastník (ukončovací), musí volat ctx.yield_output(AgentResponse(...)), aby se jeho výstup stal konečným výstupem pracovního postupu.

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

Sestavení smíšeného sekvenčního pracovního postupu

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

Ukázkový výstup s využitím vlastního exekutoru

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

Ovládání kontextu mezi agenty

Ve výchozím nastavení každý agent v SequentialBuilder pracovním postupu využívá úplnou konverzaci předchozího agenta (vstup a zprávy odpovědí). Nastavení chain_only_agent_responses=True konfiguruje všechny agenty v sekvenci tak, aby místo toho spotřebovávají pouze zprávy odpovědí předchozího agenta:

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

To je užitečné pro kanály překladu, progresivní upřesňování a další scénáře, ve kterých by se měl každý agent soustředit výhradně na transformaci výstupu předchozího agenta, bez vlivu na dřívější konverzační tahy.

Úplný příklad najdete v tématu sequential_chain_only_agent_responses.py v úložišti Agent Framework.

Návod

Pro podrobnější kontrolu nad kontextovým tokem, včetně vlastních funkcí filtru, si přečtěte Kontextové režimy v referenčních informacích k agent executor.

Zprostředkující výstupy

Ve výchozím nastavení SequentialBuilder označí posledního účastníka jako výstupní zdroj terminálu (output_from). Pouze výstup daného účastníka se objeví jako událost "output".

Pokud chcete zobrazit i výstupy předchozích účastníků, předejte intermediate_output_from je účastníkům, které chcete určit jako přechodné zdroje. To implicitně vyřadí tyto účastníky ze sady default-final — emitují události "intermediate" namísto událostí "output":

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

V režimu streamování můžete v reálném čase zpracovávat události "intermediate" i "output":

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)

Sekvenční orchestrace se zapojením člověka

Sekvenční orchestrace podporují interakce s člověkem v procesu dvěma způsoby: schválení nástrojů pro řízení volání citlivých nástrojů a požadavek na informace pro pozastavení po každé odpovědi agenta za účelem shromáždění zpětné vazby.

Návod

Další podrobnosti o modelu požadavků a odpovědí najdete v tématu Human-in-the-Loop.

Schválení nástrojů v sekvenčních pracovních postupech

Slouží @tool(approval_mode="always_require") k označení nástrojů, které před provedením potřebují schválení člověkem. Pracovní postup se pozastaví a vygeneruje request_info událost, když se agent pokusí volat nástroj.

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

Zpracování streamu událostí a zpracování žádostí o schválení:

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)

Návod

Úplný spustitelný příklad najdete v tématu sequential_builder_tool_approval.py. Schválení nástroje funguje bez SequentialBuilder jakékoli další konfigurace tvůrce.

Žádost o informace o zpětné vazbě agenta

Použijte .with_request_info() k pozastavení po odpovědi konkrétních agentů, což umožňuje externí vstup (například lidskou kontrolu) před tím, než začne další agent:

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)

Návod

Projděte si úplné ukázky: sekvenční schvalování nástrojů a informace o sekvenčních žádostech.

Klíčové koncepty

• Sdílený kontext: Ve výchozím nastavení každý agent využívá úplnou konverzaci předchozího agenta, včetně vstupních a odpovědí.
• Řízení kontextu: Slouží chain_only_agent_responses=True ke konfiguraci agentů tak, aby využívaly pouze zprávy odpovědí předchozího agenta.
• AgentResponse Output: Výstup terminálu pracovního postupu je AgentResponse obsahující odpověď posledního agenta (ne úplnou konverzaci).
• Order Matters: Agenti provádějí úkoly výhradně v pořadí uvedeném v participants seznamu.
• Flexibilní účastníci: Můžete kombinovat agenty a exekutory v libovolném pořadí.
• Vlastní ukončovací kontrakt: Vlastní exekutor použitý jako poslední účastník musí volat ctx.yield_output(AgentResponse(...)) k vytvoření výstupu terminálu.
• Přechodné výstupy: Použití intermediate_output_from=[...] nebo intermediate_output_from="all_other" zobrazení průběhu účastníků jako mezilehlých událostí pracovního postupu, nejen výstup terminálu posledního účastníka
• Schválení nástroje: Slouží @tool(approval_mode="always_require") k citlivým operacím, které potřebují kontrolu člověka.
• Informace o žádosti: Slouží .with_request_info(agents=[...]) k pozastavení po konkrétních agentech pro externí zpětnou vazbu.

Go může sestavovat sekvenční pracovní postupy agentů pomocí workflow/agentworkflownástroje . NewSequentialWorkflowBuilder hostuje každého agenta jako exekutor pracovního postupu, spojuje je v daném pořadí a poskytuje konečnou dávku zpráv jako výstup pracovního postupu.

Nastavení konfigurace 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
}

Výstraha

azidentity.NewDefaultAzureCredential 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říklad azidentity.NewManagedIdentityCredential, abyste se vyhnuli 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ů.

Definujte své agenty Go

Vytvořte specializované agenty, které budou fungovat postupně:

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

Nastavení sekvenční orchestrace

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

Spuštění sekvenčního pracovního postupu

Spusťte pracovní postup a zpracujte výstupní události:

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

Ukázkový výstup

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

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

Sekvenční orchestrace se zapojením člověka

Postupné pracovní postupy se můžou pozastavit pro schvalování nástrojů, když hostovaný agent používá nástroj vyžadovaný schválením. Zabalte nástroj tool.ApprovalRequiredFunc, pak poslouchejte workflow.RequestInfoEvent a odpovězte pomocí ToolApprovalResponseContent.

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

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
}

Sestavení a spuštění pomocí zpracování schválení

Zpracování žádostí o schválení ve streamu událostí:

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

Pokročilé: Kombinování agentů s vlastními exekutory

Pro smíšené pipeline hostujte agenty pomocí agentworkflow.New a připojte je k vlastním executorům pomocí 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
}

Ovládání kontextu mezi agenty

NewSequentialWorkflowBuilder používá výchozí konfiguraci hostovaného agenta, kde každý podřízený agent přijímá příchozí zprávy a zprávy odpovědí předchozího agenta. Pokud chcete zřetězit pouze předchozí odpovědi agenta, nastavte WithChainOnlyAgentResponses(true):

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

Zprostředkující výstupy

Ve výchozím nastavení NewSequentialWorkflowBuilder vygeneruje výstup každého účastníka jako výstup zprostředkujícího pracovního postupu a vygeneruje konečnou dávku zpráv jako výstup terminálu. Chcete-li explicitně vybrat požadované výstupy účastníků, zkombinujte WithIntermediateOutputFrom a WithOutputFrom:

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

Použijte OutputEvent.IsIntermediate() k rozlišení průběžných výstupů účastníků od koncových výstupů.

Klíčové koncepty

• Sekvenční zpracování: Každý agent nebo exekutor zpracovává výstup předchozího kroku v pořadí.
• agentworkflow.NewSequentialWorkflowBuilder(): Vytvoří sekvenční pracovní postup z kolekce agentů.
• Hostovaní agenti: agentworkflow.New Zpřístupňuje možnosti konfigurace agenta pro předávání zpráv, opětovné přiřazení rolí, události aktualizace a zachycení požadavků.
• Vlastní exekutory: Ruční workflow.NewBuilder kanály můžou kombinovat hostované agenty a deterministické exekutory.
• Schválení nástroje: Nástroje vyžadované schválením pozastaví pracovní postup a vygenerují RequestInfoEvent hodnoty obsahující ToolApprovalRequestContent.
• Zprostředkující výstupy: WithIntermediateOutputFrom označí výstupy vybraných účastníků pomocí workflow.OutputTagIntermediate.

Návod

Podívejte se na ukázku vzorů pracovních postupů agenta a ukázku agentů v pracovních postupech, kde najdete kompletní spustitelné sekvenční pracovní postupy.

Další kroky

Souběžná orchestrace