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

RealtimeHub & Tool use

Browse source
This commit is contained in:
Haiping Chen 2025-02-09 17:31:46 -06:00
parent 738d87de53
commit d9480fb0d3
15 changed files with 486 additions and 240 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
src/Infrastructure/BotSharp.Abstraction/MLTasks/IRealTimeCompletion.cs
View file

@ -8,5 +8,22 @@ public interface IRealTimeCompletion
string Model { get; }
void SetModelName(string model);
Task Connect(RealtimeHubConnection conn,
Action onModelReady,
Action<string> onModelAudioDeltaReceived,
Action onModelAudioResponseDone,
Action<string> onAudioTranscriptDone,
Action<string> onModelResponseDone,
Action onUserInterrupted);
Task AppenAudioBuffer(string message);
Task SendEventToModel(object message);
Task Disconnect();
Task<RealtimeSession> CreateSession(Agent agent, List<RoleDialogModel> conversations);
Task<string> UpdateInitialSession(RealtimeHubConnection conn);
Task<string> InertConversationItem(RoleDialogModel message);
Task<List<RoleDialogModel>> OnResponsedDone(RealtimeHubConnection conn, string response);
}

13
src/Infrastructure/BotSharp.Abstraction/Realtime/IRealtimeModelConnector.cs
View file

@ -1,13 +0,0 @@
using BotSharp.Abstraction.Realtime.Models;
namespace BotSharp.Abstraction.Realtime;
public interface IRealtimeModelConnector
{
Task Connect(RealtimeHubConnection conn,
Action<string> onAudioDeltaReceived,
Action onAudioResponseDone,
Action onUserInterrupted);
Task SendMessage(string message);
Task Disconnect();
}

1
src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs
View file

@ -6,6 +6,7 @@ public class RealtimeHubConnection
public string StreamId { get; set; } = null!;
public string ConversationId { get; set; } = null!;
public string Data { get; set; } = string.Empty;
public string Model { get; set; } = null!;
public Func<string, object> OnModelMessageReceived { get; set; } = null!;
public Func<object> OnModelAudioResponseDone { get; set; } = null!;
public Func<object> OnModelUserInterrupted { get; set; } = null!;

109
src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs
View file

@ -2,6 +2,8 @@ using BotSharp.Abstraction.Realtime;
using System.Net.WebSockets;
using System;
using BotSharp.Abstraction.Realtime.Models;
using BotSharp.Abstraction.MLTasks;
using BotSharp.Abstraction.Agents.Models;
namespace BotSharp.Core.Realtime;
@ -20,7 +22,13 @@ public class RealtimeHub : IRealtimeHub
{
var buffer = new byte[1024 * 4];
WebSocketReceiveResult result;
var modelConnector = _services.GetRequiredService<IRealtimeModelConnector>();
var llmProviderService = _services.GetRequiredService<ILlmProviderService>();
var model = llmProviderService.GetProviderModel("openai", "gpt-4",
realTime: true).Name;
var completer = _services.GetServices<IRealTimeCompletion>().First(x => x.Provider == "openai");
completer.SetModelName(model);
do
{
@ -33,46 +41,97 @@ public class RealtimeHub : IRealtimeHub
}
var conn = onUserMessageReceived(receivedText);
if (conn.Event == "connected")
conn.Model = model;
if (conn.Event == "user_connected")
{
await ConnectToModel(modelConnector, userWebSocket, conn);
await ConnectToModel(completer, userWebSocket, conn);
}
else if (conn.Event == "data_received")
else if (conn.Event == "user_data_received")
{
await modelConnector.SendMessage(conn.Data);
await completer.AppenAudioBuffer(conn.Data);
}
else if (conn.Event == "disconnected")
else if (conn.Event == "user_disconnected")
{
await modelConnector.Disconnect();
await completer.Disconnect();
}
} while (!result.CloseStatus.HasValue);
await userWebSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None);
}
private async Task ConnectToModel(IRealtimeModelConnector modelConnector, WebSocket userWebSocket, RealtimeHubConnection conn)
private async Task ConnectToModel(IRealTimeCompletion completer, WebSocket userWebSocket, RealtimeHubConnection conn)
{
await modelConnector.Connect(conn, onAudioDeltaReceived: async audioDeltaData =>
{
var data = conn.OnModelMessageReceived(audioDeltaData);
await SendEventToWebSocket(userWebSocket, data);
},
onAudioResponseDone: async () =>
{
var data = conn.OnModelAudioResponseDone();
await SendEventToWebSocket(userWebSocket, data);
},
onUserInterrupted: async () =>
{
var data = conn.OnModelUserInterrupted();
await SendEventToWebSocket(userWebSocket, data);
});
var hookProvider = _services.GetRequiredService<ConversationHookProvider>();
var storage = _services.GetRequiredService<IConversationStorage>();
var convService = _services.GetRequiredService<IConversationService>();
convService.SetConversationId(conn.ConversationId, []);
var conversation = await convService.GetConversation(conn.ConversationId);
var agentService = _services.GetRequiredService<IAgentService>();
var agent = await agentService.LoadAgent(conversation.AgentId);
var routing = _services.GetRequiredService<IRoutingService>();
var dialogs = convService.GetDialogHistory();
routing.Context.SetDialogs(dialogs);
await completer.Connect(conn,
onModelReady: async () =>
{
// Control initial session
var data = await completer.UpdateInitialSession(conn);
await completer.SendEventToModel(data);
},
onModelAudioDeltaReceived: async audioDeltaData =>
{
var data = conn.OnModelMessageReceived(audioDeltaData);
await SendEventToUser(userWebSocket, data);
},
onModelAudioResponseDone: async () =>
{
var data = conn.OnModelAudioResponseDone();
await SendEventToUser(userWebSocket, data);
},
onAudioTranscriptDone: async transcript =>
{
var message = new RoleDialogModel(AgentRole.Assistant, transcript);
// append transcript to conversation
storage.Append(conn.ConversationId, message);
foreach (var hook in hookProvider.HooksOrderByPriority)
{
hook.SetAgent(agent)
.SetConversation(conversation);
if (!string.IsNullOrEmpty(transcript))
{
await hook.OnMessageReceived(message);
}
}
},
onModelResponseDone: async response =>
{
var messages = await completer.OnResponsedDone(conn, response);
foreach (var message in messages)
{
// Invoke function
if (message.FunctionName != null)
{
await routing.InvokeFunction(message.FunctionName, message);
var data = await completer.InertConversationItem(message);
await completer.SendEventToModel(data);
}
}
},
onUserInterrupted: async () =>
{
var data = conn.OnModelUserInterrupted();
await SendEventToUser(userWebSocket, data);
});
}
private async Task SendEventToWebSocket(WebSocket webSocket, object message)
private async Task SendEventToUser(WebSocket webSocket, object message)
{
var data = JsonSerializer.Serialize(message);
var buffer = Encoding.UTF8.GetBytes(data);
await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}

1
src/Plugins/BotSharp.Plugin.OpenAI/BotSharp.Plugin.OpenAI.csproj
View file

@ -18,7 +18,6 @@
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Abstraction\BotSharp.Abstraction.csproj" />
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>
</Project>

14
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionBody.cs
View file

@ -5,23 +5,35 @@ namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class RealtimeSessionBody
{
[JsonPropertyName("id")]
[JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
public string Id { get; set; } = null!;
[JsonPropertyName("object")]
[JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
public string Object { get; set; } = null!;
[JsonPropertyName("model")]
[JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)]
public string Model { get; set; } = null!;
[JsonPropertyName("temperature")]
public float temperature { get; set; } = 0.8f;
public float Temperature { get; set; } = 0.8f;
[JsonPropertyName("modalities")]
public string[] Modalities { get; set; } = ["audio", "text"];
[JsonPropertyName("input_audio_format")]
public string InputAudioFormat { get; set; } = "pcm16";
[JsonPropertyName("output_audio_format")]
public string OutputAudioFormat { get; set; } = "pcm16";
[JsonPropertyName("instructions")]
public string Instructions { get; set; } = "You are a friendly assistant.";
[JsonPropertyName("voice")]
public string Voice { get; set; } = "sage";
[JsonPropertyName("max_response_output_tokens")]
public int MaxResponseOutputTokens { get; set; } = 512;

10
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/RealtimeSessionRequest.cs
View file

@ -1,6 +1,14 @@
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class RealtimeSessionRequest : RealtimeSessionBody
public class RealtimeSessionCreationRequest : RealtimeSessionBody
{
}
/// <summary>
/// https://platform.openai.com/docs/api-reference/realtime-client-events/session/update
/// </summary>
public class RealtimeSessionUpdateRequest : RealtimeSessionBody
{
}

19
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseAudioTranscript.cs Normal file
View file

@ -0,0 +1,19 @@
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class ResponseAudioTranscript : ServerEventResponse
{
[JsonPropertyName("response_id")]
public string ResponseId { get; set; } = null!;
[JsonPropertyName("item_id")]
public string ItemId { get; set; } = null!;
[JsonPropertyName("output_index")]
public int OutputIndex { get; set; }
[JsonPropertyName("content_index")]
public int ContentIndex { get; set; }
[JsonPropertyName("transcript")]
public string? Transcript { get; set; }
}

66
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ResponseDone.cs Normal file
View file

@ -0,0 +1,66 @@
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class ResponseDone : ServerEventResponse
{
[JsonPropertyName("response")]
public ResponseDoneBody Body { get; set; } = new();
}
public class ResponseDoneBody
{
[JsonPropertyName("id")]
public string Id { get; set; } = null!;
[JsonPropertyName("object")]
public string Object { get; set; } = null!;
[JsonPropertyName("status")]
public string Status { get; set; } = null!;
[JsonPropertyName("status_details")]
public string? StatusDetails { get; set; } = null!;
[JsonPropertyName("conversation_id")]
public string ConversationId { get; set; } = null!;
[JsonPropertyName("usage")]
public ModelTokenUsage Usage { get; set; } = new();
[JsonPropertyName("output")]
public ModelResponseDoneOutput[] Outputs { get; set; } = [];
}
public class ModelTokenUsage
{
[JsonPropertyName("total_tokens")]
public int TotalTokens { get; set; }
[JsonPropertyName("input_tokens")]
public int InputTokens { get; set; }
[JsonPropertyName("output_tokens")]
public int OutputTokens { get; set; }
}
public class ModelResponseDoneOutput
{
[JsonPropertyName("id")]
public string Id { get; set; } = null!;
[JsonPropertyName("object")]
public string Object { get; set; } = null!;
[JsonPropertyName("type")]
public string Type { get; set; } = null!;
[JsonPropertyName("status")]
public string Status { get; set; } = null!;
[JsonPropertyName("name")]
public string Name { get; set; } = null!;
[JsonPropertyName("call_id")]
public string CallId { get; set; } = null!;
[JsonPropertyName("arguments")]
public string Arguments { get; set; } = null!;
}

19
src/Plugins/BotSharp.Plugin.OpenAI/Models/Realtime/ServerEventErrorResponse.cs Normal file
View file

@ -0,0 +1,19 @@
namespace BotSharp.Plugin.OpenAI.Models.Realtime;
public class ServerEventErrorResponse : ServerEventResponse
{
[JsonPropertyName("error")]
public ServerEventErrorBody Body { get; set; } = new();
}
public class ServerEventErrorBody
{
[JsonPropertyName("type")]
public string Type { get; set; } = null!;
[JsonPropertyName("code")]
public string Code { get; set; } = null!;
[JsonPropertyName("message")]
public string? Message { get; set; }
}

3
src/Plugins/BotSharp.Plugin.OpenAI/OpenAiPlugin.cs
View file

@ -8,8 +8,6 @@ using BotSharp.Plugin.OpenAI.Providers.Audio;
using Microsoft.Extensions.Configuration;
using Refit;
using BotSharp.Plugin.OpenAI.Providers.Realtime;
using BotSharp.Plugin.Twilio.Services.Stream;
using BotSharp.Abstraction.Realtime;
namespace BotSharp.Plugin.OpenAI;
@ -37,7 +35,6 @@ public class OpenAiPlugin : IBotSharpPlugin
services.AddScoped<IImageCompletion, ImageCompletionProvider>();
services.AddScoped<IAudioCompletion, AudioCompletionProvider>();
services.AddScoped<IRealTimeCompletion, RealTimeCompletionProvider>();
services.AddScoped<IRealtimeModelConnector, OpenAiRealtimeModelConnector>();
services.AddRefitClient<IOpenAiRealtimeApi>()
.ConfigureHttpClient(c => c.BaseAddress = new Uri("https://api.openai.com"));

2
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/IOpenAiRealtimeApi.cs
View file

@ -7,5 +7,5 @@ namespace BotSharp.Plugin.OpenAI.Providers.Realtime;
public interface IOpenAiRealtimeApi
{
[Post("/v1/realtime/sessions")]
Task<RealtimeSession> GetSessionAsync(RealtimeSessionRequest model, [Authorize("Bearer")] string token);
Task<RealtimeSession> GetSessionAsync(RealtimeSessionCreationRequest model, [Authorize("Bearer")] string token);
}

186
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/OpenAiRealtimeModelConnector.cs
View file

@ -1,186 +0,0 @@
using BotSharp.Abstraction.Realtime;
using BotSharp.Abstraction.Realtime.Models;
using BotSharp.Core.Infrastructures;
using BotSharp.Plugin.OpenAI.Models.Realtime;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using System.Threading;
using Task = System.Threading.Tasks.Task;
namespace BotSharp.Plugin.Twilio.Services.Stream;
public class OpenAiRealtimeModelConnector : IRealtimeModelConnector
{
private readonly IServiceProvider _services;
private readonly ILogger _logger;
private ClientWebSocket _webSocket;
public OpenAiRealtimeModelConnector(IServiceProvider services, ILogger<OpenAiRealtimeModelConnector> logger)
{
_services = services;
_logger = logger;
}
public async Task Connect(RealtimeHubConnection conn, Action<string> onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted)
{
var convService = _services.GetRequiredService<IConversationService>();
var conv = await convService.GetConversation(conn.ConversationId);
var agentService = _services.GetRequiredService<IAgentService>();
var agent = await agentService.LoadAgent(conv.AgentId);
var completion = CompletionProvider.GetRealTimeCompletion(_services, provider: "openai", modelId: "gpt-4");
var model = completion.Model;
var settingsService = _services.GetRequiredService<ILlmProviderService>();
var settings = settingsService.GetSetting(provider: completion.Provider, model);
_webSocket = new ClientWebSocket();
_webSocket.Options.SetRequestHeader("Authorization", $"Bearer {settings.ApiKey}");
_webSocket.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1");
await _webSocket.ConnectAsync(new Uri($"wss://api.openai.com/v1/realtime?model={model}"), CancellationToken.None);
if (_webSocket.State == WebSocketState.Open)
{
// Receive a message
ReceiveMessage(onAudioDeltaReceived, onAudioResponseDone, onUserInterrupted);
// Control initial session with OpenAI
var sessionUpdate = new
{
type = "session.update",
session = new
{
turn_detection = new { type = "server_vad" },
input_audio_format = "g711_ulaw",
output_audio_format = "g711_ulaw",
voice = "alloy",
instructions = agent.Description,
modalities = new string[] { "text", "audio" },
temperature = 0.8f,
}
};
await SendEventToWebSocket(sessionUpdate);
/*var initialConversationItem = new
{
type = "conversation.item.create",
item = new
{
type = "message",
role = "user",
content = new object[]
{
new {
type = "input_text",
text = "Greet the user with \"Hello there! I am an AI voice assistant powered by Twilio and the OpenAI Realtime API. You can ask me for facts, jokes, or anything you can imagine. How can I help you?\""
}
}
}
};
await SendEventToWebSocket(initialConversationItem);*/
await SendEventToWebSocket(new { type = "response.create" });
}
}
public async Task Disconnect()
{
await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None);
}
public async Task SendMessage(string message)
{
var audioAppend = new
{
type = "input_audio_buffer.append",
audio = message
};
await SendEventToWebSocket(audioAppend);
}
private async Task ReceiveMessage(Action<string> onAudioDeltaReceived, Action onAudioResponseDone, Action onUserInterrupted)
{
var buffer = new byte[1024 * 1024 * 1];
WebSocketReceiveResult result;
string lastAssistantItem = "";
do
{
result = await _webSocket.ReceiveAsync(
new ArraySegment<byte>(buffer), CancellationToken.None);
// Convert received data to text/audio (Twilio sends Base64-encoded audio)
string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count);
if (string.IsNullOrEmpty(receivedText))
{
continue;
}
_logger.LogDebug($"{nameof(OpenAiRealtimeModelConnector)} received: {receivedText}");
var response = JsonSerializer.Deserialize<ServerEventResponse>(receivedText);
if (response.Type == "session.created")
{
}
else if (response.Type == "session.updated")
{
}
else if (response.Type == "response.audio_transcript.delta")
{
}
else if (response.Type == "response.audio_transcript.done")
{
}
else if (response.Type == "response.audio.delta")
{
var audio = JsonSerializer.Deserialize<ResponseAudioDelta>(receivedText);
lastAssistantItem = audio?.ItemId ?? "";
if (audio != null && audio.Delta != null)
{
onAudioDeltaReceived(audio.Delta);
}
}
else if (response.Type == "response.audio.done")
{
onAudioResponseDone();
}
else if (response.Type == "response.done")
{
}
else if (response.Type == "input_audio_buffer.speech_started")
{
// var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio;
// handle use interuption
var truncateEvent = new
{
type = "conversation.item.truncate",
item_id = lastAssistantItem,
content_index = 0,
audio_end_ms = 100
};
await SendEventToWebSocket(truncateEvent);
onUserInterrupted();
}
} while (!result.CloseStatus.HasValue);
await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None);
}
private async Task SendEventToWebSocket(object message)
{
var data = JsonSerializer.Serialize(message);
var buffer = Encoding.UTF8.GetBytes(data);
await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
}

256
src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs
View file

@ -3,10 +3,16 @@ using BotSharp.Abstraction.Functions.Models;
using BotSharp.Abstraction.Realtime.Models;
using BotSharp.Plugin.OpenAI.Models.Realtime;
using OpenAI.Chat;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using System.Threading;
namespace BotSharp.Plugin.OpenAI.Providers.Realtime;
/// <summary>
/// Reference to https://platform.openai.com/docs/api-reference/realtime-server-events
/// </summary>
public class RealTimeCompletionProvider : IRealTimeCompletion
{
public string Provider => "openai";
@ -17,7 +23,7 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
protected readonly ILogger<RealTimeCompletionProvider> _logger;
protected string _model = "gpt-4o-mini-realtime-preview-2024-12-17";
private ClientWebSocket _webSocket;
public RealTimeCompletionProvider(
OpenAiSettings settings,
@ -29,6 +35,150 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
_services = services;
}
public async Task Connect(RealtimeHubConnection conn,
Action onModelReady,
Action<string> onModelAudioDeltaReceived,
Action onModelAudioResponseDone,
Action<string> onAudioTranscriptDone,
Action<string> onModelResponseDone,
Action onUserInterrupted)
{
var settingsService = _services.GetRequiredService<ILlmProviderService>();
var settings = settingsService.GetSetting(provider: "openai", conn.Model);
_webSocket = new ClientWebSocket();
_webSocket.Options.SetRequestHeader("Authorization", $"Bearer {settings.ApiKey}");
_webSocket.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1");
await _webSocket.ConnectAsync(new Uri($"wss://api.openai.com/v1/realtime?model={conn.Model}"), CancellationToken.None);
if (_webSocket.State == WebSocketState.Open)
{
onModelReady();
// Receive a message
_ = ReceiveMessage(onModelAudioDeltaReceived,
onModelAudioResponseDone,
onAudioTranscriptDone,
onModelResponseDone,
onUserInterrupted);
// Triggering model inference
await SendEventToModel(new { type = "response.create" });
}
}
public async Task Disconnect()
{
await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None);
}
public async Task AppenAudioBuffer(string message)
{
var audioAppend = new
{
type = "input_audio_buffer.append",
audio = message
};
await SendEventToModel(audioAppend);
}
private async Task ReceiveMessage(Action<string> onModelAudioDeltaReceived,
Action onModelAudioResponseDone,
Action<string> onAudioTranscriptDone,
Action<string> onModelResponseDone,
Action onUserInterrupted)
{
var buffer = new byte[1024 * 1024 * 1];
WebSocketReceiveResult result;
string lastAssistantItem = "";
do
{
result = await _webSocket.ReceiveAsync(
new ArraySegment<byte>(buffer), CancellationToken.None);
// Convert received data to text/audio (Twilio sends Base64-encoded audio)
string receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count);
if (string.IsNullOrEmpty(receivedText))
{
continue;
}
_logger.LogDebug($"{nameof(RealTimeCompletionProvider)} received: {receivedText}");
var response = JsonSerializer.Deserialize<ServerEventResponse>(receivedText);
if (response.Type == "error")
{
var error = JsonSerializer.Deserialize<ServerEventErrorResponse>(receivedText);
_logger.LogError($"Error: {error.Body.Message}");
}
else if (response.Type == "session.created")
{
}
else if (response.Type == "session.updated")
{
}
else if (response.Type == "response.audio_transcript.delta")
{
}
else if (response.Type == "response.audio_transcript.done")
{
var data = JsonSerializer.Deserialize<ResponseAudioTranscript>(receivedText);
onAudioTranscriptDone(data.Transcript);
}
else if (response.Type == "response.audio.delta")
{
var audio = JsonSerializer.Deserialize<ResponseAudioDelta>(receivedText);
lastAssistantItem = audio?.ItemId ?? "";
if (audio != null && audio.Delta != null)
{
onModelAudioDeltaReceived(audio.Delta);
}
}
else if (response.Type == "response.audio.done")
{
onModelAudioResponseDone();
}
else if (response.Type == "response.done")
{
onModelResponseDone(receivedText);
}
else if (response.Type == "input_audio_buffer.speech_started")
{
// var elapsedTime = latestMediaTimestamp - responseStartTimestampTwilio;
// handle use interuption
var truncateEvent = new
{
type = "conversation.item.truncate",
item_id = lastAssistantItem,
content_index = 0,
audio_end_ms = 100
};
await SendEventToModel(truncateEvent);
onUserInterrupted();
}
} while (!result.CloseStatus.HasValue);
await _webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None);
}
public async Task SendEventToModel(object message)
{
if (message is not string data)
{
data = JsonSerializer.Serialize(message);
}
var buffer = Encoding.UTF8.GetBytes(data);
await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
}
public async Task<RealtimeSession> CreateSession(Agent agent, List<RoleDialogModel> conversations)
{
var contentHooks = _services.GetServices<IContentGeneratingHook>().ToList();
@ -37,9 +187,11 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
var chatClient = client.GetChatClient(_model);
var (prompt, messages, options) = PrepareOptions(agent, conversations);
var args = new RealtimeSessionRequest
var instruction = messages.FirstOrDefault()?.Content.FirstOrDefault()?.Text ?? agent.Description;
var args = new RealtimeSessionCreationRequest
{
Instructions = prompt,
Instructions = instruction,
ToolChoice = "auto",
Tools = options.Tools.Select(x =>
{
@ -61,6 +213,71 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
return session;
}
public async Task<string> UpdateInitialSession(RealtimeHubConnection conn)
{
var convService = _services.GetRequiredService<IConversationService>();
var conv = await convService.GetConversation(conn.ConversationId);
var agentService = _services.GetRequiredService<IAgentService>();
var agent = await agentService.LoadAgent(conv.AgentId);
var client = ProviderHelper.GetClient(Provider, _model, _services);
var chatClient = client.GetChatClient(_model);
var (prompt, messages, options) = PrepareOptions(agent, []);
var instruction = messages.FirstOrDefault()?.Content.FirstOrDefault()?.Text ?? agent.Description;
var sessionUpdate = new
{
type = "session.update",
session = new RealtimeSessionUpdateRequest
{
InputAudioFormat = "g711_ulaw",
OutputAudioFormat = "g711_ulaw",
Voice = "alloy",
Instructions = instruction,
ToolChoice = "auto",
Tools = options.Tools.Select(x =>
{
var fn = new FunctionDef
{
Name = x.FunctionName,
Description = x.FunctionDescription
};
fn.Parameters = JsonSerializer.Deserialize<FunctionParametersDef>(x.FunctionParameters);
return fn;
}).ToArray(),
Modalities = [ "text", "audio" ],
Temperature = Math.Max(options.Temperature ?? 0f, 0.6f)
}
};
return JsonSerializer.Serialize(sessionUpdate);
}
public async Task<string> InertConversationItem(RoleDialogModel message)
{
var conversationItem = new
{
type = "conversation.item.create",
item = new
{
type = "message",
role = message.Role,
content = new object[]
{
new
{
type = "text",
text = message.Content
}
}
}
};
return JsonSerializer.Serialize(conversationItem);
}
protected (string, IEnumerable<ChatMessage>, ChatCompletionOptions) PrepareOptions(Agent agent, List<RoleDialogModel> conversations)
{
var agentService = _services.GetRequiredService<IAgentService>();
@ -174,7 +391,6 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
return (prompt, messages, options);
}
private string GetPrompt(IEnumerable<ChatMessage> messages, ChatCompletionOptions options)
{
var prompt = string.Empty;
@ -246,4 +462,36 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
{
_model = model;
}
public async Task<List<RoleDialogModel>> OnResponsedDone(RealtimeHubConnection conn, string response)
{
var outputs = new List<RoleDialogModel>();
var data = JsonSerializer.Deserialize<ResponseDone>(response).Body;
foreach (var output in data.Outputs)
{
if (output.Type == "function_call")
{
outputs.Add(new RoleDialogModel(AgentRole.Assistant, output.Arguments)
{
FunctionName = output.Name,
FunctionArgs = output.Arguments
});
}
else if (output.Type == "message")
{
outputs.Add(new RoleDialogModel(AgentRole.Assistant, "")
{
FunctionName = output.Name,
FunctionArgs = output.Arguments
});
}
else
{
throw new NotImplementedException($"not implemented for output type {output.Type}");
}
}
return outputs;
}
}

10
src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs
View file

@ -30,6 +30,7 @@ public class TwilioStreamMiddleware
var services = httpContext.RequestServices;
using WebSocket webSocket = await httpContext.WebSockets.AcceptWebSocketAsync();
await HandleWebSocket(services, webSocket);
httpContext.Abort();
}
}
@ -47,10 +48,9 @@ public class TwilioStreamMiddleware
conn.StreamId = response.StreamSid;
conn.Event = response.Event switch
{
"connected" => string.Empty,
"start" => "connected",
"media" => "data_received",
"stop" => "disconnected",
"start" => "user_connected",
"media" => "user_data_received",
"stop" => "user_disconnected",
_ => response.Event
};
@ -83,7 +83,7 @@ public class TwilioStreamMiddleware
if (response.Event == "start")
{
var startResponse = JsonSerializer.Deserialize<StreamEventStartResponse>(receivedText);
conn.Data = startResponse.Body.CallSid;
conn.Data = JsonSerializer.Serialize(startResponse.Body.CustomParameters);
conn.ConversationId = startResponse.Body.CallSid;
}
else if (response.Event == "media")