mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: Implement browse/clear functionality for state (#513)
* feat: Implement browse/clear functionality for state * fix: Assertion comments in StateManagerTests
This commit is contained in:
committed by
GitHub
Unverified
parent
1134350fcf
commit
80e89a7d87
@@ -27,6 +27,73 @@ internal class StateManager
|
||||
return scope;
|
||||
}
|
||||
|
||||
private IEnumerable<UpdateKey> GetUpdatesForScopeStrict(ScopeId scopeId)
|
||||
{
|
||||
Throw.IfNull(scopeId);
|
||||
|
||||
return this._queuedUpdates.Keys.Where(key => key.IsMatchingScope(scopeId, strict: true));
|
||||
}
|
||||
|
||||
public ValueTask ClearStateAsync(string executorId, string? scopeName)
|
||||
=> this.ClearStateAsync(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName));
|
||||
|
||||
public async ValueTask ClearStateAsync(ScopeId scopeId)
|
||||
{
|
||||
Throw.IfNull(scopeId);
|
||||
|
||||
if (this._scopes.TryGetValue(scopeId, out StateScope? scope))
|
||||
{
|
||||
HashSet<string> keysToDelete = await scope.ReadKeysAsync().ConfigureAwait(false);
|
||||
|
||||
foreach (UpdateKey updateKey in this.GetUpdatesForScopeStrict(scopeId))
|
||||
{
|
||||
StateUpdate update = this._queuedUpdates[updateKey];
|
||||
if (!update.IsDelete)
|
||||
{
|
||||
this._queuedUpdates[updateKey] = StateUpdate.Delete(update.Key);
|
||||
}
|
||||
|
||||
keysToDelete.Remove(update.Key);
|
||||
}
|
||||
|
||||
foreach (string key in keysToDelete)
|
||||
{
|
||||
UpdateKey updateKey = new(scopeId, key);
|
||||
this._queuedUpdates[updateKey] = StateUpdate.Delete(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private HashSet<string> ApplyUnpublishedUpdates(ScopeId scopeId, HashSet<string> keys)
|
||||
{
|
||||
// Apply any queued updates for this scope
|
||||
foreach (UpdateKey key in this.GetUpdatesForScopeStrict(scopeId))
|
||||
{
|
||||
StateUpdate update = this._queuedUpdates[key];
|
||||
if (update.IsDelete)
|
||||
{
|
||||
keys.Remove(update.Key);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Add is idempotent on Sets
|
||||
keys.Add(update.Key);
|
||||
}
|
||||
}
|
||||
|
||||
return keys;
|
||||
}
|
||||
|
||||
public ValueTask<HashSet<string>> ReadKeysAsync(string executorId, string? scopeName = null)
|
||||
=> this.ReadKeysAsync(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName));
|
||||
|
||||
public async ValueTask<HashSet<string>> ReadKeysAsync(ScopeId scopeId)
|
||||
{
|
||||
StateScope scope = this.GetOrCreateScope(scopeId);
|
||||
HashSet<string> keys = await scope.ReadKeysAsync().ConfigureAwait(false);
|
||||
return this.ApplyUnpublishedUpdates(scopeId, keys);
|
||||
}
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string executorId, string? scopeName, string key)
|
||||
=> this.ReadStateAsync<T>(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key);
|
||||
|
||||
@@ -91,7 +158,7 @@ internal class StateManager
|
||||
stateUpdates.Add(this._queuedUpdates[key]);
|
||||
}
|
||||
|
||||
if (updatesByScope.Count > 0 && tracer != null)
|
||||
if (tracer != null && (updatesByScope.Count > 0))
|
||||
{
|
||||
tracer.TraceStatePublished();
|
||||
}
|
||||
|
||||
@@ -23,6 +23,13 @@ internal class StateScope
|
||||
{
|
||||
}
|
||||
|
||||
public ValueTask<HashSet<string>> ReadKeysAsync()
|
||||
{
|
||||
HashSet<string> keys = new(this._stateData.Keys, this._stateData.Comparer);
|
||||
|
||||
return new(keys);
|
||||
}
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string key)
|
||||
{
|
||||
Throw.IfNullOrEmpty(key);
|
||||
@@ -40,7 +47,7 @@ internal class StateScope
|
||||
|
||||
foreach (string key in updates.Keys)
|
||||
{
|
||||
if (updates[key].Count == 0)
|
||||
if (updates == null || updates[key].Count == 0)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
@@ -50,14 +57,14 @@ internal class StateScope
|
||||
throw new InvalidOperationException($"Expected exactly one update for key '{key}'.");
|
||||
}
|
||||
|
||||
StateUpdate upadte = updates[key][0];
|
||||
if (upadte.IsDelete)
|
||||
StateUpdate update = updates[key][0];
|
||||
if (update.IsDelete)
|
||||
{
|
||||
this._stateData.Remove(key);
|
||||
}
|
||||
else
|
||||
{
|
||||
this._stateData[key] = upadte.Value!;
|
||||
this._stateData[key] = update.Value!;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -29,13 +29,17 @@ internal class UpdateKey(ScopeId scopeId, string key)
|
||||
return $"{this.ScopeId}/{this.Key}";
|
||||
}
|
||||
|
||||
public bool IsMatchingScope(ScopeId scopeId, bool strict = false)
|
||||
{
|
||||
return this.ScopeId == scopeId && (!strict || this.ScopeId.ExecutorId == scopeId.ExecutorId);
|
||||
}
|
||||
|
||||
public override bool Equals(object? obj)
|
||||
{
|
||||
if (obj is UpdateKey other)
|
||||
{
|
||||
// Unlike ScopeId, UpdateKey is equal only if both the Executor and ScopeName are the same
|
||||
return this.ScopeId.ExecutorId == other.ScopeId.ExecutorId &&
|
||||
this.ScopeId.ScopeName == other.ScopeId.ScopeName &&
|
||||
return this.IsMatchingScope(other.ScopeId, strict: true) &&
|
||||
this.Key == other.Key;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace Microsoft.Agents.Workflows;
|
||||
@@ -33,10 +34,18 @@ public interface IWorkflowContext
|
||||
/// </summary>
|
||||
/// <typeparam name="T">The type of the state value.</typeparam>
|
||||
/// <param name="key">The key of the state value.</param>
|
||||
/// <param name="scopeName">The name of the scope.</param>
|
||||
/// <param name = "scopeName" > An optional name that specifies the scope to read.If null, the default scope is
|
||||
/// used.</param>
|
||||
/// <returns>A <see cref="ValueTask{T}"/> representing the asynchronous operation.</returns>
|
||||
ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null);
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronously reads all state keys within the specified scope.
|
||||
/// </summary>
|
||||
/// <param name="scopeName">An optional name that specifies the scope to read. If null, the default scope is
|
||||
/// used.</param>
|
||||
ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null);
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronously updates the state of a queue entry identified by the specified key and optional scope.
|
||||
/// </summary>
|
||||
@@ -48,8 +57,21 @@ public interface IWorkflowContext
|
||||
/// <param name="key">The unique identifier for the queue entry to update. Cannot be null or empty.</param>
|
||||
/// <param name="value">The value to set for the queue entry. If null, the entry's state may be cleared or reset depending on
|
||||
/// implementation.</param>
|
||||
/// <param name="scopeName">An optional name that specifies the scope within which the queue entry resides. If null, the default scope is
|
||||
/// <param name="scopeName">An optional name that specifies the scope to update. If null, the default scope is
|
||||
/// used.</param>
|
||||
/// <returns>A ValueTask that represents the asynchronous update operation.</returns>
|
||||
ValueTask QueueStateUpdateAsync<T>(string key, T? value, string? scopeName = null);
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronously clears all state entries within the specified scope.
|
||||
///
|
||||
/// This semantically equivalent to retrieving all keys in the scope and deleting them one-by-one.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Subsequent reads by this executor will not find any entries in the cleared scope. Other executors will only
|
||||
/// see the cleared state starting from the next SuperStep.
|
||||
/// </remarks>
|
||||
/// <param name="scopeName">An optional name that specifies the scope to clear. If null, the default scope is used.</param>
|
||||
/// <returns>A ValueTask that represents the asynchronous clear operation.</returns>
|
||||
ValueTask QueueClearScopeAsync(string? scopeName = null);
|
||||
}
|
||||
|
||||
@@ -105,11 +105,17 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
|
||||
public ValueTask AddEventAsync(WorkflowEvent workflowEvent) => RunnerContext.AddEventAsync(workflowEvent);
|
||||
public ValueTask SendMessageAsync(object message, string? targetId = null) => RunnerContext.SendMessageAsync(ExecutorId, message, targetId);
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null)
|
||||
=> RunnerContext.StateManager.ReadStateAsync<T>(ExecutorId, scopeName, key);
|
||||
|
||||
public ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null)
|
||||
=> RunnerContext.StateManager.ReadKeysAsync(ExecutorId, scopeName);
|
||||
|
||||
public ValueTask QueueStateUpdateAsync<T>(string key, T? value, string? scopeName = null)
|
||||
=> RunnerContext.StateManager.WriteStateAsync(ExecutorId, scopeName, key, value);
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null)
|
||||
=> RunnerContext.StateManager.ReadStateAsync<T>(ExecutorId, scopeName, key);
|
||||
public ValueTask QueueClearScopeAsync(string? scopeName = null)
|
||||
=> RunnerContext.StateManager.ClearStateAsync(ExecutorId, scopeName);
|
||||
}
|
||||
|
||||
internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default)
|
||||
|
||||
@@ -52,6 +52,28 @@ public class ScopeId(string executorId, string? scopeName = null)
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public static bool operator ==(ScopeId? left, ScopeId? right)
|
||||
{
|
||||
if (left is null && right == null)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (right is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// The inversion here is necessary because the null analysis is incapable of proving to itself
|
||||
// that left cannot be null here: If it was, either right is null, and we returned true, or right
|
||||
// is not null, and we returned false.
|
||||
return right.Equals(left);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public static bool operator !=(ScopeId? left, ScopeId? right) => !(left == right);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override int GetHashCode()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user