diff --git a/DEPTH_ROADMAP.md b/DEPTH_ROADMAP.md index 1533ee7..71fe00c 100644 --- a/DEPTH_ROADMAP.md +++ b/DEPTH_ROADMAP.md @@ -183,8 +183,8 @@ RBAC 未做,暂以单管理员账号代理;概览口径必须是**系统级* - [ ] mcp-py 算法层去桩:文档解析接 PaddleOCR/magic-pdf(mineru.py:11 返回空块)| M - [ ] 报告原生 PDF(现仅 Word,office/unioffice.go)| M -### [ ] T4.E 编排引擎边角 -- [ ] Map 节点错误传播 + 汇总(eino/graph.go:65 单子项失败无感知)| M +### [~] T4.E 编排引擎边角 +- [x] Map 节点错误传播 + 汇总 ✅ —— writeSection 返回 (body,err) 仅真·LLM 失败计错(预算/无模型是主动降级不算);writeSections 汇总失败数、失败项 Body 带可见标记;mapNode 全失败→置 b.fatalErr(判 failed 非静默 done-空)、部分失败→trace+流式告警。report handleReport 同步提示部分失败。3 单测(全失败/部分精确计数/mapNode 置 fatalErr)。 - [ ] 熔断器接回 failover(harness/circuitbreaker.go 与 llm/failover.go 未接合,可能长卡备用)| M - [ ] Branch else 兜底 + coordinator 专家超时(专家卡死拖垮全局,coordinator.go:109)| M - [ ] DSL 拓扑/节点-工具映射校验(dsl/parser.go:23 TODO,现仅 JSON 格式校验)| M diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 8185918..e78f2ab 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -68,7 +68,7 @@ func (o *Orchestrator) mapNode(ctx context.Context, taskID string, n dsl.Node, b end(fmt.Sprintf("拆出 %d 项:%s", len(items), strings.Join(items, " / ")), nil) o.emit(taskID, fmt.Sprintf("\n> 并行处理 %d 项…\n\n", len(items))) - secs := o.writeSections(ctx, b.query, b.kb, items, tr) // 有界并发,trace 出 section:i 各项 + secs, failed := o.writeSections(ctx, b.query, b.kb, items, tr) // 有界并发,trace 出 section:i 各项 b.sections = secs for _, s := range secs { chunk := "## " + s.Heading + "\n\n" + s.Body + "\n\n" @@ -76,6 +76,20 @@ func (o *Orchestrator) mapNode(ctx context.Context, taskID string, n dsl.Node, b b.answer += chunk b.refs = append(b.refs, s.Heading+":"+s.Body) } + // 错误传播:并行子项不再静默丢失——汇总成败;全失败则置致命错,让任务判 failed 而非 done-空。 + if failed > 0 { + endMap := tr.span("map:"+n.ID+":result", "plan", "并行结果汇总") + if failed >= len(items) && len(items) > 0 { + if b.fatalErr == nil { + b.fatalErr = fmt.Errorf("并行 fan-out 全部 %d 项撰写失败", failed) + } + endMap(fmt.Sprintf("全部 %d 项失败", failed), b.fatalErr) + } else { + note := fmt.Sprintf("%d/%d 项失败(已标注),其余成功", failed, len(items)) + endMap(note, nil) + o.emit(taskID, fmt.Sprintf("\n> ⚠️ %s\n\n", note)) + } + } } // execToolNode 执行工具节点:调 MCP 工具,产出累计进黑板;失败降级不阻断。 diff --git a/sundynix-dispatcher/internal/eino/map_error_test.go b/sundynix-dispatcher/internal/eino/map_error_test.go new file mode 100644 index 0000000..965f9f6 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/map_error_test.go @@ -0,0 +1,70 @@ +package eino + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/sundynix/sundynix-dispatcher/internal/dsl" + "github.com/sundynix/sundynix-dispatcher/internal/llm" +) + +// writeSections:真·LLM 失败应逐项标记(Body 带可见失败标记)并汇总失败数,不静默丢失。 +func TestWriteSections_PropagatesFailures(t *testing.T) { + ll := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) { + return "", errors.New("boom") + }} + o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{}) + + secs, failed := o.writeSections(context.Background(), "主题", "", []string{"A", "B", "C"}, o.tracer("t1")) + if failed != 3 { + t.Fatalf("全失败应 failed=3,got %d", failed) + } + if len(secs) != 3 { + t.Fatalf("应返回 3 个 section,got %d", len(secs)) + } + for _, s := range secs { + if !strings.Contains(s.Body, "撰写失败") { + t.Errorf("失败项 Body 应含可见失败标记,got %q", s.Body) + } + } +} + +// 部分失败:仅命中 heading B 的调用失败 → failed 精确为 1(非 all-or-nothing),A/C 正常。 +func TestWriteSections_PartialFailureCount(t *testing.T) { + ll := &fakeLLM{ready: true, chat: func(m []llm.ChatMessage) (string, error) { + for _, msg := range m { + if strings.Contains(msg.Content, "本章标题:B") { + return "", errors.New("boom-B") + } + } + return "正文", nil + }} + o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{}) + + secs, failed := o.writeSections(context.Background(), "主题", "", []string{"A", "B", "C"}, o.tracer("t1")) + if failed != 1 { + t.Fatalf("仅 B 失败应 failed=1,got %d", failed) + } + if !strings.Contains(secs[1].Body, "撰写失败") { + t.Errorf("B 应标记失败,got %q", secs[1].Body) + } + if strings.Contains(secs[0].Body, "撰写失败") || strings.Contains(secs[2].Body, "撰写失败") { + t.Errorf("A/C 不应被误判失败") + } +} + +// mapNode:并行 fan-out 全部子项失败 → 置 fatalErr,任务判 failed 而非静默 done-空。 +func TestMapNode_AllFailSetsFatal(t *testing.T) { + ll := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) { + return "", errors.New("boom") + }} + o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{}) + b := &board{query: "分析一下"} + + o.mapNode(context.Background(), "t1", dsl.Node{ID: "m1"}, b, o.tracer("t1")) + if b.fatalErr == nil { + t.Fatalf("全部子项失败时应置 fatalErr(否则任务会静默判 done-空)") + } +} diff --git a/sundynix-dispatcher/internal/eino/rag_integration_test.go b/sundynix-dispatcher/internal/eino/rag_integration_test.go index a9175bd..ce2eef4 100644 --- a/sundynix-dispatcher/internal/eino/rag_integration_test.go +++ b/sundynix-dispatcher/internal/eino/rag_integration_test.go @@ -96,7 +96,7 @@ func TestRAG_ReportSectionInjectsRefs(t *testing.T) { }} o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{}) - body := o.writeSection(context.Background(), "杭州旅游", "u1/travel", "西湖", o.tracer("t1"), "section:0") + body, _ := o.writeSection(context.Background(), "杭州旅游", "u1/travel", "西湖", o.tracer("t1"), "section:0") if !strings.Contains(body, ragSnippet) { t.Errorf("报告章节撰写 prompt 应含检索资料,got %q", body) } diff --git a/sundynix-dispatcher/internal/eino/report.go b/sundynix-dispatcher/internal/eino/report.go index a825082..45b65b7 100644 --- a/sundynix-dispatcher/internal/eino/report.go +++ b/sundynix-dispatcher/internal/eino/report.go @@ -7,6 +7,7 @@ import ( "log" "strings" "sync" + "sync/atomic" "time" "github.com/sundynix/sundynix-dispatcher/internal/dsl" @@ -75,7 +76,12 @@ func (o *Orchestrator) handleReport(ctx context.Context, t *contract.Task, tr *e o.emit(t.ID, "\n> 正在并行撰写各章…\n\n") } - sections := o.writeSections(ctx, topic, kb, outline.Sections, tr) + sections, failed := o.writeSections(ctx, topic, kb, outline.Sections, tr) + if failed > 0 { + note := fmt.Sprintf("⚠️ %d/%d 章撰写失败(下方已标注),其余照常成稿。", failed, len(sections)) + tr.info("plan", "system", "部分章节失败", note) + o.emit(t.ID, "\n> "+note+"\n\n") + } // 把完整报告正文流式呈现给客户端。 o.emit(t.ID, "\n---\n\n# "+firstNonEmpty(outline.Title, topic)+"\n\n") @@ -162,11 +168,13 @@ func (o *Orchestrator) planItems(ctx context.Context, topic, splitBy string) []s return items } -// writeSections 各章节并行撰写(有界并发),结果按原顺序返回。 -func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, headings []string, tr *execTracer) []reportSection { +// writeSections 各章节并行撰写(有界并发),结果按原顺序返回;第二返回值为失败项数。 +// 失败项的 Body 会带明确的失败标记(可见,不静默),失败计数供上游汇总/判定整体成败。 +func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, headings []string, tr *execTracer) ([]reportSection, int) { out := make([]reportSection, len(headings)) sem := make(chan struct{}, reportFanout) var wg sync.WaitGroup + var failed atomic.Int64 for i, h := range headings { wg.Add(1) go func(i int, h string) { @@ -175,26 +183,33 @@ func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, head defer func() { <-sem }() node := fmt.Sprintf("section:%d", i) end := tr.span(node, "section", fmt.Sprintf("第%d章 %s", i+1, h)) - body := o.writeSection(ctx, topic, kb, h, tr, node) - end(fmt.Sprintf("成稿 %d 字", len([]rune(body))), nil) + body, err := o.writeSection(ctx, topic, kb, h, tr, node) + if err != nil { + failed.Add(1) + end("撰写失败:"+err.Error(), err) // trace 记为该 section 失败(非静默) + } else { + end(fmt.Sprintf("成稿 %d 字", len([]rune(body))), nil) + } out[i] = reportSection{Heading: h, Body: body} }(i, h) } wg.Wait() - return out + return out, int(failed.Load()) } // writeSection 撰写一章:先 RAG 检索参考资料(若挂了知识库),再让模型成稿。 -func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading string, tr *execTracer, node string) string { +// 返回 (正文, 失败错误):仅「真·LLM 调用失败」返回非 nil err;预算触顶 / 模型未配置是 +// 主动降级(出可见的降级正文、err=nil),不计入失败。 +func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading string, tr *execTracer, node string) (string, error) { refs := o.retrieve(ctx, kb, topic+" "+heading) if refs != "" { tr.info(node, "section", "检索参考资料", truncate(strings.ReplaceAll(refs, "\n", " "), 120)) } if !o.pool.Ready() { if refs != "" { - return "(模型未配置,以下为检索到的参考资料)\n" + refs + return "(模型未配置,以下为检索到的参考资料)\n" + refs, nil } - return "(模型未配置,无法撰写本章。)" + return "(模型未配置,无法撰写本章。)", nil } sys := "你是专业报告撰稿人,语言严谨、条理清晰,使用中文书面语。" var ub strings.Builder @@ -210,7 +225,7 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri bud.AddPrompt(sys + ub.String()) if bud.Exceeded() { tr.info(node, "system", "token 预算", "已达预算上限,跳过本章") - return "(已达 token 预算上限,本章自动跳过。)" + return "(已达 token 预算上限,本章自动跳过。)", nil } } cctx, cancel := llmCtx(ctx) @@ -218,12 +233,12 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri txt, err := o.pool.Chat(cctx, []llm.ChatMessage{{Role: "system", Content: sys}, {Role: "user", Content: ub.String()}}) if err != nil { log.Printf("[report] 撰写「%s」失败: %v", heading, err) - return "(本章撰写失败:" + err.Error() + ")" + return "(本章撰写失败:" + err.Error() + ")", err } if bud := harness.BudgetFrom(ctx); bud != nil { bud.AddComplete(txt) // 成本护栏:计入输出 } - return strings.TrimSpace(txt) + return strings.TrimSpace(txt), nil } // retrieve 经 Eino Retriever 组件(包 mcp-go kb_search)检索知识库,整理为可读参考资料。