mirror of
https://github.com/Ed1s0nZ/CyberStrikeAI.git
synced 2026-08-29 22:20:37 +02:00
fix: surface summarization model errors
This commit is contained in:
@@ -747,7 +747,7 @@ func (h *AgentHandler) finalizeRobotAgentError(ctx context.Context, assistantMes
|
|||||||
if shouldPersistEinoAgentTraceAfterRunError(ctx) {
|
if shouldPersistEinoAgentTraceAfterRunError(ctx) {
|
||||||
h.persistEinoAgentTraceForResume(conversationID, resultMA)
|
h.persistEinoAgentTraceForResume(conversationID, resultMA)
|
||||||
}
|
}
|
||||||
errMsg := "执行失败: " + errMA.Error()
|
errMsg := "执行失败: " + multiagent.EinoClientRunErrorMessage(errMA)
|
||||||
if assistantMessageID != "" {
|
if assistantMessageID != "" {
|
||||||
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
||||||
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
||||||
|
|||||||
@@ -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))
|
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 assistantMessageID != "" {
|
||||||
if _, updateErr := h.db.Exec(
|
if _, updateErr := h.db.Exec(
|
||||||
"UPDATE messages SET content = ?, updated_at = ? WHERE id = ?",
|
"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.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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -371,15 +371,17 @@ func (h *AgentHandler) EinoSingleAgentLoopStream(c *gin.Context) {
|
|||||||
h.logger.Error("Eino ADK 单代理执行失败", zap.Error(runErr))
|
h.logger.Error("Eino ADK 单代理执行失败", zap.Error(runErr))
|
||||||
taskStatus = "failed"
|
taskStatus = "failed"
|
||||||
h.tasks.UpdateTaskStatus(conversationID, taskStatus)
|
h.tasks.UpdateTaskStatus(conversationID, taskStatus)
|
||||||
errMsg := "执行失败: " + runErr.Error()
|
clientErr := multiagent.EinoClientRunErrorMessage(runErr)
|
||||||
|
errMsg := "执行失败: " + clientErr
|
||||||
if assistantMessageID != "" {
|
if assistantMessageID != "" {
|
||||||
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
||||||
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
||||||
}
|
}
|
||||||
sendEvent("error", errMsg, map[string]interface{}{
|
errData := multiagent.EinoClientRunErrorFields(runErr)
|
||||||
"conversationId": conversationID,
|
errData["conversationId"] = conversationID
|
||||||
"messageId": assistantMessageID,
|
errData["messageId"] = assistantMessageID
|
||||||
})
|
errData["error"] = errMsg
|
||||||
|
sendEvent("error", errMsg, errData)
|
||||||
sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
|
sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
|
||||||
timeoutCancel()
|
timeoutCancel()
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -385,15 +385,17 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) {
|
|||||||
h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
|
h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
|
||||||
taskStatus = "failed"
|
taskStatus = "failed"
|
||||||
h.tasks.UpdateTaskStatus(conversationID, taskStatus)
|
h.tasks.UpdateTaskStatus(conversationID, taskStatus)
|
||||||
errMsg := "执行失败: " + runErr.Error()
|
clientErr := multiagent.EinoClientRunErrorMessage(runErr)
|
||||||
|
errMsg := "执行失败: " + clientErr
|
||||||
if assistantMessageID != "" {
|
if assistantMessageID != "" {
|
||||||
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), assistantMessageID)
|
||||||
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
_ = h.db.AddProcessDetail(assistantMessageID, conversationID, "error", errMsg, nil)
|
||||||
}
|
}
|
||||||
sendEvent("error", errMsg, map[string]interface{}{
|
errData := multiagent.EinoClientRunErrorFields(runErr)
|
||||||
"conversationId": conversationID,
|
errData["conversationId"] = conversationID
|
||||||
"messageId": assistantMessageID,
|
errData["messageId"] = assistantMessageID
|
||||||
})
|
errData["error"] = errMsg
|
||||||
|
sendEvent("error", errMsg, errData)
|
||||||
sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
|
sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
|
||||||
timeoutCancel()
|
timeoutCancel()
|
||||||
return
|
return
|
||||||
@@ -513,11 +515,14 @@ func (h *AgentHandler) MultiAgentLoop(c *gin.Context) {
|
|||||||
h.persistEinoAgentTraceForResume(prep.ConversationID, result)
|
h.persistEinoAgentTraceForResume(prep.ConversationID, result)
|
||||||
}
|
}
|
||||||
h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
|
h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
|
||||||
errMsg := "执行失败: " + runErr.Error()
|
clientErr := multiagent.EinoClientRunErrorMessage(runErr)
|
||||||
|
errMsg := "执行失败: " + clientErr
|
||||||
if prep.AssistantMessageID != "" {
|
if prep.AssistantMessageID != "" {
|
||||||
_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), 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
|
return
|
||||||
}
|
}
|
||||||
mw := &h.config.MultiAgent.EinoMiddleware
|
mw := &h.config.MultiAgent.EinoMiddleware
|
||||||
|
|||||||
@@ -12,15 +12,24 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type capturingAgenticChatModel struct {
|
type capturingAgenticChatModel struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
inputs [][]*schema.AgenticMessage
|
inputs [][]*schema.AgenticMessage
|
||||||
output *schema.AgenticMessage
|
output *schema.AgenticMessage
|
||||||
|
outputs []*schema.AgenticMessage
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *capturingAgenticChatModel) Generate(_ context.Context, input []*schema.AgenticMessage, _ ...model.Option) (*schema.AgenticMessage, error) {
|
func (m *capturingAgenticChatModel) Generate(_ context.Context, input []*schema.AgenticMessage, _ ...model.Option) (*schema.AgenticMessage, error) {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
m.inputs = append(m.inputs, input)
|
m.inputs = append(m.inputs, input)
|
||||||
|
callNo := len(m.inputs)
|
||||||
m.mu.Unlock()
|
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 {
|
if m.output != nil {
|
||||||
return m.output, nil
|
return m.output, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -111,7 +111,7 @@ func newEinoAgenticSummarizationMiddleware(
|
|||||||
summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "agentic", &appCfg.OpenAI, logger)
|
summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "agentic", &appCfg.OpenAI, logger)
|
||||||
|
|
||||||
mw, err := summarization.NewTyped[*schema.AgenticMessage](ctx, &summarization.TypedConfig[*schema.AgenticMessage]{
|
mw, err := summarization.NewTyped[*schema.AgenticMessage](ctx, &summarization.TypedConfig[*schema.AgenticMessage]{
|
||||||
Model: summaryModel,
|
Model: newNonEmptyAgenticSummaryModel(summaryModel),
|
||||||
ModelOptions: summaryModelOpts,
|
ModelOptions: summaryModelOpts,
|
||||||
GenModelInput: func(ctx context.Context, sysInstruction, userInstruction *schema.AgenticMessage, originalMsgs []*schema.AgenticMessage) ([]*schema.AgenticMessage, error) {
|
GenModelInput: func(ctx context.Context, sysInstruction, userInstruction *schema.AgenticMessage, originalMsgs []*schema.AgenticMessage) ([]*schema.AgenticMessage, error) {
|
||||||
classicOriginal := AgenticMessagesToEino(originalMsgs)
|
classicOriginal := AgenticMessagesToEino(originalMsgs)
|
||||||
|
|||||||
@@ -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("<summary>有效摘要:继续验证 SQL 注入路径</summary>"),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
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) {
|
func TestAppendEinoAgenticChatModelTailMiddlewaresIncludesTypedSummarization(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
mw := newAgenticSystemMessageNormalizerMiddleware(nil, "summary")
|
mw := newAgenticSystemMessageNormalizerMiddleware(nil, "summary")
|
||||||
|
|||||||
@@ -115,11 +115,7 @@ func (h *einoRunErrorHandler) emitError(err error, kind string) {
|
|||||||
} else if userErr.retryExhausted {
|
} else if userErr.retryExhausted {
|
||||||
data["hasModelOriginalError"] = false
|
data["hasModelOriginalError"] = false
|
||||||
}
|
}
|
||||||
message := err.Error()
|
h.progress("error", EinoClientRunErrorMessage(err), data)
|
||||||
if userErr.message != "" {
|
|
||||||
message = userErr.message
|
|
||||||
}
|
|
||||||
h.progress("error", message, data)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type einoRunUserError struct {
|
type einoRunUserError struct {
|
||||||
@@ -131,6 +127,7 @@ type einoRunUserError struct {
|
|||||||
retryExhausted bool
|
retryExhausted bool
|
||||||
totalRetries int
|
totalRetries int
|
||||||
hasModelOriginalError bool
|
hasModelOriginalError bool
|
||||||
|
summarizationModelErr bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func einoUserFacingRunError(err error) einoRunUserError {
|
func einoUserFacingRunError(err error) einoRunUserError {
|
||||||
@@ -152,6 +149,10 @@ func einoUserFacingRunError(err error) einoRunUserError {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
out.rawLastError = strings.TrimSpace(lastErr.Error())
|
out.rawLastError = strings.TrimSpace(lastErr.Error())
|
||||||
|
if raw, ok := einoSummarizationModelRawErrorText(lastErr); ok {
|
||||||
|
out.rawLastError = raw
|
||||||
|
out.summarizationModelErr = true
|
||||||
|
}
|
||||||
if isEinoShouldRetryOutputRejected(lastErr) {
|
if isEinoShouldRetryOutputRejected(lastErr) {
|
||||||
out.kind = "model_output_rejected"
|
out.kind = "model_output_rejected"
|
||||||
out.summary = "模型未返回原始错误;输出被重试策略拒绝。"
|
out.summary = "模型未返回原始错误;输出被重试策略拒绝。"
|
||||||
@@ -163,6 +164,9 @@ func einoUserFacingRunError(err error) einoRunUserError {
|
|||||||
if strings.TrimSpace(summary) == "" {
|
if strings.TrimSpace(summary) == "" {
|
||||||
summary = einoTrimRetryErrorSummary(lastErr.Error())
|
summary = einoTrimRetryErrorSummary(lastErr.Error())
|
||||||
}
|
}
|
||||||
|
if out.summarizationModelErr {
|
||||||
|
summary = einoTrimRetryErrorSummary(out.rawLastError)
|
||||||
|
}
|
||||||
if kind == "" {
|
if kind == "" {
|
||||||
kind = "model_retry_exhausted"
|
kind = "model_retry_exhausted"
|
||||||
}
|
}
|
||||||
@@ -190,3 +194,105 @@ func formatEinoRetryExhaustedMessage(summary string, totalRetries int) string {
|
|||||||
}
|
}
|
||||||
return "模型调用重试已耗尽:" + summary
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -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) {
|
func TestEinoRunErrorHandlerIterationLimitProgress(t *testing.T) {
|
||||||
var events []string
|
var events []string
|
||||||
var errorKind interface{}
|
var errorKind interface{}
|
||||||
|
|||||||
@@ -167,7 +167,7 @@ func newEinoSummarizationMiddleware(
|
|||||||
summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "classic", &appCfg.OpenAI, logger)
|
summaryModelOpts := newEinoSummarizationModelOptions(outputReserve, modelName, "classic", &appCfg.OpenAI, logger)
|
||||||
|
|
||||||
mw, err := summarization.New(ctx, &summarization.Config{
|
mw, err := summarization.New(ctx, &summarization.Config{
|
||||||
Model: summaryModel,
|
Model: newNonEmptySummaryChatModel(summaryModel),
|
||||||
ModelOptions: summaryModelOpts,
|
ModelOptions: summaryModelOpts,
|
||||||
GenModelInput: func(ctx context.Context, sysInstruction, userInstruction adk.Message, originalMsgs []adk.Message) ([]adk.Message, error) {
|
GenModelInput: func(ctx context.Context, sysInstruction, userInstruction adk.Message, originalMsgs []adk.Message) ([]adk.Message, error) {
|
||||||
if transcriptPath != "" && len(originalMsgs) > 0 {
|
if transcriptPath != "" && len(originalMsgs) > 0 {
|
||||||
|
|||||||
@@ -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, " ")
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -21,12 +21,15 @@ func shouldDisableDeepSeekThinkingForSummarization(oa *config.OpenAIConfig) bool
|
|||||||
if oa == nil {
|
if oa == nil {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
if oa.IsDeepSeekEndpointOrModel() {
|
||||||
|
return true
|
||||||
|
}
|
||||||
profile := strings.ToLower(strings.TrimSpace(oa.Reasoning.ProfileEffective()))
|
profile := strings.ToLower(strings.TrimSpace(oa.Reasoning.ProfileEffective()))
|
||||||
switch profile {
|
switch profile {
|
||||||
case "deepseek", "deepseek_compat":
|
case "deepseek", "deepseek_compat":
|
||||||
return true
|
return true
|
||||||
case "", "auto":
|
case "", "auto":
|
||||||
return oa.IsDeepSeekEndpointOrModel()
|
return false
|
||||||
default:
|
default:
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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"}`)
|
in := []byte(`{"model":"deepseek-v4-flash","messages":[],"thinking":{"type":"enabled"},"reasoning_effort":"high"}`)
|
||||||
oa := &config.OpenAIConfig{
|
oa := &config.OpenAIConfig{
|
||||||
BaseURL: "https://api.deepseek.com/v1",
|
BaseURL: "https://api.deepseek.com/v1",
|
||||||
@@ -66,8 +66,30 @@ func TestStripReasoningFromSummarizationPayloadHonorsOpenAICompatProfile(t *test
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
s := string(out)
|
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") {
|
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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
|
|
||||||
"github.com/cloudwego/eino/adk"
|
"github.com/cloudwego/eino/adk"
|
||||||
"github.com/cloudwego/eino/adk/middlewares/summarization"
|
"github.com/cloudwego/eino/adk/middlewares/summarization"
|
||||||
|
"github.com/cloudwego/eino/components/model"
|
||||||
"github.com/cloudwego/eino/schema"
|
"github.com/cloudwego/eino/schema"
|
||||||
"go.uber.org/zap"
|
"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("<summary>有效摘要:继续验证 SQL 注入路径</summary>", 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 消息都能在更前面找到一个
|
// assertNoOrphanTool 断言消息列表里的每个 role=tool 消息都能在更前面找到一个
|
||||||
// assistant(tool_calls) 提供相同 ID,否则说明产生了孤儿(触发 LLM 400 的根因)。
|
// assistant(tool_calls) 提供相同 ID,否则说明产生了孤儿(触发 LLM 400 的根因)。
|
||||||
func assertNoOrphanTool(t *testing.T, msgs []adk.Message) {
|
func assertNoOrphanTool(t *testing.T, msgs []adk.Message) {
|
||||||
|
|||||||
Reference in New Issue
Block a user