Microsoft Agent Framework-munkafolyamatok – Human-in-the-loop (HITL)

Ez a lap áttekintést nyújt a Microsoft Agent Framework munkafolyamat-rendszerében az emberi beavatkozásokról (HITL ). A HITL a munkafolyamatok kérés- és válaszkezelési mechanizmusával érhető el, amely lehetővé teszi a végrehajtók számára, hogy kéréseket küldjenek külső rendszereknek (például emberi operátoroknak), és várják meg a válaszaikat, mielőtt folytatják a munkafolyamat végrehajtását.

Áttekintés

A munkafolyamat végrehajtói kéréseket küldhetnek a munkafolyamaton kívülre, és várhatják a válaszokat. Ez olyan helyzetekben hasznos, amikor a végrehajtónak külső rendszerekkel kell kommunikálnia, például a cikluson belüli emberi interakciókkal vagy bármely más aszinkron művelettel.

Hozzunk létre egy munkafolyamatot, amely arra kéri az emberi operátort, hogy találjon ki egy számot, és egy végrehajtó segítségével állapítsa meg, hogy a becslés helyes-e.

Kérelem- és válaszkezelés engedélyezése munkafolyamatban

A kérések és a válaszok kezelése egy speciális típus, RequestPort, által történik.

Az A RequestPort egy kommunikációs csatorna, amely lehetővé teszi a végrehajtók számára a kérések küldését és a válaszok fogadását. Amikor egy végrehajtó üzenetet küld egy RequestPortcímzettnek, a kérelemport olyan RequestInfoEvent üzenetet bocsát ki, amely tartalmazza a kérés részleteit. A külső rendszerek figyelhetik ezeket az eseményeket, feldolgozhatják a kéréseket, és válaszokat küldhetnek vissza a munkafolyamatnak. A keretrendszer automatikusan visszairányítja a válaszokat a megfelelő végrehajtóhoz az eredeti kérés alapján.

// Create a request port that receives requests of type NumberSignal and responses of type int.
var numberRequestPort = RequestPort.Create<NumberSignal, int>("GuessNumber");

Adja hozzá a bemeneti portot egy munkafolyamathoz.

JudgeExecutor judgeExecutor = new(42);
var workflow = new WorkflowBuilder(numberRequestPort)
    .AddEdge(numberRequestPort, judgeExecutor)
    .AddEdge(judgeExecutor, numberRequestPort)
    .WithOutputFrom(judgeExecutor)
    .Build();

A definíciónak JudgeExecutor egy célszámra van szüksége, és képesnek kell lennie eldönteni, hogy a becslés helyes-e. Ha nem helyes, akkor egy másik kérést küld, amely új becslést kér a RequestPort.

internal enum NumberSignal
{
    Init,
    Above,
    Below,
}

internal sealed class JudgeExecutor() : Executor<int>("Judge")
{
    private readonly int _targetNumber;
    private int _tries;

    public JudgeExecutor(int targetNumber) : this()
    {
        this._targetNumber = targetNumber;
    }

    public override async ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        this._tries++;
        if (message == this._targetNumber)
        {
            await context.YieldOutputAsync($"{this._targetNumber} found in {this._tries} tries!", cancellationToken);
        }
        else if (message < this._targetNumber)
        {
            await context.SendMessageAsync(NumberSignal.Below, cancellationToken: cancellationToken);
        }
        else
        {
            await context.SendMessageAsync(NumberSignal.Above, cancellationToken: cancellationToken);
        }
    }
}

Pythonban a végrehajtók ctx.request_info() segítségével küldenek kéréseket és a @response_handler dekorátorral kezelik a válaszokat.

Hozzunk létre egy munkafolyamatot, amely arra kéri az emberi operátort, hogy találjon ki egy számot, és egy végrehajtó segítségével állapítsa meg, hogy a becslés helyes-e.

Kérelem- és válaszkezelés engedélyezése munkafolyamatban

from dataclasses import dataclass

from agent_framework import (
    Executor,
    WorkflowBuilder,
    WorkflowContext,
    handler,
    response_handler,
)


@dataclass
class NumberSignal:
    hint: str  # "init", "above", or "below"


class JudgeExecutor(Executor):
    def __init__(self, target_number: int):
        super().__init__(id="judge")
        self._target_number = target_number
        self._tries = 0

    @handler
    async def handle_guess(self, guess: int, ctx: WorkflowContext[int, str]) -> None:
        self._tries += 1
        if guess == self._target_number:
            await ctx.yield_output(f"{self._target_number} found in {self._tries} tries!")
        elif guess < self._target_number:
            await ctx.request_info(request_data=NumberSignal(hint="below"), response_type=int)
        else:
            await ctx.request_info(request_data=NumberSignal(hint="above"), response_type=int)

    @response_handler
    async def on_human_response(
        self,
        original_request: NumberSignal,
        response: int,
        ctx: WorkflowContext[int, str],
    ) -> None:
        await self.handle_guess(response, ctx)


judge = JudgeExecutor(target_number=42)
workflow = WorkflowBuilder(start_executor=judge).build()

A @response_handler dekoratőr automatikusan regisztrálja a metódust a megadott kérés- és választípusok válaszainak kezelésére. A keretrendszer a beérkező válaszokat a megfelelő kezelőhöz rendeli a original_request és response paraméterek típusjegyzetei alapján.

A munkafolyamatok a RequestPort segítségével támogatják a human-in-the-loop mintákat, amely szünetelteti a végrehajtást, és külső bemenetre vár.

approvalPort := workflow.RequestPort{
    ID:       "ApprovalPort",
    Request:  reflect.TypeFor[string](),
    Response: reflect.TypeFor[bool](),
}

approval := approvalPort.Bind()
finalize := workflow.NewExecutor("FinalizeExecutor", func(approved bool) string {
    if approved {
        return "Request approved by the human reviewer"
    }
    return "Request rejected by the human reviewer"
}).Bind()

wf, err := workflow.NewBuilder(approval).
    AddEdge(approval, finalize).
    WithOutputFrom(finalize).
    Build()

Az A RequestPort a munkafolyamat és a külvilág közötti beírt kérés-válasz csatornát definiálja. Amikor egy végrehajtó elér egy kérésportot, a munkafolyamat szünetel, és külső kéréseseményt bocsát ki. A munkafolyamat akkor folytatódik, amikor külső válasz érkezik.

Kérelmek és válaszok kezelése

Amikor RequestPort egy kérést kap, egy e-mailt bocsát ki RequestInfoEvent . Ezekre az eseményekre feliratkozva kezelheti a munkafolyamat bejövő kéréseit. Amikor egy külső rendszer válaszát kapja, küldje vissza a munkafolyamatba a válaszmechanizmus használatával. A keretrendszer automatikusan átirányítja a választ az eredeti kérést küldő végrehajtónak.

await using StreamingRun handle = await InProcessExecution.RunStreamingAsync(workflow, NumberSignal.Init);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    switch (evt)
    {
        case RequestInfoEvent requestInputEvt:
            // Handle `RequestInfoEvent` from the workflow
            int guess = ...; // Get the guess from the human operator or any external system
            await handle.SendResponseAsync(requestInputEvt.Request.CreateResponse(guess));
            break;

        case WorkflowOutputEvent outputEvt:
            // The workflow has yielded output
            Console.WriteLine($"Workflow completed with result: {outputEvt.Data}");
            return;
    }
}

Jótanács

Tekintse meg a teljes futtatható projekt teljes mintáját .

A végrehajtók közvetlenül, külön összetevő nélkül küldhetnek kéréseket. Amikor egy végrehajtó meghívja ctx.request_info(), a munkafolyamat egy WorkflowEvent-t ad ki type == "request_info"-vel. Ezekre az eseményekre feliratkozva kezelheti a munkafolyamat bejövő kéréseit. Amikor egy külső rendszer válaszát kapja, küldje vissza a munkafolyamatba a válaszmechanizmus használatával. A keretrendszer automatikusan átirányítja a választ a végrehajtó metódusára @response_handler .

from collections.abc import AsyncIterable

from agent_framework import WorkflowEvent


async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str, int] | None:
    """Process events from the workflow stream to capture requests."""
    requests: list[tuple[str, NumberSignal]] = []
    async for event in stream:
        if event.type == "request_info":
            requests.append((event.request_id, event.data))

    # Handle any pending human feedback requests.
    if requests:
        responses: dict[str, int] = {}
        for request_id, request in requests:
            guess = ...  # Get the guess from the human operator or any external system.
            responses[request_id] = guess
        return responses

    return None

# Initiate the first run of the workflow with an initial guess.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(25, stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    # Run the workflow until there is no more human feedback to provide,
    # in which case this workflow completes.
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Jótanács

Teljes futtatható fájlért tekintse meg ezt a teljes mintát .

Figyeljen a(z) workflow.RequestInfoEvent eseményre, hozzon létre egy választ a kérés alapján, és folytassa a futtatást ezzel a válasszal:

run, err := inproc.Default.Run(ctx, wf, "Approve deployment to production?")
if err != nil {
    return err
}

var request *workflow.ExternalRequest
for evt := range run.NewEvents() {
    if requestEvent, ok := evt.(workflow.RequestInfoEvent); ok {
        request = requestEvent.Request
        break
    }
}

response, err := request.CreateResponse(true)
if err != nil {
    return err
}

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

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Println(output.Output)
    }
}

Jótanács

A teljes futtatható fájlhoz tekintse meg a human-in-the-loop mintát .

Az Ember-a-hurokban az Ügynök Orkesztrációkkal

A fent leírt RequestPort minta együttműködik egyedi végrehajtókkal és WorkflowBuilder. Ügynökközvetítések (például szekvenciális, egyidejű vagy csoportos chat munkafolyamatok) használatakor az eszközjóváhagyás a humán-közreműködéses kérés-válasz mechanizmussal érhető el.

Az ügynökök olyan eszközöket használhatnak, amelyek végrehajtás előtt emberi jóváhagyást igényelnek. Amikor az ügynök megpróbál meghívni egy jóváhagyást igénylő eszközt, a munkafolyamat szünetel, és egy RequestInfoEvent elemet bocsát ki, akárcsak a RequestPort minta esetében, de az esemény adatcsomagja egy egyéni kéréstípus helyett egy ToolApprovalRequestContent típust (C# és Go) vagy egy Content mezővel rendelkező type == "function_approval_request" típust (Python) tartalmaz.

Olyan interaktív forgatókönyvek esetén, ahol az ügynöknek több információt kell gyűjtenie a felhasználótól, és iterálnia kell a folytatás előtt; ahelyett, hogy csak egy eszközhívást hagyjon jóvá vagy utasítson el; használja a handoff vezénylést. A továbbadás alapértelmezés szerint interaktív: amikor egy ügynök úgy válaszol, hogy nem adja át a feladatot egy másik ügynöknek, a vezérlés visszakerül a felhasználóhoz a következő bevitelhez, ami lehetővé teszi a többfordulós oda-vissza párbeszédet az orchestráción belül. A szekvenciális, egyidejű és csoportos csevegési orchesztrációk maguktól nem állnak meg szabad formátumú felhasználói bevitelre várva; kombinálja őket egy RequestPort elemmel egy egyéni WorkflowBuilder-munkafolyamatban, ha a lépések között ilyen vezérlésre van szüksége.

Jótanács

A kóddal kapcsolatos teljes példákért lásd:

Ellenőrzőpontok és kérések

Az ellenőrzőpontokról további információt az Ellenőrzőpontok című témakörben talál.

Ellenőrzőpont létrehozásakor a függőben lévő kérések is az ellenőrzőpont állapotának részeként lesznek mentve. Amikor visszaállít egy ellenőrzőpontot, a függőben lévő kérések objektumként RequestInfoEvent lesznek újra kibocsátva, így rögzítheti és megválaszolhatja őket. Az ellenőrzőpontról is folytathatja, és ugyanabban a hívásban válaszokat is megadhat, ha a(z) checkpoint_id számára átadja a(z) responses és workflow.run(...) elemet is.

A visszaállítás után figyelje meg az újra kibocsátott kéréseseményeket, és válaszoljon a nyelvére korábban bemutatott válaszmechanizmussal.

Következő lépések

  • Last updated on