From d57a8694cfe17fd57a2030df06574748ab1d8e6e Mon Sep 17 00:00:00 2001 From: yepwait <1239738103@qq.com> Date: Fri, 7 Aug 2026 13:32:35 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E6=8F=90=E9=AB=98=E5=88=86=E9=95=9C?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=BB=98=E8=AE=A4=E8=B6=85=E6=97=B6=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E8=87=B315=E5=88=86=E9=92=9F=EF=BC=8C=E8=A7=A3?= =?UTF-8?q?=E5=86=B3=E8=85=BE=E8=AE=AF1=E7=8E=AF=E5=A2=83=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=9C=A82=E5=88=86=E9=92=9F=E5=86=85=E5=BF=AB?= =?UTF-8?q?=E9=80=9F=E5=A4=B1=E8=B4=A5=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/service/runtime_policy.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/backend/internal/service/runtime_policy.go b/backend/internal/service/runtime_policy.go index 1230eecce..90f7a7963 100644 --- a/backend/internal/service/runtime_policy.go +++ b/backend/internal/service/runtime_policy.go @@ -119,7 +119,8 @@ func defaultRuntimePolicy() RuntimePolicySetting { TextTimeoutMinutes: 8, AudioTimeoutMinutes: 8, VideoTimeoutMinutes: 30, - StoryboardTimeoutMinutes: 12, + StoryboardTimeoutMinutes: 15, + DefaultTimeoutMinutes: 10, }, Request: RuntimeRequestPolicy{ From 79d2ffe34de06aadea79336b4e3b5c938ac45797 Mon Sep 17 00:00:00 2001 From: yepwait <1239738103@qq.com> Date: Fri, 7 Aug 2026 13:40:18 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E6=8F=90=E9=AB=98=E5=88=86=E9=95=9C?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=BB=98=E8=AE=A4=E8=B6=85=E6=97=B6=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E8=87=B320=E5=88=86=E9=92=9F=EF=BC=8C=E8=A7=A3?= =?UTF-8?q?=E5=86=B3=E8=85=BE=E8=AE=AF1=E7=8E=AF=E5=A2=83=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=9C=A82=E5=88=86=E9=92=9F=E5=86=85=E5=BF=AB?= =?UTF-8?q?=E9=80=9F=E5=A4=B1=E8=B4=A5=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/service/runtime_policy.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/internal/service/runtime_policy.go b/backend/internal/service/runtime_policy.go index 90f7a7963..7ce9fb232 100644 --- a/backend/internal/service/runtime_policy.go +++ b/backend/internal/service/runtime_policy.go @@ -119,7 +119,7 @@ func defaultRuntimePolicy() RuntimePolicySetting { TextTimeoutMinutes: 8, AudioTimeoutMinutes: 8, VideoTimeoutMinutes: 30, - StoryboardTimeoutMinutes: 15, + StoryboardTimeoutMinutes: 20, // 默认分镜任务超时提高到20分钟，避免腾讯1环境2分钟快速失败（已修改，详见runtime_policy.go:122） DefaultTimeoutMinutes: 10, }, From ecccc84d6c8bb9077689e988847d2648c1206920 Mon Sep 17 00:00:00 2001 From: yepwait <1239738103@qq.com> Date: Fri, 7 Aug 2026 17:06:30 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(backend):=20=E5=88=86=E9=95=9C=E7=94=9F?= =?UTF-8?q?=E6=88=90=20-=20=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F=E4=BF=9D?= =?UTF-8?q?=E6=B4=BB=E5=B9=B6=E5=BB=B6=E9=95=BF=E4=BB=BB=E5=8A=A1=E8=B6=85?= =?UTF-8?q?=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/service/provider.go | 197 +++++++++++++++++++--- backend/internal/service/provider_test.go | 54 ++++++ backend/internal/service/service.go | 8 +- 3 files changed, 231 insertions(+), 28 deletions(-) diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index b308e63f6..cd0a4ab39 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -1,6 +1,7 @@ package service import ( + "bufio" "bytes" "context" "encoding/base64" @@ -31,6 +32,7 @@ type canvasGenerationInput struct { ReferenceAudios []providerMedia `json:"referenceAudios"` Mask *providerMedia `json:"mask"` Metadata map[string]interface{} `json:"metadata"` + StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。 VideoCapability *VideoCapabilityConfig `json:"-"` } @@ -585,13 +587,13 @@ func runTextTask(ctx context.Context, input canvasGenerationInput) (map[string]i } func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { - var payload map[string]interface{} responseInput, err := textResponseInput(input) if err != nil { return nil, err } body := map[string]interface{}{"model": input.Config.Model, "input": responseInput} - if err := postJSON(ctx, input.Config, "/responses", body, &payload); err != nil { + text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText) + if err != nil { if !shouldFallbackTextToChat(err) { return nil, err } @@ -601,37 +603,23 @@ func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[st } return nil, fmt.Errorf("文本接口请求失败：Responses API %v；Chat Completions %v", err, chatErr) } - text := stringField(payload, "output_text") - if text == "" { - text = extractResponseText(payload) - } - if text == "" { - return nil, errors.New("文本接口没有返回内容") - } return map[string]interface{}{"mode": "text", "text": text}, nil } func runResponsesTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { - var payload map[string]interface{} responseInput, err := textResponseInput(input) if err != nil { return nil, err } - if err := postJSON(ctx, input.Config, "/responses", map[string]interface{}{"model": input.Config.Model, "input": responseInput}, &payload); err != nil { + body := map[string]interface{}{"model": input.Config.Model, "input": responseInput} + text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText) + if err != nil { return nil, err } - text := stringField(payload, "output_text") - if text == "" { - text = extractResponseText(payload) - } - if text == "" { - return nil, errors.New("文本接口没有返回内容") - } return map[string]interface{}{"mode": "text", "text": text}, nil } func runChatCompletionsTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { - var payload map[string]interface{} messages := []map[string]interface{}{} if systemPrompt := strings.TrimSpace(input.Config.SystemPrompt); systemPrompt != "" { messages = append(messages, map[string]interface{}{"role": "system", "content": systemPrompt}) @@ -642,13 +630,10 @@ func runChatCompletionsTextTask(ctx context.Context, input canvasGenerationInput } messages = append(messages, map[string]interface{}{"role": "user", "content": userContent}) body := map[string]interface{}{"model": input.Config.Model, "messages": messages} - if err := postJSON(ctx, input.Config, "/chat/completions", body, &payload); err != nil { + text, err := requestTextProvider(ctx, input.Config, "/chat/completions", body, "chat-completion", input.StreamText) + if err != nil { return nil, err } - text := extractChatCompletionText(payload) - if text == "" { - return nil, errors.New("文本接口没有返回内容") - } return map[string]interface{}{"mode": "text", "text": text}, nil } @@ -1893,6 +1878,170 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn return nil, fmt.Errorf("%s视频生成超时", providerName) } +func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool) (string, error) { + if stream { + return postStreamingText(ctx, config, path, body, protocol) + } + var payload map[string]interface{} + if err := postJSON(ctx, config, path, body, &payload); err != nil { + return "", err + } + text := extractTextPayload(payload, protocol) + if text == "" { + return "", errors.New("文本接口没有返回内容") + } + return text, nil +} + +func postStreamingText(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string) (string, error) { + // ponytail: 只把分镜规划/修复切到上游 SSE，完整 JSON 仍在结束后校验，避免半截结构污染画布。 + body["stream"] = true + data, mimeType, err := postStreamingBinary(ctx, config, path, body) + if err != nil { + return "", err + } + if !strings.Contains(strings.ToLower(mimeType), "event-stream") { + var payload map[string]interface{} + if err := json.Unmarshal(data, &payload); err != nil { + return "", fmt.Errorf("流式文本接口返回格式无效：%w", err) + } + if err := validateTextPayload(payload); err != nil { + return "", err + } + text := extractTextPayload(payload, protocol) + if text == "" { + return "", errors.New("文本接口没有返回内容") + } + return text, nil + } + return parseTextEventStream(data, protocol) +} + +func extractTextPayload(payload map[string]interface{}, protocol string) string { + if protocol == "responses" { + text := stringField(payload, "output_text") + if text == "" { + text = extractResponseText(payload) + } + return text + } + return extractChatCompletionText(payload) +} + +func validateTextPayload(payload map[string]interface{}) error { + if code, ok := payload["code"].(float64); ok && code != 0 { + return errors.New(defaultString(stringField(payload, "msg"), "请求失败")) + } + if errValue, ok := payload["error"].(map[string]interface{}); ok { + if message := stringField(errValue, "message"); message != "" { + return errors.New(message) + } + } + return nil +} + +func parseTextEventStream(data []byte, protocol string) (string, error) { + scanner := bufio.NewScanner(bytes.NewReader(data)) + scanner.Buffer(make([]byte, 64<<10), len(data)+1) + var text strings.Builder + var eventName string + var dataLines []string + + flush := func() error { + if len(dataLines) == 0 { + eventName = "" + return nil + } + raw := strings.TrimSpace(strings.Join(dataLines, "\n")) + dataLines = nil + if raw == "" || raw == "[DONE]" { + eventName = "" + return nil + } + var payload map[string]interface{} + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + return fmt.Errorf("流式文本事件解析失败：%w", err) + } + if eventName == "error" { + if err := validateTextPayload(payload); err != nil { + return err + } + return errors.New("上游流式文本请求失败") + } + if err := validateTextPayload(payload); err != nil { + return err + } + if protocol == "responses" { + text.WriteString(stringField(payload, "delta")) + } else { + choices, _ := payload["choices"].([]interface{}) + for _, choice := range choices { + record, _ := choice.(map[string]interface{}) + delta, _ := record["delta"].(map[string]interface{}) + text.WriteString(streamContentText(delta["content"])) + } + } + eventName = "" + return nil + } + + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + if err := flush(); err != nil { + return "", err + } + continue + } + switch { + case strings.HasPrefix(line, "event:"): + eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + case strings.HasPrefix(line, "data:"): + value := strings.TrimPrefix(line, "data:") + dataLines = append(dataLines, strings.TrimPrefix(value, " ")) + } + } + if err := scanner.Err(); err != nil { + return "", fmt.Errorf("读取流式文本响应失败：%w", err) + } + if err := flush(); err != nil { + return "", err + } + if text.Len() == 0 { + return "", errors.New("流式文本接口没有返回内容") + } + return text.String(), nil +} + +func streamContentText(value interface{}) string { + if text, ok := value.(string); ok { + return text + } + parts, ok := value.([]interface{}) + if !ok { + return "" + } + var result strings.Builder + for _, part := range parts { + record, _ := part.(map[string]interface{}) + result.WriteString(stringField(record, "text")) + } + return result.String() +} + +func postStreamingBinary(ctx context.Context, config providerConfig, path string, body interface{}) ([]byte, string, error) { + data, _ := json.Marshal(body) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL(config.BaseURL, path), bytes.NewReader(data)) + if err != nil { + return nil, "", err + } + req.Header.Set("Authorization", "Bearer "+config.APIKey) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + ApplyOutboundHeaders(req, config.Headers) + return doBinary(req) +} + func postJSON(ctx context.Context, config providerConfig, path string, body interface{}, target interface{}) error { data, _ := json.Marshal(body) req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL(config.BaseURL, path), bytes.NewReader(data)) diff --git a/backend/internal/service/provider_test.go b/backend/internal/service/provider_test.go index 019162627..31c5fa1b9 100644 --- a/backend/internal/service/provider_test.go +++ b/backend/internal/service/provider_test.go @@ -55,6 +55,60 @@ func TestWriteMediaPartSanitizesFilenameAndSetsMimeType(t *testing.T) { } } +func TestParseTextEventStreamSupportsResponsesAndChat(t *testing.T) { + responses := []byte(`event: response.output_text.delta +data: {"delta":"{\"title\":\"分镜\"}"} + +event: response.output_text.delta +data: {"delta":"}"} + +data: [DONE] + +`) + if got, err := parseTextEventStream(responses, "responses"); err != nil || got != `{"title":"分镜"}` { + t.Fatalf("Responses stream = %q, err = %v", got, err) + } + + chat := []byte(`data: {"choices":[{"delta":{"content":"第一镜"}}]} + +data: {"choices":[{"delta":{"content":"：远景"}}]} + +data: [DONE] + +`) + if got, err := parseTextEventStream(chat, "chat-completion"); err != nil || got != "第一镜：远景" { + t.Fatalf("Chat stream = %q, err = %v", got, err) + } +} + +func TestPostStreamingTextSetsStreamHeaders(t *testing.T) { + t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Accept"); got != "text/event-stream" { + t.Errorf("Accept = %q", got) + } + var body map[string]interface{} + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Errorf("decode request body: %v", err) + } + if stream, ok := body["stream"].(bool); !ok || !stream { + t.Errorf("stream body field = %#v", body["stream"]) + } + w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") + _, _ = w.Write([]byte(`data: {"choices":[{"delta":{"content":"流式分镜"}}]} + +data: [DONE] + +`)) + })) + defer server.Close() + + got, err := postStreamingText(context.Background(), providerConfig{BaseURL: server.URL, APIKey: "test-key"}, "/chat/completions", map[string]interface{}{"model": "test-model"}, "chat-completion") + if err != nil || got != "流式分镜" { + t.Fatalf("postStreamingText() = %q, err = %v", got, err) + } +} + func TestProviderHTTPErrorWarnsAboutUncertain524Billing(t *testing.T) { message := (providerHTTPError{StatusCode: 524, Status: "524 A Timeout Occurred"}).Error() if !strings.Contains(message, "可能仍在服务端执行并产生费用") || !strings.Contains(message, "请勿立即重试") { diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index 0c89b64c6..01f501671 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -1134,7 +1134,7 @@ func (s *Service) processAgentStoryboardTask(ctx context.Context, task model.Tas if err != nil { return nil, nil, err } - result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config}) + result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config, StreamText: true}) if err != nil { return nil, nil, err } @@ -1149,7 +1149,7 @@ func (s *Service) processAgentStoryboardTask(ctx context.Context, task model.Tas if promptErr != nil { return nil, nil, promptErr } - repaired, repairErr := runTextTask(withProviderRequestKind(ctx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config}) + repaired, repairErr := runTextTask(withProviderRequestKind(ctx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config, StreamText: true}) if repairErr != nil { return nil, nil, fmt.Errorf("分镜结构修复失败：%w", repairErr) } @@ -1190,7 +1190,7 @@ func (s *Service) processStoryboardRowsTask(ctx context.Context, task model.Task if err != nil { return nil, nil, err } - result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config}) + result, err := runTextTask(ctx, canvasGenerationInput{Mode: "text", Prompt: plannerPrompt, Config: config, StreamText: true}) if err != nil { return nil, nil, err } @@ -1205,7 +1205,7 @@ func (s *Service) processStoryboardRowsTask(ctx context.Context, task model.Task if promptErr != nil { return nil, nil, promptErr } - repaired, repairErr := runTextTask(withProviderRequestKind(ctx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config}) + repaired, repairErr := runTextTask(withProviderRequestKind(ctx, "repair"), canvasGenerationInput{Mode: "text", Prompt: repairPrompt, Config: config, StreamText: true}) if repairErr != nil { return nil, nil, fmt.Errorf("分镜结构修复失败：%w", repairErr) }