From d7581a63735b5f1a65f6a2bd147ae5b5e24fbd52 Mon Sep 17 00:00:00 2001 From: temp Date: Tue, 25 Aug 2026 18:28:47 +0800 Subject: [PATCH] fix: surface summarization model errors --- internal/handler/agent.go | 2 +- internal/handler/batch_queue_executor.go | 5 +- internal/handler/eino_single_agent.go | 12 +- internal/handler/multi_agent.go | 19 +- .../eino_agentic_chat_model_agent_test.go | 15 +- internal/multiagent/eino_agentic_summarize.go | 2 +- .../multiagent/eino_agentic_summarize_test.go | 53 +++++ internal/multiagent/eino_run_error_handler.go | 116 +++++++++- .../multiagent/eino_run_error_handler_test.go | 78 +++++++ internal/multiagent/eino_summarize.go | 2 +- .../multiagent/eino_summarize_model_guard.go | 214 ++++++++++++++++++ .../eino_summarize_model_guard_test.go | 101 +++++++++ internal/multiagent/eino_summarize_payload.go | 5 +- .../multiagent/eino_summarize_payload_test.go | 26 ++- internal/multiagent/eino_summarize_test.go | 79 +++++++ 15 files changed, 701 insertions(+), 28 deletions(-) create mode 100644 internal/multiagent/eino_summarize_model_guard.go create mode 100644 internal/multiagent/eino_summarize_model_guard_test.go diff --git a/internal/handler/agent.go b/internal/handler/agent.go index 80b63c92..c9eeb737 100644 --- a/internal/handler/agent.go +++ b/internal/handler/agent.go @@ -747,7 +747,7 @@ func (h *AgentHandler) finalizeRobotAgentError(ctx context.Context, assistantMes if shouldPersistEinoAgentTraceAfterRunError(ctx) { h.persistEinoAgentTraceForResume(conversationID, resultMA) } - errMsg := "执行失败: " + errMA.Error() + errMsg := "执行失败: " + multiagent.EinoClientRunErrorMessage(errMA) if assistantMessageID != "" { _, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID) _ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil) diff --git a/internal/handler/batch_queue_executor.go b/internal/handler/batch_queue_executor.go index 53eed890..5b7f58c7 100644 --- a/internal/handler/batch_queue_executor.go +++ b/internal/handler/batch_queue_executor.go @@ -391,7 +391,8 @@ func (h *AgentHandler) handleBatchSubTaskRunError( } h.logger.Error("批量任务执行失败", zap.String("queueId", queueID), zap.String("taskId", task.ID), zap.String("conversationId", conversationID), zap.Error(runErr)) - errorMsg := "执行失败: " + runErr.Error() + clientErr := multiagent.EinoClientRunErrorMessage(runErr) + errorMsg := "执行失败: " + clientErr if assistantMessageID != "" { if _, updateErr := h.db.Exec( "UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", @@ -404,5 +405,5 @@ func (h *AgentHandler) handleBatchSubTaskRunError( h.logger.Warn("保存错误详情失败", zap.String("queueId", queueID), zap.String("taskId", task.ID), zap.Error(err)) } } - h.batchTaskManager.UpdateTaskStatus(queueID, task.ID, BatchTaskStatusFailed, "", runErr.Error()) + h.batchTaskManager.UpdateTaskStatus(queueID, task.ID, BatchTaskStatusFailed, "", clientErr) } diff --git a/internal/handler/eino_single_agent.go b/internal/handler/eino_single_agent.go index 1691b40d..7e4c7f5d 100644 --- a/internal/handler/eino_single_agent.go +++ b/internal/handler/eino_single_agent.go @@ -371,15 +371,17 @@ func (h *AgentHandler) EinoSingleAgentLoopStream(c *gin.Context) { h.logger.Error("Eino ADK 单代理执行失败", zap.Error(runErr)) taskStatus = "failed" h.tasks.UpdateTaskStatus(conversationID, taskStatus) - errMsg := "执行失败: " + runErr.Error() + clientErr := multiagent.EinoClientRunErrorMessage(runErr) + errMsg := "执行失败: " + clientErr if assistantMessageID != "" { _, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID) _ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil) } - sendEvent("error", errMsg, map[string]interface{}{ - "conversationId": conversationID, - "messageId": assistantMessageID, - }) + errData := multiagent.EinoClientRunErrorFields(runErr) + errData["conversationId"] = conversationID + errData["messageId"] = assistantMessageID + errData["error"] = errMsg + sendEvent("error", errMsg, errData) sendEvent("done", "", map[string]interface{}{"conversationId": conversationID}) timeoutCancel() return diff --git a/internal/handler/multi_agent.go b/internal/handler/multi_agent.go index 7da4e76f..5c4bf392 100644 --- a/internal/handler/multi_agent.go +++ b/internal/handler/multi_agent.go @@ -385,15 +385,17 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) { h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr)) taskStatus = "failed" h.tasks.UpdateTaskStatus(conversationID, taskStatus) - errMsg := "执行失败: " + runErr.Error() + clientErr := multiagent.EinoClientRunErrorMessage(runErr) + errMsg := "执行失败: " + clientErr if assistantMessageID != "" { _, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID) _ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil) } - sendEvent("error", errMsg, map[string]interface{}{ - "conversationId": conversationID, - "messageId": assistantMessageID, - }) + errData := multiagent.EinoClientRunErrorFields(runErr) + errData["conversationId"] = conversationID + errData["messageId"] = assistantMessageID + errData["error"] = errMsg + sendEvent("error", errMsg, errData) sendEvent("done", "", map[string]interface{}{"conversationId": conversationID}) timeoutCancel() return @@ -513,11 +515,14 @@ func (h *AgentHandler) MultiAgentLoop(c *gin.Context) { h.persistEinoAgentTraceForResume(prep.ConversationID, result) } h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr)) - errMsg := "执行失败: " + runErr.Error() + clientErr := multiagent.EinoClientRunErrorMessage(runErr) + errMsg := "执行失败: " + clientErr if prep.AssistantMessageID != "" { _, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), prep.AssistantMessageID) } - c.JSON(http.StatusInternalServerError, gin.H{"error": errMsg}) + errData := multiagent.EinoClientRunErrorFields(runErr) + errData["error"] = errMsg + c.JSON(http.StatusInternalServerError, errData) return } mw := &h.config.MultiAgent.EinoMiddleware diff --git a/internal/multiagent/eino_agentic_chat_model_agent_test.go b/internal/multiagent/eino_agentic_chat_model_agent_test.go index 63761c71..4046af61 100644 --- a/internal/multiagent/eino_agentic_chat_model_agent_test.go +++ b/internal/multiagent/eino_agentic_chat_model_agent_test.go @@ -12,15 +12,24 @@ import ( ) type capturingAgenticChatModel struct { - mu sync.Mutex - inputs [][]*schema.AgenticMessage - output *schema.AgenticMessage + mu sync.Mutex + inputs [][]*schema.AgenticMessage + output *schema.AgenticMessage + outputs []*schema.AgenticMessage } func (m *capturingAgenticChatModel) Generate(_ context.Context, input []*schema.AgenticMessage, _ ...model.Option) (*schema.AgenticMessage, error) { m.mu.Lock() m.inputs = append(m.inputs, input) + callNo := len(m.inputs) m.mu.Unlock() + if len(m.outputs) > 0 { + idx := callNo - 1 + if idx >= len(m.outputs) { + idx = len(m.outputs) - 1 + } + return m.outputs[idx], nil + } if m.output != nil { return m.output, nil } diff --git a/internal/multiagent/eino_agentic_summarize.go b/internal/multiagent/eino_agentic_summarize.go index 2be3f455..a2423370 100644 --- a/internal/multiagent/eino_agentic_summarize.go +++ b/internal/multiagent/eino_agentic_summarize.go @@ -111,7 +111,7 @@ func newEinoAgenticSummarizationMiddleware( summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "agentic", &appCfg.OpenAI, logger) mw, err := summarization.NewTyped[*schema.AgenticMessage](ctx, &summarization.TypedConfig[*schema.AgenticMessage]{ - Model: summaryModel, + Model: newNonEmptyAgenticSummaryModel(summaryModel), ModelOptions: summaryModelOpts, GenModelInput: func(ctx context.Context, sysInstruction, userInstruction *schema.AgenticMessage, originalMsgs []*schema.AgenticMessage) ([]*schema.AgenticMessage, error) { classicOriginal := AgenticMessagesToEino(originalMsgs) diff --git a/internal/multiagent/eino_agentic_summarize_test.go b/internal/multiagent/eino_agentic_summarize_test.go index 4472582c..16b81e7a 100644 --- a/internal/multiagent/eino_agentic_summarize_test.go +++ b/internal/multiagent/eino_agentic_summarize_test.go @@ -171,6 +171,59 @@ func TestEinoAgenticChatModelAgentCompactsContextBeforeBusinessModel(t *testing. } } +func TestEinoAgenticSummarizationMiddlewareRetriesWhenSummaryModelReturnsEmpty(t *testing.T) { + t.Parallel() + ctx := context.Background() + emit := false + summaryModel := &capturingAgenticChatModel{ + outputs: []*schema.AgenticMessage{ + agenticAssistantTextMessage(""), + agenticAssistantTextMessage("有效摘要：继续验证 SQL 注入路径"), + }, + } + appCfg := &config.Config{} + appCfg.OpenAI.Model = "gpt-4o" + appCfg.OpenAI.MaxTotalTokens = 5000 + appCfg.Database.Path = filepath.Join(t.TempDir(), "cyberstrike.db") + mwCfg := &config.MultiAgentEinoMiddlewareConfig{ + SummarizationEmitInternalEvents: &emit, + SummarizationOutputReserveTokens: 1024, + } + + mw, err := newEinoAgenticSummarizationMiddleware(ctx, summaryModel, appCfg, mwCfg, "conv-agentic-empty-summary", nil, "", nil) + if err != nil { + t.Fatalf("newEinoAgenticSummarizationMiddleware: %v", err) + } + state := &adk.TypedChatModelAgentState[*schema.AgenticMessage]{ + Messages: []*schema.AgenticMessage{ + schema.SystemAgenticMessage("system root"), + schema.UserAgenticMessage("授权范围 example.com\n" + strings.Repeat("历史扫描输出 ", 12000)), + agenticAssistantTextMessage("已记录范围"), + schema.UserAgenticMessage("继续验证 SQL 注入路径"), + }, + } + + _, after, err := mw.BeforeModelRewriteState(ctx, state, nil) + if err != nil { + t.Fatalf("BeforeModelRewriteState should retry instead of failing on empty summary: %v", err) + } + if after == nil { + t.Fatal("after state is nil") + } + if inputs := summaryModel.snapshotInputs(); len(inputs) < 2 { + t.Fatalf("summary model calls=%d, want retry after empty output", len(inputs)) + } + joined := joinClassicMessageContent(AgenticMessagesToEino(after.Messages)) + for _, want := range []string{"有效摘要", "继续验证 SQL 注入路径", "原始用户输入与约束账本"} { + if !strings.Contains(joined, want) { + t.Fatalf("retried compacted context missing %q:\n%s", want, joined) + } + } + if strings.Contains(joined, "本地压缩摘要") { + t.Fatalf("local fallback should not be used:\n%s", joined) + } +} + func TestAppendEinoAgenticChatModelTailMiddlewaresIncludesTypedSummarization(t *testing.T) { t.Parallel() mw := newAgenticSystemMessageNormalizerMiddleware(nil, "summary") diff --git a/internal/multiagent/eino_run_error_handler.go b/internal/multiagent/eino_run_error_handler.go index 94575cd9..21bd0c77 100644 --- a/internal/multiagent/eino_run_error_handler.go +++ b/internal/multiagent/eino_run_error_handler.go @@ -115,11 +115,7 @@ func (h *einoRunErrorHandler) emitError(err error, kind string) { } else if userErr.retryExhausted { data["hasModelOriginalError"] = false } - message := err.Error() - if userErr.message != "" { - message = userErr.message - } - h.progress("error", message, data) + h.progress("error", EinoClientRunErrorMessage(err), data) } type einoRunUserError struct { @@ -131,6 +127,7 @@ type einoRunUserError struct { retryExhausted bool totalRetries int hasModelOriginalError bool + summarizationModelErr bool } func einoUserFacingRunError(err error) einoRunUserError { @@ -152,6 +149,10 @@ func einoUserFacingRunError(err error) einoRunUserError { return out } out.rawLastError = strings.TrimSpace(lastErr.Error()) + if raw, ok := einoSummarizationModelRawErrorText(lastErr); ok { + out.rawLastError = raw + out.summarizationModelErr = true + } if isEinoShouldRetryOutputRejected(lastErr) { out.kind = "model_output_rejected" out.summary = "模型未返回原始错误；输出被重试策略拒绝。" @@ -163,6 +164,9 @@ func einoUserFacingRunError(err error) einoRunUserError { if strings.TrimSpace(summary) == "" { summary = einoTrimRetryErrorSummary(lastErr.Error()) } + if out.summarizationModelErr { + summary = einoTrimRetryErrorSummary(out.rawLastError) + } if kind == "" { kind = "model_retry_exhausted" } @@ -190,3 +194,105 @@ func formatEinoRetryExhaustedMessage(summary string, totalRetries int) string { } return "模型调用重试已耗尽：" + summary } + +// EinoClientRunErrorMessage returns the error text that should be shown directly +// to clients. When native retry hides the final provider failure behind a retry +// wrapper, prefer the original last model error so summarization/model issues are +// diagnosable from the frontend without opening server logs. +func EinoClientRunErrorMessage(err error) string { + if err == nil { + return "" + } + userErr := einoUserFacingRunError(err) + if userErr.retryExhausted { + if userErr.hasModelOriginalError && userErr.rawLastError != "" { + if userErr.summarizationModelErr { + return formatEinoSummarizationRetryExhaustedRawModelMessage(userErr.rawLastError, userErr.totalRetries) + } + return formatEinoRetryExhaustedRawModelMessage(userErr.rawLastError, userErr.totalRetries) + } + if userErr.message != "" { + return userErr.message + } + } + if raw, ok := einoSummarizationModelRawErrorText(err); ok { + return formatEinoSummarizationRawModelMessage(raw) + } + return err.Error() +} + +func formatEinoRetryExhaustedRawModelMessage(raw string, totalRetries int) string { + raw = strings.TrimSpace(raw) + prefix := "模型调用重试已耗尽" + if totalRetries > 0 { + prefix = fmt.Sprintf("模型调用重试已耗尽（已重试 %d 次）", totalRetries) + } + if raw == "" { + return prefix + } + return prefix + "，最后一次模型原始错误：\n" + raw +} + +func formatEinoSummarizationRetryExhaustedRawModelMessage(raw string, totalRetries int) string { + raw = strings.TrimSpace(raw) + prefix := "摘要阶段大模型调用失败，模型调用重试已耗尽" + if totalRetries > 0 { + prefix = fmt.Sprintf("摘要阶段大模型调用失败，模型调用重试已耗尽（已重试 %d 次）", totalRetries) + } + if raw == "" { + return prefix + } + return prefix + "，最后一次大模型报错原文：\n" + raw +} + +func formatEinoSummarizationRawModelMessage(raw string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "摘要阶段大模型调用失败。" + } + return "摘要阶段大模型调用失败，大模型报错原文：\n" + raw +} + +// EinoClientRunErrorFields returns structured diagnostic fields that handlers can +// attach to their final error event in addition to the visible message. +func EinoClientRunErrorFields(err error) map[string]interface{} { + fields := make(map[string]interface{}) + if err == nil { + return fields + } + userErr := einoUserFacingRunError(err) + if userErr.kind != "" { + fields["errorKind"] = userErr.kind + } + if userErr.summary != "" { + fields["errorSummary"] = userErr.summary + } + if userErr.retryExhausted { + fields["retryExhausted"] = true + if userErr.totalRetries > 0 { + fields["totalRetries"] = userErr.totalRetries + } + } + if userErr.rawLastError != "" { + fields["lastError"] = userErr.rawLastError + } + if userErr.technicalError != "" { + fields["technicalError"] = userErr.technicalError + } + if userErr.hasModelOriginalError { + fields["modelOriginalError"] = userErr.rawLastError + } else if userErr.retryExhausted { + fields["hasModelOriginalError"] = false + } + if userErr.summarizationModelErr { + fields["errorPhase"] = "summarization" + fields["summarizationModelError"] = true + fields["modelOriginalError"] = userErr.rawLastError + } else if raw, ok := einoSummarizationModelRawErrorText(err); ok { + fields["errorPhase"] = "summarization" + fields["summarizationModelError"] = true + fields["modelOriginalError"] = raw + fields["lastError"] = raw + } + return fields +} diff --git a/internal/multiagent/eino_run_error_handler_test.go b/internal/multiagent/eino_run_error_handler_test.go index f25865e5..d1e22e7a 100644 --- a/internal/multiagent/eino_run_error_handler_test.go +++ b/internal/multiagent/eino_run_error_handler_test.go @@ -155,6 +155,84 @@ func TestEinoRunErrorHandlerRetryExhaustedOriginalErrorProgress(t *testing.T) { } } +func TestEinoClientRunErrorMessageUsesRawRetryExhaustedModelError(t *testing.T) { + raw := "summary content is empty: role=assistant content_runes=0 reasoning_runes=42\nprovider request id: req_123" + err := &adk.RetryExhaustedError{ + LastErr: errors.New(raw), + TotalRetries: 4, + } + + got := EinoClientRunErrorMessage(err) + if !strings.Contains(got, "模型调用重试已耗尽（已重试 4 次）") { + t.Fatalf("message missing retry prefix: %q", got) + } + if !strings.Contains(got, raw) { + t.Fatalf("message should include raw model error:\n%s", got) + } + + fields := EinoClientRunErrorFields(err) + if fields["modelOriginalError"] != raw || fields["lastError"] != raw { + t.Fatalf("raw fields = %#v", fields) + } +} + +func TestEinoClientRunErrorMessageMarksSummarizationModelRetryError(t *testing.T) { + raw := `POST "https://api.deepseek.com/v1/chat/completions": 429 Too Many Requests: {"error":{"message":"Rate limit reached"}}` + err := &adk.RetryExhaustedError{ + LastErr: newEinoSummarizationModelError(errors.New(raw)), + TotalRetries: 4, + } + + got := EinoClientRunErrorMessage(err) + for _, want := range []string{ + "摘要阶段大模型调用失败", + "模型调用重试已耗尽（已重试 4 次）", + "最后一次大模型报错原文", + raw, + } { + if !strings.Contains(got, want) { + t.Fatalf("message missing %q:\n%s", want, got) + } + } + if strings.Contains(got, "summarization model error") { + t.Fatalf("message leaked internal wrapper:\n%s", got) + } + + fields := EinoClientRunErrorFields(err) + if fields["errorPhase"] != "summarization" || fields["summarizationModelError"] != true { + t.Fatalf("phase fields = %#v", fields) + } + if fields["modelOriginalError"] != raw || fields["lastError"] != raw { + t.Fatalf("raw fields = %#v", fields) + } +} + +func TestEinoClientRunErrorMessageMarksDirectSummarizationModelError(t *testing.T) { + raw := `POST "https://api.deepseek.com/v1/chat/completions": 400 Bad Request: {"error":{"message":"invalid thinking parameter"}}` + err := newEinoSummarizationModelError(errors.New(raw)) + + got := EinoClientRunErrorMessage(err) + for _, want := range []string{ + "摘要阶段大模型调用失败", + "大模型报错原文", + raw, + } { + if !strings.Contains(got, want) { + t.Fatalf("message missing %q:\n%s", want, got) + } + } + if strings.Contains(got, "summarization model error") { + t.Fatalf("message leaked internal wrapper:\n%s", got) + } + + fields := EinoClientRunErrorFields(err) + if fields["errorPhase"] != "summarization" || + fields["modelOriginalError"] != raw || + fields["lastError"] != raw { + t.Fatalf("fields = %#v", fields) + } +} + func TestEinoRunErrorHandlerIterationLimitProgress(t *testing.T) { var events []string var errorKind interface{} diff --git a/internal/multiagent/eino_summarize.go b/internal/multiagent/eino_summarize.go index 9e144ef6..b7cd7e41 100644 --- a/internal/multiagent/eino_summarize.go +++ b/internal/multiagent/eino_summarize.go @@ -167,7 +167,7 @@ func newEinoSummarizationMiddleware( summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "classic", &appCfg.OpenAI, logger) mw, err := summarization.New(ctx, &summarization.Config{ - Model: summaryModel, + Model: newNonEmptySummaryChatModel(summaryModel), ModelOptions: summaryModelOpts, GenModelInput: func(ctx context.Context, sysInstruction, userInstruction adk.Message, originalMsgs []adk.Message) ([]adk.Message, error) { if transcriptPath != "" && len(originalMsgs) > 0 { diff --git a/internal/multiagent/eino_summarize_model_guard.go b/internal/multiagent/eino_summarize_model_guard.go new file mode 100644 index 00000000..3a770d95 --- /dev/null +++ b/internal/multiagent/eino_summarize_model_guard.go @@ -0,0 +1,214 @@ +package multiagent + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" +) + +type einoSummarizationModelError struct { + err error +} + +func newEinoSummarizationModelError(err error) error { + if err == nil { + return nil + } + var existing *einoSummarizationModelError + if errors.As(err, &existing) { + return err + } + return &einoSummarizationModelError{err: err} +} + +func (e *einoSummarizationModelError) Error() string { + if e == nil || e.err == nil { + return "summarization model error" + } + return "summarization model error: " + e.err.Error() +} + +func (e *einoSummarizationModelError) Unwrap() error { + if e == nil { + return nil + } + return e.err +} + +func einoSummarizationModelRawErrorText(err error) (string, bool) { + var summaryErr *einoSummarizationModelError + if !errors.As(err, &summaryErr) || summaryErr == nil || summaryErr.err == nil { + return "", false + } + return strings.TrimSpace(summaryErr.err.Error()), true +} + +type nonEmptySummaryChatModel struct { + base model.BaseChatModel +} + +func newNonEmptySummaryChatModel(base model.BaseChatModel) model.BaseChatModel { + return &nonEmptySummaryChatModel{base: base} +} + +func (m *nonEmptySummaryChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) { + out, err := m.base.Generate(ctx, input, opts...) + if err != nil { + return out, newEinoSummarizationModelError(err) + } + if strings.TrimSpace(classicAssistantTextContent(out)) == "" { + return out, newEinoSummarizationModelError(fmt.Errorf("summary content is empty: %s", classicSummaryEmptyDiagnostics(out))) + } + return out, nil +} + +func (m *nonEmptySummaryChatModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) { + return m.base.Stream(ctx, input, opts...) +} + +type nonEmptyAgenticSummaryModel struct { + base model.BaseModel[*schema.AgenticMessage] +} + +func newNonEmptyAgenticSummaryModel(base model.BaseModel[*schema.AgenticMessage]) model.BaseModel[*schema.AgenticMessage] { + return &nonEmptyAgenticSummaryModel{base: base} +} + +func (m *nonEmptyAgenticSummaryModel) Generate(ctx context.Context, input []*schema.AgenticMessage, opts ...model.Option) (*schema.AgenticMessage, error) { + out, err := m.base.Generate(ctx, input, opts...) + if err != nil { + return out, newEinoSummarizationModelError(err) + } + if strings.TrimSpace(agenticAssistantTextContent(out)) == "" { + return out, newEinoSummarizationModelError(fmt.Errorf("summary content is empty: %s", agenticSummaryEmptyDiagnostics(out))) + } + return out, nil +} + +func (m *nonEmptyAgenticSummaryModel) Stream(ctx context.Context, input []*schema.AgenticMessage, opts ...model.Option) (*schema.StreamReader[*schema.AgenticMessage], error) { + return m.base.Stream(ctx, input, opts...) +} + +func classicAssistantTextContent(msg *schema.Message) string { + if msg == nil || msg.Role != schema.Assistant { + return "" + } + parts := make([]string, 0, len(msg.AssistantGenMultiContent)) + for _, part := range msg.AssistantGenMultiContent { + if part.Type == schema.ChatMessagePartTypeText && part.Text != "" { + parts = append(parts, part.Text) + } + } + if len(parts) > 0 { + return strings.Join(parts, "\n") + } + return msg.Content +} + +func agenticAssistantTextContent(msg *schema.AgenticMessage) string { + if msg == nil || msg.Role != schema.AgenticRoleTypeAssistant { + return "" + } + parts := make([]string, 0, len(msg.ContentBlocks)) + for _, block := range msg.ContentBlocks { + if block != nil && block.AssistantGenText != nil { + parts = append(parts, block.AssistantGenText.Text) + } + } + return strings.Join(parts, "\n") +} + +func classicSummaryEmptyDiagnostics(msg *schema.Message) string { + if msg == nil { + return "model returned nil message" + } + var textParts, reasoningParts, otherParts int + var multiTextRunes, multiReasoningRunes int + for _, part := range msg.AssistantGenMultiContent { + switch part.Type { + case schema.ChatMessagePartTypeText: + textParts++ + multiTextRunes += len([]rune(strings.TrimSpace(part.Text))) + case schema.ChatMessagePartTypeReasoning: + reasoningParts++ + if part.Reasoning != nil { + multiReasoningRunes += len([]rune(strings.TrimSpace(part.Reasoning.Text))) + } + default: + otherParts++ + } + } + reasoningRunes := len([]rune(strings.TrimSpace(msg.ReasoningContent))) + multiReasoningRunes + fields := []string{ + fmt.Sprintf("role=%s", msg.Role), + fmt.Sprintf("content_runes=%d", len([]rune(strings.TrimSpace(msg.Content)))+multiTextRunes), + fmt.Sprintf("reasoning_runes=%d", reasoningRunes), + fmt.Sprintf("text_parts=%d", textParts), + fmt.Sprintf("reasoning_parts=%d", reasoningParts), + fmt.Sprintf("other_parts=%d", otherParts), + fmt.Sprintf("tool_calls=%d", len(msg.ToolCalls)), + } + if msg.ResponseMeta != nil { + fields = append(fields, fmt.Sprintf("finish_reason=%q", msg.ResponseMeta.FinishReason)) + if usage := msg.ResponseMeta.Usage; usage != nil { + fields = append(fields, + fmt.Sprintf("prompt_tokens=%d", usage.PromptTokens), + fmt.Sprintf("completion_tokens=%d", usage.CompletionTokens), + fmt.Sprintf("total_tokens=%d", usage.TotalTokens), + fmt.Sprintf("reasoning_tokens=%d", usage.CompletionTokensDetails.ReasoningTokens), + ) + } + } + if reasoningRunes > 0 { + fields = append(fields, "hint=模型返回了 reasoning_content 但没有返回可作为摘要正文的 content；请检查 DeepSeek thinking 是否已在摘要请求中关闭") + } + return strings.Join(fields, " ") +} + +func agenticSummaryEmptyDiagnostics(msg *schema.AgenticMessage) string { + if msg == nil { + return "model returned nil agentic message" + } + var textBlocks, reasoningBlocks, otherBlocks int + var textRunes, reasoningRunes int + for _, block := range msg.ContentBlocks { + if block == nil { + continue + } + switch { + case block.AssistantGenText != nil: + textBlocks++ + textRunes += len([]rune(strings.TrimSpace(block.AssistantGenText.Text))) + case block.Reasoning != nil: + reasoningBlocks++ + reasoningRunes += len([]rune(strings.TrimSpace(block.Reasoning.Text))) + default: + otherBlocks++ + } + } + fields := []string{ + fmt.Sprintf("role=%s", msg.Role), + fmt.Sprintf("content_runes=%d", textRunes), + fmt.Sprintf("reasoning_runes=%d", reasoningRunes), + fmt.Sprintf("text_blocks=%d", textBlocks), + fmt.Sprintf("reasoning_blocks=%d", reasoningBlocks), + fmt.Sprintf("other_blocks=%d", otherBlocks), + } + if msg.ResponseMeta != nil && msg.ResponseMeta.TokenUsage != nil { + usage := msg.ResponseMeta.TokenUsage + fields = append(fields, + fmt.Sprintf("prompt_tokens=%d", usage.PromptTokens), + fmt.Sprintf("completion_tokens=%d", usage.CompletionTokens), + fmt.Sprintf("total_tokens=%d", usage.TotalTokens), + fmt.Sprintf("reasoning_tokens=%d", usage.CompletionTokensDetails.ReasoningTokens), + ) + } + if reasoningRunes > 0 { + fields = append(fields, "hint=模型返回了 reasoning block 但没有返回可作为摘要正文的 text block；请检查 DeepSeek thinking 是否已在摘要请求中关闭") + } + return strings.Join(fields, " ") +} diff --git a/internal/multiagent/eino_summarize_model_guard_test.go b/internal/multiagent/eino_summarize_model_guard_test.go new file mode 100644 index 00000000..b52f7c71 --- /dev/null +++ b/internal/multiagent/eino_summarize_model_guard_test.go @@ -0,0 +1,101 @@ +package multiagent + +import ( + "context" + "strings" + "testing" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" +) + +type guardClassicSummaryModel struct { + out *schema.Message +} + +func (m *guardClassicSummaryModel) Generate(context.Context, []*schema.Message, ...model.Option) (*schema.Message, error) { + return m.out, nil +} + +func (m *guardClassicSummaryModel) Stream(context.Context, []*schema.Message, ...model.Option) (*schema.StreamReader[*schema.Message], error) { + return schema.StreamReaderFromArray([]*schema.Message{m.out}), nil +} + +func TestNonEmptySummaryChatModelReportsEmptyContentDiagnostics(t *testing.T) { + msg := schema.AssistantMessage("", nil) + msg.ReasoningContent = "只返回了思考，没有最终摘要" + msg.ResponseMeta = &schema.ResponseMeta{ + FinishReason: "stop", + Usage: &schema.TokenUsage{ + PromptTokens: 10, + CompletionTokens: 3, + TotalTokens: 13, + CompletionTokensDetails: schema.CompletionTokensDetails{ + ReasoningTokens: 3, + }, + }, + } + _, err := newNonEmptySummaryChatModel(&guardClassicSummaryModel{out: msg}).Generate(context.Background(), nil) + if err == nil { + t.Fatal("expected empty summary error") + } + text := err.Error() + for _, want := range []string{ + "summary content is empty", + "reasoning_runes=", + `finish_reason="stop"`, + "reasoning_tokens=3", + "DeepSeek thinking", + } { + if !strings.Contains(text, want) { + t.Fatalf("error missing %q:\n%s", want, text) + } + } +} + +type guardAgenticSummaryModel struct { + out *schema.AgenticMessage +} + +func (m *guardAgenticSummaryModel) Generate(context.Context, []*schema.AgenticMessage, ...model.Option) (*schema.AgenticMessage, error) { + return m.out, nil +} + +func (m *guardAgenticSummaryModel) Stream(context.Context, []*schema.AgenticMessage, ...model.Option) (*schema.StreamReader[*schema.AgenticMessage], error) { + return schema.StreamReaderFromArray([]*schema.AgenticMessage{m.out}), nil +} + +func TestNonEmptyAgenticSummaryModelReportsEmptyContentDiagnostics(t *testing.T) { + msg := &schema.AgenticMessage{ + Role: schema.AgenticRoleTypeAssistant, + ContentBlocks: []*schema.ContentBlock{ + schema.NewContentBlock(&schema.Reasoning{Text: "只返回了思考，没有最终摘要"}), + }, + ResponseMeta: &schema.AgenticResponseMeta{ + TokenUsage: &schema.TokenUsage{ + PromptTokens: 10, + CompletionTokens: 3, + TotalTokens: 13, + CompletionTokensDetails: schema.CompletionTokensDetails{ + ReasoningTokens: 3, + }, + }, + }, + } + _, err := newNonEmptyAgenticSummaryModel(&guardAgenticSummaryModel{out: msg}).Generate(context.Background(), nil) + if err == nil { + t.Fatal("expected empty summary error") + } + text := err.Error() + for _, want := range []string{ + "summary content is empty", + "reasoning_runes=", + "reasoning_blocks=1", + "reasoning_tokens=3", + "DeepSeek thinking", + } { + if !strings.Contains(text, want) { + t.Fatalf("error missing %q:\n%s", want, text) + } + } +} diff --git a/internal/multiagent/eino_summarize_payload.go b/internal/multiagent/eino_summarize_payload.go index 0c24259d..0f96a68b 100644 --- a/internal/multiagent/eino_summarize_payload.go +++ b/internal/multiagent/eino_summarize_payload.go @@ -21,12 +21,15 @@ func shouldDisableDeepSeekThinkingForSummarization(oa *config.OpenAIConfig) bool if oa == nil { return false } + if oa.IsDeepSeekEndpointOrModel() { + return true + } profile := strings.ToLower(strings.TrimSpace(oa.Reasoning.ProfileEffective())) switch profile { case "deepseek", "deepseek_compat": return true case "", "auto": - return oa.IsDeepSeekEndpointOrModel() + return false default: return false } diff --git a/internal/multiagent/eino_summarize_payload_test.go b/internal/multiagent/eino_summarize_payload_test.go index 273df249..4b6890d1 100644 --- a/internal/multiagent/eino_summarize_payload_test.go +++ b/internal/multiagent/eino_summarize_payload_test.go @@ -52,7 +52,7 @@ func TestStripReasoningFromSummarizationPayloadDisablesDeepSeekThinking(t *testi } } -func TestStripReasoningFromSummarizationPayloadHonorsOpenAICompatProfile(t *testing.T) { +func TestStripReasoningFromSummarizationPayloadDisablesDeepSeekEndpointEvenWithOpenAICompatProfile(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", @@ -66,8 +66,30 @@ func TestStripReasoningFromSummarizationPayloadHonorsOpenAICompatProfile(t *test 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 official DeepSeek endpoint thinking disabled, got %s", s) + } +} + +func TestStripReasoningFromSummarizationPayloadHonorsOpenAICompatProfileForNonDeepSeekEndpoint(t *testing.T) { + in := []byte(`{"model":"deepseek-v4-flash","messages":[],"thinking":{"type":"enabled"},"reasoning_effort":"high"}`) + oa := &config.OpenAIConfig{ + BaseURL: "https://compatible.example.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) + t.Fatalf("expected non-DeepSeek OpenAI-compatible endpoint to strip reasoning fields, got %s", s) } } diff --git a/internal/multiagent/eino_summarize_test.go b/internal/multiagent/eino_summarize_test.go index e4a79371..0997c863 100644 --- a/internal/multiagent/eino_summarize_test.go +++ b/internal/multiagent/eino_summarize_test.go @@ -13,6 +13,7 @@ import ( "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/adk/middlewares/summarization" + "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/schema" "go.uber.org/zap" ) @@ -385,6 +386,84 @@ func TestSummarizeFinalize_MergesSystemMessages(t *testing.T) { } } +func TestEinoSummarizationMiddlewareRetriesWhenSummaryModelReturnsEmpty(t *testing.T) { + t.Parallel() + ctx := context.Background() + emit := false + summaryModel := &capturingClassicChatModel{outputs: []*schema.Message{ + schema.AssistantMessage("", nil), + schema.AssistantMessage("有效摘要：继续验证 SQL 注入路径", nil), + }} + appCfg := &config.Config{} + appCfg.OpenAI.Model = "gpt-4o" + appCfg.OpenAI.MaxTotalTokens = 5000 + appCfg.Database.Path = filepath.Join(t.TempDir(), "cyberstrike.db") + mwCfg := &config.MultiAgentEinoMiddlewareConfig{ + SummarizationEmitInternalEvents: &emit, + SummarizationOutputReserveTokens: 1024, + } + + mw, err := newEinoSummarizationMiddleware(ctx, summaryModel, appCfg, mwCfg, "conv-empty-summary", nil, "", nil) + if err != nil { + t.Fatalf("newEinoSummarizationMiddleware: %v", err) + } + state := &adk.ChatModelAgentState{Messages: []adk.Message{ + schema.SystemMessage("system root"), + schema.UserMessage("授权范围 example.com\n" + strings.Repeat("历史扫描输出 ", 12000)), + schema.AssistantMessage("已记录范围", nil), + schema.UserMessage("继续验证 SQL 注入路径"), + }} + + _, after, err := mw.BeforeModelRewriteState(ctx, state, nil) + if err != nil { + t.Fatalf("BeforeModelRewriteState should retry instead of failing on empty summary: %v", err) + } + if after == nil { + t.Fatal("after state is nil") + } + if summaryModel.calls < 2 { + t.Fatalf("summary model calls=%d, want retry after empty output", summaryModel.calls) + } + joined := joinClassicMessageContent(after.Messages) + for _, want := range []string{"有效摘要", "继续验证 SQL 注入路径", "原始用户输入与约束账本"} { + if !strings.Contains(joined, want) { + t.Fatalf("retried compacted context missing %q:\n%s", want, joined) + } + } + if strings.Contains(joined, "本地压缩摘要") { + t.Fatalf("local fallback should not be used:\n%s", joined) + } +} + +type capturingClassicChatModel struct { + output *schema.Message + outputs []*schema.Message + calls int +} + +func (m *capturingClassicChatModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) { + m.calls++ + if len(m.outputs) > 0 { + idx := m.calls - 1 + if idx >= len(m.outputs) { + idx = len(m.outputs) - 1 + } + return m.outputs[idx], nil + } + if m.output != nil { + return m.output, nil + } + return schema.AssistantMessage("classic answer", nil), nil +} + +func (m *capturingClassicChatModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) { + msg, err := m.Generate(ctx, input, opts...) + if err != nil { + return nil, err + } + return schema.StreamReaderFromArray([]*schema.Message{msg}), nil +} + // assertNoOrphanTool 断言消息列表里的每个 role=tool 消息都能在更前面找到一个 // assistant(tool_calls) 提供相同 ID，否则说明产生了孤儿（触发 LLM 400 的根因）。 func assertNoOrphanTool(t *testing.T, msgs []adk.Message) {