Orkiestracje przepływów pracy programu Microsoft Agent Framework — sekwencyjne

W sekwencyjnej organizacji agenci są zorganizowani w kanałach. Każdy agent przetwarza zadanie z kolei, przekazując dane wyjściowe do następnego agenta w sekwencji. Jest to idealne rozwiązanie w przypadku przepływów pracy, w których każdy krok opiera się na poprzednim, takim jak przegląd dokumentów, potoki przetwarzania danych lub rozumowanie wieloetapowe.

Important

Domyślnie każdy agent w sekwencji korzysta z pełnej konwersacji poprzedniego agenta — zarówno komunikatów wejściowych dostarczonych do poprzedniego agenta, jak i komunikatów odpowiedzi. Zamiast tego można skonfigurować agentów tak, aby używali tylko komunikatów odpowiedzi poprzedniego agenta. Aby uzyskać szczegółowe informacje, zobacz Kontrolowanie kontekstu między agentami .

Czego nauczysz się

• Jak utworzyć sekwencyjny potok agentów
• Jak utworzyć łańcuch agentów, w których każda z nich opiera się na poprzednich danych wyjściowych
• Jak dodać zatwierdzenie z udziałem człowieka w cyklu dla wywołań poufnych narzędzi
• Jak mieszać agentów z niestandardowymi wykonawcami dla wyspecjalizowanych zadań
• Jak śledzić przepływ konwersacji przez pipeline

Definiowanie agentów

W sekwencyjnej aranżacji agenci są zorganizowani w potoku, w którym każdy agent przetwarza zadanie z kolei, przekazując dane wyjściowe do następnego agenta w sekwencji.

Konfigurowanie klienta usługi 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);

Warning

DefaultAzureCredential jest wygodne do programowania, ale wymaga starannego rozważenia w środowisku produkcyjnym. W środowisku produkcyjnym rozważ użycie określonego poświadczenia (np. ManagedIdentityCredential), aby uniknąć problemów z opóźnieniami, niezamierzonego sondowania poświadczeń i potencjalnych zagrożeń bezpieczeństwa wynikających z mechanizmów awaryjnych.

Utwórz wyspecjalizowanych agentów, którzy będą działać w sekwencji:

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

Konfigurowanie sekwencyjnej orkiestracji

Skompiluj przepływ pracy przy użyciu polecenia AgentWorkflowBuilder:

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

Uruchamianie sekwencyjnego przepływu pracy

Wykonaj przepływ pracy i przetwórz zdarzenia:

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

Przykładowe dane wyjściowe

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!

Sekwencyjna orkiestracja z udziałem człowieka

Sekwencyjne orkiestracje wspierają interakcje z udziałem człowieka poprzez aprobację narzędzi. Gdy agenci używają narzędzi opakowanych w polecenie ApprovalRequiredAIFunction, przepływ pracy wstrzymuje się i emituje RequestInfoEvent, który zawiera ToolApprovalRequestContent. Systemy zewnętrzne (takie jak operator ludzki) mogą sprawdzać wywołanie narzędzia, zatwierdzać je lub odrzucać, a przepływ pracy zostaje odpowiednio wznowiony.

Tip

Aby uzyskać więcej informacji na temat modelu żądania i odpowiedzi, zobacz Human-in-the-Loop.

Definiowanie agentów za pomocą narzędzi wymagających zatwierdzenia

Utwórz agentów, w których poufne narzędzia są opakowane za pomocą polecenia 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");

Kompilowanie i uruchamianie przy użyciu obsługi zatwierdzania

Skompiluj przepływ pracy sekwencyjny normalnie. Przepływ zatwierdzania jest obsługiwany za pośrednictwem strumienia zdarzeń:

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

Uwaga / Notatka

AgentWorkflowBuilder.BuildSequential() obsługuje domyślne zatwierdzanie narzędzi — nie jest wymagana żadna dodatkowa konfiguracja. Gdy agent wywołuje narzędzie owinięte w ApprovalRequiredAIFunction, przepływ pracy automatycznie wstrzymuje się i emituje RequestInfoEvent sygnał.

Tip

Pełny przykład działania tego przepływu zatwierdzania można znaleźć w przykładzieGroupChatToolApproval. Ten sam RequestInfoEvent wzorzec obsługi ma zastosowanie do innych orkiestracji.

Więcej niż zatwierdzanie narzędzi: interaktywne informacje zwrotne

Zatwierdzanie narzędzi pozwala człowiekowi zaakceptować lub odrzucić konkretne wywołanie narzędzia, ale orkiestracja sekwencyjna nie obejmuje wbudowanego kroku umożliwiającego wstrzymanie działania w celu uzyskania swobodnej informacji zwrotnej od użytkownika między agentami i nie może przekazać sterowania z powrotem do poprzedniego agenta. Gdy agent musi interaktywnie poprosić użytkownika o więcej informacji i iterować przed kontynuowaniem; na przykład zbieranie szczegółów rezerwacji przed wywołaniem narzędzia rezerwacji; Zamiast tego należy użyć jednej z następujących metod:

• Orkiestracja przekazywania jest domyślnie interaktywna: gdy agent odpowiada bez przekazania, sterowanie wraca do użytkownika, aby mógł wprowadzić kolejne dane wejściowe, co umożliwia wieloetapową wymianę w ramach orkiestracji. Ogranicz każdego agenta do jednego celu przekazania, aby odwzorować sekwencyjny przepływ, który nadal zatrzymuje się, czekając na dane wejściowe użytkownika.
• Niestandardowy przepływ pracy zbudowany przy użyciu WorkflowBuilder i RequestPort umożliwia wysłanie do użytkownika żądania tekstowego w dowolnym momencie oraz przekierowanie odpowiedzi z powrotem do wykonawcy, którego można umieścić przed agentami lub po nich w potoku przetwarzania.

Kluczowe pojęcia

• Przetwarzanie sekwencyjne: każdy agent przetwarza dane wyjściowe poprzedniego agenta w kolejności
• AgentWorkflowBuilder.BuildSequential(): Tworzy sekwencyjny przepływ pracy z kolekcji agentów
• ChatClientAgent: reprezentuje agenta wspieranego przez klienta czatu z określonymi instrukcjami
• InProcessExecution.RunStreamingAsync(): Uruchamia przepływ pracy i zwraca element StreamingRun do przesyłania strumieniowego zdarzeń w czasie rzeczywistym
• Obsługa zdarzeń: Monitorowanie postępu agenta za pomocą AgentResponseUpdateEvent oraz ukończenie za pomocą WorkflowOutputEvent
• Zatwierdzanie narzędzi: owinięcie poufnych narzędzi za pomocą ApprovalRequiredAIFunction w celu wymagania zatwierdzenia przez człowieka przed wykonaniem
• RequestInfoEvent: emitowany, gdy narzędzie wymaga zatwierdzenia; zawiera ToolApprovalRequestContent ze szczegółami wywołania narzędzia
• Interactive HITL: Orkiestracja sekwencyjna obejmuje zatwierdzanie narzędzi; w przypadku interaktywnej wymiany, w której agent zbiera od użytkownika więcej informacji, użyj orkiestracji przekazania lub niestandardowego RequestPort przepływu pracy

W sekwencyjnej aranżacji każdy agent przetwarza zadanie z kolei, a dane wyjściowe przepływają od jednego do następnego. Zacznij od zdefiniowania agentów dla dwuetapowego procesu:

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

Konfigurowanie sekwencyjnej orkiestracji

Klasa SequentialBuilder tworzy strumień, w którym moduły przetwarzają zadania według kolejności. Każdy agent widzi pełną historię konwersacji i dodaje odpowiedź:

from agent_framework.orchestrations import SequentialBuilder

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

Uruchamianie sekwencyjnego przepływu pracy

Wykonaj przepływ pracy i zbierz końcowe dane wyjściowe. Dane wyjściowe terminalu to AgentResponse zawierające komunikaty odpowiedzi ostatniego 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}")

Przykładowe dane wyjściowe

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

Zaawansowane: Łączenie agentów z niestandardowymi wykonawcami

Sekwencyjna orkiestracja obsługuje mieszanie agentów z niestandardowymi wykonawcami dla wyspecjalizowanego przetwarzania. Jest to przydatne, gdy potrzebna jest logika niestandardowa, która nie wymaga modułu LLM:

Definiowanie niestandardowego modułu wykonawczego

Uwaga / Notatka

Gdy niestandardowy wykonawca następuje po agencie w sekwencji, jego obsługujący otrzymuje AgentExecutorResponse (ponieważ agenci są wewnętrznie opakowani przez AgentExecutor). Użyj polecenia agent_response.full_conversation , aby uzyskać dostęp do pełnej historii konwersacji. Niestandardowy wykonawca używany jako ostatni uczestnik (terminator) musi wywołać ctx.yield_output(AgentResponse(...)), aby jego dane wyjściowe stały się danymi zakończeniowymi procesu roboczego.

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

Tworzenie mieszanego sekwencyjnego przepływu pracy

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

Przykładowe dane wyjściowe z niestandardowym wykonawcą

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

Zarządzanie kontekstem między agentami

Domyślnie każdy agent w procesie SequentialBuilder korzysta z pełnej konwersacji poprzedniego agenta (wiadomości wejściowe oraz odpowiedzi). Ustawienie chain_only_agent_responses=True ustawia wszystkich agentów w sekwencji tak, aby konsumowali wyłącznie komunikaty odpowiedzi poprzedniego agenta:

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

Jest to przydatne w przypadku potoków tłumaczenia, stopniowego uściślenia i innych scenariuszy, w których każdy agent powinien skupić się wyłącznie na przekształcaniu danych wyjściowych poprzedniego agenta bez wpływu wcześniejszych etapów konwersacji.

Aby zapoznać się z kompletnym przykładem, zobacz sequential_chain_only_agent_responses.py w repozytorium Agent Framework.

Tip

Aby uzyskać bardziej szczegółową kontrolę nad przepływem kontekstu — w tym funkcjami niestandardowego filtru — zobacz Tryby kontekstu w odniesieniu do Agenta Wykonawczego.

Dodaj końcową turę użytkownika w celu zachowania zgodności dostawcy

Niektórzy dostawcy chat completions zwracają pustą odpowiedź, gdy konwersacja łańcuchowa kończy się komunikatem asystenta. Dla tych dostawców ustaw wartość ensure_trailing_user_turn=True.

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

Builder dołącza syntetyczny komunikat użytkownika "please continue." tylko do kopii wywołania wysłanej do następnego agenta. Nie modyfikuje przechowywanej konwersacji. Opcja jest domyślnie ustawiona na False. Włącz go tylko wtedy, gdy dostawca wymaga końcowego komunikatu użytkownika. Działa również z agentami opakowanymi w request-info oraz z chain_only_agent_responses=True.

Dane wyjściowe pośrednie

Domyślnie SequentialBuilder wyznacza ostatniego uczestnika jako źródło danych wyjściowych terminalu (output_from). Tylko wynik tego uczestnika pojawia się jako zdarzenie "output".

Aby wyświetlić również dane wyjściowe wcześniejszych uczestników, przekaż intermediate_output_from je uczestnikom, które chcesz wyznaczyć jako źródła pośrednie. To pośrednio degraduje tych uczestników względem domyślnego zbioru końcowego — emitują zdarzenia "intermediate", a nie "output":

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

Możesz obsługiwać zarówno zdarzenia "intermediate", jak i "output" w czasie rzeczywistym w trybie strumieniowym:

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)

Gdy aktualizacje przesyłane strumieniowo są łączone w komunikaty, zmiana między dwiema niepustymi wartościami author_name rozpoczyna nowy komunikat. Ta zmiana powoduje, że dane wyjściowe przesyłane strumieniowo przez każdego agenta pozostają oddzielone w końcowej odpowiedzi. Brak autora nie uruchamia nowej wiadomości.

Sekwencyjna orkiestracja z udziałem człowieka

Sekwencyjne aranżacje obsługują interakcje człowieka w pętli na dwa sposoby: zatwierdzanie narzędzi do kontrolowania poufnych wywołań narzędzi i żądanie wstrzymania informacji po każdej odpowiedzi agenta w celu zebrania opinii.

Tip

Aby uzyskać więcej informacji na temat modelu żądania i odpowiedzi, zobacz Human-in-the-Loop.

Zatwierdzanie narzędzi w sekwencyjnych przepływach pracy

Użyj @tool(approval_mode="always_require") do oznaczania narzędzi, które wymagają zatwierdzenia przez człowieka przed wykonaniem. Przepływ pracy wstrzymuje się i emituje request_info zdarzenie, gdy agent próbuje wywołać narzędzie.

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

Przetwarzanie strumienia zdarzeń i obsługa żądań zatwierdzenia:

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)

Tip

Aby uzyskać pełny przykład możliwych do uruchomienia, zobacz sequential_builder_tool_approval.py. Zatwierdzanie narzędzi działa z SequentialBuilder bez żadnej dodatkowej konfiguracji konstruktora.

Żądanie informacji o opinii agenta

Użyj .with_request_info() do zatrzymania po odpowiedziach określonych agentów, umożliwiając zewnętrzne dane wejściowe (takie jak przegląd przez człowieka) przed rozpoczęciem następnego agenta.

Wartości w elemecie agents muszą dokładnie odpowiadać identyfikatorom agenta aranżacji, z uwzględnieniem wielkości liter. Przepływ pracy zgłasza ValueError podczas tworzenia dla nieznanego identyfikatora lub identyfikatora o niezgodnej wielkości liter.

Zdarzenie request_info zawiera odpowiedź będącą w trakcie przeglądu. Przepływ pracy nie zwraca tej odpowiedzi jako zwykłych danych wyjściowych, dopóki jej nie zatwierdzisz. Jeśli po wznowieniu przepływ pracy natrafi na kolejną pauzę typu request-info, nadal nie generuje żadnych danych wyjściowych dla nowo oczekiwanej odpowiedzi.

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)

Tip

Zobacz pełne przykłady: sekwencyjne zatwierdzanie narzędzi i informacje o sekwencyjnym żądaniu.

Kluczowe pojęcia

• Kontekst udostępniony: domyślnie każdy agent używa pełnej konwersacji poprzedniego agenta, w tym komunikatów wejściowych i odpowiedzi
• Kontrola kontekstu: służy chain_only_agent_responses=True do konfigurowania agentów do korzystania tylko z komunikatów odpowiedzi poprzedniego agenta
• Wyjście AgentResponse: Terminalne wyjście przepływu pracy to AgentResponse zawierający odpowiedź ostatniego agenta (nie pełną konwersację)
• Order Matters: Agenci wykonują ściśle w kolejności określonej w participants liście
• Elastyczni uczestnicy: możesz mieszać agentów i niestandardowych wykonawców w dowolnej kolejności
• Niestandardowy kontrakt terminatora: Niestandardowy wykonawca używany jako ostatni uczestnik musi wywołać ctx.yield_output(AgentResponse(...)) aby wygenerować wyjściowe dane terminalowe
• Pośrednie dane wyjściowe: Użyj intermediate_output_from=[...] lub intermediate_output_from="all_other", aby udostępniać postępy uczestników jako pośrednie zdarzenia przepływu pracy, a nie tylko końcowy wynik ostatniego uczestnika
• Aprobata narzędzia: Używaj @tool(approval_mode="always_require") do obsługi wrażliwych operacji wymagających przeglądu przez człowieka
• Żądaj informacji: wykorzystaj .with_request_info(agents=[...]) , aby zatrzymać się przy określonych agentach w celu uzyskania opinii zewnętrznej

W języku Go można tworzyć sekwencyjne przepływy pracy agenta za pomocą polecenia workflow/agentworkflow. NewSequentialWorkflowBuilder obsługuje każdego agenta jako wykonawcę przepływu pracy, łączy ich kolejno i zwraca końcową partię komunikatów jako wynik przepływu pracy.

Skonfiguruj konfigurację 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
}

Warning

azidentity.NewDefaultAzureCredential jest wygodne do programowania, ale wymaga starannego rozważenia w środowisku produkcyjnym. W środowisku produkcyjnym rozważ użycie konkretnego poświadczenia, takiego jak azidentity.NewManagedIdentityCredential, aby uniknąć opóźnień, niezamierzonego sprawdzania poświadczeń oraz potencjalnych zagrożeń bezpieczeństwa wynikających z mechanizmów awaryjnych.

Zdefiniuj swoich agentów Go

Utwórz wyspecjalizowanych agentów, którzy będą działać w sekwencji:

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

Konfigurowanie sekwencyjnej orkiestracji

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

Uruchamianie sekwencyjnego przepływu pracy

Wykonaj przepływ pracy i przetwórz zdarzenia wyjściowe:

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 accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
} else if !accepted {
    return fmt.Errorf("workflow doesn't accept turn tokens")
}

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

Przykładowe dane wyjściowe

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

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

Sekwencyjna orkiestracja z udziałem człowieka

Sekwencyjne przepływy pracy mogą wstrzymywać zatwierdzanie narzędzi, gdy hostowany agent używa narzędzia wymaganego do zatwierdzenia. Opakuj narzędzie za pomocą tool.ApprovalRequiredFunc, a następnie nasłuchuj na workflow.RequestInfoEvent i odpowiadaj za pomocą ToolApprovalResponseContent.

Definiowanie agentów za pomocą narzędzi wymagających zatwierdzenia

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
}

Kompilowanie i uruchamianie przy użyciu obsługi zatwierdzania

Obsługa żądań zatwierdzenia w strumieniu zdarzeń:

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

Zaawansowane: Łączenie agentów z niestandardowymi wykonawcami

W przypadku mieszanych potoków hostuj agentów za pomocą agentworkflow.New i połącz ich z niestandardowymi executorami za 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
}

Zarządzanie kontekstem między agentami

NewSequentialWorkflowBuilder używa domyślnej konfiguracji hostowanego agenta, gdzie każdy agent podrzędny odbiera komunikaty przychodzące poprzedniego agenta i komunikaty odpowiedzi. Aby połączyć łańcuch tylko z poprzednimi odpowiedziami agenta, ustaw wartość WithChainOnlyAgentResponses(true):

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

Dane wyjściowe pośrednie

Domyślnie NewSequentialWorkflowBuilder emituje dane wyjściowe każdego uczestnika jako dane wyjściowe pośredniego przepływu pracy i emituje ostateczną partię komunikatów jako dane wyjściowe terminalu. Aby jawnie wybrać żądane dane wyjściowe uczestnika, połącz WithIntermediateOutputFrom i WithOutputFrom:

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

Użyj OutputEvent.IsIntermediate(), aby odróżnić dane wyjściowe uczestnika pośredniego od danych wyjściowych terminala.

Kluczowe pojęcia

• Przetwarzanie sekwencyjne: każdy agent lub funkcja wykonawcza przetwarza dane wyjściowe poprzedniego kroku w kolejności.
• agentworkflow.NewSequentialWorkflowBuilder(): Tworzy potokowy przepływ pracy z kolekcji agentów.
• Hostowani agenci: agentworkflow.New udostępnia opcje konfiguracji agenta na potrzeby przekazywania komunikatów, ponownego przypisania roli, zdarzeń aktualizacji i przechwytywania żądań.
• Niestandardowe funkcje wykonawcze: potoki ręczne workflow.NewBuilder mogą mieszać hostowanych agentów i deterministyczne funkcje wykonawcze.
• Zatwierdzenie narzędzia: narzędzia wymagane do zatwierdzenia wstrzymują przepływ pracy i emitują RequestInfoEvent wartości zawierające ToolApprovalRequestContent.
• Dane wyjściowe pośrednie: WithIntermediateOutputFrom oznacza wybrane dane wyjściowe uczestnika za pomocą polecenia workflow.OutputTagIntermediate.

Tip

Zobacz przykład wzorców przepływów pracy agentów oraz przykład agentów w przepływach pracy, aby poznać kompletne, gotowe do uruchomienia sekwencyjne przepływy pracy.

Następne kroki

Współbieżna orkiestracja