Microsoft Ügynök-keretrendszer munkafolyamatai – Ellenőrzőpontok

Ez a lap áttekintést nyújt a Checkpoints Microsoft Agent Framework workflow rendszerében.

Áttekintés

Az ellenőrzőpontok lehetővé teszik a munkafolyamat állapotának mentését a végrehajtás bizonyos pontjain, és később folytathatja azokat. Ez a funkció különösen hasznos a következő helyzetekben:

  • Hosszú ideig futó munkafolyamatok, ahol hibák esetén el szeretné kerülni az előrehaladás elvesztését.
  • Hosszú ideig futó munkafolyamatok, ahol később szüneteltetni és folytatni szeretné a végrehajtást.
  • Olyan munkafolyamatok, amelyek naplózási vagy megfelelőségi célokra rendszeres állapotmentést igényelnek.
  • Különböző környezetekben vagy példányokban áttelepítendő munkafolyamatok.

Mikor jönnek létre ellenőrzőpontok?

Ne feledje, hogy a munkafolyamatok szupersztepsekben vannak végrehajtva, a munkafolyamat-végrehajtási modellben dokumentált módon. Az ellenőrzőpontok az egyes szupersztepek végén jönnek létre, miután az adott szuperstep összes végrehajtója befejezte a végrehajtást. Egy ellenőrzőpont rögzíti a munkafolyamat teljes állapotát, beleértve a következőket:

  • Az összes végrehajtó aktuális állapota
  • A munkafolyamat összes függőben lévő üzenete a következő szuperlépéshez
  • Függőben lévő kérelmek és válaszok
  • Megosztott állapotok

Megjegyzés:

Python 1.13.0-s verziójától kezdve a munkafolyamatok egy belépési ellenőrzőpontot is létrehoznak az első szupersztep előtt a munkafolyamat bemenetének rögzítéséhez, valamint egy másik belépési ellenőrzőpontot a kérési eseményekre adott válaszok kézbesítésekor. Ezek az ellenőrzőpontok a teljes munkafolyamatot újrajátszhatóvá teszik. Ez a kiadás kisebb kompatibilitástörő módosításokat tartalmaz az iterációk számától, az üzenetforrás azonosítóitól vagy az ellenőrzőpontok sorrendjétől függő alkalmazások esetében. A meglévő ellenőrzőpontok továbbra is támogatottak. Az áttelepítés részleteiért lásd: Python munkafolyamat-ellenőrzőpontok frissítése 1.13.0-ra.

Ellenőrzőpontok rögzítése

Az ellenőrzőpontok engedélyezéséhez szükséges egy CheckpointManager megadása a munkafolyamat futtatásakor. Ezután egy ellenőrzőpont egy SuperStepCompletedEvent-en keresztül, vagy a futtatás Checkpoints tulajdonságán keresztül érhető el.

using Microsoft.Agents.AI.Workflows;

// Create a checkpoint manager to manage checkpoints
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();

// Run the workflow with checkpointing enabled
StreamingRun run = await InProcessExecution
    .RunStreamingAsync(workflow, input, checkpointManager)
    .ConfigureAwait(false);
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is SuperStepCompletedEvent superStepCompletedEvt)
    {
        // Access the checkpoint
        CheckpointInfo? checkpoint = superStepCompletedEvt.CompletionInfo?.Checkpoint;
    }
}

// Checkpoints can also be accessed from the run directly
IReadOnlyList<CheckpointInfo> checkpoints = run.Checkpoints;

Az ellenőrzőpontok engedélyezéséhez meg kell adni egy CheckpointStorage, amikor munkafolyamatot hozunk létre. Ezután egy ellenőrzőpont a tárterületen keresztül érhető el. Az Agent Framework három beépített implementációt szállít – válassza ki a tartóssági és üzembehelyezési igényeknek megfelelőt:

Szolgáltató Szoftvercsomag Durability A következőkre alkalmas
InMemoryCheckpointStorage agent-framework Csak folyamatban Tesztek, bemutatók, rövid élettartamú munkafolyamatok
FileCheckpointStorage agent-framework Helyi lemez Egygépes munkafolyamatok, helyi fejlesztés
CosmosCheckpointStorage agent-framework-azure-cosmos Azure Cosmos DB Gyártási, elosztott, folyamatok közötti munkafolyamatok

Mindhárom ugyanazt CheckpointStorage a protokollt valósítja meg, így a munkafolyamat vagy a végrehajtó kód módosítása nélkül is felcserélheti a szolgáltatókat.

InMemoryCheckpointStorage a folyamat memóriájában tartja az ellenőrzőpontokat. Olyan tesztekhez, bemutatókhoz és rövid élettartamú munkafolyamatokhoz ajánlott, ahol nincs szükség tartósságra az újraindítások során.

from agent_framework import (
    InMemoryCheckpointStorage,
    WorkflowBuilder,
)

# Create a checkpoint storage to manage checkpoints
checkpoint_storage = InMemoryCheckpointStorage()

# Build a workflow with checkpointing enabled
builder = WorkflowBuilder(start_executor=start_executor, checkpoint_storage=checkpoint_storage)
builder.add_edge(start_executor, executor_b)
builder.add_edge(executor_b, executor_c)
builder.add_edge(executor_b, end_executor)
workflow = builder.build()

# Run the workflow
async for event in workflow.run(input, stream=True):
    ...

# Access checkpoints from the storage
checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=workflow.name)

Az ellenőrzőpontok engedélyezéséhez konfigurálja a végrehajtási környezetet egy ellenőrzőpont-kezelővel. Az ellenőrzőpont ezután a(z) workflow.SuperStepCompletedEvent elemről vagy a futás ellenőrzőpontlistáján keresztül érhető el.

checkpointManager := checkpoint.NewInMemoryManager()

run, err := inproc.Default.
    WithCheckpointing(checkpointManager).
    RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer run.Close(ctx)

var checkpoints []workflow.CheckpointInfo
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if completed, ok := evt.(workflow.SuperStepCompletedEvent); ok && completed.CompletionInfo != nil {
        if completed.CompletionInfo.CheckpointInfo != nil {
            checkpoints = append(checkpoints, *completed.CompletionInfo.CheckpointInfo)
        }
    }
}

// Checkpoints can also be accessed from the run directly.
checkpoints = run.Checkpoints()

Folytatás ellenőrzőpontokból

A munkafolyamatot közvetlenül ugyanazon a futtatáson folytathatja egy adott ellenőrzőpontról.

// Assume we want to resume from the 6th checkpoint
CheckpointInfo savedCheckpoint = run.Checkpoints[5];
// Restore the state directly on the same run instance.
await run.RestoreCheckpointAsync(savedCheckpoint).ConfigureAwait(false);
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is WorkflowOutputEvent workflowOutputEvt)
    {
        Console.WriteLine($"Workflow completed with result: {workflowOutputEvt.Data}");
    }
}

A munkafolyamatot közvetlenül ugyanazon a munkafolyamat-példányon folytathatja egy adott ellenőrzőpontról.

# Assume we want to resume from the 6th checkpoint
saved_checkpoint = checkpoints[5]
async for event in workflow.run(checkpoint_id=saved_checkpoint.checkpoint_id, stream=True):
    ...

A streamelési futtatásokat közvetlenül ugyanazon a futtatáson állíthatja vissza egy adott ellenőrzőpontra.

// Assume we want to resume from the 6th checkpoint.
savedCheckpoint := checkpoints[5]
if err := run.RestoreCheckpoint(ctx, savedCheckpoint); err != nil {
    return err
}

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if outputEvent, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed with result: %v\n", outputEvent.Output)
    }
}

Rehidratálás ellenőrzőpontokból

A rehidratált munkafolyamatnak meg kell őriznie az ellenőrzőpontot létrehozó munkafolyamat topológiáját és végrehajtói identitásait. A végrehajtói identitás feloldása az SDK-tól és a végrehajtó típusától függ.

Vagy újrahidratálhat egy munkafolyamatot egy ellenőrzőpontból egy új futtatási példányba.

// A rehydrated workflow must preserve the topology and executor identities of the workflow that
// created the checkpoint. This executor-only workflow rebuilds identically because its executors
// use fixed ids. Agent-based workflows must recreate each local agent with the same
// ChatClientAgentOptions.Id (and, if set, the same Name), otherwise the executor ids no longer
// match the checkpoint and resume fails.
var newWorkflow = WorkflowFactory.BuildWorkflow();
const int CheckpointIndex = 5;
Console.WriteLine($"\n\nHydrating a new workflow instance from the {CheckpointIndex + 1}th checkpoint.");
CheckpointInfo savedCheckpoint = checkpoints[CheckpointIndex];

await using StreamingRun newCheckpointedRun =
    await InProcessExecution.ResumeStreamingAsync(newWorkflow, savedCheckpoint, checkpointManager);

Fontos

Az átadott munkafolyamatnak ResumeStreamingAsync ugyanazzal a struktúrával és végrehajtói identitásokkal kell rendelkeznie, mint az ellenőrzőpontot létrehozó munkafolyamatnak. Ha a munkafolyamat olyan helyi ChatClientAgent példányokat tartalmaz, amelyek a kérések, a függőséginjektálási hatókörök, a folyamatok vagy az üzemelő példányok között rekonstruálva vannak, minden ügynökhöz rendeljen egy stabilat ChatClientAgentOptions.Id. Ha egy ügynök is beállít egy Namebeállítást, ezt Name is hagyja változatlanul.

Rendeljen például egy azonosítót, amely az ügynök logikai szerepkörét jelöli:

// Give each agent a stable, unique Id so its workflow executor identity stays the same when the
// workflow is reconstructed (for example per request or dependency-injection scope), which keeps
// checkpoints resumable. If an agent also has a Name, keep that stable too, since the executor
// identity includes it. Use a fixed logical role here, not a conversation, request, or user id.
internal const string IntakeAgentName = "Assistant";
public AIAgent IntakeAgent { get; } = chatClient.AsAIAgent(new ChatClientAgentOptions
{
    Id = "intake-agent",
    Name = IntakeAgentName,
    ChatOptions = new()
    {
        Instructions =
            """
            You receive a user request and are responsible for routing to the correct initial expert agent.
            """,
    },
});

Alkalmazza ezt a mintát a munkafolyamatban részt vevő összes ügynökre. Az ügynökazonosítóknak egyedinek kell lenniük a munkafolyamaton belül, és ugyanazon logikai ügynök rekonstruálásakor újra fel kell használni. Ne használjon beszélgetési azonosítókat, kérésazonosítókat, felhasználói azonosítókat, személyazonosításra alkalmas adatokat vagy titkos kulcsokat ügynökazonosítóként.

Ha egy ügynök Name be van állítva, az aktuális .NET munkafolyamat-végrehajtói identitás a saját Name és Idaz , így az egyik érték módosítása miatt az újraépített munkafolyamat nem kompatibilis az ellenőrzőponttal. A stabil értékek hozzárendelése nem javítja a különböző vagy véletlenszerűen létrehozott azonosítókkal létrehozott ellenőrzőpontokat; indítsa el az új munkamenetet, és helyette ellenőrizze a pontsort.

A kapcsolódó forgatókönyvekért tekintse meg a Munkafolyamatok ügynökök ésa Handoff vezénylés című témakört.

Egy új munkafolyamat-példányt egy ellenőrzőpontról is újrahidratálhat.

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder(start_executor=start_executor)
builder.add_edge(start_executor, executor_b)
builder.add_edge(executor_b, executor_c)
builder.add_edge(executor_b, end_executor)
# This workflow instance doesn't require checkpointing enabled.
workflow = builder.build()

# Assume we want to resume from the 6th checkpoint
saved_checkpoint = checkpoints[5]
async for event in workflow.run(
    checkpoint_id=saved_checkpoint.checkpoint_id,
    checkpoint_storage=checkpoint_storage,
    stream=True,
):
    ...

Egy új munkafolyamat-példányt egy ellenőrzőpontról is újrahidratálhat.

// Assume we want to resume from the 6th checkpoint
savedCheckpoint := checkpoints[5]
newWorkflow := buildWorkflow()

newRun, err := inproc.Default.
    WithCheckpointing(checkpointManager).
    ResumeStreaming(ctx, newWorkflow, savedCheckpoint)
if err != nil {
    return err
}
defer newRun.Close(ctx)

for evt, err := range newRun.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if outputEvent, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed with result: %v\n", outputEvent.Output)
    }
}

Végrehajtói állapotok mentése

Annak érdekében, hogy a végrehajtó állapota egy ellenőrzőponton legyen rögzítve, a végrehajtónak felül kell bírálnia a OnCheckpointingAsync metódust, és mentenie kell az állapotát a munkafolyamat-környezetbe.

using Microsoft.Agents.AI.Workflows;

internal sealed partial class CustomExecutor() : Executor("CustomExecutor")
{
    private const string StateKey = "CustomExecutorState";

    private List<string> messages = new();

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this.messages.Add(message);
        // Executor logic...
    }

    protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
    {
        return context.QueueStateUpdateAsync(StateKey, this.messages);
    }
}

Emellett annak érdekében, hogy az állapot helyesen legyen visszaállítva egy ellenőrzőpontról való folytatáskor, a végrehajtónak felül kell bírálnia a OnCheckpointRestoredAsync metódust, és be kell töltenie az állapotát a munkafolyamat-környezetből.

protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
    this.messages = await context.ReadStateAsync<List<string>>(StateKey).ConfigureAwait(false);
}

Annak biztosítása érdekében, hogy a végrehajtó állapota egy ellenőrzőpontban legyen rögzítve, a végrehajtónak felül kell bírálnia a on_checkpoint_save metódust, és szótárként kell visszaadnia az állapotát.

class CustomExecutor(Executor):
    def __init__(self, id: str) -> None:
        super().__init__(id=id)
        self._messages: list[str] = []

    @handler
    async def handle(self, message: str, ctx: WorkflowContext):
        self._messages.append(message)
        # Executor logic...

    async def on_checkpoint_save(self) -> dict[str, Any]:
        return {"messages": self._messages}

Emellett annak érdekében, hogy az állapot helyesen legyen visszaállítva az ellenőrzőpontról való folytatáskor, a végrehajtónak felül kell bírálnia a on_checkpoint_restore metódust, és vissza kell állítania az állapotát a megadott állapotszótárból.

async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
    self._messages = state.get("messages", [])

Annak érdekében, hogy a végrehajtó állapota rögzítve legyen egy ellenőrzőpontban, csatolja az ellenőrzőpont-horgokat a végrehajtóhoz, és tárolja az állapotot a munkafolyamat-környezeten keresztül.

type customExecutor struct {
    messages []string
}

func (e *customExecutor) Handle(message string) {
    e.messages = append(e.messages, message)
}

func (e *customExecutor) OnCheckpoint(ctx *workflow.Context) error {
    return ctx.QueueStateUpdate("CustomExecutorState", "", slices.Clone(e.messages))
}

Állítsa vissza az állapotot itt: OnCheckpointRestoredFunc

func (e *customExecutor) OnCheckpointRestored(ctx *workflow.Context) error {
    value, err := ctx.ReadState("CustomExecutorState", "")
    if err != nil {
        return err
    }
    if value == nil {
        e.messages = nil
        return nil
    }

    messages, ok := value.([]string)
    if !ok {
        return fmt.Errorf("unexpected custom executor state type %T", value)
    }
    e.messages = slices.Clone(messages)
    return nil
}

executorState := &customExecutor{}
custom := workflow.NewExecutor("CustomExecutor", executorState).Extend(&workflow.Executor{
    OnCheckpointFunc:         executorState.OnCheckpoint,
    OnCheckpointRestoredFunc: executorState.OnCheckpointRestored,
}).Bind()

Biztonsági szempontok

Fontos

A Checkpoint Storage egy megbízhatósági határ. Függetlenül attól, hogy a beépített tároló-implementációkat vagy egy egyénit használja, a tároló háttérrendszerét megbízható, privát infrastruktúraként kell kezelni. Soha ne töltsön be ellenőrzőpontokat nem megbízható vagy esetleg illetéktelen forrásokból.

Győződjön meg arról, hogy az ellenőrzőpontokhoz használt tárolási hely megfelelően van biztosítva. Csak a jogosult szolgáltatásoknak és a felhasználóknak kell olvasási vagy írási hozzáféréssel rendelkezniük az ellenőrzőpont-adatokhoz.

Pickle szerializálás

A FileCheckpointStorage és a CosmosCheckpointStorage Python pickle moduljával szerializálja a nem JSON-natív állapotokat, például az adatosztályokat, a datetime-okat és az egyéni objektumokat. Az önkényes kódvégrehajtás kockázatának csökkentése érdekében a deszerializálás során mindkét szolgáltató alapértelmezés szerint korlátozott unpicklert használ. A deszerializálás során csak biztonságos Python típusok (primitívek, datetime, , uuidközös Decimalgyűjtemények stb.) és támogatott Ügynök-keretrendszer- vagy OpenAI SDK-típusok beépített készlete engedélyezett. A modulelőtagok engedélyezési listája csak típusalapú: a segédfüggvények és más nem típusú globálisok elutasítva. A nem támogatott típusok miatt a deszerializálás sikertelen lesz egy WorkflowCheckpointException.

További alkalmazásspecifikus típusok engedélyezéséhez adja át őket a allowed_checkpoint_types paraméteren keresztül a "module:qualname" formátum használatával:

from agent_framework import FileCheckpointStorage

storage = FileCheckpointStorage(
    "/tmp/checkpoints",
    allowed_checkpoint_types=[
        "my_app.models:SafeState",
        "my_app.models:UserProfile",
    ],
)

Minden allowed_checkpoint_types bejegyzésnek egy típusra kell feloldania. A modulszintű függvények vagy más nem típusú globális függvények hozzáadása nem teszi ezt a globális deszerializálhatóvá.

CosmosCheckpointStorage ugyanazt a paramétert fogadja el:

from azure.identity.aio import DefaultAzureCredential
from agent_framework_azure_cosmos import CosmosCheckpointStorage

storage = CosmosCheckpointStorage(
    endpoint="https://my-account.documents.azure.com:443/",
    credential=DefaultAzureCredential(),
    database_name="agent-db",
    container_name="checkpoints",
    allowed_checkpoint_types=[
        "my_app.models:SafeState",
        "my_app.models:UserProfile",
    ],
)

Ha a fenyegetésmodell egyáltalán nem teszi lehetővé a pickle-alapú szerializálást, használjon InMemoryCheckpointStorage, vagy valósítson meg saját megoldást CheckpointStorage egy alternatív szerializálási stratégiával.

Tárolási hely felelőssége

FileCheckpointStorage explicit storage_path paramétert igényel – nincs alapértelmezett könyvtár. Bár a keretrendszer érvényesíti az elérési utakat érintő támadásokat, a tárolókönyvtár védelme (fájlengedélyek, inaktív titkosítás, hozzáférés-vezérlés) a fejlesztő feladata. Csak az engedélyezett folyamatoknak kell olvasási vagy írási hozzáféréssel rendelkezniük az ellenőrzőpont-címtárhoz.

CosmosCheckpointStorage az Azure Cosmos DB-ra támaszkodik tárolás céljából. Ha lehetséges, használjon felügyelt identitást/ RBAC-t, hatókört adjon az adatbázisnak és a tárolónak a munkafolyamat-szolgáltatásnak, és forgassa el a fiókkulcsokat, ha kulcsalapú hitelesítést használ. A fájltároláshoz hasonlóan csak a jogosult tagoknak kell olvasási vagy írási hozzáféréssel rendelkezniük az ellenőrzőpont-dokumentumokat tartalmazó Cosmos DB-tárolóhoz.

A Go ellenőrzőpont-kezelők JSON formátumban szerializálják az ellenőrzőpont állapotát, de az ellenőrzőpontok tárolása továbbra is megbízhatónak tekintett alkalmazásállapot. Ha használja checkpoint.NewFileSystemJSONStore, az ellenőrzőpont-fájlokat egy védett könyvtárban tárolja, és csak az engedélyezett folyamatok olvasási/írási hozzáférését korlátozza. Az egyéni üzletek felelősek saját hozzáférés-vezérlési, integritási és tartóssági garanciáikért.

Következő lépések

  • Last updated on