Bagikan melalui

Membuat Alur Kerja Berurutan Sederhana

Tutorial ini menunjukkan cara membuat alur kerja berurutan sederhana menggunakan Alur Kerja Kerangka Kerja Agen.

Alur kerja berurutan adalah fondasi membangun sistem agen AI yang kompleks. Tutorial ini menunjukkan cara membuat alur kerja dua langkah sederhana di mana setiap langkah memproses data dan meneruskannya ke langkah berikutnya.

Gambaran Umum

Dalam tutorial ini, Anda akan membuat alur kerja dengan dua pelaksana:

  1. Pengonversi Huruf Besar - Mengonversi teks input menjadi huruf besar
  2. Pelaksana Teks Terbalik - Membalikkan teks dan menghasilkan hasil akhir

Alur kerja menunjukkan konsep inti seperti:

  • Membuat pelaksana kustom dengan satu handler
  • Membuat pelaksana kustom dari fungsi
  • Menggunakan WorkflowBuilder untuk menyambungkan eksekutor dengan sisi
  • Memproses data melalui langkah-langkah berurutan
  • Mengamati eksekusi alur kerja melalui peristiwa

Konsep yang Tercakup

Prasyarat

Implementasi Langkah demi Langkah

Bagian berikut menunjukkan cara membuat alur kerja berurutan langkah demi langkah.

Langkah 1: Instal paket NuGet

Pertama, instal paket yang diperlukan untuk proyek .NET Anda:

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

Langkah 2: Tentukan Pelaksana Huruf Besar

Tentukan pelaksana yang mengonversi teks menjadi huruf besar:

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

Poin Utama:

  • Membuat fungsi yang mengambil string dan mengembalikan versi huruf besar
  • Gunakan BindExecutor() untuk membuat pelaksana dari fungsi

Langkah 3: Tentukan Pengolah Teks yang Dibalik

Tentukan pelaksana yang membalikkan teks:

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

Poin Utama:

  • Membuat kelas yang mewarisi dari Executor<TInput, TOutput>
  • Terapkan HandleAsync() untuk memproses input dan mengembalikan output

Langkah 4: Membangun dan Menyambungkan Alur Kerja

Sambungkan eksekutor menggunakan WorkflowBuilder

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

Poin Utama:

  • WorkflowBuilder konstruktor mengambil pelaksana awal
  • AddEdge() membuat koneksi yang diarahkan dari huruf besar ke terbalik
  • WithOutputFrom() menentukan eksekutor mana yang menghasilkan output alur kerja
  • Build() membuat alur kerja yang tidak dapat diubah

Langkah 5: Jalankan Alur Kerja

Jalankan alur kerja dan amati hasilnya:

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

Langkah 6: Memahami Output Alur Kerja

Saat menjalankan alur kerja, Anda akan melihat output seperti:

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

Input "Hello, World!" pertama kali diubah menjadi huruf kapital ("HELLO, WORLD!"), lalu dibalik ("!DLROW ,OLLEH").

Konsep Utama Dijelaskan

Antarmuka Pelaksana

Pelaksana dari fungsi:

  • Gunakan BindExecutor() untuk membuat pelaksana dari fungsi

Pelaksana mengimplementasikan Executor<TInput, TOutput>:

  • TInput: Jenis data yang diterima pelaksana ini
  • TOutput: Jenis data yang dihasilkan pelaksana ini
  • HandleAsync: Metode yang memproses input dan mengembalikan output

Pola Penyusun Alur Kerja .NET

WorkflowBuilder menyediakan API yang mudah digunakan untuk membangun alur kerja.

  • Konstruktor: Menetapkan pelaksana awal
  • AddEdge(): Membuat koneksi yang diarahkan antar eksekutor
  • WithOutputFrom(): Menentukan eksekutor mana yang menghasilkan output alur kerja
  • Build(): Membuat alur kerja akhir yang tidak dapat diubah

Jenis Peristiwa .NET

Selama eksekusi, Anda dapat mengamati jenis peristiwa ini:

  • ExecutorCompletedEvent - Ketika pengolah selesai memproses

Contoh .NET Lengkap

Untuk implementasi lengkap dan siap dijalankan, lihat sampel 01_ExecutorsAndEdges di repositori Kerangka Kerja Agen.

Sampel ini mencakup:

  • Implementasi penuh dengan semua menggunakan pernyataan dan struktur kelas
  • Komentar tambahan yang menjelaskan konsep alur kerja
  • Menyelesaikan penyiapan dan konfigurasi proyek

Gambaran Umum

Dalam tutorial ini, Anda akan membuat alur kerja dengan dua pelaksana:

  1. Pengubah Huruf Besar - Mengonversi teks input menjadi huruf besar
  2. Pelaksana Teks Terbalik - Membalikkan teks dan menghasilkan hasil akhir

Alur kerja menunjukkan konsep inti seperti:

  • Dua cara untuk menentukan unit kerja (simpul pelaksana):
    1. Kelas kustom yang subkelas Executor dengan metode asinkron yang ditandai dengan @handler
    2. Fungsi asinkron mandiri yang didekorasi dengan @executor
  • Menyambungkan pelaksana dengan WorkflowBuilder
  • Meneruskan data di antara langkah-langkah dengan ctx.send_message()
  • Menghasilkan output akhir dengan ctx.yield_output()
  • Streaming peristiwa untuk observabilitas waktu nyata

Konsep yang Tercakup

Prasyarat

  • Python 3.10 atau yang lebih baru
  • Paket Agent Framework Core Python terinstal: pip install agent-framework-core --pre
  • Tidak diperlukan layanan AI eksternal untuk contoh dasar ini

Implementasi Langkah demi Langkah

Bagian berikut menunjukkan cara membuat alur kerja berurutan langkah demi langkah.

Langkah 1: Impor modul yang Diperlukan

Pertama, impor modul yang diperlukan dari Agent Framework:

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

Langkah 2: Buat Pelaksana Pertama

Buat pelaksana yang mengonversi teks menjadi huruf besar dengan menerapkan pelaksana dengan metode handler:

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)

Poin Utama:

  • Subkelas Executor memungkinkan Anda menentukan node bernama dengan lifecycle hooks jika diperlukan
  • Dekorator @handler menandai metode asinkron yang melakukan pekerjaan
  • Tanda tangan handler mengikuti kontrak:
    • Parameter pertama adalah input yang ditik ke simpul ini (di sini: text: str)
    • Parameter kedua adalah WorkflowContext[T_Out], di mana T_Out adalah jenis data yang akan dipancarkan simpul ini melalui ctx.send_message() (di sini: str)
  • Dalam handler, Anda biasanya menghitung hasil dan meneruskannya ke simpul hilir menggunakan ctx.send_message(result)

Langkah 3: Buat Pelaksana Kedua

Untuk langkah-langkah sederhana, Anda dapat melewati subkelas dan menentukan fungsi asinkron dengan pola tanda tangan yang sama (input yang diketik + WorkflowContext) dan menghiasnya dengan @executor. Ini menciptakan node yang berfungsi penuh yang dapat dikabeli ke dalam alur:

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

Poin Utama:

  • Dekorator @executor mengubah fungsi asinkron mandiri menjadi simpul alur kerja
  • WorkflowContext diparameterkan dengan dua jenis:
    • T_Out = Never: simpul ini tidak mengirim pesan ke simpul hilir
    • T_W_Out = str: simpul ini menghasilkan output alur kerja dari jenis str
  • Node terminal menghasilkan keluaran menggunakan ctx.yield_output() untuk memberikan hasil alur kerja
  • Alur kerja selesai ketika diam (tidak ada lagi pekerjaan yang harus dilakukan)

Langkah 4: Bangun Alur Kerja

Sambungkan eksekutor menggunakan WorkflowBuilder

upper_case = UpperCase(id="upper_case_executor")

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

Poin Utama:

  • add_edge() membuat koneksi terarah antar pelaksana
  • set_start_executor() mendefinisikan titik masuk
  • build() menyelesaikan alur kerja

Langkah 5: Jalankan Alur Kerja dengan Streaming

Jalankan alur kerja dan amati peristiwa secara real time:

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

Langkah 6: Memahami Output

Saat menjalankan alur kerja, Anda akan melihat peristiwa seperti:

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

Konsep Utama Dijelaskan

Dua Cara untuk Menentukan Pelaksana

  1. Kelas kustom (subkelas Executor): Terbaik saat Anda membutuhkan kait siklus hidup atau status kompleks. Definisikan metode asinkron dengan dekorator @handler.
  2. Berbasis fungsi (@executor dekorator): Terbaik untuk langkah-langkah sederhana. Tentukan fungsi asinkron mandiri dengan pola tanda tangan yang sama.

Kedua pendekatan menggunakan tanda tangan handler yang sama:

  • Parameter pertama: input yang ditik ke simpul ini
  • Parameter kedua: a WorkflowContext[T_Out, T_W_Out]

Jenis Konteks Alur Kerja

WorkflowContext Jenis generik menentukan aliran data apa di antara pelaksana:

  • WorkflowContext[T_Out] - Digunakan untuk simpul yang mengirim pesan jenis T_Out ke simpul hilir melalui ctx.send_message()
  • WorkflowContext[T_Out, T_W_Out] - Digunakan untuk simpul yang juga menghasilkan output alur kerja jenis T_W_Out melalui ctx.yield_output()
  • WorkflowContext tanpa parameter tipe adalah setara dengan WorkflowContext[Never, Never], yang berarti simpul ini tidak mengirim pesan ke simpul hilir ataupun menghasilkan keluaran alur kerja.

Jenis Peristiwa

Selama eksekusi streaming, Anda akan mengamati jenis peristiwa ini:

  • ExecutorInvokedEvent - Ketika pelaksana mulai memproses
  • ExecutorCompletedEvent - Ketika pengolah selesai memproses
  • WorkflowOutputEvent - Berisi hasil alur kerja akhir

Pola Penyusun Alur Kerja Python

WorkflowBuilder menyediakan API yang mudah digunakan untuk membangun alur kerja.

  • add_edge(): Membuat koneksi yang diarahkan antar pelaksana
  • set_start_executor(): Menentukan titik entri alur kerja
  • build(): Menyelesaikan dan mengembalikan objek alur kerja yang tidak dapat diubah

Contoh Lengkap

Untuk implementasi lengkap dan siap dijalankan, lihat sampel di repositori Kerangka Kerja Agen.

Sampel ini mencakup:

  • Implementasi penuh dengan semua impor dan dokumentasi
  • Komentar tambahan yang menjelaskan konsep alur kerja
  • Output sampel memperlihatkan hasil yang diharapkan

Langkah Selanjutnya

Pelajari tentang membuat alur kerja bersamaan sederhana

  • Last updated on