diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index 3d4565c29..3fc62eb24 100644 --- a/backend/internal/handler/auth.go +++ b/backend/internal/handler/auth.go @@ -715,8 +715,10 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, modelName := proxyRequestModelForPath(path, c.GetHeader("Content-Type"), body) protocol := model.ChannelInterfaceType("") capability := "text" + var channelModel *model.ChannelModel if !(c.Request.Method == http.MethodGet && path == "/models") { - channelModel, modelErr := svc.SystemChannelModel(channel.ID, modelName) + var modelErr error + channelModel, modelErr = svc.SystemChannelModel(channel.ID, modelName) if modelErr != nil || channelModel.Protocol == "" { fail(c, http.StatusForbidden, errors.New("当前系统渠道未授权该模型或模型协议尚未配置")) return @@ -728,6 +730,13 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, fail(c, http.StatusForbidden, err) return } + if channelModel != nil && channelModel.BillingMode == "token" && protocol == model.ChannelInterfaceChatCompletion { + body, err = service.EnsureChatCompletionStreamUsageRequest(body) + if err != nil { + fail(c, http.StatusBadRequest, err) + return + } + } billingOrderID := "" query := c.Request.URL.Query() for _, key := range []string{"key", "api_key", "access_token", "token"} { diff --git a/backend/internal/service/analytics.go b/backend/internal/service/analytics.go index 581a099c4..62abd9f18 100644 --- a/backend/internal/service/analytics.go +++ b/backend/internal/service/analytics.go @@ -1,6 +1,7 @@ package service import ( + "bufio" "bytes" "encoding/csv" "encoding/json" @@ -939,13 +940,16 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { if log.ProviderRequestID == "" { log.ProviderRequestID = providerRequestIDFromPath(log.Path) } - if len(responseBody) == 0 || !json.Valid(responseBody) { + payloads := providerResponsePayloads(responseBody) + if len(payloads) == 0 { return } - var payload map[string]any - if json.Unmarshal(responseBody, &payload) != nil { - return + for _, payload := range payloads { + s.enrichAPICallLogPayload(log, payload) } +} + +func (s *Service) enrichAPICallLogPayload(log *model.ApiCallLog, payload map[string]any) { if data, ok := payload["data"].(map[string]any); ok { for key, value := range data { if _, exists := payload[key]; !exists { @@ -953,6 +957,14 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { } } } + // Responses API 的终态 SSE 把实际响应(包括 usage)放在 response 字段中。 + if response, ok := payload["response"].(map[string]any); ok { + for key, value := range response { + if _, exists := payload[key]; !exists { + payload[key] = value + } + } + } if log.Status == model.ApiCallStatusFailed { errorCode, errorMessage := providerFailureDetails(payload) log.ErrorCode = errorCode @@ -962,9 +974,17 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { } usage, _ := payload["usage"].(map[string]any) if usage != nil { - log.UsageAvailable = true - log.InputTokens = firstInt64(usage, "input_tokens", "prompt_tokens") - log.OutputTokens = firstInt64(usage, "output_tokens", "completion_tokens") + inputTokens, inputAvailable := firstInt64Value(usage, "input_tokens", "prompt_tokens") + outputTokens, outputAvailable := firstInt64Value(usage, "output_tokens", "completion_tokens") + if inputAvailable { + log.InputTokens = inputTokens + } + if outputAvailable { + log.OutputTokens = outputTokens + } + if inputAvailable || outputAvailable { + log.UsageAvailable = true + } // 火山方舟视频的输入 Token 恒为 0;查询任务以 completion_tokens 为实际用量, // 兼容只返回 total_tokens 的同协议中转实现。 if log.Capability == "video" && strings.Contains(log.Path, "/contents/generations/tasks") && log.OutputTokens == 0 { @@ -981,9 +1001,17 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { } } if usageMetadata, ok := payload["usageMetadata"].(map[string]any); ok { - log.UsageAvailable = true - log.InputTokens = firstInt64(usageMetadata, "promptTokenCount") - log.OutputTokens = firstInt64(usageMetadata, "candidatesTokenCount") + inputTokens, inputAvailable := firstInt64Value(usageMetadata, "promptTokenCount") + outputTokens, outputAvailable := firstInt64Value(usageMetadata, "candidatesTokenCount") + if inputAvailable { + log.InputTokens = inputTokens + } + if outputAvailable { + log.OutputTokens = outputTokens + } + if inputAvailable || outputAvailable { + log.UsageAvailable = true + } log.CachedTokens = firstInt64(usageMetadata, "cachedContentTokenCount") } log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id"), stringField(payload, "name"), log.ProviderRequestID) @@ -1003,6 +1031,45 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { } } +func providerResponsePayloads(responseBody []byte) []map[string]any { + if len(responseBody) == 0 { + return nil + } + var payload map[string]any + if json.Unmarshal(responseBody, &payload) == nil { + return []map[string]any{payload} + } + + // 流式文本的用量只出现在最后一个 SSE data 事件中,不能把整段响应当作 JSON。 + result := make([]map[string]any, 0) + scanner := bufio.NewScanner(bytes.NewReader(responseBody)) + scanner.Buffer(make([]byte, 64<<10), max(len(responseBody)+1, 64<<10)) + dataLines := make([]string, 0, 1) + flush := func() { + raw := strings.TrimSpace(strings.Join(dataLines, "\n")) + dataLines = dataLines[:0] + if raw == "" || raw == "[DONE]" { + return + } + var event map[string]any + if json.Unmarshal([]byte(raw), &event) == nil { + result = append(result, event) + } + } + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + flush() + continue + } + if strings.HasPrefix(line, "data:") { + dataLines = append(dataLines, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ")) + } + } + flush() + return result +} + func providerRequestIDFromPath(path string) string { parts := strings.Split(strings.Trim(strings.TrimSpace(path), "/"), "/") for index := len(parts) - 1; index >= 0; index-- { @@ -1019,18 +1086,23 @@ func providerRequestIDFromPath(path string) string { } func firstInt64(values map[string]any, keys ...string) int64 { + value, _ := firstInt64Value(values, keys...) + return value +} + +func firstInt64Value(values map[string]any, keys ...string) (int64, bool) { for _, key := range keys { switch value := values[key].(type) { case float64: - return int64(value) + return int64(value), true case int64: - return value + return value, true case json.Number: - parsed, _ := value.Int64() - return parsed + parsed, err := value.Int64() + return parsed, err == nil } } - return 0 + return 0, false } func (s *Service) recordActivity(userID string, event string, count int) { diff --git a/backend/internal/service/finance_token_test.go b/backend/internal/service/finance_token_test.go index a83bbab26..b222b4724 100644 --- a/backend/internal/service/finance_token_test.go +++ b/backend/internal/service/finance_token_test.go @@ -1,6 +1,7 @@ package service import ( + "encoding/json" "testing" "infinite-canvas/backend/internal/model" @@ -83,3 +84,49 @@ func TestEnrichAPICallLogReadsArkVideoUsage(t *testing.T) { t.Fatalf("EnrichAPICallLog() accepted empty Ark usage: %#v", missing) } } + +func TestEnrichAPICallLogReadsChatCompletionStreamUsage(t *testing.T) { + log := &model.ApiCallLog{Capability: "text", Path: "/v1/chat/completions"} + stream := []byte("data: {\"id\":\"chatcmpl-test\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n" + + "data: {\"id\":\"chatcmpl-test\",\"choices\":[],\"usage\":{\"prompt_tokens\":23,\"completion_tokens\":7,\"prompt_tokens_details\":{\"cached_tokens\":5}}}\n\n" + + "data: [DONE]\n\n") + (&Service{}).EnrichAPICallLog(log, stream) + if !log.UsageAvailable || log.InputTokens != 23 || log.OutputTokens != 7 || log.CachedTokens != 5 { + t.Fatalf("EnrichAPICallLog() = %#v", log) + } +} + +func TestEnrichAPICallLogReadsResponsesStreamUsage(t *testing.T) { + log := &model.ApiCallLog{Capability: "text", Path: "/v1/responses"} + stream := []byte("event: response.output_text.delta\n" + + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"ok\"}\n\n" + + "event: response.completed\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-test\",\"status\":\"completed\",\"usage\":{\"input_tokens\":31,\"output_tokens\":11,\"input_tokens_details\":{\"cached_tokens\":9}}}}\n\n") + (&Service{}).EnrichAPICallLog(log, stream) + if !log.UsageAvailable || log.InputTokens != 31 || log.OutputTokens != 11 || log.CachedTokens != 9 || log.ProviderRequestID != "resp-test" { + t.Fatalf("EnrichAPICallLog() = %#v", log) + } +} + +func TestEnrichAPICallLogRejectsEmptyTextUsage(t *testing.T) { + log := &model.ApiCallLog{Capability: "text", Path: "/v1/chat/completions"} + (&Service{}).EnrichAPICallLog(log, []byte(`{"usage":{}}`)) + if log.UsageAvailable { + t.Fatalf("EnrichAPICallLog() accepted empty text usage: %#v", log) + } +} + +func TestEnsureChatCompletionStreamUsageRequest(t *testing.T) { + data, err := EnsureChatCompletionStreamUsageRequest([]byte(`{"model":"gpt-test","stream":true,"stream_options":{"custom":true}}`)) + if err != nil { + t.Fatalf("EnsureChatCompletionStreamUsageRequest() error = %v", err) + } + var payload map[string]any + if err := json.Unmarshal(data, &payload); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + options, _ := payload["stream_options"].(map[string]any) + if options["include_usage"] != true || options["custom"] != true { + t.Fatalf("stream_options = %#v", options) + } +} diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index f6b71360f..11866b29d 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -113,6 +113,7 @@ type providerAnalyticsContext struct { UserID string TaskID string BillingOrderID string + BillingMode string Capability string Operation string ChannelID string @@ -125,6 +126,12 @@ type providerAnalyticsContext struct { func withProviderAnalytics(ctx context.Context, service *Service, task model.Task) context.Context { metadata := providerAnalyticsContext{Service: service, UserID: task.UserID, TaskID: task.ID, BillingOrderID: task.BillingOrderID, Capability: capabilityFromTaskType(task.Type), Operation: task.Operation, Model: task.Model, ProviderRequestID: task.ProviderRequestID} + // 账单模式随请求上下文传递,流式协议据此只为 Token 计费开启 usage 终态块。 + if service != nil && task.BillingOrderID != "" { + if order, err := service.repo.BillingOrder(task.BillingOrderID); err == nil { + metadata.BillingMode = order.BillingMode + } + } var input struct { Mode string `json:"mode"` Config providerConfig `json:"config"` @@ -2417,6 +2424,12 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool) (string, error) { if stream { + metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext) + if protocol == "chat-completion" && metadata.BillingMode == "token" { + if err := ensureChatCompletionStreamUsage(body); err != nil { + return "", err + } + } return postStreamingText(ctx, config, path, body, protocol) } var payload map[string]interface{} diff --git a/backend/internal/service/text_stream_usage.go b/backend/internal/service/text_stream_usage.go new file mode 100644 index 000000000..501f9027d --- /dev/null +++ b/backend/internal/service/text_stream_usage.go @@ -0,0 +1,37 @@ +package service + +import ( + "encoding/json" + "errors" +) + +// EnsureChatCompletionStreamUsageRequest 为 Token 计费的流式 Chat Completions +// 请求开启最终 usage 块;非流式请求保持原样,避免改变普通接口合同。 +func EnsureChatCompletionStreamUsageRequest(data []byte) ([]byte, error) { + var payload map[string]any + if err := json.Unmarshal(data, &payload); err != nil { + return nil, err + } + stream, _ := payload["stream"].(bool) + if !stream { + return data, nil + } + if err := ensureChatCompletionStreamUsage(payload); err != nil { + return nil, err + } + return json.Marshal(payload) +} + +func ensureChatCompletionStreamUsage(payload map[string]any) error { + options := map[string]any{} + if value, exists := payload["stream_options"]; exists { + var ok bool + options, ok = value.(map[string]any) + if !ok { + return errors.New("stream_options 必须是 JSON 对象") + } + } + options["include_usage"] = true + payload["stream_options"] = options + return nil +}