Add OpenAIRealtimeUpstream + RealtimeSessionFactory + DI registrations

2026-06-11 21:48:21 +00:00
parent 8f53f9daa9
commit 78373b0912
3 changed files with 98 additions and 0 deletions
@@ -0,0 +1,56 @@
+using System.Net.WebSockets;
+using System.Text;
+using System.Text.Json.Nodes;
+
+namespace backend.Realtime;
+
+public class OpenAIRealtimeUpstream : IRealtimeUpstream
+{
+    private readonly ClientWebSocket _ws = new();
+
+    public async Task ConnectAsync(string apiKey, string model, CancellationToken ct)
+    {
+        _ws.Options.SetRequestHeader("Authorization", $"Bearer {apiKey}");
+        _ws.Options.SetRequestHeader("OpenAI-Beta", "realtime=v1");
+        var uri = new Uri($"wss://api.openai.com/v1/realtime?model={Uri.EscapeDataString(model)}");
+        await _ws.ConnectAsync(uri, ct);
+    }
+
+    public async Task SendJsonAsync(JsonObject envelope, CancellationToken ct)
+    {
+        var json = envelope.ToJsonString();
+        var bytes = Encoding.UTF8.GetBytes(json);
+        await _ws.SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, ct);
+    }
+
+    public async Task<JsonObject?> ReceiveJsonAsync(CancellationToken ct)
+    {
+        var buffer = new byte[8192];
+        using var stream = new MemoryStream();
+        while (true)
+        {
+            WebSocketReceiveResult result;
+            try { result = await _ws.ReceiveAsync(buffer, ct); }
+            catch (WebSocketException) { return null; }
+            catch (OperationCanceledException) { return null; }
+
+            if (result.MessageType == WebSocketMessageType.Close) return null;
+            stream.Write(buffer, 0, result.Count);
+            if (result.EndOfMessage)
+            {
+                var json = Encoding.UTF8.GetString(stream.ToArray());
+                return JsonNode.Parse(json)?.AsObject();
+            }
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        if (_ws.State == WebSocketState.Open)
+        {
+            try { await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None); }
+            catch { }
+        }
+        _ws.Dispose();
+    }
+}
@@ -0,0 +1,37 @@
+using backend.Conversations;
+using backend.Tools;
+using Microsoft.Extensions.Logging;
+
+namespace backend.Realtime;
+
+public class RealtimeSessionFactory
+{
+    protected readonly OpenAIKeyProvider keys;
+    protected readonly ToolRegistry registry;
+    protected readonly ConversationLog log;
+    protected readonly ILoggerFactory loggerFactory;
+
+    public RealtimeSessionFactory(
+        OpenAIKeyProvider keys, ToolRegistry registry,
+        ConversationLog log, ILoggerFactory loggerFactory)
+    {
+        this.keys = keys;
+        this.registry = registry;
+        this.log = log;
+        this.loggerFactory = loggerFactory;
+    }
+
+    public virtual async Task<RealtimeSession> CreateAsync(
+        Guid deviceId,
+        RealtimeSettings cfg,
+        ISessionOutput output,
+        IDeviceChannel channel,
+        CancellationToken ct)
+    {
+        var upstream = new OpenAIRealtimeUpstream();
+        await upstream.ConnectAsync(keys.GetRequired(), cfg.Model, ct);
+        return new RealtimeSession(
+            deviceId, upstream, output, log, registry, cfg, channel,
+            loggerFactory.CreateLogger<RealtimeSession>());
+    }
+}