refactor(dispatcher): T1.1 退役 graph.go —— 编排引擎收成单一 compose
compose 已默认数天、HITL/多智能体/评测全在其上真实跑过,soak 充分。删掉自研拓扑 解释器这第二套引擎,消灭"双实现 drift"税: - 删 runGraph(graph.go 的自研解释器)+ composeEnabled/EINO_COMPOSE 逃生舱开关 + runConversation(仅 runGraph 用的死代码)。 - executeGraph 直接走 runComposeGraph;compose 编译失败兜底改单轮对话(不再回退 graph.go);清掉仅 runGraph 用的 import(otel attribute/trace/otelx)。 - 保留 board / 各节点执行器(retriever/tool/agent/branch/approval/map/render/aggregate) / 工具函数 —— 它们是 compose 各节点 lambda 复用的,非 graph.go 专属。 - 测试:8 处 runGraph→runComposeGraph;等价测试(对照两引擎)转为 compose 正确性测试; runConversation 的开关测试转为「无 ChatModel 降级 runAgent」。 go test ./... 全绿 + vet 干净;冒烟 简单 agent/分支图 跑通。此后每个编排改动不再两边 对齐,成本减半。DEPTH_ROADMAP T1.1 ✅。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+9
-6
@@ -39,12 +39,15 @@
|
|||||||
|
|
||||||
## 🥈 Tier 1 — 收口 / 去税
|
## 🥈 Tier 1 — 收口 / 去税
|
||||||
|
|
||||||
### [ ] T1.1 退役 graph.go — 消灭双引擎
|
### [x] T1.1 退役 graph.go — 消灭双引擎 ✅
|
||||||
现状:compose 已默认,HITL/多智能体只在 compose 上;graph.go 是逃生舱。
|
现状:compose 已默认,HITL/多智能体只在 compose 上;graph.go 是逃生舱。
|
||||||
- [ ] compose soak 一段真实流量(无回归)
|
- [x] compose soak(默认跑数天,HITL/多智能体/评测全在其上真实运行)
|
||||||
- [ ] 删 graph.go(runGraph 及其专属分支)+ `EINO_COMPOSE` 开关 + 降级回退
|
- [x] 删 `runGraph`(自研拓扑解释器)+ `composeEnabled`/`EINO_COMPOSE` 开关 + `runConversation` 死代码
|
||||||
- [ ] 把仅 graph.go 用到的辅助函数收编/删除,等价测试转为只测 compose
|
- [x] compose 编译失败兜底改单轮对话(不再回退 graph.go);清掉仅 runGraph 用的 import(otel)
|
||||||
- 验收:单引擎,全测试绿,HITL/多智能体/分支/map 全在 compose 上。
|
- [x] 保留 board/各节点执行器/工具函数(compose 各 lambda 复用);8 处测试 runGraph→runComposeGraph,
|
||||||
|
等价测试转为 compose 正确性测试,runConversation 测试转为「无 ChatModel 降级 runAgent」
|
||||||
|
- 验收 ✅:单引擎,go test ./... 全绿 + vet 干净;冒烟 简单 agent/分支图 跑通;HITL/多智能体/branch/map
|
||||||
|
全在 compose 上。
|
||||||
|
|
||||||
### [ ] T1.2 前端深度补齐
|
### [ ] T1.2 前端深度补齐
|
||||||
- [ ] 评测质量面板(运行页展示 overall/质量/忠实/纠偏/flags,驱动 T0.1 的可见性)
|
- [ ] 评测质量面板(运行页展示 overall/质量/忠实/纠偏/flags,驱动 T0.1 的可见性)
|
||||||
@@ -92,7 +95,7 @@
|
|||||||
| Tier | 完成 / 总 |
|
| Tier | 完成 / 总 |
|
||||||
|---|---|
|
|---|---|
|
||||||
| T0 激活 | 2 / 2 ✅ |
|
| T0 激活 | 2 / 2 ✅ |
|
||||||
| T1 收口 | 0 / 2 |
|
| T1 收口 | 1 / 2 |
|
||||||
| T2 深化 | 0 / 3 |
|
| T2 深化 | 0 / 3 |
|
||||||
| T3 硬化 ⏸ | 0 / 5 |
|
| T3 硬化 ⏸ | 0 / 5 |
|
||||||
|
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ func TestAgentCollaborationPassesOutput(t *testing.T) {
|
|||||||
return "研究产出XYZ"
|
return "研究产出XYZ"
|
||||||
}}
|
}}
|
||||||
o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
|
o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
|
||||||
ans, _, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
|
ans, _, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,13 +26,10 @@ func registerFlowMerge() {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// executeGraph 按灰度开关选编排实现:compose.Graph(Phase C)或自研 graph.go(默认/权威)。
|
// executeGraph 把任务 DSL 图编译为 Eino compose.Graph 并执行(唯一编排引擎;自研解释器 graph.go
|
||||||
// 返回 (成稿, 检索来源, error);来源供忠实度评测,两条路径都回传(compose 已对齐 graph.go)。
|
// 的 runGraph 已退役)。返回 (成稿, 检索来源, error);来源供忠实度评测。
|
||||||
func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
|
func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
|
||||||
if composeEnabled() {
|
return o.runComposeGraph(ctx, t, tr)
|
||||||
return o.runComposeGraph(ctx, t, tr)
|
|
||||||
}
|
|
||||||
return o.runGraph(ctx, t, tr)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行(Phase C 编排归一):
|
// runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行(Phase C 编排归一):
|
||||||
@@ -212,8 +209,10 @@ func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, t
|
|||||||
}
|
}
|
||||||
r, cerr := g.Compile(ctx, compileOpts...)
|
r, cerr := g.Compile(ctx, compileOpts...)
|
||||||
if cerr != nil {
|
if cerr != nil {
|
||||||
tr.info("task", "system", "compose 编译失败", "退回自研 graph.go:"+cerr.Error())
|
// 编译失败(罕见)→ 降级为单轮对话兜底(自研 graph.go 已退役,不再回退)。
|
||||||
return o.runGraph(ctx, t, tr) // 降级回权威实现(带 refs / 终态)
|
tr.info("task", "system", "compose 编译失败", "降级单轮对话:"+cerr.Error())
|
||||||
|
o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
|
||||||
|
return b.answer, refsOf(b), b.fatalErr
|
||||||
}
|
}
|
||||||
// checkpoint id = task id:审批中断时 compose 据此把整图状态(含 board)落进 store。
|
// checkpoint id = task id:审批中断时 compose 据此把整图状态(含 board)落进 store。
|
||||||
var invokeOpts []compose.Option
|
var invokeOpts []compose.Option
|
||||||
|
|||||||
@@ -28,21 +28,15 @@ func echoLLM() *fakeLLM {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func runBoth(t *testing.T, graph string) (interp, comp string) {
|
// runCompose 跑一个图并返回成稿(自研 graph.go 已退役,编排引擎统一为 compose)。
|
||||||
|
func runCompose(t *testing.T, graph string) string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
task := &contract.Task{ID: "t_eq", Graph: []byte(graph)}
|
o := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
|
||||||
|
ans, _, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "t_eq", Graph: []byte(graph)}, &execTracer{})
|
||||||
o1 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
|
|
||||||
a1, _, err := o1.runGraph(context.Background(), task, &execTracer{})
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("runGraph: %v", err)
|
|
||||||
}
|
|
||||||
o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
|
|
||||||
a2, _, err := o2.runComposeGraph(context.Background(), task, &execTracer{})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("runComposeGraph: %v", err)
|
t.Fatalf("runComposeGraph: %v", err)
|
||||||
}
|
}
|
||||||
return a1, a2
|
return ans
|
||||||
}
|
}
|
||||||
|
|
||||||
// rejectWaiter 是恒拒绝的 HITL 审批替身(fail-safe 路径回归用)。
|
// rejectWaiter 是恒拒绝的 HITL 审批替身(fail-safe 路径回归用)。
|
||||||
@@ -257,8 +251,8 @@ func TestComposeReturnsRefs(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestComposeEquivalentLinear 多节点线性图:input→memory→agent→output,两路径成稿应逐字一致。
|
// TestComposeLinear 多节点线性图:input→memory→agent→output,compose 应跑通并产出成稿。
|
||||||
func TestComposeEquivalentLinear(t *testing.T) {
|
func TestComposeLinear(t *testing.T) {
|
||||||
graph := `{"version":"1","nodes":[
|
graph := `{"version":"1","nodes":[
|
||||||
{"id":"in","kind":"input","config":{"text":"什么是图编排"}},
|
{"id":"in","kind":"input","config":{"text":"什么是图编排"}},
|
||||||
{"id":"m","kind":"memory","config":{}},
|
{"id":"m","kind":"memory","config":{}},
|
||||||
@@ -267,14 +261,14 @@ func TestComposeEquivalentLinear(t *testing.T) {
|
|||||||
],"edges":[
|
],"edges":[
|
||||||
{"source":"in","target":"m"},{"source":"m","target":"a"},{"source":"a","target":"out"}
|
{"source":"in","target":"m"},{"source":"m","target":"a"},{"source":"a","target":"out"}
|
||||||
]}`
|
]}`
|
||||||
interp, comp := runBoth(t, graph)
|
if ans := runCompose(t, graph); !strings.Contains(ans, "ANS:") {
|
||||||
if interp == "" || interp != comp {
|
t.Fatalf("线性图 compose 应产出成稿,got %q", ans)
|
||||||
t.Fatalf("线性图不等价: interp=%q compose=%q", interp, comp)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestComposeEquivalentBranch 分支图:input→branch→(真)A/(假)B,条件恒真应都走 A,两路径一致。
|
// TestComposeBranch 分支图:input→branch→(真)A/(假)B,条件恒真走 A,compose 应跑通产出。
|
||||||
func TestComposeEquivalentBranch(t *testing.T) {
|
// 分支真假边的精确选路另由 TestRunGraph_BranchRouting(compose)钉死。
|
||||||
|
func TestComposeBranch(t *testing.T) {
|
||||||
graph := `{"version":"1","nodes":[
|
graph := `{"version":"1","nodes":[
|
||||||
{"id":"in","kind":"input","config":{"text":"hi"}},
|
{"id":"in","kind":"input","config":{"text":"hi"}},
|
||||||
{"id":"br","kind":"branch","config":{"condition":""}},
|
{"id":"br","kind":"branch","config":{"condition":""}},
|
||||||
@@ -285,8 +279,7 @@ func TestComposeEquivalentBranch(t *testing.T) {
|
|||||||
{"source":"br","target":"a","sourceHandle":"true"},
|
{"source":"br","target":"a","sourceHandle":"true"},
|
||||||
{"source":"br","target":"b","sourceHandle":"false"}
|
{"source":"br","target":"b","sourceHandle":"false"}
|
||||||
]}`
|
]}`
|
||||||
interp, comp := runBoth(t, graph)
|
if ans := runCompose(t, graph); !strings.Contains(ans, "ANS:") {
|
||||||
if interp == "" || interp != comp {
|
t.Fatalf("分支图 compose 应产出成稿,got %q", ans)
|
||||||
t.Fatalf("分支图不等价: interp=%q compose=%q", interp, comp)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -14,28 +13,13 @@ import (
|
|||||||
"github.com/sundynix/sundynix-dispatcher/internal/harness"
|
"github.com/sundynix/sundynix-dispatcher/internal/harness"
|
||||||
)
|
)
|
||||||
|
|
||||||
// composeEnabled 报告是否启用 compose.Graph 编排路径。
|
|
||||||
// Phase C 灰度收尾:compose 已与 graph.go 终态对齐(refs / 审批拒绝 / 预算·模型失败 全部传播),
|
|
||||||
// 默认翻为 compose 运行时;保留逃生舱 EINO_COMPOSE=0 → 回退权威 graph.go。soak 无回归后即可退役 graph.go。
|
|
||||||
func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") != "0" }
|
|
||||||
|
|
||||||
// runConversation 是对话/模型节点的统一入口:按灰度开关选 compose.Graph 或自研 runAgent。
|
|
||||||
// 二者对外行为一致(据黑板拼消息 → 流式回流 token → 累计成稿),便于等价回归。
|
|
||||||
func (o *Orchestrator) runConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
|
|
||||||
if composeEnabled() {
|
|
||||||
o.runComposeConversation(ctx, taskID, b, system, tr, node)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
o.runAgent(ctx, taskID, b, system, tr, node)
|
|
||||||
}
|
|
||||||
|
|
||||||
// runComposeConversation 用 Eino compose.Graph 跑对话主流程:
|
// runComposeConversation 用 Eino compose.Graph 跑对话主流程:
|
||||||
// START → ChatModel 节点 → END,编译为 Runnable 后流式执行;可观测经 callbacks 桥到 ExecEvent。
|
// START → ChatModel 节点 → END,编译为 Runnable 后流式执行;可观测经 callbacks 桥到 ExecEvent。
|
||||||
// 模型未就绪 / 编译失败时降级回自研 runAgent,保证不回归。
|
// 模型未就绪 / 编译失败时降级回 runAgent(同样的流式回流,保证不回归)。
|
||||||
func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
|
func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
|
||||||
cm := o.pool.ChatModel()
|
cm := o.pool.ChatModel()
|
||||||
if cm == nil {
|
if cm == nil {
|
||||||
o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 走自研路径的降级桩
|
o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 降级桩
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -50,10 +50,8 @@ func TestComposeGraphRuns(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestComposeConversationEquivalent 验证 EINO_COMPOSE=1 时对话主流程走 compose 路径,
|
// TestComposeConversation 验证对话主流程走 compose 路径:模型输出流式回流到 sink 并累计成稿。
|
||||||
// 把模型输出流式回流到 sink 并累计成稿——与自研 runAgent 行为等价(都逐片转发模型输出)。
|
func TestComposeConversation(t *testing.T) {
|
||||||
func TestComposeConversationEquivalent(t *testing.T) {
|
|
||||||
t.Setenv("EINO_COMPOSE", "1")
|
|
||||||
fs := &fakeSink{}
|
fs := &fakeSink{}
|
||||||
o := &Orchestrator{
|
o := &Orchestrator{
|
||||||
pool: &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}},
|
pool: &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}},
|
||||||
@@ -61,8 +59,7 @@ func TestComposeConversationEquivalent(t *testing.T) {
|
|||||||
sink: fs,
|
sink: fs,
|
||||||
}
|
}
|
||||||
b := &board{query: "你好"}
|
b := &board{query: "你好"}
|
||||||
tr := &execTracer{} // sink 为 nil → 轨迹发射空操作
|
o.runComposeConversation(context.Background(), "task_compose", b, "", &execTracer{}, "agent")
|
||||||
o.runConversation(context.Background(), "task_compose", b, "", tr, "agent")
|
|
||||||
|
|
||||||
if !strings.Contains(b.answer, "compose 路径") {
|
if !strings.Contains(b.answer, "compose 路径") {
|
||||||
t.Fatalf("成稿未含模型输出: %q", b.answer)
|
t.Fatalf("成稿未含模型输出: %q", b.answer)
|
||||||
@@ -72,18 +69,18 @@ func TestComposeConversationEquivalent(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestComposeDisabledUsesRunAgent 验证开关关闭时仍走自研 runAgent(默认行为不变)。
|
// TestComposeConversationDegradesToRunAgent 验证无 Eino ChatModel 时降级回 runAgent(流式回流不丢)。
|
||||||
func TestComposeDisabledUsesRunAgent(t *testing.T) {
|
func TestComposeConversationDegradesToRunAgent(t *testing.T) {
|
||||||
t.Setenv("EINO_COMPOSE", "0")
|
|
||||||
fs := &fakeSink{}
|
fs := &fakeSink{}
|
||||||
o := &Orchestrator{
|
o := &Orchestrator{
|
||||||
pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "自研路径回答" }},
|
// cm 为 nil → ChatModel() 返回 nil → runComposeConversation 降级 runAgent(走 stream 桩)。
|
||||||
|
pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "降级路径回答" }},
|
||||||
breaker: harness.NewCircuitBreaker(),
|
breaker: harness.NewCircuitBreaker(),
|
||||||
sink: fs,
|
sink: fs,
|
||||||
}
|
}
|
||||||
b := &board{query: "你好"}
|
b := &board{query: "你好"}
|
||||||
o.runConversation(context.Background(), "task_legacy", b, "", &execTracer{}, "agent")
|
o.runComposeConversation(context.Background(), "task_degrade", b, "", &execTracer{}, "agent")
|
||||||
if !strings.Contains(b.answer, "自研路径") {
|
if !strings.Contains(b.answer, "降级路径") {
|
||||||
t.Fatalf("未走自研路径: %q", b.answer)
|
t.Fatalf("无 ChatModel 应降级 runAgent 出稿: %q", b.answer)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,13 +8,10 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/cloudwego/eino/schema"
|
"github.com/cloudwego/eino/schema"
|
||||||
"go.opentelemetry.io/otel/attribute"
|
|
||||||
"go.opentelemetry.io/otel/trace"
|
|
||||||
|
|
||||||
"github.com/sundynix/sundynix-dispatcher/internal/dsl"
|
"github.com/sundynix/sundynix-dispatcher/internal/dsl"
|
||||||
"github.com/sundynix/sundynix-dispatcher/internal/harness"
|
"github.com/sundynix/sundynix-dispatcher/internal/harness"
|
||||||
"github.com/sundynix/sundynix-shared/contract"
|
"github.com/sundynix/sundynix-shared/contract"
|
||||||
"github.com/sundynix/sundynix-shared/otelx"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// defaultAgentSystem 是 agent 节点未填系统提示词时的兜底。
|
// defaultAgentSystem 是 agent 节点未填系统提示词时的兜底。
|
||||||
@@ -32,156 +29,12 @@ type board struct {
|
|||||||
sections []reportSection // map 并行 fan-out 产出的分项成稿(供 render 多章渲染)
|
sections []reportSection // map 并行 fan-out 产出的分项成稿(供 render 多章渲染)
|
||||||
answer string // 当前成稿(多 agent 协作时 = 最近一个 agent 的产出 = 成品)
|
answer string // 当前成稿(多 agent 协作时 = 最近一个 agent 的产出 = 成品)
|
||||||
agentOut []string // 各上游 agent 的产出(按序),注入下游 agent 上下文以实现接力协作
|
agentOut []string // 各上游 agent 的产出(按序),注入下游 agent 上下文以实现接力协作
|
||||||
rejected bool // HITL 审批节点拒绝/超时 → 置位,runGraph 中止并返回 errRejected
|
rejected bool // HITL 审批节点拒绝/超时 → 置位,执行中止并返回 errRejected
|
||||||
fatalErr error // agent 节点 LLM 调用失败 → 置位,runGraph 中止并上抛 → 任务判 failed(而非 done-空)
|
fatalErr error // agent 节点 LLM 调用失败 → 置位,执行中止并上抛 → 任务判 failed(而非 done-空)
|
||||||
}
|
}
|
||||||
|
|
||||||
// runGraph 按 DSL 图的真实拓扑与连线执行(替代旧的线性拍平 compileFlow)。
|
// 注:本文件原有自研图解释器 runGraph 已退役(编排引擎统一为 compose,见 compose_compiler.go
|
||||||
//
|
// 的 execComposeGraph)。以下是被 compose 各节点 lambda 复用的节点执行器与黑板辅助函数。
|
||||||
// 入度0 入口 → 沿连线传播 active → 每个节点按 kind 执行真实行为 →
|
|
||||||
// branch 按条件只激活选中的下游(剪枝)→ agent 节点流式回流 token。
|
|
||||||
//
|
|
||||||
// 逐节点点亮"运行·观测"。返回终端 agent 的完整产出(供写回历史)。
|
|
||||||
func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
|
|
||||||
flow, ferr := dsl.Parse(t.Graph)
|
|
||||||
plan := dsl.Compile(t.Graph)
|
|
||||||
b := &board{
|
|
||||||
uid: meta(t, contract.MetaUserID),
|
|
||||||
sid: meta(t, contract.MetaSessionID),
|
|
||||||
query: plan.Query,
|
|
||||||
}
|
|
||||||
|
|
||||||
// 无法解析或空图:退化为"无图单轮对话"(注入默认记忆 + 直接出模型)。
|
|
||||||
if ferr != nil || flow == nil || len(flow.Nodes) == 0 {
|
|
||||||
tr.info("task", "system", "无结构化图", "按单轮对话执行")
|
|
||||||
b.profile = o.fetchMemory(ctx, b.uid, b.query)
|
|
||||||
b.history = o.fetchHistory(ctx, b.sid)
|
|
||||||
o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
|
|
||||||
return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed
|
|
||||||
}
|
|
||||||
|
|
||||||
// 建邻接与入度(只认两端都存在的边)。保留整条边以便 branch 按 true/false 标签选路。
|
|
||||||
nodeByID := make(map[string]dsl.Node, len(flow.Nodes))
|
|
||||||
outE := make(map[string][]dsl.Edge)
|
|
||||||
indeg := make(map[string]int, len(flow.Nodes))
|
|
||||||
for _, n := range flow.Nodes {
|
|
||||||
nodeByID[n.ID] = n
|
|
||||||
indeg[n.ID] = 0
|
|
||||||
}
|
|
||||||
for _, e := range flow.Edges {
|
|
||||||
if _, ok := nodeByID[e.Source]; !ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if _, ok := nodeByID[e.Target]; !ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
outE[e.Source] = append(outE[e.Source], e)
|
|
||||||
indeg[e.Target]++
|
|
||||||
}
|
|
||||||
|
|
||||||
// 入口节点(入度 0)置 active;执行时沿连线把下游激活,branch 只激活选中分支。
|
|
||||||
active := make(map[string]bool)
|
|
||||||
for _, n := range flow.Nodes {
|
|
||||||
if indeg[n.ID] == 0 {
|
|
||||||
active[n.ID] = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 图里没有 memory 节点 → 沿用旧默认:注入画像+历史(避免回归)。
|
|
||||||
hasMemory := false
|
|
||||||
for _, n := range flow.Nodes {
|
|
||||||
if n.Kind == "memory" {
|
|
||||||
hasMemory = true
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if !hasMemory {
|
|
||||||
b.profile = o.fetchMemory(ctx, b.uid, b.query)
|
|
||||||
b.history = o.fetchHistory(ctx, b.sid)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, n := range flow.Topo() {
|
|
||||||
if !active[n.ID] {
|
|
||||||
continue // 被 branch 剪掉的下游,不执行
|
|
||||||
}
|
|
||||||
// 每个节点一个 span:下游的工具调用 / LLM span 会自动挂到本节点之下,
|
|
||||||
// Jaeger 里即呈现「节点 → 工具 → embedding/milvus」的层级。
|
|
||||||
nctx, nspan := otelx.Tracer().Start(ctx, "node."+n.Kind,
|
|
||||||
trace.WithAttributes(
|
|
||||||
attribute.String("sundynix.node.id", n.ID),
|
|
||||||
attribute.String("sundynix.node.kind", n.Kind),
|
|
||||||
))
|
|
||||||
propagate := targetsOf(outE[n.ID]) // 默认激活全部出边;branch 会改写
|
|
||||||
switch n.Kind {
|
|
||||||
case "input":
|
|
||||||
if txt := cstr(n.Config, "text"); txt != "" {
|
|
||||||
b.query = txt
|
|
||||||
}
|
|
||||||
tr.info("input:"+n.ID, "system", labelOf(n, "输入"), truncate(b.query, 80))
|
|
||||||
case "memory":
|
|
||||||
if cbool(n.Config, "profile") {
|
|
||||||
b.profile = o.fetchMemory(nctx, b.uid, b.query)
|
|
||||||
}
|
|
||||||
if cbool(n.Config, "history") {
|
|
||||||
b.history = o.fetchHistory(nctx, b.sid)
|
|
||||||
}
|
|
||||||
tr.info("memory:"+n.ID, "memory", labelOf(n, "记忆"),
|
|
||||||
fmt.Sprintf("画像 %d 字 · 历史 %d 条", len([]rune(b.profile)), len(b.history)))
|
|
||||||
case "retriever":
|
|
||||||
o.retrieverNode(nctx, n, b, tr)
|
|
||||||
case "tool":
|
|
||||||
o.execToolNode(nctx, t.ID, n, b, tr)
|
|
||||||
case "agent":
|
|
||||||
sys := firstNonEmpty(cstr(n.Config, "system"), plan.System)
|
|
||||||
if cbool(n.Config, "autonomous") { // 开启自主工具 → ReAct(模型自己选工具)
|
|
||||||
o.runReactAgent(nctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
|
|
||||||
} else {
|
|
||||||
o.runConversation(nctx, t.ID, b, sys, tr, "agent:"+n.ID)
|
|
||||||
}
|
|
||||||
case "coordinator": // 多智能体协调:orchestrator 自主把子任务派给专家(agent-as-tool)再综合
|
|
||||||
o.runCoordinator(nctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), n, tr, "coordinator:"+n.ID)
|
|
||||||
case "aggregate":
|
|
||||||
merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
|
|
||||||
b.refs, b.toolOut = merged, nil
|
|
||||||
tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略:"+firstNonEmpty(cstr(n.Config, "strategy"), "拼接"))
|
|
||||||
case "render":
|
|
||||||
o.renderNode(nctx, t.ID, n, b, tr)
|
|
||||||
case "branch":
|
|
||||||
propagate = o.branchNode(n, b, outE[n.ID], nodeByID, tr)
|
|
||||||
case "approval":
|
|
||||||
propagate = o.approvalNode(nctx, t.ID, n, b, tr, outE[n.ID])
|
|
||||||
case "map":
|
|
||||||
o.mapNode(nctx, t.ID, n, b, tr)
|
|
||||||
case "output":
|
|
||||||
tr.info("output:"+n.ID, "system", labelOf(n, "输出"), "目标:"+firstNonEmpty(cstr(n.Config, "target"), "屏幕"))
|
|
||||||
default:
|
|
||||||
tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点,跳过")
|
|
||||||
}
|
|
||||||
nspan.End()
|
|
||||||
if b.rejected || b.fatalErr != nil {
|
|
||||||
break // 审批拒绝/超时 或 模型失败:中止后续节点
|
|
||||||
}
|
|
||||||
for _, tgt := range propagate {
|
|
||||||
active[tgt] = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if b.rejected {
|
|
||||||
return b.answer, nil, errRejected // 合法终态,Handle 据此判 rejected 并优雅收尾
|
|
||||||
}
|
|
||||||
if b.fatalErr != nil {
|
|
||||||
return b.answer, nil, b.fatalErr // 上抛 → Handle 判 failed(带原因)
|
|
||||||
}
|
|
||||||
|
|
||||||
// 图里无 agent 节点(纯工具/检索图)也要出一段模型答复,否则没有输出。
|
|
||||||
if b.answer == "" {
|
|
||||||
o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
|
|
||||||
}
|
|
||||||
if b.fatalErr != nil {
|
|
||||||
return b.answer, nil, b.fatalErr
|
|
||||||
}
|
|
||||||
return b.answer, refsOf(b), nil // 成功:带回检索来源供忠实度评测
|
|
||||||
}
|
|
||||||
|
|
||||||
// refsOf 汇总本次执行的检索来源(检索资料 + 工具产出),供忠实度评测。
|
// refsOf 汇总本次执行的检索来源(检索资料 + 工具产出),供忠实度评测。
|
||||||
func refsOf(b *board) []string {
|
func refsOf(b *board) []string {
|
||||||
|
|||||||
@@ -134,7 +134,7 @@ func TestRunGraph_BranchRouting(t *testing.T) {
|
|||||||
ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }}
|
ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }}
|
||||||
run := func(cond string) string {
|
run := func(cond string) string {
|
||||||
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
|
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
|
||||||
ans, _, err := o.runGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1"))
|
ans, _, err := o.runComposeGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -160,7 +160,7 @@ func TestRunGraph_ToolFeedsAgent(t *testing.T) {
|
|||||||
return &contract.ToolResult{OK: true, Content: "TOOLDATA"}
|
return &contract.ToolResult{OK: true, Content: "TOOLDATA"}
|
||||||
}}
|
}}
|
||||||
o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
|
o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
|
||||||
ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
|
ans, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -186,7 +186,7 @@ func TestRunGraph_MapFanout(t *testing.T) {
|
|||||||
return "正文XYZ", nil
|
return "正文XYZ", nil
|
||||||
}}
|
}}
|
||||||
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
|
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
|
||||||
ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
|
ans, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -203,7 +203,7 @@ func TestRunGraph_OutputRedaction(t *testing.T) {
|
|||||||
}}
|
}}
|
||||||
fs := &fakeSink{}
|
fs := &fakeSink{}
|
||||||
o := newOrch(ll, &fakeTools{}, fs, &fakeExec{})
|
o := newOrch(ll, &fakeTools{}, fs, &fakeExec{})
|
||||||
if _, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")); err != nil {
|
if _, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1")); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
out := fs.text()
|
out := fs.text()
|
||||||
|
|||||||
@@ -261,7 +261,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
|
|||||||
slog.InfoContext(ctx, "task received", "task_id", t.ID, "graph_bytes", len(t.Graph))
|
slog.InfoContext(ctx, "task received", "task_id", t.ID, "graph_bytes", len(t.Graph))
|
||||||
tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节,按图执行", len(t.Graph)))
|
tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节,按图执行", len(t.Graph)))
|
||||||
|
|
||||||
// 按 DSL 图执行:compose.Graph(EINO_COMPOSE=1)或自研 graph.go(默认);agent 节点流式回流 token。
|
// 按 DSL 图执行:编译为 Eino compose.Graph 并跑(唯一编排引擎);agent 节点流式回流 token。
|
||||||
answer, refs, err := o.executeGraph(tctx, t, tr)
|
answer, refs, err := o.executeGraph(tctx, t, tr)
|
||||||
if errors.Is(err, errInterrupted) {
|
if errors.Is(err, errInterrupted) {
|
||||||
// HITL 持久化中断:checkpoint 已落、任务停在 waiting(状态在审批节点内已置)。
|
// HITL 持久化中断:checkpoint 已落、任务停在 waiting(状态在审批节点内已置)。
|
||||||
|
|||||||
@@ -41,7 +41,7 @@ func TestRAG_ConversationInjectsAndReturnsRefs(t *testing.T) {
|
|||||||
o := newOrch(ll, kbTool(&captured), &fakeSink{}, &fakeExec{})
|
o := newOrch(ll, kbTool(&captured), &fakeSink{}, &fakeExec{})
|
||||||
tk := &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{contract.MetaUserID: "u42"}}
|
tk := &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{contract.MetaUserID: "u42"}}
|
||||||
|
|
||||||
ans, refs, err := o.runGraph(context.Background(), tk, o.tracer("t1"))
|
ans, refs, err := o.runComposeGraph(context.Background(), tk, o.tracer("t1"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -119,7 +119,7 @@ func TestRAG_DegradesWhenRetrieverDown(t *testing.T) {
|
|||||||
}}
|
}}
|
||||||
o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
|
o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
|
||||||
|
|
||||||
ans, refs, err := o.runGraph(context.Background(), &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{}}, o.tracer("t1"))
|
ans, refs, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{}}, o.tracer("t1"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("检索失败不应让整图失败: %v", err)
|
t.Fatalf("检索失败不应让整图失败: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user