wangdefa/oci-portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 18 Wiki Activity

修复 AI 网关流式断流:错误透传、自动降级,max_tokens 可缺省
CI / test (push) Successful in 33s
Details
Release / release (push) Successful in 56s
Details

Browse Source
This commit is contained in:
wangdefa
2026-07-13 15:06:01 +08:00
parent 81d4650f3d
commit 9309ad1ffc
11 changed files with 420 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/aigateway.go
+111 -10
View File
@@ -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
}
}
}