Skip to content

Commit e4b6ab0

Browse files
committed
feat: 排队重开、非流式保活与兼容层修补
- 排队信号(Qoder/Trae 的 isQueued / RESOURCE_LIMIT):不冷却凭证、不换号, 等 retryAfterSeconds 后在原号重开;轮次与单轮等待都有上限,已出帧不重开 - 非流式出口保活(NONSTREAM_KEEPALIVE_SECONDS,默认关):宽限期内完成仍是 逐字节相同的 JSONResponse,超时才转 chunked 先吐空白心跳再补正文 - Anthropic 心跳改具名 event: ping;OpenAI 流式工具调用缺 id 时按 index 补 稳定 id(按 id 聚合的客户端不再丢分片) - trae 上下文窗口取 context_window_tokens.max,1M 档不再报 200k - 同步 PROPOSAL / README / TECHNICAL
1 parent 6e6fa30 commit e4b6ab0

31 files changed

Lines changed: 3082 additions & 1414 deletions

‎PROPOSAL.md‎

Lines changed: 131 additions & 154 deletions
Large diffs are not rendered by default.

‎README.en.md‎

Lines changed: 483 additions & 358 deletions
Large diffs are not rendered by default.

‎README.md‎

Lines changed: 396 additions & 486 deletions
Large diffs are not rendered by default.

‎TECHNICAL.md‎

Lines changed: 451 additions & 342 deletions
Large diffs are not rendered by default.

‎docker-compose.yml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,8 @@ services:
159159
STREAM_IDLE_TIMEOUT_SECONDS: ${STREAM_IDLE_TIMEOUT_SECONDS:-0}
160160
# 短冷却「等待续上」(秒):全池都在短冷却时最多等这么久再取一次;0 关闭
161161
SHORT_COOLDOWN_WAIT_SECONDS: ${SHORT_COOLDOWN_WAIT_SECONDS:-0}
162+
# 非流式出口保活(秒):N 秒内没聚合完就转 chunked 吐空白心跳;0 关闭
163+
NONSTREAM_KEEPALIVE_SECONDS: ${NONSTREAM_KEEPALIVE_SECONDS:-0}
162164
# 单账号在途请求上限(跨渠道全局);0 = 不限
163165
MAX_INFLIGHT_PER_CREDENTIAL: ${MAX_INFLIGHT_PER_CREDENTIAL:-0}
164166
# Prometheus 指标出口(GET /metrics,无鉴权,与 /healthz 同口径);false 关闭

‎src/api/endpoints.py‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,11 @@
1919
from ..config import Settings
2020
from ..engine.compress import estimate_messages
2121
from ..engine.executor import Executor, StreamSink
22+
from ..engine.sse import SSE_COMMENT
2223
from ..key_quota import Lease, QuotaExceeded
2324
from ..runtime_settings import RuntimeSettings
2425
from .deps import ApiKeyPrincipal, get_services, read_json_body
25-
from .streaming import with_keepalive
26+
from .streaming import respond_with_keepalive, with_keepalive
2627

2728

2829
def ensure_model_allowed(chat_request: ChatRequest, principal: ApiKeyPrincipal,
@@ -98,12 +99,26 @@ async def dispatch_compat(
9899
lease.release() # 400 不该占着并发名额
99100
raise
100101
translator = translator_for(target.model) if translator_for else None
102+
# 心跳帧随出口协议走:Anthropic 要 `event: ping` 具名事件(只解析
103+
# `event:` 行的客户端会忽略注释帧),chat / Responses 保持注释帧
104+
# (Responses 无 ping 事件,未知事件类型会让严格解析器报错)。
105+
keepalive_frame = (translator.keepalive() if translator is not None
106+
else SSE_COMMENT)
101107
frames = with_keepalive(executor.stream_guarded(chat_request,
102108
username=principal.username,
103109
translator=translator,
104-
provider_binding=binding))
110+
provider_binding=binding),
111+
keepalive_frame=keepalive_frame)
105112
return StreamingResponse(
106113
_release_when_done(frames, lease), media_type="text/event-stream")
114+
# 非流式保活(`nonstream_keepalive_seconds`,默认 0 = 关闭):宽限期内
115+
# 完成仍是逐字节相同的 JSONResponse,超时才转 chunked 吐空白心跳。
116+
keepalive = float(getattr(settings, "nonstream_keepalive_seconds", 0) or 0)
117+
if keepalive > 0:
118+
return await respond_with_keepalive(
119+
lambda: executor.complete(chat_request, username=principal.username,
120+
provider_binding=binding),
121+
grace=keepalive, convert=convert, release=lease.release)
107122
try:
108123
completion = await executor.complete(chat_request, username=principal.username,
109124
provider_binding=binding)

‎src/api/streaming.py‎

Lines changed: 118 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,24 +3,46 @@
33
from __future__ import annotations
44

55
import asyncio
6-
from collections.abc import AsyncIterator
6+
import json
7+
import logging
8+
from collections.abc import AsyncIterator, Awaitable, Callable
9+
from typing import Any
10+
11+
from starlette.responses import JSONResponse, Response, StreamingResponse
712

813
from ..engine.sse import SSE_COMMENT
914

15+
logger = logging.getLogger(__name__)
16+
1017
# 上游思考/排队时可能长时间不吐字节;nginx proxy_read_timeout 默认 60s,
1118
# 空闲超过它就会断开连接。SSE 注释帧被客户端忽略,只为保活。
1219
KEEPALIVE_SECONDS = 15.0
1320

21+
# 非流式保活字节:JSON 允许任意前导空白(RFC 8259 §2 `JSON-text = ws value ws`),
22+
# 解析器(Python json / JS JSON.parse / Go encoding/json)都照常接受
23+
# `"\n\n{\"a\":1}"`。发换行比发空格更不容易被中间盒做行首嗅探改写。
24+
JSON_KEEPALIVE_BYTES = b"\n"
25+
1426

1527
async def with_keepalive(frames: AsyncIterator[bytes],
16-
interval: float = KEEPALIVE_SECONDS) -> AsyncIterator[bytes]:
17-
"""在上游无输出的空隙插入 SSE 注释帧。
28+
interval: float | None = None,
29+
keepalive_frame: bytes = SSE_COMMENT) -> AsyncIterator[bytes]:
30+
"""在上游无输出的空隙插入心跳帧。
31+
32+
`keepalive_frame` 由出口协议决定:chat / Responses 用 SSE 注释帧(被客户端
33+
忽略),Anthropic 用具名 `event: ping`(部分客户端只解析 `event:` 行,注释帧
34+
对它们等于没有字节,长静默下仍会判超时)。
35+
36+
`interval=None` 读模块常量 `KEEPALIVE_SECONDS`(运行期读而非默认参数,
37+
否则 import 时就定死,测试无法缩短)。
1838
1939
实现要点:取下一帧用**可复用的 Task** 而不是 `wait_for(anext(...))`。
2040
后者在超时时会 cancel 掉 `anext` 协程,取消会穿透进上游生成器并把
2141
它关闭——心跳反而变成了断流。这里用 `asyncio.wait` 与定时器竞争,
22-
超时只发注释、待取帧的 Task 继续存活。
42+
超时只发心跳、待取帧的 Task 继续存活。
2343
"""
44+
if interval is None:
45+
interval = KEEPALIVE_SECONDS
2446
iterator = frames.__aiter__()
2547
pending: asyncio.Task[bytes] | None = None
2648
try:
@@ -41,11 +63,102 @@ async def with_keepalive(frames: AsyncIterator[bytes],
4163
continue
4264
# 定时器先到:上游暂无输出 → 发心跳,保留 pending 继续等
4365
ticker.cancel()
44-
yield SSE_COMMENT
66+
yield keepalive_frame
4567
finally:
4668
# 客户端断开 / 生成器关闭:清掉在途任务,避免 "Task was destroyed"。
4769
# cancel() 只是「请求取消」,必须 await 到任务真正结束再退出——否则
4870
# 生成器所在事件循环很快回收,任务以 pending 状态被销毁并打印告警。
4971
if pending is not None and not pending.done():
5072
pending.cancel()
5173
await asyncio.gather(pending, return_exceptions=True)
74+
75+
76+
# 保活已提交 200 之后上游才失败时的正文:状态码已经发出去了,改不回来,只能
77+
# 给一个可解析的 JSON 错误体(比截断强:截断的 JSON 在客户端是解码异常,
78+
# 完全看不出是上游问题)。文案固定,不回显上游正文。
79+
LATE_FAILURE_PAYLOAD: dict[str, Any] = {
80+
"error": {
81+
"message": "upstream request failed after the response had started",
82+
"type": "api_error",
83+
"code": "upstream_error",
84+
"status": 502,
85+
}
86+
}
87+
88+
89+
def _dumps(payload: Any) -> bytes:
90+
"""与 `JSONResponse.render` 逐字节一致(分隔符/allow_nan 都对齐)。"""
91+
return json.dumps(payload, ensure_ascii=False, allow_nan=False,
92+
indent=None, separators=(",", ":")).encode("utf-8")
93+
94+
95+
async def respond_with_keepalive(
96+
work: Callable[[], Awaitable[dict[str, Any]]],
97+
*, grace: float,
98+
convert: Callable[[dict[str, Any]], Any] | None = None,
99+
release: Callable[[], None] | None = None) -> Response:
100+
"""非流式出口的保活(`nonstream_keepalive_seconds`,默认 0 关闭)。
101+
102+
**宽限期设计**:先等 `grace` 秒。
103+
104+
- 期间完成 → 返回普通 `JSONResponse`:状态码、`Content-Length`、字节内容
105+
与不开保活时**完全一致**,异常也照常抛给上层异常处理器(400/503 不变)。
106+
- 超时 → 才提交 200 + chunked,先吐空白字节再补正文。
107+
108+
为什么不一开始就用 `StreamingResponse`:那样 200 在聚合前就发出去了,
109+
之后任何失败(无可用凭证 503 / 请求无效 400)都改不回来。宽限期把
110+
「改不回来」的窗口压到只有真正超过 `grace` 的请求。
111+
112+
空白字节合法:RFC 8259 §2 规定 `JSON-text = ws value ws`,值前允许任意
113+
空白,故 `"\\n\\n{...}"` 仍是合法 JSON(Python `json.loads` / JS
114+
`JSON.parse` 均已实测接受)。
115+
"""
116+
task = asyncio.ensure_future(work())
117+
try:
118+
done, _ = await asyncio.wait({task}, timeout=grace)
119+
except BaseException:
120+
# 客户端断开 / 上层取消:必须把工作协程一起收掉,否则它继续跑完整个
121+
# 上游调用(真实计费)却没人接收结果。
122+
task.cancel()
123+
await asyncio.gather(task, return_exceptions=True)
124+
if release is not None:
125+
release()
126+
raise
127+
if task in done:
128+
try:
129+
completion = task.result()
130+
finally:
131+
if release is not None:
132+
release()
133+
return JSONResponse(convert(completion) if convert else completion)
134+
return StreamingResponse(
135+
_json_frames(task, grace, convert, release),
136+
media_type="application/json",
137+
headers={"X-Accel-Buffering": "no", "Cache-Control": "no-store"})
138+
139+
140+
async def _json_frames(task: asyncio.Task[dict[str, Any]], interval: float,
141+
convert: Callable[[dict[str, Any]], Any] | None,
142+
release: Callable[[], None] | None) -> AsyncIterator[bytes]:
143+
"""慢请求的正文生成器:空闲 `interval` 秒就吐一个空白字节。"""
144+
try:
145+
while not task.done():
146+
await asyncio.wait({task}, timeout=interval)
147+
if not task.done():
148+
yield JSON_KEEPALIVE_BYTES
149+
try:
150+
completion = task.result()
151+
except asyncio.CancelledError:
152+
raise
153+
except Exception: # noqa: BLE001 - 已提交 200,只能写进正文
154+
logger.warning("非流式保活:200 已提交后上游才失败,正文降级为错误体",
155+
exc_info=True)
156+
yield _dumps(LATE_FAILURE_PAYLOAD)
157+
return
158+
yield _dumps(convert(completion) if convert else completion)
159+
finally:
160+
if not task.done():
161+
task.cancel()
162+
await asyncio.gather(task, return_exceptions=True)
163+
if release is not None:
164+
release()

‎src/compat/anthropic/response.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -157,10 +157,13 @@ def finish(self) -> Iterator[bytes]:
157157
yield from self._close("stop")
158158

159159
def keepalive(self) -> bytes:
160-
"""SSE 注释帧心跳(与 chat / Responses 出口同一约定)。"""
161-
from ...engine.sse import SSE_COMMENT
160+
"""Anthropic 心跳:具名 `ping` 事件(不是 SSE 注释帧)。
162161
163-
return SSE_COMMENT
162+
官方流式规范里的 `ping` 是**具名事件**(`event: ping` +
163+
`data: {"type": "ping"}`),客户端只解析 `event:` 行时注释帧等于没有
164+
字节,长静默(推理模型 TTFT 可达 40s)下仍会判超时断开。
165+
"""
166+
return b'event: ping\ndata: {"type": "ping"}\n\n'
164167

165168
def error_frame(self, message: str, code: str) -> bytes:
166169
"""出口错误帧:Anthropic 用 `event: error`(无 [DONE] 哨兵)。"""

‎src/compat/openai/response.py‎

Lines changed: 30 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@
22
33
约定(继承 codebuddy2api 的语义):
44
- 首块补 role:assistant
5-
- 上游缺失 index 的 tool_calls 补稳定 index,不重生成 id
5+
- 上游缺失 index 的 tool_calls 补稳定 index,不重生成已有 id;`id` 缺失时
6+
按 index 补一个稳定值(OpenAI 客户端按 id 聚合分片,缺 id 会丢工具调用)
67
- reasonng 走 delta.reasoning_content
78
- 流以 data: [DONE] 结束
89
"""
@@ -58,19 +59,26 @@ def _completion_id() -> str:
5859

5960
@dataclass
6061
class ToolIndexState:
61-
"""优先沿用上游 index,缺失时按出现顺序补稳定位置。"""
62+
"""优先沿用上游 index,缺失时按出现顺序补稳定位置;`id` 缺失时补稳定 id。
63+
64+
上游(尤其聚合型中转)偶尔只发 `function.arguments` 分片而不发 `id`。
65+
OpenAI 客户端按 `id` 聚合流式工具调用分片,缺 id 的分片会被丢弃,表现为
66+
「模型调了工具但客户端收不到」。故按 index 生成并记忆 id,后续同 index 的
67+
分片沿用同一个值(每次现生成会让同一调用被拆成多个)。
68+
"""
6269

6370
id_to_index: dict[str, int] = field(default_factory=dict)
71+
index_to_id: dict[int, str] = field(default_factory=dict)
6472
used: set[int] = field(default_factory=set)
73+
_seq: int = 0
6574

6675
def resolve(self, tool_call: dict[str, Any]) -> tuple[dict[str, Any], int]:
6776
item = dict(tool_call)
6877
index = item.get("index")
6978
tool_id = item.get("id")
7079
if isinstance(index, int) and not isinstance(index, bool):
7180
self.used.add(index)
72-
if isinstance(tool_id, str) and tool_id:
73-
self.id_to_index[tool_id] = index
81+
self._bind(index, tool_id, item)
7482
return item, index
7583
if isinstance(tool_id, str) and tool_id and tool_id in self.id_to_index:
7684
resolved = self.id_to_index[tool_id]
@@ -80,11 +88,27 @@ def resolve(self, tool_call: dict[str, Any]) -> tuple[dict[str, Any], int]:
8088
while candidate in self.used:
8189
candidate += 1
8290
self.used.add(candidate)
83-
if isinstance(tool_id, str) and tool_id:
84-
self.id_to_index[tool_id] = candidate
91+
self._bind(candidate, tool_id, item)
8592
item["index"] = candidate
8693
return item, candidate
8794

95+
def _bind(self, index: int, tool_id: Any, item: dict[str, Any]) -> None:
96+
"""登记 id↔index 双向映射;`id` 缺失时按 index 补一个稳定值。
97+
98+
每片都补同一个 id(而不是只补首片):按 id 聚合的客户端若在续片上
99+
拿到空 id,会把一次调用拆成两条;重复同一个 id 则各种聚合方式都安全。
100+
"""
101+
if isinstance(tool_id, str) and tool_id:
102+
self.id_to_index[tool_id] = index
103+
self.index_to_id.setdefault(index, tool_id)
104+
return
105+
existing = self.index_to_id.get(index)
106+
if existing is None:
107+
self._seq += 1
108+
existing = f"call_{index}_{self._seq}"
109+
self.index_to_id[index] = existing
110+
item["id"] = existing
111+
88112

89113
class StreamTranslator:
90114
"""把 Event 序列翻译为 OpenAI SSE 帧序列。"""

‎src/compat/responses/response.py‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,12 @@ def finish(self) -> Iterator[bytes]:
190190
yield from self._close("stop")
191191

192192
def keepalive(self) -> bytes:
193-
"""SSE 注释帧心跳(与 chat 出口同一约定)。"""
193+
"""SSE 注释帧心跳(与 chat 出口同一约定)。
194+
195+
**不能改成 `event: ping`**:Responses API 没有 ping 事件类型,严格
196+
解析器收到未知事件会直接报错(`MAXeaglet/commandcode-proxy` 的实测
197+
结论,与 Anthropic 出口的处理相反)。
198+
"""
194199
from ...engine.sse import SSE_COMMENT
195200

196201
return SSE_COMMENT

0 commit comments

Comments
 (0)