diff options
Diffstat (limited to 'libpanto/src/provider_openai_chat.zig')
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 24 |
1 files changed, 20 insertions, 4 deletions
diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index e1b9f81..c03d797 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -29,10 +29,10 @@ const ActiveBlock = enum { none, text, thinking }; pub const OpenAIChatProvider = struct { allocator: Allocator, io: Io, - config: config_mod.Config, + config: config_mod.OpenAIChatConfig, http_client: http.Client, - pub fn init(allocator: Allocator, io: Io, cfg: config_mod.Config) OpenAIChatProvider { + pub fn init(allocator: Allocator, io: Io, cfg: config_mod.OpenAIChatConfig) OpenAIChatProvider { return .{ .allocator = allocator, .io = io, @@ -64,9 +64,15 @@ pub const OpenAIChatProvider = struct { return self.streamStep(conv, receiver); } + /// Called via the `Provider` interface. Tears down the impl AND frees + /// its heap allocation, since `Provider.init` is the one that allocated + /// it. Direct stack-allocated users (tests, embedders) call `deinit` + /// themselves and never hit this path. fn vtableDeinit(ptr: *anyopaque) void { const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + const allocator = self.allocator; self.deinit(); + allocator.destroy(self); } pub fn streamStep( @@ -120,6 +126,9 @@ pub const OpenAIChatProvider = struct { // response; we want to stream the body as it arrives. var req = try self.http_client.request(.POST, uri, .{ .extra_headers = &extra_headers, + // Disable compression: gzip buffers small SSE frames, defeating + // the streaming property we paid for `stream: true` to get. + .headers = .{ .accept_encoding = .{ .override = "identity" } }, .keep_alive = false, .redirect_behavior = .not_allowed, }); @@ -167,9 +176,16 @@ pub const OpenAIChatProvider = struct { var state: StreamState = .init(self.allocator); defer state.deinit(); + // Use `readVec` so we return to the event loop as soon as *any* + // bytes arrive, rather than waiting for the buffer to fill. + // `readSliceShort` blocks until EOF or full, which defeats streaming. var chunk: [4096]u8 = undefined; + var vecs: [1][]u8 = .{&chunk}; while (true) { - const n = try body_reader.readSliceShort(&chunk); + const n = body_reader.readVec(&vecs) catch |err| switch (err) { + error.EndOfStream => break, + else => return err, + }; if (n == 0) break; const events = try parser.feed(chunk[0..n]); @@ -232,7 +248,7 @@ const StreamState = struct { const block: conversation.ContentBlock = switch (self.active) { .text => .{ .Text = self.current_buf }, - .thinking => .{ .Thinking = self.current_buf }, + .thinking => .{ .Thinking = .{ .text = self.current_buf } }, .none => unreachable, }; // The buffer ownership has moved into `block`; replace with empty. |
