Zustand für lang laufende Agents verwalten (Vorschauversion)

Ein langlebiger gehosteter Agent erholt sich nur dann von Abstürzen, wenn sein Fortschritt dauerhaft gespeichert wird. Dieser Artikel zeigt die zwei Ebenen des dauerhaften Zustands – kleine Mengen an Aufgabenmetadaten als Prüfpunkt-Index und den Foundry-Zustandsspeicher als Prüfpunkt-Speicher. Außerdem wird erörtert, wie man einen Framework-Checkpointer so sichert, dass die Wiederherstellung auf Frameworkebene Neustarts übersteht.

Zwei Ebenen des Zustands

Die Regel: Metadaten sind ein Checkpoint-Index, kein Checkpoint-Speicher. Kleine Schreibvorgänge sind kostengünstig und schnell; umfangreiche Schreibvorgänge stoßen an die Payload-Grenzen des Task-Stores und verlangsamen die Wiederherstellung.

Verwenden von Aufgabenmetadaten für Wasserzeichen

ctx.metadata ist ein kleiner Schlüssel-Wert-Namespace, der Abstürze übersteht und über mehrere Runden einer Kette hinweg sichtbar ist. Werte müssen JSON-serialisierbar sein.

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

Persistenz ist nicht implizit. Rufen Sie flush() auf, wenn das Schreiben der Metadaten vor einem Nebeneffekt erfolgen muss, der nicht dedupliziert werden kann. Namen, die mit _ beginnen, sind für das Framework reserviert und lösen ValueError aus.

Drei nützliche Bereiche:

Massenzustand im Foundry-Zustandsspeicher speichern

FoundryStateStore ist ein dauerhafter, servergestützter Schlüssel-Wert-Speicher für Zustandsdaten, die Abstürze und Entfernung bei Inaktivität überstehen müssen. Ein Speicher ist an einen vom Anrufer gewählten Namen gebunden; codieren Sie Ihren Geltungsbereich (Sitzung, Thread oder Lauf) in diesem Namen.

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

Wichtige Verhaltensweisen:

• get_or_create() ruft den Store in einem Aufruf ab oder erstellt ihn; get_or_create() wendet user_isolation / item_ttl_seconds nur bei der ersten Erstellung an.
• Name des Stores = Geltungsbereich. Namen können / enthalten; verwenden Sie dieses als Hierarchietrennzeichen und wählen Sie von Anfang an ein stabiles Schema.
• Optimistische Parallelität. Verwenden Sie if_match=item.etag für veränderbare Elemente wie Zähler; überspringen Sie es bei Nur-Anfüge-Prüfpunkten. Eine nicht erfüllte Vorbedingung löst FoundryStoragePreconditionError aus.
• Grenzwerte. Elementwert ≤ 1 MB serialisiert; Speichername 1–128 Zeichen; bis zu 16 Tags pro Element.

Framework-Checkpointer sichern

Zeigen Sie auf einen LangGraph- oder Microsoft Agent Framework (MAF)-Checkpointer im FoundryStateStore, und die eigene Wiederherstellung des Frameworks wird über Abstürze hinweg dauerhaft – kein benutzerdefinierter Wiederherstellungscode.

Prüfpunkte werden nur angefügt – bei jedem Speichern wird eine neue ID verwendet, daher gibt es keine Schreibkonflikte. Außerdem wird für den Prüfpunktpfad niemals if_match benötigt.

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

Eingaben klein halten

Aufgabeneingaben werden persistiert, bevor der Handler ausgeführt wird (darauf basiert die Wiederherstellung), also halten Sie sie klein: Das Limit pro Eingabe beträgt nach der JSON-Serialisierung etwa 10 MiB, und bei größeren Eingaben wird noch vor einem Netzwerkaufruf InputTooLarge ausgelöst. Lagern Sie umfangreiche Nutzdaten in einen Blob Storage aus, und übergeben Sie stattdessen einen Verweis.

Verwandte Inhalte

• Resilienz für gehostete Agenten mit langer Laufzeit
• API-Referenz für langlebige Agenten
• Lang andauernde Arbeit nach einem Absturz wiederherstellen
• Genehmigungsschritt mit menschlicher Freigabe hinzufügen