Microsoft Agent Framework Munkafolyamat-vezérlések – Szekvenciális

A szekvenciális szervezés során az ügynökök egy folyamatba vannak rendezve. Minden ügynök egymás után dolgozza fel a feladatot, és átadja a kimenetét a sorozat következő ügynökének. Ez olyan munkafolyamatokhoz ideális, ahol minden lépés az előzőre épül, például dokumentum-felülvizsgálatra, adatfeldolgozási folyamatokra vagy többfázisú érvelésre.

Fontos

Alapértelmezés szerint a sorozat minden ügynöke az előző ügynök teljes beszélgetését használja fel – az előző ügynöknek küldött bemeneti üzeneteket és a válaszüzeneteket is. Az ügynököket úgy konfigurálhatja, hogy ehelyett csak az előző ügynök válaszüzeneteit használják fel. Részletekért lásd: Az ügynökök közötti környezet szabályozása .

Tudnivalók

• Ügynökök szekvenciális folyamatláncának létrehozása
• Ügynökök láncba kapcsolása, ahol mindegyik az előző kimenetre épül
• Emberi szerepvállalás hozzáadása érzékeny eszközhívásokhoz
• Ügynökök és egyéni végrehajtók keverése speciális feladatokhoz
• A beszélgetés áramlásának nyomon követése a csatornán keresztül

Az ügynökök definiálása

A szekvenciális vezénylés során az ügynökök egy folyamatba vannak rendezve, amelyben minden ügynök feldolgozza a feladatot, és átadja a kimenetet a következő ügynöknek a sorrendben.

Az Azure OpenAI-ügyfél beállítása

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

Figyelmeztetés

DefaultAzureCredential a fejlesztéshez kényelmes, de a termelési környezetben gondos megfontolást igényel. Éles környezetben fontolja meg egy adott hitelesítő adat (pl. ManagedIdentityCredential) használatát a késési problémák elkerülése, a hitelesítő adatok nem szándékos próbálgatásának és a tartalék mechanizmusokból eredő esetleges biztonsági kockázatok elkerülése érdekében.

Speciális ügynökök létrehozása, amelyek sorrendben fognak működni:

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

Az szekvenciális orchestráció beállítása

Hozza létre a munkafolyamatot a következővel AgentWorkflowBuilder:

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

A szekvenciális munkafolyamat futtatása

Hajtsa végre a munkafolyamatot, és dolgozza fel az eseményeket:

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

Mintakimenet

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!

Szekvenciális vezénylés a Human-in-the-Loop használatával

A szekvenciális vezénylések az ember általi közbeavatkozást támogatják a folyamatba történő eszközjóváhagyás révén. Ha az ügynökök ApprovalRequiredAIFunction-vel körülvett eszközöket használnak, a munkafolyamat szünetel, és kibocsát egy RequestInfoEvent tartalmazó ToolApprovalRequestContent. A külső rendszerek (például egy emberi operátor) megvizsgálhatják az eszközhívást, jóváhagyhatják vagy elutasíthatják azt, és a munkafolyamat ennek megfelelően folytatódik.

Jótanács

A kérelem- és válaszmodell további részleteiért lásd: Human-in-the-Loop.

Ügynökök definiálása jóváhagyás-köteles eszközökkel

Hozzon létre olyan ügynököket, amelyekbe a bizalmas eszközök be vannak csomagolva 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");

Építés és futtatás jóváhagyási folyamatkezeléssel

A szekvenciális munkafolyamatot normál módon hozhatja létre. A jóváhagyási folyamat az eseményfolyamon keresztül történik:

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

Megjegyzés:

AgentWorkflowBuilder.BuildSequential() alapértelmezetten támogatja az eszköz jóváhagyását – nincs szükség további konfigurációra. Amikor egy ügynök becsomagolt ApprovalRequiredAIFunction eszközt hív meg, a munkafolyamat automatikusan szünetel és RequestInfoEvent-t bocsát ki.

Jótanács

A jóváhagyási folyamat teljes futtatható példáját a mintábanGroupChatToolApproval találhatja meg. Ugyanez a RequestInfoEvent kezelési minta érvényes más orchestrationökre is.

Alapfogalmak

• Szekvenciális feldolgozás: Minden ügynök az előző ügynök kimenetét dolgozza fel sorrendben
• AgentWorkflowBuilder.BuildSequential(): Folyamat-munkafolyamatot hoz létre ügynökök gyűjteményéből
• ChatClientAgent: Egy csevegőügyfél által támogatott ügynököt jelöl, konkrét útmutatással
• InProcessExecution.RunStreamingAsync(): Futtatja a munkafolyamatot StreamingRun , és valós idejű eseménystreamelést ad vissza
• Eseménykezelés: Az ügynök előrehaladásának AgentResponseUpdateEvent és befejezésének monitorozása WorkflowOutputEvent
• Eszközjóváhagyás: A bizalmas eszközöket ApprovalRequiredAIFunction-val látja el, hogy emberi jóváhagyást igényeljen a végrehajtás előtt.
• RequestInfoEvent: Kibocsátva, ha egy eszköz jóváhagyást igényel; tartalmazza ToolApprovalRequestContent az eszköz hívásának részleteit

A szekvenciális vezénylés során minden ügynök egymás után dolgozza fel a feladatot, és a kimenet egyikről a másikra halad. Először definiálja az ügynököket egy kétfázisú folyamathoz:

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

Az szekvenciális orchestráció beállítása

Az SequentialBuilder osztály létrehoz egy folyamatot, amelyben az ügynökök sorrendben dolgozzák fel a feladatokat. Minden ügynök látja a teljes beszélgetési előzményeket, és hozzáadja a válaszukat:

from agent_framework.orchestrations import SequentialBuilder

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

A szekvenciális munkafolyamat futtatása

Hajtsa végre a munkafolyamatot, és gyűjtse össze a végső kimenetet. A terminálkimenet az AgentResponse utolsó ügynök válaszüzeneteit tartalmazza:

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

Mintakimenet

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

Speciális: Ügynökök összekeverése testreszabott végrehajtókkal

A szekvenciális vezénylés támogatja az ügynökök és az egyéni végrehajtók speciális feldolgozásra való keverését. Ez akkor hasznos, ha olyan egyéni logikára van szüksége, amely nem igényel LLM-et:

Egyéni végrehajtó definiálása

Megjegyzés:

Amikor egy testreszabott végrehajtó követ egy ügynököt a sorrendben, a kezelő kap egy AgentExecutorResponse (mert az ügynökök belsőleg AgentExecutor csomagolva vannak). A teljes beszélgetési előzmények eléréséhez használható agent_response.full_conversation . Az utolsó résztvevőként (terminátorként) használt egyéni végrehajtónak meg kell hívnia ctx.yield_output(AgentResponse(...)) , hogy a kimenete a munkafolyamat terminálkimenetévé váljon.

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

Vegyes szekvenciális munkafolyamat létrehozása

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

Mintakimenet egyéni végrehajtóval

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

Az ügynökök közötti környezet szabályozása

Alapértelmezés szerint a SequentialBuilder munkafolyamat minden ügynöke az előző ügynök teljes beszélgetését használja (bemenet + válaszüzenetek). A beállítás chain_only_agent_responses=True a sorrendben szereplő összes ügynököt úgy konfigurálja, hogy ehelyett csak az előző ügynök válaszüzeneteit használja fel:

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

Ez a fordítási folyamatok, a fokozatos finomítás és más olyan forgatókönyvek esetében hasznos, ahol minden ügynöknek kizárólag az előző ügynök kimenetének átalakítására kell összpontosítania, anélkül, hogy a korábbi beszélgetési fordulatok befolyásolták volna.

Egy teljes példáért lásd a sequential_chain_only_agent_responses.py fájlt az Agent Framework adattárában.

Jótanács

A környezeti áramlás — beleértve az egyedi szűrőfüggvényeket is — részletesebb szabályozásához tekintse meg az Agent Executor Környezeti módok című referenciáját.

Köztes kimenetek

Alapértelmezés szerint SequentialBuilder az utolsó résztvevőt jelöli ki terminálkimeneti forrásként (final_output_from). Csak az adott résztvevő kimenete jelenik meg "output" eseményként.

A korábbi résztvevők kimenetének megjelenítéséhez is adja meg a(z) intermediate_output_from elemet azokkal a résztvevőkkel, amelyeket köztes forrásként szeretne kijelölni. Ez hallgatólagosan kiveszi ezeket a résztvevőket az alapértelmezett végső halmazból — "intermediate" eseményeket generálnak "output" események helyett:

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

Valós időben, streaming módban kezelheti mind a "intermediate", mind a "output" eseményeket:

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)

Szekvenciális vezénylés a Human-in-the-Loop használatával

A szekvenciális összehangolások kétféleképpen támogatják az emberi beavatkozásokat: a bizalmas eszközhívások vezérlésére szolgáló eszközjóváhagyás, valamint az ügynökök válaszait követő visszajelzések összegyűjtése érdekében történő szüneteltetés.

Jótanács

A kérelem- és válaszmodell további részleteiért lásd: Human-in-the-Loop.

Eszközjóváhagyás szekvenciális munkafolyamatokban

A végrehajtás előtt emberi jóváhagyást igénylő eszközök megjelölésére használható @tool(approval_mode="always_require") . A munkafolyamat szünetel, és eseményt request_info bocsát ki, amikor az ügynök megpróbálja meghívni az eszközt.

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

Az eseményfolyam feldolgozása és a jóváhagyási kérelmek kezelése:

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)

Jótanács

Egy teljes futtatható példáért lásd: sequential_builder_tool_approval.py. Az eszközjóváhagyás SequentialBuilder további építő konfigurációs beállítás nélkül működik.

Információ kérése az ügynök visszajelzéséhez

Az .with_request_info() adott ügynökök válasza után szünetelhet, lehetővé téve a külső bemenetet (például az emberi felülvizsgálatot) a következő ügynök kezdete előtt:

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)

Jótanács

Tekintse meg a teljes mintákat: szekvenciális eszköz jóváhagyása és szekvenciális kérelmek adatai.

Alapfogalmak

• Megosztott környezet: Alapértelmezés szerint minden ügynök az előző ügynök teljes beszélgetését használja, beleértve a bemeneti és válaszüzeneteket is
• Környezetvezérlés: Ügynökök chain_only_agent_responses=True konfigurálása csak az előző ügynök válaszüzeneteinek felhasználására
• AgentResponse Kimenet: A munkafolyamat terminálkimenete AgentResponse az utolsó ügynök válaszát tartalmazza (nem a teljes beszélgetést)
• Rendelési ügyek: Az ügynökök szigorúan a listában megadott sorrendben hajtanak végre participants
• Rugalmas résztvevők: Az ügynököket és az egyéni végrehajtókat tetszőleges sorrendben keverheti
• Egyéni Terminator-szerződés: Az utolsó résztvevőként használt egyéni végrehajtónak meg kell hívnia ctx.yield_output(AgentResponse(...)) a terminálkimenet előállításához
• Köztes kimenetek: Úgy van beállítva intermediate_outputs=True , hogy minden résztvevő kimenete munkafolyamat-eseményként output jelenik meg, nem csak az utolsó résztvevő kimeneteként
• Eszközjóváhagyás: Olyan érzékeny műveletekhez használható @tool(approval_mode="always_require") , amelyekhez emberi felülvizsgálatra van szükség
• Információ kérése: .with_request_info(agents=[...]) segítségével szüneteltetni lehet az egyes ügynökök utáni folyamatokat a külső visszajelzések érdekében.

Következő lépések

Egyidejű vezénylés