# CheckPoints in Microsoft Agent Workflow

Imagine you have a Microsoft Agent Workflow that continuously processes a high-volume stream of incoming data. The data flows through multiple executors responsible for validation, summarization and downstream actions.

Now suppose that after processing thousands of events, you realize that a particular subset of the stream data failed critical checks. Now you would want to re-execute those events.

But there is a catch. The original stream data is no longer available.

The workflow processed the data in real time without workflow checkpoints. Of course the incoming data is available in a persistent data storage but that would mean recreating the stream to the input source to be re processed but it can be challenging and to an extent cumbersome.

Also imagine a different scenario where in an a operation goes through a sequence of executors and during execution one of the executors fails. In such a case, the input data associated with the failed operation cannot simply be replayed or reprocessed from the beginning because the original execution context may no longer be available.

This becomes even more challenging in long-running or streaming workflows where the operation may have already passed through multiple intermediate stages before the failure occurred. Restarting the entire workflow from scratch can lead to unnecessary re computation, duplicate processing, increased operational cost and possible data inconsistencies.

These are the most important challenges in streaming AI workflows and distributed agent systems. All you need in such scenarios is an native inbuilt AI workflow capability to rerun the process for those events without requiring them to be recreated manually. This is where MAF Checkpoints help.

To demonstrate MAF checkpoints we will use the use case from my earlier [article](https://www.azureguru.net/deep-dive-into-workflow-execution-in-microsoft-agent-framework).

### CheckpointManager

The most important component in MAF checkpointing is the **CheckpointManager**.

It is responsible for storing and retrieval of the workflow checkpoint execution. It acts as one of the parameter to **RunStreamingAsync** of workflow execution. Setting this up enables the workflow to maintain checkpoints during execution.

The **CheckpointManager** can create a JSON object or store the checkpoint details in memory. You can use the JSON option incase you want to store the checkpoint details to persistent storage.

In this article we will create a workflow checkpoint in JSON format and store it on the filesystem.

### Structure

The checkpoint manager creates an **index.jsonl** file that maintains a mapping between the session ID and the checkpoint ID.

**index.jsonl** is always used as reference for all executions and each checkpoint entries are appended into it as the workflow progresses through multiple executions.

**First Execution:**

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/83e1b48f-1cd6-481b-a14a-d1f8b86719b5.png align="center")

**Second Execution:**

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/2e40da05-4b98-431b-ae8f-2e108fbf48c1.png align="center")

**Third Execution:**

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/e643d0a3-6fb5-49db-85db-73525685f807.png align="center")

If you observe the third session, you will realize that the number of checkpoint ID's are comparatively more than the previous two executions. This is because the input data for the third execution had to go through an additional executor compared to the previous two executions.

The no of checkpoint ID's created correspond to the no of executor's the workflow passes through.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/d05fcb64-33fa-434f-979b-611977f2b81f.png align="center")

### Code

As mentioned earlier, we will reuse the example from my previous [article](https://www.azureguru.net/deep-dive-into-workflow-execution-in-microsoft-agent-framework) and fit in the checkpoint manager into the code from that article.

We create object of **FileSystemJsonCheckpointStore** that persists the checkpoint data on the disk in JSON format.

```csharp
string input = "13";
DirectoryInfo dirInfo = new DirectoryInfo("Json Directory location");
var store = new FileSystemJsonCheckpointStore(dirinfo);
```

and then create an instance of the**CheckpointManager** from the store.

```csharp
CheckpointManager checkpointManager = CheckpointManager.CreateJson(store);
```

and finally use that instance of the**CheckpointManager** in **InProcessExecution**.

```csharp
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, input), checkpointManager);
```

Assume that we executed the workflow for an input value of **13** that resulted in the following output along with creation of multiple checkpoints in JSON format.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/f2f08628-5e82-43c7-a8fb-6c2ef08da89b.png align="center")

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/c41f9f5c-0c40-4ef7-8646-73c79e41f3b0.png align="center")

In the next execution we re run the workflow for an input value of **"a"**, as expected it results in the following output

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/8d891822-f8d2-453c-b700-258fc48a91ff.png align="center")

This resulted in creation of two additional checkpoint files.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/8e21549d-9437-4132-8c84-b6b333928450.png align="center")

We now want to re-execute the workflow for value of **13** through checkpoints.

The session ID of execution for **13** is **49d7868b625943a59a804c6c3363e1b7** and the checkpoint ID's are

*   **854b727b737b4d31be82e265564d66ab**
    
*   **55ffcd9eacc54efca3ba38e23ead83bf**
    
*   **7e2007b7c29c4bc2884b67e6d92fcc91**
    

First check if **index.jsonl** exists in the directory and if it does , then ensure that the file is not empty.

```csharp
DirectoryInfo dirinfo = new DirectoryInfo("Json Directory location");

string filepath = Path.Combine(dirinfo.FullName, "index.jsonl");

var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };

var list = new List<CheckpointDetails>();

if (File.Exists(filepath))
{
    foreach (var line in File.ReadLines(filepath))
    {
        var item = JsonSerializer.Deserialize<CheckpointDetails>(line, options);
        if (item != null)
        {
            list.Add(item);
        }
    }
}
```

If confirmed that file isn't empty, instance of **FileSystemJsonCheckpointStore** is created against the underlying checkpoint directory.

```csharp
 if (list.Count > 0)
{
    var store_checkpoint = new FileSystemJsonCheckpointStore(dirinfo);

    CheckpointInfo chk = new CheckpointInfo(SessionId, CheckpointId);
    CheckpointManager checkpointManagers = CheckpointManager.CreateJson(store_checkpoint);
    await using StreamingRun runa = await InProcessExecution.ResumeStreamingAsync(workflow, chk, checkpointManagers);
    await runa.RestoreCheckpointAsync(chk, CancellationToken.None).ConfigureAwait(false);

    await foreach (WorkflowEvent evt in runa.WatchStreamAsync().ConfigureAwait(false))
    {
        if (evt is SuperStepCompletedEvent completedEvent)
        {
            break;
        }

        if (evt is ExecutorInvokedEvent outputEvent)
        {
            var property = outputEvent.Data.GetType().GetProperty("Reason");

            var reason = property?.GetValue(outputEvent.Data);

            Console.WriteLine(reason);
        }
    }

    store_checkpoint.Dispose();
}
```

The iteration is required so that **InProcessExecution** processes all the recorded checkpoints for the provided Session ID.

Incase you retrieve all the checkpoints for a provided **SessionId**

```csharp
await GetCheckpointList(dirinfo, workflow, "SessionId");
```

**GetCheckpointList >>**

```csharp
public static async Task GetCheckpointList(DirectoryInfo dirinfo, Workflow workflow, string SessionId)
 {
     var store_checkpoint = new FileSystemJsonCheckpointStore(dirinfo);

     List<CheckpointInfo> checkpoints_t = [.. await store_checkpoint.RetrieveIndexAsync(SessionId!)];

     foreach (var session in checkpoints_t)
     {
         if (session.SessionId != SessionId) { continue; }            

     }

 }
```

You might wonder what the need for the following check

```csharp
if (chk.SessionId != sessionId) { continue; }
```

in the function when the checkpoints for a given **SessionId** is already filtered by the following line of code.

```csharp
List<CheckpointInfo> checkpoints_t = [.. await store_checkpoint.RetrieveIndexAsync(sessionId!)];
```

This is because of a bug related to the behavior of **RetrieveIndexAsync** where it does not filter the Session list for the provided **SessionId.**

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/ebc24272-cf6b-4599-a536-6851c67183bc.png align="center")

> I have raised this issue on the official Microsoft Agent Framework GitHub repository. The fix is in process and the issue can be tracked here [https://github.com/microsoft/agent-framework/issues/5942](https://github.com/microsoft/agent-framework/issues/5942)

Back to our main function **CreateCheckpointAsync** where we pass **SessionId** and **CheckPointId** values.

```csharp
public static async Task CreateCheckpointAsync(DirectoryInfo dirinfo, Workflow workflow, string SessionId, string CheckpointId)
{

    string filepath = Path.Combine(dirinfo.FullName, "index.jsonl");
    var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
    var list = new List<CheckpointDetails>();

    if (File.Exists(filepath))
    {
        foreach (var line in File.ReadLines(filepath))
        {
            var item = JsonSerializer.Deserialize<CheckpointDetails>(line, options);
            if (item != null)
            {
                list.Add(item);
            }
        }
    }

    if (list.Count > 0)
    {
        var store_checkpoint = new FileSystemJsonCheckpointStore(dirinfo);

        CheckpointInfo chk = new CheckpointInfo(SessionId, CheckpointId);
        CheckpointManager checkpointManagers = CheckpointManager.CreateJson(store_checkpoint);
        await using StreamingRun runa = await InProcessExecution.ResumeStreamingAsync(workflow, chk, checkpointManagers);
        await runa.RestoreCheckpointAsync(chk, CancellationToken.None).ConfigureAwait(false);

        await foreach (WorkflowEvent evt in runa.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is SuperStepCompletedEvent completedEvent)
            {
                break;
            }

            if (evt is ExecutorInvokedEvent outputEvent)
            {
                var property = outputEvent.Data.GetType().GetProperty("Reason");

                var reason = property?.GetValue(outputEvent.Data);

                Console.WriteLine(reason);
            }

        }

        store_checkpoint.Dispose();
    }

}
```

Now lets pick a checkpoint that we want to execute from. This is our workflow path.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/935e7dbd-9093-4351-8ac2-f3e41132a516.png align="center")

Since we want to execute the checkpoint for an input value of **13**, the workflow will follow the **PrimeNumber** path. As a result, there are two executors involved in this execution path.

*   **PrimeNo\_SquareRootExecutor**
    
*   **SendDetails\_Executor\_RouteA**
    

If you open the checkpoint files you will notice that there is a property named `stepNumber` that helps identify the sequence of executor execution.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/dc471fb4-1dcc-427d-8772-5f17d42948ae.png align="center")

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/65d9e91a-a1bb-42ce-a542-860a5368c28a.png align="center")

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/0adb20bf-53e6-405c-b323-52ff4c1a3dc3.png align="center")

The most important property in the checkpoint file is **queuedMessages**. It helps identify which executor is next in line for execution within the workflow pipeline.

For step 0 in the checkpoint file we have the **SquareRootExecutor** under **queuedMessages**.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/55b83497-62da-4b5e-a9ac-ecaa9704c8db.png align="center")

For step 1 in checkpoint file we have **PrimeNo\_SendDetails\_RouteA\_Executor** under **queuedMessages**.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/72163f79-47ff-415d-921b-890d4bc5c90e.png align="center")

**Note** : In checkpoint files, the executor name corresponds to the actual executor name defined in the code, rather than the object name used to reference it.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/04d2349b-f252-4fa9-8964-a3bf27678de8.png align="center")

Lets execute for SessionId **49d7868b625943a59a804c6c3363e1b7** and CheckpointId **7e2007b7c29c4bc2884b67e6d92fcc91** .

To test the functionality lets make some change to the checkpoint file corresponding to the CheckPointId **7e2007b7c29c4bc2884b67e6d92fcc91** . In the checkpoint file I will make some changes to the **reason** property.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/a2ad34f3-fddb-43e7-9726-2ccafd40977b.png align="center")

I will append additional text : **Hello from Mumbai** to the value of reason property.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/2d73723d-3b92-41b4-9896-52141de3cdbb.png align="center")

and when I reference this checkpoint file in the checkpoint manager , the output should be

*"The square root of 13 is a non-integer value and approximately equals 3.6055512755.Hello from Mumbai"*

instead of just

*"The square root of 13 is a non-integer value and approximately equals 3.6055512755"*

which was the output from the original execution.

**Execution >>**

```csharp
var workflow = builder.Build();
string input = "a";
DirectoryInfo dirinfo = new DirectoryInfo("Your checkpoint folder");
await CreateCheckpointAsync(dirinfo, workflow, "SessionID", "CheckpointID");
```

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/17bb9385-8941-43da-8afd-06002e2a5f31.gif align="center")

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/412791af-4315-4564-b9b5-c408ea02f8b8.png align="center")

**Very Important**

Workflow execution through checkpoint reference will create its own checkpoint file. The above execution resulted in creation of a new checkpoint file.

New checkpoint file

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/289db07d-78e5-4a45-ae3c-daf4a63aae98.png align="center")

the step no marked in this newly created checkpoint file is 0 with the reason property value picked up from the checkpoint file referenced in the previous execution.

![](https://cdn.hashnode.com/uploads/covers/6693c62c166ee9c594cffda0/0e5491b4-b8af-48b4-af6e-04bcbb169515.png align="center")

### Conclusion:

In conclusion, checkpoints in MAF act as persisted snapshots of runtime that could be restored for an in point in time execution.

Although there isn't any solid documentation on checkpoint in MAF for C# , I explored the functionality to best of my abilities. As a result, there may still be some inaccuracies or gaps in the explanation :)

On a side note , during my days when working extensively with SQL Server a number of times I used to manually run Checkpoint commands to flush out dirty pages from memory to disk (a complicated topic). Though not as similar to Checkpoints in MAF , I just remembered the Checkpoint concept of SQL Server.

Thanks for reading !!!