[ResponsesAPI] Implement streaming mode (#1582)

Wauplin · web-flow · commit e47b3f9fc4ba · 2025-07-02T10:09:10.000+02:00
Built on top of #1576. This PR adds support for streaming mode to the Responses API. Tested it using the [openai-responses-starter-app](https://github.com/openai/openai-responses-starter-app): [Screencast from 02-07-2025 07:43:52.webm](https://github.com/user-attachments/assets/6eb77c9c-5796-4841-af55-f526da8da847) ``` pnpm run example streaming ``` ```js { type: 'response.created', response: { object: 'response', id: 'resp_861131785bfb75f24f944aa7cbc4767b194a2ea320cff258', status: 'in_progress', error: null, instructions: null, model: 'Qwen/Qwen2.5-VL-7B-Instruct', temperature: 1, top_p: 1, created_at: 1751383702199, output: [] }, sequence_number: 0 } { type: 'response.in_progress', response: { object: 'response', id: 'resp_861131785bfb75f24f944aa7cbc4767b194a2ea320cff258', status: 'in_progress', error: null, instructions: null, model: 'Qwen/Qwen2.5-VL-7B-Instruct', temperature: 1, top_p: 1, created_at: 1751383702199, output: [] }, sequence_number: 1 } { type: 'response.output_item.added', output_index: 0, item: { id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', type: 'message', role: 'assistant', status: 'in_progress', content: [] }, sequence_number: 2 } { type: 'response.content_part.added', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, part: { type: 'output_text', text: '', annotations: [] }, sequence_number: 3 } { type: 'response.output_text.delta', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, delta: 'Double', sequence_number: 4 } { type: 'response.output_text.delta', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, delta: ' bubble', sequence_number: 5 } ... { type: 'response.output_text.delta', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, delta: '!', sequence_number: 43 } { type: 'response.output_text.done', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, text: 'Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath!', sequence_number: 44 } { type: 'response.content_part.done', item_id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', output_index: 0, content_index: 0, part: { type: 'output_text', text: 'Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath! Double bubble bath!', annotations: [] }, sequence_number: 45 } { type: 'response.output_item.done', output_index: 0, item: { id: 'msg_def4b731a2654f7eab4fb2efdff217079da37154709c0f0b', type: 'message', role: 'assistant', status: 'completed', content: [ [Object] ] }, sequence_number: 46 } { type: 'response.completed', response: { object: 'response', id: 'resp_861131785bfb75f24f944aa7cbc4767b194a2ea320cff258', status: 'completed', error: null, instructions: null, model: 'Qwen/Qwen2.5-VL-7B-Instruct', temperature: 1, top_p: 1, created_at: 1751383702199, output: [ [Object] ] }, sequence_number: 47 } ```
diff --git a/packages/responses-server/examples/streaming.js b/packages/responses-server/examples/streaming.js
@@ -0,0 +1,17 @@
+import { OpenAI } from "openai";
+const openai = new OpenAI({ baseURL: "http://localhost:3000/v1", apiKey: process.env.HF_TOKEN });
+
+const stream = await openai.responses.create({
+	model: "Qwen/Qwen2.5-VL-7B-Instruct",
+	input: [
+		{
+			role: "user",
+			content: "Say 'double bubble bath' ten times fast.",
+		},
+	],
+	stream: true,
+});
+
+for await (const event of stream) {
+	console.log(event);
+}
diff --git a/packages/responses-server/src/routes/responses.ts b/packages/responses-server/src/routes/responses.ts
@@ -5,7 +5,12 @@ import { generateUniqueId } from "../lib/generateUniqueId.js";
 import { InferenceClient } from "@huggingface/inference";
 import type { ChatCompletionInputMessage, ChatCompletionInputMessageChunkType } from "@huggingface/tasks";
 
-import { type Response as OpenAIResponse } from "openai/resources/responses/responses";
+import type {
+	Response,
+	ResponseStreamEvent,
+	ResponseOutputItem,
+	ResponseContentPartAddedEvent,
+} from "openai/resources/responses/responses";
 
 export const postCreateResponse = async (
 	req: ValidatedRequest<CreateResponseParams>,
@@ -33,27 +38,189 @@ export const postCreateResponse = async (
 				content:
 					typeof item.content === "string"
 						? item.content
-						: item.content.map((content) => {
-								if (content.type === "input_image") {
-									return {
-										type: "image_url" as ChatCompletionInputMessageChunkType,
-										image_url: {
-											url: content.image_url,
-										},
-									};
-								}
-								// content.type must be "input_text" at this point
-								return {
-									type: "text" as ChatCompletionInputMessageChunkType,
-									text: content.text,
-								};
-						  }),
+						: item.content
+								.map((content) => {
+									switch (content.type) {
+										case "input_image":
+											return {
+												type: "image_url" as ChatCompletionInputMessageChunkType,
+												image_url: {
+													url: content.image_url,
+												},
+											};
+										case "output_text":
+											return {
+												type: "text" as ChatCompletionInputMessageChunkType,
+												text: content.text,
+											};
+										case "refusal":
+											return undefined;
+										case "input_text":
+											return {
+												type: "text" as ChatCompletionInputMessageChunkType,
+												text: content.text,
+											};
+									}
+								})
+								.filter((item) => item !== undefined),
 			}))
 		);
 	} else {
 		messages.push({ role: "user", content: req.body.input });
 	}
 
+	const payload = {
+		model: req.body.model,
+		messages: messages,
+		temperature: req.body.temperature,
+		top_p: req.body.top_p,
+		stream: req.body.stream,
+	};
+
+	const responseObject: Omit<
+		Response,
+		"incomplete_details" | "metadata" | "output_text" | "parallel_tool_calls" | "tool_choice" | "tools"
+	> = {
+		object: "response",
+		id: generateUniqueId("resp"),
+		status: "in_progress",
+		error: null,
+		instructions: req.body.instructions,
+		model: req.body.model,
+		temperature: req.body.temperature,
+		top_p: req.body.top_p,
+		created_at: new Date().getTime(),
+		output: [],
+	};
+
+	if (req.body.stream) {
+		res.setHeader("Content-Type", "text/event-stream");
+		res.setHeader("Connection", "keep-alive");
+		let sequenceNumber = 0;
+
+		// Emit events in sequence
+		const emitEvent = (event: ResponseStreamEvent) => {
+			res.write(`data: ${JSON.stringify(event)}\n\n`);
+		};
+
+		try {
+			// Response created event
+			emitEvent({
+				type: "response.created",
+				response: responseObject as Response,
+				sequence_number: sequenceNumber++,
+			});
+
+			// Response in progress event
+			emitEvent({
+				type: "response.in_progress",
+				response: responseObject as Response,
+				sequence_number: sequenceNumber++,
+			});
+
+			const stream = client.chatCompletionStream(payload);
+
+			const outputObject: ResponseOutputItem = {
+				id: generateUniqueId("msg"),
+				type: "message",
+				role: "assistant",
+				status: "in_progress",
+				content: [],
+			};
+			responseObject.output = [outputObject];
+
+			// Response output item added event
+			emitEvent({
+				type: "response.output_item.added",
+				output_index: 0,
+				item: outputObject,
+				sequence_number: sequenceNumber++,
+			});
+
+			// Response content part added event
+			const contentPart: ResponseContentPartAddedEvent["part"] = {
+				type: "output_text",
+				text: "",
+				annotations: [],
+			};
+			outputObject.content.push(contentPart);
+
+			emitEvent({
+				type: "response.content_part.added",
+				item_id: outputObject.id,
+				output_index: 0,
+				content_index: 0,
+				part: contentPart,
+				sequence_number: sequenceNumber++,
+			});
+
+			for await (const chunk of stream) {
+				if (chunk.choices[0].delta.content) {
+					contentPart.text += chunk.choices[0].delta.content;
+
+					// Response output text delta event
+					emitEvent({
+						type: "response.output_text.delta",
+						item_id: outputObject.id,
+						output_index: 0,
+						content_index: 0,
+						delta: chunk.choices[0].delta.content,
+						sequence_number: sequenceNumber++,
+					});
+				}
+			}
+
+			// Response output text done event
+			emitEvent({
+				type: "response.output_text.done",
+				item_id: outputObject.id,
+				output_index: 0,
+				content_index: 0,
+				text: contentPart.text,
+				sequence_number: sequenceNumber++,
+			});
+
+			// Response content part done event
+			emitEvent({
+				type: "response.content_part.done",
+				item_id: outputObject.id,
+				output_index: 0,
+				content_index: 0,
+				part: contentPart,
+				sequence_number: sequenceNumber++,
+			});
+
+			// Response output item done event
+			outputObject.status = "completed";
+			emitEvent({
+				type: "response.output_item.done",
+				output_index: 0,
+				item: outputObject,
+				sequence_number: sequenceNumber++,
+			});
+
+			// Response completed event
+			responseObject.status = "completed";
+			emitEvent({
+				type: "response.completed",
+				response: responseObject as Response,
+				sequence_number: sequenceNumber++,
+			});
+		} catch (streamError: any) {
+			console.error("Error in streaming chat completion:", streamError);
+
+			emitEvent({
+				type: "error",
+				code: null,
+				message: streamError.message || "An error occurred while streaming from inference server.",
+				param: null,
+				sequence_number: sequenceNumber++,
+			});
+		}
+		res.end();
+		return;
+	}
+
 	try {
 		const chatCompletionResponse = await client.chatCompletion({
 			model: req.body.model,
@@ -62,37 +229,24 @@ export const postCreateResponse = async (
 			top_p: req.body.top_p,
 		});
 
-		const responseObject: Omit<
-			OpenAIResponse,
-			"incomplete_details" | "metadata" | "output_text" | "parallel_tool_calls" | "tool_choice" | "tools"
-		> = {
-			object: "response",
-			id: generateUniqueId("resp"),
-			status: "completed",
-			error: null,
-			instructions: req.body.instructions,
-			model: req.body.model,
-			temperature: req.body.temperature,
-			top_p: req.body.top_p,
-			created_at: chatCompletionResponse.created,
-			output: chatCompletionResponse.choices[0].message.content
-				? [
-						{
-							id: generateUniqueId("msg"),
-							type: "message",
-							role: "assistant",
-							status: "completed",
-							content: [
-								{
-									type: "output_text",
-									text: chatCompletionResponse.choices[0].message.content,
-									annotations: [],
-								},
-							],
-						},
-				  ]
-				: [],
-		};
+		responseObject.status = "completed";
+		responseObject.output = chatCompletionResponse.choices[0].message.content
+			? [
+					{
+						id: generateUniqueId("msg"),
+						type: "message",
+						role: "assistant",
+						status: "completed",
+						content: [
+							{
+								type: "output_text",
+								text: chatCompletionResponse.choices[0].message.content,
+								annotations: [],
+							},
+						],
+					},
+			  ]
+			: [];
 
 		res.json(responseObject);
 	} catch (error) {
diff --git a/packages/responses-server/src/schemas.ts b/packages/responses-server/src/schemas.ts
