mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
* .NET: Persist input messages on streaming errors in PerServiceCallChatHistoryPersistingChatClient When the underlying chat service emits an in-stream error (for example a `response.error` SSE event from the OpenAI Responses API on rate limit), the OpenAI client surfaces it as an `ErrorContent` update and ends the stream without throwing. Previously, `PerServiceCallChatHistoryPersistingChatClient` only persisted history when the streaming loop completed successfully and `NotifyProvidersOfNewMessagesAsync` was called at the end. On the in-stream-error path, the input messages handed to that iteration - typically `FunctionResultContent` produced by `FunctionInvokingChatClient` in the previous iteration - were never persisted. The next run would replay session history with a dangling `FunctionCallContent` and the service would reject the request with `No tool output found for function call <id>`. This change: - Adds a `PersistInputOnErrorAsync` helper that persists the input messages (with no response messages) so function-call/function-result pairings are not split across failures. - Calls the helper from every error path: pre-loop enumerator creation, the first `MoveNextAsync`, the in-loop `MoveNextAsync`, and a new `finally` that handles abnormal iterator disposal. - After the streaming loop, scans the assembled response for any `ErrorContent` and, if present, persists the input, notifies providers of failure, and throws `InvalidOperationException` so the error is surfaced to the caller instead of silently corrupting history. - Hardens `InMemoryChatHistoryProvider.StoreChatHistoryAsync` to treat a null `RequestMessages` as empty, since the new error path can invoke it with no response messages. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix dropped FunctionResultContent on streaming pipeline early-disposal When a consumer of ChatClientAgent.RunStreamingAsync stops iterating early (e.g. ToolApprovalAgent yields the approval request and then `yield break`), the framework cascades DisposeAsync down the stream. C# async iterators do not auto-dispose IAsyncDisposable locals, so the inner enumerator returned by IChatClient.GetStreamingResponseAsync(...).GetAsyncEnumerator(ct) was left suspended. That suspended FunctionInvokingChatClient downstream, which suspended PerServiceCallChatHistoryPersistingChatClient at its `yield return`, so its finally block never ran and the in-flight FunctionResultContent for the just-completed tool call was not persisted to chat history. The next turn then loaded a session that contained a FunctionCallContent with no matching FunctionResultContent and the model returned HTTP 400 `No tool output found for function call`. Fixes: * ChatClientAgent.RunStreamingAsync: wrap the iteration in try/finally that disposes the inner enumerator. Disposal now cascades through the pipeline and PerService's finally runs on early exit. * PerServiceCallChatHistoryPersistingChatClient: in the streaming path, snapshot input messages with `messages.ToList()` (the caller, FICC, reuses a single mutable buffer across iterations and may mutate it before our finally / error path persists), wrap GetAsyncEnumerator, the first MoveNextAsync, and in-loop MoveNextAsync in try/catch each calling PersistInputOnErrorAsync + NotifyProvidersOfFailureAsync, and add a finally that calls PersistInputOnErrorAsync when the loop did not exit normally so per-iteration FRCs are persisted on early disposal as well as on errors. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Add tests for PerService streaming error/dispose persistence paths Adds five regression tests covering the new error-path persistence in PerServiceCallChatHistoryPersistingChatClient.GetStreamingResponseInnerAsync: - Persists input messages when GetStreamingResponseAsync throws synchronously. - Persists input messages when the first MoveNextAsync throws. - Persists input messages when a mid-stream MoveNextAsync throws. - Persists input messages when the consumer abandons enumeration early (the ToolApprovalAgent yield-break / disposal-cascade case). - Throws and persists input when the stream emits an in-band ErrorContent. All 66 tests in the class pass on net10.0 and net472. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Address PR feedback on PerService streaming error persistence Two follow-ups from PR #5744 review: 1. Prevent duplicate persistence on the in-loop MoveNextAsync catch path. The inner catch persists input messages, then rethrows, which propagates through the surrounding try/finally where loopExitedNormally is still false, causing the finally to persist again. Introduced an inputPersisted flag that the inner catch sets after persisting; the finally now skips when inputPersisted is true. 2. Use the caller's CancellationToken in the abnormal-exit finally instead of CancellationToken.None, so cleanup remains responsive to cancellation. Fall back to CancellationToken.None only when the caller's token is already canceled (otherwise the persist call would observe the cancellation, throw, and mask the original early-exit reason). Tightened all five new streaming-error tests from Times.AtLeastOnce to Times.Once on the input-persistence matcher to regression-guard against duplicate persistence. All 66 tests in the class still pass (net10.0 + net472). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Scope PerService streaming changes to cooperative early-exit only Per discussion on PR #5744, scope this PR back to fix only the original ToolApprovalAgent dropped-FunctionResultContent bug and address the enumerator-disposal review comment. Specifically: - Remove input-message persistence from the GetAsyncEnumerator and MoveNextAsync error paths. Routing failed service calls through the success notification channel was breaking the provider contract; we will instead rely on inner-agent retries for transient errors. Failure paths still call NotifyProvidersOfFailureAsync as before. - Remove the in-stream ErrorContent detection block (same rationale). - Keep the try/finally that calls the (now narrower) early-exit input notification on cooperative disposal (e.g. ToolApprovalAgent yield break). A new serviceErrorOccurred flag ensures we do NOT renotify on exception paths. - Always DisposeAsync the underlying enumerator on every exit path, addressing the copilot-reviewer comment about leaked HTTP/streams. - Rename PersistInputOnErrorAsync -> NotifyProvidersOfEarlyExitInputAsync to better reflect what it does and when it runs (rogerbarreto nit). - Apply rogerbarreto nit on InMemoryChatHistoryProvider null-coalescing. - Drop the four tests that covered the removed error-path behavior; keep RunStreamingAsync_PersistsInputMessages_WhenConsumerAbandons EnumerationAsync (regression guard for the cooperative-pause path). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: alliscode <bentho@microsoft.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
135 lines
5.8 KiB
C#
135 lines
5.8 KiB
C#
// Copyright (c) Microsoft. All rights reserved.
|
|
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text.Json.Serialization;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.AI;
|
|
using Microsoft.Shared.Diagnostics;
|
|
|
|
namespace Microsoft.Agents.AI;
|
|
|
|
/// <summary>
|
|
/// Provides an in-memory implementation of <see cref="ChatHistoryProvider"/> with support for message reduction.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// <see cref="InMemoryChatHistoryProvider"/> stores chat messages in the <see cref="AgentSession.StateBag"/>,
|
|
/// providing fast access and manipulation capabilities integrated with session state management.
|
|
/// </para>
|
|
/// <para>
|
|
/// This <see cref="ChatHistoryProvider"/> maintains all messages in memory. For long-running conversations or high-volume scenarios, consider using
|
|
/// message reduction strategies or alternative storage implementations.
|
|
/// </para>
|
|
/// </remarks>
|
|
public sealed class InMemoryChatHistoryProvider : ChatHistoryProvider
|
|
{
|
|
private readonly ProviderSessionState<State> _sessionState;
|
|
private IReadOnlyList<string>? _stateKeys;
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="InMemoryChatHistoryProvider"/> class.
|
|
/// </summary>
|
|
/// <param name="options">
|
|
/// Optional configuration options that control the provider's behavior, including state initialization,
|
|
/// message reduction, and serialization settings. If <see langword="null"/>, default settings will be used.
|
|
/// </param>
|
|
public InMemoryChatHistoryProvider(InMemoryChatHistoryProviderOptions? options = null)
|
|
: base(
|
|
options?.ProvideOutputMessageFilter,
|
|
options?.StorageInputRequestMessageFilter,
|
|
options?.StorageInputResponseMessageFilter)
|
|
{
|
|
this._sessionState = new ProviderSessionState<State>(
|
|
options?.StateInitializer ?? (_ => new State()),
|
|
options?.StateKey ?? this.GetType().Name,
|
|
options?.JsonSerializerOptions);
|
|
this.ChatReducer = options?.ChatReducer;
|
|
this.ReducerTriggerEvent = options?.ReducerTriggerEvent ?? InMemoryChatHistoryProviderOptions.ChatReducerTriggerEvent.BeforeMessagesRetrieval;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override IReadOnlyList<string> StateKeys => this._stateKeys ??= [this._sessionState.StateKey];
|
|
|
|
/// <summary>
|
|
/// Gets the chat reducer used to process or reduce chat messages. If null, no reduction logic will be applied.
|
|
/// </summary>
|
|
public IChatReducer? ChatReducer { get; }
|
|
|
|
/// <summary>
|
|
/// Gets the event that triggers the reducer invocation in this provider.
|
|
/// </summary>
|
|
public InMemoryChatHistoryProviderOptions.ChatReducerTriggerEvent ReducerTriggerEvent { get; }
|
|
|
|
/// <summary>
|
|
/// Gets the chat messages stored for the specified session.
|
|
/// </summary>
|
|
/// <param name="session">The agent session containing the state.</param>
|
|
/// <returns>A list of chat messages, or an empty list if no state is found.</returns>
|
|
public List<ChatMessage> GetMessages(AgentSession? session)
|
|
=> this._sessionState.GetOrInitializeState(session).Messages;
|
|
|
|
/// <summary>
|
|
/// Sets the chat messages for the specified session.
|
|
/// </summary>
|
|
/// <param name="session">The agent session containing the state.</param>
|
|
/// <param name="messages">The messages to store.</param>
|
|
/// <exception cref="ArgumentNullException"><paramref name="messages"/> is <see langword="null"/>.</exception>
|
|
public void SetMessages(AgentSession? session, List<ChatMessage> messages)
|
|
{
|
|
Throw.IfNull(messages);
|
|
|
|
State state = this._sessionState.GetOrInitializeState(session);
|
|
state.Messages = messages;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override async ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default)
|
|
{
|
|
State state = this._sessionState.GetOrInitializeState(context.Session);
|
|
|
|
if (this.ReducerTriggerEvent is InMemoryChatHistoryProviderOptions.ChatReducerTriggerEvent.BeforeMessagesRetrieval && this.ChatReducer is not null)
|
|
{
|
|
// Apply pre-retrieval reduction if configured
|
|
await ReduceMessagesAsync(this.ChatReducer, state, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
|
|
return state.Messages;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override async ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default)
|
|
{
|
|
State state = this._sessionState.GetOrInitializeState(context.Session);
|
|
|
|
// Add request and response messages to the provider
|
|
var allNewMessages = (context.RequestMessages ?? []).Concat(context.ResponseMessages ?? []);
|
|
state.Messages.AddRange(allNewMessages);
|
|
|
|
if (this.ReducerTriggerEvent is InMemoryChatHistoryProviderOptions.ChatReducerTriggerEvent.AfterMessageAdded && this.ChatReducer is not null)
|
|
{
|
|
// Apply pre-write reduction strategy if configured
|
|
await ReduceMessagesAsync(this.ChatReducer, state, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private static async Task ReduceMessagesAsync(IChatReducer reducer, State state, CancellationToken cancellationToken = default)
|
|
{
|
|
state.Messages = [.. await reducer.ReduceAsync(state.Messages, cancellationToken).ConfigureAwait(false)];
|
|
}
|
|
|
|
/// <summary>
|
|
/// Represents the state of a <see cref="InMemoryChatHistoryProvider"/> stored in the <see cref="AgentSession.StateBag"/>.
|
|
/// </summary>
|
|
public sealed class State
|
|
{
|
|
/// <summary>
|
|
/// Gets or sets the list of chat messages.
|
|
/// </summary>
|
|
[JsonPropertyName("messages")]
|
|
public List<ChatMessage> Messages { get; set; } = [];
|
|
}
|
|
}
|