Alur Kerja Kerangka Kerja Agen Microsoft - Titik Pemeriksaan

Halaman ini menyediakan gambaran umum Checkpoints dalam sistem Alur Kerja Kerangka Kerja Agen Microsoft.

Gambaran Umum

Titik pemeriksaan memungkinkan Anda menyimpan status alur kerja pada titik tertentu selama eksekusinya, dan melanjutkan dari titik-titik tersebut nanti. Fitur ini sangat berguna untuk skenario berikut:

  • Alur kerja yang berjalan lama di mana Anda ingin menghindari kehilangan kemajuan jika terjadi kegagalan.
  • Alur kerja yang berjalan lama di mana Anda ingin menjeda dan melanjutkan eksekusi di lain waktu.
  • Alur kerja yang memerlukan penyimpanan status berkala untuk tujuan audit atau kepatuhan.
  • Alur kerja yang perlu dimigrasikan di berbagai lingkungan atau instans.

Kapan Titik Pemeriksaan Dibuat?

Ingatlah bahwa alur kerja dijalankan dalam supersteps, seperti yang didokumenkan dalam model eksekusi alur kerja. Titik pemeriksaan dibuat di akhir setiap superstep, setelah semua pelaksana dalam superstep tersebut telah menyelesaikan eksekusi mereka. Titik pemeriksaan menangkap seluruh status alur kerja, termasuk:

  • Status saat ini dari semua pelaksana
  • Semua pesan yang tertunda dalam alur kerja untuk superstep berikutnya
  • Permintaan dan respons tertunda
  • Status bersama

Nota

Mulai dari Python versi 1.13.0, alur kerja juga membuat titik pemeriksaan entri sebelum langkah pertama untuk merekam input alur kerja, dan titik pemeriksaan entri lain saat respons terhadap peristiwa permintaan dikirimkan. Titik pemeriksaan ini membuat alur kerja lengkap dapat diputar ulang. Rilis ini mencakup perubahan kecil untuk aplikasi yang bergantung pada jumlah iterasi, ID sumber pesan, atau urutan titik pemeriksaan. Titik pemeriksaan yang ada tetap didukung. Untuk detail migrasi, lihat Meningkatkan Python titik pemeriksaan alur kerja ke 1.13.0.

Menangkap Titik Pemeriksaan

Untuk mengaktifkan titik pemeriksaan, CheckpointManager perlu disediakan saat menjalankan alur kerja. Titik pemeriksaan kemudian dapat diakses melalui SuperStepCompletedEvent, atau melalui properti Checkpoints pada run.

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;

Untuk mengaktifkan titik pemeriksaan, CheckpointStorage perlu disediakan saat membuat alur kerja. Titik pemeriksaan kemudian dapat diakses melalui penyimpanan. Agent Framework mengirimkan tiga implementasi bawaan — pilih salah satu yang sesuai dengan durabilitas dan kebutuhan penyebaran Anda:

Provider Package Durability Paling cocok untuk
InMemoryCheckpointStorage agent-framework Hanya dalam pemrosesan Pengujian, demo, alur kerja sementara
FileCheckpointStorage agent-framework Disk lokal Alur kerja mesin tunggal, pengembangan lokal
CosmosCheckpointStorage agent-framework-azure-cosmos Azure Cosmos DB Alur kerja produksi, terdistribusi, lintas proses

Ketiganya menerapkan protokol yang sama CheckpointStorage , sehingga Anda dapat menukar penyedia tanpa mengubah alur kerja atau kode pelaksana.

InMemoryCheckpointStorage menyimpan titik pemeriksaan dalam memori proses. Terbaik untuk pengujian, demo, dan alur kerja berumur pendek di mana Anda tidak memerlukan ketahanan saat melakukan restart.

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)

Untuk mengaktifkan titik pemeriksaan, konfigurasikan lingkungan eksekusi dengan manajer titik pemeriksaan. Checkpoint selanjutnya dapat diakses dari workflow.SuperStepCompletedEvent, atau melalui daftar checkpoint run tersebut.

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

Lanjutkan dari Titik Pemeriksaan

Anda dapat melanjutkan alur kerja dari titik pemeriksaan tertentu langsung dalam proses yang sama.

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

Anda dapat melanjutkan alur kerja dari checkpoint tertentu langsung pada instans alur kerja yang sama.

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

Anda dapat memulihkan proses streaming ke checkpoint tertentu langsung pada proses yang sama.

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

Rehidrasi dari Titik Pemeriksaan

Alur kerja yang direhidrasi harus mempertahankan identitas topologi dan pelaksana alur kerja yang membuat titik pemeriksaan. Bagaimana identitas pelaksana diselesaikan tergantung pada jenis SDK dan pelaksana.

Atau Anda dapat merehidrasi alur kerja dari titik pemeriksaan ke dalam instans eksekusi baru.

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

Penting

Alur kerja yang diteruskan ke ResumeStreamingAsync harus memiliki struktur dan identitas pelaksana yang sama dengan alur kerja yang membuat titik pemeriksaan. Jika alur kerja berisi instans lokal ChatClientAgent yang direkonstruksi di seluruh permintaan, cakupan injeksi dependensi, proses, atau penyebaran, tetapkan setiap agen dengan stabil ChatClientAgentOptions.Id. Jika agen juga mengatur Name, jaga agar Name tidak berubah juga.

Misalnya, tetapkan ID yang mewakili peran logis agen:

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

Terapkan pola ini ke setiap agen yang berpartisipasi dalam alur kerja. ID agen harus unik dalam alur kerja dan harus digunakan kembali saat menyusun ulang agen logis yang sama. Jangan gunakan ID percakapan, ID permintaan, ID pengguna, informasi identitas pribadi, atau rahasia sebagai ID agen.

Ketika agen Name diatur, identitas pelaksana alur kerja .NET saat ini berasal dari Name dan Id, sehingga mengubah salah satu nilai membuat alur kerja pembangunan ulang tidak kompatibel dengan titik pemeriksaan. Menetapkan nilai stabil tidak memperbaiki titik pemeriksaan yang dibuat dengan ID yang berbeda atau dihasilkan secara acak; mulai sesi baru dan silsilah titik pemeriksaan sebagai gantinya.

Untuk skenario terkait, lihat Alur kerja sebagai Agen dan orkestrasi Handoff.

Atau Anda dapat merehidrasi instans alur kerja baru dari titik pemeriksaan.

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

Atau Anda dapat merehidrasi instans alur kerja baru dari titik pemeriksaan.

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

Simpan Status Pelaksana

Untuk memastikan bahwa status pelaksana diambil dalam titik pemeriksaan, pelaksana harus mengambil alih OnCheckpointingAsync metode dan menyimpan statusnya ke konteks alur kerja.

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

Selain itu, untuk memastikan status dipulihkan dengan benar saat melanjutkan dari titik pemeriksaan, pelaksana harus mengambil alih OnCheckpointRestoredAsync metode dan memuat statusnya dari konteks alur kerja.

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

Untuk memastikan bahwa status pelaksana ditangkap dalam titik pemeriksaan, pelaksana harus mengambil on_checkpoint_save alih metode dan mengembalikan statusnya sebagai kamus.

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}

Selain itu, untuk memastikan status dipulihkan dengan benar saat melanjutkan dari titik pemeriksaan, pelaksana harus mengambil alih on_checkpoint_restore metode dan memulihkan statusnya dari kamus status yang disediakan.

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

Untuk memastikan bahwa status eksekutor tercatat dalam checkpoint, lampirkan hook checkpoint ke eksekutor dan simpan statusnya melalui konteks alur kerja.

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

Pulihkan status di 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()

Pertimbangan Keamanan

Penting

Penyimpanan checkpoint berfungsi sebagai batas keamanan kepercayaan. Baik Anda menggunakan implementasi penyimpanan bawaan atau yang kustom, backend penyimpanan harus diperlakukan sebagai infrastruktur privat tepercaya. Jangan pernah memuat titik pemeriksaan dari sumber yang tidak terpercaya atau berpotensi telah diubah.

Pastikan bahwa lokasi penyimpanan yang digunakan untuk titik pemeriksaan diamankan dengan tepat. Hanya layanan dan pengguna yang berwenang yang seharusnya memiliki akses baca atau tulis ke data titik pemeriksaan.

Serialisasi Pickle

Baik FileCheckpointStorage dan CosmosCheckpointStorage menggunakan modul pickle Python untuk menserialisasikan status non-JSON-asli seperti kelas data, tanggalwaktu, dan objek kustom. Untuk mengurangi risiko eksekusi kode arbitrer selama deserialisasi, kedua penyedia menggunakan unpickler terbatas secara bawaan. Hanya sekumpulan jenis Python aman bawaan (primitif, datetime, , uuidDecimal, koleksi umum, dll.) dan jenis Agent Framework atau OpenAI SDK yang didukung yang diizinkan selama deserialisasi. Daftar izin awalan modul adalah jenis-saja: fungsi pembantu dan global non-jenis lainnya ditolak. Jenis apa pun yang tidak didukung menyebabkan deserialisasi gagal dengan WorkflowCheckpointException.

Untuk mengizinkan jenis khusus aplikasi tambahan, lewatkan melalui parameter allowed_checkpoint_types dengan menggunakan format "module:qualname".

from agent_framework import FileCheckpointStorage

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

Setiap allowed_checkpoint_types entri harus diselesaikan ke jenis. Menambahkan fungsi tingkat modul atau global non-jenis lain tidak membuat deserialisasi global tersebut dapat dideserialisasi.

CosmosCheckpointStorage menerima parameter yang sama:

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

Jika model ancaman Anda tidak mengizinkan serialisasi berbasis pickle sama sekali, gunakan InMemoryCheckpointStorage atau menetapkan CheckpointStorage melalui strategi serialisasi alternatif.

Tanggung jawab lokasi penyimpanan

FileCheckpointStorage memerlukan parameter eksplisit storage_path — tidak ada direktori default. Meskipun kerangka kerja memvalidasi terhadap serangan traversal path, tanggung jawab pengembang adalah mengamankan direktori penyimpanan itu sendiri, termasuk izin file, enkripsi dalam keadaan diam, dan kontrol akses. Hanya proses yang berwenang yang seharusnya memiliki akses baca atau tulis ke direktori titik pemeriksaan.

CosmosCheckpointStorage bergantung pada Azure Cosmos DB untuk penyimpanan. Gunakan identitas terkelola / RBAC jika memungkinkan, tetapkan cakupan database dan kontainer ke dalam layanan alur kerja, dan rotasikan kunci akun jika menggunakan autentikasi berbasis kunci. Seperti halnya penyimpanan file, hanya pihak yang berwenang yang seharusnya memiliki akses baca atau tulis ke kontainer Cosmos DB yang menyimpan dokumen penanda posisi.

Manajer checkpoint Go menyerialkan status checkpoint sebagai JSON, tetapi penyimpanan checkpoint tetap merupakan status aplikasi yang tepercaya. Jika Anda menggunakan checkpoint.NewFileSystemJSONStore, simpan file titik pemeriksaan di direktori yang dilindungi dan batasi akses baca/tulis ke proses yang diotorisasi saja. Penyimpanan kustom bertanggung jawab atas kontrol akses, integritas, dan jaminan durabilitas mereka sendiri.

Langkah Selanjutnya

  • Last updated on