Agent：  ”你看着办”（LLM 动态决策，灵活但不确定）工作流：  ”按这个流程图走”（预定义路径，确定但不灵活）

┌──→ [切菜] ──→ [炒菜] ──┐[洗菜] ─┤                        ├──→ [上桌]        └──→ [煮饭] ────────────┘                    │                    ▼              [如果辣？]              ╱        ╲           是            否           ▼              ▼       [多放糖]      [正常调味]

# 万悟工作流真实架构# 1. 独立仓库：UnicomAI/wanwu-workflow（DAG 引擎、节点执行器、调度器）# 2. 本仓库中的集成点：internal/bff-service/# BFF 网关（暴露工作流 API + 可视化画布）docker-compose*.yaml# 拉起 workflow 容器（8998/8999）# 工作流容器通过 gRPC 与 bff-service 通信# bff-service 负责工作流 CRUD、执行触发、状态查询# workflow 容器负责 DAG 解析、节点调度、执行

// 教学重建版（真实实现：UnicomAI/wanwu-workflow 独立仓库）// Execute 是工作流执行的主入口// 它按照 DAG 拓扑顺序执行所有节点func(e *Executor) Execute(ctx context.Context, wf *Workflow, input map[string]interface{}) (*ExecutionResult, error) {    // Step 1: 构建执行上下文（存储所有节点的输入/输出）    execCtx := NewExecutionContext(input)    // Step 2: 拓扑排序，确定执行顺序    // 将 DAG 转为层级结构：同一层的节点可以并行执行    layers, err := e.scheduler.TopologicalSort(wf.Nodes, wf.Edges)    if err != nil {        return nil, fmt.Errorf(”topological sort failed (cycle detected?): %w”, err)    }    // Step 3: 逐层执行    for layerIdx, layer := range layers {        log.Infof(”executing layer %d, %d nodes”, layerIdx, len(layer))        // 同一层的节点并行执行        var wg sync.WaitGroup        errCh := make(chan error, len(layer))        for _, nodeID := range layer {            node := wf.GetNode(nodeID)            // 检查条件：如果上游条件不满足，跳过此节点            if !e.shouldExecute(node, execCtx) {                execCtx.SetNodeStatus(nodeID, NodeSkipped)                continue            }            wg.Add(1)            go func(n *Node) {                defer wg.Done()                // 执行节点（带重试）                output, err := e.executeNodeWithRetry(ctx, n, execCtx)                if err != nil {                    errCh <- fmt.Errorf(”node %s failed: %w”, n.ID, err)                    execCtx.SetNodeStatus(n.ID, NodeFailed)                    return                }                // 将输出存入上下文，供下游节点引用                execCtx.SetNodeOutput(n.ID, output)                execCtx.SetNodeStatus(n.ID, NodeSuccess)            }(node)        }        wg.Wait()        close(errCh)        // 检查是否有节点失败        for err := range errCh {            if wf.ErrorStrategy == ”fail_fast” {                return nil, err // 快速失败            }            // ”continue” 策略：记录错误但继续执行其他分支            log.Warnf(”node error (continuing): %v”, err)        }    }    // Step 4: 收集最终输出    return execCtx.GetFinalOutput(wf.EndNodeID), nil}

// 教学重建版（真实实现：UnicomAI/wanwu-workflow 独立仓库）// TopologicalSort 将 DAG 节点分层// 返回 [][]string，每个内层 slice 是可以并行执行的节点 IDfunc(s *Scheduler) TopologicalSort(nodes []*Node, edges []*Edge) ([][]string, error) {    // 构建邻接表和入度表    inDegree := make(map[string]int)    adjacency := make(map[string][]string)    for _, node := range nodes {        inDegree[node.ID] = 0    }    for _, edge := range edges {        adjacency[edge.From] = append(adjacency[edge.From], edge.To)        inDegree[edge.To]++    }    // BFS 分层（Kahn 算法）    var layers [][]string    queue := []string{}    // 入度为 0 的节点是第一层    for id, deg := range inDegree {        if deg == 0 {            queue = append(queue, id)        }    }    for len(queue) > 0 {        // 当前层的所有节点        layers = append(layers, queue)        var nextQueue []string        for _, nodeID := range queue {            for _, neighbor := range adjacency[nodeID] {                inDegree[neighbor]--                if inDegree[neighbor] == 0 {                    nextQueue = append(nextQueue, neighbor)                }            }        }        queue = nextQueue    }    // 检测环：如果还有节点入度 > 0，说明有环    for _, deg := range inDegree {        if deg > 0 {            return nil, fmt.Errorf(”cycle detected in workflow DAG”)        }    }    return layers, nil}

// 教学重建版（真实实现：UnicomAI/wanwu-workflow 独立仓库）// Execute 评估条件表达式，决定走哪个分支func(c *ConditionNode) Execute(ctx *ExecutionContext) (*NodeOutput, error) {    // 条件表达式示例：{{node_2.output.total_sales}} < 100000    expression := c.Config.Expression    // 模板变量替换：将 {{node_2.output.total_sales}} 替换为实际值    resolved := ctx.ResolveTemplate(expression)    // 替换后：85000 < 100000    // 安全评估表达式（不用 eval，用白名单解析器）    result, err := c.evaluator.Evaluate(resolved)    if err != nil {        return nil, fmt.Errorf(”condition evaluation failed: %w”, err)    }    // 返回分支选择    if result.Bool() {        return &NodeOutput{Branch: ”true”}, nil   // 走”是”分支    }    return &NodeOutput{Branch: ”false”}, nil      // 走”否”分支}

// 教学重建版（真实实现：UnicomAI/wanwu-workflow 独立仓库）// ResolveTemplate 解析模板变量引用// 输入：”销售额为 {{node_2.output.total}}，目标为 {{workflow.input.target}}”// 输出：”销售额为 85000，目标为 100000”func(ctx *ExecutionContext) ResolveTemplate(template string) string {    // 正则匹配 {{...}} 占位符    re := regexp.MustCompile(`\{\{(.+?)\}\}`)    return re.ReplaceAllStringFunc(template, func(match string) string {        // 去掉 {{ }}        path := strings.Trim(match, ”{} ”)        // 路径解析：node_2.output.total → 节点2的输出中的total字段        value, err := ctx.ResolvePath(path)        if err != nil {            return match // 解析失败，保留原文        }        return fmt.Sprintf(”%v”, value)    })}// ResolvePath 按路径查找变量值func(ctx *ExecutionContext) ResolvePath(path string) (interface{}, error) {    parts := strings.Split(path, ”.”)    switch parts[0] {    case ”workflow”:        // workflow.input.xxx → 工作流全局输入        return getNestedValue(ctx.globalInput, parts[2:])    case ”node”:        // node_2.output.xxx → 某个节点的输出        nodeID := parts[0] + ”.” + parts[1] // ”node_2”        output, ok := ctx.nodeOutputs[nodeID]        if !ok {            return nil, fmt.Errorf(”node %s not executed yet”, nodeID)        }        return getNestedValue(output, parts[3:])    default:        return nil, fmt.Errorf(”unknown variable scope: %s”, parts[0])    }}

// 教学重建版（真实实现：UnicomAI/wanwu-workflow 独立仓库）// executeNodeWithRetry 带重试的节点执行func(e *Executor) executeNodeWithRetry(ctx context.Context, node *Node, execCtx *ExecutionContext) (*NodeOutput, error) {    maxRetries := node.RetryConfig.MaxRetries     // 默认 3    backoff := node.RetryConfig.InitialBackoff    // 默认 1s    var lastErr error    for attempt := 0; attempt <= maxRetries; attempt++ {        if attempt > 0 {            log.Infof(”retrying node %s, attempt %d/%d”, node.ID, attempt, maxRetries)            time.Sleep(backoff)            backoff *= 2 // 指数退避：1s → 2s → 4s        }        output, err := e.executeNode(ctx, node, execCtx)        if err == nil {            return output, nil        }        lastErr = err        // 判断是否可重试（网络超时可重试，参数错误不可重试）        if !isRetryable(err) {            break        }    }    return nil, fmt.Errorf(”node %s failed after %d retries: %w”, node.ID, maxRetries, lastErr)}

[开始] → [查数据库：本周销售数据] → [LLM：生成分析摘要]                                         │                                    [条件：是否达标？]                                    ╱              ╲                                 是                 否                                 ▼                  ▼                          [LLM：写周报]      [LLM：写预警报告]                                 │                  │                                 ▼                  ▼                          [发邮件：周报]     [发邮件：预警]                                 ╲                ╱                                  ╲              ╱                                   ▼            ▼                              [归档到知识库]                                   │                                   ▼                                [结束]

# mini_workflow.py# 零依赖！仅用标准库实现 DAG 工作流引擎import jsonfrom collections import defaultdict, dequefrom concurrent.futures import ThreadPoolExecutor# ===== 1. 定义节点 =====class Node:    def __init__(self, id, func, deps=None):        self.id = id        self.func = func# 可调用对象        self.deps = deps or []# 依赖的节点 ID 列表# ===== 2. DAG 引擎 =====class WorkflowEngine:    def __init__(self):        self.nodes = {}        self.outputs = {}# node_id → output    def add_node(self, node):        self.nodes[node.id] = node    def _topo_layers(self):        ”””Kahn 算法分层”””        in_deg = {nid: 0 for nid in self.nodes}        adj = defaultdict(list)        for nid, node in self.nodes.items():            for dep in node.deps:                adj[dep].append(nid)                in_deg[nid] += 1        layers, queue = [], deque([n for n, d in in_deg.items() if d == 0])        while queue:            layer = list(queue)            layers.append(layer)            next_q = deque()            for nid in layer:                for nb in adj[nid]:                    in_deg[nb] -= 1                    if in_deg[nb] == 0:                        next_q.append(nb)            queue = next_q        if sum(in_deg.values()) > 0:            raise ValueError(”Cycle detected!”)        return layers    def run(self, global_input):        ”””执行工作流”””        self.outputs = {”__input__”: global_input}        layers = self._topo_layers()        for layer_idx, layer in enumerate(layers):            print(f”--- Layer {layer_idx}: {layer} ---”)# 同层并行执行            with ThreadPoolExecutor(max_workers=4) as pool:                futures = {}                for nid in layer:                    node = self.nodes[nid]# 收集依赖节点的输出作为输入                    node_input = {dep: self.outputs[dep] for dep in node.deps}                    node_input[”__input__”] = global_input                    futures[nid] = pool.submit(node.func, node_input)                for nid, future in futures.items():                    self.outputs[nid] = future.result()                    print(f”  ✅ {nid} → {self.outputs[nid]}”)        return self.outputs# ===== 3. 定义工作流 =====engine = WorkflowEngine()# 节点函数def fetch_sales(ctx):    return {”total”: 85000, ”orders”: 342}def analyze(ctx):    data = ctx[”fetch_sales”]    return f”销售额{data['total']}，订单{data['orders']}单，环比下降12%”def check_target(ctx):    data = ctx[”fetch_sales”]    return ”pass” if data[”total”] >= 100000 else ”fail”def write_report(ctx):    analysis = ctx[”analyze”]    return f”📊 周报：{analysis}。整体平稳。”def write_alert(ctx):    analysis = ctx[”analyze”]    return f”⚠️ 预警：{analysis}。建议加大推广力度。”def send_email(ctx):# 根据上游分支选择内容    content = ctx.get(”write_report”) or ctx.get(”write_alert”)    return f”邮件已发送: {content[:30]}...”# 注册节点（deps 定义依赖关系）engine.add_node(Node(”fetch_sales”, fetch_sales))engine.add_node(Node(”analyze”, analyze, deps=[”fetch_sales”]))engine.add_node(Node(”check_target”, check_target, deps=[”fetch_sales”]))engine.add_node(Node(”write_report”, write_report, deps=[”analyze”, ”check_target”]))engine.add_node(Node(”write_alert”, write_alert, deps=[”analyze”, ”check_target”]))engine.add_node(Node(”send_email”, send_email, deps=[”write_report”, ”write_alert”]))# ===== 4. 执行 =====result = engine.run({”target”: 100000, ”week”: ”2026-W30”})print(f”\n🎉 最终输出: {result['send_email']}”)# 预期输出：# --- Layer 0: ['fetch_sales'] ---#   ✅ fetch_sales → {'total': 85000, 'orders': 342}# --- Layer 1: ['analyze', 'check_target'] ---#   ✅ analyze → 销售额85000，订单342单，环比下降12%#   ✅ check_target → fail# --- Layer 2: ['write_report', 'write_alert'] ---#   ✅ write_report → 📊 周报：...#   ✅ write_alert → ⚠️ 预警：...# --- Layer 3: ['send_email'] ---#   ✅ send_email → 邮件已发送: ⚠️ 预警：销售额85000...# 🎉 最终输出: 邮件已发送: ⚠️ 预警：销售额85000...