diff --git a/packages/console/app/src/routes/zen/util/provider/openai-compatible.ts b/packages/console/app/src/routes/zen/util/provider/openai-compatible.ts index 666ed52f97e4..43af6605e531 100644 --- a/packages/console/app/src/routes/zen/util/provider/openai-compatible.ts +++ b/packages/console/app/src/routes/zen/util/provider/openai-compatible.ts @@ -1,4 +1,4 @@ -import { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider" +import type { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider" type Usage = { prompt_tokens?: number diff --git a/packages/console/app/src/routes/zen/util/provider/openai.ts b/packages/console/app/src/routes/zen/util/provider/openai.ts index cebc78a127a3..d79101167a9a 100644 --- a/packages/console/app/src/routes/zen/util/provider/openai.ts +++ b/packages/console/app/src/routes/zen/util/provider/openai.ts @@ -1,4 +1,4 @@ -import { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider" +import type { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider" type Usage = { input_tokens?: number @@ -556,75 +556,160 @@ export function fromOpenaiChunk(chunk: string): CommonChunk | string { return out } -export function toOpenaiChunk(chunk: CommonChunk): string { - if (!chunk.choices || !Array.isArray(chunk.choices) || chunk.choices.length === 0) { - return "" - } +export function createToOpenaiChunk() { + let responseId = "" + let itemId = "" + let started = false + let text = "" + let itemAdded = false + let partAdded = false + let itemDone = false + let toolCallId = "" + let toolName = "" + let toolArguments = "" + + const messageItem = (status: string) => ({ + id: itemId, + type: "message", + status, + role: "assistant", + content: partAdded ? [{ type: "output_text", text, annotations: [] }] : [], + }) + + const toolCallItem = () => ({ + id: toolCallId, + type: "function_call", + call_id: toolCallId, + name: toolName, + arguments: toolArguments, + }) + + return (chunk: CommonChunk): string => { + if (!chunk.choices || !Array.isArray(chunk.choices) || chunk.choices.length === 0) { + return "" + } - const choice = chunk.choices[0] - const d = choice.delta - if (!d) return "" + const choice = chunk.choices[0] + const d = choice.delta + if (!d) return "" - const id = chunk.id - const model = chunk.model + const events: string[] = [] + const emit = (event: string, data: Record) => { + events.push(`event: ${event}\ndata: ${JSON.stringify(data)}`) + } - if (d.content) { - const data = { - id, - type: "response.output_text.delta", - delta: d.content, - response: { id, model }, + if (!started) { + started = true + responseId = chunk.id || `resp_${Math.random().toString(36).slice(2)}` + itemId = `msg_${Math.random().toString(36).slice(2)}` + const response = { id: responseId, object: "response", status: "in_progress", model: chunk.model, output: [] } + emit("response.created", { type: "response.created", response }) + emit("response.in_progress", { type: "response.in_progress", response }) } - return `event: response.output_text.delta\ndata: ${JSON.stringify(data)}` - } - if (d.tool_calls) { - for (const tc of d.tool_calls) { - if (tc.function?.name) { - const data = { + if (d.content) { + if (!itemAdded) { + itemAdded = true + emit("response.output_item.added", { type: "response.output_item.added", output_index: 0, - item: { - id: tc.id, - type: "function_call", - name: tc.function.name, - call_id: tc.id, - arguments: "", - }, - } - return `event: response.output_item.added\ndata: ${JSON.stringify(data)}` + item: messageItem("in_progress"), + }) } - if (tc.function?.arguments) { - const data = { - type: "response.function_call_arguments.delta", + if (!partAdded) { + partAdded = true + emit("response.content_part.added", { + type: "response.content_part.added", output_index: 0, - delta: tc.function.arguments, + item_id: itemId, + part: { type: "output_text", text: "", annotations: [] }, + }) + } + text += d.content + emit("response.output_text.delta", { + id: responseId, + type: "response.output_text.delta", + delta: d.content, + response: { id: responseId, model: chunk.model }, + }) + } + + if (d.tool_calls) { + for (const tc of d.tool_calls) { + if (!tc || tc.type !== "function") continue + if (tc.function?.name) { + toolCallId = tc.id || `call_${Math.random().toString(36).slice(2)}` + toolName = tc.function.name + itemAdded = true + emit("response.output_item.added", { + type: "response.output_item.added", + output_index: 0, + item: toolCallItem(), + }) + } + if (tc.function?.arguments) { + toolArguments += tc.function.arguments + emit("response.function_call_arguments.delta", { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: toolCallId, + delta: tc.function.arguments, + }) } - return `event: response.function_call_arguments.delta\ndata: ${JSON.stringify(data)}` } } - } - if (choice.finish_reason) { - const u = chunk.usage - const usage = u - ? { - input_tokens: u.prompt_tokens, - output_tokens: u.completion_tokens, - total_tokens: u.total_tokens, - ...(u.prompt_tokens_details?.cached_tokens - ? { input_tokens_details: { cached_tokens: u.prompt_tokens_details.cached_tokens } } - : {}), + if (choice.finish_reason) { + if (itemAdded && !itemDone) { + itemDone = true + if (toolCallId) { + emit("response.function_call_arguments.done", { + type: "response.function_call_arguments.done", + output_index: 0, + item_id: toolCallId, + arguments: toolArguments, + }) + emit("response.output_item.done", { + type: "response.output_item.done", + output_index: 0, + item: toolCallItem(), + }) + } else { + emit("response.output_item.done", { + type: "response.output_item.done", + output_index: 0, + item: messageItem("completed"), + }) } - : undefined + } - const data: any = { - id, - type: "response.completed", - response: { id, model, ...(usage ? { usage } : {}) }, + const u = chunk.usage + const usage = u + ? { + input_tokens: u.prompt_tokens, + output_tokens: u.completion_tokens, + total_tokens: u.total_tokens, + ...(u.prompt_tokens_details?.cached_tokens + ? { input_tokens_details: { cached_tokens: u.prompt_tokens_details.cached_tokens } } + : {}), + } + : undefined + + const stop_reason = (() => { + const r = choice.finish_reason + if (r === "stop") return "stop" + if (r === "tool_calls") return "tool_call" + if (r === "length") return "max_output_tokens" + if (r === "content_filter") return "content_filter" + return null + })() + + const output = toolCallId ? [toolCallItem()] : itemAdded ? [messageItem("completed")] : [] + const response: any = { id: responseId, model: chunk.model, output, stop_reason } + if (usage) response.usage = usage + emit("response.completed", { id: responseId, type: "response.completed", response }) } - return `event: response.completed\ndata: ${JSON.stringify(data)}` - } - return "" + return events.join("\n\n") + } } diff --git a/packages/console/app/src/routes/zen/util/provider/provider.ts b/packages/console/app/src/routes/zen/util/provider/provider.ts index 9b26c82377b1..50bdfce08e4c 100644 --- a/packages/console/app/src/routes/zen/util/provider/provider.ts +++ b/packages/console/app/src/routes/zen/util/provider/provider.ts @@ -8,10 +8,10 @@ import { toAnthropicResponse, } from "./anthropic" import { + createToOpenaiChunk, fromOpenaiChunk, fromOpenaiRequest, fromOpenaiResponse, - toOpenaiChunk, toOpenaiRequest, toOpenaiResponse, } from "./openai" @@ -195,6 +195,7 @@ export function createBodyConverter(from: ZenData.Format, to: ZenData.Format) { } export function createStreamPartConverter(from: ZenData.Format, to: ZenData.Format) { + const toOpenaiChunk = to === "openai" ? createToOpenaiChunk() : undefined return (part: any): any => { if (from === to) return part @@ -207,7 +208,7 @@ export function createStreamPartConverter(from: ZenData.Format, to: ZenData.Form if (typeof raw === "string") return raw if (to === "anthropic") return toAnthropicChunk(raw) - if (to === "openai") return toOpenaiChunk(raw) + if (to === "openai") return toOpenaiChunk!(raw) if (to === "oa-compat") return toOaCompatibleChunk(raw) } } diff --git a/packages/console/app/test/openaiResponsesStream.test.ts b/packages/console/app/test/openaiResponsesStream.test.ts new file mode 100644 index 000000000000..cfb6d2c0a2a2 --- /dev/null +++ b/packages/console/app/test/openaiResponsesStream.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, test } from "bun:test" +import { createToOpenaiChunk } from "../src/routes/zen/util/provider/openai" +import { fromOaCompatibleChunk } from "../src/routes/zen/util/provider/openai-compatible" + +function convert(chunks: string[]) { + const toOpenai = createToOpenaiChunk() + return chunks + .map((part) => { + const raw = fromOaCompatibleChunk(part) + return typeof raw === "string" ? raw : toOpenai(raw) + }) + .filter((part) => part.length > 0) + .join("\n\n") +} + +const chunk = (delta: string, finishReason?: string, usage?: Record) => + JSON.stringify({ + id: "chatcmpl-1", + object: "chat.completion.chunk", + created: 1, + model: "deepseek-v4-flash", + choices: [{ index: 0, delta: JSON.parse(delta), finish_reason: finishReason ?? null }], + ...(usage ? { usage } : {}), + }) + +const eventOrder = (stream: string) => [...stream.matchAll(/event: ([a-z_.]+)/g)].map((m) => m[1]) + +const eventData = (stream: string, event: string) => { + const match = stream.match(new RegExp(`event: ${event}\\ndata: (\\{.*?\\})(?:\\n\\n|$)`)) + if (!match) throw new Error(`missing ${event}`) + return JSON.parse(match[1]) +} + +describe("createToOpenaiChunk", () => { + test("text stream emits the full Responses-API lifecycle in order", () => { + const stream = convert([ + `data: ${chunk('{"role":"assistant","content":""}')}`, + `data: ${chunk('{"content":"ok"}')}`, + `data: ${chunk('{"content":"!"}')}`, + `data: ${chunk("{}", "stop", { prompt_tokens: 10, completion_tokens: 3, total_tokens: 13 })}`, + "data: [DONE]", + ]) + + expect(eventOrder(stream)).toEqual([ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.delta", + "response.output_item.done", + "response.completed", + ]) + expect((stream.match(/event: response\.created/g) ?? []).length).toBe(1) + expect(stream).toContain("data: [DONE]") + + const item = eventData(stream, "response.output_item.done").item + expect(item).toMatchObject({ type: "message", status: "completed", role: "assistant" }) + expect(item.content[0].text).toBe("ok!") + + const completed = eventData(stream, "response.completed").response + expect(completed.output[0].content[0].text).toBe("ok!") + expect(completed.usage).toEqual({ + input_tokens: 10, + output_tokens: 3, + total_tokens: 13, + }) + }) + + test("[DONE] passes through unchanged", () => { + expect(convert(["data: [DONE]"])).toBe("data: [DONE]") + }) + + test("tool call stream emits the function-call lifecycle", () => { + const stream = convert([ + `data: ${chunk('{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]}')}`, + `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"{\\"city\\":"}}]}')}`, + `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"\\"Sydney\\""}}]}')}`, + `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"}"}}]}')}`, + `data: ${chunk("{}", "tool_calls")}`, + ]) + + expect(eventOrder(stream)).toEqual([ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.function_call_arguments.delta", + "response.function_call_arguments.delta", + "response.function_call_arguments.delta", + "response.function_call_arguments.done", + "response.output_item.done", + "response.completed", + ]) + + const item = eventData(stream, "response.output_item.done").item + expect(item).toMatchObject({ type: "function_call", name: "get_weather" }) + expect(item.arguments).toBe('{"city":"Sydney"}') + }) + + test("stream with no visible text still opens and closes the response", () => { + const stream = convert([`data: ${chunk("{}", "stop")}`]) + expect(eventOrder(stream)).toEqual(["response.created", "response.in_progress", "response.completed"]) + expect(eventData(stream, "response.completed").response.output).toEqual([]) + }) +})