Files
sundynix-agentix/sundynix-dispatcher/internal/eino/graph.go
T

511 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package eino
import (
"context"
"fmt"
"strconv"
"strings"
"time"
"github.com/cloudwego/eino/schema"
"github.com/sundynix/sundynix-dispatcher/internal/dsl"
"github.com/sundynix/sundynix-dispatcher/internal/harness"
"github.com/sundynix/sundynix-shared/contract"
)
// defaultAgentSystem 是 agent 节点未填系统提示词时的兜底。
const defaultAgentSystem = "你是 sundynix-agentix 平台的 AI 助手。"
// board 是图执行的"黑板":节点按拓扑序流转时读写它。
type board struct {
uid, sid string
tenant string // 发起任务的租户(平台工具计费口径要跟 HTTP 提交一致,经 inject 传给 platform_*
query string
useVoice bool // 该任务用 JARVIS 语音模型池(网关 Meta[model_profile]==voice
profile string
history []*schema.Message
kb string // 最近一个检索节点的 owner 作用域库名(供 map 并行各项检索)
refs []string // 检索 / 聚合得到的参考资料
toolOut []string // 工具节点产出
sections []reportSection // map 并行 fan-out 产出的分项成稿(供 render 多章渲染)
answer string // 当前成稿(多 agent 协作时 = 最近一个 agent 的产出 = 成品)
agentOut []string // 各上游 agent 的产出(按序),注入下游 agent 上下文以实现接力协作
rejected bool // HITL 审批节点拒绝/超时 → 置位,执行中止并返回 errRejected
fatalErr error // agent 节点 LLM 调用失败 → 置位,执行中止并上抛 → 任务判 failed(而非 done-空)
}
// 注:本文件原有自研图解释器 runGraph 已退役(编排引擎统一为 compose,见 compose_compiler.go
// 的 execComposeGraph)。以下是被 compose 各节点 lambda 复用的节点执行器与黑板辅助函数。
// refsOf 汇总本次执行的检索来源(检索资料 + 工具产出),供忠实度评测。
func refsOf(b *board) []string {
out := append([]string{}, b.refs...)
return append(out, b.toolOut...)
}
// retrieverNode 执行检索节点:kb 按 owner 作用域 → kb_search → 累计参考资料。
func (o *Orchestrator) retrieverNode(ctx context.Context, n dsl.Node, b *board, tr *execTracer) {
kb := cstr(n.Config, "kb")
scoped := kb
if b.uid != "" && kb != "" && !strings.Contains(kb, "/") {
scoped = b.uid + "/" + kb
}
if scoped != "" {
b.kb = scoped // 记下作用域库名,供后续 map 并行各项检索复用
}
end := tr.span("retriever:"+n.ID, "tool", labelOf(n, "检索"))
refs := o.retrieve(ctx, scoped, b.query)
if refs != "" {
b.refs = append(b.refs, refs)
}
end(fmt.Sprintf("kb=%s · 命中 %d 段", firstNonEmpty(kb, "(未指定)"), countLines(refs)), nil)
}
// mapNode 执行并行 fan-out:把主题拆成若干项,各项有界并发撰写(复用 report 的 writeSections),
// 成稿存入黑板(结构化 sections + 拼进 answer),并流式呈现进度。
func (o *Orchestrator) mapNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer) {
end := tr.span("map:"+n.ID, "plan", labelOf(n, "并行 fan-out"))
items := o.planItems(ctx, b.query, cstr(n.Config, "splitBy"))
end(fmt.Sprintf("拆出 %d 项:%s", len(items), strings.Join(items, " / ")), nil)
o.emit(taskID, fmt.Sprintf("\n> 并行处理 %d 项…\n\n", len(items)))
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"
o.emit(taskID, chunk)
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 工具,产出累计进黑板;失败降级不阻断。
func (o *Orchestrator) execToolNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer) {
tool, args := dsl.ToolBinding(n)
if tool == "" {
return
}
node := "tool:" + tool
if o.tools == nil {
tr.info(node, "tool", "工具 "+tool, "工具总线未接入,跳过")
return
}
call := map[string]any{}
for k, v := range args {
call[k] = v
}
if call["q"] == nil && call["query"] == nil {
call["q"] = b.query
}
if b.uid != "" {
if kbv, ok := call["kb"].(string); ok && kbv != "" && !strings.Contains(kbv, "/") {
call["kb"] = b.uid + "/" + kbv
}
}
end := tr.span(node, "tool", "调用工具 "+tool)
cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
defer cancel()
res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo(tool), &contract.ToolCall{Tool: tool, TaskID: taskID, Args: call})
if err != nil {
end("调用失败,降级跳过", err)
return
}
if res == nil || !res.OK || res.Content == "" {
end("无结果,降级跳过", nil)
return
}
end("入参 "+previewArgs(call)+" → 产出 "+truncate(res.Content, 160), nil)
b.toolOut = append(b.toolOut, "["+tool+"] "+res.Content)
}
// runAgent 执行 agent/模型节点:据黑板拼消息 → 流式回流 token → 累计成稿。
// label 是这个节点在轨迹里的显示名,一律由调用方按 labelOf(n, 兜底) 传入:
// 图里可以并排好几个 agent,写死成同一个名字的话轨迹/团队视图里根本分不出谁是谁。
func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node, label string) {
label = firstNonEmpty(label, "模型流式推理")
rc := &RunCtx{
System: firstNonEmpty(system, defaultAgentSystem),
Query: b.query,
Profile: b.profile,
History: b.history,
ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
}
msgs, _ := buildMessages(ctx, rc)
// 成本护栏:计入本节点输入 token;若已触顶则中止整图(防失控成本)。
if bud := harness.BudgetFrom(ctx); bud != nil {
for _, m := range msgs {
bud.AddPrompt(m.Content)
}
if bud.Exceeded() {
if b.fatalErr == nil {
b.fatalErr = errBudget
}
tr.emit(node, "system", "error", "token 预算", "已达单任务预算上限,中止", 0)
return
}
}
tr.emit(node, "model", "start", label, "", 0)
t0 := time.Now()
n := 0
var produced strings.Builder // 本节点自身产出(用于沿图向下游传递)
// 输出护栏:有状态流式脱敏,跨分片缓冲,杜绝密钥被切断而漏检(逐片正则会漏跨片密钥)。
red := harness.NewStreamRedactor()
emit := func(safe string) {
if safe == "" {
return
}
_ = o.sink.PublishToken(taskID, []byte(safe))
produced.WriteString(safe)
n++
}
send := func(s string) { emit(red.Push(s)) }
// reasoning 模型(本地 Qwen3 思考 / DeepSeek-R1 / QwQ 等)的「思考过程」:不进答案、
// 累计后 surface 到观测轨迹(让用户看到模型在想什么,又不污染最终回答)。
var reasoning strings.Builder
onReasoning := func(s string) { reasoning.WriteString(s) }
var err error
pool := o.agentPool(b) // 语音任务走语音模型池(否则工作池)
if pool.Ready() {
err = pool.ChatStream(ctx, toChatMessages(msgs), send, onReasoning)
} else {
err = pool.StreamText(ctx, replyFor(msgs), func(tok []byte) { send(string(tok)) })
}
if rc := reasoning.String(); rc != "" {
tr.info(node, "model", "推理过程", fmt.Sprintf("思考 %d 字:%s", len([]rune(rc)), truncate(rc, 200)))
}
if err != nil {
tr.emit(node, "model", "error", label, err.Error(), time.Since(t0).Milliseconds())
// 未产出任何 token 即失败 → 标记致命错,让任务判 failed(暴露原因,便于监控告警),
// 而非静默 done-空。已流出部分 token 的中断也算失败(结果不完整)。
if b.fatalErr == nil {
b.fatalErr = fmt.Errorf("agent 模型推理失败: %w", err)
}
return
}
emit(red.Flush()) // 吐出暂留的尾部(最后一段疑似密钥的判定)
if bud := harness.BudgetFrom(ctx); bud != nil {
bud.AddComplete(produced.String()) // 成本护栏:计入本节点输出 token
}
if red.Hits() > 0 {
tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/PII", red.Hits()))
}
o.recordAgentOutput(b, produced.String()) // 产出入黑板:成当前成稿 + 供下游接力
tr.emit(node, "model", "end", label,
fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
}
// recordAgentOutput 把一个 agent 节点的产出记入黑板:append 到 agentOut(供下游 agent 注入接力),
// 并设为当前成稿 answer(多 agent 协作时,最后一个 agent 的产出即最终成品)。空产出忽略。
func (o *Orchestrator) recordAgentOutput(b *board, out string) {
if strings.TrimSpace(out) == "" {
return
}
b.agentOut = append(b.agentOut, out)
b.answer = out
}
// renderNode 执行渲染节点:把当前成稿渲染成 Word(经 mcp-go report_render)。
func (o *Orchestrator) renderNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer) {
if strings.TrimSpace(b.answer) == "" {
tr.info("render:"+n.ID, "render", labelOf(n, "渲染"), "暂无正文可渲染(render 前需有 agent 产出)")
return
}
format := firstNonEmpty(cstr(n.Config, "format"), "docx")
end := tr.span("render:"+n.ID, "render", labelOf(n, "渲染 "+format))
title := truncate(b.query, 40)
secs := b.sections // map 产出的多章优先;否则把整段成稿当单章
if len(secs) == 0 {
secs = []reportSection{{Heading: title, Body: b.answer}}
}
if path := o.renderReport(ctx, taskID, title, secs); path != "" {
end("已落盘:"+path, nil)
o.emit(taskID, "\n\n---\n✅ 已渲染 "+format+" 文档,可在「下载」获取。\n")
} else {
end("渲染服务不可用", fmt.Errorf("render unavailable"))
}
}
// branchNode 执行分支节点:求值条件,按出边的 true/false 标签选出要激活的下游。
// 边带 sourceHandle("true"/"false") 时按标签精确选路(前端 Phase 2 给 branch 的边打标);
// 无标签的旧图退回"出边顺序 [true,false]"约定,保持向后兼容。
func (o *Orchestrator) branchNode(n dsl.Node, b *board, outs []dsl.Edge, byID map[string]dsl.Node, tr *execTracer) []string {
cond := cstr(n.Config, "condition")
res := evalCondition(cond, b)
var truthy, falsy, deflt []string
labeled := false
for _, e := range outs {
switch e.SourceHandle {
case "true":
truthy = append(truthy, e.Target)
labeled = true
case "false":
falsy = append(falsy, e.Target)
labeled = true
case "default", "else": // 显式兜底边:主选择落空时走它
deflt = append(deflt, e.Target)
}
}
var chosen []string
mode := "边标签"
if labeled {
if res {
chosen = truthy
} else {
chosen = falsy
}
} else {
mode = "边序约定[true,false]"
targets := targetsOf(outs)
switch {
case len(targets) >= 2:
if res {
chosen = targets[:1]
} else {
chosen = targets[1:2]
}
case len(targets) == 1 && !res:
chosen = nil
default:
chosen = targets
}
}
// else 兜底:主选择未命中任何下游时,走显式 default/else 边(若配了),避免悄悄落 END。
usedDefault := false
if len(chosen) == 0 && len(deflt) > 0 {
chosen = deflt
usedDefault = true
}
names := make([]string, 0, len(chosen))
for _, id := range chosen {
names = append(names, labelOf(byID[id], id))
}
// 明确记录选路:走 default / 未匹配收口结束 / 正常选路,杜绝"静默落 END"被误当 bug。
dest := "[" + strings.Join(names, ", ") + "]"
switch {
case usedDefault:
dest = "default → " + dest
case len(chosen) == 0:
dest = "未匹配任何下游 → 收口结束(END)"
}
tr.info("branch:"+n.ID, "system", labelOf(n, "分支"),
fmt.Sprintf("条件「%s」→ %v ⇒ 走 %s%s",
firstNonEmpty(cond, "(空=真)"), res, dest, mode))
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, summary := approvalSummary(n, b)
// 未接审批通道(单测/降级)→ 自动放行,避免无人应答卡死。
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)
if err != nil { // 超时 / ctx 取消 → 安全默认拒绝
b.rejected = true
b.answer = "❌ 审批超时未决,已自动拒绝:" + title
tr.emit("approval:"+n.ID, "approval", "error", title, "审批超时,自动拒绝", 0)
return nil
}
// 批准/拒绝的落盘与终态事件与中断式审批共用,杜绝两路文案/语义漂移。
o.applyApprovalDecision(taskID, n, b, dec, dec.Approved, tr)
if b.rejected {
return nil
}
return targetsOf(outs)
}
// approvalSummary 据审批节点配置 + 当前成稿,算出待审标题与摘要(带产出预览,便于审批人判断)。
func approvalSummary(n dsl.Node, b *board) (title, summary 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)
}
return title, summary
}
// applyApprovalDecision 把一次人工决定落到黑板并发终态事件:拒绝 → 置 rejected + 拒绝语;
// 批准 → 状态回 running 放行。阻塞式(approvalNode)与中断式(approvalInterruptLambda)审批共用。
func (o *Orchestrator) applyApprovalDecision(taskID string, n dsl.Node, b *board, dec *contract.ApprovalDecision, approved bool, tr *execTracer) {
title := firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批"))
if !approved {
note := "审批人拒绝"
if dec != nil {
note = firstNonEmpty(dec.Note, note)
}
b.rejected = true
b.answer = "❌ 已被拒绝:" + note
tr.emit("approval:"+n.ID, "approval", "end", title, "拒绝:"+note, 0)
return
}
note := "放行"
if dec != nil {
note = firstNonEmpty(dec.Note, note)
}
o.setStatus(taskID, contract.TaskRunning, "审批通过,继续执行")
tr.emit("approval:"+n.ID, "approval", "end", title, "批准:"+note, 0)
}
// targetsOf 取一组边的目标节点 ID(保持顺序)。
func targetsOf(edges []dsl.Edge) []string {
out := make([]string, 0, len(edges))
for _, e := range edges {
out = append(out, e.Target)
}
return out
}
// evalCondition 求值 branch 条件。支持:
//
// 空 → true;关键字 refs/tools/answer/profile 作左值(取数量/字数);
// 形如 "a op b"op: >= <= == != > <)数值比较;其余非空 → 默认真。
func evalCondition(cond string, b *board) bool {
cond = strings.TrimSpace(cond)
if cond == "" {
return true
}
for _, op := range []string{">=", "<=", "==", "!=", ">", "<"} {
if i := strings.Index(cond, op); i >= 0 {
l := resolveOperand(strings.TrimSpace(cond[:i]), b)
r := resolveOperand(strings.TrimSpace(cond[i+len(op):]), b)
switch op {
case ">":
return l > r
case "<":
return l < r
case ">=":
return l >= r
case "<=":
return l <= r
case "==":
return l == r
case "!=":
return l != r
}
}
}
return true
}
// resolveOperand 把条件里的左/右值解析为数值(关键字取运行时数量,否则按字面量)。
func resolveOperand(s string, b *board) float64 {
switch strings.ToLower(s) {
case "refs":
return float64(len(b.refs))
case "tools":
return float64(len(b.toolOut))
case "answer":
return float64(len([]rune(b.answer)))
case "profile":
return float64(len([]rune(b.profile)))
}
f, _ := strconv.ParseFloat(s, 64)
return f
}
// aggregate 按策略合并多段参考资料为一段。
func aggregate(strategy string, parts []string) []string {
var nonEmpty []string
for _, p := range parts {
if strings.TrimSpace(p) != "" {
nonEmpty = append(nonEmpty, p)
}
}
if len(nonEmpty) == 0 {
return nil
}
switch strategy {
case "去重合并":
seen := map[string]bool{}
var uniq []string
for _, p := range nonEmpty {
if !seen[p] {
seen[p] = true
uniq = append(uniq, p)
}
}
return []string{strings.Join(uniq, "\n---\n")}
case "摘要":
return []string{truncate(strings.Join(nonEmpty, "\n"), 800)}
default: // 拼接
return []string{strings.Join(nonEmpty, "\n---\n")}
}
}
// ---- 小工具 ----
func meta(t *contract.Task, key string) string {
v, _ := t.Meta[key].(string)
return v
}
func cstr(cfg map[string]any, key string) string {
if cfg == nil {
return ""
}
v, ok := cfg[key]
if !ok || v == nil {
return ""
}
if s, ok := v.(string); ok {
return strings.TrimSpace(s)
}
return strings.TrimSpace(fmt.Sprint(v))
}
func cbool(cfg map[string]any, key string) bool {
if cfg == nil {
return false
}
if v, ok := cfg[key].(bool); ok {
return v
}
return false
}
func labelOf(n dsl.Node, def string) string {
if strings.TrimSpace(n.Label) != "" {
return n.Label
}
return def
}
func countLines(s string) int {
s = strings.TrimSpace(s)
if s == "" {
return 0
}
return strings.Count(s, "\n") + 1
}