mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-26 20:35:39 -06:00
fix(azure): support Responses text stream events
Co-authored-by: thomas.krohnfuss <thomas.krohnfuss@hsu.hamburg>
This commit is contained in:
Vincent Koc
@@ -25,7 +25,12 @@ import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "./system-prompt-cache-boundary.js"
|
||||
type OpenAICompletionsOutput = Parameters<typeof testing.processOpenAICompletionsStream>[1];
|
||||
type OpenAIResponsesOutput = Parameters<typeof testing.processResponsesStream>[1];
|
||||
|
||||
type CapturedStreamEvent = { type?: string; delta?: string; partial?: unknown };
|
||||
type CapturedStreamEvent = {
|
||||
type?: string;
|
||||
delta?: string;
|
||||
content?: string;
|
||||
partial?: unknown;
|
||||
};
|
||||
|
||||
function createDeepSeekCompletionsModel(): Model<"openai-completions"> {
|
||||
return {
|
||||
@@ -1967,6 +1972,68 @@ describe("openai transport stream", () => {
|
||||
expect(output.content).toEqual([{ type: "text", text: "ab" }]);
|
||||
});
|
||||
|
||||
it("handles Azure Responses text content and text delta events", async () => {
|
||||
const model = createAzureResponsesModel();
|
||||
const output = createResponsesAssistantOutput(model);
|
||||
const events: CapturedStreamEvent[] = [];
|
||||
|
||||
await testing.processResponsesStream(
|
||||
streamChunks([
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_text",
|
||||
content: [],
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{ type: "response.text.delta", delta: "Hello" },
|
||||
{ type: "response.text.delta", delta: " from Azure!" },
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_text",
|
||||
content: [{ type: "text", text: "Hello from Azure!" }],
|
||||
status: "completed",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.completed",
|
||||
response: {
|
||||
id: "resp_azure_text",
|
||||
status: "completed",
|
||||
usage: {
|
||||
input_tokens: 4,
|
||||
output_tokens: 3,
|
||||
total_tokens: 7,
|
||||
},
|
||||
},
|
||||
},
|
||||
]),
|
||||
output,
|
||||
{ push: (event) => events.push(event as CapturedStreamEvent) },
|
||||
model,
|
||||
);
|
||||
|
||||
expect(events).toMatchObject([
|
||||
{ type: "text_start" },
|
||||
{ type: "text_delta", delta: "Hello" },
|
||||
{ type: "text_delta", delta: " from Azure!" },
|
||||
{ type: "text_end", content: "Hello from Azure!" },
|
||||
]);
|
||||
expect(output.content).toMatchObject([{ type: "text", text: "Hello from Azure!" }]);
|
||||
expectRecordFields(output.usage, {
|
||||
input: 4,
|
||||
output: 3,
|
||||
totalTokens: 7,
|
||||
});
|
||||
expect(output.responseId).toBe("resp_azure_text");
|
||||
});
|
||||
|
||||
it("skips null and non-object OpenAI-compatible stream chunks", async () => {
|
||||
const model = {
|
||||
id: "glm-5",
|
||||
|
||||
@@ -1571,7 +1571,11 @@ async function processResponsesStream(
|
||||
partial: output,
|
||||
});
|
||||
}
|
||||
} else if (type === "response.output_text.delta" || type === "response.refusal.delta") {
|
||||
} else if (
|
||||
type === "response.output_text.delta" ||
|
||||
type === "response.text.delta" ||
|
||||
type === "response.refusal.delta"
|
||||
) {
|
||||
if (currentItem?.type === "message" && currentBlock?.type === "text") {
|
||||
currentBlock.text = `${stringifyUnknown(currentBlock.text)}${stringifyUnknown(event.delta)}`;
|
||||
stream.push({
|
||||
@@ -1623,7 +1627,7 @@ async function processResponsesStream(
|
||||
currentBlock.text = content
|
||||
.map((part) => {
|
||||
const contentPart = part as { type?: string; text?: string; refusal?: string };
|
||||
return contentPart.type === "output_text"
|
||||
return contentPart.type === "output_text" || contentPart.type === "text"
|
||||
? (contentPart.text ?? "")
|
||||
: (contentPart.refusal ?? "");
|
||||
})
|
||||
|
||||
@@ -1,13 +1,40 @@
|
||||
// OpenAI Responses shared tests cover tool conversion and response item mapping.
|
||||
import type { Tool as OpenAIResponsesTool } from "openai/resources/responses/responses.js";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { AssistantMessage, Context, Model, Tool } from "../types.js";
|
||||
import type { AssistantMessage, AssistantMessageEvent, Context, Model, Tool } from "../types.js";
|
||||
import { AssistantMessageEventStream } from "../utils/event-stream.js";
|
||||
import { convertResponsesMessages, processResponsesStream } from "./openai-responses-shared.js";
|
||||
import {
|
||||
createResponsesAssistantOutput,
|
||||
convertResponsesMessages,
|
||||
type OpenAIResponsesStreamEvent,
|
||||
processResponsesStream,
|
||||
} from "./openai-responses-shared.js";
|
||||
import { convertResponsesTools } from "./openai-responses-tools.js";
|
||||
|
||||
type ResponsesFunctionTool = Extract<OpenAIResponsesTool, { type: "function" }>;
|
||||
|
||||
async function* streamResponsesEvents(
|
||||
events: readonly OpenAIResponsesStreamEvent[],
|
||||
): AsyncGenerator<OpenAIResponsesStreamEvent> {
|
||||
for (const event of events) {
|
||||
yield event;
|
||||
}
|
||||
}
|
||||
|
||||
function createCapturedAssistantMessageEventStream(): {
|
||||
stream: AssistantMessageEventStream;
|
||||
events: AssistantMessageEvent[];
|
||||
} {
|
||||
const stream = new AssistantMessageEventStream();
|
||||
const events: AssistantMessageEvent[] = [];
|
||||
const push = stream.push.bind(stream);
|
||||
stream.push = (event) => {
|
||||
events.push(event);
|
||||
push(event);
|
||||
};
|
||||
return { stream, events };
|
||||
}
|
||||
|
||||
function expectResponsesFunctionTool(tool: OpenAIResponsesTool | undefined): ResponsesFunctionTool {
|
||||
expect(tool).toHaveProperty("type", "function");
|
||||
return tool as ResponsesFunctionTool;
|
||||
@@ -481,3 +508,284 @@ describe("processResponsesStream", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Azure OpenAI Responses content type support", () => {
|
||||
const azureModel = {
|
||||
id: "gpt-5.5",
|
||||
name: "GPT-5.5 (Azure)",
|
||||
api: "azure-openai-responses",
|
||||
provider: "azure",
|
||||
baseUrl: "https://test.openai.azure.com/openai/v1",
|
||||
reasoning: true,
|
||||
input: ["text"],
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
||||
contextWindow: 200000,
|
||||
maxTokens: 8192,
|
||||
} satisfies Model<"azure-openai-responses">;
|
||||
|
||||
it("supports Azure 'text' content type in addition to 'output_text'", () => {
|
||||
const input = convertResponsesMessages(
|
||||
azureModel,
|
||||
{
|
||||
systemPrompt: "system",
|
||||
messages: [
|
||||
{
|
||||
role: "assistant",
|
||||
api: azureModel.api,
|
||||
provider: azureModel.provider,
|
||||
model: azureModel.id,
|
||||
usage: {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
},
|
||||
stopReason: "stop",
|
||||
timestamp: 1,
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: "Azure response with text content type",
|
||||
textSignature: JSON.stringify({
|
||||
v: 1,
|
||||
id: "msg_azure_text",
|
||||
}),
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
} satisfies Context,
|
||||
new Set(["azure", "azure-openai-responses"]),
|
||||
{ includeSystemPrompt: false },
|
||||
);
|
||||
|
||||
const assistantMessage = input.find(
|
||||
(item) => item && typeof item === "object" && "role" in item && item.role === "assistant",
|
||||
);
|
||||
|
||||
expect(assistantMessage).toMatchObject({
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: "output_text",
|
||||
text: "Azure response with text content type",
|
||||
annotations: [],
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it("processResponsesStream handles Azure 'text' content type streaming events", async () => {
|
||||
const azureEvents: OpenAIResponsesStreamEvent[] = [
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
sequence_number: 1,
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_1",
|
||||
content: [],
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.content_part.added",
|
||||
content_index: 0,
|
||||
item_id: "msg_azure_1",
|
||||
output_index: 0,
|
||||
sequence_number: 2,
|
||||
part: {
|
||||
type: "text",
|
||||
text: "",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.text.delta",
|
||||
delta: "Hello",
|
||||
},
|
||||
{
|
||||
type: "response.text.delta",
|
||||
delta: " from",
|
||||
},
|
||||
{
|
||||
type: "response.text.delta",
|
||||
delta: " Azure!",
|
||||
},
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
sequence_number: 6,
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_1",
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: "Hello from Azure!",
|
||||
},
|
||||
],
|
||||
status: "completed",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.completed",
|
||||
sequence_number: 7,
|
||||
response: {
|
||||
id: "resp_azure_123",
|
||||
created_at: 1,
|
||||
output_text: "Hello from Azure!",
|
||||
error: null,
|
||||
incomplete_details: null,
|
||||
instructions: null,
|
||||
metadata: null,
|
||||
model: azureModel.id,
|
||||
object: "response",
|
||||
output: [],
|
||||
parallel_tool_calls: false,
|
||||
temperature: null,
|
||||
tool_choice: "auto",
|
||||
tools: [],
|
||||
top_p: null,
|
||||
status: "completed",
|
||||
usage: {
|
||||
input_tokens: 10,
|
||||
input_tokens_details: { cached_tokens: 0 },
|
||||
output_tokens: 5,
|
||||
output_tokens_details: { reasoning_tokens: 0 },
|
||||
total_tokens: 15,
|
||||
},
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
const { stream, events } = createCapturedAssistantMessageEventStream();
|
||||
const output = createResponsesAssistantOutput(azureModel, "azure-openai-responses");
|
||||
await processResponsesStream(streamResponsesEvents(azureEvents), output, stream, azureModel);
|
||||
|
||||
expect(
|
||||
events.map((event) =>
|
||||
event.type === "text_delta"
|
||||
? { type: event.type, delta: event.delta }
|
||||
: event.type === "text_end"
|
||||
? { type: event.type, content: event.content }
|
||||
: { type: event.type },
|
||||
),
|
||||
).toEqual([
|
||||
{ type: "text_start" },
|
||||
{ type: "text_delta", delta: "Hello" },
|
||||
{ type: "text_delta", delta: " from" },
|
||||
{ type: "text_delta", delta: " Azure!" },
|
||||
{ type: "text_end", content: "Hello from Azure!" },
|
||||
]);
|
||||
|
||||
expect(output.content).toHaveLength(1);
|
||||
expect(output.content[0]).toMatchObject({
|
||||
type: "text",
|
||||
text: "Hello from Azure!",
|
||||
});
|
||||
|
||||
expect(output.usage).toMatchObject({
|
||||
input: 10,
|
||||
output: 5,
|
||||
totalTokens: 15,
|
||||
});
|
||||
|
||||
expect(output.stopReason).toBe("stop");
|
||||
});
|
||||
|
||||
it("processResponsesStream handles Azure text deltas without a content_part.added event", async () => {
|
||||
const azureEvents: OpenAIResponsesStreamEvent[] = [
|
||||
{
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
sequence_number: 1,
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_without_part",
|
||||
content: [],
|
||||
status: "in_progress",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.text.delta",
|
||||
delta: "No explicit",
|
||||
},
|
||||
{
|
||||
type: "response.text.delta",
|
||||
delta: " part",
|
||||
},
|
||||
{
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
sequence_number: 4,
|
||||
item: {
|
||||
type: "message",
|
||||
role: "assistant",
|
||||
id: "msg_azure_without_part",
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: "No explicit part",
|
||||
},
|
||||
],
|
||||
status: "completed",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "response.completed",
|
||||
sequence_number: 5,
|
||||
response: {
|
||||
id: "resp_azure_without_part",
|
||||
created_at: 1,
|
||||
output_text: "No explicit part",
|
||||
error: null,
|
||||
incomplete_details: null,
|
||||
instructions: null,
|
||||
metadata: null,
|
||||
model: azureModel.id,
|
||||
object: "response",
|
||||
output: [],
|
||||
parallel_tool_calls: false,
|
||||
temperature: null,
|
||||
tool_choice: "auto",
|
||||
tools: [],
|
||||
top_p: null,
|
||||
status: "completed",
|
||||
usage: {
|
||||
input_tokens: 3,
|
||||
input_tokens_details: { cached_tokens: 0 },
|
||||
output_tokens: 3,
|
||||
output_tokens_details: { reasoning_tokens: 0 },
|
||||
total_tokens: 6,
|
||||
},
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
const { stream, events } = createCapturedAssistantMessageEventStream();
|
||||
const output = createResponsesAssistantOutput(azureModel, "azure-openai-responses");
|
||||
|
||||
await processResponsesStream(streamResponsesEvents(azureEvents), output, stream, azureModel);
|
||||
|
||||
expect(
|
||||
events.map((event) =>
|
||||
event.type === "text_delta"
|
||||
? event.delta
|
||||
: event.type === "text_end"
|
||||
? `[END:${event.content}]`
|
||||
: event.type,
|
||||
),
|
||||
).toEqual(["text_start", "No explicit", " part", "[END:No explicit part]"]);
|
||||
|
||||
expect(output.content[0]).toMatchObject({
|
||||
type: "text",
|
||||
text: "No explicit part",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -43,6 +43,33 @@ import { transformMessages } from "./transform-messages.js";
|
||||
|
||||
type ReplayableResponseOutputMessage = Omit<ResponseOutputMessage, "id"> & { id?: string };
|
||||
type ReplayableResponseReasoningItem = Omit<ResponseReasoningItem, "id"> & { id?: string };
|
||||
type AzureResponsesTextContentPart = { type: "text"; text: string };
|
||||
type ResponsesTextContentPart =
|
||||
| ResponseOutputMessage["content"][number]
|
||||
| AzureResponsesTextContentPart;
|
||||
type ResponsesStreamOutputMessage = Omit<ResponseOutputMessage, "content"> & {
|
||||
content: ResponsesTextContentPart[];
|
||||
};
|
||||
type ResponsesContentPartAddedEvent = Extract<
|
||||
ResponseStreamEvent,
|
||||
{ type: "response.content_part.added" }
|
||||
>;
|
||||
type ResponsesOutputItemDoneEvent = Extract<
|
||||
ResponseStreamEvent,
|
||||
{ type: "response.output_item.done" }
|
||||
>;
|
||||
type AzureResponsesContentPartAddedEvent = Omit<ResponsesContentPartAddedEvent, "part"> & {
|
||||
part: AzureResponsesTextContentPart;
|
||||
};
|
||||
type AzureResponsesOutputItemDoneEvent = Omit<ResponsesOutputItemDoneEvent, "item"> & {
|
||||
item: ResponsesStreamOutputMessage;
|
||||
};
|
||||
|
||||
export type OpenAIResponsesStreamEvent =
|
||||
| ResponseStreamEvent
|
||||
| AzureResponsesContentPartAddedEvent
|
||||
| AzureResponsesOutputItemDoneEvent
|
||||
| { type: "response.text.delta"; delta: string };
|
||||
|
||||
function normalizeResponsesReasoningReplayItem(params: {
|
||||
item: ReplayableResponseReasoningItem;
|
||||
@@ -526,14 +553,17 @@ export async function runResponsesStreamLifecycle<TApi extends Api>(params: {
|
||||
// =============================================================================
|
||||
|
||||
export async function processResponsesStream<TApi extends Api>(
|
||||
openaiStream: AsyncIterable<ResponseStreamEvent>,
|
||||
openaiStream: AsyncIterable<OpenAIResponsesStreamEvent>,
|
||||
output: AssistantMessage,
|
||||
stream: AssistantMessageEventStream,
|
||||
model: Model<TApi>,
|
||||
options?: OpenAIResponsesStreamOptions,
|
||||
): Promise<void> {
|
||||
let currentItem: ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | null =
|
||||
null;
|
||||
let currentItem:
|
||||
| ResponseReasoningItem
|
||||
| ResponsesStreamOutputMessage
|
||||
| ResponseFunctionToolCall
|
||||
| null = null;
|
||||
let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null =
|
||||
null;
|
||||
const blocks = output.content;
|
||||
@@ -614,8 +644,13 @@ export async function processResponsesStream<TApi extends Api>(
|
||||
} else if (event.type === "response.content_part.added") {
|
||||
if (currentItem?.type === "message") {
|
||||
currentItem.content = currentItem.content || [];
|
||||
// Filter out ReasoningText, only accept output_text and refusal
|
||||
if (event.part.type === "output_text" || event.part.type === "refusal") {
|
||||
// Accept output_text, text (Azure), and refusal content parts
|
||||
// Azure OpenAI Responses may return "text" instead of "output_text"
|
||||
if (
|
||||
event.part.type === "output_text" ||
|
||||
event.part.type === "text" ||
|
||||
event.part.type === "refusal"
|
||||
) {
|
||||
currentItem.content.push(event.part);
|
||||
}
|
||||
}
|
||||
@@ -636,6 +671,24 @@ export async function processResponsesStream<TApi extends Api>(
|
||||
});
|
||||
}
|
||||
}
|
||||
} else if (event.type === "response.text.delta") {
|
||||
// Azure OpenAI Responses may emit "text" events instead of "output_text"
|
||||
if (currentItem?.type === "message" && currentBlock?.type === "text") {
|
||||
currentItem.content = currentItem.content || [];
|
||||
let lastPart = currentItem.content[currentItem.content.length - 1];
|
||||
if (lastPart?.type !== "text") {
|
||||
lastPart = { type: "text", text: "" };
|
||||
currentItem.content.push(lastPart);
|
||||
}
|
||||
currentBlock.text += event.delta;
|
||||
lastPart.text += event.delta;
|
||||
stream.push({
|
||||
type: "text_delta",
|
||||
contentIndex: blockIndex(),
|
||||
delta: event.delta,
|
||||
partial: output,
|
||||
});
|
||||
}
|
||||
} else if (event.type === "response.refusal.delta") {
|
||||
if (currentItem?.type === "message" && currentBlock?.type === "text") {
|
||||
if (!currentItem.content || currentItem.content.length === 0) {
|
||||
@@ -705,8 +758,9 @@ export async function processResponsesStream<TApi extends Api>(
|
||||
});
|
||||
currentBlock = null;
|
||||
} else if (item.type === "message" && currentBlock?.type === "text") {
|
||||
// Support both OpenAI "output_text" and Azure "text" content types
|
||||
currentBlock.text = item.content
|
||||
.map((c) => (c.type === "output_text" ? c.text : c.refusal))
|
||||
.map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal))
|
||||
.join("");
|
||||
currentBlock.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined);
|
||||
stream.push({
|
||||
|
||||
Reference in New Issue
Block a user