Pracovní postupy rozhraní Microsoft Agent Framework – stav

Tento dokument obsahuje přehled stavu v systému pracovních postupů rozhraní Microsoft Agent Framework.

Přehled

Stav umožňuje více exekutorům v rámci pracovního postupu přístup k běžným datům a jejich úpravě. Tato funkce je nezbytná pro scénáře, kdy různé části pracovního postupu potřebují sdílet informace, kde přímé předávání zpráv není proveditelné nebo efektivní.

Zápis do stavu

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)

Přístup ke stavu

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

Izolace stavu

V reálných aplikacích je při zpracování více úloh nebo požadavků důležitá správná správa stavu. Bez správné izolace může sdílený stav mezi různými spuštěními pracovního postupu vést k neočekávanému chování, poškození dat nebo stavovým podmínkám závodu. Tato část vysvětluje, jak zajistit izolaci stavu v rámci pracovních postupů rozhraní Microsoft Agent Framework, a poskytuje přehledy o osvědčených postupech a běžných nástrahách.

Tvůrce proměnlivých pracovních postupů vs. Neměnné pracovní postupy

Pracovní postupy vytvářejí tvůrci pracovních postupů. Tvůrci pracovních postupů se obecně považují za proměnlivé, a je možné přidávat nebo upravovat spouštěcí mechanismus či jiné konfigurace po vytvoření tvůrce nebo dokonce i po vytvoření pracovního postupu. Pracovní postupy jsou naopak neměnné v tom, že po vytvoření pracovního postupu není možné ho změnit (žádné veřejné rozhraní API pro úpravu pracovního postupu).

Tento rozdíl je důležitý, protože ovlivňuje způsob správy stavu napříč různými prováděními pracovních postupů. Nedoporučuje se opakovaně používat jednu instanci pracovního postupu pro více úkolů nebo požadavků, protože to může vést k nezamýšleným sdílením stavu. Místo toho doporučujeme vytvořit novou instanci pracovního postupu od tvůrce pro každý úkol nebo požadavek, aby se zajistila správná izolace stavu a bezpečnost vláken.

Zajištění izolace stavu pomocí pomocných metod

Když jsou instance exekutoru vytvořeny jednou a sdíleny napříč více sestaveními pracovního postupu, jejich vnitřní stav se sdílí napříč všemi prováděními pracovních postupů. To může vést k problémům, pokud výkonný modul obsahuje proměnlivý stav, který by měl být izolován pro každý pracovní postup. Chcete-li zajistit správnou izolaci stavu a zabezpečení vlákna, zabalte instanci exekutoru a pracovní postup sestavení uvnitř pomocné metody tak, aby každé volání vytvořilo čerstvé, nezávislé instance.

Již brzy...

Neizolovaný příklad (sdílený stav):

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

Izolovaný příklad (pomocná metoda):

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

Návod

Chcete-li zajistit správnou izolaci stavu a zabezpečení vlákna, ujistěte se také, že instance exekutorů vytvořené uvnitř pomocné funkce nesdílí externí proměnlivý stav.

Správa stavu agenta

Kontext agenta se spravuje prostřednictvím vláken agenta. Ve výchozím nastavení každý agent v pracovním postupu získá vlastní vlákno, pokud agenta nespravuje vlastní exekutor. Další informace najdete v tématu Práce s agenty.

Vlákna agenta jsou uchovávána mezi jednotlivými spuštěními pracovního postupu. To znamená, že pokud je agent vyvolán při prvním spuštění pracovního postupu, bude obsah vygenerovaný agentem dostupný v následných spuštěních stejné instance pracovního postupu. I když to může být užitečné pro zachování kontinuity v rámci jednoho úkolu, může to také vést k nezamýšlenému sdílení stavu, pokud je stejná instance pracovního postupu znovu použita pro různé úlohy nebo požadavky. Aby se zajistilo, že má každá úloha stav izolovaného agenta, zabalte vytvoření agenta a pracovního postupu do pomocné metody tak, aby každé volání vytvářelo nové instance agenta s vlastními vlákny.

Již brzy...

Příklad neizolovaný (stav sdíleného agenta):

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

Izolovaný příklad (pomocná metoda):

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

Shrnutí

Izolaci stavu v pracovních postupech rozhraní Microsoft Agent Framework lze efektivně spravovat zabalením exekutoru a instance agenta spolu s vytvářením pracovních postupů uvnitř pomocných metod. Voláním pomocné metody pokaždé, když potřebujete nový pracovní postup, zajistíte, aby každá instance měla aktuální, nezávislý stav a vyhnula se nezamýšlenému sdílení stavu mezi různými spuštěními pracovního postupu.

Další kroky

• Naučte se vytvářet kontrolní body a pokračovat v nich.
• Zjistěte, jak monitorovat pracovní postupy.
• Naučte se vizualizovat pracovní postupy.