Gestire lo stato per gli agenti con esecuzione prolungata (anteprima)

Un agente ospitato con esecuzione prolungata si riprende dagli arresti anomali solo se il suo stato di avanzamento è persistente. Questo articolo illustra i due livelli di stato persistente: i metadati di piccole dimensioni delle attività come indice di checkpoint e l'archivio di stato di Foundry come archivio di checkpoint, e spiega come fornire un supporto a un checkpointer del framework affinché il ripristino a livello di framework sopravviva ai riavvii.

Due livelli di stato

La regola: metadata è un indice del checkpoint, non un archivio dei checkpoint. Le scritture di piccole dimensioni sono economiche e veloci; le scritture in blocco raggiungono i limiti del payload dell'archivio delle attività e rallentano il ripristino.

Utilizza i metadati dell'attività per le filigrane

ctx.metadata è un piccolo spazio di nomi chiave-valore che sopravvive agli arresti anomali ed è visibile attraverso i vari turni di una catena. I valori devono essere serializzabili in JSON.

@multi_turn_task(name="workflow")
async def run(ctx: TaskContext[dict]) -> dict:
    step = int(ctx.metadata.get("workflow_step", 0))
    for i in range(step, total_steps):
        await upstream_store.write_step_result(i, result)   # bulk data goes to your store
        ctx.metadata["workflow_step"] = i + 1
        await ctx.metadata.flush()                          # explicit fence before the next side effect
    return {"done": True}

La persistenza non è implicita. Chiamare flush() quando la scrittura dei metadati deve essere completata prima di un effetto collaterale che non può essere deduplicato. I nomi che iniziano con _ sono riservati per il framework e generano ValueError.

Tre ambiti utili:

Archiviare lo stato in blocco nell'archivio di stato di Foundry

FoundryStateStore è un archivio chiave-valore persistente, basato sul server, per lo stato che deve sopravvivere agli arresti anomali e all'espulsione per inattività. Un archivio è vincolato a un nome scelto dal chiamante; codifica l'ambito (sessione, thread o esecuzione) nel nome.

from azure.ai.agentserver.core.storage import FoundryStateStore

store = await FoundryStateStore.get_or_create(
    "checkpoints/thread-abc",   # store name == scope
    user_isolation=True,        # partition items per end user when the name is shared
    item_ttl_seconds=3600,      # idle items age out (store-level; renewed on write)
    description="Checkpoints for thread abc",
)
async with store:
    await store.set_item("step-1", {"done": False})
    item = await store.get_item("step-1")

Comportamenti principali:

• get_or_create() recupera o crea lo store in una sola chiamata; applica user_isolation / item_ttl_seconds solo durante la prima creazione.
• Nome dello store = ambito. I nomi possono contenere /; usarlo come separatore di gerarchia e scegliere uno schema stabile in anticipo.
• Concorrenza ottimistica. Usare if_match=item.etag per elementi modificabili, ad esempio i contatori, ignorarlo nei checkpoint di sola aggiunta. Una precondizione non riuscita genera FoundryStoragePreconditionError.
• Limiti. Valore dell'elemento serializzato ≤ 1 MB; nome dell'archivio 1–128 caratteri; fino a 16 tag per elemento.

Eseguire il backup del checkpointer del framework

Puntate un checkpointer di LangGraph o di Microsoft Agent Framework (MAF) a FoundryStateStore e il meccanismo di ripristino del framework rimane persistente anche in caso di arresti anomali, senza bisogno di codice di ripristino personalizzato.

I checkpoint sono di tipo solo append: ogni salvataggio usa un nuovo ID, quindi non si verifica alcuna contesa in scrittura e non è mai necessario if_match sul percorso del checkpoint.

# LangGraph: one thread = one store
async def _store(thread_id: str) -> FoundryStateStore:
    return await FoundryStateStore.get_or_create(
        f"langGraphCheckpoints/{thread_id}", user_isolation=True
    )

Mantenere gli input di piccole dimensioni

Gli input dell'attività vengono salvati in modo persistente prima che il gestore venga eseguito (è su questo che si basa il ripristino), quindi è bene che restino di dimensioni ridotte: il limite per ciascun input è di circa 10 MiB dopo la serializzazione JSON e gli input più grandi provocano InputTooLarge prima di qualsiasi chiamata di rete. Esternalizzare i payload di grandi dimensioni in un archivio BLOB e passare un riferimento.

Contenuti correlati

• Resilienza per gli agenti ospitati di lunga durata
• Informazioni di riferimento sulle API dell'agente a esecuzione prolungata
• Recuperare il lavoro con esecuzione prolungata dopo un arresto anomalo.
• Aggiungere un passaggio di approvazione con intervento umano