Freigeben über

Microsoft Agent Framework Workflows – Status

Dieses Dokument enthält eine Übersicht über den Status im Microsoft Agent Framework-Workflowsystem.

Überblick

Status ermöglicht es mehreren Executoren innerhalb eines Workflows, auf allgemeine Daten zuzugreifen und diese zu ändern. Dieses Feature ist für Szenarien unerlässlich, in denen unterschiedliche Teile des Workflows Informationen freigeben müssen, bei denen direkte Nachrichtenübergaben nicht machbar oder effizient sind.

In den Zustand schreiben

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)

Zugriff auf den 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()))

Zustandsisolation

In realen Anwendungen ist die ordnungsgemäße Verwaltung des Zustands bei der Behandlung mehrerer Aufgaben oder Anforderungen von entscheidender Bedeutung. Ohne ordnungsgemäße Isolierung kann der gemeinsame Zustand zwischen verschiedenen Workflowausführungen zu unerwartetem Verhalten, Datenbeschädigungen und Rennbedingungen führen. In diesem Abschnitt wird erläutert, wie Sie die Zustandsisolation innerhalb von Microsoft Agent Framework-Workflows sicherstellen und Einblicke in bewährte Methoden und häufige Fallstricke bieten.

Veränderbare Workflow-Generatoren im Vergleich zu unveränderlichen Workflows

Workflows werden von Workflow-Generatoren erstellt. Workflow-Ersteller werden in der Regel als änderbar angesehen, wobei man Konfigurationen wie den Start-Executor oder andere Einstellungen nach der Erstellung des Erstellers oder sogar nachdem ein Workflow gebaut wurde, hinzufügen oder ändern kann. Andererseits sind Workflows unveränderlich, wenn ein Workflow erstellt wurde, kann er nicht mehr geändert werden (keine öffentliche API zum Ändern eines Workflows).

Diese Unterscheidung ist wichtig, da sie sich darauf auswirkt, wie der Zustand in verschiedenen Workflowausführungen verwaltet wird. Es ist nicht ratsam, eine einzelne Workflowinstanz für mehrere Aufgaben oder Anforderungen wiederzuverwenden, da dies zu unbeabsichtigter Zustandsteilung führen kann. Stattdessen wird empfohlen, eine neue Workflowinstanz aus dem Generator für jede Aufgabe oder Anforderung zu erstellen, um eine ordnungsgemäße Zustandsisolation und Threadsicherheit sicherzustellen.

Sicherstellen der Zustandsisolation mit Hilfsmethoden

Wenn Executorinstanzen einmal erstellt und für mehrere Workflowbuilds freigegeben werden, wird ihr interner Zustand für alle Workflowausführungen freigegeben. Dies kann zu Problemen führen, wenn ein Executor den veränderbaren Zustand enthält, der pro Workflow isoliert werden soll. Um eine ordnungsgemäße Zustandsisolation und Threadsicherheit sicherzustellen, schließen Sie die Instanziierung des Executors und den Workflow in einer Hilfsmethode ein, sodass jeder Aufruf frische, unabhängige Instanzen erzeugt.

Demnächst...

Nicht isoliertes Beispiel (gemeinsamer Zustand):

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

Isoliertes Beispiel (Hilfsmethode):

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

Tipp

Um eine ordnungsgemäße Zustandsisolation und Threadsicherheit sicherzustellen, stellen Sie außerdem sicher, dass in der Hilfsmethode erstellte Executorinstanzen keinen externen veränderbaren Zustand freigeben.

Agentstatusverwaltung

Der Agentkontext wird über Agent-Threads verwaltet. Standardmäßig erhält jeder Agent in einem Workflow seinen eigenen Thread, es sei denn, der Agent wird von einem benutzerdefinierten Executor verwaltet. Weitere Informationen finden Sie unter "Arbeiten mit Agents".

Agentthreads werden über Workflowausführungen hinweg beibehalten. Dies bedeutet, dass Inhalte, die vom Agent generiert werden, in nachfolgenden Ausführungen derselben Workflowinstanz verfügbar sind, wenn ein Agent in der ersten Ausführung eines Workflows aufgerufen wird. Dies kann zwar nützlich sein, um die Kontinuität innerhalb einer einzelnen Aufgabe aufrechtzuerhalten, kann aber auch zu unbeabsichtigter Zustandsfreigabe führen, wenn dieselbe Workflowinstanz für verschiedene Aufgaben oder Anforderungen wiederverwendet wird. Um sicherzustellen, dass jede Aufgabe einen isolierten Agentstatus aufweist, erstellen Sie den Agent und den Workflow innerhalb einer Hilfsmethode, damit jeder Aufruf neue Agentinstanzen mit eigenen Threads erzeugt.

Demnächst...

Nicht isoliertes Beispiel (geteilter Agentenstatus):

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

Isoliertes Beispiel (Hilfsmethode):

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

Zusammenfassung

Die Zustandsisolation in Microsoft Agent Framework-Workflows kann effektiv verwaltet werden, indem die Ausführung und die Instanziierung von Agenten zusammen mit dem Erstellen von Workflows in Hilfsmethoden eingebettet werden. Wenn Sie die Hilfsmethode jedes Mal aufrufen, wenn Sie einen neuen Workflow benötigen, stellen Sie sicher, dass jede Instanz über einen neuen, unabhängigen Zustand verfügt, und vermeiden Sie die unbeabsichtigte Zustandsfreigabe zwischen verschiedenen Workflowausführungen.

Nächste Schritte

  • Last updated on