Microsoft Agent Framework-munkafolyamatok – Állapot

Ez a dokumentum áttekintést nyújt a Microsoft Agent Framework munkafolyamat-rendszer állapotáról .

Áttekintés

Az állapot lehetővé teszi, hogy a munkafolyamaton belül több végrehajtó is hozzáférhessen és módosíthassa a közös adatokat. Ez a funkció elengedhetetlen olyan helyzetekben, ahol a munkafolyamat különböző részeinek olyan információkat kell megosztaniuk, ahol a közvetlen üzenetátadás nem megvalósítható vagy hatékony.

Állapot írása

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)

Az állapot elérése

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

Állapotelkülönítés

A valós alkalmazásokban az állapot megfelelő kezelése kritikus fontosságú több feladat vagy kérés kezelésekor. Megfelelő elkülönítés nélkül a különböző munkafolyamat-végrehajtások közötti megosztott állapot váratlan viselkedéshez, adatsérüléshez és versenyfeltételekhez vezethet. Ez a szakasz bemutatja, hogyan biztosítható az állapotelkülönítés a Microsoft Agent Framework-munkafolyamatokon belül, és hogyan nyújt betekintést az ajánlott eljárásokba és a gyakori buktatókba.

Mutable Workflow Builders és nem módosítható munkafolyamatok

A munkafolyamatokat munkafolyamat-készítők hozzák létre. A munkafolyamat-szerkesztőket általában mutable-nak tekintik, ahol a szerkesztő létrehozása után vagy akár a munkafolyamat létrehozása után is hozzáadhat, módosíthat indítási végrehajtót vagy más konfigurációkat. A munkafolyamatok viszont nem módosíthatók, mivel a munkafolyamatok létrehozása után nem módosíthatók (a munkafolyamatok módosításához nincs nyilvános API).

Ez a megkülönböztetés azért fontos, mert hatással van az állapot különböző munkafolyamat-végrehajtások közötti felügyeletére. Nem ajánlott egyetlen munkafolyamat-példányt újra felhasználni több tevékenységhez vagy kéréshez, mivel ez nem kívánt állapotmegosztáshoz vezethet. Ehelyett javasoljuk, hogy minden feladathoz vagy kéréshez hozzon létre egy új munkafolyamat-példányt a szerkesztőből a megfelelő állapotelkülönítés és a szálbiztonság biztosítása érdekében.

Állapotelkülönítés biztosítása segédfüggvényekkel

Ha a végrehajtópéldányok létrehozása és megosztása több munkafolyamat-buildben történik, a rendszer az összes munkafolyamat-végrehajtásban megosztja a belső állapotukat. Ez problémákhoz vezethet, ha egy végrehajtó olyan mutable állapotot tartalmaz, amelyet munkafolyamatonként el kell különíteni. A megfelelő állapotelkülönítés és szálbiztonság biztosítása érdekében helyezze el minden végrehajtói példányosítást és munkafolyamat-építést egy segédmetódusba, hogy minden hívás friss, független példányokat állítson elő.

Hamarosan...

Nem izolált példa (megosztott állapot):

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

Izolált példa (segédmetódus):

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

Jótanács

A megfelelő állapotelkülönítés és szálbiztonság érdekében győződjön meg arról is, hogy a segédmetóduson belül létrehozott végrehajtópéldányok nem osztoznak külső, módosítható állapoton.

Ügynökállapot-kezelés

Az ügynökkörnyezet kezelése ügynökszálakon keresztül történik. Alapértelmezés szerint a munkafolyamat minden ügynöke saját szálat kap, kivéve, ha az ügynököt egyéni végrehajtó kezeli. További információt az Ügynökök használata című témakörben talál.

Az ügynöki szálak fennmaradnak a munkafolyamat-futtatások során. Ez azt jelenti, hogy ha egy ügynököt egy munkafolyamat első futtatásakor hív meg, az ügynök által létrehozott tartalom elérhető lesz ugyanannak a munkafolyamat-példánynak a későbbi futtatásaiban. Ez hasznos lehet egy tevékenység folytonosságának fenntartásához, de nem kívánt állapotmegosztáshoz is vezethet, ha ugyanazt a munkafolyamat-példányt újra felhasználják a különböző tevékenységekhez vagy kérésekhez. Annak biztosítása érdekében, hogy minden feladat izolált ügynökállapottal rendelkezzen, helyezze be az ügynök és a munkafolyamat létrehozását egy segédmetódusba, így minden hívás új ügynökpéldányokat hoz létre a saját szálaikkal.

Hamarosan...

Nem izolált példa (megosztott ügynök állapota):

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

Izolált példa (segédmetódus):

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

Összefoglalás

A Microsoft Agent Framework munkafolyamatokban az állapot elkülönítése hatékonyan kezelhető a végrehajtó és az ügynök példányosításának bekapcsolásával, valamint a munkafolyamatok segédmetódusokon belüli létrehozásával. Ha minden alkalommal meghívja a segédmetódust, amikor új munkafolyamatra van szüksége, győződjön meg arról, hogy minden példány friss, független állapotú, és elkerülheti a különböző munkafolyamat-végrehajtások közötti nem szándékos állapotmegosztást.

Következő lépések

• Megtudhatja, hogyan hozhat létre ellenőrzőpontokat, és hogyan folytathatja azokból.
• Megtudhatja, hogyan figyelheti a munkafolyamatokat.
• Megtudhatja, hogyan jelenítheti meg a munkafolyamatokat.