mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Remove create/delete from AgentThread to avoid needing a thread implementation for each service. (#64)
* Remove create/delete from thread to avoid needing a thread implementation for each service. * Address pr feedback.
This commit is contained in:
westey
committed by
GitHub
Unverified
GitHub
parent
6882e003f5
commit
46a117d581
@@ -39,14 +39,19 @@ public abstract class Agent
|
||||
public virtual string? Instructions { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Create a new <see cref="AgentThread"/> that is compatible with the agent.
|
||||
/// Get a new <see cref="AgentThread"/> instance that is compatible with the agent.
|
||||
/// </summary>
|
||||
/// <returns>A new <see cref="AgentThread"/> instance that is in the created state.</returns>
|
||||
/// <returns>A new <see cref="AgentThread"/> instance.</returns>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// If an agent supports multiple thread types, this method should return the default thread
|
||||
/// type for the agent or whatever the agent was configured to use.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// If the thread needs to be created via a service call it would be created on first use.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public abstract Task<AgentThread> CreateThreadAsync();
|
||||
public abstract AgentThread GetNewThread();
|
||||
|
||||
/// <summary>
|
||||
/// Run the agent with no message assuming that all required instructions are already provided to the agent or on the thread.
|
||||
@@ -189,23 +194,17 @@ public abstract class Agent
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Ensures that the thread exists, is of the expected type, and is active, plus adds the provided message to the thread.
|
||||
/// Checks that the thread is of the expected type, or if null, creates the default thread type.
|
||||
/// </summary>
|
||||
/// <typeparam name="TThreadType">The expected type of the thead.</typeparam>
|
||||
/// <param name="messages">The messages to add to the thread once it is setup.</param>
|
||||
/// <param name="thread">The thread to create if it's null, validate it's type if not null, and start if it is not active.</param>
|
||||
/// <param name="thread">The thread to create if it's null and validate its type if not null.</param>
|
||||
/// <param name="constructThread">A callback to use to construct the thread if it's null.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>An async task that completes once all update are complete.</returns>
|
||||
protected virtual async Task<TThreadType> EnsureThreadExistsWithMessagesAsync<TThreadType>(
|
||||
IReadOnlyCollection<ChatMessage> messages,
|
||||
protected virtual TThreadType ValidateOrCreateThreadType<TThreadType>(
|
||||
AgentThread? thread,
|
||||
Func<TThreadType> constructThread,
|
||||
CancellationToken cancellationToken)
|
||||
Func<TThreadType> constructThread)
|
||||
where TThreadType : AgentThread
|
||||
{
|
||||
Throw.IfNull(messages);
|
||||
|
||||
thread ??= constructThread is not null ? constructThread() : throw new ArgumentNullException(nameof(constructThread));
|
||||
|
||||
if (thread is not TThreadType concreteThreadType)
|
||||
@@ -213,23 +212,11 @@ public abstract class Agent
|
||||
throw new NotSupportedException($"{this.GetType().Name} currently only supports agent threads of type {nameof(TThreadType)}.");
|
||||
}
|
||||
|
||||
// We have to explicitly call create here to ensure that the thread is created
|
||||
// before we run using the thread. While threads will be created when
|
||||
// notified of new messages, some agents support invoking without a message,
|
||||
// and in that case no messages will be sent in the next step.
|
||||
await thread.CreateAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
// Notify the thread that new messages are available.
|
||||
foreach (var message in messages)
|
||||
{
|
||||
await this.NotifyThreadOfNewMessage(thread, message, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return concreteThreadType;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Notfiy the given thread that a new message is available.
|
||||
/// Notfiy the given thread that new messages are available.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -248,12 +235,15 @@ public abstract class Agent
|
||||
/// require the message.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="thread">The thread to notify of the new message.</param>
|
||||
/// <param name="message">The message to pass to the thread.</param>
|
||||
/// <param name="thread">The thread to notify of the new messages.</param>
|
||||
/// <param name="messages">The messages to pass to the thread.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>An async task that completes once the notification is complete.</returns>
|
||||
protected Task NotifyThreadOfNewMessage(AgentThread thread, ChatMessage message, CancellationToken cancellationToken)
|
||||
protected async Task NotifyThreadOfNewMessagesAsync(AgentThread thread, IReadOnlyCollection<ChatMessage> messages, CancellationToken cancellationToken)
|
||||
{
|
||||
return thread.OnNewMessageAsync(message, cancellationToken);
|
||||
if (messages.Count > 0)
|
||||
{
|
||||
await thread.OnNewMessagesAsync(messages, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.AI;
|
||||
@@ -11,113 +12,25 @@ namespace Microsoft.Agents;
|
||||
/// Base abstraction for all agent threads.
|
||||
/// A thread represents a specific conversation with an agent.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This class is used to manage the lifecycle of an agent thread.
|
||||
/// The thread can be not-start, started or ended.
|
||||
/// </remarks>
|
||||
public abstract class AgentThread
|
||||
public class AgentThread
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets the id of the current thread.
|
||||
/// Gets or sets the id of the current thread.
|
||||
/// </summary>
|
||||
public string? Id { get; protected set; }
|
||||
public string? Id { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets a value indicating whether the thread has been deleted.
|
||||
/// </summary>
|
||||
public bool IsDeleted { get; protected set; } = false;
|
||||
|
||||
/// <summary>
|
||||
/// Creates the thread and returns the thread id.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task that completes when the thread has been created.</returns>
|
||||
/// <exception cref="InvalidOperationException">The thread has been deleted.</exception>
|
||||
internal async Task CreateAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (this.IsDeleted)
|
||||
{
|
||||
throw new InvalidOperationException("This thread has been deleted and cannot be recreated.");
|
||||
}
|
||||
|
||||
if (this.Id is not null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
this.Id = await this.CreateCoreAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the current thread.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task that completes when the thread has been deleted.</returns>
|
||||
/// <exception cref="InvalidOperationException">The thread was never created.</exception>
|
||||
public async Task DeleteAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (this.IsDeleted)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.Id is null)
|
||||
{
|
||||
throw new InvalidOperationException("This thread cannot be deleted, since it has not been created.");
|
||||
}
|
||||
|
||||
await this.DeleteCoreAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
this.IsDeleted = true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// This method is called when a new message has been contributed to the chat by any participant.
|
||||
/// This method is called when new messages have been contributed to the chat by any participant.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Inheritors can use this method to update their context based on the new message.
|
||||
/// </remarks>
|
||||
/// <param name="newMessage">The new message.</param>
|
||||
/// <param name="newMessages">The new messages.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task that completes when the context has been updated.</returns>
|
||||
/// <exception cref="InvalidOperationException">The thread has been deleted.</exception>
|
||||
internal async Task OnNewMessageAsync(ChatMessage newMessage, CancellationToken cancellationToken = default)
|
||||
protected internal virtual Task OnNewMessagesAsync(IReadOnlyCollection<ChatMessage> newMessages, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (this.IsDeleted)
|
||||
{
|
||||
throw new InvalidOperationException("This thread has been deleted and cannot be used anymore.");
|
||||
}
|
||||
|
||||
if (this.Id is null)
|
||||
{
|
||||
await this.CreateAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await this.OnNewMessageCoreAsync(newMessage, cancellationToken).ConfigureAwait(false);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates the thread and returns the thread id.
|
||||
/// Checks have already been completed in the <see cref="CreateAsync"/> method to ensure that the thread can be created.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>The id of the thread that was created if one is available.</returns>
|
||||
protected abstract Task<string?> CreateCoreAsync(CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the current thread.
|
||||
/// Checks have already been completed in the <see cref="DeleteAsync"/> method to ensure that the thread can be deleted.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task that completes when the thread has been deleted.</returns>
|
||||
protected abstract Task DeleteCoreAsync(CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// This method is called when a new message has been contributed to the chat by any participant.
|
||||
/// Checks have already been completed in the <see cref="OnNewMessageAsync"/> method to ensure that the thread can be updated.
|
||||
/// </summary>
|
||||
/// <param name="newMessage">The new message.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task that completes when the context has been updated.</returns>
|
||||
protected abstract Task OnNewMessageCoreAsync(ChatMessage newMessage, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
@@ -212,25 +212,35 @@ public class AgentTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnsureThreadExistsWithMessagesVerifiesAndCreatesThreadAndNotifiesThreadAsync()
|
||||
public void ValidateOrCreateThreadTypeVerifiesAndCreatesThread()
|
||||
{
|
||||
var messages = new[] { new ChatMessage(ChatRole.User, "msg1"), new ChatMessage(ChatRole.User, "msg2") };
|
||||
var cancellationToken = new CancellationToken();
|
||||
|
||||
// Custom thread type for type checking
|
||||
var threadMock = new Mock<TestAgentThread>() { CallBase = true };
|
||||
|
||||
var agent = new MockAgent();
|
||||
|
||||
// Should create and notify
|
||||
var result = await agent.EnsureThreadExistsWithMessagesAsync<TestAgentThread>(messages, null, () => threadMock.Object, cancellationToken);
|
||||
// Should create
|
||||
var result = agent.ValidateOrCreateThreadType<TestAgentThread>(null, () => threadMock.Object);
|
||||
Assert.Same(threadMock.Object, result);
|
||||
threadMock.Protected().Verify("OnNewMessageCoreAsync", Times.Once(), messages[0], CancellationToken.None);
|
||||
threadMock.Protected().Verify("OnNewMessageCoreAsync", Times.Once(), messages[1], CancellationToken.None);
|
||||
|
||||
// Should throw if wrong type
|
||||
var wrongThread = new Mock<AgentThread>().Object;
|
||||
await Assert.ThrowsAsync<NotSupportedException>(() => agent.EnsureThreadExistsWithMessagesAsync<TestAgentThread>(messages, wrongThread, () => threadMock.Object, cancellationToken));
|
||||
Assert.Throws<NotSupportedException>(() => agent.ValidateOrCreateThreadType<TestAgentThread>(wrongThread, () => threadMock.Object));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task NotifyThreadOfNewMessagesNotifiesThreadAsync()
|
||||
{
|
||||
var cancellationToken = new CancellationToken();
|
||||
|
||||
var messages = new[] { new ChatMessage(ChatRole.User, "msg1"), new ChatMessage(ChatRole.User, "msg2") };
|
||||
|
||||
var threadMock = new Mock<TestAgentThread>() { CallBase = true };
|
||||
var agent = new MockAgent();
|
||||
|
||||
await agent.NotifyThreadOfNewMessagesAsync(threadMock.Object, messages, cancellationToken);
|
||||
|
||||
threadMock.Protected().Verify("OnNewMessagesAsync", Times.Once(), messages, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -241,25 +251,26 @@ public class AgentTests
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Mock class to test the <see cref="Agent.EnsureThreadExistsWithMessagesAsync{TThreadType}"/> method.
|
||||
/// Mock class to test the <see cref="Agent.ValidateOrCreateThreadType{TThreadType}"/> method.
|
||||
/// </summary>
|
||||
private sealed class MockAgent : Agent
|
||||
{
|
||||
public new Task<TThreadType> EnsureThreadExistsWithMessagesAsync<TThreadType>(
|
||||
IReadOnlyCollection<ChatMessage> messages,
|
||||
public new TThreadType ValidateOrCreateThreadType<TThreadType>(
|
||||
AgentThread? thread,
|
||||
Func<TThreadType> constructThread,
|
||||
CancellationToken cancellationToken)
|
||||
Func<TThreadType> constructThread)
|
||||
where TThreadType : AgentThread
|
||||
{
|
||||
return base.EnsureThreadExistsWithMessagesAsync<TThreadType>(
|
||||
messages,
|
||||
return base.ValidateOrCreateThreadType<TThreadType>(
|
||||
thread,
|
||||
constructThread,
|
||||
cancellationToken);
|
||||
constructThread);
|
||||
}
|
||||
|
||||
public override Task<AgentThread> CreateThreadAsync()
|
||||
public new Task NotifyThreadOfNewMessagesAsync(AgentThread thread, IReadOnlyCollection<ChatMessage> messages, CancellationToken cancellationToken)
|
||||
{
|
||||
return base.NotifyThreadOfNewMessagesAsync(thread, messages, cancellationToken);
|
||||
}
|
||||
|
||||
public override AgentThread GetNewThread()
|
||||
{
|
||||
throw new System.NotImplementedException();
|
||||
}
|
||||
|
||||
@@ -1,171 +0,0 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Microsoft.Agents.Abstractions.UnitTests;
|
||||
|
||||
/// <summary>
|
||||
/// Contains tests for the <see cref="AgentThread"/> class.
|
||||
/// </summary>
|
||||
public class AgentThreadTests
|
||||
{
|
||||
/// <summary>
|
||||
/// Tests that the CreateAsync method sets the Id and invokes CreateInternalAsync once.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task CreateShouldSetIdAndInvokeCreateInternalOnceAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
|
||||
// Act
|
||||
await thread.CreateAsync();
|
||||
await thread.CreateAsync();
|
||||
|
||||
// Assert
|
||||
Assert.Equal("test-thread-id", thread.Id);
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the CreateAsync method throws an InvalidOperationException if the thread is deleted.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task CreateShouldThrowIfThreadDeletedAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
await thread.CreateAsync();
|
||||
await thread.DeleteAsync();
|
||||
|
||||
// Act & Assert
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(() => thread.CreateAsync());
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(1, thread.DeleteInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the DeleteAsync method sets IsDeleted and invokes DeleteInternalAsync.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task DeleteShouldSetIsDeletedAndInvokeDeleteInternalAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
await thread.CreateAsync();
|
||||
|
||||
// Act
|
||||
await thread.DeleteAsync();
|
||||
|
||||
// Assert
|
||||
Assert.True(thread.IsDeleted);
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(1, thread.DeleteInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the DeleteAsync method does not invoke DeleteInternalAsync if the thread is already deleted.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task DeleteShouldNotInvokeDeleteInternalIfAlreadyDeletedAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
await thread.CreateAsync();
|
||||
await thread.DeleteAsync();
|
||||
|
||||
// Act
|
||||
await thread.DeleteAsync();
|
||||
|
||||
// Assert
|
||||
Assert.True(thread.IsDeleted);
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(1, thread.DeleteInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the DeleteAsync method throws an InvalidOperationException if the thread was never created.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task DeleteShouldThrowIfNeverCreatedAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
|
||||
// Act & Assert
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(() => thread.DeleteAsync());
|
||||
Assert.Equal(0, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(0, thread.DeleteInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the OnNewMessageAsync method creates the thread if it is not already created.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task OnNewMessageShouldCreateThreadIfNotCreatedAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
var message = new ChatMessage();
|
||||
|
||||
// Act
|
||||
await thread.OnNewMessageAsync(message);
|
||||
|
||||
// Assert
|
||||
Assert.Equal("test-thread-id", thread.Id);
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(1, thread.OnNewMessageInternalAsyncCount);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests that the OnNewMessageAsync method throws an InvalidOperationException if the thread is deleted.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task OnNewMessageShouldThrowIfThreadDeletedAsync()
|
||||
{
|
||||
// Arrange
|
||||
var thread = new TestAgentThread();
|
||||
await thread.CreateAsync();
|
||||
await thread.DeleteAsync();
|
||||
var message = new ChatMessage();
|
||||
|
||||
// Act & Assert
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(() => thread.OnNewMessageAsync(message));
|
||||
Assert.Equal(1, thread.CreateInternalAsyncCount);
|
||||
Assert.Equal(1, thread.DeleteInternalAsyncCount);
|
||||
Assert.Equal(0, thread.OnNewMessageInternalAsyncCount);
|
||||
}
|
||||
|
||||
private sealed class TestAgentThread : AgentThread
|
||||
{
|
||||
public int CreateInternalAsyncCount { get; private set; }
|
||||
public int DeleteInternalAsyncCount { get; private set; }
|
||||
public int OnNewMessageInternalAsyncCount { get; private set; }
|
||||
|
||||
public new Task CreateAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
return base.CreateAsync(cancellationToken);
|
||||
}
|
||||
|
||||
protected override Task<string?> CreateCoreAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
this.CreateInternalAsyncCount++;
|
||||
return Task.FromResult<string?>("test-thread-id");
|
||||
}
|
||||
|
||||
protected override Task DeleteCoreAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
this.DeleteInternalAsyncCount++;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
protected override Task OnNewMessageCoreAsync(ChatMessage newMessage, CancellationToken cancellationToken = default)
|
||||
{
|
||||
this.OnNewMessageInternalAsyncCount++;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user