Microsoft Agent Framework-munkafolyamatok – Állapot

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

Overview

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 láthatósága és hatókör viselkedése

QueueStateUpdateAsync és ReadStateAsync mindkettő hatókörrel tisztában van:

• Ha a(z) scopeName értéke null, a rendszer a végrehajtó saját alapértelmezett hatókörét használja.
• Ha scopeName be van állítva (például), az érték egy megosztott hatókörbe lesz írva, "SharedResponse"amelyet bármely végrehajtó elolvashat, ha ugyanazt a hatókörnevet használja.

A láthatóság időzítése szupersztep-szabályokat követ:

• A hívásokat kezdeményező QueueStateUpdateAsync végrehajtó azonnal beolvassa a frissített értéket ugyanabban a kezelőben.
• A többi végrehajtó a következő szupersteptől kezdve látja ezt a frissítést.

Ha meg szeretné osztani az állapotot a végrehajtók között, használja ugyanazt a nem null hatókörnevet írási és olvasási hívásokban is:

private const string SharedScope = "SharedResponse";

await context.QueueStateUpdateAsync("Response", blanketResponse, scopeName: SharedScope, cancellationToken);

var finalResponse = await context.ReadStateAsync<string>("Response", scopeName: SharedScope, cancellationToken);

WorkflowContext.set_state() és WorkflowContext.get_state() olyan munkafolyamat-állapoton működjön, amely a munkafolyamat végrehajtása során elérhető az alsóbb rétegbeli végrehajtók számára.

Azonos érték írásához és olvasásához használjon konzisztens kulcsokat a végrehajtók között:

ctx.set_state("response", blanket_response)
final_response = ctx.get_state("response")

Állapot írása

using Microsoft.Agents.AI.Workflows;

internal sealed class FileReadExecutor() : Executor<string, string>("FileReadExecutor")
{
    public override async ValueTask<string> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // 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("N");
        await context.QueueStateUpdateAsync(fileID, fileContent, scopeName: "FileContent", cancellationToken);

        return fileID;
    }
}

import uuid

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)

fileRead := workflow.NewExecutor("FileReadExecutor", func(ctx *workflow.Context, path string) (string, error) {
    fileContent, err := os.ReadFile(path)
    if err != nil {
        return "", err
    }

    fileID := uuid.NewString()
    if err := ctx.QueueStateUpdate(fileID, "FileContent", string(fileContent)); err != nil {
        return "", err
    }

    return fileID, nil
}).Bind()

Az állapot elérése

using Microsoft.Agents.AI.Workflows;

internal sealed class WordCountingExecutor() : Executor<string, int>("WordCountingExecutor")
{
    public override async ValueTask<int> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // Retrieve the file content from the shared state
        var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent", cancellationToken)
            ?? 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()))

fileProcess := workflow.NewExecutor("FileProcessExecutor", func(ctx *workflow.Context, fileID string) (FileSummary, error) {
    value, err := ctx.ReadState(fileID, "FileContent")
    if err != nil {
        return FileSummary{}, err
    }

    fileContent, ok := value.(string)
    if !ok {
        return FileSummary{}, fmt.Errorf("file content %q was not found", fileID)
    }

    return FileSummary{
        FileID:  fileID,
        Summary: summarize(fileContent),
    }, nil
}).Bind()

Futtatókörnyezet munkafolyamat-hatókörrel – kwargs

Azoknál az értékeknél, amelyeknek nem szabad megosztott munkafolyamat-állapottá válnia, továbbítsa azokat az ügynökök és eszközök felé workflow.run() formájában function_invocation_kwargs= vagy client_kwargs=.

• Ha a legfelső szintű kulcsok egyike sem egyezik meg egy végrehajtóazonosítóval, a leképezés globálisként lesz kezelve, és minden egyező ügynök végrehajtója ugyanazt a diktált értéket kapja.
• Ha egy vagy több legfelső szintű kulcs egyezik a végrehajtóazonosítókkal, a rendszer a teljes leképezést végrehajtónkénti célzásként kezeli, és minden végrehajtó csak a saját bejegyzését kapja meg.
• Ugyanezek a globális és a célzott szabályok vonatkoznak mind a function_invocation_kwargs és a client_kwargs egységekre.

await workflow.run(
    "Create the report",
    function_invocation_kwargs={
        "tenant": "contoso",
        "request_id": "req-42",
    },
)

await workflow.run(
    "Create the report",
    function_invocation_kwargs={
        "researcher": {
            "db_config": {"connection_string": "..."},
        },
        "writer": {
            "user_preferences": {"format": "markdown"},
        },
    },
)

Tip

A végrehajtó által célzott kwargok munkafolyamat-végrehajtó azonosítókat használnak. Burkolt ügynökök esetén alapértelmezés szerint ez az ügynök neve, vagy az explicit id, amelyet AgentExecutor(...) átad.

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

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

executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
    return input, nil
}).Bind()
executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
    return input, nil
}).Bind()

builder := workflow.NewBuilder(executorA).AddEdge(executorA, executorB)

workflowA, err := builder.Build()
if err != nil {
    return err
}
workflowB, err := builder.Build()
if err != nil {
    return err
}

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

func createWorkflow() (*workflow.Workflow, error) {
    executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
        return input, nil
    }).Bind()
    executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
        return input, nil
    }).Bind()

    return workflow.NewBuilder(executorA).AddEdge(executorA, executorB).Build()
}

workflowA, err := createWorkflow()
if err != nil {
    return err
}
workflowB, err := createWorkflow()
if err != nil {
    return err
}

Tip

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.

Megosztott végrehajtók alaphelyzetbe állítása

Ha végrehajtópéldányokat kell megosztania a munkafolyamat-futtatások között – például ha a végrehajtó felépítése költséges, vagy ha egy munkafolyamat ügynökként van közzétéve – az állapotalapú végrehajtóknak implementálniuk IResettableExecutorkell. Ez a felület egy ResetAsync() olyan módszert biztosít, amelyet a munkafolyamat-futtatókörnyezet automatikusan hív a futtatások között az elavult állapot törlése érdekében.

A IResettableExecutor megvalósításának időpontjával és módjával kapcsolatos részletekért lásd: Visszaállítható végrehajtók.

Megosztott végrehajtók alaphelyzetbe állítása

A Go végrehajtó kötései visszaállíthatják a megosztott végrehajtó állapotát a ResetFunc használatával. A BindNewExecutorFunc használatával létrehozott kötések minden munkafolyamat-munkamenethez új végrehajtót hoznak létre, és nincs szükségük visszaállítási horogra.

A részletekért lásd: Visszaállítható executorok.

Ü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 = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    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 = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    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 = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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 = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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()

A Go-ügynök állapotának kezelése a következőn keresztül történik: agent.Session. A munkafolyamatokban lévő ügynökök megőrzik a munkamenetüket a fordulók között, hacsak nem jön létre új ügynök, munkafolyamat vagy munkamenet.

session, err := writerAgent.CreateSession(ctx)
if err != nil {
    return err
}

_, err = writerAgent.RunText(ctx, "first request", agent.WithSession(session)).Collect()
if err != nil {
    return err
}

_, err = writerAgent.RunText(ctx, "follow-up request", agent.WithSession(session)).Collect()
if err != nil {
    return err
}

A(z) agentworkflow.New használatával létrehozott hosztolt ügynökvégrehajtók a(z) agentworkflow.ResetSignal{} elküldésével új ügynökmunkamenetet is indíthatnak.

Ö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.