SemianiakaVY/BotSharp
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

replace conversation progress service

Browse source
This commit is contained in:
Jicheng Lu 2025-08-01 15:47:42 -05:00
parent 20de3f7565
commit ae0abecd3b
11 changed files with 58 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationProgressService.cs
View file

@ -1,10 +0,0 @@
namespace BotSharp.Abstraction.Conversations;
public delegate Task FunctionExecuting(RoleDialogModel msg);
public delegate Task FunctionExecuted(RoleDialogModel msg);
public interface IConversationProgressService
{
FunctionExecuted OnFunctionExecuted { get; set; }
FunctionExecuting OnFunctionExecuting { get; set; }
}

9
src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/BotSharpObserverBase.cs
View file

@ -3,7 +3,8 @@ namespace BotSharp.Abstraction.MessageHub.Observers;
public abstract class BotSharpObserverBase<T> : IBotSharpObserver<T>
{
protected bool _active = false;
private bool _active = false;
protected Dictionary<string, Func<T, Task>> _listeners = [];
protected BotSharpObserverBase()
{
@ -22,6 +23,12 @@ public abstract class BotSharpObserverBase<T> : IBotSharpObserver<T>
public virtual void Deactivate()
{
_active = false;
_listeners = [];
}
public virtual void SetEventListeners(Dictionary<string, Func<T, Task>> listeners)
{
_listeners = listeners;
}
public virtual void OnCompleted()

2
src/Infrastructure/BotSharp.Abstraction/MessageHub/Observers/IBotSharpObserver.cs
View file

@ -4,6 +4,8 @@ public interface IBotSharpObserver<T> : IObserver<T>
{
string Name { get; }
bool Active { get; }
void SetEventListeners(Dictionary<string, Func<T, Task>> listeners);
void Activate();
void Deactivate();
}

5
src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/IObserverService.cs
View file

@ -4,7 +4,10 @@ namespace BotSharp.Abstraction.MessageHub.Services;
public interface IObserverService
{
IDisposable SubscribeObservers<T>(string refId, IEnumerable<string>? names = null) where T : ObserveDataBase;
IDisposable SubscribeObservers<T>(
string refId,
IEnumerable<string>? names = null,
Dictionary<string, Func<T, Task>>? listeners = null) where T : ObserveDataBase;
void UnSubscribeObservers<T>(IEnumerable<string>? names = null) where T : ObserveDataBase;
}

1
src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs
View file

@ -56,7 +56,6 @@ public class ConversationPlugin : IBotSharpPlugin
services.AddScoped<IConversationStorage, ConversationStorage>();
services.AddScoped<IConversationService, ConversationService>();
services.AddScoped<IConversationProgressService, ConversationProgressService>();
services.AddScoped<IConversationStateService, ConversationStateService>();
services.AddScoped<ITranslationService, TranslationService>();

11
src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationProgressService.cs
View file

@ -1,11 +0,0 @@
namespace BotSharp.Core.Conversations.Services
{
public class ConversationProgressService : IConversationProgressService
{
public FunctionExecuting OnFunctionExecuting { get; set; }
public FunctionExecuted OnFunctionExecuted { get; set; }
}
}

2
src/Infrastructure/BotSharp.Core/Demo/Functions/GetFunEventsFn.cs
View file

@ -34,7 +34,7 @@ public class GetFunEventsFn : IFunctionCallback
await Task.Delay(1500);
message.Indication = $"Still searching in {args?.City}";
message.Indication = $"Still searching events in {args?.City}";
messageHub.Push(new()
{
EventName = ChatEvent.OnIndicationReceived,

5
src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs
View file

@ -36,7 +36,10 @@ public class ConversationObserver : BotSharpObserverBase<HubObserveData<RoleDial
#if DEBUG
_logger.LogCritical($"Receiving {value.EventName} ({value.Data.Indication}) in {nameof(ConversationObserver)} - {conv.ConversationId}");
#endif
//progress.OnFunctionExecuting(value.Data).ConfigureAwait(false).GetAwaiter().GetResult();
if (_listeners.TryGetValue(value.EventName, out var func) && func != null)
{
func(value).ConfigureAwait(false).GetAwaiter().GetResult();
}
}
}
}

13
src/Infrastructure/BotSharp.Core/MessageHub/Services/ObserverService.cs
View file

@ -17,7 +17,10 @@ public class ObserverService : IObserverService
_logger = logger;
}
public IDisposable SubscribeObservers<T>(string refId, IEnumerable<string>? names = null) where T : ObserveDataBase
public IDisposable SubscribeObservers<T>(
string refId,
IEnumerable<string>? names = null,
Dictionary<string, Func<T, Task>>? listeners = null) where T : ObserveDataBase
{
var container = _services.GetRequiredService<ObserverSubscriptionContainer<T>>();
var observers = _services.GetServices<IBotSharpObserver<T>>()
@ -34,6 +37,14 @@ public class ObserverService : IObserverService
return container;
}
if (!listeners.IsNullOrEmpty())
{
foreach (var observer in observers)
{
observer.SetEventListeners(listeners ?? []);
}
}
#if DEBUG
_logger.LogCritical($"Subscribe observers: {string.Join(",", observers.Select(x => x.Name))}");
#endif

31
src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs
View file

@ -7,7 +7,6 @@ using BotSharp.Abstraction.Options;
using BotSharp.Abstraction.Routing;
using BotSharp.Abstraction.Users.Dtos;
using BotSharp.Core.Infrastructures;
using BotSharp.Core.MessageHub.Observers;
namespace BotSharp.OpenAPI.Controllers;
@ -386,6 +385,12 @@ public class ConversationController : ControllerBase
[HttpPost("/conversation/{agentId}/{conversationId}/sse")]
public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input)
{
var observer = _services.GetRequiredService<IObserverService>();
using var container = observer.SubscribeObservers<HubObserveData<RoleDialogModel>>(conversationId, listeners: new()
{
{ ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) }
});
var conv = _services.GetRequiredService<IConversationService>();
var inputMsg = new RoleDialogModel(AgentRole.User, input.Text)
{
@ -411,7 +416,6 @@ public class ConversationController : ControllerBase
Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.ContentType, "text/event-stream");
Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.CacheControl, "no-cache");
Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive");
InitProgressService(conversationId);
await conv.SendMessage(agentId, inputMsg,
replyMessage: input.Postback,
@ -435,23 +439,18 @@ public class ConversationController : ControllerBase
// await OnEventCompleted(Response);
}
private void InitProgressService(string conversationId)
private async Task OnReceiveToolCallIndication(string conversationId, RoleDialogModel msg)
{
var progressService = _services.GetService<IConversationProgressService>();
progressService.OnFunctionExecuting = async msg =>
var indicator = new ChatResponseModel
{
var indicator = new ChatResponseModel
{
ConversationId = conversationId,
MessageId = msg.MessageId,
Text = msg.Indication,
Function = "indicating",
Instruction = msg.Instruction,
States = new Dictionary<string, string>()
};
await OnChunkReceived(Response, indicator);
ConversationId = conversationId,
MessageId = msg.MessageId,
Text = msg.Indication,
Function = "indicating",
Instruction = msg.Instruction,
States = new Dictionary<string, string>()
};
progressService.OnFunctionExecuted = async msg => { };
await OnChunkReceived(Response, indicator);
}
#endregion

22
src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs
View file

@ -1,4 +1,6 @@
using BotSharp.Abstraction.Files;
using BotSharp.Abstraction.MessageHub.Models;
using BotSharp.Abstraction.MessageHub.Services;
using BotSharp.Abstraction.Realtime;
using BotSharp.Abstraction.Routing;
using BotSharp.Core.Infrastructures;
@ -82,8 +84,12 @@ public class TwilioMessageQueueService : BackgroundService
var routing = sp.GetRequiredService<IRoutingService>();
var config = sp.GetRequiredService<TwilioSetting>();
var sessionManager = sp.GetRequiredService<ITwilioSessionManager>();
var progressService = sp.GetRequiredService<IConversationProgressService>();
InitProgressService(message, sessionManager, progressService);
var observer = sp.GetRequiredService<IObserverService>();
using var container = observer.SubscribeObservers<HubObserveData<RoleDialogModel>>(message.ConversationId, listeners: new()
{
{ ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(data.Data, message, sessionManager) }
});
InitConversation(message, inputMsg, conv, routing);
// Need to consider Inbound and Outbound call
@ -185,15 +191,11 @@ public class TwilioMessageQueueService : BackgroundService
return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse());
}
private static void InitProgressService(CallerMessage message, ITwilioSessionManager sessionManager, IConversationProgressService progressService)
private static async Task OnReceiveToolCallIndication(RoleDialogModel msg, CallerMessage message, ITwilioSessionManager sessionManager)
{
progressService.OnFunctionExecuting = async msg =>
if (!string.IsNullOrEmpty(msg.Indication))
{
if (!string.IsNullOrEmpty(msg.Indication))
{
await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication);
}
};
progressService.OnFunctionExecuted = async msg => { };
await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication);
}
}
}