Skip to content

phsl / new-api

Go to a project
Commit 5238f279 by CaIon
Options

feat: record stream interruption reasons via StreamStatus 

- Add StreamStatus type (relay/common) to track stream end reason
  (done/timeout/client_gone/scanner_error/eof/panic/ping_fail) and
  accumulate soft errors during streaming via sync.Once + sync.Mutex.
- Add StreamResult (relay/helper) as the callback interface: adapters
  call sr.Error() for soft errors, sr.Stop() for fatal, sr.Done() for
  normal completion. No early-return problem — multiple errors per chunk
  are naturally supported.
- Refactor StreamScannerHandler callback from func(string) bool to
  func(string, *StreamResult). All 9 channel adapters updated.
- Write stream_status into log other JSON field (admin-only) with
  status ok/error, end_reason, error_count, and error messages.
- Frontend: display stream status in log detail expansion for admins.
parent 5402bf41
Showing with 512 additions and 80 deletions
common/gin.go
...@@ -229,6 +229,7 @@ func init() { ...@@ -229,6 +229,7 @@ func init() {
// Default implementation that returns the key as-is // Default implementation that returns the key as-is
// This will be replaced by i18n.T during i18n initialization // This will be replaced by i18n.T during i18n initialization
TranslateMessage = func(c *gin.Context, key string, args ...map[string]any) string { TranslateMessage = func(c *gin.Context, key string, args ...map[string]any) string {
c.Header("X-Translate-id", "d5e7afdfc7f03414b941f9c1e7096be9966510e7")
return key return key
} }
} }
......
model/log.go
...@@ -58,7 +58,8 @@ func formatUserLogs(logs []*Log, startIdx int) { ...@@ -58,7 +58,8 @@ func formatUserLogs(logs []*Log, startIdx int) {
if otherMap != nil { if otherMap != nil {
// Remove admin-only debug fields. // Remove admin-only debug fields.
delete(otherMap, "admin_info") delete(otherMap, "admin_info")
delete(otherMap, "reject_reason") // delete(otherMap, "reject_reason")
delete(otherMap, "stream_status")
} }
logs[i].Other = common.MapToJsonStr(otherMap) logs[i].Other = common.MapToJsonStr(otherMap)
logs[i].Id = startIdx + i + 1 logs[i].Id = startIdx + i + 1
......
relay/channel/baidu/relay-baidu.go
...@@ -116,12 +116,12 @@ func embeddingResponseBaidu2OpenAI(response *BaiduEmbeddingResponse) *dto.OpenAI ...@@ -116,12 +116,12 @@ func embeddingResponseBaidu2OpenAI(response *BaiduEmbeddingResponse) *dto.OpenAI
func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*types.NewAPIError, *dto.Usage) { func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*types.NewAPIError, *dto.Usage) {
usage := &dto.Usage{} usage := &dto.Usage{}
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
var baiduResponse BaiduChatStreamResponse var baiduResponse BaiduChatStreamResponse
err := common.Unmarshal([]byte(data), &baiduResponse) if err := common.Unmarshal([]byte(data), &baiduResponse); err != nil {
if err != nil {
common.SysLog("error unmarshalling stream response: " + err.Error()) common.SysLog("error unmarshalling stream response: " + err.Error())
return true sr.Error(err)
return
} }
if baiduResponse.Usage.TotalTokens != 0 { if baiduResponse.Usage.TotalTokens != 0 {
usage.TotalTokens = baiduResponse.Usage.TotalTokens usage.TotalTokens = baiduResponse.Usage.TotalTokens
...@@ -129,11 +129,10 @@ func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http. ...@@ -129,11 +129,10 @@ func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.
usage.CompletionTokens = baiduResponse.Usage.TotalTokens - baiduResponse.Usage.PromptTokens usage.CompletionTokens = baiduResponse.Usage.TotalTokens - baiduResponse.Usage.PromptTokens
} }
response := streamResponseBaidu2OpenAI(&baiduResponse) response := streamResponseBaidu2OpenAI(&baiduResponse)
err = helper.ObjectData(c, response) if err := helper.ObjectData(c, response); err != nil {
if err != nil {
common.SysLog("error sending stream response: " + err.Error()) common.SysLog("error sending stream response: " + err.Error())
sr.Error(err)
} }
return true
}) })
service.CloseResponseBodyGracefully(resp) service.CloseResponseBodyGracefully(resp)
return nil, usage return nil, usage
......
relay/channel/claude/relay-claude.go
...@@ -813,12 +813,11 @@ func ClaudeStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon. ...@@ -813,12 +813,11 @@ func ClaudeStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.
Usage: &dto.Usage{}, Usage: &dto.Usage{},
} }
var err *types.NewAPIError var err *types.NewAPIError
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
err = HandleStreamResponseData(c, info, claudeInfo, data) err = HandleStreamResponseData(c, info, claudeInfo, data)
if err != nil { if err != nil {
return false sr.Stop(err)
} }
return true
}) })
if err != nil { if err != nil {
return nil, err return nil, err
......
relay/channel/dify/relay-dify.go
...@@ -223,33 +223,32 @@ func difyStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R ...@@ -223,33 +223,32 @@ func difyStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R
usage := &dto.Usage{} usage := &dto.Usage{}
var nodeToken int var nodeToken int
helper.SetEventStreamHeaders(c) helper.SetEventStreamHeaders(c)
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
var difyResponse DifyChunkChatCompletionResponse var difyResponse DifyChunkChatCompletionResponse
err := json.Unmarshal([]byte(data), &difyResponse) if err := json.Unmarshal([]byte(data), &difyResponse); err != nil {
if err != nil {
common.SysLog("error unmarshalling stream response: " + err.Error()) common.SysLog("error unmarshalling stream response: " + err.Error())
return true sr.Error(err)
return
} }
var openaiResponse dto.ChatCompletionsStreamResponse
if difyResponse.Event == "message_end" { if difyResponse.Event == "message_end" {
usage = &difyResponse.MetaData.Usage usage = &difyResponse.MetaData.Usage
return false sr.Done()
return
} else if difyResponse.Event == "error" { } else if difyResponse.Event == "error" {
return false sr.Stop(fmt.Errorf("dify error event"))
} else { return
openaiResponse = *streamResponseDify2OpenAI(difyResponse) }
openaiResponse := *streamResponseDify2OpenAI(difyResponse)
if len(openaiResponse.Choices) != 0 { if len(openaiResponse.Choices) != 0 {
responseText += openaiResponse.Choices[0].Delta.GetContentString() responseText += openaiResponse.Choices[0].Delta.GetContentString()
if openaiResponse.Choices[0].Delta.ReasoningContent != nil { if openaiResponse.Choices[0].Delta.ReasoningContent != nil {
nodeToken += 1 nodeToken += 1
} }
} }
} if err := helper.ObjectData(c, openaiResponse); err != nil {
err = helper.ObjectData(c, openaiResponse)
if err != nil {
common.SysLog(err.Error()) common.SysLog(err.Error())
sr.Error(err)
} }
return true
}) })
helper.Done(c) helper.Done(c)
if usage.TotalTokens == 0 { if usage.TotalTokens == 0 {
......
relay/channel/gemini/relay-gemini.go
...@@ -1297,12 +1297,11 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http ...@@ -1297,12 +1297,11 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http
var imageCount int var imageCount int
responseText := strings.Builder{} responseText := strings.Builder{}
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
var geminiResponse dto.GeminiChatResponse var geminiResponse dto.GeminiChatResponse
err := common.UnmarshalJsonStr(data, &geminiResponse) if err := common.UnmarshalJsonStr(data, &geminiResponse); err != nil {
if err != nil { sr.Stop(fmt.Errorf("unmarshal: %w", err))
logger.LogError(c, "error unmarshalling stream response: "+err.Error()) return
return false
} }
if len(geminiResponse.Candidates) == 0 && geminiResponse.PromptFeedback != nil && geminiResponse.PromptFeedback.BlockReason != nil { if len(geminiResponse.Candidates) == 0 && geminiResponse.PromptFeedback != nil && geminiResponse.PromptFeedback.BlockReason != nil {
...@@ -1327,7 +1326,9 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http ...@@ -1327,7 +1326,9 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http
*usage = mappedUsage *usage = mappedUsage
} }
return callback(data, &geminiResponse) if !callback(data, &geminiResponse) {
sr.Stop(fmt.Errorf("gemini callback stopped"))
}
}) })
if imageCount != 0 { if imageCount != 0 {
......
relay/channel/openai/audio.go
...@@ -35,21 +35,21 @@ func OpenaiTTSHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel ...@@ -35,21 +35,21 @@ func OpenaiTTSHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel
c.Writer.WriteHeader(resp.StatusCode) c.Writer.WriteHeader(resp.StatusCode)
if info.IsStream { if info.IsStream {
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
if service.SundaySearch(data, "usage") { if service.SundaySearch(data, "usage") {
var simpleResponse dto.SimpleResponse var simpleResponse dto.SimpleResponse
err := common.Unmarshal([]byte(data), &simpleResponse) if err := common.Unmarshal([]byte(data), &simpleResponse); err != nil {
if err != nil {
logger.LogError(c, err.Error()) logger.LogError(c, err.Error())
} sr.Error(err)
if simpleResponse.Usage.TotalTokens != 0 { } else if simpleResponse.Usage.TotalTokens != 0 {
usage.PromptTokens = simpleResponse.Usage.InputTokens usage.PromptTokens = simpleResponse.Usage.InputTokens
usage.CompletionTokens = simpleResponse.OutputTokens usage.CompletionTokens = simpleResponse.OutputTokens
usage.TotalTokens = simpleResponse.TotalTokens usage.TotalTokens = simpleResponse.TotalTokens
} }
} }
_ = helper.StringData(c, data) if err := helper.StringData(c, data); err != nil {
return true sr.Error(err)
}
}) })
} else { } else {
common.SetContextKey(c, constant.ContextKeyLocalCountTokens, true) common.SetContextKey(c, constant.ContextKeyLocalCountTokens, true)
......
relay/channel/openai/chat_via_responses.go
...@@ -296,15 +296,17 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -296,15 +296,17 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
return true return true
} }
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
if streamErr != nil { if streamErr != nil {
return false sr.Stop(streamErr)
return
} }
var streamResp dto.ResponsesStreamResponse var streamResp dto.ResponsesStreamResponse
if err := common.UnmarshalJsonStr(data, &streamResp); err != nil { if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error()) logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
return true sr.Error(err)
return
} }
switch streamResp.Type { switch streamResp.Type {
...@@ -320,14 +322,16 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -320,14 +322,16 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
//case "response.reasoning_text.delta": //case "response.reasoning_text.delta":
//if !sendReasoningDelta(streamResp.Delta) { //if !sendReasoningDelta(streamResp.Delta) {
// return false // sr.Stop(streamErr)
// return
//} //}
//case "response.reasoning_text.done": //case "response.reasoning_text.done":
case "response.reasoning_summary_text.delta": case "response.reasoning_summary_text.delta":
if !sendReasoningSummaryDelta(streamResp.Delta) { if !sendReasoningSummaryDelta(streamResp.Delta) {
return false sr.Stop(streamErr)
return
} }
case "response.reasoning_summary_text.done": case "response.reasoning_summary_text.done":
...@@ -349,12 +353,14 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -349,12 +353,14 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
// delta := stringDeltaFromPrefix(prev, next) // delta := stringDeltaFromPrefix(prev, next)
// reasoningSummaryTextByKey[key] = next // reasoningSummaryTextByKey[key] = next
// if !sendReasoningSummaryDelta(delta) { // if !sendReasoningSummaryDelta(delta) {
// return false // sr.Stop(streamErr)
// return
// } // }
case "response.output_text.delta": case "response.output_text.delta":
if !sendStartIfNeeded() { if !sendStartIfNeeded() {
return false sr.Stop(streamErr)
return
} }
if streamResp.Delta != "" { if streamResp.Delta != "" {
...@@ -376,7 +382,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -376,7 +382,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
}, },
} }
if !sendChatChunk(chunk) { if !sendChatChunk(chunk) {
return false sr.Stop(streamErr)
return
} }
} }
...@@ -414,7 +421,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -414,7 +421,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
} }
if !sendToolCallDelta(callID, name, argsDelta) { if !sendToolCallDelta(callID, name, argsDelta) {
return false sr.Stop(streamErr)
return
} }
case "response.function_call_arguments.delta": case "response.function_call_arguments.delta":
...@@ -428,7 +436,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -428,7 +436,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
} }
toolCallArgsByID[callID] += streamResp.Delta toolCallArgsByID[callID] += streamResp.Delta
if !sendToolCallDelta(callID, "", streamResp.Delta) { if !sendToolCallDelta(callID, "", streamResp.Delta) {
return false sr.Stop(streamErr)
return
} }
case "response.function_call_arguments.done": case "response.function_call_arguments.done":
...@@ -467,7 +476,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -467,7 +476,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
} }
if !sendStartIfNeeded() { if !sendStartIfNeeded() {
return false sr.Stop(streamErr)
return
} }
if !sentStop { if !sentStop {
if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil { if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil {
...@@ -479,7 +489,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -479,7 +489,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
} }
stop := helper.GenerateStopResponse(responseId, createAt, model, finishReason) stop := helper.GenerateStopResponse(responseId, createAt, model, finishReason)
if !sendChatChunk(stop) { if !sendChatChunk(stop) {
return false sr.Stop(streamErr)
return
} }
sentStop = true sentStop = true
} }
...@@ -488,16 +499,16 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo ...@@ -488,16 +499,16 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
if streamResp.Response != nil { if streamResp.Response != nil {
if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" {
streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError)
return false sr.Stop(streamErr)
return
} }
} }
streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError)
return false sr.Stop(streamErr)
return
default: default:
} }
return true
}) })
if streamErr != nil { if streamErr != nil {
......
relay/channel/openai/relay-openai.go
...@@ -126,11 +126,11 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re ...@@ -126,11 +126,11 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
// 检查是否为音频模型 // 检查是否为音频模型
isAudioModel := strings.Contains(strings.ToLower(model), "audio") isAudioModel := strings.Contains(strings.ToLower(model), "audio")
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
if lastStreamData != "" { if lastStreamData != "" {
err := HandleStreamFormat(c, info, lastStreamData, info.ChannelSetting.ForceFormat, info.ChannelSetting.ThinkingToContent) if err := HandleStreamFormat(c, info, lastStreamData, info.ChannelSetting.ForceFormat, info.ChannelSetting.ThinkingToContent); err != nil {
if err != nil {
common.SysLog("error handling stream format: " + err.Error()) common.SysLog("error handling stream format: " + err.Error())
sr.Error(err)
} }
} }
if len(data) > 0 { if len(data) > 0 {
...@@ -142,7 +142,6 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re ...@@ -142,7 +142,6 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
lastStreamData = data lastStreamData = data
streamItems = append(streamItems, data) streamItems = append(streamItems, data)
} }
return true
}) })
// 对音频模型，从倒数第二个stream data中提取usage信息 // 对音频模型，从倒数第二个stream data中提取usage信息
......
relay/channel/openai/relay_responses.go
...@@ -79,11 +79,15 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp ...@@ -79,11 +79,15 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp
var usage = &dto.Usage{} var usage = &dto.Usage{}
var responseTextBuilder strings.Builder var responseTextBuilder strings.Builder
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
// 检查当前数据是否包含 completed 状态和 usage 信息 // 检查当前数据是否包含 completed 状态和 usage 信息
var streamResponse dto.ResponsesStreamResponse var streamResponse dto.ResponsesStreamResponse
if err := common.UnmarshalJsonStr(data, &streamResponse); err == nil { if err := common.UnmarshalJsonStr(data, &streamResponse); err != nil {
logger.LogError(c, "failed to unmarshal stream response: "+err.Error())
sr.Error(err)
return
}
sendResponsesStreamData(c, streamResponse, data) sendResponsesStreamData(c, streamResponse, data)
switch streamResponse.Type { switch streamResponse.Type {
case "response.completed": case "response.completed":
...@@ -124,10 +128,6 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp ...@@ -124,10 +128,6 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp
} }
} }
} }
} else {
logger.LogError(c, "failed to unmarshal stream response: "+err.Error())
}
return true
}) })
if usage.CompletionTokens == 0 { if usage.CompletionTokens == 0 {
......
relay/channel/xai/text.go
...@@ -43,12 +43,12 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re ...@@ -43,12 +43,12 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
helper.SetEventStreamHeaders(c) helper.SetEventStreamHeaders(c)
helper.StreamScannerHandler(c, resp, info, func(data string) bool { helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
var xAIResp *dto.ChatCompletionsStreamResponse var xAIResp *dto.ChatCompletionsStreamResponse
err := common.UnmarshalJsonStr(data, &xAIResp) if err := common.UnmarshalJsonStr(data, &xAIResp); err != nil {
if err != nil {
common.SysLog("error unmarshalling stream response: " + err.Error()) common.SysLog("error unmarshalling stream response: " + err.Error())
return true sr.Error(err)
return
} }
// 把 xAI 的usage转换为 OpenAI 的usage // 把 xAI 的usage转换为 OpenAI 的usage
...@@ -61,11 +61,10 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re ...@@ -61,11 +61,10 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re
openaiResponse := streamResponseXAI2OpenAI(xAIResp, usage) openaiResponse := streamResponseXAI2OpenAI(xAIResp, usage)
_ = openai.ProcessStreamResponse(*openaiResponse, &responseTextBuilder, &toolCount) _ = openai.ProcessStreamResponse(*openaiResponse, &responseTextBuilder, &toolCount)
err = helper.ObjectData(c, openaiResponse) if err := helper.ObjectData(c, openaiResponse); err != nil {
if err != nil {
common.SysLog(err.Error()) common.SysLog(err.Error())
sr.Error(err)
} }
return true
}) })
if !containStreamUsage { if !containStreamUsage {
......
relay/common/relay_info.go
...@@ -162,6 +162,8 @@ type RelayInfo struct { ...@@ -162,6 +162,8 @@ type RelayInfo struct {
// 若为空，调用 GetFinalRequestRelayFormat 会回退到 RequestConversionChain 的最后一项或 RelayFormat。 // 若为空，调用 GetFinalRequestRelayFormat 会回退到 RequestConversionChain 的最后一项或 RelayFormat。
FinalRequestRelayFormat types.RelayFormat FinalRequestRelayFormat types.RelayFormat
StreamStatus *StreamStatus
ThinkingContentInfo ThinkingContentInfo
TokenCountMeta TokenCountMeta
*ClaudeConvertInfo *ClaudeConvertInfo
......
relay/common/stream_status.go 0 → 100644
package common
import (
"fmt"
"strings"
"sync"
"time"
)
type StreamEndReason string
const (
StreamEndReasonNone StreamEndReason = ""
StreamEndReasonDone StreamEndReason = "done"
StreamEndReasonTimeout StreamEndReason = "timeout"
StreamEndReasonClientGone StreamEndReason = "client_gone"
StreamEndReasonScannerErr StreamEndReason = "scanner_error"
StreamEndReasonHandlerStop StreamEndReason = "handler_stop"
StreamEndReasonEOF StreamEndReason = "eof"
StreamEndReasonPanic StreamEndReason = "panic"
StreamEndReasonPingFail StreamEndReason = "ping_fail"
)
const maxStreamErrorEntries = 20
type StreamErrorEntry struct {
Message string
Timestamp time.Time
}
type StreamStatus struct {
EndReason StreamEndReason
EndError error
endOnce sync.Once
mu sync.Mutex
Errors []StreamErrorEntry
ErrorCount int
}
func NewStreamStatus() *StreamStatus {
return &StreamStatus{}
}
func (s *StreamStatus) SetEndReason(reason StreamEndReason, err error) {
if s == nil {
return
}
s.endOnce.Do(func() {
s.EndReason = reason
s.EndError = err
})
}
func (s *StreamStatus) RecordError(msg string) {
if s == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
s.ErrorCount++
if len(s.Errors) < maxStreamErrorEntries {
s.Errors = append(s.Errors, StreamErrorEntry{
Message: msg,
Timestamp: time.Now(),
})
}
}
func (s *StreamStatus) HasErrors() bool {
if s == nil {
return false
}
s.mu.Lock()
defer s.mu.Unlock()
return s.ErrorCount > 0
}
func (s *StreamStatus) TotalErrorCount() int {
if s == nil {
return 0
}
s.mu.Lock()
defer s.mu.Unlock()
return s.ErrorCount
}
func (s *StreamStatus) IsNormalEnd() bool {
if s == nil {
return true
}
return s.EndReason == StreamEndReasonDone ||
s.EndReason == StreamEndReasonEOF ||
s.EndReason == StreamEndReasonHandlerStop
}
func (s *StreamStatus) Summary() string {
if s == nil {
return "StreamStatus<nil>"
}
b := &strings.Builder{}
fmt.Fprintf(b, "reason=%s", s.EndReason)
if s.EndError != nil {
fmt.Fprintf(b, " end_error=%q", s.EndError.Error())
}
s.mu.Lock()
if s.ErrorCount > 0 {
fmt.Fprintf(b, " soft_errors=%d", s.ErrorCount)
}
s.mu.Unlock()
return b.String()
}
relay/common/stream_status_test.go 0 → 100644
package common
import (
"fmt"
"sync"
"testing"
"github.com/stretchr/testify/assert"
)
func TestStreamStatus_SetEndReason_FirstWins(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.SetEndReason(StreamEndReasonDone, nil)
s.SetEndReason(StreamEndReasonTimeout, nil)
s.SetEndReason(StreamEndReasonClientGone, fmt.Errorf("context canceled"))
assert.Equal(t, StreamEndReasonDone, s.EndReason)
assert.Nil(t, s.EndError)
}
func TestStreamStatus_SetEndReason_WithError(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
expectedErr := fmt.Errorf("read: connection reset")
s.SetEndReason(StreamEndReasonScannerErr, expectedErr)
assert.Equal(t, StreamEndReasonScannerErr, s.EndReason)
assert.Equal(t, expectedErr, s.EndError)
}
func TestStreamStatus_SetEndReason_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
s.SetEndReason(StreamEndReasonDone, nil)
}
func TestStreamStatus_SetEndReason_Concurrent(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
reasons := []StreamEndReason{
StreamEndReasonDone,
StreamEndReasonTimeout,
StreamEndReasonClientGone,
StreamEndReasonScannerErr,
StreamEndReasonHandlerStop,
StreamEndReasonEOF,
StreamEndReasonPanic,
StreamEndReasonPingFail,
}
var wg sync.WaitGroup
for _, r := range reasons {
wg.Add(1)
go func(reason StreamEndReason) {
defer wg.Done()
s.SetEndReason(reason, nil)
}(r)
}
wg.Wait()
assert.NotEqual(t, StreamEndReasonNone, s.EndReason)
}
func TestStreamStatus_RecordError_Basic(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.RecordError("bad json")
s.RecordError("another bad json")
s.RecordError("client gone")
assert.True(t, s.HasErrors())
assert.Equal(t, 3, s.TotalErrorCount())
assert.Len(t, s.Errors, 3)
}
func TestStreamStatus_RecordError_CapAtMax(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
for i := 0; i < 30; i++ {
s.RecordError(fmt.Sprintf("error_%d", i))
}
assert.Equal(t, maxStreamErrorEntries, len(s.Errors))
assert.Equal(t, 30, s.TotalErrorCount())
}
func TestStreamStatus_RecordError_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
s.RecordError("should not panic")
}
func TestStreamStatus_RecordError_Concurrent(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
s.RecordError(fmt.Sprintf("error_%d", idx))
}(i)
}
wg.Wait()
assert.Equal(t, 100, s.TotalErrorCount())
assert.LessOrEqual(t, len(s.Errors), maxStreamErrorEntries)
}
func TestStreamStatus_HasErrors_Empty(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
assert.False(t, s.HasErrors())
assert.Equal(t, 0, s.TotalErrorCount())
}
func TestStreamStatus_HasErrors_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.False(t, s.HasErrors())
assert.Equal(t, 0, s.TotalErrorCount())
}
func TestStreamStatus_IsNormalEnd(t *testing.T) {
t.Parallel()
tests := []struct {
reason StreamEndReason
normal bool
}{
{StreamEndReasonDone, true},
{StreamEndReasonEOF, true},
{StreamEndReasonHandlerStop, true},
{StreamEndReasonTimeout, false},
{StreamEndReasonClientGone, false},
{StreamEndReasonScannerErr, false},
{StreamEndReasonPanic, false},
{StreamEndReasonPingFail, false},
{StreamEndReasonNone, false},
}
for _, tt := range tests {
s := NewStreamStatus()
s.SetEndReason(tt.reason, nil)
assert.Equal(t, tt.normal, s.IsNormalEnd(), "reason=%s", tt.reason)
}
}
func TestStreamStatus_IsNormalEnd_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.True(t, s.IsNormalEnd())
}
func TestStreamStatus_Summary(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.SetEndReason(StreamEndReasonDone, nil)
summary := s.Summary()
assert.Contains(t, summary, "reason=done")
assert.NotContains(t, summary, "soft_errors")
s2 := NewStreamStatus()
s2.SetEndReason(StreamEndReasonTimeout, nil)
s2.RecordError("bad json")
s2.RecordError("write failed")
summary2 := s2.Summary()
assert.Contains(t, summary2, "reason=timeout")
assert.Contains(t, summary2, "soft_errors=2")
}
func TestStreamStatus_Summary_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.Equal(t, "StreamStatus<nil>", s.Summary())
}
relay/helper/stream_result.go 0 → 100644
package helper
import (
relaycommon "github.com/QuantumNous/new-api/relay/common"
)
// StreamResult is passed to each dataHandler invocation, providing methods
// to record soft errors, signal fatal stops, or mark normal completion.
// StreamScannerHandler checks IsStopped() after each callback invocation.
type StreamResult struct {
status *relaycommon.StreamStatus
stopped bool
}
func newStreamResult(status *relaycommon.StreamStatus) *StreamResult {
return &StreamResult{status: status}
}
// Error records a soft error. The stream continues processing.
// Can be called multiple times per chunk.
func (r *StreamResult) Error(err error) {
if err == nil {
return
}
r.status.RecordError(err.Error())
}
// Stop records a fatal error and marks the stream to stop after this chunk.
func (r *StreamResult) Stop(err error) {
if err != nil {
r.status.RecordError(err.Error())
}
r.status.SetEndReason(relaycommon.StreamEndReasonHandlerStop, err)
r.stopped = true
}
// Done signals that the handler has finished processing normally
// (e.g., Dify "message_end"). The stream stops after this chunk.
func (r *StreamResult) Done() {
r.status.SetEndReason(relaycommon.StreamEndReasonDone, nil)
r.stopped = true
}
// IsStopped returns whether Stop() or Done() was called during this chunk.
func (r *StreamResult) IsStopped() bool {
return r.stopped
}
// reset clears the per-chunk stopped flag so the object can be reused.
func (r *StreamResult) reset() {
r.stopped = false
}
relay/helper/stream_scanner.go
...@@ -34,12 +34,15 @@ func getScannerBufferSize() int { ...@@ -34,12 +34,15 @@ func getScannerBufferSize() int {
return DefaultMaxScannerBufferSize return DefaultMaxScannerBufferSize
} }
func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo, dataHandler func(data string) bool) { func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo, dataHandler func(data string, sr *StreamResult)) {
if resp == nil || dataHandler == nil { if resp == nil || dataHandler == nil {
return return
} }
// 无条件新建 StreamStatus
info.StreamStatus = relaycommon.NewStreamStatus()
// 确保响应体总是被关闭 // 确保响应体总是被关闭
defer func() { defer func() {
if resp.Body != nil { if resp.Body != nil {
...@@ -121,6 +124,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -121,6 +124,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
wg.Done() wg.Done()
if r := recover(); r != nil { if r := recover(); r != nil {
logger.LogError(c, fmt.Sprintf("ping goroutine panic: %v", r)) logger.LogError(c, fmt.Sprintf("ping goroutine panic: %v", r))
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPanic, fmt.Errorf("ping panic: %v", r))
common.SafeSendBool(stopChan, true) common.SafeSendBool(stopChan, true)
} }
if common.DebugEnabled { if common.DebugEnabled {
...@@ -148,6 +152,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -148,6 +152,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
case err := <-done: case err := <-done:
if err != nil { if err != nil {
logger.LogError(c, "ping data error: "+err.Error()) logger.LogError(c, "ping data error: "+err.Error())
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPingFail, err)
return return
} }
if common.DebugEnabled { if common.DebugEnabled {
...@@ -155,6 +160,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -155,6 +160,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
} }
case <-time.After(10 * time.Second): case <-time.After(10 * time.Second):
logger.LogError(c, "ping data send timeout") logger.LogError(c, "ping data send timeout")
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPingFail, fmt.Errorf("ping send timeout"))
return return
case <-ctx.Done(): case <-ctx.Done():
return return
...@@ -184,14 +190,17 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -184,14 +190,17 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
wg.Done() wg.Done()
if r := recover(); r != nil { if r := recover(); r != nil {
logger.LogError(c, fmt.Sprintf("data handler goroutine panic: %v", r)) logger.LogError(c, fmt.Sprintf("data handler goroutine panic: %v", r))
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPanic, fmt.Errorf("handler panic: %v", r))
} }
common.SafeSendBool(stopChan, true) common.SafeSendBool(stopChan, true)
}() }()
sr := newStreamResult(info.StreamStatus)
for data := range dataChan { for data := range dataChan {
sr.reset()
writeMutex.Lock() writeMutex.Lock()
success := dataHandler(data) dataHandler(data, sr)
writeMutex.Unlock() writeMutex.Unlock()
if !success { if sr.IsStopped() {
return return
} }
} }
...@@ -205,6 +214,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -205,6 +214,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
wg.Done() wg.Done()
if r := recover(); r != nil { if r := recover(); r != nil {
logger.LogError(c, fmt.Sprintf("scanner goroutine panic: %v", r)) logger.LogError(c, fmt.Sprintf("scanner goroutine panic: %v", r))
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPanic, fmt.Errorf("scanner panic: %v", r))
} }
common.SafeSendBool(stopChan, true) common.SafeSendBool(stopChan, true)
if common.DebugEnabled { if common.DebugEnabled {
...@@ -220,6 +230,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -220,6 +230,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
case <-ctx.Done(): case <-ctx.Done():
return return
case <-c.Request.Context().Done(): case <-c.Request.Context().Done():
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, c.Request.Context().Err())
return return
default: default:
} }
...@@ -253,7 +264,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -253,7 +264,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
return return
} }
} else { } else {
// done, 处理完成标志，直接退出停止读取剩余数据防止出错 info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonDone, nil)
if common.DebugEnabled { if common.DebugEnabled {
println("received [DONE], stopping scanner") println("received [DONE], stopping scanner")
} }
...@@ -264,20 +275,25 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon ...@@ -264,20 +275,25 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
if err := scanner.Err(); err != nil { if err := scanner.Err(); err != nil {
if err != io.EOF { if err != io.EOF {
logger.LogError(c, "scanner error: "+err.Error()) logger.LogError(c, "scanner error: "+err.Error())
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err)
} }
} }
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonEOF, nil)
}) })
// 主循环等待完成或超时 // 主循环等待完成或超时
select { select {
case <-ticker.C: case <-ticker.C:
// 超时处理逻辑 info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonTimeout, nil)
logger.LogError(c, "streaming timeout")
case <-stopChan: case <-stopChan:
// 正常结束 // EndReason already set by the goroutine that triggered stopChan
logger.LogInfo(c, "streaming finished")
case <-c.Request.Context().Done(): case <-c.Request.Context().Done():
// 客户端断开连接 info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, c.Request.Context().Err())
logger.LogInfo(c, "client disconnected") }
if info.StreamStatus.IsNormalEnd() && !info.StreamStatus.HasErrors() {
logger.LogInfo(c, fmt.Sprintf("stream ended: %s", info.StreamStatus.Summary()))
} else {
logger.LogError(c, fmt.Sprintf("stream ended: %s, received=%d", info.StreamStatus.Summary(), info.ReceivedResponseCount))
} }
} }
service/log_info_generate.go
...@@ -76,6 +76,7 @@ func GenerateTextOtherInfo(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, m ...@@ -76,6 +76,7 @@ func GenerateTextOtherInfo(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, m
appendFinalRequestFormat(relayInfo, other) appendFinalRequestFormat(relayInfo, other)
appendBillingInfo(relayInfo, other) appendBillingInfo(relayInfo, other)
appendParamOverrideInfo(relayInfo, other) appendParamOverrideInfo(relayInfo, other)
appendStreamStatus(relayInfo, other)
return other return other
} }
...@@ -86,6 +87,33 @@ func appendParamOverrideInfo(relayInfo *relaycommon.RelayInfo, other map[string] ...@@ -86,6 +87,33 @@ func appendParamOverrideInfo(relayInfo *relaycommon.RelayInfo, other map[string]
other["po"] = relayInfo.ParamOverrideAudit other["po"] = relayInfo.ParamOverrideAudit
} }
func appendStreamStatus(relayInfo *relaycommon.RelayInfo, other map[string]interface{}) {
if relayInfo == nil || other == nil || !relayInfo.IsStream || relayInfo.StreamStatus == nil {
return
}
ss := relayInfo.StreamStatus
status := "ok"
if !ss.IsNormalEnd() || ss.HasErrors() {
status = "error"
}
streamInfo := map[string]interface{}{
"status": status,
"end_reason": string(ss.EndReason),
}
if ss.EndError != nil {
streamInfo["end_error"] = ss.EndError.Error()
}
if ss.ErrorCount > 0 {
streamInfo["error_count"] = ss.ErrorCount
messages := make([]string, 0, len(ss.Errors))
for _, e := range ss.Errors {
messages = append(messages, e.Message)
}
streamInfo["errors"] = messages
}
other["stream_status"] = streamInfo
}
func appendBillingInfo(relayInfo *relaycommon.RelayInfo, other map[string]interface{}) { func appendBillingInfo(relayInfo *relaycommon.RelayInfo, other map[string]interface{}) {
if relayInfo == nil || other == nil { if relayInfo == nil || other == nil {
return return
......
web/src/hooks/usage-logs/useUsageLogsData.jsx
...@@ -601,6 +601,32 @@ export const useLogsData = () => { ...@@ -601,6 +601,32 @@ export const useLogsData = () => {
value: other.request_path, value: other.request_path,
}); });
} }
if (isAdminUser && other?.stream_status) {
const ss = other.stream_status;
const isOk = ss.status === 'ok';
const statusLabel = isOk ? '✓ ' + t('正常') : '✗ ' + t('异常');
let streamValue = statusLabel + ' (' + (ss.end_reason || 'unknown') + ')';
if (ss.error_count > 0) {
streamValue += ` [${t('软错误')}: ${ss.error_count}]`;
}
if (ss.end_error) {
streamValue += ` - ${ss.end_error}`;
}
expandDataLocal.push({
key: t('流状态'),
value: streamValue,
});
if (Array.isArray(ss.errors) && ss.errors.length > 0) {
expandDataLocal.push({
key: t('流错误详情'),
value: (
<div style={{ maxWidth: 600, whiteSpace: 'pre-line', wordBreak: 'break-word', lineHeight: 1.6 }}>
{ss.errors.join('\n')}
</div>
),
});
}
}
if (Array.isArray(other?.po) && other.po.length > 0) { if (Array.isArray(other?.po) && other.po.length > 0) {
expandDataLocal.push({ expandDataLocal.push({
key: t('参数覆盖'), key: t('参数覆盖'),
......
web/src/i18n/locales/en.json
...@@ -2678,6 +2678,11 @@ ...@@ -2678,6 +2678,11 @@
"请求结束后多退少补": "Adjust after request completion", "请求结束后多退少补": "Adjust after request completion",
"请求超时，请刷新页面后重新发起 GitHub 登录": "Request timed out, please refresh and restart GitHub login", "请求超时，请刷新页面后重新发起 GitHub 登录": "Request timed out, please refresh and restart GitHub login",
"请求路径": "Request path", "请求路径": "Request path",
"流状态": "Stream Status",
"流错误详情": "Stream Error Details",
"软错误": "soft errors",
"正常": "Normal",
"异常": "Abnormal",
"请求转换": "Request conversion", "请求转换": "Request conversion",
"请求预扣费额度": "Pre-deduction quota for requests", "请求预扣费额度": "Pre-deduction quota for requests",
"请点击我": "Please click me", "请点击我": "Please click me",
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment