@@ -73,6 +73,8 @@ const DEFAULT_AZURE_OPENAI_API_VERSION = "preview";
7373const OPENAI_CODEX_RESPONSES_EMPTY_INPUT_TEXT = " ";
7474const GEMINI_THOUGHT_SIGNATURE_VALIDATOR_SKIP = "skip_thought_signature_validator";
7575const AZURE_RESPONSES_FIRST_EVENT_TIMEOUT_MS = 30_000;
76+const MODEL_STREAM_COOPERATIVE_YIELD_INTERVAL_MS = 12;
77+const MODEL_STREAM_COOPERATIVE_YIELD_MAX_EVENTS = 64;
7678const log = createSubsystemLogger("openai-transport");
77797880type ReplayableResponseOutputMessage = Omit<ResponseOutputMessage, "id"> & { id?: string };
@@ -92,6 +94,42 @@ type BaseStreamOptions = {
9294responseFormat?: Record<string, unknown>;
9395};
949697+type ModelStreamCooperativeScheduler = {
98+afterEvent: () => Promise<void>;
99+};
100+101+function throwIfModelStreamAborted(signal?: AbortSignal): void {
102+if (signal?.aborted) {
103+throw new Error("Request was aborted");
104+}
105+}
106+107+function createModelStreamCooperativeScheduler(
108+signal?: AbortSignal,
109+): ModelStreamCooperativeScheduler {
110+let lastYieldedAt = Date.now();
111+let eventsSinceYield = 0;
112+return {
113+async afterEvent() {
114+throwIfModelStreamAborted(signal);
115+eventsSinceYield += 1;
116+const now = Date.now();
117+if (
118+eventsSinceYield < MODEL_STREAM_COOPERATIVE_YIELD_MAX_EVENTS &&
119+now - lastYieldedAt < MODEL_STREAM_COOPERATIVE_YIELD_INTERVAL_MS
120+) {
121+return;
122+}
123+eventsSinceYield = 0;
124+lastYieldedAt = now;
125+await new Promise<void>((resolve) => {
126+setImmediate(resolve);
127+});
128+throwIfModelStreamAborted(signal);
129+},
130+};
131+}
132+95133type OpenAIResponsesOptions = BaseStreamOptions & {
96134reasoning?: OpenAIReasoningEffort;
97135reasoningEffort?: OpenAIReasoningEffort;
@@ -722,6 +760,7 @@ async function processResponsesStream(
722760serviceTier?: ResponseCreateParamsStreaming["service_tier"],
723761) => void;
724762firstEventTimeoutMs?: number;
763+signal?: AbortSignal;
725764},
726765) {
727766let currentItem: Record<string, unknown> | null = null;
@@ -736,7 +775,9 @@ async function processResponsesStream(
736775model,
737776options?.firstEventTimeoutMs,
738777);
778+const cooperativeScheduler = createModelStreamCooperativeScheduler(options?.signal);
739779for await (const rawEvent of guardedStream) {
780+throwIfModelStreamAborted(options?.signal);
740781const event = rawEvent as Record<string, unknown>;
741782const type = stringifyUnknown(event.type);
742783eventCount += 1;
@@ -933,6 +974,7 @@ async function processResponsesStream(
933974 : "Unknown error (no error details in response)";
934975throw new Error(msg);
935976}
977+await cooperativeScheduler.afterEvent();
936978}
937979const eventTypeSummary = [...eventTypes.entries()]
938980.slice(0, 12)
@@ -1141,6 +1183,7 @@ export function createOpenAIResponsesTransportStreamFn(): StreamFn {
11411183await processResponsesStream(responseStream, output, stream, model, {
11421184serviceTier: (options as OpenAIResponsesOptions | undefined)?.serviceTier,
11431185 applyServiceTierPricing,
1186+signal: options?.signal,
11441187});
11451188if (options?.signal?.aborted) {
11461189throw new Error("Request was aborted");
@@ -1538,6 +1581,7 @@ export function createAzureOpenAIResponsesTransportStreamFn(): StreamFn {
15381581stream.push({ type: "start", partial: output as never });
15391582await processResponsesStream(responseStream, output, stream, model, {
15401583firstEventTimeoutMs: AZURE_RESPONSES_FIRST_EVENT_TIMEOUT_MS,
1584+signal: options?.signal,
15411585});
15421586if (options?.signal?.aborted) {
15431587throw new Error("Request was aborted");
@@ -1738,7 +1782,9 @@ export function createOpenAICompletionsTransportStreamFn(): StreamFn {
17381782buildOpenAISdkRequestOptions(model, options?.signal),
17391783)) as unknown as AsyncIterable<ChatCompletionChunk>;
17401784stream.push({ type: "start", partial: output as never });
1741-await processOpenAICompletionsStream(responseStream, output, model, stream);
1785+await processOpenAICompletionsStream(responseStream, output, model, stream, {
1786+signal: options?.signal,
1787+});
17421788if (options?.signal?.aborted) {
17431789throw new Error("Request was aborted");
17441790}
@@ -1760,6 +1806,7 @@ async function processOpenAICompletionsStream(
17601806output: MutableAssistantOutput,
17611807model: Model<Api>,
17621808stream: { push(event: unknown): void },
1809+options?: { signal?: AbortSignal },
17631810) {
17641811const MAX_POST_TOOL_CALL_BUFFER_BYTES = 256_000;
17651812const MAX_TOOL_CALL_ARGUMENT_BUFFER_BYTES = 256_000;
@@ -1907,8 +1954,11 @@ async function processOpenAICompletionsStream(
19071954appendVisibleTextDelta(part);
19081955}
19091956};
1957+const cooperativeScheduler = createModelStreamCooperativeScheduler(options?.signal);
19101958for await (const rawChunk of responseStream as AsyncIterable<unknown>) {
1959+throwIfModelStreamAborted(options?.signal);
19111960if (!rawChunk || typeof rawChunk !== "object") {
1961+await cooperativeScheduler.afterEvent();
19121962continue;
19131963}
19141964const chunk = rawChunk as ChatCompletionChunk;
@@ -1918,6 +1968,7 @@ async function processOpenAICompletionsStream(
19181968}
19191969const choice = Array.isArray(chunk.choices) ? chunk.choices[0] : undefined;
19201970if (!choice) {
1971+await cooperativeScheduler.afterEvent();
19211972continue;
19221973}
19231974const choiceUsage = (choice as unknown as { usage?: ChatCompletionChunk["usage"] }).usage;
@@ -1935,6 +1986,7 @@ async function processOpenAICompletionsStream(
19351986choice.delta ??
19361987(choice as unknown as { message?: ChatCompletionChunk["choices"][number]["delta"] }).message;
19371988if (!choiceDelta) {
1989+await cooperativeScheduler.afterEvent();
19381990continue;
19391991}
19401992if (choiceDelta.content) {
@@ -2026,6 +2078,7 @@ async function processOpenAICompletionsStream(
20262078}
20272079}
20282080flushPendingPostToolCallDeltas();
2081+await cooperativeScheduler.afterEvent();
20292082}
20302083flushDeepSeekTextFilterAtEnd();
20312084finishCurrentBlock();
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。