源码
packages/ai/src/api/openai-responses.ts
import OpenAI from "openai";
import type {
ResponseCreateParamsStreaming,
ResponseInput,
ResponseOutputItem,
} from "openai/resources/responses/responses";
import type { Model, SimpleStreamOptions, ThinkingContent, ToolCall, TranscriptContext } from "../types.ts";
import { measureContextBreakdown } from "../utils/context-breakdown.ts";
import { calculateCost } from "../utils/cost.ts";
import { formatProviderError, normalizeProviderError } from "../utils/error-body.ts";
import { projectContextForModelMediaPolicy } from "../utils/media-policy.ts";
import { MeasuredAssistantStream } from "../utils/model-performance.ts";
import { modelFailure } from "../utils/model-retry.ts";
import { retryProviderRequest } from "../utils/provider-retry.ts";
import { reasoningParameters } from "../utils/reasoning.ts";
import { redactProviderError } from "../utils/redact-error.ts";
import { renderSystemMessageUpdate } from "../utils/text.ts";
import {
assertSupportedOptions,
contentText,
emptyAssistant,
getCurrentTools,
getInitialSystemMessage,
isJsonObject,
} from "../utils/transcript.ts";
import { responsesCache } from "./responses-cache.ts";
import { responsesEvents } from "./responses-events.ts";
import { transformMessages } from "./transform-messages.ts";
export function responsesInput(model: Model, context: TranscriptContext): ResponseInput {
const input: ResponseInput = [];
for (const [index, message] of transformMessages(context.messages, model).entries()) {
if (message.role === "system") {
if (index > 0) {
const text = renderSystemMessageUpdate(message);
if (text) input.push({ role: "developer", content: text });
}
continue;
}
if (message.role === "user") {
input.push({
role: "user",
content:
typeof message.content === "string"
? message.content
: message.content.map((block) =>
block.type === "text"
? { type: "input_text", text: block.text }
: {
type: "input_image",
image_url: `data:${block.mimeType};base64,${block.data}`,
detail: "auto",
},
),
});
} else if (message.role === "assistant") {
for (const block of message.content) {
if (block.type === "text" && block.text.trim()) {
input.push({
role: "assistant",
content: block.text,
...(block.phase ? { phase: block.phase } : {}),
} as ResponseInput[number]);
} else if (block.type === "thinking" && block.thinkingSignature) {
const item: unknown = JSON.parse(block.thinkingSignature);
if (isJsonObject(item) && item.type === "reasoning")
input.push(item as unknown as ResponseInput[number]);
} else if (block.type === "toolCall") {
const [callId, itemId] = block.id.split("|");
input.push({
type: "function_call",
call_id: callId,
...(itemId ? { id: itemId } : {}),
name: block.name,
arguments: JSON.stringify(block.arguments),
...(model.auth === "chatgpt"
? { namespace: "ZPI" }
: block.namespace
? { namespace: block.namespace }
: {}),
});
}
}
} else {
const text = contentText(message.content);
input.push({
type: "function_call_output",
call_id: message.toolCallId.split("|")[0],
output: text || "Image result attached.",
});
const images = message.content.filter((c) => c.type === "image");
if (images.length && model.input.includes("image"))
input.push({
role: "user",
content: [
{ type: "input_text", text: `Images from tool ${message.toolName}:` },
...images.map((c) => ({
type: "input_image" as const,
image_url: `data:${c.mimeType};base64,${c.data}`,
detail: "auto" as const,
})),
],
});
}
}
return input;
}
export function streamSimple(model: Model, context: TranscriptContext, options: SimpleStreamOptions = {}) {
assertSupportedOptions(
options,
[
"signal",
"sessionId",
"responsesSession",
"responsesWarmup",
"cacheRetention",
"skipCacheWrite",
"apiKey",
"fetch",
"headers",
"temperature",
"maxTokens",
"timeoutMs",
"maxRetries",
"maxRetryDelayMs",
"onRetry",
"onPayload",
"onResponse",
"onPerformance",
"onMediaProjection",
"samplingParams",
"reasoning",
"toolChoice",
],
"streamSimple",
);
if (model.api !== "openai-responses") throw new Error(`Unsupported API: ${model.api}`);
if (options.apiKey === undefined) throw new Error("API key must be configured");
const events = new MeasuredAssistantStream(options);
let output = emptyAssistant(model);
const complete = new Set<ToolCall>();
void consume(events).catch((error: unknown) => {
output.content = output.content.filter((block) => block.type !== "toolCall" || complete.has(block));
output.stopReason = options.signal?.aborted ? "aborted" : "error";
const message = formatProviderError(
normalizeProviderError(error),
`${model.provider === "openai" ? "OpenAI" : model.provider} API error`,
);
output.errorMessage = redactProviderError(message, [
options.apiKey,
...Object.values(model.headers ?? {}),
...Object.values(options.headers ?? {}),
]);
output.errorDetails = modelFailure(error, options.signal);
events.push({ type: "error", reason: output.stopReason, error: output });
});
return events;
async function consume(events: MeasuredAssistantStream): Promise<void> {
output = emptyAssistant(model);
complete.clear();
context = projectContextForModelMediaPolicy(context, model, { onProjection: options.onMediaProjection });
const client = new OpenAI({
apiKey: options.apiKey || "ZPI-no-auth",
adminAPIKey: null,
organization: null,
project: null,
baseURL: model.baseUrl,
maxRetries: 0,
timeout: options.timeoutMs,
fetch: options.fetch,
defaultHeaders: {
...(options.sessionId
? { session_id: options.sessionId, "x-client-request-id": options.sessionId }
: {}),
...model.headers,
...options.headers,
...(options.apiKey === "" ? { Authorization: null } : {}),
},
});
const tools = getCurrentTools(context.messages).map((t) => ({
type: "function" as const,
name: t.name,
description: t.description,
parameters: t.parameters,
strict: false,
}));
const thinking = reasoningParameters(model, options.reasoning);
const effort = thinking?.reasoning_effort;
const reasoning = isJsonObject(thinking?.reasoning)
? { summary: "auto", ...thinking.reasoning }
: typeof effort === "string"
? { effort, summary: "auto" }
: undefined;
const initialSystem = getInitialSystemMessage(context.messages);
let payload: unknown = {
model: model.id,
...responsesCache(model, options, responsesInput(model, context)),
instructions: initialSystem ? renderSystemMessageUpdate(initialSystem) : "",
store: false,
stream: true,
include: ["reasoning.encrypted_content"],
...(tools.length
? {
tools:
model.auth === "chatgpt"
? [{ type: "namespace", name: "ZPI", description: "Local coding tools", tools }]
: tools,
}
: {}),
...(options.toolChoice ? { tool_choice: options.toolChoice } : {}),
...(reasoning ? { reasoning } : {}),
...(model.auth !== "chatgpt"
? {
max_output_tokens: options.maxTokens ?? model.maxTokens,
...(options.temperature !== undefined ? { temperature: options.temperature } : {}),
}
: {}),
...model.samplingParams,
...options.samplingParams,
};
const replacement = await options.onPayload?.(payload, model);
if (replacement !== undefined) payload = replacement;
if (!isJsonObject(payload) || payload.stream !== true || !Array.isArray(payload.input))
throw new Error("Payload must be a streaming Responses request object");
if (isJsonObject(payload.reasoning) && payload.reasoning.summary === undefined)
payload.reasoning.summary = "auto";
// Keep callers' structured-output preferences portable across both transports.
const format = payload.response_format;
if (isJsonObject(format)) {
const schema = format.json_schema;
payload.text = {
...(isJsonObject(payload.text) ? payload.text : {}),
format:
format.type === "json_schema" && isJsonObject(schema) ? { type: "json_schema", ...schema } : format,
};
delete payload.response_format;
}
if (isJsonObject(payload.reasoning) && typeof payload.reasoning.effort === "string")
output.providerThinkingLevel = payload.reasoning.effort;
if (model.auth === "chatgpt") {
for (const key of [
"background",
"conversation",
"max_output_tokens",
"max_tool_calls",
"metadata",
"moderation",
"multi_agent",
"prompt",
"prompt_cache_retention",
"prompt_cache_options",
"safety_identifier",
"temperature",
"top_logprobs",
"top_p",
"truncation",
"user",
"previous_response_id",
])
delete payload[key];
payload.store = false;
}
output.contextBreakdown = measureContextBreakdown(payload);
async function http() {
const { data: chunks, response } = await retryProviderRequest(
() => {
events.startRequest();
return client.responses
.create(payload as unknown as ResponseCreateParamsStreaming, { signal: options.signal })
.withResponse();
},
{ ...options, maxRetries: options.maxRetries ?? 2 },
);
await options.onResponse?.(
{ status: response.status, headers: Object.fromEntries(response.headers) },
model,
);
if (options.signal?.aborted) throw new Error("Request aborted");
return responsesEvents(chunks, response);
}
const session = options.responsesSession;
if (options.responsesWarmup && !session) throw new Error("Responses WebSocket warmup unavailable");
const chunks = session
? session.events(
client,
model,
payload as Record<string, unknown> & { input: unknown[] },
options,
http,
() => events.startRequest(),
)
: await http();
events.push({ type: "start", partial: output });
const blocks = new Map<string, number>();
const thinkingParts = new Map<string, string[]>();
const args = new Map<string, string>();
const ended = new Set<string>();
const invalidToolInputs = new Map<string, unknown>();
const contentKey = (itemId: string, index: number) => `${itemId}:${index}`;
function append(key: string, type: "text" | "thinking", delta: string) {
let index = blocks.get(key);
if (index === undefined) {
index = output.content.length;
blocks.set(key, index);
output.content.push(type === "text" ? { type, text: "" } : { type, thinking: "" });
events.push({
type: type === "text" ? "text_start" : "thinking_start",
contentIndex: index,
partial: output,
});
}
const block = output.content[index];
if (block.type === "text") block.text += delta;
if (block.type === "thinking") block.thinking += delta;
if (delta)
events.push({
type: type === "text" ? "text_delta" : "thinking_delta",
contentIndex: index,
delta,
partial: output,
});
}
function close(key: string) {
if (ended.has(key)) return;
const index = blocks.get(key);
if (index === undefined) return;
const block = output.content[index];
if (block.type === "text")
events.push({ type: "text_end", contentIndex: index, content: block.text, partial: output });
if (block.type === "thinking")
events.push({ type: "thinking_end", contentIndex: index, content: block.thinking, partial: output });
ended.add(key);
}
function completeThinking(key: string, text: string) {
const index = blocks.get(key);
const block = index === undefined ? undefined : output.content[index];
const current = block?.type === "thinking" ? block.thinking : "";
if (text.startsWith(current)) append(key, "thinking", text.slice(current.length));
else if (block?.type === "thinking") block.thinking = text;
}
function updateThinkingPart(itemId: string, index: number, text: string, delta = false) {
const parts = thinkingParts.get(itemId) ?? [];
parts[index] = delta ? (parts[index] ?? "") + text : text;
thinkingParts.set(itemId, parts);
completeThinking(itemId, parts.filter(Boolean).join("\n\n"));
}
function completeText(key: string, text: string) {
const index = blocks.get(key);
const block = index === undefined ? undefined : output.content[index];
const current = block?.type === "text" ? block.text : "";
if (current !== text) ended.delete(key);
if (text.startsWith(current)) append(key, "text", text.slice(current.length));
else if (block?.type === "text") {
block.text = text;
events.generated();
}
}
function itemAdded(item: ResponseOutputItem) {
if (item.type === "reasoning") {
if (!blocks.has(item.id)) append(item.id, "thinking", "");
return;
}
if (item.type !== "function_call") return;
if (!item.id) throw new Error("Incomplete tool call item ID");
if (blocks.has(item.id)) return;
const call: ToolCall = {
type: "toolCall",
id: `${item.call_id}|${item.id}`,
name: item.name,
arguments: {},
...(item.namespace ? { namespace: item.namespace } : {}),
};
blocks.set(item.id, output.content.length);
args.set(item.id, item.arguments ?? "");
output.content.push(call);
events.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
}
function itemDone(item: ResponseOutputItem) {
if (item.type === "function_call") {
if (!item.id) throw new Error("Incomplete tool call item ID");
itemAdded(item);
if (ended.has(item.id)) return;
const index = blocks.get(item.id);
if (index === undefined) throw new Error("Incomplete tool call");
const block = output.content[index];
if (item.arguments && item.arguments !== args.get(item.id)) events.generated();
try {
const argumentsValue: unknown = JSON.parse(item.arguments || args.get(item.id) || "{}");
if (block.type !== "toolCall" || !item.call_id || !item.name || !isJsonObject(argumentsValue))
throw new Error("Incomplete tool call");
block.arguments = argumentsValue;
} catch (error) {
// Item completion precedes response.incomplete; wait for the terminal reason before failing.
invalidToolInputs.set(item.id, error);
return;
}
invalidToolInputs.delete(item.id);
complete.add(block);
ended.add(item.id);
events.push({ type: "toolcall_end", contentIndex: index, toolCall: block, partial: output });
} else if (item.type === "reasoning") {
itemAdded(item);
const parts = item.summary.length ? item.summary : (item.content ?? []);
if (parts.length)
completeThinking(
item.id,
parts
.map((part) => part.text)
.filter(Boolean)
.join("\n\n"),
);
const first = blocks.get(item.id);
if (first !== undefined)
(output.content[first] as ThinkingContent).thinkingSignature = JSON.stringify(item);
close(item.id);
} else if (item.type === "message") {
item.content.forEach((part, index) => {
const key = contentKey(item.id, index);
const text = part.type === "output_text" ? part.text : part.refusal;
completeText(key, text);
const blockIndex = blocks.get(key);
if (blockIndex !== undefined && output.content[blockIndex].type === "text") {
output.content[blockIndex].textSignature = item.id;
if (item.phase) output.content[blockIndex].phase = item.phase;
}
close(key);
});
}
}
let terminal = false;
for await (const event of chunks) {
events.received();
if (options.signal?.aborted) throw new Error("Request aborted");
switch (event.type) {
case "response.created":
output.responseId = event.response.id;
output.responseModel = event.response.model;
break;
case "response.output_item.added":
itemAdded(event.item);
break;
case "response.output_text.delta":
case "response.refusal.delta":
append(contentKey(event.item_id, event.content_index), "text", event.delta);
break;
case "response.output_text.done":
completeText(contentKey(event.item_id, event.content_index), event.text);
break;
case "response.refusal.done":
completeText(contentKey(event.item_id, event.content_index), event.refusal);
break;
case "response.reasoning_summary_text.delta":
updateThinkingPart(event.item_id, event.summary_index, event.delta, true);
break;
case "response.reasoning_summary_text.done":
updateThinkingPart(event.item_id, event.summary_index, event.text);
break;
case "response.reasoning_summary_part.done":
updateThinkingPart(event.item_id, event.summary_index, event.part.text);
break;
case "response.reasoning_text.delta":
updateThinkingPart(event.item_id, event.content_index, event.delta, true);
break;
case "response.reasoning_text.done":
updateThinkingPart(event.item_id, event.content_index, event.text);
break;
case "response.function_call_arguments.delta": {
args.set(event.item_id, (args.get(event.item_id) ?? "") + event.delta);
const index = blocks.get(event.item_id);
if (index !== undefined)
events.push({ type: "toolcall_delta", contentIndex: index, delta: event.delta, partial: output });
break;
}
case "response.output_item.done":
itemDone(event.item);
break;
case "error":
throw Object.assign(new Error(`OpenAI Responses: ${event.code ?? "error"}: ${event.message}`), {
code: event.code,
});
case "response.failed":
throw Object.assign(
new Error(
`OpenAI Responses: ${event.response.error?.code ?? "failed"}: ${event.response.error?.message ?? "Request failed"}`,
),
{ code: event.response.error?.code },
);
case "response.incomplete":
case "response.completed": {
if (
event.type === "response.incomplete" &&
event.response.incomplete_details?.reason !== "max_output_tokens"
)
throw new Error(
`OpenAI Responses incomplete: ${event.response.incomplete_details?.reason ?? "unknown"}`,
);
for (const item of event.response.output) itemDone(item);
const usage = event.response.usage;
if (usage) {
const cached = usage.input_tokens_details?.cached_tokens ?? 0;
const written = usage.input_tokens_details?.cache_write_tokens ?? 0;
if (
[usage.input_tokens, usage.output_tokens, usage.total_tokens, cached, written].every(
(n) => Number.isSafeInteger(n) && n >= 0,
) &&
cached + written <= usage.input_tokens
) {
output.usageAvailable = true;
output.cacheUsageAvailable = usage.input_tokens_details?.cached_tokens !== undefined;
output.usage.input = usage.input_tokens - cached - written;
output.usage.cacheRead = cached;
output.usage.cacheWrite = written;
output.usage.output = usage.output_tokens;
output.usage.totalTokens = usage.total_tokens;
output.usage.reasoning = usage.output_tokens_details?.reasoning_tokens;
}
events.recordUsage(usage.output_tokens);
}
output.responseId = event.response.id;
output.responseModel = event.response.model;
output.rawStopReason = event.response.status;
output.stopReason =
event.type === "response.incomplete" ? "length" : complete.size ? "toolUse" : "stop";
terminal = true;
break;
}
}
if (terminal) break;
}
if (options.signal?.aborted) throw new Error("Request aborted");
if (!terminal) throw new Error("Stream ended without response.completed");
if (invalidToolInputs.size && output.stopReason !== "length")
throw invalidToolInputs.values().next().value;
for (const key of blocks.keys()) close(key);
output.content = output.content.filter((block) => block.type !== "toolCall" || complete.has(block));
if (
!options.responsesWarmup &&
output.stopReason === "stop" &&
output.content.length === 0 &&
output.usage.totalTokens === 0
) {
throw Object.assign(new Error("Model returned an empty response without text, tool calls or usage."), {
code: "EMPTY_MODEL_RESPONSE",
});
}
calculateCost(model, output.usage);
events.push({
type: "done",
reason: output.stopReason as "stop" | "toolUse" | "length",
message: output,
});
}
}