Mendaftarkan Pabrik ke Penyusun Alur Kerja

Hingga saat ini, kami telah membuat instans pelaksana dan meneruskannya langsung ke WorkflowBuilder. Pendekatan ini berfungsi dengan baik untuk skenario sederhana di mana Anda hanya memerlukan satu instans alur kerja. Namun, dalam kasus yang lebih kompleks, Anda mungkin ingin membuat beberapa instans terisolasi dari alur kerja yang sama. Untuk mendukung ini, setiap instans alur kerja harus menerima sekumpulan instans pelaksananya sendiri. Menggunakan kembali pelaksana yang sama akan menyebabkan status internal mereka dibagikan di seluruh alur kerja, menghasilkan efek samping yang tidak diinginkan. Untuk menghindari hal ini, Anda dapat mendaftarkan "executor factories" dengan WorkflowBuilder, memastikan bahwa instans pelaksana baru dibuat untuk setiap instans alur kerja.

Mendaftarkan Pabrik ke Penyusun Alur Kerja

Segera datang...

Untuk mendaftarkan pabrik pelaksana ke WorkflowBuilder, Anda dapat menggunakan metode register_executor. Metode ini mengambil dua parameter: fungsi pabrik yang membuat instans pelaksana (jenis Executor atau turunan Executor) dan nama pabrik yang akan digunakan dalam konfigurasi alur kerja.

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."""
        result = text.upper()

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

class Accumulate(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)
        # Executor internal state that should not be shared among different workflow instances.
        self._text_length = 0

    @handler
    async def accumulate(self, text: str, ctx: WorkflowContext) -> None:
        """Accumulate the length of the input text and log it."""
        self._text_length += len(text)
        print(f"Accumulated text length: {self._text_length}")

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[str]) -> None:
    """Reverse the input string and send it downstream."""
    result = text[::-1]

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

workflow_builder = (
    WorkflowBuilder()
    .register_executor(
        factory_func=lambda: UpperCase(id="UpperCaseExecutor"),
        name="UpperCase",
    )
    .register_executor(
        factory_func=lambda: Accumulate(id="AccumulateExecutor"),
        name="Accumulate",
    )
    .register_executor(
        factory_func=lambda: reverse_text,
        name="ReverseText",
    )
    # Use the factory name to configure the workflow
    .add_fan_out_edges("UpperCase", ["Accumulate", "ReverseText"])
    .set_start_executor("UpperCase")
)

Membangun alur kerja menggunakan penyusun

# Build the workflow using the builder
workflow_a = workflow_builder.build()
await workflow_a.run("hello world")
await workflow_a.run("hello world")

Output yang diharapkan:

Accumulated text length: 22

Sekarang mari kita buat instans alur kerja lain dan jalankan. Pelaksana Accumulate harus memiliki status internal sendiri dan tidak berbagi status dengan instans alur kerja pertama.

# Build another workflow using the builder
# This workflow will have its own set of executors, including a new instance of the Accumulate executor.
workflow_b = workflow_builder.build()
await workflow_b.run("hello world")

Output yang diharapkan:

Accumulated text length: 11

Untuk mendaftarkan pabrik agen ke WorkflowBuilder, Anda dapat menggunakan metode register_agent. Metode ini mengambil dua parameter: fungsi pabrik yang membuat instans agen (jenis yang mengimplementasikan AgentProtocol) dan nama pabrik yang akan digunakan dalam konfigurasi alur kerja.

def create_agent() -> ChatAgent:
    """Factory function to create a Writer agent."""
    return AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
        instructions=("You are a helpful assistant.",),
        name="assistant",
    )

workflow_builder = (
    WorkflowBuilder()
    .register_agent(
        factory_func=create_agent,
        name="Assistant",
    )
    # Register other executors or agents as needed and configure the workflow
    ...
)

# Build the workflow using the builder
workflow = workflow_builder.build()

Setiap kali instans alur kerja baru dibuat, agen di dalam alur kerja akan menjadi instans baru yang dihasilkan oleh fungsi factory, dan akan memperoleh instans utas baru.

Isolasi Status Alur Kerja

Untuk mempelajari selengkapnya tentang isolasi status alur kerja, lihat dokumentasi Isolasi Status Alur Kerja .