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

refine observers

Browse source
This commit is contained in:
Jicheng Lu 2025-07-31 17:48:51 -05:00
parent 6419ca3f8f
commit d523798d41
15 changed files with 223 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/ObserveDataBase.cs
View file

@ -1,6 +1,7 @@
namespace BotSharp.Abstraction.MessageHub.Models;
public abstract class ObserveDataBase
public class ObserveDataBase
{
public IServiceProvider ServiceProvider { get; set; } = null!;
public string RefId { get; set; } = null!;
}

22
src/Infrastructure/BotSharp.Abstraction/MessageHub/Models/ObserverSubscription.cs Normal file
View file

@ -0,0 +1,22 @@
using BotSharp.Abstraction.MessageHub.Observers;
namespace BotSharp.Abstraction.MessageHub.Models;
public class ObserverSubscription<T>
{
public IBotSharpObserver<T> Observer { get; set; }
public IDisposable Subscription { get; set; }
public ObserverSubscription()
{
}
public ObserverSubscription(
IBotSharpObserver<T> observer,
IDisposable subscription)
{
Observer = observer;
Subscription = subscription;
}
}

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

@ -0,0 +1,9 @@
namespace BotSharp.Abstraction.MessageHub.Observers;
public abstract class BotSharpObserverBase<T>
{
protected BotSharpObserverBase()
{
}
}

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

@ -0,0 +1,9 @@
namespace BotSharp.Abstraction.MessageHub.Observers;
public interface IBotSharpObserver<T> : IObserver<T>
{
//string Name { get; }
bool IsActive { get; }
void Activate();
void Deactivate();
}

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

@ -0,0 +1,8 @@
using BotSharp.Abstraction.MessageHub.Models;
namespace BotSharp.Abstraction.MessageHub.Services;
public interface IObserverService
{
ObserverSubscriptionContainer<T> RegisterObservers<T>(string refId) where T : ObserveDataBase;
}

49
src/Infrastructure/BotSharp.Abstraction/MessageHub/Services/ObserverSubscriptionContainer.cs Normal file
View file

@ -0,0 +1,49 @@
using BotSharp.Abstraction.MessageHub.Models;
namespace BotSharp.Abstraction.MessageHub.Services;
public class ObserverSubscriptionContainer<T> : IDisposable
{
private IList<ObserverSubscription<T>> _subscriptions = [];
private bool _disposed = false;
public ObserverSubscriptionContainer()
{
}
public ObserverSubscriptionContainer(
IList<ObserverSubscription<T>> subscriptions)
{
_subscriptions = subscriptions;
}
public void Dispose()
{
Dispose(true);
}
protected virtual void Dispose(bool disposing)
{
if (!_disposed)
{
if (disposing)
{
#if DEBUG
Console.WriteLine($"Start disposing subscriptions...");
#endif
// Unregister all observers
foreach (var item in _subscriptions)
{
item.Observer.Deactivate();
item.Subscription.Dispose();
}
_subscriptions.Clear();
#if DEBUG
Console.WriteLine($"End disposing subscriptions...");
#endif
}
_disposed = true;
}
}
}

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

@ -1,5 +1,8 @@
using BotSharp.Abstraction.Google.Settings;
using BotSharp.Abstraction.Instructs;
using BotSharp.Abstraction.MessageHub;
using BotSharp.Abstraction.MessageHub.Observers;
using BotSharp.Abstraction.MessageHub.Services;
using BotSharp.Abstraction.Messaging;
using BotSharp.Abstraction.Planning;
using BotSharp.Abstraction.Plugins.Models;
@ -8,6 +11,7 @@ using BotSharp.Abstraction.Templating;
using BotSharp.Core.Instructs;
using BotSharp.Core.MessageHub;
using BotSharp.Core.MessageHub.Observers;
using BotSharp.Core.MessageHub.Services;
using BotSharp.Core.Messaging;
using BotSharp.Core.Routing.Reasoning;
using BotSharp.Core.Templating;
@ -45,6 +49,8 @@ public class ConversationPlugin : IBotSharpPlugin, IBotSharpAppPlugin
});
services.AddSingleton<MessageHub<HubObserveData<RoleDialogModel>>>();
//services.AddScoped<IBotSharpObserver<HubObserveData<RoleDialogModel>>, ConversationObserver>();
services.AddScoped<IObserverService, ObserverService>();
services.AddScoped<IConversationStorage, ConversationStorage>();
services.AddScoped<IConversationService, ConversationService>();
@ -71,9 +77,8 @@ public class ConversationPlugin : IBotSharpPlugin, IBotSharpAppPlugin
public void Configure(IApplicationBuilder app)
{
var services = app.ApplicationServices;
var queue = services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData<RoleDialogModel>>>>();
queue.Events.Subscribe(new ConversationObserver(logger));
//var services = app.ApplicationServices;
//var queue = services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
//var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData<RoleDialogModel>>>>();
}
}

14
src/Infrastructure/BotSharp.Core/Demo/Functions/GetWeatherFn.cs
View file

@ -1,5 +1,6 @@
using BotSharp.Abstraction.Functions;
using BotSharp.Core.MessageHub;
using System.Text.Json.Serialization;
namespace BotSharp.Core.Demo.Functions;
@ -17,26 +18,29 @@ public class GetWeatherFn : IFunctionCallback
public async Task<bool> Execute(RoleDialogModel message)
{
var args = JsonSerializer.Deserialize<WeatherLocation>(message.FunctionArgs);
var conv = _services.GetRequiredService<IConversationService>();
var messageHub = _services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
await Task.Delay(1000);
message.Indication = "Start querying weather data";
message.Indication = $"Start querying weather data in {args?.City}";
messageHub.Push(new()
{
EventName = ChatEvent.OnIndicationReceived,
Data = message,
RefId = conv.ConversationId,
ServiceProvider = _services
});
await Task.Delay(1500);
message.Indication = "Still working on it";
message.Indication = $"Still working on it, {args?.City}";
messageHub.Push(new()
{
EventName = ChatEvent.OnIndicationReceived,
Data = message,
RefId = conv.ConversationId,
ServiceProvider = _services
});
@ -46,4 +50,10 @@ public class GetWeatherFn : IFunctionCallback
message.StopCompletion = false;
return true;
}
}
class WeatherLocation
{
[JsonPropertyName("city")]
public string City { get; set; }
}

2
src/Infrastructure/BotSharp.Core/MessageHub/MessageHub.cs
View file

@ -2,7 +2,7 @@ using System.Reactive.Subjects;
namespace BotSharp.Core.MessageHub;
public class MessageHub<T> where T : class, new()
public class MessageHub<T> where T : ObserveDataBase
{
private readonly ILogger<MessageHub<T>> _logger;
private readonly ISubject<T> _observable = Subject.Synchronize(new Subject<T>());

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

@ -1,15 +1,31 @@
using BotSharp.Abstraction.MessageHub.Observers;
namespace BotSharp.Core.MessageHub.Observers;
public class ConversationObserver : IObserver<HubObserveData<RoleDialogModel>>
public class ConversationObserver : IBotSharpObserver<HubObserveData<RoleDialogModel>>
{
private readonly ILogger _logger;
private readonly ILogger<ConversationObserver> _logger;
private IServiceProvider _services;
private bool _isActive;
public ConversationObserver(ILogger logger)
public ConversationObserver(
ILogger<ConversationObserver> logger)
{
_logger = logger;
}
public bool IsActive => _isActive;
public void Activate()
{
_isActive = true;
}
public void Deactivate()
{
_isActive = false;
}
public void OnCompleted()
{
_logger.LogWarning($"{nameof(ConversationObserver)} receives complete notification.");
@ -23,15 +39,14 @@ public class ConversationObserver : IObserver<HubObserveData<RoleDialogModel>>
public void OnNext(HubObserveData<RoleDialogModel> value)
{
_services = value.ServiceProvider;
var progress = _services.GetRequiredService<IConversationProgressService>();
//var progress = _services.GetRequiredService<IConversationProgressService>();
if (value.EventName == ChatEvent.OnIndicationReceived
&& progress.OnFunctionExecuting != null)
if (value.EventName == ChatEvent.OnIndicationReceived)
{
#if DEBUG
_logger.LogCritical($"Receiving {value.EventName} in {nameof(ConversationObserver)}");
#if !DEBUG
_logger.LogCritical($"Receiving {value.EventName} ({value.Data.Indication}) in {nameof(ConversationObserver)}");
#endif
progress.OnFunctionExecuting(value.Data).ConfigureAwait(false).GetAwaiter().GetResult();
//progress.OnFunctionExecuting(value.Data).ConfigureAwait(false).GetAwaiter().GetResult();
}
}
}

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

@ -0,0 +1,43 @@
using BotSharp.Abstraction.MessageHub.Observers;
using BotSharp.Abstraction.MessageHub.Services;
using System.Reactive.Linq;
namespace BotSharp.Core.MessageHub.Services;
public class ObserverService : IObserverService
{
private readonly IServiceProvider _services;
public ObserverService(
IServiceProvider services)
{
_services = services;
}
public ObserverSubscriptionContainer<T> RegisterObservers<T>(string refId) where T : ObserveDataBase
{
var subscriptions = new List<ObserverSubscription<T>>();
var observers = _services.GetServices<IBotSharpObserver<T>>()
.Where(x => !x.IsActive)
.ToList();
if (observers.IsNullOrEmpty())
{
return new();
}
var messageHub = _services.GetRequiredService<MessageHub<T>>();
foreach (var observer in observers)
{
observer.Activate();
var sub = messageHub.Events.Where(x => x.RefId == refId).Subscribe(observer);
subscriptions.Add(new()
{
Observer = observer,
Subscription = sub
});
}
return new(subscriptions);
}
}

2
src/Infrastructure/BotSharp.Core/Routing/RoutingService.InvokeFunction.cs
View file

@ -25,11 +25,13 @@ public partial class RoutingService
clonedMessage.FunctionName = name;
clonedMessage.Indication = await funcExecutor.GetIndicatorAsync(message);
var conv = _services.GetRequiredService<IConversationService>();
var messageHub = _services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
messageHub.Push(new()
{
EventName = ChatEvent.OnIndicationReceived,
Data = clonedMessage,
RefId = conv.ConversationId,
ServiceProvider = _services
});

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

@ -1,10 +1,13 @@
using BotSharp.Abstraction.Files.Constants;
using BotSharp.Abstraction.Files.Enums;
using BotSharp.Abstraction.Files.Utilities;
using BotSharp.Abstraction.MessageHub.Models;
using BotSharp.Abstraction.MessageHub.Services;
using BotSharp.Abstraction.Options;
using BotSharp.Abstraction.Routing;
using BotSharp.Abstraction.Users.Dtos;
using BotSharp.Core.Infrastructures;
using System.ComponentModel;
namespace BotSharp.OpenAPI.Controllers;
@ -343,6 +346,9 @@ public class ConversationController : ControllerBase
[FromRoute] string conversationId,
[FromBody] NewMessageModel input)
{
var observerService = _services.GetRequiredService<IObserverService>();
using var container = observerService.RegisterObservers<HubObserveData<RoleDialogModel>>(conversationId);
var conv = _services.GetRequiredService<IConversationService>();
var inputMsg = new RoleDialogModel(AgentRole.User, input.Text)
{

11
src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs
View file

@ -1,6 +1,8 @@
using BotSharp.Abstraction.Crontab;
using BotSharp.Abstraction.MessageHub.Models;
using BotSharp.Abstraction.MessageHub.Observers;
using BotSharp.Core.MessageHub;
using BotSharp.Core.MessageHub.Observers;
using BotSharp.Plugin.ChatHub.Hooks;
using BotSharp.Plugin.ChatHub.Observers;
using Microsoft.AspNetCore.Builder;
@ -24,6 +26,8 @@ public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin
config.Bind("ChatHub", settings);
services.AddSingleton(x => settings);
services.AddScoped<IBotSharpObserver<HubObserveData<RoleDialogModel>>, ChatHubObserver>();
// Register hooks
services.AddScoped<IConversationHook, ChatHubConversationHook>();
services.AddScoped<IConversationHook, StreamingLogHook>();
@ -35,9 +39,8 @@ public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin
public void Configure(IApplicationBuilder app)
{
var services = app.ApplicationServices;
var queue = services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData<RoleDialogModel>>>>();
queue.Events.Subscribe(new ChatHubObserver(logger));
//var services = app.ApplicationServices;
//var queue = services.GetRequiredService<MessageHub<HubObserveData<RoleDialogModel>>>();
//var logger = services.GetRequiredService<ILogger<MessageHub<HubObserveData<RoleDialogModel>>>>();
}
}

22
src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs
View file

@ -1,21 +1,37 @@
using BotSharp.Abstraction.Conversations.Dtos;
using BotSharp.Abstraction.Conversations.Enums;
using BotSharp.Abstraction.MessageHub.Models;
using BotSharp.Abstraction.MessageHub.Observers;
using BotSharp.Abstraction.SideCar;
using BotSharp.Core.MessageHub.Observers;
using System.Runtime.CompilerServices;
namespace BotSharp.Plugin.ChatHub.Observers;
public class ChatHubObserver : IObserver<HubObserveData<RoleDialogModel>>
public class ChatHubObserver : IBotSharpObserver<HubObserveData<RoleDialogModel>>
{
private readonly ILogger _logger;
private IServiceProvider _services;
private bool _isActive = false;
public ChatHubObserver(ILogger logger)
public ChatHubObserver(
ILogger<ChatHubObserver> logger)
{
_logger = logger;
}
public bool IsActive => _isActive;
public void Activate()
{
_isActive = true;
}
public void Deactivate()
{
_isActive = false;
}
public void OnCompleted()
{
_logger.LogWarning($"{nameof(ChatHubObserver)} receives complete notification.");
@ -117,6 +133,8 @@ public class ChatHubObserver : IObserver<HubObserveData<RoleDialogModel>>
Role = AgentRole.Assistant
}
};
_logger.LogCritical($"Receiving {value.EventName} ({value.Data.Indication}) in {nameof(ChatHubObserver)} - {conv.ConversationId}");
break;
}