From 9309ad1ffcd92b2a604158ff88bc768a49fd1d1d Mon Sep 17 00:00:00 2001 From: Wang Defa <1+wangdefa@noreply.gitea.bcde.io> Date: Mon, 13 Jul 2026 15:06:01 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20AI=20=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E6=96=AD=E6=B5=81:=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E9=80=8F=E4=BC=A0=E3=80=81=E8=87=AA=E5=8A=A8=E9=99=8D=E7=BA=A7?= =?UTF-8?q?,max=5Ftokens=20=E5=8F=AF=E7=BC=BA=E7=9C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 14 +++ README.md | 18 +++- docs/docs.go | 2 +- docs/swagger.json | 2 +- docs/swagger.yaml | 3 +- internal/api/aigateway.go | 121 +++++++++++++++++++++-- internal/api/aigateway_test.go | 2 + internal/service/airesponses.go | 22 +++++ internal/service/anthresponses.go | 130 +++++++++++++++++++++++-- internal/service/anthresponses_test.go | 105 +++++++++++++++++++- internal/service/chatresponses.go | 29 ++++++ 11 files changed, 420 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 89e7809..09e67fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ 格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/)，版本号遵循语义化版本。 +## [0.3.1] - 2026-07-13 + +### Added + +- AI 网关流式断流自动降级：Messages 与 Chat Completions 流式请求在客户端尚未收到任何输出时遭遇上游断流，自动降级为非流式重做，结果按标准事件 / chunk 序列一次推送，调用日志记 `retries=1` 与降级标记。实测 OCI 兼容面对 `instructions` 与 `tools` 合计超约 64.5KB 的流式请求会静默断连（无错误事件，非流式正常，消息正文不计入），Claude Code 等大体量系统提示客户端极易触发；Responses 直通因初始事件已转发无法透明降级，日志记「上游流提前终止」；README 增补「已知上游限制：大 system 区流式断流」小节 + +### Changed + +- Messages 的 `max_tokens` 改为可缺省：缺省或 ≤0 时按默认值 8192 放行（此前返回 400；部分客户端将该字段视为选填） + +### Fixed + +- AI 网关流式假成功：上游 error / `response.failed` 事件此前被吞掉，流提前 EOF 也被伪装成正常结束，调用日志呈现 200 · 0/0 且无错误信息；现 Messages 将上游失败转为 Anthropic `error` 事件，三个流式端点日志均记录上游错误消息或「上游流提前终止，未返回终态事件」 + ## [0.3.0] - 2026-07-13 ### Added diff --git a/README.md b/README.md index ac84aee..39e1149 100644 --- a/README.md +++ b/README.md @@ -216,6 +216,16 @@ AI 网关使用面板创建的独立密钥鉴权，支持 `Authorization: Bearer - Chat Completions 只承担协议转换与兼容修复；新能力优先在 Responses 和 Messages 提供 - 单次请求最多尝试三个渠道；可重试错误会切换渠道，流式响应建立后不会换渠道重试 +### 已知上游限制：大 system 区流式断流 + +实测（2026-07-13）OCI 兼容面对 `instructions` 与 `tools` 合计超约 64.5KB 的**流式**请求会在发出少量事件后静默断开连接（无任何错误事件；同请求非流式正常），与模型、字符集、消息正文大小均无关——消息正文（`input`）不计入该限制。Chat Completions 与 Messages 的 system/developer 提示会转换为 `instructions`，因此 Claude Code 等自带大体量系统提示与工具定义的客户端极易触发。 + +网关侧应对： + +- Messages 与 Chat Completions 的流式请求在客户端尚未收到任何输出时遭遇上游断流，会自动降级为非流式重做，并按标准事件/chunk 序列一次推送；调用日志记 `retries=1` 与降级标记 +- Responses 直通因初始事件已转发、协议上无法透明降级，调用日志记「上游流提前终止」，客户端需自行回退非流式 +- 应急规避：将超长 system 内容移入首条 user 消息正文可绕过该限制（正文不计入），但语义有别，根治有待上游修复 + 这里提供的是兼容接口而非 OpenAI / Anthropic 协议的完整实现。OCI OpenAI 兼容面的部分行为来自实测，未见 Oracle 文档承诺，可能随上游调整。路由与鉴权定义以 [Swagger YAML](docs/swagger.yaml) 或运行时 Swagger UI 为准；无法由 OpenAPI 完整表达的兼容边界列于上方。 ### 字段兼容矩阵 @@ -289,7 +299,7 @@ Chat Completions 会先转换为 Responses 请求，再把 OCI Responses 响应 | `max_completion_tokens` | 🔄 | 转为 `max_output_tokens`，优先于 `max_tokens` | | `max_tokens` | 🔄 | 未提供 `max_completion_tokens` 时转为 `max_output_tokens` | | `temperature`、`top_p`、`parallel_tool_calls` | ◐ | 原值写入 Responses 请求，但不校验范围或模型能力 | -| `stream` | 🔄 | OCI Responses SSE 桥接为 `chat.completion.chunk`，末尾补 `data: [DONE]` | +| `stream` | 🔄 | OCI Responses SSE 桥接为 `chat.completion.chunk`，末尾补 `data: [DONE]`；上游断流且尚无输出时自动降级非流式重做，结果按 chunk 序列一次推送 | | `stream_options.include_usage` | 🔄 | 控制网关在终块后追加 `choices: []` 的 usage 块 | | `stream_options.include_obfuscation` | ⚠️ | 未建模，静默忽略 | | `tools[].type=function` | ◐ | `name`、`description`、`parameters` 支持；`function.strict` 被忽略 | @@ -346,13 +356,13 @@ Chat Completions 会先转换为 Responses 请求，再把 OCI Responses 响应 | 标准字段 | 状态 | 网关行为 | | --- | :---: | --- | | `model` | ◐ | 用于模型与渠道选择，但空字符串不会在 handler 中按参数错误拒绝，通常最终返回模型不存在 | -| `max_tokens` | 🔄 | 必填且必须大于 0，转换为 `max_output_tokens` | +| `max_tokens` | 🔄 | 可缺省（缺省或 ≤0 时按默认值 8192），转换为 `max_output_tokens` | | `messages` | ◐ | 必须非空；角色、交替顺序和空内容不做完整校验 | | `system` | ◐ | 支持字符串或 text 块数组；多个文本块直接拼接，`cache_control` 等附加字段被忽略 | | `temperature`、`top_p` | ◐ | 写入 Responses 请求，不做取值范围或模型能力校验 | | `top_k` | ⚠️ | 能解析但不会传给上游 | | `stop_sequences` | ⚠️ | 能解析但不会传给上游；响应 `stop_sequence` 恒为 `null` | -| `stream` | 🔄 | Responses SSE 桥接为 Anthropic 事件序列 | +| `stream` | 🔄 | Responses SSE 桥接为 Anthropic 事件序列；上游断流且尚无输出时自动降级非流式重做，结果按事件序列一次推送 | | `tools` | ◐ | 每个工具都转换成 Responses `function`；自定义客户端工具可用，Anthropic 服务端工具类型不保留原语义 | | `tool_choice` | ◐ | 支持 `auto`、`any`、`none`、具名 `tool`；`disable_parallel_tool_use` 等附加字段被忽略 | | `metadata` | ⚠️ | 能解析但不会传给上游 | @@ -380,7 +390,7 @@ Chat Completions 会先转换为 Responses 请求，再把 OCI Responses 响应 - `stop_reason` 只生成 `end_turn`、`tool_use`、`max_tokens`；`stop_sequence` 恒为 `null` - reasoning 不会生成 Anthropic `thinking` / `redacted_thinking` 块，也没有 signature - usage 只保留 `input_tokens`、`output_tokens` 和 `cache_read_input_tokens`，不提供 `cache_creation_input_tokens` -- 流式输出标准事件骨架，但不生成 `thinking_delta`、`signature_delta` 或上游失败对应的 Anthropic `error` 事件 +- 流式输出标准事件骨架，但不生成 `thinking_delta` 和 `signature_delta`；上游错误事件与无终态断流会转成 Anthropic `error` 事件 实现依据：[`anthresponses.go`](internal/service/anthresponses.go) · [`anthropic.go`](internal/aiwire/anthropic.go) · [`aigateway.go`](internal/api/aigateway.go) diff --git a/docs/docs.go b/docs/docs.go index 8ee4b86..d49d016 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -79,7 +79,7 @@ const docTemplate = `{ "summary": "Anthropic Messages 兼容端点", "parameters": [ { - "description": "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通)", + "description": "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通;max_tokens 可缺省,默认 8192)", "name": "body", "in": "body", "required": true, diff --git a/docs/swagger.json b/docs/swagger.json index d1a391c..1c98a76 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -72,7 +72,7 @@ "summary": "Anthropic Messages 兼容端点", "parameters": [ { - "description": "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通)", + "description": "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通;max_tokens 可缺省,默认 8192)", "name": "body", "in": "body", "required": true, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 71407ef..636a6c5 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -617,7 +617,8 @@ paths: /ai/v1/messages: post: parameters: - - description: Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通) + - description: Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通;max_tokens + 可缺省,默认 8192) in: body name: body required: true diff --git a/internal/api/aigateway.go b/internal/api/aigateway.go index 0f2766f..5859550 100644 --- a/internal/api/aigateway.go +++ b/internal/api/aigateway.go @@ -210,11 +210,15 @@ func sseHeaders(c *gin.Context) { c.Writer.Flush() } +// anthDefaultMaxTokens 是 max_tokens 缺省时的默认输出上限:Anthropic 协议 +// 该字段必填,但部分客户端当选填不传,按默认值放行而非 400 拒绝。 +const anthDefaultMaxTokens = 8192 + // messages 是 Anthropic /ai/v1/messages 端点。 // // @Summary Anthropic Messages 兼容端点 // @Tags AI 网关 -// @Param body body object true "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通)" +// @Param body body object true "Anthropic messages 请求体(支持 stream;经 OCI OpenAI 兼容面直通;max_tokens 可缺省,默认 8192)" // @Success 200 {object} map[string]any "Anthropic 兼容响应(流式为 SSE)" // @Router /ai/v1/messages [post] func (h *aiGatewayHandler) messages(c *gin.Context) { @@ -224,8 +228,7 @@ func (h *aiGatewayHandler) messages(c *gin.Context) { return } if req.MaxTokens <= 0 { - aiError(c, http.StatusBadRequest, "invalid_request_error", "max_tokens 必填且需大于 0") - return + req.MaxTokens = anthDefaultMaxTokens } if len(req.Messages) == 0 { aiError(c, http.StatusBadRequest, "invalid_request_error", "messages 不能为空") @@ -281,10 +284,30 @@ func (h *aiGatewayHandler) streamAnthropic(c *gin.Context, body []byte, req aiwi defer upstream.Close() sseHeaders(c) bridge := service.NewAnthRespBridge(aiRandID("msg_"), req.Model) - if err := forwardSSEData(upstream, func(data []byte) { writeAnthEvents(c, bridge.Feed(data)) }); err != nil { + emitted := 0 + if err := forwardSSEData(upstream, func(data []byte) { + evs := bridge.Feed(data) + emitted += len(evs) + writeAnthEvents(c, evs) + }); err != nil { entry.ErrMsg = err.Error() } + // 上游断流且客户端尚未收到任何事件(OCI 兼容面对大请求 + 推理模型的 + // 流式通道会在 reasoning 阶段掐断):降级非流式重做,结果按事件序列推送 + if emitted == 0 && !bridge.SawTerminal() && c.Request.Context().Err() == nil { + if h.anthFallback(c, &entry, req) { + entry.Status = http.StatusOK + entry.LatencyMs = time.Since(start).Milliseconds() + callID := h.gw.LogCall(entry) + h.maybeLogContent(c, callID, "anthropic", req.Model, true, req, nil) + return + } + } writeAnthEvents(c, bridge.Finish()) + // 上游错误事件 / 提前终止:SSE 头已发出维持 200,错误落日志可查 + if msg := bridge.Err(); msg != "" && entry.ErrMsg == "" { + entry.ErrMsg = msg + } entry.Status = http.StatusOK entry.LatencyMs = time.Since(start).Milliseconds() usage := bridge.Usage() @@ -295,6 +318,30 @@ func (h *aiGatewayHandler) streamAnthropic(c *gin.Context, body []byte, req aiwi h.maybeLogContent(c, callID, "anthropic", req.Model, true, req, nil) } +// anthFallback 用非流式重做同一请求并把完整结果按事件序列推送; +// 成功返回 true 并把渠道 / 用量 / 降级标注写入日志条目。 +func (h *aiGatewayHandler) anthFallback(c *gin.Context, entry *model.AiCallLog, req aiwire.MessagesRequest) bool { + req.Stream = false + body, err := service.AnthropicToResponsesBody(req) + if err != nil { + return false + } + payload, meta, err := h.gw.RespPassthrough(c.Request.Context(), body, req.Model, keyGroup(c)) + if err != nil { + return false + } + out, err := service.ResponsesToAnthropic(payload, aiRandID("msg_")) + if err != nil { + return false + } + writeAnthEvents(c, service.AnthMessageEvents(out)) + entry.ChannelID, entry.ChannelName = meta.ChannelID, meta.ChannelName + entry.Retries++ + entry.ErrMsg = "流式上游断流,已降级非流式完成" + fillUsage(entry, service.RespPassthroughUsage(payload)) + return true +} + // forwardSSEData 逐行读上游 SSE,把 data 行交给 emit 即时转换写出; // Anthropic 与 Chat Completions 两条流式桥共用。 func forwardSSEData(upstream io.Reader, emit func([]byte)) error { @@ -403,12 +450,31 @@ func (h *aiGatewayHandler) streamChat(c *gin.Context, body []byte, req aiwire.Ch sseHeaders(c) includeUsage := req.StreamOptions != nil && req.StreamOptions.IncludeUsage bridge := service.NewChatRespBridge(aiRandID("chatcmpl-"), req.Model, time.Now().Unix(), includeUsage) - if err := forwardSSEData(upstream, func(data []byte) { writeChatChunks(c, bridge.Feed(data)) }); err != nil { + emitted := 0 + if err := forwardSSEData(upstream, func(data []byte) { + chunks := bridge.Feed(data) + emitted += len(chunks) + writeChatChunks(c, chunks) + }); err != nil { entry.ErrMsg = err.Error() } + // 上游断流且客户端尚未收到任何块:降级非流式重做(成因同 messages 端点) + if emitted == 0 && bridge.Usage() == nil && c.Request.Context().Err() == nil { + if h.chatFallback(c, &entry, req, includeUsage) { + entry.Status = http.StatusOK + entry.LatencyMs = time.Since(start).Milliseconds() + callID := h.gw.LogCall(entry) + h.maybeLogContent(c, callID, "openai", req.Model, true, req, nil) + return + } + } writeChatChunks(c, bridge.Finish()) c.Writer.WriteString("data: [DONE]\n\n") c.Writer.Flush() + // 未见 completed 即结束:usage 缺失说明上游流异常提前终止,落日志可查 + if bridge.Usage() == nil && entry.ErrMsg == "" { + entry.ErrMsg = "上游流提前终止,未返回终态事件" + } entry.Status = http.StatusOK entry.LatencyMs = time.Since(start).Milliseconds() fillUsage(&entry, bridge.Usage()) @@ -416,6 +482,32 @@ func (h *aiGatewayHandler) streamChat(c *gin.Context, body []byte, req aiwire.Ch h.maybeLogContent(c, callID, "openai", req.Model, true, req, nil) } +// chatFallback 用非流式重做同一请求并把完整结果按 chunk 序列推送; +// 成功返回 true 并把渠道 / 用量 / 降级标注写入日志条目。 +func (h *aiGatewayHandler) chatFallback(c *gin.Context, entry *model.AiCallLog, req aiwire.ChatRequest, includeUsage bool) bool { + req.Stream = false + body, err := service.ChatToResponsesBody(req) + if err != nil { + return false + } + payload, meta, err := h.gw.RespPassthrough(c.Request.Context(), body, req.Model, keyGroup(c)) + if err != nil { + return false + } + out, err := service.ResponsesToChat(payload, aiRandID("chatcmpl-"), time.Now().Unix()) + if err != nil { + return false + } + writeChatChunks(c, service.ChatResponseChunks(out, includeUsage)) + c.Writer.WriteString("data: [DONE]\n\n") + c.Writer.Flush() + entry.ChannelID, entry.ChannelName = meta.ChannelID, meta.ChannelName + entry.Retries++ + entry.ErrMsg = "流式上游断流,已降级非流式完成" + fillUsage(entry, service.RespPassthroughUsage(payload)) + return true +} + // writeChatChunks 逐块写出 chunk 的 SSE data 行并 flush。 func writeChatChunks(c *gin.Context, chunks []aiwire.ChatChunk) { for _, ch := range chunks { @@ -526,9 +618,14 @@ func (h *aiGatewayHandler) responsesPassthroughStream(c *gin.Context, body []byt } defer upstream.Close() sseHeaders(c) - usage, err := forwardSSE(c, upstream) + usage, upErr, err := forwardSSE(c, upstream) if err != nil { entry.ErrMsg = err.Error() + } else if upErr != "" { + // 上游错误事件已原样转发给客户端,这里落日志可查 + entry.ErrMsg = upErr + } else if usage == nil { + entry.ErrMsg = "上游流提前终止,未返回终态事件" } entry.Status = http.StatusOK entry.LatencyMs = time.Since(start).Milliseconds() @@ -538,10 +635,11 @@ func (h *aiGatewayHandler) responsesPassthroughStream(c *gin.Context, body []byt } // forwardSSE 把上游 SSE 逐行转发给客户端,空行(事件边界)即 flush; -// 顺带从 data 行提取 response.completed 的 usage。 -func forwardSSE(c *gin.Context, upstream io.Reader) (*aiwire.Usage, error) { +// 顺带从 data 行提取 response.completed 的 usage 与错误事件消息。 +func forwardSSE(c *gin.Context, upstream io.Reader) (*aiwire.Usage, string, error) { reader := bufio.NewReader(upstream) var usage *aiwire.Usage + var upErr string for { line, err := reader.ReadBytes('\n') if len(line) > 0 { @@ -553,14 +651,17 @@ func forwardSSE(c *gin.Context, upstream io.Reader) (*aiwire.Usage, error) { if u := service.RespStreamCompletedUsage(data); u != nil { usage = u } + if m := service.RespStreamErrorMsg(data); m != "" && upErr == "" { + upErr = m + } } } if err != nil { c.Writer.Flush() if errors.Is(err, io.EOF) { - return usage, nil + return usage, upErr, nil } - return usage, err + return usage, upErr, err } } } diff --git a/internal/api/aigateway_test.go b/internal/api/aigateway_test.go index 0d22f65..081662c 100644 --- a/internal/api/aigateway_test.go +++ b/internal/api/aigateway_test.go @@ -85,6 +85,8 @@ func TestAiGatewayKeyModelRestrict(t *testing.T) { `{"model":"meta.llama-3.3-70b-instruct","messages":[{"role":"user","content":"hi"}]}`, 404, deny}, {"chat completions 白名单内穿透", "limited-key-1234", "/ai/v1/chat/completions", `{"model":"ghost-model","messages":[{"role":"user","content":"hi"}]}`, 404, pass}, + {"messages 缺 max_tokens 按默认值放行", "open-key-12345", "/ai/v1/messages", + `{"model":"ghost-model","messages":[{"role":"user","content":"hi"}]}`, 404, pass}, {"chat completions 缺 messages 拒绝", "open-key-12345", "/ai/v1/chat/completions", `{"model":"ghost-model"}`, 400, []string{"invalid_request_error"}}, {"chat completions 非 function 工具拒绝", "open-key-12345", "/ai/v1/chat/completions", diff --git a/internal/service/airesponses.go b/internal/service/airesponses.go index 87cdaf0..d4ed5ea 100644 --- a/internal/service/airesponses.go +++ b/internal/service/airesponses.go @@ -104,3 +104,25 @@ func RespStreamCompletedUsage(data []byte) *aiwire.Usage { } return RespPassthroughUsage(ev.Response) } + +// RespStreamErrorMsg 从一行 SSE data JSON 中提取 error / response.failed 事件的 +// 错误消息;非错误事件返回空串。流式直通据此把上游错误写入调用日志。 +func RespStreamErrorMsg(data []byte) string { + var ev respStreamEvent + if json.Unmarshal(data, &ev) != nil { + return "" + } + switch ev.Type { + case "error": + if ev.Message != "" { + return ev.Message + } + return "上游返回错误事件 error" + case "response.failed": + if m := respErrorMsg(ev.Response); m != "" { + return m + } + return "上游返回错误事件 response.failed" + } + return "" +} diff --git a/internal/service/anthresponses.go b/internal/service/anthresponses.go index b574bc3..2552334 100644 --- a/internal/service/anthresponses.go +++ b/internal/service/anthresponses.go @@ -258,7 +258,7 @@ type AnthEvent struct { // AnthRespBridge 把直通 SSE 事件流桥接为 Anthropic 事件序列: // message_start → content_block_start/delta/stop(text 与 tool_use 分块)→ message_delta → message_stop。 -// reasoning 系列事件丢弃。 +// reasoning 系列事件丢弃;上游 error / response.failed 转 Anthropic error 事件透传。 type AnthRespBridge struct { id, model string started bool @@ -267,6 +267,11 @@ type AnthRespBridge struct { blockIndex int stopReason string usage aiwire.AnthUsage + // sawTerminal 标记收到过终态事件(completed/incomplete/failed); + // 上游流提前 EOF 时据此发 error 事件而非伪装正常结束 + sawTerminal bool + // errMsg 记录上游错误消息,非空即本流已失败(供调用日志) + errMsg string } // NewAnthRespBridge 构造桥;id 为响应消息 ID。 @@ -278,6 +283,7 @@ func NewAnthRespBridge(id, model string) *AnthRespBridge { type respStreamEvent struct { Type string `json:"type"` Delta string `json:"delta"` + Message string `json:"message"` Item *respOutputItem `json:"item"` Response *respPayload `json:"response"` } @@ -288,27 +294,130 @@ func (st *AnthRespBridge) Feed(data []byte) []AnthEvent { if json.Unmarshal(data, &ev) != nil { return nil } - var events []AnthEvent - if !st.started { - st.started = true - events = append(events, st.startEvent()) + if ev.Type == "error" || ev.Type == "response.failed" { + return st.failWith(ev) } + // message_start 延迟到首个可见输出事件:created / reasoning 阶段不向客户端 + // 写任何字节,上游此段断流时 handler 才有降级非流式重做的无感窗口 + var events []AnthEvent switch ev.Type { case "response.output_item.added": if ev.Item != nil && ev.Item.Type == "function_call" { + events = append(events, st.ensureStarted()...) events = append(events, st.openBlock(true, ev.Item.CallID, ev.Item.Name)...) st.stopReason = "tool_use" } case "response.output_text.delta": + events = append(events, st.ensureStarted()...) events = append(events, st.textDelta(ev.Delta)...) case "response.function_call_arguments.delta": events = append(events, st.argsDelta(ev.Delta)...) - case "response.completed", "response.incomplete", "response.failed": + case "response.completed", "response.incomplete": + st.sawTerminal = true st.finishFrom(ev.Response) } return events } +// ensureStarted 在首个可见输出事件前补发 message_start(仅一次)。 +func (st *AnthRespBridge) ensureStarted() []AnthEvent { + if st.started { + return nil + } + st.started = true + return []AnthEvent{st.startEvent()} +} + +// failWith 记录上游错误并产出 Anthropic error 事件(每流至多一次)。 +func (st *AnthRespBridge) failWith(ev respStreamEvent) []AnthEvent { + if st.errMsg != "" { + return nil + } + st.sawTerminal = true + msg := ev.Message + if ev.Type == "response.failed" { + st.finishFrom(ev.Response) + if m := respErrorMsg(ev.Response); m != "" { + msg = m + } + } + if msg == "" { + msg = "上游返回错误事件 " + ev.Type + } + st.errMsg = msg + return []AnthEvent{st.errorEvent()} +} + +// respErrorMsg 提取 response.failed 载荷中的错误消息。 +func respErrorMsg(resp *respPayload) string { + if resp == nil || resp.Error == nil { + return "" + } + return (*resp.Error)["message"] +} + +// errorEvent 按 Anthropic 流式协议产出 error 事件。 +func (st *AnthRespBridge) errorEvent() AnthEvent { + return AnthEvent{Event: "error", Data: map[string]any{ + "type": "error", + "error": map[string]string{"type": "api_error", "message": st.errMsg}, + }} +} + +// Err 返回上游错误消息;空串表示流正常(供调用日志)。 +func (st *AnthRespBridge) Err() string { return st.errMsg } + +// SawTerminal 报告是否收到过终态事件;false 即上游流提前终止。 +func (st *AnthRespBridge) SawTerminal() bool { return st.sawTerminal } + +// AnthMessageEvents 把完整 Messages 响应展开为标准事件序列, +// 供流式上游断流后的非流式降级结果推送(客户端协议不变)。 +func AnthMessageEvents(m *aiwire.MessagesResponse) []AnthEvent { + events := []AnthEvent{{Event: "message_start", Data: map[string]any{ + "type": "message_start", + "message": map[string]any{ + "id": m.ID, "type": "message", "role": m.Role, "model": m.Model, + "content": []any{}, "stop_reason": nil, + "usage": map[string]int{"input_tokens": 0, "output_tokens": 0}, + }, + }}} + for i, b := range m.Content { + events = append(events, anthBlockEvents(i, b)...) + } + usage := map[string]int{"input_tokens": m.Usage.InputTokens, "output_tokens": m.Usage.OutputTokens} + if m.Usage.CacheReadInputTokens > 0 { + usage["cache_read_input_tokens"] = m.Usage.CacheReadInputTokens + } + events = append(events, + AnthEvent{Event: "message_delta", Data: map[string]any{ + "type": "message_delta", + "delta": map[string]any{"stop_reason": m.StopReason, "stop_sequence": nil}, + "usage": usage, + }}, + AnthEvent{Event: "message_stop", Data: map[string]any{"type": "message_stop"}}) + return events +} + +// anthBlockEvents 把一个内容块展开为 start / delta / stop 三事件。 +func anthBlockEvents(index int, b aiwire.AnthBlock) []AnthEvent { + var start, delta map[string]any + if b.Type == "tool_use" { + start = map[string]any{"type": "tool_use", "id": b.ID, "name": b.Name, "input": map[string]any{}} + delta = map[string]any{"type": "input_json_delta", "partial_json": string(b.Input)} + } else { + start = map[string]any{"type": "text", "text": ""} + delta = map[string]any{"type": "text_delta", "text": b.Text} + } + return []AnthEvent{ + {Event: "content_block_start", Data: map[string]any{ + "type": "content_block_start", "index": index, "content_block": start}}, + {Event: "content_block_delta", Data: map[string]any{ + "type": "content_block_delta", "index": index, "delta": delta}}, + {Event: "content_block_stop", Data: map[string]any{ + "type": "content_block_stop", "index": index}}, + } +} + func (st *AnthRespBridge) textDelta(delta string) []AnthEvent { var events []AnthEvent if !st.blockOpen || st.blockIsTool { @@ -378,7 +487,16 @@ func (st *AnthRespBridge) closeBlockEvent() AnthEvent { } // Finish 在上游流结束后收尾:关块 → message_delta(stop_reason+usage)→ message_stop。 +// 已发过 error 事件的流不再补终态;未见终态事件即 EOF 视为上游提前终止, +// 发 error 事件而非伪装正常结束(否则客户端拿到"成功的空消息")。 func (st *AnthRespBridge) Finish() []AnthEvent { + if st.errMsg != "" { + return nil + } + if !st.sawTerminal { + st.errMsg = "上游流提前终止,未返回终态事件" + return []AnthEvent{st.errorEvent()} + } var events []AnthEvent if !st.started { st.started = true diff --git a/internal/service/anthresponses_test.go b/internal/service/anthresponses_test.go index 328a725..a059d1f 100644 --- a/internal/service/anthresponses_test.go +++ b/internal/service/anthresponses_test.go @@ -164,10 +164,105 @@ func TestAnthRespBridge(t *testing.T) { } } -// TestAnthRespBridgeEmpty 断言空流也产出完整事件骨架。 -func TestAnthRespBridgeEmpty(t *testing.T) { - st := NewAnthRespBridge("msg_1", "m1") - if got := bridgeEventTypes(st.Finish()); got != "message_start,message_delta,message_stop" { - t.Errorf("空流事件序列 = %s", got) +// TestAnthRespBridgeFailure 断言异常流的错误透传:上游 error / response.failed +// 事件转 Anthropic error 事件,提前 EOF(未见终态)不再伪装正常结束。 +func TestAnthRespBridgeFailure(t *testing.T) { + cases := []struct { + name string + lines []string + wantKinds string + wantErr string + }{ + { + name: "上游 error 事件透传", + lines: []string{`{"type":"error","message":"model overloaded"}`}, + wantKinds: "error", + wantErr: "model overloaded", + }, + { + name: "response.failed 提取错误消息", + lines: []string{ + `{"type":"response.output_text.delta","delta":"你"}`, + `{"type":"response.failed","response":{"status":"failed","error":{"message":"content filtered"}}}`, + }, + wantKinds: "message_start,content_block_start,content_block_delta,error", + wantErr: "content filtered", + }, + { + name: "空流提前终止", + lines: nil, + wantKinds: "error", + wantErr: "上游流提前终止,未返回终态事件", + }, + { + name: "输出中途 EOF 无终态", + lines: []string{ + `{"type":"response.output_text.delta","delta":"你"}`, + }, + wantKinds: "message_start,content_block_start,content_block_delta,error", + wantErr: "上游流提前终止,未返回终态事件", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + st := NewAnthRespBridge("msg_1", "m1") + var events []AnthEvent + for _, l := range tc.lines { + events = append(events, st.Feed([]byte(l))...) + } + events = append(events, st.Finish()...) + if got := bridgeEventTypes(events); got != tc.wantKinds { + t.Fatalf("事件序列 = %s, want %s", got, tc.wantKinds) + } + if st.Err() != tc.wantErr { + t.Fatalf("Err() = %q, want %q", st.Err(), tc.wantErr) + } + }) + } +} + +// TestRespStreamErrorMsg 断言直通流错误事件消息提取。 +func TestRespStreamErrorMsg(t *testing.T) { + cases := []struct { + name string + data string + want string + }{ + {name: "error 事件", data: `{"type":"error","message":"boom"}`, want: "boom"}, + {name: "error 无消息用占位", data: `{"type":"error"}`, want: "上游返回错误事件 error"}, + {name: "failed 事件", data: `{"type":"response.failed","response":{"error":{"message":"bad"}}}`, want: "bad"}, + {name: "正常事件返回空", data: `{"type":"response.completed","response":{}}`, want: ""}, + {name: "非 JSON 返回空", data: ``, want: ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := RespStreamErrorMsg([]byte(tc.data)); got != tc.want { + t.Fatalf("RespStreamErrorMsg = %q, want %q", got, tc.want) + } + }) + } +} + +// TestAnthMessageEvents 断言非流式降级结果展开的事件序列与 usage。 +func TestAnthMessageEvents(t *testing.T) { + msg := &aiwire.MessagesResponse{ + ID: "msg_1", Type: "message", Role: "assistant", Model: "m1", + Content: []aiwire.AnthBlock{ + {Type: "text", Text: "好"}, + {Type: "tool_use", ID: "c1", Name: "f", Input: json.RawMessage(`{"a":1}`)}, + }, + StopReason: "tool_use", + Usage: aiwire.AnthUsage{InputTokens: 9, OutputTokens: 3, CacheReadInputTokens: 5}, + } + events := AnthMessageEvents(msg) + want := "message_start,content_block_start,content_block_delta,content_block_stop," + + "content_block_start,content_block_delta,content_block_stop,message_delta,message_stop" + if got := bridgeEventTypes(events); got != want { + t.Fatalf("事件序列:\n got %s\nwant %s", got, want) + } + delta := events[len(events)-2].Data.(map[string]any) + usage := delta["usage"].(map[string]int) + if usage["input_tokens"] != 9 || usage["output_tokens"] != 3 || usage["cache_read_input_tokens"] != 5 { + t.Fatalf("usage = %+v", usage) } } diff --git a/internal/service/chatresponses.go b/internal/service/chatresponses.go index 84bc8ef..12be4b1 100644 --- a/internal/service/chatresponses.go +++ b/internal/service/chatresponses.go @@ -358,3 +358,32 @@ func (b *ChatRespBridge) Finish() []aiwire.ChatChunk { // Usage 返回聚合到的用量(供调用日志),上游未报告时为 nil。 func (b *ChatRespBridge) Usage() *aiwire.Usage { return b.usage } + +// ChatResponseChunks 把完整 Chat 响应展开为 chunk 序列(内容与工具调用 → +// 终块 → 可选 usage 块),供流式上游断流后的非流式降级结果推送。 +func ChatResponseChunks(resp *aiwire.ChatResponse, includeUsage bool) []aiwire.ChatChunk { + if len(resp.Choices) == 0 { + return nil + } + choice := resp.Choices[0] + mk := func(delta aiwire.Delta, finish *string) aiwire.ChatChunk { + return aiwire.ChatChunk{ID: resp.ID, Object: "chat.completion.chunk", Created: resp.Created, + Model: resp.Model, Choices: []aiwire.ChunkChoice{{Index: 0, Delta: delta, FinishReason: finish}}} + } + delta := aiwire.Delta{Role: "assistant", Content: choice.Message.Content.Text} + for i, tc := range choice.Message.ToolCalls { + delta.ToolCalls = append(delta.ToolCalls, aiwire.ToolCallDelta{Index: i, ID: tc.ID, + Type: "function", Function: aiwire.FunctionCallDelta{Name: tc.Function.Name, Arguments: tc.Function.Arguments}}) + } + finish := choice.FinishReason + chunks := []aiwire.ChatChunk{mk(delta, nil), mk(aiwire.Delta{}, &finish)} + if includeUsage { + usage := resp.Usage + if usage == nil { + usage = &aiwire.Usage{} + } + chunks = append(chunks, aiwire.ChatChunk{ID: resp.ID, Object: "chat.completion.chunk", + Created: resp.Created, Model: resp.Model, Choices: []aiwire.ChunkChoice{}, Usage: usage}) + } + return chunks +}