Skip to content

Repository navigation

WP1: 出口序列化完整映射 #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ggbdpq merged 1 commit into from
Oct 20, 2026
Merged

WP1: 出口序列化完整映射 #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 90 additions & 34 deletions apps/gateway/src/protocols/anthropic/messages.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,42 +4,47 @@ import type { ProtocolHandlerArgs } from "../../server/http.ts";
import { beginSse, writeAnthropicSse } from "../../server/sse.ts";
import { anthropicToCanonical } from "./request.ts";
import {
STOP_REASON,
TEXT_BLOCK,
THINKING_BLOCK,
contentBlockInputJsonDelta,
contentBlockStart,
contentBlockStop,
contentBlockTextDelta,
contentBlockThinkingDelta,
messageDelta,
messageStart,
messageStop,
streamError,
} from "./stream.ts";
import type { AggregateResult } from "../../core/aggregate.ts";
import type { UsageEvent } from "../../core/events.ts";

// Provider usage 的最小形状（AggregateUsage / UsageEvent 的公共部分）。
export interface AnthropicUsageSummary {
inputTokens?: number;
outputTokens?: number;
}

export function buildAnthropicMessage({
id,
model,
text,
usage,
}: {
id: string;
model: string;
text: string;
usage?: AnthropicUsageSummary;
}) {
export function buildAnthropicMessage(id: string, model: string, result: AggregateResult) {
const content: unknown[] = [];
if (result.thinking) content.push({ ...THINKING_BLOCK, thinking: result.thinking });
if (result.text || !result.toolCalls.length) content.push({ type: "text", text: result.text });
for (const call of result.toolCalls) {
// 参数不是合法 JSON 属于上游协议错误，交给 server 的错误映射，不静默吞掉。
content.push({
type: "tool_use",
id: call.id,
name: call.name,
input: call.arguments ? JSON.parse(call.arguments) : {},
});
}
return {
id,
type: "message",
role: "assistant",
model,
content: [{ type: "text", text }],
stop_reason: "end_turn",
content,
stop_reason: STOP_REASON[result.stopReason],
stop_sequence: null,
usage: { input_tokens: usage?.inputTokens ?? 0, output_tokens: usage?.outputTokens ?? 0 },
usage: {
input_tokens: result.usage.inputTokens ?? 0,
output_tokens: result.usage.outputTokens ?? 0,
},
};
}

Expand All @@ -59,27 +64,78 @@ export async function handleAnthropicMessages({
if (body.stream) {
beginSse(res);
writeAnthropicSse(res, "message_start", messageStart(id, model));
writeAnthropicSse(res, "content_block_start", contentBlockStart(0));
let usage: UsageEvent | undefined;
let stopReason: string = STOP_REASON.stop;
// 同一时刻只有一个打开的 content block，换类型时先 stop 上一个；index 按出现顺序编号。
let index = -1;
let current: string | null = null;
const closeBlock = () => {
if (current === null) return;
writeAnthropicSse(res, "content_block_stop", contentBlockStop(index));
current = null;
};
const openBlock = (key: string, block: object) => {
if (current === key) return;
closeBlock();
current = key;
writeAnthropicSse(res, "content_block_start", contentBlockStart(++index, block));
};

for await (const event of run.stream(request)) {
if (event.type === "text.delta") {
writeAnthropicSse(res, "content_block_delta", contentBlockTextDelta(0, event.text));
} else if (event.type === "usage") {
usage = event;
} else if (event.type === "error") {
// 官方 Anthropic 流式语义：错误以 error 事件收尾，不再发 message_stop。
writeAnthropicSse(res, "error", streamError(event.error));
res.end();
return;
switch (event.type) {
case "text.delta":
openBlock("text", TEXT_BLOCK);
writeAnthropicSse(res, "content_block_delta", contentBlockTextDelta(index, event.text));
break;
case "thinking.delta":
openBlock("thinking", THINKING_BLOCK);
writeAnthropicSse(
res,
"content_block_delta",
contentBlockThinkingDelta(index, event.text),
);
break;
case "tool.start":
openBlock(`tool:${event.index}`, {
type: "tool_use",
id: event.id,
name: event.name,
input: {},
});
break;
case "tool.arguments.delta":
if (current === `tool:${event.index}`) {
writeAnthropicSse(
res,
"content_block_delta",
contentBlockInputJsonDelta(index, event.json),
);
}
break;
case "tool.end":
if (current === `tool:${event.index}`) closeBlock();
break;
case "message.end":
stopReason = STOP_REASON[event.stopReason];
break;
case "usage":
usage = event;
break;
case "error":
// 官方 Anthropic 流式语义：错误以 error 事件收尾，不再发 message_stop。
writeAnthropicSse(res, "error", streamError(event.error));
res.end();
return;
}
}
writeAnthropicSse(res, "content_block_stop", contentBlockStop(0));
writeAnthropicSse(res, "message_delta", messageDelta("end_turn", usage?.outputTokens ?? 0));
// 没有任何内容块时保持 v0.1 的空 text 块。
if (index === -1) openBlock("text", TEXT_BLOCK);
closeBlock();
writeAnthropicSse(res, "message_delta", messageDelta(stopReason, usage?.outputTokens ?? 0));
writeAnthropicSse(res, "message_stop", messageStop());
res.end();
return;
}

const result = await run.aggregate(request);
writeJson(res, 200, buildAnthropicMessage({ id, model, text: result.text, usage: result.usage }));
writeJson(res, 200, buildAnthropicMessage(id, model, await run.aggregate(request)));
}
28 changes: 26 additions & 2 deletions apps/gateway/src/protocols/anthropic/stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,38 @@ export function messageStart(id: string, model: string) {
};
}

export function contentBlockStart(index: number) {
return { type: "content_block_start", index, content_block: { type: "text", text: "" } };
export const TEXT_BLOCK = { type: "text", text: "" };
// 上游不提供 thinking signature，按 Anthropic 形状给空串。
export const THINKING_BLOCK = { type: "thinking", thinking: "", signature: "" };

// Canonical stopReason → Anthropic stop_reason。
export const STOP_REASON = {
stop: "end_turn",
tool_use: "tool_use",
length: "max_tokens",
cancelled: "end_turn",
} as const;

export function contentBlockStart(index: number, contentBlock: object = TEXT_BLOCK) {
return { type: "content_block_start", index, content_block: contentBlock };
}

export function contentBlockTextDelta(index: number, text: string) {
return { type: "content_block_delta", index, delta: { type: "text_delta", text } };
}

export function contentBlockThinkingDelta(index: number, thinking: string) {
return { type: "content_block_delta", index, delta: { type: "thinking_delta", thinking } };
}

export function contentBlockInputJsonDelta(index: number, partialJson: string) {
return {
type: "content_block_delta",
index,
delta: { type: "input_json_delta", partial_json: partialJson },
};
}

export function contentBlockStop(index: number) {
return { type: "content_block_stop", index };
}
Expand Down
100 changes: 63 additions & 37 deletions apps/gateway/src/protocols/openai/chat.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@ import { wireId, writeJson } from "../../server/http.ts";
import type { ProtocolHandlerArgs } from "../../server/http.ts";
import { beginSse, writeSseData, writeSseDone } from "../../server/sse.ts";
import { assertHasText, chatToCanonical } from "./request.ts";
import { chatChunk } from "./stream.ts";
import { FINISH_REASON, chatChunk } from "./stream.ts";
import type { ChatCompletionChunk, UsageSummary } from "./stream.ts";
import type { AggregateResult } from "../../core/aggregate.ts";
import type { UsageEvent } from "../../core/events.ts";

// Provider usage（inputTokens/outputTokens）→ OpenAI usage；无数据保持 0。
Expand All @@ -14,32 +15,26 @@ function toOpenAiUsage(usage?: UsageSummary) {
return { prompt_tokens: input, completion_tokens: output, total_tokens: input + output };
}

export function buildChatCompletion({
id,
model,
text,
finishReason = "stop",
usage,
}: {
id: string;
model: string;
text: string;
finishReason?: string;
usage?: UsageSummary;
}) {
export function buildChatCompletion(id: string, model: string, result: AggregateResult) {
const message: Record<string, unknown> = {
role: "assistant",
content: result.text || (result.toolCalls.length ? null : ""),
};
if (result.thinking) message.reasoning_content = result.thinking;
if (result.toolCalls.length) {
message.tool_calls = result.toolCalls.map((call) => ({
id: call.id,
type: "function",
function: { name: call.name, arguments: call.arguments },
}));
}
return {
id,
object: "chat.completion",
created: Math.floor(Date.now() / 1000),
model,
choices: [
{
index: 0,
message: { role: "assistant", content: text || "" },
finish_reason: finishReason,
},
],
usage: toOpenAiUsage(usage),
choices: [{ index: 0, message, finish_reason: FINISH_REASON[result.stopReason] }],
usage: toOpenAiUsage(result.usage),
};
}

Expand All @@ -57,22 +52,54 @@ export async function handleChatCompletions({
beginSse(res);
writeSseData(res, chatChunk(id, model, { role: "assistant" }));
let usage: UsageEvent | undefined;
let finishReason: string = FINISH_REASON.stop;
// Canonical tool index → 下游 tool_calls[].index，按 tool.start 出现顺序编号。
const toolIndex = new Map<number, number>();
for await (const event of run.stream(request)) {
if (event.type === "text.delta") {
writeSseData(res, chatChunk(id, model, { content: event.text }));
} else if (event.type === "usage") {
usage = event;
} else if (event.type === "error") {
// v0.1 在此会触发 headers-sent 崩溃并挂住连接；现按 OpenAI 惯例发 error 数据后干净收尾。
writeSseData(res, {
error: { message: event.error.message, type: "server_error", code: event.error.code },
});
writeSseDone(res);
res.end();
return;
switch (event.type) {
case "text.delta":
writeSseData(res, chatChunk(id, model, { content: event.text }));
break;
case "thinking.delta":
writeSseData(res, chatChunk(id, model, { reasoning_content: event.text }));
break;
case "tool.start": {
const index = toolIndex.size;
toolIndex.set(event.index, index);
const call = { name: event.name, arguments: "" };
writeSseData(
res,
chatChunk(id, model, {
tool_calls: [{ index, id: event.id, type: "function", function: call }],
}),
);
break;
}
case "tool.arguments.delta": {
const index = toolIndex.get(event.index) ?? event.index;
writeSseData(
res,
chatChunk(id, model, { tool_calls: [{ index, function: { arguments: event.json } }] }),
);
break;
}
case "message.end":
finishReason = FINISH_REASON[event.stopReason];
break;
case "usage":
usage = event;
break;
case "error":
// v0.1 在此会触发 headers-sent 崩溃并挂住连接；现按 OpenAI 惯例发 error 数据后干净收尾。
writeSseData(res, {
error: { message: event.error.message, type: "server_error", code: event.error.code },
});
writeSseDone(res);
res.end();
return;
}
}
writeSseData(res, chatChunk(id, model, {}, "stop"));
writeSseData(res, chatChunk(id, model, {}, finishReason));
if (usage) {
// OpenAI include_usage 约定：空 choices 的收尾 usage chunk。
const final: ChatCompletionChunk = chatChunk(id, model, {}, "stop");
Expand All @@ -85,6 +112,5 @@ export async function handleChatCompletions({
return;
}

const result = await run.aggregate(request);
writeJson(res, 200, buildChatCompletion({ id, model, text: result.text, usage: result.usage }));
writeJson(res, 200, buildChatCompletion(id, model, await run.aggregate(request)));
}
Loading
Loading