Alur Kerja Kerangka Kerja Agen Microsoft - Status

Dokumen ini menyediakan gambaran umum Status dalam sistem Microsoft Agent Framework Workflow.

Gambaran Umum

Status memungkinkan beberapa pelaksana dalam alur kerja untuk mengakses dan memodifikasi data umum. Fitur ini sangat penting untuk skenario di mana berbagai bagian alur kerja perlu berbagi informasi di mana pengiriman pesan langsung tidak layak atau efisien.

Menulis ke Status

using Microsoft.Agents.AI.Workflows;

internal sealed partial class FileReadExecutor(): Executor("FileReadExecutor")
{
    /// <summary>
    /// Reads a file and stores its content in a shared state.
    /// </summary>
    /// <param name="message">The path to the embedded resource file.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The ID of the shared state where the file content is stored.</returns>
    [MessageHandler]
    private async ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        // Read file content from embedded resource
        string fileContent = File.ReadAllText(message);
        // Store file content in a shared state for access by other executors
        string fileID = Guid.NewGuid().ToString();
        await context.QueueStateUpdateAsync<string>(fileID, fileContent, scopeName: "FileContent");

        return fileID;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class FileReadExecutor(Executor):

    @handler
    async def handle(self, file_path: str, ctx: WorkflowContext[str]):
        # Read file content from embedded resource
        with open(file_path, 'r') as file:
            file_content = file.read()
        # Store file content in state for access by other executors
        file_id = str(uuid.uuid4())
        ctx.set_state(file_id, file_content)

        await ctx.send_message(file_id)

Mengakses Status

using Microsoft.Agents.AI.Workflows;

internal sealed partial class WordCountingExecutor() : Executor("WordCountingExecutor")
{
    /// <summary>
    /// Counts the number of words in the file content stored in a shared state.
    /// </summary>
    /// <param name="message">The ID of the shared state containing the file content.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The number of words in the file content.</returns>
    [MessageHandler]
    private async ValueTask<int> HandleAsync(string message, IWorkflowContext context)
    {
        // Retrieve the file content from the shared state
        var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent")
            ?? throw new InvalidOperationException("File content state not found");

        return fileContent.Split([' ', '\n', '\r'], StringSplitOptions.RemoveEmptyEntries).Length;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class WordCountingExecutor(Executor):

    @handler
    async def handle(self, file_id: str, ctx: WorkflowContext[int]):
        # Retrieve the file content from state
        file_content = ctx.get_state(file_id)
        if file_content is None:
            raise ValueError("File content state not found")

        await ctx.send_message(len(file_content.split()))

Isolasi Negara

Dalam aplikasi dunia nyata, mengelola status dengan benar sangat penting saat menangani beberapa tugas atau permintaan. Tanpa isolasi yang tepat, status bersama antara eksekusi alur kerja yang berbeda dapat menyebabkan perilaku tak terduga, kerusakan data, dan kondisi ras. Bagian ini menjelaskan cara memastikan isolasi status dalam Alur Kerja Kerangka Kerja Agen Microsoft, memberikan wawasan tentang praktik terbaik dan jebakan umum.

Penyusun Alur Kerja yang Dapat Diubah vs Alur Kerja yang Tidak Dapat Diubah

Alur kerja dibuat oleh penyusun alur kerja. Penyusun alur kerja umumnya dianggap dapat diubah-ubah, di mana kita dapat menambahkan, memodifikasi pelaksana mula atau konfigurasi lain setelah penyusun dibuat atau bahkan setelah alur kerja dibangun. Di sisi lain, alur kerja tidak dapat diubah karena setelah alur kerja dibuat, alur kerja tidak dapat dimodifikasi (tidak ada API publik untuk memodifikasi alur kerja).

Perbedaan ini penting karena memengaruhi bagaimana status dikelola di berbagai eksekusi alur kerja. Tidak disarankan untuk menggunakan kembali satu instans alur kerja untuk beberapa tugas atau permintaan, karena ini dapat menyebabkan berbagi status yang tidak diinginkan. Sebagai gantinya, disarankan untuk membuat instans alur kerja baru dari penyusun untuk setiap tugas atau permintaan untuk memastikan isolasi status dan keamanan utas yang tepat.

Memastikan Isolasi Status dengan Metode Bantu

Saat instans eksekutor dibuat satu kali dan dibagikan di beberapa kompilasi alur kerja, status internalnya dibagikan di semua eksekusi alur kerja. Ini dapat menyebabkan masalah jika eksekutor berisi status yang dapat diubah yang harus diisolasi per alur kerja. Untuk memastikan isolasi status dan keamanan thread yang tepat, bungkus pembuatan executor dan alur kerja di dalam metode pembantu sehingga setiap panggilan menghasilkan instans baru yang mandiri.

Segera datang...

Contoh yang tidak terisolasi (keadaan bersama):

executor_a = CustomExecutorA()
executor_b = CustomExecutorB()

# executor_a and executor_b are shared across all workflows built from this builder
workflow_builder = WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b)

workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()
# workflow_a and workflow_b share the same executor instances and their mutable state

Contoh terisolasi (metode pembantu):

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated state.

    Each call produces independent executor instances, ensuring no state
    leaks between workflow runs.
    """
    executor_a = CustomExecutorA()
    executor_b = CustomExecutorB()

    return WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b).build()

# Each workflow has its own executor instances with independent state
workflow_a = create_workflow()
workflow_b = create_workflow()

Petunjuk / Saran

Untuk memastikan isolasi status dan keamanan thread yang tepat, pastikan juga bahwa instance eksekutor yang dibuat di dalam metode bantu tidak berbagi keadaan eksternal yang dapat diubah.

Manajemen Status Agen

Konteks agen dikelola melalui thread agen. Secara default, setiap agen dalam alur kerja akan mendapatkan utasnya sendiri kecuali agen dikelola oleh pelaksana kustom. Untuk informasi selengkapnya, lihat Bekerja dengan Agen.

Utas agen dipertahankan di seluruh pelaksanaan alur kerja. Ini berarti bahwa jika agen dipanggil dalam eksekusi pertama alur kerja, konten yang dihasilkan oleh agen akan tersedia dalam eksekusi berikutnya dari instans alur kerja yang sama. Meskipun ini dapat berguna untuk mempertahankan kelangsungan dalam satu tugas, ini juga dapat menyebabkan berbagi status yang tidak diinginkan jika instans alur kerja yang sama digunakan kembali untuk tugas atau permintaan yang berbeda. Untuk memastikan setiap tugas memiliki status agen terisolasi, bungkus agen dan pembuatan alur kerja di dalam fungsi pembantu sehingga setiap panggilan menghasilkan instans agen baru dengan thread mereka sendiri.

Segera datang...

Contoh yang tidak terisolasi (status agen bersama):

writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

# writer_agent and reviewer_agent are shared across all workflows
workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

Contoh terisolasi (metode pembantu):

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated agent state.

    Each call produces new agent instances with their own threads,
    ensuring no conversation history leaks between workflow runs.
    """
    writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
        name="writer_agent",
    )
    reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
        instructions=(
            "You are an excellent content reviewer."
            "Provide actionable feedback to the writer about the provided content."
            "Provide the feedback in the most concise manner possible."
        ),
        name="reviewer_agent",
    )

    return WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

# Each workflow has its own agent instances and threads
workflow_a = create_workflow()
workflow_b = create_workflow()

Ringkasan

Isolasi status dalam Microsoft Agent Framework Workflows dapat dikelola secara efektif dengan mengemas pelaksanaan dan inisiasi agen serta pembangunan alur kerja di dalam metode pembantu. Dengan memanggil metode pembantu setiap kali Anda memerlukan alur kerja baru, Anda memastikan setiap instans memiliki status segar dan independen dan menghindari berbagi status yang tidak diinginkan antara eksekusi alur kerja yang berbeda.

Langkah Selanjutnya

• Pelajari cara membuat titik pemeriksaan dan melanjutkan dari titik pemeriksaan.
• Pelajari cara memantau alur kerja.
• Pelajari cara memvisualisasikan alur kerja.