Microsoft Agent Framework-werkstromen - Controlepunten

Deze pagina bevat een overzicht van Checkpoints in het Microsoft Agent Framework Workflow-systeem.

Overview

Met controlepunten kunt u de status van een werkstroom opslaan op specifieke punten tijdens de uitvoering en later hervatten vanaf die punten. Deze functie is met name handig voor de volgende scenario's:

  • Langlopende werkstromen waarbij u wilt voorkomen dat de voortgang verloren gaat in het geval van fouten.
  • Langlopende werkstromen waarin u de uitvoering op een later tijdstip wilt onderbreken en hervatten.
  • Werkstromen waarvoor periodieke statusbesparing is vereist voor controle- of nalevingsdoeleinden.
  • Werkstromen die moeten worden gemigreerd naar verschillende omgevingen of instanties.

Wanneer worden controlepunten gemaakt?

Houd er rekening mee dat werkstromen worden uitgevoerd in supersteps, zoals beschreven in het werkstroomuitvoeringsmodel. Controlepunten worden gemaakt aan het einde van elke superstep, nadat alle uitvoerders in die superstep hun uitvoering hebben voltooid. Een controlepunt legt de volledige status van de werkstroom vast, waaronder:

  • De huidige status van alle uitvoerders
  • Alle berichten in behandeling in de werkstroom voor de volgende superstep
  • Aanvragen en antwoorden in behandeling
  • Gedeelde toestanden

Opmerking

Vanaf Python versie 1.13.0 maken werkstromen ook een invoercontrolepunt voordat de eerste superstap de invoer van de werkstroom registreert en een ander controlepunt voor invoer wanneer reacties op aanvraag-gebeurtenissen worden geleverd. Met deze controlepunten kan de volledige werkstroom opnieuw worden afgespeeld. Deze release bevat kleine belangrijke wijzigingen voor toepassingen die afhankelijk zijn van iteratieaantallen, berichtbron-id's of volgorde van controlepunten. Bestaande controlepunten blijven ondersteund. Zie Upgrade Python werkstroomcontrolepunten naar 1.13.0 voor meer informatie over de migratie.

Controlepunten vastleggen

Als u checkpointing wilt inschakelen, moet tijdens het uitvoeren van de workflow een CheckpointManager worden opgegeven. Een controlepunt kan vervolgens worden geopend via een SuperStepCompletedEvent, of met behulp van de Checkpoints eigenschap tijdens de uitvoering van het programma.

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;

Als u controlepunten wilt inschakelen, moet er een CheckpointStorage worden opgegeven bij het maken van een workflow. Een controlepunt kan vervolgens worden geopend via de opslag. Agent Framework verzendt drie ingebouwde implementaties: kies de implementatie die overeenkomt met uw duurzaamheid en implementatiebehoeften:

Provider Package Durability Geschikt voor
InMemoryCheckpointStorage agent-framework In bewerking alleen Tests, demo's, kortdurende werkstromen
FileCheckpointStorage agent-framework Lokale schijf Werkstromen met één machine, lokale ontwikkeling
CosmosCheckpointStorage agent-framework-azure-cosmos Azure Cosmos DB Productie-, gedistribueerde, werkstromen voor meerdere processen

Alle drie implementeren hetzelfde CheckpointStorage protocol, zodat u providers kunt wisselen zonder werkstroom- of uitvoercode te wijzigen.

Ingebouwde orchestratie-builders accepteren een optionele name, die de workflownaam wordt die wordt gebruikt om controlepunten te groeperen en terug te vinden. Hun stabiele standaardwaarden zijn Concurrent, Sequential, GroupChat, en HandoffMagentic. Stel een expliciete naam in wanneer meerdere workflows van hetzelfde orchestratietype dezelfde checkpointopslag delen. Gebruik dezelfde naam wanneer u de controlepunten weer in een lijst opneemt of hervat.

InMemoryCheckpointStorage houdt controlepunten in procesgeheugen. Het meest geschikt voor tests, demo's en kortdurende werkstromen waarvoor u geen duurzaamheid nodig hebt bij het opnieuw opstarten.

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)

Als u controlepunten wilt inschakelen, configureert u de uitvoeringsomgeving met een controlepuntbeheer. Een controlepunt kan vervolgens worden geopend vanuit workflow.SuperStepCompletedEvent, of via de lijst met controlepunten van de uitvoering.

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

Hervatten vanaf controlepunten

U kunt een werkstroom rechtstreeks vanuit een specifiek controlepunt hervatten tijdens dezelfde uitvoering.

// 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}");
    }
}

U kunt een werkstroom hervatten vanaf een specifiek controlepunt op dezelfde workflowinstantie.

# 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):
    ...

U kunt een streamingrun direct op diezelfde run herstellen naar een specifiek controlepunt.

// 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)
    }
}

Reactiveren vanuit controlepunten

Een gerehydrateerde workflow moet de topologie en de identiteiten van de executors behouden van de workflow waarmee het controlepunt is gemaakt. Hoe de identiteit van de executor wordt bepaald, hangt af van de SDK en het type executor.

Of u kunt een werkstroom vanuit een controlepunt reactiveren in een nieuw run-exemplaar.

// 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);

Belangrijk

De doorgegeven ResumeStreamingAsync werkstroom moet dezelfde structuur- en uitvoerdersidentiteiten hebben als de werkstroom die het controlepunt heeft gemaakt. Als de workflow lokale ChatClientAgent-instanties bevat die opnieuw worden opgebouwd tussen aanvragen, scopes voor dependency injection, processen of deployments, wijs dan elke agent een stabiele ChatClientAgentOptions.Id toe. Als een agent ook een Name instelt, laat dan ook dat Name ongewijzigd.

Wijs bijvoorbeeld een id toe die de logische rol van de agent vertegenwoordigt:

// 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.
            """,
    },
});

Pas dit patroon toe op elke agent die deelneemt aan de werkstroom. Agent-id's moeten uniek zijn binnen de werkstroom en moeten opnieuw worden gebruikt bij het reconstrueren van dezelfde logische agent. Gebruik geen gespreks-id's, aanvraag-id's, gebruikers-id's, persoonsgegevens of geheimen als agent-id's.

Wanneer een agent Name is ingesteld, wordt de identiteit van de huidige .NET-workflowexecutor afgeleid van zowel de Name als de Id, dus door een van beide waarden te wijzigen wordt de opnieuw opgebouwde workflow incompatibel met het checkpoint. Het toewijzen van stabiele waarden herstelt geen controlepunten die zijn gemaakt met verschillende of willekeurig gegenereerde id's; start in plaats daarvan een nieuwe sessie en controlepuntherkomst.

Zie Werkstromen als agents en handoff-indeling voor gerelateerde scenario's.

U kunt ook een nieuwe werkstroomexemplaar herstellen vanaf een controlepunt.

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,
):
    ...

U kunt ook een nieuwe werkstroomexemplaar herstellen vanaf een controlepunt.

// 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)
    }
}

Uitvoerdersstatussen opslaan

Om ervoor te zorgen dat de status van een uitvoerder wordt vastgelegd in een controlepunt, moet de uitvoerder de methode overschrijven en de OnCheckpointingAsync status ervan opslaan in de werkstroomcontext.

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);
    }
}

Om ervoor te zorgen dat de status correct wordt hersteld bij het hervatten van een controlepunt, moet de uitvoerder de OnCheckpointRestoredAsync methode overschrijven en de status ervan laden vanuit de werkstroomcontext.

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

Het openbare AgentExecutorCheckpointStateTypedDict bestand beschrijft de status die de ingebouwde AgentExecutor opslaat en herstelt:

  • cache: Berichten gebufferd vóór de volgende aanroep van de agent.
  • full_conversation: Eerdere invoer en de uitvoer van de assistent of het hulpprogramma van de meest recente uitvoering.
  • agent_session: De geserialiseerde agentsessie.
  • pending_agent_requests: Aanvragen van de in-flight-agent die zijn gesleuteld op aanvraag-id.
  • pending_responses_to_agent: Antwoorden in wachtrij die wachten om naar de agent te worden verzonden.

Ontbrekende bekende velden gebruiken hun standaardinstellingen, inclusief een nieuwe agentsessie wanneer agent_session deze afwezig is. Het systeem negeert onbekende velden. Een onjuist gevormd bekend veld werpt WorkflowCheckpointExceptionop.

Voor een aangepaste uitvoerder definieert u een toegewezen waarde TypedDict voor de status van het controlepunt en valideert u de bekende velden in on_checkpoint_restore. Genereer WorkflowCheckpointException voor onjuist gevormde checkpointgegevens.

Als u de status van een aangepaste executor in een controlepunt wilt vastleggen, overschrijft u de on_checkpoint_save methode en retourneert u de status als een woordenlijst.

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}

Om ervoor te zorgen dat de status correct wordt hersteld bij het hervatten van een controlepunt, moet de uitvoerder de on_checkpoint_restore methode overschrijven en de status ervan herstellen vanuit de opgegeven statuswoordenlijst.

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

Als u ervoor wilt zorgen dat de uitvoeringsstatus wordt vastgelegd in een controlepunt, koppelt u controlepunthaken aan de uitvoerder en slaat u de status op via de werkstroomcontext.

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

Herstel de toestand in 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()

Beveiligingsoverwegingen

Belangrijk

Controlepuntopslag vormt een vertrouwensgrens. Of u nu gebruikmaakt van de ingebouwde opslag-implementaties of een aangepaste implementatie, de back-end van de opslag moet worden behandeld als vertrouwde, privé-infrastructuur. Laad nooit controlepunten van niet-vertrouwde of mogelijk gemanipuleerde bronnen.

Zorg ervoor dat de opslaglocatie die wordt gebruikt voor controlepunten op de juiste manier is beveiligd. Alleen geautoriseerde services en gebruikers moeten lees- of schrijftoegang hebben tot controlepuntgegevens.

Pickle-serialisatie

FileCheckpointStorage, FoundryCheckpointStore en CosmosCheckpointStorage gebruiken de module pickle van Python om status die niet van nature JSON is te serialiseren, zoals dataclasses, datum- en tijdobjecten en aangepaste objecten. Om de risico's van willekeurige uitvoering van code tijdens de deserialisatie te beperken, gebruiken alle drie de winkels standaard een beperkte uitkiezer . Alleen een ingebouwde set veilige Python typen (primitieven, datetime, , uuidalgemene Decimalverzamelingen, enzovoort) en ondersteunde Agent Framework- of OpenAI SDK-typen zijn toegestaan tijdens deserialisatie. Allowlisting op basis van modulevoorvoegsel geldt alleen voor typen: hulpfuncties en andere niet-type globale elementen worden geweigerd.

Agent Framework laat die tijdelijke raw_representation-waarden weg uit frameworkeigen objecten voordat ze worden gepickeld en herstelt die velden als None. Elk ander niet-ondersteund type zorgt ervoor dat deserialisatie mislukt met een WorkflowCheckpointException.

Importeer een ingebouwde orchestratieklasse uit agent_framework.orchestrations voordat u een checkpoint laadt dat de status van de orchestratie bevat. Wanneer u het orkestratiepakket laadt, worden de systeemeigen typen van het framework automatisch geregistreerd, zodat u ze niet aan allowed_checkpoint_types hoeft toe te voegen.

Als u aanvullende toepassingsspecifieke typen wilt toestaan, geeft u deze door via de allowed_checkpoint_types parameter met behulp van "module:qualname" de indeling:

from agent_framework import FileCheckpointStorage

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

Elke allowed_checkpoint_types vermelding moet worden omgezet in een type. Het toevoegen van een functie op moduleniveau of een andere globale variabele die geen type is, maakt die globale variabele nog niet deserialiseerbaar.

CosmosCheckpointStorage accepteert dezelfde parameter:

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",
    ],
)

FoundryCheckpointStore accepteert ook allowed_checkpoint_types. Wanneer ResponsesHostServer werkstroomcontrolepunten beheert, maakt u met deze parameter een CheckpointStoreProvider en geeft u de provider door via checkpoint_store_provider.

Als u een type wilt registreren voor elke beperkte controlepuntdecoder in het huidige Python proces, geeft u de klasse door aanregister_checkpoint_type():

from agent_framework import register_checkpoint_type
from my_app.models import SafeState

register_checkpoint_type(SafeState)

Globale registratie is ook van toepassing op opslagexemplaren die vóór de aanroep zijn gemaakt. Daarentegen geldt dit alleen voor de checkpointopslaginstantie allowed_checkpoint_types of CheckpointStoreProvider die deze ontvangt. Geef de voorkeur aan de parameter per winkel of per provider wanneer uw code opslag configureert. Gebruik alleen globale registratie wanneer uw code het opslagexemplaren of de provider niet kan configureren. Registreer het type voordat u een controlepunt laadt dat het bevat.

Beide mechanismen breiden de toegestane lijst van de beperkte unpickler uit; geen van beide maakt niet-vertrouwde pickledata veilig. Registreer alleen vertrouwde toepassingsklassen omdat een toegestane klasse aangepaste pickle-reconstructiegedrag kan definiëren. Globale registratie verhoogt het deserialisatieoppervlak voor elke beperkte belasting van controlepunten in het proces, dus behoud het globale register tot de minimaal vereiste set.

Als uw bedreigingsmodel helemaal geen serialisatie op basis van pickle toestaat, gebruikt InMemoryCheckpointStorage of implementeert u een aangepaste CheckpointStorage met een alternatieve serialisatiestrategie.

Verantwoordelijkheid voor opslaglocatie

FileCheckpointStorage vereist een expliciete storage_path parameter. Er is geen standaardmap. Hoewel het framework valideert tegen padtraversalaanvallen, is het beveiligen van de opslagmap zelf (bestandsmachtigingen, versleuteling bij opslag, toegangsbeheer) de verantwoordelijkheid van de ontwikkelaar. Alleen geautoriseerde processen moeten lees- of schrijftoegang hebben tot de checkpointdirectory.

CosmosCheckpointStorage is afhankelijk van Azure Cosmos DB voor opslag. Gebruik waar mogelijk beheerde identiteit en rolgebaseerde toegangscontrole (RBAC), beperk de toegang van de database en container tot de workflowservice en ververs accountsleutels als u sleutelgebaseerde authenticatie gebruikt. Net als bij bestandsopslag moeten alleen geautoriseerde principals lees- of schrijftoegang hebben tot de Cosmos DB-container die checkpoint-documenten bevat.

Go-checkpointbeheerders serialiseren checkpointstatus als JSON, maar checkpointopslag is nog steeds een vertrouwde applicatiestatus. Als u dit gebruikt checkpoint.NewFileSystemJSONStore, slaat u controlepuntbestanden op in een beveiligde map en beperkt u alleen lees-/schrijftoegang tot geautoriseerde processen. Aangepaste winkels zijn verantwoordelijk voor hun eigen garanties voor toegangsbeheer, integriteit en duurzaamheid.

Volgende stappen

  • Last updated on