From e4441f91adc1c974730ceac41337a04621958128 Mon Sep 17 00:00:00 2001 From: temp Date: Tue, 25 Aug 2026 14:54:02 +0800 Subject: [PATCH] fix: stabilize eino summarization for deepseek --- internal/multiagent/eino_agentic_summarize.go | 5 +- internal/multiagent/eino_summarize.go | 46 +++++++++++-------- internal/multiagent/eino_summarize_payload.go | 23 +++++++++- .../multiagent/eino_summarize_payload_test.go | 44 +++++++++++++++++- internal/multiagent/eino_transient_retry.go | 8 ++++ .../multiagent/eino_transient_retry_test.go | 1 + internal/openai/reasoning_payload.go | 17 +++++++ 7 files changed, 119 insertions(+), 25 deletions(-) diff --git a/internal/multiagent/eino_agentic_summarize.go b/internal/multiagent/eino_agentic_summarize.go index d4f1e45b..2be3f455 100644 --- a/internal/multiagent/eino_agentic_summarize.go +++ b/internal/multiagent/eino_agentic_summarize.go @@ -10,7 +10,6 @@ import ( "cyberstrike-ai/internal/config" "cyberstrike-ai/internal/database" - einoopenai "github.com/cloudwego/eino-ext/components/model/openai" "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/adk/middlewares/summarization" "github.com/cloudwego/eino/components/model" @@ -109,9 +108,7 @@ func newEinoAgenticSummarizationMiddleware( retryPolicy := einoTransientRunRetryPolicyFromMW(mwCfg) retryMax := retryPolicy.maxAttempts var summaryOverflowRetries int - summaryModelOpts := []model.Option{ - einoopenai.WithMaxCompletionTokens(outputReserve), - } + summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "agentic", &appCfg.OpenAI, logger) mw, err := summarization.NewTyped[*schema.AgenticMessage](ctx, &summarization.TypedConfig[*schema.AgenticMessage]{ Model: summaryModel, diff --git a/internal/multiagent/eino_summarize.go b/internal/multiagent/eino_summarize.go index 114a7626..d7fa27d8 100644 --- a/internal/multiagent/eino_summarize.go +++ b/internal/multiagent/eino_summarize.go @@ -164,24 +164,7 @@ func newEinoSummarizationMiddleware( retryMax := retryPolicy.maxAttempts var summaryOverflowRetries int - // ModelOptions apply only to summarization Generate (same ChatModel instance as the agent). - // Strip thinking/reasoning on this call path; mark requests for empty-choices diagnostics. - summaryModelOpts := []model.Option{ - einoopenai.WithMaxCompletionTokens(outputReserve), - einoopenai.WithExtraHeader(map[string]string{ - copenai.SummarizationRequestHeader: "1", - }), - einoopenai.WithRequestPayloadModifier(func(_ context.Context, in []*schema.Message, rawBody []byte) ([]byte, error) { - if logger != nil { - logger.Info("eino summarization generate request", - zap.Int("input_messages", len(in)), - zap.Int("payload_bytes", len(rawBody)), - zap.String("model", modelName), - ) - } - return stripReasoningFromSummarizationPayload(rawBody) - }), - } + summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "classic", &appCfg.OpenAI, logger) mw, err := summarization.New(ctx, &summarization.Config{ Model: summaryModel, @@ -308,6 +291,33 @@ func newEinoSummarizationMiddleware( return mw, nil } +// newEinoSummarizationModelOptions applies only to summarization Generate calls +// on the shared main model. Summary generation should be plain-text and cheap: +// strip provider reasoning/thinking controls so DeepSeek/OpenAI-compatible +// endpoints do not spend the reserved output budget on invisible reasoning. +func newEinoSummarizationModelOptions(outputReserve int, modelName, kind string, oa *config.OpenAIConfig, logger *zap.Logger) []model.Option { + label := "eino summarization generate request" + if strings.TrimSpace(kind) != "" && kind != "classic" { + label = "eino " + kind + " summarization generate request" + } + return []model.Option{ + einoopenai.WithMaxCompletionTokens(outputReserve), + einoopenai.WithExtraHeader(map[string]string{ + copenai.SummarizationRequestHeader: "1", + }), + einoopenai.WithRequestPayloadModifier(func(_ context.Context, in []*schema.Message, rawBody []byte) ([]byte, error) { + if logger != nil { + logger.Info(label, + zap.Int("input_messages", len(in)), + zap.Int("payload_bytes", len(rawBody)), + zap.String("model", modelName), + ) + } + return stripReasoningFromSummarizationPayload(rawBody, oa) + }), + } +} + // summarizationInputBudgetOpts controls spill/truncation behavior when a round alone exceeds budget. type summarizationInputBudgetOpts struct { toolMaxBytes int diff --git a/internal/multiagent/eino_summarize_payload.go b/internal/multiagent/eino_summarize_payload.go index 254e7849..0c24259d 100644 --- a/internal/multiagent/eino_summarize_payload.go +++ b/internal/multiagent/eino_summarize_payload.go @@ -1,12 +1,33 @@ package multiagent import ( + "strings" + + "cyberstrike-ai/internal/config" copenai "cyberstrike-ai/internal/openai" ) // stripReasoningFromSummarizationPayload removes thinking / reasoning fields from a // chat-completions JSON body. Applied only to summarization Generate calls via // model.ModelOptions on the shared ChatModel — main-agent requests are unchanged. -func stripReasoningFromSummarizationPayload(rawBody []byte) ([]byte, error) { +func stripReasoningFromSummarizationPayload(rawBody []byte, oa *config.OpenAIConfig) ([]byte, error) { + if shouldDisableDeepSeekThinkingForSummarization(oa) { + return copenai.DisableThinkingForChatCompletionBody(rawBody) + } return copenai.StripReasoningFromChatCompletionBody(rawBody) } + +func shouldDisableDeepSeekThinkingForSummarization(oa *config.OpenAIConfig) bool { + if oa == nil { + return false + } + profile := strings.ToLower(strings.TrimSpace(oa.Reasoning.ProfileEffective())) + switch profile { + case "deepseek", "deepseek_compat": + return true + case "", "auto": + return oa.IsDeepSeekEndpointOrModel() + default: + return false + } +} diff --git a/internal/multiagent/eino_summarize_payload_test.go b/internal/multiagent/eino_summarize_payload_test.go index a84ce33f..0b075852 100644 --- a/internal/multiagent/eino_summarize_payload_test.go +++ b/internal/multiagent/eino_summarize_payload_test.go @@ -3,11 +3,13 @@ package multiagent import ( "strings" "testing" + + "cyberstrike-ai/internal/config" ) func TestStripReasoningFromSummarizationPayload(t *testing.T) { in := []byte(`{"model":"deepseek-chat","messages":[],"thinking":{"type":"enabled"},"reasoning_effort":"high"}`) - out, err := stripReasoningFromSummarizationPayload(in) + out, err := stripReasoningFromSummarizationPayload(in, nil) if err != nil { t.Fatal(err) } @@ -20,7 +22,7 @@ func TestStripReasoningFromSummarizationPayload(t *testing.T) { } plain := []byte(`{"model":"gpt-4o","messages":[]}`) - out2, err := stripReasoningFromSummarizationPayload(plain) + out2, err := stripReasoningFromSummarizationPayload(plain, nil) if err != nil { t.Fatal(err) } @@ -28,3 +30,41 @@ func TestStripReasoningFromSummarizationPayload(t *testing.T) { t.Fatalf("expected unchanged payload, got %s", out2) } } + +func TestStripReasoningFromSummarizationPayloadDisablesDeepSeekThinking(t *testing.T) { + in := []byte(`{"model":"deepseek-v4-flash","messages":[],"thinking":{"type":"enabled"},"reasoning_effort":"high"}`) + oa := &config.OpenAIConfig{ + BaseURL: "https://api.deepseek.com/v1", + Model: "deepseek-v4-flash", + } + out, err := stripReasoningFromSummarizationPayload(in, oa) + if err != nil { + t.Fatal(err) + } + s := string(out) + if strings.Contains(s, "reasoning_effort") { + t.Fatalf("expected reasoning_effort stripped, got %s", s) + } + if !strings.Contains(s, `"thinking":{"type":"disabled"}`) { + t.Fatalf("expected DeepSeek thinking disabled, got %s", s) + } +} + +func TestStripReasoningFromSummarizationPayloadHonorsOpenAICompatProfile(t *testing.T) { + in := []byte(`{"model":"deepseek-v4-flash","messages":[],"thinking":{"type":"enabled"},"reasoning_effort":"high"}`) + oa := &config.OpenAIConfig{ + BaseURL: "https://api.deepseek.com/v1", + Model: "deepseek-v4-flash", + Reasoning: config.OpenAIReasoningConfig{ + Profile: "openai_compat", + }, + } + out, err := stripReasoningFromSummarizationPayload(in, oa) + if err != nil { + t.Fatal(err) + } + s := string(out) + if strings.Contains(s, "thinking") || strings.Contains(s, "reasoning_effort") { + t.Fatalf("expected OpenAI-compatible profile to strip reasoning fields, got %s", s) + } +} diff --git a/internal/multiagent/eino_transient_retry.go b/internal/multiagent/eino_transient_retry.go index 64a74bac..1bfbd7a0 100644 --- a/internal/multiagent/eino_transient_retry.go +++ b/internal/multiagent/eino_transient_retry.go @@ -52,6 +52,9 @@ func isEinoTransientRunError(err error) bool { if msg == "" { return false } + if isEinoEmptySummaryContentErrorText(msg) { + return true + } if status := httpStatusFromErrorText(msg); status > 0 { return isRetryableHTTPStatus(status) } @@ -94,6 +97,11 @@ func isEinoTransientRunError(err error) bool { return false } +func isEinoEmptySummaryContentErrorText(msg string) bool { + return strings.Contains(msg, "summary content is empty") || + strings.Contains(msg, "agentic summarization returned empty summary") +} + func isRetryableHTTPStatus(status int) bool { switch status { case 408, 409, 425, 429: diff --git a/internal/multiagent/eino_transient_retry_test.go b/internal/multiagent/eino_transient_retry_test.go index be22250d..c8dd19fb 100644 --- a/internal/multiagent/eino_transient_retry_test.go +++ b/internal/multiagent/eino_transient_retry_test.go @@ -36,6 +36,7 @@ func TestIsEinoTransientRunError(t *testing.T) { {"http2 goaway", errors.New("failed to receive stream chunk: error, http2: server sent GOAWAY and closed the connection; LastStreamID=791, ErrCode=NO_ERROR"), true}, {"unexpected internal stream chunk", errors.New("failed to receive stream chunk: error, The service encountered an unexpected internal error. Request id: 0217851391106464f01ec66621d0980a42fd45436ed75957a6a0a"), true}, {"unexpected eof", errors.New("unexpected EOF"), true}, + {"empty summarization output", errors.New("[NodeRunError] summary content is empty\nnode path: [node_1, ChatModel]"), true}, {"503", errors.New("upstream returned 503"), true}, {"iteration limit", errors.New("max iteration reached"), false}, {"canceled", context.Canceled, false}, diff --git a/internal/openai/reasoning_payload.go b/internal/openai/reasoning_payload.go index 2110c90d..3e418627 100644 --- a/internal/openai/reasoning_payload.go +++ b/internal/openai/reasoning_payload.go @@ -32,6 +32,23 @@ func StripReasoningFromChatCompletionBody(rawBody []byte) ([]byte, error) { return out, nil } +// DisableThinkingForChatCompletionBody removes generic reasoning controls and +// explicitly disables DeepSeek-style thinking. Use only for providers where +// omitting the field would leave thinking enabled by default. +func DisableThinkingForChatCompletionBody(rawBody []byte) ([]byte, error) { + var payload map[string]any + if err := sonic.Unmarshal(rawBody, &payload); err != nil { + return rawBody, nil + } + stripReasoningFields(payload) + payload["thinking"] = map[string]any{"type": "disabled"} + out, err := sonic.Marshal(payload) + if err != nil { + return rawBody, err + } + return out, nil +} + // StripReasoningIfForcedToolChoice removes thinking / reasoning fields when the // request sets tool_choice to "required" or an object. Several providers reject // that combination (e.g. DashScope: "tool_choice does not support being set to