Repository navigation
.NET: Add Redis checkpoint store for workflows #9253
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Draft
Murat Ay (Murat7Ay)
wants to merge
2
commits into
microsoft:main
Choose a base branch
from
Murat7Ay:dotnet-redis-checkpoint-store
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+1,307
−0
Draft
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
dotnet/samples/03-workflows/Checkpoint/CheckpointWithRedis/CheckpointWithRedis.csproj
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFrameworks>net10.0</TargetFrameworks> | ||
|
|
||
| <Nullable>enable</Nullable> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.Redis\Microsoft.Agents.AI.Redis.csproj" /> | ||
| <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
90 changes: 90 additions & 0 deletions
90
dotnet/samples/03-workflows/Checkpoint/CheckpointWithRedis/Program.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
|
|
||
| using Microsoft.Agents.AI.Workflows; | ||
| using Microsoft.Agents.AI.Workflows.Checkpointing; | ||
| using StackExchange.Redis; | ||
|
|
||
| namespace WorkflowCheckpointWithRedisSample; | ||
|
|
||
| /// <summary> | ||
| /// This sample shows how to store workflow checkpoints in Redis with <see cref="RedisCheckpointStore"/>, | ||
| /// so that a workflow can be resumed by another process, container or machine. | ||
| /// Key concepts: | ||
| /// - The store keeps the checkpoints of each session in Redis, in the order they were committed. | ||
| /// - Any process that connects to the same Redis server can find the latest checkpoint of a session | ||
| /// and resume the workflow from it. | ||
| /// - An optional time to live removes the checkpoints of a session some time after its last checkpoint. | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// Pre-requisites: | ||
| /// - Foundational samples should be completed first. | ||
| /// - A Redis server, for example: docker run -d -p 6379:6379 redis:7-alpine | ||
| /// - Optionally set REDIS_CONNECTION_STRING (defaults to localhost:6379). | ||
| /// </remarks> | ||
| public static class Program | ||
| { | ||
| private static async Task Main() | ||
| { | ||
| string connectionString = Environment.GetEnvironmentVariable("REDIS_CONNECTION_STRING") ?? "localhost:6379"; | ||
| string sessionId = $"guess-number-{Guid.NewGuid():N}"; | ||
|
|
||
| // <run_with_redis_checkpoints> | ||
| // The application owns the connection; share one ConnectionMultiplexer per application. | ||
| await using ConnectionMultiplexer redis = await ConnectionMultiplexer.ConnectAsync(connectionString); | ||
|
|
||
| RedisCheckpointStoreOptions options = new() | ||
| { | ||
| KeyPrefix = "samples:checkpoints", | ||
| TimeToLive = TimeSpan.FromHours(1), | ||
| }; | ||
| CheckpointManager checkpointManager = CheckpointManager.CreateJson(new RedisCheckpointStore(redis, options)); | ||
|
|
||
| Workflow workflow = WorkflowFactory.BuildWorkflow(); | ||
| await using (StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, NumberSignal.Init, checkpointManager, sessionId)) | ||
| { | ||
| // Stop the run part way through, as if the process had crashed. | ||
| int checkpointCount = 0; | ||
| await foreach (WorkflowEvent evt in run.WatchStreamAsync()) | ||
| { | ||
| if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: not null } && ++checkpointCount == 4) | ||
| { | ||
| Console.WriteLine($"Stopping the first run after {checkpointCount} checkpoints."); | ||
| break; | ||
| } | ||
| } | ||
| } | ||
| // </run_with_redis_checkpoints> | ||
|
|
||
| // <resume_from_redis> | ||
| // Later, possibly in another process: create a new store and workflow, find the latest | ||
| // checkpoint of the session in Redis and resume the workflow from it. | ||
| CheckpointManager resumeManager = CheckpointManager.CreateJson(new RedisCheckpointStore(redis, options)); | ||
| CheckpointInfo latest = await resumeManager.GetLatestCheckpointAsync(sessionId) | ||
| ?? throw new InvalidOperationException($"No checkpoint found for session '{sessionId}'."); | ||
| Console.WriteLine($"Resuming session '{sessionId}' from checkpoint '{latest.CheckpointId}'."); | ||
|
|
||
| Workflow resumedWorkflow = WorkflowFactory.BuildWorkflow(); | ||
| await using StreamingRun resumedRun = await InProcessExecution.ResumeStreamingAsync(resumedWorkflow, latest, resumeManager); | ||
| // </resume_from_redis> | ||
|
|
||
| await foreach (WorkflowEvent evt in resumedRun.WatchStreamAsync()) | ||
| { | ||
| switch (evt) | ||
| { | ||
| case ExecutorCompletedEvent executorCompleted: | ||
| Console.WriteLine($"* Executor {executorCompleted.ExecutorId} completed."); | ||
| break; | ||
|
|
||
| case WorkflowOutputEvent output: | ||
| Console.WriteLine($"Workflow completed with result: {output.Data}"); | ||
| break; | ||
|
|
||
| case WorkflowErrorEvent workflowError: | ||
| Console.ForegroundColor = ConsoleColor.Red; | ||
| Console.Error.WriteLine(workflowError.Exception?.ToString() ?? "Unknown workflow error occurred."); | ||
| Console.ResetColor(); | ||
| break; | ||
| } | ||
| } | ||
| } | ||
| } | ||
23 changes: 23 additions & 0 deletions
23
dotnet/samples/03-workflows/Checkpoint/CheckpointWithRedis/README.md
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| # Checkpoint with Redis | ||
|
|
||
| This sample stores workflow checkpoints in Redis with `RedisCheckpointStore` from the `Microsoft.Agents.AI.Redis` package, so a workflow can be resumed by another process, container or machine. | ||
|
|
||
| It runs a number guessing workflow, stops it after a few super steps as if the process had crashed, then creates a new store and workflow, finds the latest checkpoint of the session in Redis with `CheckpointManager.GetLatestCheckpointAsync` and resumes from it. | ||
|
|
||
| ## Prerequisites | ||
|
|
||
| - A Redis server, for example: | ||
|
|
||
| ```bash | ||
| docker run -d -p 6379:6379 redis:7-alpine | ||
| ``` | ||
|
|
||
| - Optionally set `REDIS_CONNECTION_STRING` (defaults to `localhost:6379`). | ||
|
|
||
| ## Run the sample | ||
|
|
||
| ```bash | ||
| dotnet run | ||
| ``` | ||
|
|
||
| The checkpoints are written under the `samples:checkpoints` key prefix and expire one hour after the last checkpoint of a session. |
126 changes: 126 additions & 0 deletions
126
dotnet/samples/03-workflows/Checkpoint/CheckpointWithRedis/WorkflowFactory.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
|
|
||
| using Microsoft.Agents.AI.Workflows; | ||
|
|
||
| namespace WorkflowCheckpointWithRedisSample; | ||
|
|
||
| internal static class WorkflowFactory | ||
| { | ||
| /// <summary> | ||
| /// Get a workflow that plays a number guessing game with checkpointing support. | ||
| /// The workflow consists of two executors that are connected in a feedback loop: | ||
| /// 1. GuessNumberExecutor: Makes a guess based on the current known bounds. | ||
| /// 2. JudgeExecutor: Evaluates the guess and provides feedback. | ||
| /// The workflow continues until the correct number is guessed. | ||
| /// </summary> | ||
| internal static Workflow BuildWorkflow() | ||
| { | ||
| GuessNumberExecutor guessNumberExecutor = new(); | ||
| JudgeExecutor judgeExecutor = new(42); | ||
|
|
||
| return new WorkflowBuilder(guessNumberExecutor) | ||
| .AddEdge(guessNumberExecutor, judgeExecutor) | ||
| .AddEdge(judgeExecutor, guessNumberExecutor) | ||
| .WithOutputFrom(judgeExecutor) | ||
| .Build(); | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Signals used for communication between GuessNumberExecutor and JudgeExecutor. | ||
| /// </summary> | ||
| internal enum NumberSignal | ||
| { | ||
| Init, | ||
| Above, | ||
| Below, | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// The guessing range. Executor state is stored in the checkpoints as JSON. | ||
| /// </summary> | ||
| internal sealed record GuessRange(int LowerBound, int UpperBound); | ||
|
|
||
| /// <summary> | ||
| /// Executor that makes a guess based on the current bounds. | ||
| /// </summary> | ||
| [SendsMessage(typeof(int))] | ||
| internal sealed class GuessNumberExecutor() : Executor<NumberSignal>("Guess") | ||
| { | ||
| private const string StateKey = "GuessNumberExecutorState"; | ||
|
|
||
| private GuessRange _range = new(1, 100); | ||
|
|
||
| private int NextGuess => (this._range.LowerBound + this._range.UpperBound) / 2; | ||
|
|
||
| public override async ValueTask HandleAsync(NumberSignal message, IWorkflowContext context, CancellationToken cancellationToken = default) | ||
| { | ||
| this._range = message switch | ||
| { | ||
| NumberSignal.Above => this._range with { UpperBound = this.NextGuess - 1 }, | ||
| NumberSignal.Below => this._range with { LowerBound = this.NextGuess + 1 }, | ||
| _ => this._range, | ||
| }; | ||
|
|
||
| await context.SendMessageAsync(this.NextGuess, cancellationToken: cancellationToken); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Checkpoint the current state of the executor. | ||
| /// </summary> | ||
| protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => | ||
| context.QueueStateUpdateAsync(StateKey, this._range, cancellationToken: cancellationToken); | ||
|
|
||
| /// <summary> | ||
| /// Restore the state of the executor from a checkpoint. | ||
| /// </summary> | ||
| protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => | ||
| this._range = await context.ReadStateAsync<GuessRange>(StateKey, cancellationToken: cancellationToken) ?? this._range; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Executor that judges the guess and provides feedback. | ||
| /// </summary> | ||
| [SendsMessage(typeof(NumberSignal))] | ||
| [YieldsOutput(typeof(string))] | ||
| internal sealed class JudgeExecutor() : Executor<int>("Judge") | ||
| { | ||
| private const string StateKey = "JudgeExecutorState"; | ||
|
|
||
| private readonly int _targetNumber; | ||
| private int _tries; | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="JudgeExecutor"/> class. | ||
| /// </summary> | ||
| /// <param name="targetNumber">The number to be guessed.</param> | ||
| public JudgeExecutor(int targetNumber) : this() | ||
| { | ||
| this._targetNumber = targetNumber; | ||
| } | ||
|
|
||
| public override async ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default) | ||
| { | ||
| this._tries++; | ||
| if (message == this._targetNumber) | ||
| { | ||
| await context.YieldOutputAsync($"{this._targetNumber} found in {this._tries} tries!", cancellationToken: cancellationToken); | ||
| } | ||
| else | ||
| { | ||
| await context.SendMessageAsync(message < this._targetNumber ? NumberSignal.Below : NumberSignal.Above, cancellationToken: cancellationToken); | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Checkpoint the current state of the executor. | ||
| /// </summary> | ||
| protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => | ||
| context.QueueStateUpdateAsync(StateKey, this._tries, cancellationToken: cancellationToken); | ||
|
|
||
| /// <summary> | ||
| /// Restore the state of the executor from a checkpoint. | ||
| /// </summary> | ||
| protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => | ||
| this._tries = await context.ReadStateAsync<int>(StateKey, cancellationToken: cancellationToken); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
dotnet/src/Microsoft.Agents.AI.Redis/Microsoft.Agents.AI.Redis.csproj
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFrameworks>$(TargetFrameworksCore)</TargetFrameworks> | ||
| <RootNamespace>Microsoft.Agents.AI.Redis</RootNamespace> | ||
| <VersionSuffix>alpha</VersionSuffix> | ||
| </PropertyGroup> | ||
|
|
||
| <PropertyGroup> | ||
| <InjectSharedThrow>true</InjectSharedThrow> | ||
| </PropertyGroup> | ||
|
|
||
| <Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" /> | ||
| <PropertyGroup> | ||
| <!-- Disable packing until we are ready to release this as a nuget --> | ||
| <IsPackable>false</IsPackable> | ||
| </PropertyGroup> | ||
|
|
||
| <PropertyGroup> | ||
| <!-- NuGet Package Settings --> | ||
| <Title>Microsoft Agent Framework - Redis integration</Title> | ||
| <Description>Provides Redis implementations for Microsoft Agent Framework storage abstractions, starting with a workflow CheckpointStore built on StackExchange.Redis.</Description> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="StackExchange.Redis" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.