feat(hitl): 人工审批中断（Eino Phase D）—— 审批节点暂停→批准续跑/拒绝中止
专用「审批」节点方案，全栈打通。 后端： - contract：TaskWaiting/TaskRejected 状态 + ApprovalSubject/ApprovalDecision。 - bus：PublishApproval + WaitApproval（订阅决定主题，带超时）；消费者 AckWait 30s→15min（阻塞等人审期间消息未 ack，否则重投成重复任务）。 - dispatcher：ApprovalWaiter 接口 + approvalNode——执行到审批节点发 await 事件 + 置 waiting，阻塞等决定。批准→回 running 放行下游；拒绝/超时→errRejected 哨兵→ 剪下游→rejected，优雅收尾不计熔断。超时安全默认拒绝。 taskExecTimeout 3→10min（审批5 < 执行10 < AckWait15）。 - 网关：POST /tasks/:id/approve（仅 waiting 态受理，幂等）。 桌面端： - Studio 新增「人工审批」节点（nodeCatalog，可填标题/说明）。 - run.ts pendingApproval() 从 exec 流派生待审中断 + waiting 节点状态。 - BottomDrawer ApprovalBar：琥珀审批条（摘要 + 批准/拒绝 + 备注，调 api.approveTask）。 - ExecTrace waiting 状态灯。 验证：后端 curl 实测 waiting→批准→running→done；waiting→拒绝→rejected（下游未跑）。 前端 tsc + vitest 36 过（pendingApproval 4 例 + waiting 状态）。全模块 build+vet+test 全绿。 EINO_ADOPTION Phase D 标记 HITL 完成。 未覆盖：compose 路径（EINO_COMPOSE，默认关）的 approval 节点；桌面端审批条未在 GUI 实点。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Blizzard
@@ -45,8 +45,9 @@ func main() {
|
||||
}
|
||||
go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig)
|
||||
|
||||
// sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)与任务状态回写(StatusSink)出口。
|
||||
orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub)
|
||||
// sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、
|
||||
// 任务状态回写(StatusSink)与 HITL 审批等待(ApprovalWaiter)出口。
|
||||
orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub)
|
||||
if err != nil {
|
||||
log.Fatalf("[dispatcher] build eino graph: %v", err)
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ type board struct {
|
||||
sections []reportSection // map 并行 fan-out 产出的分项成稿（供 render 多章渲染）
|
||||
answer string // 当前成稿（多 agent 协作时 = 最近一个 agent 的产出 = 成品）
|
||||
agentOut []string // 各上游 agent 的产出（按序），注入下游 agent 上下文以实现接力协作
|
||||
rejected bool // HITL 审批节点拒绝/超时 → 置位，runGraph 中止并返回 errRejected
|
||||
}
|
||||
|
||||
// runGraph 按 DSL 图的真实拓扑与连线执行（替代旧的线性拍平 compileFlow）。
|
||||
@@ -144,6 +145,8 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
|
||||
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":
|
||||
@@ -152,11 +155,18 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
|
||||
tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点，跳过")
|
||||
}
|
||||
nspan.End()
|
||||
if b.rejected {
|
||||
break // 审批拒绝/超时：中止后续节点，整图判 rejected
|
||||
}
|
||||
for _, tgt := range propagate {
|
||||
active[tgt] = true
|
||||
}
|
||||
}
|
||||
|
||||
if b.rejected {
|
||||
return b.answer, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾
|
||||
}
|
||||
|
||||
// 图里无 agent 节点（纯工具/检索图）也要出一段模型答复，否则没有输出。
|
||||
if b.answer == "" {
|
||||
o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
|
||||
@@ -369,6 +379,47 @@ func (o *Orchestrator) branchNode(n dsl.Node, b *board, outs []dsl.Edge, byID ma
|
||||
return chosen
|
||||
}
|
||||
|
||||
// approvalNode 是 HITL 人工审批中断：执行到此暂停，把待审摘要推给 UI（exec 事件 kind=approval/phase=await
|
||||
// + 状态 waiting），阻塞等人工批准/拒绝（带超时，安全默认拒绝）。
|
||||
// 批准 → 状态回 running 并放行下游；拒绝/超时 → 置 b.rejected 中止全图。返回应激活的下游（拒绝=空）。
|
||||
func (o *Orchestrator) approvalNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer, outs []dsl.Edge) []string {
|
||||
title := firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批"))
|
||||
prompt := firstNonEmpty(cstr(n.Config, "prompt"), "请审批是否继续执行后续步骤")
|
||||
summary := prompt
|
||||
if b.answer != "" { // 带上当前产出预览，便于审批人判断
|
||||
summary = prompt + "\n—— 当前产出预览 ——\n" + truncate(b.answer, 400)
|
||||
}
|
||||
|
||||
// 未接审批通道（单测/降级）→ 自动放行，避免无人应答卡死。
|
||||
if o.approval == nil {
|
||||
tr.info("approval:"+n.ID, "approval", title, "未接审批通道，自动放行")
|
||||
return targetsOf(outs)
|
||||
}
|
||||
|
||||
// 暂停：发待审事件（UI 据 kind=approval & phase=await 弹批准/拒绝）+ 置任务 waiting。
|
||||
tr.emit("approval:"+n.ID, "approval", "await", title, summary, 0)
|
||||
o.setStatus(taskID, contract.TaskWaiting, title)
|
||||
|
||||
dec, err := o.approval.WaitApproval(ctx, taskID, approvalTimeout)
|
||||
switch {
|
||||
case err != nil: // 超时 / ctx 取消 → 安全默认拒绝
|
||||
b.rejected = true
|
||||
b.answer = "❌ 审批超时未决，已自动拒绝：" + title
|
||||
tr.emit("approval:"+n.ID, "approval", "error", title, "审批超时，自动拒绝", 0)
|
||||
return nil
|
||||
case !dec.Approved:
|
||||
b.rejected = true
|
||||
note := firstNonEmpty(dec.Note, "审批人拒绝")
|
||||
b.answer = "❌ 已被拒绝：" + note
|
||||
tr.emit("approval:"+n.ID, "approval", "end", title, "拒绝："+note, 0)
|
||||
return nil
|
||||
default: // 批准 → 恢复执行，放行下游
|
||||
o.setStatus(taskID, contract.TaskRunning, "审批通过，继续执行")
|
||||
tr.emit("approval:"+n.ID, "approval", "end", title, "批准："+firstNonEmpty(dec.Note, "放行"), 0)
|
||||
return targetsOf(outs)
|
||||
}
|
||||
}
|
||||
|
||||
// targetsOf 取一组边的目标节点 ID（保持顺序）。
|
||||
func targetsOf(edges []dsl.Edge) []string {
|
||||
out := make([]string, 0, len(edges))
|
||||
|
||||
@@ -40,6 +40,15 @@ type StatusSink interface {
|
||||
PublishTaskStatus(taskID, status, detail string) error
|
||||
}
|
||||
|
||||
// ApprovalWaiter 阻塞等待审批节点的人工决定（由 NATS bus 实现；可为 nil → 审批节点自动放行）。
|
||||
type ApprovalWaiter interface {
|
||||
WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error)
|
||||
}
|
||||
|
||||
// errRejected 是审批节点拒绝（或超时）时图执行返回的哨兵错误：它是合法终态而非故障，
|
||||
// Handle 据此判 rejected 并优雅收尾（不计熔断失败）。
|
||||
var errRejected = errors.New("approval rejected")
|
||||
|
||||
// LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。
|
||||
type LLM interface {
|
||||
Ready() bool
|
||||
@@ -56,17 +65,22 @@ type LLM interface {
|
||||
const toolCallTimeout = 3 * time.Second
|
||||
|
||||
// taskExecTimeout 是单个任务整体执行上限；超时即判 timeout（状态机），避免无限期"运行中"。
|
||||
const taskExecTimeout = 3 * time.Minute
|
||||
// 含 HITL 审批等待预算（approvalTimeout）+ 常规图执行；须 < bus 消费者 AckWait(15min) 以免重投。
|
||||
const taskExecTimeout = 10 * time.Minute
|
||||
|
||||
// approvalTimeout 是单个审批节点等待人工决定的上限；超时安全默认拒绝（fail-safe）。
|
||||
const approvalTimeout = 5 * time.Minute
|
||||
|
||||
// Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行（记忆召回 → 工具节点 → 注入 → 流式）。
|
||||
type Orchestrator struct {
|
||||
pool LLM
|
||||
breaker *harness.CircuitBreaker
|
||||
eval *harness.Evaluator
|
||||
sink TokenSink
|
||||
tools ToolCaller
|
||||
exec ExecSink
|
||||
status StatusSink // 任务生命周期状态回写（可为 nil）
|
||||
pool LLM
|
||||
breaker *harness.CircuitBreaker
|
||||
eval *harness.Evaluator
|
||||
sink TokenSink
|
||||
tools ToolCaller
|
||||
exec ExecSink
|
||||
status StatusSink // 任务生命周期状态回写（可为 nil）
|
||||
approval ApprovalWaiter // HITL 审批等待（可为 nil → 审批节点自动放行）
|
||||
|
||||
turnMu sync.Mutex // 保护 turns（攒批计数，多任务 goroutine 共享）
|
||||
turns map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate
|
||||
@@ -74,9 +88,9 @@ type Orchestrator struct {
|
||||
|
||||
// NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。
|
||||
// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）；
|
||||
// status 为任务生命周期状态回写出口（可为 nil）。
|
||||
func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink) (*Orchestrator, error) {
|
||||
return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status}, nil
|
||||
// status 为任务生命周期状态回写出口（可为 nil）；approval 为 HITL 审批等待（可为 nil）。
|
||||
func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink, approval ApprovalWaiter) (*Orchestrator, error) {
|
||||
return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status, approval: approval}, nil
|
||||
}
|
||||
|
||||
// setStatus 回写一次任务状态流转（status 为 nil 时静默跳过）。
|
||||
@@ -144,6 +158,17 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
|
||||
|
||||
// 按 DSL 图执行：compose.Graph（EINO_COMPOSE=1）或自研 graph.go（默认）；agent 节点流式回流 token。
|
||||
answer, err := o.executeGraph(tctx, t, tr)
|
||||
if errors.Is(err, errRejected) {
|
||||
// HITL 拒绝：合法终态，非故障。收尾流 + 置 rejected，不计熔断、不重投。
|
||||
slog.InfoContext(ctx, "task rejected by approval", "task_id", t.ID)
|
||||
if answer != "" {
|
||||
_ = o.sink.PublishToken(t.ID, []byte(answer))
|
||||
}
|
||||
_ = o.sink.CompleteStream(t.ID)
|
||||
o.breaker.Report(true) // 拒绝是人为决策，不算后端失败
|
||||
o.setStatus(t.ID, contract.TaskRejected, truncate(answer, 120))
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
span.RecordError(err)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
|
||||
@@ -81,6 +81,11 @@ func (s *Subscriber) PublishTaskStatus(taskID, status, detail string) error {
|
||||
})
|
||||
}
|
||||
|
||||
// WaitApproval 让 Subscriber 满足 eino.ApprovalWaiter，阻塞等待审批节点的人工决定。
|
||||
func (s *Subscriber) WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) {
|
||||
return s.inner.WaitApproval(ctx, taskID, timeout)
|
||||
}
|
||||
|
||||
// RequestModelConfig 向控制面（Gateway）取当前激活的对话模型配置。
|
||||
func (s *Subscriber) RequestModelConfig(ctx context.Context) (*contract.ModelConfig, error) {
|
||||
return s.inner.RequestConfig(ctx, contract.ConfigKindChat)
|
||||
|
||||
Reference in New Issue
Block a user