mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: Adds Valkey to chat message history - issue 5445 (#5542)
* Adds Valkey to chat message history * Address review: switch to Valkey.Glide, add options class, remove context provider - Switch from StackExchange.Redis to Valkey.Glide 1.1.0 (official Valkey .NET client) - Extract optional params into ValkeyChatHistoryProviderOptions - Add JsonSerializerOptions support, remove [RequiresUnreferencedCode] - Make MaxMessages/MaxMessagesToRetrieve readonly via options - Remove ValkeyContextProvider (overlaps with ChatHistoryMemoryProvider + MEVD) - Remove ValkeyProviderScope (only used by context provider) - Remove connection string constructors (caller manages IConnectionMultiplexer) - Update samples to use new API and gpt-5.4-mini * Use type-safe JsonSerializer overloads, remove suppress attributes Use JsonSerializerOptions.GetTypeInfo() for Serialize/Deserialize calls to enable NativeAOT/trimming compatibility without suppress attributes. Default to AgentAbstractionsJsonUtilities.DefaultOptions when no options provided. Signed-off-by: Matthias Howell <matthias.howell@improving.com> * Update READMEs: remove context provider references Remove ValkeyContextProvider and long-term memory references from sample READMEs since the context provider was removed from this PR. Simplify Valkey server requirements (no search module needed for chat history). Signed-off-by: Matthias Howell <matthias.howell@improving.com> * Apply suggestion from @westey-m * Fix formatting (dotnet format) Signed-off-by: Matthias Howell <matthias.howell@improving.com> * Update dotnet/src/Microsoft.Agents.AI.Valkey/Microsoft.Agents.AI.Valkey.csproj Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com> --------- Signed-off-by: Matthias Howell <matthias.howell@improving.com> Co-authored-by: Matthias Howell <matthias.howell@yoppworks.com> Co-authored-by: westey <164392973+westey-m@users.noreply.github.com> Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com>
This commit is contained in:
Matthias Howell
committed by
GitHub
Unverified
GitHub
parent
4149f24791
commit
8e1998ddcb
@@ -0,0 +1,40 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>$(TargetFrameworksCore)</TargetFrameworks>
|
||||
<RootNamespace>Microsoft.Agents.AI.Valkey</RootNamespace>
|
||||
<VersionSuffix>alpha</VersionSuffix>
|
||||
<NoWarn>$(NoWarn);CA1873</NoWarn>
|
||||
</PropertyGroup>
|
||||
|
||||
<PropertyGroup>
|
||||
<InjectSharedThrow>true</InjectSharedThrow>
|
||||
<InjectTrimAttributesOnLegacy>true</InjectTrimAttributesOnLegacy>
|
||||
</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 - Valkey integration</Title>
|
||||
<Description>Provides Valkey integration for Microsoft Agent Framework, including chat history persistence and context provider with full-text search.</Description>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Microsoft.Agents.AI.Abstractions\Microsoft.Agents.AI.Abstractions.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Valkey.Glide" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<InternalsVisibleTo Include="Microsoft.Agents.AI.Valkey.UnitTests" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,225 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.AI;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Shared.Diagnostics;
|
||||
using Valkey.Glide;
|
||||
|
||||
namespace Microsoft.Agents.AI.Valkey;
|
||||
|
||||
/// <summary>
|
||||
/// Provides a Valkey-backed implementation of <see cref="ChatHistoryProvider"/> for persistent chat history storage.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Uses basic Valkey list operations via Valkey.Glide.
|
||||
/// No search module is required — this provider works with any Valkey server.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <strong>Data retention:</strong> Stored messages have no TTL and persist indefinitely.
|
||||
/// Use <see cref="ValkeyChatHistoryProviderOptions.MaxMessages"/> to limit per-conversation storage, and <see cref="ClearMessagesAsync"/>
|
||||
/// for explicit cleanup. Callers are responsible for implementing data retention policies.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <strong>Security considerations:</strong>
|
||||
/// <list type="bullet">
|
||||
/// <item><description><strong>PII and sensitive data:</strong> Chat history stored in Valkey may contain PII and sensitive
|
||||
/// conversation content. Ensure the Valkey server is configured with appropriate access controls and encryption in transit
|
||||
/// (TLS). The <see cref="ValkeyChatHistoryProviderOptions.MaxMessages"/> property can limit stored messages per conversation.</description></item>
|
||||
/// <item><description><strong>Compromised store risks:</strong> Agent Framework does not validate or filter messages loaded
|
||||
/// from the store — they are accepted as-is. If the Valkey store is compromised, adversarial content could be injected
|
||||
/// into the conversation context.</description></item>
|
||||
/// </list>
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ValkeyChatHistoryProvider : ChatHistoryProvider
|
||||
{
|
||||
private readonly ProviderSessionState<State> _sessionState;
|
||||
private IReadOnlyList<string>? _stateKeys;
|
||||
private readonly IConnectionMultiplexer _connection;
|
||||
private readonly string _keyPrefix;
|
||||
private readonly int? _maxMessages;
|
||||
private readonly int? _maxMessagesToRetrieve;
|
||||
private readonly JsonSerializerOptions _jsonSerializerOptions;
|
||||
private readonly ILogger<ValkeyChatHistoryProvider>? _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ValkeyChatHistoryProvider"/> class.
|
||||
/// </summary>
|
||||
/// <param name="connection">An existing <see cref="IConnectionMultiplexer"/> instance.</param>
|
||||
/// <param name="stateInitializer">A delegate that initializes the provider state on the first invocation.</param>
|
||||
/// <param name="options">Optional configuration options.</param>
|
||||
/// <param name="loggerFactory">Optional logger factory.</param>
|
||||
public ValkeyChatHistoryProvider(
|
||||
IConnectionMultiplexer connection,
|
||||
Func<AgentSession?, State> stateInitializer,
|
||||
ValkeyChatHistoryProviderOptions? options = null,
|
||||
ILoggerFactory? loggerFactory = null)
|
||||
: base(options?.ProvideOutputMessageFilter, options?.StoreInputRequestMessageFilter, options?.StoreInputResponseMessageFilter)
|
||||
{
|
||||
this._sessionState = new ProviderSessionState<State>(
|
||||
Throw.IfNull(stateInitializer),
|
||||
options?.StateKey ?? this.GetType().Name,
|
||||
options?.JsonSerializerOptions);
|
||||
this._connection = Throw.IfNull(connection);
|
||||
this._keyPrefix = options?.KeyPrefix ?? "chat_history";
|
||||
this._maxMessages = options?.MaxMessages;
|
||||
this._maxMessagesToRetrieve = options?.MaxMessagesToRetrieve;
|
||||
this._jsonSerializerOptions = options?.JsonSerializerOptions ?? AgentAbstractionsJsonUtilities.DefaultOptions;
|
||||
this._logger = loggerFactory?.CreateLogger<ValkeyChatHistoryProvider>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override IReadOnlyList<string> StateKeys => this._stateKeys ??= [this._sessionState.StateKey];
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Throw.IfNull(context);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
var state = this._sessionState.GetOrInitializeState(context.Session);
|
||||
var db = this._connection.GetDatabase();
|
||||
var key = this.BuildKey(state);
|
||||
|
||||
// Fetch only the tail when MaxMessagesToRetrieve is set [Low: avoid fetching all then trimming]
|
||||
ValkeyValue[] values;
|
||||
if (this._maxMessagesToRetrieve.HasValue)
|
||||
{
|
||||
values = await db.ListRangeAsync(key, -this._maxMessagesToRetrieve.Value, -1).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
values = await db.ListRangeAsync(key).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var messages = new List<ChatMessage>(values.Length);
|
||||
|
||||
foreach (var value in values)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
if (value.IsNullOrEmpty)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var message = JsonSerializer.Deserialize(value.ToString(), this._jsonSerializerOptions.GetTypeInfo(typeof(ChatMessage))) as ChatMessage;
|
||||
if (message is not null)
|
||||
{
|
||||
messages.Add(message);
|
||||
}
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
// Skip malformed entries rather than crashing the session [VERIFY-002]
|
||||
this._logger?.LogWarning(ex, "ValkeyChatHistoryProvider: Skipping malformed message in conversation '{ConversationId}'.", state.ConversationId);
|
||||
}
|
||||
}
|
||||
|
||||
this._logger?.LogInformation(
|
||||
"ValkeyChatHistoryProvider: Retrieved {Count} messages for conversation.",
|
||||
messages.Count);
|
||||
|
||||
return messages;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Throw.IfNull(context);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
var state = this._sessionState.GetOrInitializeState(context.Session);
|
||||
var messageList = context.RequestMessages.Concat(context.ResponseMessages ?? []).ToList();
|
||||
if (messageList.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var db = this._connection.GetDatabase();
|
||||
var key = this.BuildKey(state);
|
||||
|
||||
// Batch push — single round-trip [Medium-8]
|
||||
var serialized = new ValkeyValue[messageList.Count];
|
||||
for (int i = 0; i < messageList.Count; i++)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
serialized[i] = JsonSerializer.Serialize(messageList[i], this._jsonSerializerOptions.GetTypeInfo(typeof(ChatMessage)));
|
||||
}
|
||||
|
||||
await db.ListRightPushAsync(key, serialized).ConfigureAwait(false);
|
||||
|
||||
// Trim to max messages if configured
|
||||
if (this._maxMessages.HasValue)
|
||||
{
|
||||
await db.ListTrimAsync(key, -this._maxMessages.Value, -1).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
this._logger?.LogInformation(
|
||||
"ValkeyChatHistoryProvider: Stored {Count} messages for conversation.",
|
||||
messageList.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Clears all messages for the specified session's conversation.
|
||||
/// </summary>
|
||||
/// <param name="session">The session containing the conversation state.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>A task representing the asynchronous operation.</returns>
|
||||
public async Task ClearMessagesAsync(AgentSession? session, CancellationToken cancellationToken = default)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
var state = this._sessionState.GetOrInitializeState(session);
|
||||
var db = this._connection.GetDatabase();
|
||||
var key = this.BuildKey(state);
|
||||
await db.KeyDeleteAsync(key).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the count of stored messages for the specified session's conversation.
|
||||
/// </summary>
|
||||
/// <param name="session">The session containing the conversation state.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>The number of stored messages.</returns>
|
||||
public async Task<long> GetMessageCountAsync(AgentSession? session, CancellationToken cancellationToken = default)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
var state = this._sessionState.GetOrInitializeState(session);
|
||||
var db = this._connection.GetDatabase();
|
||||
var key = this.BuildKey(state);
|
||||
return await db.ListLengthAsync(key).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private string BuildKey(State state) => $"{this._keyPrefix}:{state.ConversationId}";
|
||||
|
||||
/// <summary>
|
||||
/// Represents the per-session state of a <see cref="ValkeyChatHistoryProvider"/>.
|
||||
/// </summary>
|
||||
public sealed class State
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="State"/> class.
|
||||
/// </summary>
|
||||
/// <param name="conversationId">The unique identifier for this conversation thread.</param>
|
||||
[JsonConstructor]
|
||||
public State(string conversationId)
|
||||
{
|
||||
this.ConversationId = Throw.IfNullOrWhitespace(conversationId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the conversation ID associated with this state.
|
||||
/// </summary>
|
||||
public string ConversationId { get; }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Microsoft.Agents.AI.Valkey;
|
||||
|
||||
/// <summary>
|
||||
/// Options for configuring <see cref="ValkeyChatHistoryProvider"/>.
|
||||
/// </summary>
|
||||
public sealed class ValkeyChatHistoryProviderOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the prefix for Valkey keys. Defaults to "chat_history".
|
||||
/// </summary>
|
||||
public string KeyPrefix { get; set; } = "chat_history";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the maximum number of messages to retain per conversation.
|
||||
/// When exceeded, oldest messages are automatically trimmed. Null means unlimited.
|
||||
/// </summary>
|
||||
public int? MaxMessages { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the maximum number of messages to retrieve from the provider.
|
||||
/// Null means no limit.
|
||||
/// </summary>
|
||||
public int? MaxMessagesToRetrieve { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional key for storing state in the session's StateBag.
|
||||
/// </summary>
|
||||
public string? StateKey { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets optional JSON serializer options for serializing the state of this provider.
|
||||
/// </summary>
|
||||
public JsonSerializerOptions? JsonSerializerOptions { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional filter for messages when retrieving from history.
|
||||
/// </summary>
|
||||
public Func<IEnumerable<ChatMessage>, IEnumerable<ChatMessage>>? ProvideOutputMessageFilter { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional filter for request messages before storing.
|
||||
/// </summary>
|
||||
public Func<IEnumerable<ChatMessage>, IEnumerable<ChatMessage>>? StoreInputRequestMessageFilter { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional filter for response messages before storing.
|
||||
/// </summary>
|
||||
public Func<IEnumerable<ChatMessage>, IEnumerable<ChatMessage>>? StoreInputResponseMessageFilter { get; set; }
|
||||
}
|
||||
Reference in New Issue
Block a user