Megosztás:

Egyszerű szekvenciális munkafolyamat létrehozása

Ez az oktatóanyag bemutatja, hogyan hozható létre egyszerű szekvenciális munkafolyamat az Agent Framework-munkafolyamatok használatával.

A szekvenciális munkafolyamatok a komplex AI-ügynökrendszerek létrehozásának alapjai. Ez az oktatóanyag bemutatja, hogyan hozhat létre egy egyszerű kétlépéses munkafolyamatot, amelyben minden lépés feldolgozza az adatokat, és átadja azokat a következő lépésnek.

Áttekintés

Ebben az oktatóanyagban egy munkafolyamatot hoz létre két végrehajtóval:

  1. Nagybetűs végrehajtó – A bemeneti szöveg nagybetűssé alakítása
  2. Fordított szöveg végrehajtója – A szöveg megfordítása és a végeredmény kimenete

A munkafolyamat a következő alapvető fogalmakat mutatja be:

  • Egyéni végrehajtó létrehozása egy kezelővel
  • Egyéni végrehajtó létrehozása függvényből
  • Végrehajtók WorkflowBuilder csatlakoztatása élekhez
  • Adatok feldolgozása szekvenciális lépéseken keresztül
  • Munkafolyamat-végrehajtás megfigyelése eseményeken keresztül

A tárgyalt fogalmak

Előfeltételek

Részletes megvalósítás

A következő szakaszok bemutatják, hogyan lehet lépésről lépésre felépíteni a szekvenciális munkafolyamatot.

1. lépés: NuGet-csomagok telepítése

Először telepítse a szükséges csomagokat a .NET-projekthez:

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

2. lépés: A nagybetű konvertáló modul definiálása

Adjon meg egy olyan végrehajtót, amely nagybetűssé konvertálja a szöveget:

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

Főbb pontok:

  • Sztringet használó függvény létrehozása, amely a nagybetűs verziót adja vissza
  • Használja a BindExecutor()-t, hogy létrehozzon egy végrehajtót a függvényből.

3. lépés: A fordított szöveg végrehajtójának meghatározása

Adjon meg egy olyan végrehajtót, amely megfordítja a szöveget:

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

Főbb pontok:

  • Olyan osztály létrehozása, amely örököl Executor<TInput, TOutput>-től
  • A HandleAsync() megvalósítása a bemenet feldolgozására és a kimenet visszaadására

4. lépés: A munkafolyamat létrehozása és csatlakoztatása

Csatlakoztassa a végrehajtókat a következővel WorkflowBuilder:

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

Főbb pontok:

  • WorkflowBuilder a konstruktor átveszi a kezdő végrehajtót
  • AddEdge() irányított kapcsolatot hoz létre nagybetűről fordítottra
  • WithOutputFrom() meghatározza, hogy mely végrehajtók hoznak létre munkafolyamat-kimeneteket
  • Build() a nem módosítható munkafolyamat létrehozása

5. lépés: A munkafolyamat végrehajtása

Futtassa a munkafolyamatot, és figyelje meg az eredményeket:

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

6. lépés: A munkafolyamat kimenetének ismertetése

A munkafolyamat futtatásakor a következő kimenet jelenik meg:

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

A "Hello, World!" bemenetet először nagybetűssé konvertálja ("HELLO, WORLD!"), majd megfordítja ("! DLROW ,OLLEH").

Alapvető fogalmak magyarázata

Végrehajtó felület

Függvényekből származó végrehajtók:

  • Használja a BindExecutor()-t, hogy végrehajtót hozzon létre egy függvényből

Végrehajtók implementálása Executor<TInput, TOutput>:

  • TInput: Az a típusú adat, amelyet ez a végrehajtó elfogad
  • TOutput: A végrehajtó által előállított adatok típusa
  • HandleAsync: A bemenetet feldolgozó és a kimenetet visszaadó metódus

.NET-munkafolyamat-szerkesztő minta

A WorkflowBuilder munkafolyamatok létrehozásához folyékony API-t biztosít:

  • Konstruktor: Átveszi a kezdő végrehajtót
  • AddEdge(): Irányított kapcsolatokat hoz létre a végrehajtók között
  • WithOutputFrom(): Meghatározza, hogy mely végrehajtók hoznak létre munkafolyamat-kimeneteket
  • Build(): Létrehozza a végleges nem módosítható munkafolyamatot

.NET-eseménytípusok

A végrehajtás során megfigyelheti az alábbi eseménytípusokat:

  • ExecutorCompletedEvent – Amikor egy végrehajtó befejezi a feldolgozást

.NET-példa befejezése

A teljes, használatra kész implementációért tekintse meg az Ügynök keretrendszer adattárában található 01_ExecutorsAndEdges mintát .

Ez a minta a következőket tartalmazza:

  • Teljes körű implementáció az összes utasítással és osztálystruktúrával
  • A munkafolyamat fogalmait ismertető további megjegyzések
  • Projekt beállításának és konfigurálásának befejezése

Áttekintés

Ebben az oktatóanyagban egy munkafolyamatot hoz létre két végrehajtóval:

  1. Nagybetűs végrehajtó – A bemeneti szöveg nagybetűssé alakítása
  2. Fordított szöveg végrehajtója – A szöveg megfordítása és a végeredmény kimenete

A munkafolyamat a következő alapvető fogalmakat mutatja be:

  • A munkaegységek (végrehajtó csomópontok) meghatározásának két módja:
    1. Olyan egyéni osztály, amely aszinkron módszerrel alosztályozódik Executor@handler
    2. Önálló aszinkron függvény, amely @executor dekorátorral van ellátva
  • Végrehajtók csatlakoztatása WorkflowBuilder
  • Adatok továbbítása a lépések között a következővel: ctx.send_message()
  • Végső kimenet a következővel: ctx.yield_output()
  • Streamelési események a valós idejű megfigyelhetőség érdekében

A tárgyalt fogalmak

Előfeltételek

  • Python 3.10 vagy újabb
  • Az Agent Framework Core Python-csomag telepítve van: pip install agent-framework-core --pre
  • Ehhez az alap példához nincs szükség külső AI-szolgáltatásokra

Részletes megvalósítás

A következő szakaszok bemutatják, hogyan lehet lépésről lépésre felépíteni a szekvenciális munkafolyamatot.

1. lépés: Kötelező modulok importálása

Először importálja a szükséges modulokat az Agent Frameworkből:

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

2. lépés: Az első végrehajtó létrehozása

Hozzon létre egy végrehajtót, amely nagybetűssé konvertálja a szöveget egy végrehajtó kezelői metódussal történő implementálásával:

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

Főbb pontok:

  • Az alosztályozással Executor szükség esetén definiálhat egy elnevezett csomópontot életciklus-horgokkal
  • A @handler dekoratőr megjelöli az aszinkron metódust, amely elvégzi a munkát
  • A kezelői aláírás egy szerződést követ:
    • Az első paraméter a csomópont beírt bemenete (itt: text: str)
    • A második paraméter egy WorkflowContext[T_Out], ahol T_Out a csomópont által ctx.send_message() kibocsátott adatok típusa (itt: str)
  • Egy kezelőn belül általában kiszámít egy eredményt, és továbbítja azt a lefelé irányuló csomópontokra a ctx.send_message(result)

3. lépés: A második végrehajtó létrehozása

Az egyszerű lépésekhez kihagyhatja az alosztályozást, és definiálhat egy aszinkron függvényt ugyanazzal az aláírási mintával (típusos bemenet + WorkflowContext), és díszítheti azt a @executor-el. Ez létrehoz egy teljesen működőképes csomópontot, amely egy folyamatba kapcsolható:

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

Főbb pontok:

  • A @executor dekorátor munkafolyamat-csomóponttá alakít át egy önálló aszinkron függvényt
  • A WorkflowContext paraméterezés két típussal történik:
    • T_Out = Never: ez a csomópont nem küld üzeneteket az alsóbb rétegbeli csomópontoknak
    • T_W_Out = str: ez a csomópont típusú munkafolyamat-kimenetet eredményez str
  • A terminálcsomópontok ctx.yield_output()-val kimeneteket nyújtanak a munkafolyamat eredményeinek biztosítására.
  • A munkafolyamat akkor fejeződik be, amikor tétlenné válik (nincs több teendő)

4. lépés: A munkafolyamat létrehozása

Csatlakoztassa a végrehajtókat a következővel WorkflowBuilder:

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

Főbb pontok:

  • add_edge() irányított kapcsolatokat hoz létre a végrehajtók között
  • set_start_executor() meghatározza a belépési pontot
  • build() véglegesíti a munkafolyamatot

5. lépés: A munkafolyamat futtatása streameléssel

Hajtsa végre a munkafolyamatot, és figyelje meg az eseményeket valós időben:

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

6. lépés: A kimenet ismertetése

A munkafolyamat futtatásakor a következő eseményeket fogja látni:

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

Alapvető fogalmak magyarázata

A végrehajtók definiálásának két módja

  1. Egyéni osztály (alosztályozás Executor): Akkor ideális, ha életciklus-horgokra vagy összetett állapotra van szüksége. Aszinkron metódus definiálása a @handler dekoratorral.
  2. Függvényalapú (@executor dekoratőr): Egyszerű lépésekhez a legjobb. Definiáljon egy önálló aszinkron függvényt ugyanazzal az aláírási mintával.

Mindkét módszer ugyanazt a kezelői aláírást használja:

  • Első paraméter: a csomópont beírt bemenete
  • Második paraméter: a WorkflowContext[T_Out, T_W_Out]

Munkafolyamat-környezettípusok

Az WorkflowContext általános típus határozza meg, hogy milyen adatfolyamok futnak a végrehajtók között:

  • WorkflowContext[T_Out] - Olyan csomópontokhoz használatos, amelyek T_Out típusú üzeneteket küldenek leágazó csomópontoknak a ctx.send_message()
  • WorkflowContext[T_Out, T_W_Out] - Olyan csomópontokhoz használatos, amelyek a ctx.yield_output()-en keresztül T_W_Out típusú munkafolyamat-kimenetet is szolgáltatnak
  • WorkflowContext típusparaméterek nélkül egyenértékű, ami azt jelenti, hogy WorkflowContext[Never, Never]ez a csomópont nem küld üzeneteket az alsóbb rétegbeli csomópontoknak, és nem eredményez munkafolyamat-kimenetet

Eseménytípusok

A streamelés végrehajtása során az alábbi eseménytípusokat fogja megfigyelni:

  • ExecutorInvokedEvent – Amikor egy végrehajtó elkezdi a feldolgozást
  • ExecutorCompletedEvent – Amikor egy végrehajtó befejezi a feldolgozást
  • WorkflowOutputEvent – A munkafolyamat végső eredményét tartalmazza

Python-munkafolyamat-szerkesztő minta

A WorkflowBuilder munkafolyamatok létrehozásához folyékony API-t biztosít:

  • add_edge(): Irányított kapcsolatokat hoz létre a végrehajtók között
  • set_start_executor(): Meghatározza a munkafolyamat belépési pontját
  • build(): Véglegesít és visszaad egy nem módosítható munkafolyamat-objektumot

Teljes példa

A teljes, futtatásra kész megvalósításért tekintse meg az Ügynök keretrendszer adattárában található mintát .

Ez a minta a következőket tartalmazza:

  • Teljes körű implementáció az összes importtal és dokumentációval
  • A munkafolyamat fogalmait ismertető további megjegyzések
  • A várt eredményeket megjelenítő mintakimenet

Következő lépések

Tudnivalók egy egyszerű, egyidejű munkafolyamat létrehozásáról

  • Last updated on