QQs'Note

这篇解决什么问题

上篇：Flow Graph 入门：用 TypeScript 实现工业视觉流程执行器

上篇已经得到一条可运行的最小链路：

1

schema -> validate -> compile -> execute -> trace

但工业视觉和机器人应用软件还会遇到更难的问题：画布如何保存完整语义、一帧一帧的数据如何流动、下游变慢时如何限制上游，以及一个名字叫 execute 的节点为什么不能自动获得真实硬件能力。

这篇不再复制一套完整代码，而是建立下一阶段的工程边界。重点是：

• React Flow 只是编辑器，不是运行时。
• 流式不是“拓扑排序执行得快一点”。
• 能自由拖线，不代表能自由获得设备副作用。
• 工业安全必须由运行端能力和 fail-closed 规则保证。

1. 先回顾上篇的边界

上篇的执行器是确定性的批式 DAG：一张图运行一次，每个节点运行一次，下游等上游产出后继续。

它已经能证明：

• 边精确连接端口。
• 端口类型在加载期检查。
• 必填输入不能漏接。
• 顶层图不能有环。
• 图先编译，再按稳定顺序运行。

它还不能表达：

• 一个 source 连续产生多帧。
• branch 未选中的路径如何标记。
• 多个上游何时算“到齐”。
• 队列满时上游该等待、丢弃还是降采样。
• 什么条件满足后才允许真实设备动作。

接下来不是给现有 for 循环不断加 if，而是逐项定义这些语义。

2. React Flow 只负责编辑，不定义运行语义

React Flow 的 handle 很适合表示端口。关键是保存时不能把端口信息压扁成 A -> B。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import type { Edge, Node } from "@xyflow/react";

interface FlowEdge {
  id: string;
  from: { node: string; port: string };
  to: { node: string; port: string };
}

function toFlowEdge(edge: Edge): FlowEdge {
  if (!edge.sourceHandle || !edge.targetHandle) {
    throw new Error(`edge ${edge.id} must connect explicit ports`);
  }
  return {
    id: edge.id,
    from: { node: edge.source, port: edge.sourceHandle },
    to: { node: edge.target, port: edge.targetHandle },
  };
}

对应关系必须直接、可逆：

1
2

sourceHandle -> 输出端口 ID
targetHandle -> 输入端口 ID

节点坐标不影响执行，应单独保存：

1
2
3
4
5
6
7
8
9

interface EditorMetadata {
  positions: Record<string, { x: number; y: number }>;
  viewport?: { x: number; y: number; zoom: number };
}

interface EditorDocument {
  graph: FlowDocument;
  metadata: EditorMetadata;
}

拖线时的类型校验只是体验优化。后端或本地 runtime 仍要重新校验，因为 JSON 可能来自文件、CLI、测试 fixture 或旧客户端。

3. 编辑模型、编译模型和运行模型

把三种模型塞进一个巨大对象，往往会让 UI 状态、持久化格式和临时运行状态互相污染。

1
2
3

编辑模型：节点配置、端口边、坐标、分组和注释
编译模型：拓扑顺序、直接边绑定和资源引用
运行模型：消息 handle、节点状态、取消信号和 trace

1
2
3
4
5
6
7

flowchart LR
    E[Editor Document] --> V[Validate]
    V --> C[Compile]
    C --> P[Static Plan]
    P --> R[Runtime State]
    R --> T[Trace]
    T --> E

Flow 编辑不等于运行时必须创建重型消息系统

端口图是语义层。单进程执行时，可以在加载期把边解析为直接绑定：

1

detect.detection -> adjust.detection

运行时只需要拿到已绑定的上游输出并调用 runner，不必每个节点都查全局表，也不必为低频 DAG 引入分布式 broker。

图像和点云也不应沿边复制本体。端口消息传 FrameHandle 或引用，实际数据由资源存储管理：

1
2
3
4

interface FrameHandle {
  frameId: string;
  kind: "image_2d" | "point_cloud";
}

选择这种分离的代价是：schema 变更时要同时考虑编辑文档迁移和编译器兼容，不能靠一个对象“到处都能用”来逃避版本治理。

4. Fan-out、fan-in、barrier 和 branch

这几个术语经常同时出现，最好逐个定义。

Fan-out：一个输出发给多个下游

1
2
3

flowchart LR
    C[Capture] --> D1[Detect Defect]
    C --> D2[Estimate Pose]

若 payload 很大，两个下游应共享只读 handle，而不是各复制一份图像。

Fan-in：多个上游汇入同一节点

1
2
3

flowchart LR
    D[Detection] --> A[Decision]
    P[Pose Estimate] --> A

fan-in 只描述结构，不自动说明节点何时执行。节点需要一个明确的等待规则。

Barrier：等一组条件全部满足

例如 decision 必须同时收到 DetectionResult 和 PoseEstimate。在批式 DAG 中，required inputs 到齐即可运行；在流式系统中，还必须知道哪些消息属于同一个工件或同一帧。

关联键至少包含：

1

(runId, itemId)

不能只按端口收集，否则可能把工件 A 的检测结果和工件 B 的位姿拼成一组。

Branch：只激活一条路径

branch 的两个输出可以是 pass 和 review。未选中的路径应该标记为 skipped，不是 failed。

这三种状态不能混为一谈：

如果不显式区分，fan-in 节点很难判断应该继续等待、跳过还是报错。

5. 从批式 DAG 到流式 Flow

批式 DAG 的执行单位是“节点的一次运行”；流式 Flow 的执行单位是“消息的一次到达”。

一种最小流式 runner 接口是：

1
2
3
4
5
6
7
8
9
10
11
12

interface OutputEvent {
  port: string;
  itemId: string;
  value: unknown;
}

interface StreamingNodeRunner {
  run(
    inputs: AsyncIterable<OutputEvent>,
    context: RunContext,
  ): AsyncIterable<OutputEvent>;
}

Backpressure 不是性能优化，而是内存上界

假设相机每秒输出 30 帧，检测节点每秒只能处理 10 帧。若输入队列无限增长，系统只是把“处理不过来”延迟成“稍后内存耗尽”。

使用有界 channel 时，队列满后必须选择策略：

• await：让 producer 等待，适合不能丢数据的离线处理。
• drop-oldest：只保留新数据，适合实时预览。
• sample：按频率采样，适合监控界面。
• reject：明确失败，适合必须保证完整批次的流程。

策略属于业务契约，不能由底层队列随意决定。

实用升级顺序

1. 先让一个 source 能连续 emit。
2. 让一条线性链处理多消息。
3. 增加 end 信号和取消。
4. 增加有界队列和背压策略。
5. 再加 fan-out。
6. 最后实现按 (runId, itemId) 收集的 barrier。

不要一开始同时引入分支、循环、并行、流式和分布式执行。

6. 取消、超时、重试和 trace

取消必须传到真实 I/O

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

async function withTimeout<T>(
  operation: (signal: AbortSignal) => Promise<T>,
  timeoutMs: number,
  parentSignal: AbortSignal,
): Promise<T> {
  const controller = new AbortController();
  const timer = setTimeout(
    () => controller.abort(new Error(`timeout after ${timeoutMs}ms`)),
    timeoutMs,
  );
  const cancelFromParent = () => controller.abort(parentSignal.reason);
  parentSignal.addEventListener("abort", cancelFromParent, { once: true });

  try {
    return await operation(controller.signal);
  } finally {
    clearTimeout(timer);
    parentSignal.removeEventListener("abort", cancelFromParent);
  }
}

只在外层 Promise.race() 一个 timeout，会让调用方停止等待，却不一定停止底层 I/O。对设备动作而言，这是假取消：UI 显示已停止，设备请求仍可能继续。

重试要同时满足三个条件

1. 错误明确标记为 transient（暂时性故障）。
2. 操作幂等，或带稳定 idempotency key。
3. 重复执行不会制造额外物理副作用。

网络读取可能重试；机器人移动、PLC 写入和工艺派发不能默认重试。

Trace 是统一的可观察性基础

至少记录：

1
2

graphVersion, runId, nodeId, status, durationMs,
retryCount, runMode, input/output summary, errorCode

图像和点云只记录 handle、尺寸和摘要，不把整个 payload 写进日志。Trace 同时服务于节点高亮、性能分析、失败定位、运行回放和安全审计。

7. L1 原子节点与 L2 复合节点

L1（Level 1）原子节点只做一件事，例如：

• capture：产出 frame handle。
• detect：frame -> detection。
• collect：按关联键收集消息。
• log：输出摘要。

L2（Level 2）复合节点把常用链封装成一个外部契约，例如：

1

InspectPart = capture -> detect -> aggregate

复合节点需要定义：

• 对外 typed ports。
• 外部端口到内部悬空端口的映射。
• config 如何传入内部节点。
• trace 默认折叠还是展开。
• 子图版本和依赖。

选择 L1/L2 分层的收益是：开发者可展开调试，操作员可使用更少、更贴近工艺概念的节点。代价是复合节点需要单独的版本、迁移和 trace 展示规则。

8. 六个值得保留的架构决议

下面把设计演进整理成六张决议卡。它们是通用工程结论，不依赖某个具体项目。

8.1 从隐式黑板到 typed edges

背景问题：节点通过字符串后缀、全局变量名或扫描共享对象寻找上游数据。图能运行，但数据依赖不在图上，错误通常到运行期才出现。

考虑方案：继续靠文档约定；统一显式变量名；让业务产物通过 typed edges 传递。

最终选择：业务数据沿 typed edges 传递；context 只保留 run ID、日志、取消信号等运行能力。

理由：依赖可见，端口可在加载期校验，多个相同类型的 source 也不会因命名猜测而冲突。

代价：需要维护端口 schema 和图迁移，旧的隐式数据引用不能自动继续工作。

8.2 编辑是 Flow，执行是编译后的静态计划

背景问题：直接让运行时解释画布对象，会不断扫描边、读取 UI 字段，并把展示状态带入执行逻辑。

考虑方案：运行时直接解释编辑 JSON；每次执行动态查找；加载时编译静态计划。

最终选择：保存完整端口图，加载时校验、拓扑排序并绑定边，运行时消费静态计划。

理由：编辑语义完整，同时减少运行时动态查找；确定性也更容易测试。

代价：编译模型需要缓存失效和版本管理，编辑后必须重新编译。

8.3 L1 原子节点与 L2 复合节点分层

背景问题：全用原子节点会让操作画布被大量胶水步骤淹没；全用大黑盒又难以组合和调试。

考虑方案：只保留原子节点；只提供固定流程；原子节点与用户可复用复合节点并存。

最终选择：L1 提供清晰、可测试的原子能力，L2 封装高频工艺和受控流程。

理由：兼顾组合能力和操作员认知负担，调试时还能展开内部链。

代价：复合节点的端口映射、版本和 trace 展开需要额外设计。

8.4 教学节点不等于真实硬件能力

背景问题：把 register、adjust、execute 做成可自由组合且都能真实运行的节点，可能允许用户绕过必要的安全判断。

考虑方案：相信画布连线正确；在每个节点重复安全逻辑；把真实能力收敛到受控复合节点和 runner。

最终选择：细粒度节点可用于 simulate、教学和 dry-run；真实纠偏通过受控复合节点调用单一 runner。

理由：安全边界不依赖 UI，不会因为多一种连法就出现第二条设备写入路径。

代价：模拟图和真实能力不完全一一对应，UI 必须明确标识哪些节点仅用于模拟。

8.5 工位 barrier 与数值 gate 正交

背景问题：数据齐全不代表设备已到位；设备到位也不代表调整量安全。

考虑方案：只设一个综合布尔值；只检查工位；分别建模两道门。

最终选择：工位 barrier 检查数据和设备状态，数值 gate 判断调整量，两者都通过才允许真实动作。

理由：两道门处理不同故障，失败原因可观察，也便于分别测试。

代价：运行状态更多，需要明确超时、错误码和操作员处理流程。

8.6 批量流程把计算与派发分开

背景问题：计算节点内部直接下发设备，会让中间产物无法预览、重放和统一审核，多项计算也难以在最终派发前叠加。

考虑方案：边算边发；每个调整节点自行下发；先计算全部调整，再通过 barrier 统一派发。

最终选择：纯计算节点产出新数据，受控派发节点执行副作用。

理由：中间结果可 trace、可 dry-run、可比较；批量数据齐备后再进入统一安全路径。

代价：需要保存中间产物，并处理它们的版本、生命周期和一致性。

9. Simulate、real 与能力注入

run mode 不是 UI 上的装饰开关，而是 runtime 的强制策略。

节点名称不是权限

图中出现 execute，不代表它天然能控制设备。真实副作用还必须同时满足：

1
2
3
4

runMode == real
AND runner 已注入
AND barrier ready
AND numeric gate accept

runner 是 capability（能力）对象，只由应用装配层注入：

1
2
3
4
5
6
7

interface MotionRunner {
  dispatch(command: ApprovedCommand, signal: AbortSignal): Promise<void>;
}

interface RuntimeServices {
  motionRunner?: MotionRunner;
}

在 simulate 模式，即使 runner 存在也不能调用。real 模式缺 runner 必须拒绝，不能悄悄退化成“假成功”。

这种设计的代价是依赖注入和测试装配更复杂，但安全能力不会因节点配置或前端请求而凭空出现。

10. 两道正交的安全门

考虑“质检工位计算调整量，执行工位应用结果”的流程：

1
2
3
4
5
6

flowchart LR
    I[Inspection Data] --> B[Station Barrier]
    S[Station Ready] --> B
    B --> G[Numeric Gate]
    A[Adjustment] --> G
    G --> R[Controlled Runner]

工位 barrier 检查

• 所需检测、位姿和调整数据是否属于同一 runId/itemId。
• 设备是否到达指定工位。
• 必需的握手信号是否在超时内满足。

数值 gate 检查

• 调整量是否在允许范围。
• 输入标定和坐标变换是否有效。
• 决策是 Apply、Skip、Clamp 还是 Reject。

真实 runner 不接受未经 gate 的原始调整量。更稳妥的做法是让类型也表达这一点：

1
2
3
4
5

interface GatedCommand {
  decision: "apply" | "clamp";
  effectiveOffsetMm: number;
  auditId: string;
}

fail-closed 的含义是：runner 缺失、barrier 未满足、gate 拒绝或状态无法确认时，结果都是“不执行真实动作”，同时留下可诊断错误，而不是猜测一个默认值继续。

11. 批量流程为什么要把计算与派发分开

工业视觉不总是拍一帧就立即动作。常见批量流程是：

1
2
3
4
5
6
7

采集多视点
  -> 分别检测/配准
  -> 按 itemId 映射到目标工艺对象
  -> 汇总全部调整
  -> 工位 barrier
  -> 统一安全检查
  -> 批量派发

建议的数据形态：

1
2
3
4
5
6
7
8
9
10
11

interface PlannedAdjustment {
  itemId: string;
  targetId: string;
  offsetMm: number;
  sourceFrameId: string;
}

interface AdjustmentBatch {
  runId: string;
  items: PlannedAdjustment[];
}

计算节点只产生 AdjustmentBatch，不持有 runner。派发节点只接受经过映射、barrier 和 gate 的批次。

这样做可在派发前完成：

• 预览每个目标对应哪一帧、哪项调整。
• 检查是否漏项或重复映射。
• dry-run 比较调整前后的结果。
• 记录统一审计 trace。

代价是必须定义批次完成条件。若某帧永远不到，需要明确 timeout 后是整批失败、跳过该项还是进入人工确认，不能无限等待。

12. React Flow 编辑器的最小实现边界

第一版编辑器只需要做到六件事：

1. 从 registry 展示节点面板。
2. 根据 typed ports 渲染 handles。
3. 拖线时检查方向和类型。
4. 用 schema 驱动属性编辑。
5. 保存、加载完整 FlowDocument + EditorMetadata。
6. 把 trace 映射为节点状态和端口产物摘要。

状态展示应稳定，不因文案或图标改变节点尺寸：

不要在第一版加入多人协作、无限画布优化、自动布局市场、插件商店或真实设备控制。这些都不是验证 Flow 语义的必要条件。

工程验收清单

1. Typed graph editor

交付物：React Flow 节点面板、typed handles、拖线校验、保存加载和后端二次校验。

验收：错误类型不能连接；手改 JSON 后 runtime 仍能拒绝；保存再加载保持端口和配置不变。

面试价值：证明前端交互、schema 建模和执行契约能连成一个系统，而不只是画一张流程图。

2. Streaming frame demo

交付物：模拟相机连续产生 frame handle，检测节点较慢，有界队列提供可切换背压策略。

验收：长时间运行内存不随消息数无限增长；trace 能显示丢弃、等待或采样数量；取消后 source 和下游都停止。

面试价值：证明理解实时数据流、资源上界和可观察性，适合机器人应用软件与工业数据平台方向。

3. Simulate-only safety workflow

交付物：工位 barrier、数值 gate、run-mode 矩阵和审计 trace；runner 使用 mock，默认 simulate。

验收：缺 runner、barrier 未就绪或 gate 拒绝时均 fail-closed；任何测试都不能产生真实 I/O。

面试价值：证明能讨论工业副作用、权限边界和故障模式，而不是把“工作流执行”简化为函数调用。

对转型和面试的价值

这两篇最适合支撑以下定位：

• 工业数字孪生与机器人可视化工程师。
• 机器人应用软件工程师。
• 设备集成、HMI 和工业数据流方向。
• 工业视觉工作流与 AI 应用工程师。

它利用了已有的 TypeScript、前端交互和系统架构优势，同时补上图校验、调度语义、背压、trace 和工业安全边界。

需要保持职业表述克制：typed graph demo 和 simulate workflow 能证明系统软件思维，但不能证明实时控制、机器人全身控制、真实产线调试或商业机器人交付。要让能力更可信，下一步应补一段可运行演示视频、架构图、失败案例和测试报告。

这篇要解决什么问题

我熟悉 TypeScript、组件状态和异步调用，但第一次看 Flow Graph 时，容易被 DAG、typed port、拓扑排序、barrier 等术语同时淹没。

这篇只解决一个小问题：如何把一组有类型的 TypeScript 函数，组织成一张能在运行前校验、按依赖顺序执行、并留下 trace 的图。

我们用一个虚构的工业视觉流程贯穿全文：

1

启动 -> 模拟采集图像 -> 模拟缺陷检测 -> 计算调整量 -> 安全检查 -> 模拟执行

所有数据都是固定 fixture，不连接相机、PLC 或机器人。最终产物是一个最小批式 DAG 执行器，不是假装已经可用于生产的设备控制系统。

1. 先用熟悉的 TypeScript 理解端口

先不谈图。下面是普通函数调用：

1
2
3

function detect(frame: ImageFrame): DetectionResult {
  return { frameId: frame.frameId, defectFound: true, score: 0.92 };
}

可以把 Flow Graph 中的概念理解成：

如果 capture 输出 ImageFrame，detect 输入也要求 ImageFrame，这条边可以连接。若把 DetectionResult 接到要求 trigger 的端口，就像把对象传给只接收布尔信号的函数，应该在运行前报错。

图比手写函数调用多做四件事：

1. 校验：端口、类型和必填输入是否正确。
2. 调度：根据依赖计算执行顺序。
3. 观察：记录哪个节点成功、失败和耗时。
4. 持久化：把节点和边保存成 JSON，而不是把流程写死在调用栈中。

2. DAG、数据流图和状态机不是一回事

“都是方框和连线”不代表运行语义相同。

本文做的是 typed dataflow（有类型的数据流图），同时把顶层限制为 DAG（Directed Acyclic Graph，有向无环图）。

判断一张图是否真正表达了 typed dataflow，可以问：

• 边是否精确连接到端口，而不只是连接节点？
• 端口是否声明数据类型？
• 下游是否直接消费上游产物，而不是从全局对象里猜变量名？
• 图加载时能否发现类型错误和漏接输入？

3. 本文要实现的工业视觉流程

1
2
3
4
5
6

flowchart LR
    S[Start] -->|trigger| C[Capture]
    C -->|ImageFrame| D[Detect]
    D -->|DetectionResult| A[Compute Adjustment]
    A -->|Adjustment| G[Safety Check]
    G -->|ApprovedAdjustment| E[Simulate Execute]

示例数据固定为：

• 图像编号：frame-001。
• 缺陷分数：0.92。
• 建议调整量：0.6 mm。
• 允许的最大调整量：2 mm。

这里的 safety check 只是解释数据流和 fail-closed（条件不满足就拒绝继续）的最小示例，不代表真实机器人系统只需要比较一个数值。

4. 初始化实验目录

以下代码按 Node.js 24 编写，利用其直接运行可擦除 TypeScript 类型的能力：

1
2
3
4
5
6

mkdir flow-graph-lab
cd flow-graph-lab
npm init -y
npm pkg set type=module
npm install -D typescript @types/node
mkdir -p src/flow

创建 tsconfig.json：

1
2
3
4
5
6
7
8
9
10
11

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true
  },
  "include": ["src/**/*.ts"]
}

相对 import 显式带 .ts 后缀，既能通过上述配置检查，也能由 Node 24 直接执行。

5. 定义图、节点、端口和边

新建 src/flow/types.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

export type PortDataType =
  | "trigger"
  | "image_frame"
  | "detection"
  | "adjustment"
  | "approved_adjustment";

export interface ImageFrame {
  frameId: string;
  width: number;
  height: number;
}

export interface DetectionResult {
  frameId: string;
  defectFound: boolean;
  score: number;
}

export interface Adjustment {
  frameId: string;
  offsetMm: number;
}

export interface ApprovedAdjustment extends Adjustment {
  approved: true;
}

export interface PortDefinition {
  id: string;
  dataType: PortDataType;
  required?: boolean;
}

export interface FlowNode {
  id: string;
  type: string;
  config: Record<string, unknown>;
}

export interface FlowEdge {
  id: string;
  from: { node: string; port: string };
  to: { node: string; port: string };
}

export interface FlowDocument {
  schemaVersion: 1;
  id: string;
  name: string;
  nodes: FlowNode[];
  edges: FlowEdge[];
}

export type PortValues = Record<string, unknown>;

export interface RunContext {
  runId: string;
  signal: AbortSignal;
  log(message: string, fields?: Record<string, unknown>): void;
}

export interface NodeDefinition {
  type: string;
  inputs: PortDefinition[];
  outputs: PortDefinition[];
  run(
    inputs: PortValues,
    config: Record<string, unknown>,
    context: RunContext,
  ): Promise<PortValues>;
}

export function isTypeCompatible(
  output: PortDataType,
  input: PortDataType,
): boolean {
  return output === input;
}

这里刻意没有 any。示例规模小，先要求类型完全相等，能更清楚地观察类型校验的价值。

RunContext 只保存运行控制和日志能力，不用来传 ImageFrame 等业务数据。业务数据沿边传递，才能看清来源和去向。

6. 用注册表统一节点契约

图实例只保存节点 ID、节点类型和配置。某种节点有哪些端口、如何运行，由注册表统一定义。

新建 src/flow/registry.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import type { NodeDefinition } from "./types.ts";

export class NodeRegistry {
  private readonly definitions = new Map<string, NodeDefinition>();

  register(definition: NodeDefinition): this {
    if (this.definitions.has(definition.type)) {
      throw new Error(`node type already registered: ${definition.type}`);
    }
    this.definitions.set(definition.type, definition);
    return this;
  }

  get(type: string): NodeDefinition | undefined {
    return this.definitions.get(type);
  }

  require(type: string): NodeDefinition {
    const definition = this.get(type);
    if (!definition) throw new Error(`unknown node type: ${type}`);
    return definition;
  }
}

再新建 src/flow/example-nodes.ts。这些节点全部使用固定数据，只用于模拟：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

import { NodeRegistry } from "./registry.ts";
import type {
  Adjustment,
  ApprovedAdjustment,
  DetectionResult,
  ImageFrame,
  NodeDefinition,
} from "./types.ts";

function objectInput(value: unknown, name: string): Record<string, unknown> {
  if (typeof value !== "object" || value === null) {
    throw new Error(`${name} must be an object`);
  }
  return value as Record<string, unknown>;
}

const start: NodeDefinition = {
  type: "start",
  inputs: [],
  outputs: [{ id: "trigger", dataType: "trigger" }],
  async run() {
    return { trigger: true };
  },
};

const capture: NodeDefinition = {
  type: "capture",
  inputs: [{ id: "trigger", dataType: "trigger", required: true }],
  outputs: [{ id: "frame", dataType: "image_frame" }],
  async run(_inputs, config) {
    if (typeof config.frameId !== "string") {
      throw new Error("capture.config.frameId must be a string");
    }
    const frame: ImageFrame = {
      frameId: config.frameId,
      width: 1280,
      height: 1024,
    };
    return { frame };
  },
};

const detect: NodeDefinition = {
  type: "detect",
  inputs: [{ id: "frame", dataType: "image_frame", required: true }],
  outputs: [{ id: "detection", dataType: "detection" }],
  async run(inputs, config) {
    const frame = objectInput(inputs.frame, "detect.frame");
    if (typeof frame.frameId !== "string") {
      throw new Error("detect.frame.frameId must be a string");
    }
    if (typeof config.defectFound !== "boolean" || typeof config.score !== "number") {
      throw new Error("detect config is invalid");
    }
    const detection: DetectionResult = {
      frameId: frame.frameId,
      defectFound: config.defectFound,
      score: config.score,
    };
    return { detection };
  },
};

const computeAdjustment: NodeDefinition = {
  type: "compute_adjustment",
  inputs: [{ id: "detection", dataType: "detection", required: true }],
  outputs: [{ id: "adjustment", dataType: "adjustment" }],
  async run(inputs, config) {
    const detection = objectInput(inputs.detection, "compute_adjustment.detection");
    if (typeof detection.frameId !== "string" || typeof config.offsetMm !== "number") {
      throw new Error("compute_adjustment input or config is invalid");
    }
    const adjustment: Adjustment = {
      frameId: detection.frameId,
      offsetMm: detection.defectFound === true ? config.offsetMm : 0,
    };
    return { adjustment };
  },
};

const safetyCheck: NodeDefinition = {
  type: "safety_check",
  inputs: [{ id: "adjustment", dataType: "adjustment", required: true }],
  outputs: [{ id: "approved", dataType: "approved_adjustment" }],
  async run(inputs, config) {
    const adjustment = objectInput(inputs.adjustment, "safety_check.adjustment");
    if (typeof adjustment.frameId !== "string" ||
        typeof adjustment.offsetMm !== "number" ||
        typeof config.maxOffsetMm !== "number") {
      throw new Error("safety_check input or config is invalid");
    }
    if (Math.abs(adjustment.offsetMm) > config.maxOffsetMm) {
      throw new Error("adjustment rejected by safety check");
    }
    const approved: ApprovedAdjustment = {
      frameId: adjustment.frameId,
      offsetMm: adjustment.offsetMm,
      approved: true,
    };
    return { approved };
  },
};

const simulateExecute: NodeDefinition = {
  type: "simulate_execute",
  inputs: [{ id: "adjustment", dataType: "approved_adjustment", required: true }],
  outputs: [],
  async run(inputs, _config, context) {
    const adjustment = objectInput(inputs.adjustment, "simulate_execute.adjustment");
    if (adjustment.approved !== true) {
      throw new Error("simulate_execute requires an approved adjustment");
    }
    context.log("simulation only: no hardware command sent", {
      frameId: adjustment.frameId,
      offsetMm: adjustment.offsetMm,
    });
    return {};
  },
};

export function createExampleRegistry(): NodeRegistry {
  return new NodeRegistry()
    .register(start)
    .register(capture)
    .register(detect)
    .register(computeAdjustment)
    .register(safetyCheck)
    .register(simulateExecute);
}

注册表是契约的单一真源：校验器读取端口，执行器读取 runner，将来编辑器也可以据此生成节点面板。

7. 在运行前校验图

可靠 Flow Graph 的第一价值不是拖线，而是把错误从“设备运行到一半”提前为“图加载失败”。

1
2
3
4
5
6

flowchart LR
    A[节点 ID 与类型] --> B[边端点]
    B --> C[端口]
    C --> D[类型]
    D --> E[必填输入]
    E --> F[无环]

新建 src/flow/validate.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

import type { NodeRegistry } from "./registry.ts";
import { isTypeCompatible } from "./types.ts";
import type { FlowDocument, PortDefinition } from "./types.ts";

export class GraphValidationError extends Error {
  readonly issues: string[];

  constructor(issues: string[]) {
    super(`invalid flow graph:\n${issues.map((issue) => `- ${issue}`).join("\n")}`);
    this.issues = issues;
  }
}

function findPort(
  ports: PortDefinition[],
  portId: string,
): PortDefinition | undefined {
  return ports.find((port) => port.id === portId);
}

export function validateGraph(
  graph: FlowDocument,
  registry: NodeRegistry,
): void {
  const issues: string[] = [];
  const seenNodeIds = new Set<string>();

  for (const node of graph.nodes) {
    if (seenNodeIds.has(node.id)) issues.push(`duplicate node id: ${node.id}`);
    seenNodeIds.add(node.id);
    if (!registry.get(node.type)) issues.push(`unknown node type: ${node.type}`);
  }

  const nodes = new Map(graph.nodes.map((node) => [node.id, node]));
  const incomingCount = new Map<string, number>();

  for (const edge of graph.edges) {
    const source = nodes.get(edge.from.node);
    const target = nodes.get(edge.to.node);

    if (!source) {
      issues.push(`edge ${edge.id} references missing source node ${edge.from.node}`);
      continue;
    }
    if (!target) {
      issues.push(`edge ${edge.id} references missing target node ${edge.to.node}`);
      continue;
    }

    const sourceDefinition = registry.get(source.type);
    const targetDefinition = registry.get(target.type);
    if (!sourceDefinition || !targetDefinition) continue;

    const output = findPort(sourceDefinition.outputs, edge.from.port);
    const input = findPort(targetDefinition.inputs, edge.to.port);

    if (!output) {
      issues.push(`${source.id} has no output port ${edge.from.port}`);
      continue;
    }
    if (!input) {
      issues.push(`${target.id} has no input port ${edge.to.port}`);
      continue;
    }
    if (!isTypeCompatible(output.dataType, input.dataType)) {
      issues.push(
        `type mismatch: ${source.id}.${output.id}(${output.dataType}) -> ` +
        `${target.id}.${input.id}(${input.dataType})`,
      );
    }

    const inputKey = `${target.id}.${input.id}`;
    incomingCount.set(inputKey, (incomingCount.get(inputKey) ?? 0) + 1);
  }

  for (const node of graph.nodes) {
    const definition = registry.get(node.type);
    if (!definition) continue;
    for (const input of definition.inputs) {
      if (input.required && (incomingCount.get(`${node.id}.${input.id}`) ?? 0) === 0) {
        issues.push(`required input is unconnected: ${node.id}.${input.id}`);
      }
    }
  }

  issues.push(...detectCycle(graph));

  if (issues.length > 0) throw new GraphValidationError(issues);
}

function detectCycle(graph: FlowDocument): string[] {
  const indegree = new Map(graph.nodes.map((node) => [node.id, 0]));
  const outgoing = new Map(graph.nodes.map((node) => [node.id, [] as string[]]));

  for (const edge of graph.edges) {
    if (!indegree.has(edge.from.node) || !indegree.has(edge.to.node)) continue;
    indegree.set(edge.to.node, (indegree.get(edge.to.node) ?? 0) + 1);
    outgoing.get(edge.from.node)?.push(edge.to.node);
  }

  const queue = graph.nodes
    .map((node) => node.id)
    .filter((nodeId) => indegree.get(nodeId) === 0);
  let visited = 0;

  while (queue.length > 0) {
    const nodeId = queue.shift();
    if (!nodeId) break;
    visited += 1;
    for (const targetId of outgoing.get(nodeId) ?? []) {
      const next = (indegree.get(targetId) ?? 0) - 1;
      indegree.set(targetId, next);
      if (next === 0) queue.push(targetId);
    }
  }

  return visited === graph.nodes.length
    ? []
    : ["graph contains a cycle"];
}

前端可以在拖线时立即检查类型，改善体验；运行端仍必须重新校验，因为 JSON 也可能来自文件、CLI 或旧版本客户端。

8. 编译为静态执行计划

图只在加载时变化，执行器不必每次都扫描所有边和重复计算顺序。编译阶段完成两件事：

1. 用 Kahn 算法得到稳定的拓扑顺序。
2. 把每个节点的入边预先绑定好。

1
2
3
4
5

flowchart LR
    G[FlowDocument] --> V[Validate]
    V --> T[Topological order]
    T --> B[Bind incoming edges]
    B --> P[ExecutionPlan]

新建 src/flow/compile.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import type { NodeRegistry } from "./registry.ts";
import type { FlowDocument, FlowEdge } from "./types.ts";
import { validateGraph } from "./validate.ts";

export interface ExecutionPlan {
  graph: FlowDocument;
  order: string[];
  incomingByNode: Map<string, FlowEdge[]>;
}

export function compileGraph(
  graph: FlowDocument,
  registry: NodeRegistry,
): ExecutionPlan {
  validateGraph(graph, registry);

  const indegree = new Map(graph.nodes.map((node) => [node.id, 0]));
  const outgoing = new Map(graph.nodes.map((node) => [node.id, [] as string[]]));
  const incomingByNode = new Map(
    graph.nodes.map((node) => [node.id, [] as FlowEdge[]]),
  );

  for (const edge of graph.edges) {
    indegree.set(edge.to.node, (indegree.get(edge.to.node) ?? 0) + 1);
    outgoing.get(edge.from.node)?.push(edge.to.node);
    incomingByNode.get(edge.to.node)?.push(edge);
  }

  const queue = graph.nodes
    .map((node) => node.id)
    .filter((nodeId) => indegree.get(nodeId) === 0);
  const order: string[] = [];

  while (queue.length > 0) {
    const nodeId = queue.shift();
    if (!nodeId) break;
    order.push(nodeId);
    for (const targetId of outgoing.get(nodeId) ?? []) {
      const next = (indegree.get(targetId) ?? 0) - 1;
      indegree.set(targetId, next);
      if (next === 0) queue.push(targetId);
    }
  }

  return { graph, order, incomingByNode };
}

“编辑是 Flow，执行是静态计划”并不矛盾。Flow 是用户看见和保存的语义；静态计划是运行端为了确定性和效率生成的内部结构。

9. 实现确定性执行器

执行器按计划依次运行节点，把上游输出放入目标输入端口，并记录最小 trace。

新建 src/flow/executor.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

import type { ExecutionPlan } from "./compile.ts";
import type { NodeRegistry } from "./registry.ts";
import type { PortValues, RunContext } from "./types.ts";

export interface TraceEvent {
  nodeId: string;
  status: "running" | "success" | "failed";
  durationMs?: number;
  error?: string;
}

export interface RunResult {
  outputs: Map<string, unknown>;
  trace: TraceEvent[];
}

function outputKey(nodeId: string, portId: string): string {
  return `${nodeId}.${portId}`;
}

export class FlowExecutor {
  private readonly plan: ExecutionPlan;
  private readonly registry: NodeRegistry;

  constructor(
    plan: ExecutionPlan,
    registry: NodeRegistry,
  ) {
    this.plan = plan;
    this.registry = registry;
  }

  async run(context: RunContext): Promise<RunResult> {
    const nodes = new Map(this.plan.graph.nodes.map((node) => [node.id, node]));
    const outputs = new Map<string, unknown>();
    const trace: TraceEvent[] = [];

    for (const nodeId of this.plan.order) {
      if (context.signal.aborted) throw new Error(`run ${context.runId} cancelled`);

      const node = nodes.get(nodeId);
      if (!node) throw new Error(`compiled node is missing: ${nodeId}`);
      const definition = this.registry.require(node.type);
      const inputs: PortValues = {};

      for (const edge of this.plan.incomingByNode.get(nodeId) ?? []) {
        const key = outputKey(edge.from.node, edge.from.port);
        if (!outputs.has(key)) throw new Error(`upstream output is missing: ${key}`);
        inputs[edge.to.port] = outputs.get(key);
      }

      const startedAt = performance.now();
      trace.push({ nodeId, status: "running" });

      try {
        const result = await definition.run(inputs, node.config, context);
        for (const port of definition.outputs) {
          if (!(port.id in result)) {
            throw new Error(`declared output is missing: ${nodeId}.${port.id}`);
          }
          outputs.set(outputKey(nodeId, port.id), result[port.id]);
        }
        trace.push({
          nodeId,
          status: "success",
          durationMs: performance.now() - startedAt,
        });
      } catch (error) {
        const message = error instanceof Error ? error.message : String(error);
        trace.push({
          nodeId,
          status: "failed",
          durationMs: performance.now() - startedAt,
          error: message,
        });
        throw new Error(`node ${nodeId} failed: ${message}`, { cause: error });
      }
    }

    return { outputs, trace };
  }
}

这个版本刻意顺序执行。先把输入收集、错误传播和结果可重复做正确，再讨论并行与流式，否则很难判断错误来自基础语义还是并发调度。

10. 运行完整示例

新建 src/main.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import { compileGraph } from "./flow/compile.ts";
import { createExampleRegistry } from "./flow/example-nodes.ts";
import { FlowExecutor } from "./flow/executor.ts";
import type { FlowDocument, RunContext } from "./flow/types.ts";

const graph: FlowDocument = {
  schemaVersion: 1,
  id: "vision-inspection-demo",
  name: "Vision Inspection Demo",
  nodes: [
    { id: "start", type: "start", config: {} },
    { id: "capture", type: "capture", config: { frameId: "frame-001" } },
    { id: "detect", type: "detect", config: { defectFound: true, score: 0.92 } },
    { id: "adjust", type: "compute_adjustment", config: { offsetMm: 0.6 } },
    { id: "gate", type: "safety_check", config: { maxOffsetMm: 2 } },
    { id: "execute", type: "simulate_execute", config: {} },
  ],
  edges: [
    { id: "e1", from: { node: "start", port: "trigger" }, to: { node: "capture", port: "trigger" } },
    { id: "e2", from: { node: "capture", port: "frame" }, to: { node: "detect", port: "frame" } },
    { id: "e3", from: { node: "detect", port: "detection" }, to: { node: "adjust", port: "detection" } },
    { id: "e4", from: { node: "adjust", port: "adjustment" }, to: { node: "gate", port: "adjustment" } },
    { id: "e5", from: { node: "gate", port: "approved" }, to: { node: "execute", port: "adjustment" } },
  ],
};

const registry = createExampleRegistry();
const executor = new FlowExecutor(compileGraph(graph, registry), registry);
const context: RunContext = {
  runId: "run-001",
  signal: new AbortController().signal,
  log(message, fields) {
    console.log(message, fields ?? {});
  },
};

const result = await executor.run(context);
console.log("detection", result.outputs.get("detect.detection"));
console.log("approved", result.outputs.get("gate.approved"));
console.log(
  "order",
  result.trace
    .filter((event) => event.status === "success")
    .map((event) => event.nodeId),
);

运行：

1

node src/main.ts

输出重点如下：

1
2
3
4

simulation only: no hardware command sent { frameId: 'frame-001', offsetMm: 0.6 }
detection { frameId: 'frame-001', defectFound: true, score: 0.92 }
approved { frameId: 'frame-001', offsetMm: 0.6, approved: true }
order [ 'start', 'capture', 'detect', 'adjust', 'gate', 'execute' ]

11. 用测试锁定行为

新建 src/flow/flow.test.ts：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

import assert from "node:assert/strict";
import test from "node:test";
import { compileGraph } from "./compile.ts";
import { createExampleRegistry } from "./example-nodes.ts";
import { FlowExecutor } from "./executor.ts";
import { NodeRegistry } from "./registry.ts";
import type { FlowDocument, NodeDefinition, RunContext } from "./types.ts";
import { GraphValidationError, validateGraph } from "./validate.ts";

function createGraph(): FlowDocument {
  return {
    schemaVersion: 1,
    id: "vision-inspection-demo",
    name: "Vision Inspection Demo",
    nodes: [
      { id: "start", type: "start", config: {} },
      { id: "capture", type: "capture", config: { frameId: "frame-001" } },
      { id: "detect", type: "detect", config: { defectFound: true, score: 0.92 } },
      { id: "adjust", type: "compute_adjustment", config: { offsetMm: 0.6 } },
      { id: "gate", type: "safety_check", config: { maxOffsetMm: 2 } },
      { id: "execute", type: "simulate_execute", config: {} },
    ],
    edges: [
      { id: "e1", from: { node: "start", port: "trigger" }, to: { node: "capture", port: "trigger" } },
      { id: "e2", from: { node: "capture", port: "frame" }, to: { node: "detect", port: "frame" } },
      { id: "e3", from: { node: "detect", port: "detection" }, to: { node: "adjust", port: "detection" } },
      { id: "e4", from: { node: "adjust", port: "adjustment" }, to: { node: "gate", port: "adjustment" } },
      { id: "e5", from: { node: "gate", port: "approved" }, to: { node: "execute", port: "adjustment" } },
    ],
  };
}

test("executes the industrial vision flow in deterministic order", async () => {
  const registry = createExampleRegistry();
  const plan = compileGraph(createGraph(), registry);
  const executor = new FlowExecutor(plan, registry);
  const logs: string[] = [];
  const context: RunContext = {
    runId: "run-001",
    signal: new AbortController().signal,
    log(message) { logs.push(message); },
  };

  const result = await executor.run(context);

  assert.deepEqual(
    result.trace.filter((event) => event.status === "success").map((event) => event.nodeId),
    ["start", "capture", "detect", "adjust", "gate", "execute"],
  );
  assert.deepEqual(result.outputs.get("detect.detection"), {
    frameId: "frame-001",
    defectFound: true,
    score: 0.92,
  });
  assert.deepEqual(result.outputs.get("gate.approved"), {
    frameId: "frame-001",
    offsetMm: 0.6,
    approved: true,
  });
  assert.deepEqual(logs, ["simulation only: no hardware command sent"]);
});

test("rejects incompatible port types", () => {
  const graph = createGraph();
  graph.edges[1] = {
    id: "bad-type",
    from: { node: "detect", port: "detection" },
    to: { node: "capture", port: "trigger" },
  };

  assert.throws(
    () => validateGraph(graph, createExampleRegistry()),
    (error) => error instanceof GraphValidationError &&
      error.issues.some((issue) => issue.includes("type mismatch")),
  );
});

test("rejects an unconnected required input", () => {
  const graph = createGraph();
  graph.edges = graph.edges.filter((edge) => edge.to.node !== "detect");

  assert.throws(
    () => validateGraph(graph, createExampleRegistry()),
    (error) => error instanceof GraphValidationError &&
      error.issues.includes("required input is unconnected: detect.frame"),
  );
});

test("rejects an unknown port", () => {
  const graph = createGraph();
  graph.edges[1] = {
    id: "bad-port",
    from: { node: "capture", port: "missing" },
    to: { node: "detect", port: "frame" },
  };

  assert.throws(
    () => validateGraph(graph, createExampleRegistry()),
    (error) => error instanceof GraphValidationError &&
      error.issues.includes("capture has no output port missing"),
  );
});

test("rejects a cycle", () => {
  const loop: NodeDefinition = {
    type: "loop",
    inputs: [{ id: "in", dataType: "trigger", required: true }],
    outputs: [{ id: "out", dataType: "trigger" }],
    async run() { return { out: true }; },
  };
  const registry = new NodeRegistry().register(loop);
  const graph: FlowDocument = {
    schemaVersion: 1,
    id: "cycle",
    name: "Cycle",
    nodes: [
      { id: "a", type: "loop", config: {} },
      { id: "b", type: "loop", config: {} },
    ],
    edges: [
      { id: "ab", from: { node: "a", port: "out" }, to: { node: "b", port: "in" } },
      { id: "ba", from: { node: "b", port: "out" }, to: { node: "a", port: "in" } },
    ],
  };

  assert.throws(
    () => validateGraph(graph, registry),
    (error) => error instanceof GraphValidationError &&
      error.issues.some((issue) => issue.includes("cycle")),
  );
});

运行类型检查和测试：

1
2

npx tsc
node src/flow/flow.test.ts

五个测试分别证明：正常链路顺序稳定、类型错误会被拒绝、必填输入不能漏接、端口名不能写错、顶层图不能有环。

12. 当前版本没有解决什么

这个执行器已经形成 schema -> validate -> compile -> execute -> trace 闭环，但它仍然只是最小批式 DAG：

• 没有 branch 的 skipped 语义。
• 没有并行执行和 fan-in barrier 的运行时状态。
• 没有多消息流、有界队列和 backpressure（背压）。
• 没有 timeout、retry 和真实 I/O 的取消传播。
• 没有运行时 payload schema，只校验端口声明的类型名称。
• 没有任何真实设备能力。

这些不是“少写几段代码”的问题，而是需要先定义清楚运行语义。下篇会逐项拆开。

验收与下一步

完成本文后，应能独立回答：

1. typed edge 相比共享黑板解决了什么问题？
2. 为什么前端拖线校验不能替代运行端校验？
3. 为什么图要先编译再反复运行？
4. RunContext 为什么不应该承载主要业务数据？
5. 当前执行器为什么还不能连接真实机器人？

练习：

1. 增加重复 edge ID 校验，并写一个失败测试。
2. 增加无硬件能力的 manual_review 节点，输出人工确认结果。
3. 为 ImageFrame 和 DetectionResult 增加运行时 payload 校验。

这份最小实现能证明 typed dataflow、图校验和确定性调度的基础能力。它对工业可视化、机器人应用软件和系统集成岗位有价值，但不证明真实设备调试或运动控制能力。

下一篇：Flow Graph 进阶：可视化、流式执行与工业安全

AI Agent 已经不只是“模型加几个工具”。真正拉开实际效果差距的，往往是任务怎样拆分、上下文怎样隔离、工具怎样授权、状态怎样持久化，以及失败后怎样验证和恢复。

这篇文章不是产品大全，也不试图追完所有论文。它更像一份工作手册：遇到调研、编码、业务自动化、工业软件或机器人任务时，先判断应该采用哪种工作模式，再选择合适的 Harness、Skills、协议和运行框架。

最后核对：2026-07-22。

一、先看结论：按任务选择工作模式

推荐的复杂度升级顺序是：

1
2
3
4
5
6

强单 Agent 基线
  -> 把重复做法提炼为 Skills
  -> 为上下文隔离引入少量 subagents
  -> 为稳定分支引入确定性 workflow
  -> 为长任务增加持久化与恢复
  -> 只有任务宽度和并行收益可测时才使用 swarm

多 Agent 不是默认升级。它只有在任务可分解、子结果可验收、协调成本低于收益时才值得使用。

二、从 Chat、Agent 到 Agent Team

普通 Chat 的核心是生成回答；Agent 的核心是形成“观察环境、选择动作、读取结果、继续或停止”的循环。Agent Team 则进一步增加委派、隔离、同步、仲裁和聚合。

这几个概念经常被混在一起，可以用下面的分层理解：

1
2
3
4
5
6
7

Model
  -> Agent loop / reasoning policy
    -> Skills and instructions
      -> Harness and runtime
        -> MCP / native tools
          -> A2A
            -> Multi-agent workflow

这不是强制调用栈，而是一张选型地图：

因此，“某产品支持 MCP”不等于它具备多智能体协作；“写了一组角色提示”也不等于得到了可恢复的工作流。

三、六类值得掌握的工作模式

1. Skills-driven Agent：把经验变成可复用流程

Skills 替代的是散落在 Wiki、聊天记录和个人记忆中的 SOP。它适合已经反复出现、步骤和验收逐渐稳定，但每次输入仍不同的任务。

一个实用 Skill 通常包含：

• 何时触发，以及何时不触发；
• 固定的检查顺序和禁止动作；
• 需要读取的参考资料、脚本或模板；
• 产物格式和验收命令；
• 失败时停止、升级或交给人工的条件。

Skill 本身不会创建并行，也不会提供凭证、持久状态和事务恢复。它编码的是流程知识，不是运行时。

最小试验：把“ROS2 故障信息采集”做成一个 Skill，只规定 topic、TF、日志和参数的只读检查顺序，在两个不同故障上比较漏项数、上下文用量和结论可复现性。

2. Subagents：用隔离上下文处理少量独立子任务

Subagent 适合少量、边界清楚、可独立交付摘要或 patch 的任务。例如主 Agent 负责整体功能，子 Agent 分别调查 API 契约、前端实现位置和测试策略。

关键不是给它们起“架构师”“批评家”的名字，而是明确：

• 输入范围；
• 文件所有权；
• 输出契约；
• 局部验收；
• 与其他子任务的依赖。

只读调查和不同文件所有权可以并行；共享接口应先约定；同一文件最好维持单写者。子 Agent 的同意不能替代测试，汇总后仍要运行跨模块验证。

最小试验：把一个小功能拆成“接口核对、UI 实现、测试设计”三个任务，与单 Agent 对照总 token、墙钟时间、冲突数和最终测试结果。

3. Agent Swarm：让宽搜索空间动态扩张

Swarm 适合大量或数量未知的同构分片，例如批量扫描文档、并行搜索候选、对大量文件抽取相同字段。它的核心不只是“并发更多”，而是协调者可以动态派生任务，并通过共享 artifact 聚合结果。

Swarm 的前置条件比演示中看起来更严格：

• 有明确的分片键和去重规则；
• 每个分片可以独立失败和重试；
• 聚合器能验证覆盖率和引用；
• 有最大深度、并发、工具调用、费用和超时限制；
• 有强单 Agent 基线作为对照。

Kimi 在 2026 年 2 月的 Agent Swarm 技术博客中披露：基于当时 K2.5 的系统可部署最多 100 个 subagents、执行超过 1,500 次工具调用，并在其测试场景中相对串行执行快 4.5 倍。这是厂商对特定系统和任务的披露，不是独立通用基准，也不能直接归因给 2026 年 7 月发布的 K3。

最小试验：对 30 份设备手册并行提取故障码，最多使用 5 个工作 Agent 和固定工具预算，与单 Agent 比较字段召回、错误引用、总成本和墙钟时间。

4. 确定性 Workflow：让 LLM 只处理不确定节点

当步骤、状态转移、重试、幂等和审计能够提前定义时，应让普通软件控制流程，让 Agent 只处理分类、抽取、检索或候选生成等非结构化节点。

例如：

1
2
3
4
5
6
7

告警进入
  -> 拉取设备上下文
  -> Agent 生成归因候选
  -> 规则校验
  -> 人工批准
  -> 确定性执行器创建工单
  -> 记录结果与审计

这类架构比“让 Agent 自己决定下一步”更容易做状态测试、故障注入、重试和补偿。LangGraph、Google ADK、Microsoft Agent Framework 可用于表达 Agent workflow；Temporal 等持久工作流系统则适合承载更严格的长任务恢复，但 Temporal 本身不是 Agent 框架。

最小试验：实现五步告警流程，模拟超时、重复消息和进程重启，验证不会重复建单，并能从中断点恢复。

5. 后台长任务：把状态从聊天窗口移出去

跨分钟、小时或天的任务，不能只依赖一个持续增长的上下文窗口。计划、检查点、artifact、权限租约、事件日志和预算应该外置；每次唤醒只重建最小上下文。

长任务至少要具备：

• 可查询进度；
• 可取消、暂停和恢复；
• 心跳、租约和超时；
• 幂等键与重复投递处理；
• artifact 和原始证据可追溯；
• 外部环境或仓库版本漂移检测。

最小试验：让后台 Agent 每小时读取一次模拟设备日志并更新报告，运行 8 小时，中途杀死 worker，验证恢复后没有重复告警。

6. Human-in-the-loop：在具体动作前设置治理闸门

人工审批适合发布、付费、删除、外部通信、权限升级、设备写入，以及超出预授权风险包络的动作。审批界面应展示具体目标、参数、证据、diff、影响范围和回滚方案，而不是让人对一句模糊意图点“同意”。

已经过风险评估、位于预授权确定性流程内、受限额和监控约束的低风险动作，可以按策略自动执行。闸门过多会制造审批疲劳。

必须强调：人工审批是治理措施，不是功能安全措施，不能替代安全 PLC、安全控制器、参数限制、联锁、风险评估和适用标准。

四、复杂任务应该怎样拆

高质量拆分不是生成一份自然语言待办列表，而是建立一张可执行、可验证的任务图。

1. 先画依赖，再谈并行

每个节点至少写清：

1
2
3
4
5
6
7
8

目标
输入与允许读取的上下文
输出 artifact
文件或资源所有权
依赖节点
验收条件
预算与超时
失败后的重试、降级或升级路径

任务图中“已经就绪且互不冲突的节点数”只是结构上的并行上限。实际并发还会被 API 限流、token 预算、工具资源、文件冲突和协调成本进一步压缩。

2. 上下文按职责分区

不要把完整仓库、全部聊天历史和所有工具权限广播给每个 Agent。主 Agent 保存全局目标和依赖，子 Agent 只获得完成任务所需的最小材料；共享事实落入文件、数据库或 artifact，而不是依赖转述记忆。

3. 单写者，多个读者

多 Agent 编码最常见的失败不是不会写代码，而是同时修改共享文件。可以并行搜索、阅读和独立模块实现，但共享接口与高冲突文件应由一个 Agent 写入，其他 Agent 返回建议或 patch 草案。

4. 验收器必须提供独立证据

好的验收器包括测试、类型检查、schema 校验、截图对比、查询结果、仿真输出和人工检查。另一个 LLM reviewer 可以发现问题，但“两个 Agent 都认为正确”并不是独立证据。

5. 重规划应该由事件触发

只有依赖变化、验收失败、预算超限、外部资源不可用或新证据推翻假设时才重规划。每一步都重新生成全局计划，会浪费上下文，也容易让目标漂移。

五、Skills：Superpowers、Matt Pocock与方法论复用

Agent Skills Specification正在形成一种可移植的目录约定：入口 SKILL.md 描述触发条件和流程，脚本、模板、参考资料按需加载。它的重要价值是渐进式上下文，而不是把所有知识塞进系统提示。

Superpowers

obra/superpowers把 brainstorming、计划、TDD、调试、代码审查、验证和分支收尾做成强约束的流程 Skills。它适合需要提高工程纪律、减少 Agent 跳步的任务，尤其适合已有测试和 Git 工作流的代码库。

它的限制也很明确：

• 流程有额外调用和审查成本；
• 依赖宿主 Agent 真正遵守 Skill；
• 对一次性小改动可能显得过重；
• Skill 不能弥补错误的验收标准或缺失的测试环境。

对我更有价值的用法，不是完整照搬所有流程，而是挑出“先复现、隔离工作树、规格审查、质量审查、完成前验证”这些高收益约束。

Matt Pocock Skills

mattpocock/skills更像一组可阅读、可改写的编码技能样例。它的价值在于展示如何把领域经验拆成清晰的触发条件、步骤和检查表，适合学习 TypeScript、前端和编码 Agent 的 Skill 写法。

它不是权威标准，也没有统一的跨模型效果基准。个人维护项目的 star、传播量和使用体验只能作为线索，不能证明在自己的仓库里一定有效。

何时自己写 Skill

出现下面三个信号时值得提炼：

• 同类任务反复发生，人工每次都在补充相同约束；
• 成败依赖固定检查顺序或遗漏防护；
• 产物可以通过命令、schema、截图或清单稳定验收。

建议优先写与自己方向直接相关的 Skills：ROS2 故障采集、URDF/TF 检查、工业协议联调记录、前端视觉回归、Agent 工具权限审计。

六、Harness：模型之外真正决定效果的系统

Harness 是模型实际工作的运行环境。它决定：

• 能看到哪些仓库规则和上下文；
• 如何搜索、编辑、运行命令和浏览页面；
• 工具调用怎样授权和隔离；
• 是否支持 Skills、subagents、后台任务和工作树；
• 如何压缩上下文、保存 artifact、运行测试和呈现 diff；
• 失败后能否恢复、取消和审计。

因此同一个模型放在 Codex、Claude Code、KimiCode 或自建 CLI runtime 中，最终效果可能明显不同。Kimi K3 技术页本身也展示了不同评测会使用 KimiCode、Claude Code 或 Codex 等 Harness，这说明成绩不能简单归因于“裸模型”。

评测一个 coding Harness 时，至少锁定：模型版本、Harness 版本、工具集、权限、推理预算、超时、并发、数据集版本、重试次数和验收命令。

对个人开发者，Harness 的优先级通常高于从零自建 Agent 框架。先在真实仓库中比较 Codex、Claude Code、KimiCode 或 OpenHands 的任务通过率、人工返工和工具轨迹，再决定是否需要下沉到 SDK。

七、MCP、A2A与Agent Skills怎样组合

MCP：连接工具和数据

Model Context Protocol定义客户端怎样发现并调用 tools、resources 和 prompts。它适合把数据库、代码托管、文档、浏览器或内部服务以统一方式暴露给 Agent。

MCP 不负责：

• 怎样拆分复杂任务；
• 多个 Agent 怎样协调；
• server 是否可信；
• 业务权限是否正确；
• 调用结果是否真实或安全。

MCP Registry提供发现入口，但“被收录”不等于安全和质量背书。安装第三方 server 前仍要固定版本、审查代码、隔离凭证并授予最小权限。

A2A：连接独立 Agent 系统

A2A 1.0关注独立 Agent 服务之间的发现、任务委派、消息、artifact 和异步协作。它更适合跨团队、跨产品或跨部署域，而不是一个进程内部的简单 subagent。

协议互通只证明消息可以交换。身份、授权委托、租户隔离、参数校验、速率限制、审计和对端信任仍需要应用层设计。

一种实用组合

1
2
3
4
5

Agent Skills：告诉 Agent 应该怎样完成某类任务
Harness：装配模型、上下文、工具、权限和验证
MCP：让 Agent 访问外部工具与数据
A2A：让独立 Agent 服务互相委派与回传
Workflow：管理状态、依赖、重试、审批和收敛

小团队通常先需要 Skills、Harness 和少量 MCP。只有出现独立 Agent 服务、组织边界或异步任务契约时，A2A 才值得进入架构。

八、工具地图：不要按厂商堆功能

日常编码 Harness

Agent 与 Workflow 框架

不要因为框架支持的 Agent 类型最多就选它。对一个可在单进程内完成的小工具，SDK 和 runtime 的学习、部署、观测成本可能比模型调用本身更高。

可观测与评测

Langfuse、OpenTelemetry GenAI 语义、框架自带 tracing 可帮助定位失败发生在检索、推理、工具选择还是执行阶段。但 trace 不是正确性证明，也可能包含 prompt、凭证、代码和设备数据，必须做脱敏和访问控制。

公开 benchmark 也要按任务使用：SWE-bench 看仓库修复，Terminal-Bench 看终端长任务，GAIA 和 BrowseComp 看检索研究，tau-bench 看带规则的工具交互，OSWorld 看 GUI 操作，BFCL 看函数调用。分数只有在版本、Harness、预算和工具条件一致时才可比较。

九、六类场景手册

1. 调研与知识工作

首选模式： 强单 Agent + 动态检索 + 来源台账。材料数量很大、可按主题或文档独立分片时，再增加 subagents；只有分片宽度和聚合指标明确时才试 swarm。

候选工具： Codex/Claude/Kimi 的研究与文件工具、OpenAI Agents SDK、LlamaIndex、浏览器工具、MCP 数据源。

检查项： 来源日期、证据等级、原文链接、重复来源、版本归因、厂商声明与独立评测分离。

最小试验： 调研 20 个同类工具，要求每个结论回链一手来源；比较单 Agent 与 4 个主题 subagents 的覆盖率、错误引用、成本和汇总返工。

2. 编码与大型仓库

首选模式： coding Harness + 仓库规则 + 测试闭环。单仓库修改保持单写者；跨模块只在文件所有权清楚时并行；跨仓库迁移使用任务图和检查点。

候选工具： Codex、Claude Code、KimiCode、OpenHands、Superpowers、自定义 Skills。

检查项： 复现命令、允许修改范围、接口契约、测试与类型检查、工作树隔离、diff 审查、回滚方式。

最小试验： 选择两组难度匹配的真实 bug，在干净基线中交换工具执行顺序，比较一次通过率、人工干预、工具调用和测试覆盖，避免学习效应造成偏差。

3. 产品设计与前端实现

首选模式： 单 Agent 负责整体一致性，视觉素材、设计核对和无重叠组件可并行调查。使用截图、浏览器交互和像素检查作为独立反馈。

候选工具： coding Harness、浏览器自动化、Figma/设计系统连接、图像生成、视觉回归工具。

检查项： 目标用户、既有设计系统、桌面与移动视口、真实交互状态、文本溢出、可访问性、截图验收。

最小试验： 实现一个机器人状态面板，在两个视口运行交互和截图检查；比较仅看代码与加入视觉闭环后的返工项。

4. 运维与业务自动化

首选模式： 确定性 workflow 包围受限 Agent。Agent 负责读日志、分类、提取和提出动作；普通软件负责幂等、限额、重试、补偿、审批和审计。

候选工具： LangGraph、Temporal、Google ADK、Microsoft Agent Framework、MCP、策略与审批系统。

检查项： 最小权限、凭证隔离、dry-run、幂等键、超时、回滚、人工闸门、审计、prompt injection 和数据外泄。

最小试验： 在 mock 工单系统中处理告警，只允许 Agent 生成分类和工单草案；模拟重复消息、API 超时和恶意日志文本。

5. 工业软件与设备集成

首选模式： 设备状态模型和确定性状态机在核心层，Agent 作为只读解释、资料检索和候选诊断节点。低风险、预授权动作也应经过白名单、参数范围和状态前置条件。

候选工具： 普通 workflow runtime、MQTT/OPC UA/Modbus 适配器、时序与事件存储、LangGraph/Agents SDK、只读 MCP server。

检查项： 设备身份、状态新鲜度、数据质量、命令与反馈关联、离线/重连、重放、告警风暴、权限和审计。

最小试验： 建立 Modbus/OPC UA/MQTT 模拟设备平台，Agent 只读事件日志并生成诊断，禁止直接写设备；通过重放比较诊断一致性。

6. 机器人应用与AI辅助运维

首选模式： Agent 位于意图解析、手册检索、候选计划、工具编排和解释层；ROS2 action、行为树、规划器、普通控制软件和安全控制系统负责可验证执行。

1
2
3
4
5
6
7

ROS2 / 设备状态
  -> 确定性采集与状态模型
  -> 受限 Agent + 排障 Skill
  -> 证据、命令草案、影响范围
  -> 策略校验与必要的人工批准
  -> 确定性执行器
  -> 反馈、审计与独立安全系统

候选工具： ROS2 Jazzy、仿真器、行为树、Codex/Claude/Kimi 作为开发 Harness、Agents SDK/LangGraph 作为非实时编排层、只读 MCP 工具。

检查项： frame 与时间戳、机器人和环境状态、命令白名单、速度/区域/参数限制、仿真验证、操作者权限、动作反馈、急停和安全系统独立性。

最小试验： 做一个只连接仿真或只读 ROS2 状态的 Robot Operations Assistant。它可以解释告警和生成命令草案，但不能进入实时控制回路，也不能绕过规划器、联锁或安全控制器。

十、工具选型检查清单

在引入一个新框架、协议或多 Agent 方案前，逐项回答：

如果这些问题无法回答，多 Agent 通常只会把不确定性放大。

十一、常见失败模式

1. 伪并行

把强依赖步骤同时派发，最后仍要等待或返工；多个 Agent 同时编辑共享文件，合并成本超过节省的时间。

2. 角色很多，证据相同

同一模型、同一上下文、同一检索源，仅靠不同角色提示产生的意见高度相关。需要独立性时，更值得测试不同数据切片、工具、模型或确定性验证器，但目前“异构一定更好”仍缺少足够广泛的外部验证。

3. 上下文广播与摘要损失

全量广播导致 token 和权限膨胀；过度摘要又会丢失跨域约束。应保留原始 artifact 引用，并让验收节点能够回查证据。

4. 把模型错误变成系统副作用

没有参数校验、最小权限、dry-run、幂等和回滚时，一次错误工具调用就可能产生真实损失。MCP、A2A 或框架自带 guardrail 都不能自动建立业务安全边界。

5. 验证器与生成器共享错误规格

测试、LLM judge、critic 和仿真都可能验证了错误目标。高风险任务需要不同性质的证据，并保留人工或现场验证。

6. 把长对话当持久状态

上下文压缩会丢信息，外部环境会变化，权限也会过期。长任务必须将状态、事件和 artifact 外置。

7. 只看榜单，不看 Harness

模型分数可能来自不同工具、并发、预算、超时和重试策略。公开 benchmark 也不代表自己的代码库、业务流程或设备环境。

8. 过度自治

能调用工具不等于应该自动执行。尤其在工业和机器人系统中，LLM 不应进入实时控制或功能安全回路；人工批准也不能替代独立安全设计。

十二、我的试用顺序与持续观察渠道

结合现有 Web/full-stack、可视化和 AI 应用经验，最有价值的路线不是先搭一个庞大的多 Agent 平台，而是逐层证明工程能力。

第一阶段：把现有 coding Agent 用扎实

1. 用 Codex 作为主力 Harness，固定复现、测试、diff 和完成前验证流程。
2. 提炼 3 个高频 Skills：仓库诊断、前端视觉验收、技术调研来源审计。
3. 用同一组真实任务对照 Claude Code、KimiCode 或 OpenHands，记录成功率、返工、成本和工具轨迹。

第二阶段：增加 MCP 与少量 subagents

1. 先连接只读、低风险的数据源，例如文档、Git 仓库或模拟设备状态。
2. 用 2 到 4 个边界清晰的 subagents 做调研和分文件任务。
3. 建立单写者、任务契约、局部验收和最终集成测试。

第三阶段：做可恢复、可观测的 workflow

1. 用 LangGraph、ADK 或 Agents SDK 做“确定性流程 + Agent 节点”。
2. 引入 trace、数据集评测、checkpoint、幂等和故障注入。
3. 任务确实跨小时并需要强恢复时，再评估 Temporal 等持久工作流系统。

第四阶段：连接机器人与工业场景

最值得做的作品是一个只读或仿真的 AI Agent for Robot Operations：

• RAG 检索设备和机器人手册；
• 读取 ROS2 topic、TF、日志和状态；
• 使用 Skills 固化排障步骤；
• 必要时用 subagents 并行调查不同证据；
• 输出诊断、候选计划、命令草案和影响范围；
• 通过策略、仿真和人工审批后，才进入确定性执行层；
• 全程保留 trace、事件日志、拒绝原因和失败样本。

这个项目能同时证明系统集成、状态建模、实时数据 UI、Agent 工具设计、权限、安全边界和可观测性，比展示“创建了十个 Agent 角色”更有面试价值。

持续观察渠道

热点不应只靠社交媒体时间线。更稳定的信息雷达是：

• 官方产品与实现： OpenAI、Anthropic、Google、Moonshot、Microsoft、LangChain、LlamaIndex 的文档、release notes 和 GitHub；
• 协议与生态： MCP specification/Registry、A2A releases、Agent Skills specification、OpenTelemetry GenAI、OWASP Agentic Security；
• 公开评测： SWE-bench、Terminal-Bench、GAIA、BrowseComp、tau-bench、OSWorld、BFCL、MCP Atlas；
• 论文发现： arXiv、OpenReview、ACL Anthology、Hugging Face Papers，但新预印本只作为待验证线索；
• 实践者信号： Simon Willison、Latent Space、Matt Pocock、Superpowers、GitHub Trending、Hacker News，用于发现项目，不用于证明能力。

当前最值得持续验证的热点包括：K3 上是否出现可审计的 Agent Swarm 结果、Agent Skills 跨宿主兼容与供应链安全、A2A 1.0 与 MCP 的联合身份和 trace、原生 Harness 长程评测，以及强单 Agent 是否继续缩小同质多 Agent 的收益。

最终的判断标准始终相同：这个新东西是否改变了任务成功率、墙钟时间、成本、可恢复性、权限边界或验证质量。如果没有，就先把它留在观察清单，而不是加入生产架构。

主要参考资料

• Kimi K3 技术页
• Kimi Agent Swarm 技术博客
• Anthropic: How we built our multi-agent research system
• OpenAI Agents SDK: Multi-agent orchestration
• Claude Agent SDK: Subagents
• Google ADK: Workflow agents
• LangChain: Multi-agent
• Agent Skills Specification
• Model Context Protocol 2025-11-25
• A2A Protocol 1.0
• Towards a Science of Scaling Agent Systems
• Rethinking the Value of Multi-Agent Workflow
• When Do Multi-Agent Systems Help?
• WildClawBench
• OWASP Agentic Security Initiative

先说结论：基础没有消失，资深标准正在变化

Vibe Coding 降低了代码生成和 API 查询成本，但没有降低软件失败的代价。代码来得越快，审查边界、验证行为、定位故障和对结果负责越重要。

面试也不会整齐地从“八股”切换到 AI 协作。不同公司、岗位和面试官差异很大，标准化基础题仍是成本较低的筛选手段。对资深工程师，更常见的有效追问是：

• 为什么这个机制会产生当前现象？
• 结论在哪些条件下失效？
• 出现性能、并发或安全问题时怎样定位？
• 为什么选择这个方案，替代方案有什么代价？
• 这是生产经验、个人实验，还是根据原理作出的推断？
• AI 生成的代码看似能跑，怎样证明它可以进入系统？

因此复习策略不是“继续背完所有题”或“以后全靠 AI”，而是：

基础知识达到稳定过线水平，主要精力投入项目深挖、工程取舍、代码审查和验证证据。

对已有 11 年 Web/full-stack 经验、正在转向机器人系统软件和工业可视化的工程师，Vue、TypeScript 与 Node.js 是能力底座和差异化资产，不应成为占满全部学习时间的舒适区。

四层知识地图

哪些内容不值得长期投入

• 仅对某个旧版本成立、又能快速查询的执行顺序。
• 为展示技巧而构造的 TypeScript 递归类型谜题。
• Vue 2 的冷门 API 和已经退出主流项目的配置细节。
• 没有输入规模、环境和测量数据的性能结论。
• 脱离业务不变量的“最佳实践”列表。

这不表示这些内容永远不会被问到，而是它们不应挤占建立新职业证据的时间。遇到目标公司明确使用旧技术栈时，再进行定向准备。

三条专题主线

• Vue 进阶与响应式机制
• Vue Router
• TypeScript 类型系统
• Node.js 运行时
• Node.js Stream
• 具身智能与机器人系统软件路线

Vue：从 API 使用推进到更新模型

闭卷掌握响应式依赖、更新调度、computed/watch、key 和组件状态所有权。资深追问集中在高频更新、异步竞态、组件边界、大列表和 Web3D 对象如何避免无意义深代理。

目标不是背 Vue 源码函数名，而是能把一次状态写入追踪到 effect、scheduler、组件渲染和 DOM patch，并知道如何用 Vue Devtools 与浏览器 Performance 验证。

TypeScript：从“有类型”推进到可信边界

闭卷掌握结构化类型、unknown、判别联合、泛型关系、类型收窄和类型擦除。资深追问集中在公共 API 类型、方差、严格配置、声明文件、运行时输入和类型测试。

看到 AI 生成的 req.body as Command、response.json() as User 时，应立即追问：数据由谁验证、验证了哪些业务约束、失败如何处理、类型断言的证据是什么。

Node.js：从“异步非阻塞”推进到服务可靠性

闭卷掌握 V8/libuv/OS 的职责、事件循环稳定关系、I/O 与 CPU 任务边界、Stream 背压和错误传播。资深追问集中在尾延迟、事件循环阻塞、内存、超时取消、幂等、优雅停机和可观测性。

回答“Node.js 是单线程”时必须说明指的是 JavaScript 执行模型，而不是整个运行时只有一个线程。回答“流省内存”时必须说明缓冲、背压、并发和测量条件。

仍需单独准备的通用基础

这三条主线不能代替 JavaScript 语言、浏览器、网络、数据结构、数据库和系统设计。至少要能解释：

• 作用域、闭包、原型、Promise、微任务和错误传播。
• 浏览器渲染、布局绘制、事件、缓存、同源策略和常见安全边界。
• HTTP 方法语义、状态码、缓存、Cookie、认证与授权。
• 常见数据结构复杂度、并发限制、数据库索引和事务边界。

本轮文章不重复扩写这些主题，避免把索引变成无法复习的百科全书。

资深回答的六步结构

面对原理题或场景题，可以按六步组织，而不是一上来倾倒术语：

1. 结论：先直接回答问题，限定讨论条件。
2. 机制：说明关键数据流、控制流或类型关系。
3. 场景：把机制放回输入规模、并发、生命周期和业务目标。
4. 风险：指出错误、安全、性能、一致性和维护风险。
5. 替代方案：比较至少一个可行替代，并说明采用成本。
6. 验证与证据：给指标、profile、日志、测试、故障复盘或最小实验；没有生产经历时明确说是实验或推理。

“我会使用虚拟列表”只是方案名称；“先证明 DOM 数量和 layout/paint 是主要瓶颈，再决定虚拟列表窗口、滚动定位和可访问性策略”才体现资深判断。

跨栈场景：机器人与工业设备状态平台

问题：5,000 台设备高频更新，页面越来越卡，如何定位？

1. 先定义现象和目标

确认是消息延迟、交互卡顿、图表掉帧还是内存上涨；记录设备数、每台更新频率、单条消息大小、允许的 UI 新鲜度和报警延迟。没有这些条件，“优化 Vue”没有明确目标。

2. 把链路拆开测量

1
2
3
4
5
6
7

设备 / 模拟器
  -> Node.js 网关接收与解析
  -> 校验、聚合和消息分发
  -> WebSocket 客户端接收
  -> TypeScript 领域模型
  -> Vue 响应式状态与组件更新
  -> DOM / Canvas / WebGL 绘制

• 网关观察吞吐量、事件循环延迟、CPU、内存、队列深度和下游耗时。
• 浏览器观察消息处理耗时、Long Task、Vue 组件更新、DOM 数量、layout/paint 和帧率。
• 给消息加时间戳与 correlation ID，区分服务端积压、网络延迟和前端处理延迟。

3. 分离采集频率与展示频率

设备数据可以高频到达，但人眼和 DOM 不需要逐条刷新。把数据写入有界缓冲，按 100 至 250 ms 合并为 UI 快照。位姿展示可按设备保留最新值；报警、审计和控制结果不能用同样策略静默覆盖。

4. 守住类型和业务边界

WebSocket 数据先视为 unknown，验证消息类型、设备标识、时间戳、数值范围和版本，再转换为判别联合。TypeScript 接口不能代替这些检查。

5. 根据证据选择优化

• Vue 更新过多：批量发布快照、稳定 props、减少无意义深响应。
• DOM 数量过多：虚拟滚动、分组展开或只展示视口数据。
• Canvas/WebGL 绘制过重：控制渲染频率、复用对象、减少材质和 draw call。
• 主线程解析或计算过重：评估 Worker，并计入数据复制与调度成本。
• Node.js 消费跟不上：使用背压、有界并发、批处理、过载拒绝或分区扩展。

6. 给出验收指标

例如：在 5,000 台设备、每台每秒 10 条模拟遥测下，网关队列不持续增长；报警端到端 p99 小于约定阈值；页面交互无超过约定时长的 Long Task；稳态内存不随运行时间持续增长。具体阈值必须由业务和测试环境确定，不能把示例数字写成行业标准。

AI 生成代码审查练习

下面是一段表面上能工作的 AI 生成代码：

1
2
3
4
5

app.post('/devices/:id/command', async (req, res) => {
  const command = req.body as RobotCommand
  await sendCommand(req.params.id, command)
  res.json({ ok: true })
})

对于普通信息系统，这段代码已经缺少关键边界；对于机器人命令接口，直接上线更不可接受。

至少应发现的问题

1. as RobotCommand 只是类型断言，没有验证请求体结构、数值范围和命令版本。
2. 没有认证，也没有验证用户是否有权操作目标设备和当前空间。
3. 没有命令 allowlist、安全门控、设备状态检查和速度/工作空间限制。
4. 没有超时、取消和下游无响应时的行为定义。
5. 不清楚命令是否幂等；客户端重试可能造成第二次物理动作。
6. 没有 command ID、审计记录、操作者、批准信息和结果状态。
7. async 失败如何进入 Express 错误边界不明确，可能产生未处理 rejection 或错误响应。
8. 没有同一设备的并发、顺序、状态冲突和急停优先级策略。
9. HTTP 返回成功是否表示“已接收”“已规划”“已下发”还是“已执行完成”不明确。
10. 没有指标、trace 和安全事件日志，故障后无法还原链路。

修正方向

安全的接口不是增加几个 try/catch。它至少需要如下分层：

1
2
3
4
5
6
7
8

HTTP 边界
  -> 认证与资源授权
  -> 运行时 schema + 领域约束校验
  -> command ID / 幂等与设备级并发策略
  -> 工作流和人工批准
  -> 确定性安全控制与硬件互锁
  -> 下发、反馈、超时与状态查询
  -> 审计、指标、日志和告警

LLM 可以生成 handler、schema 初稿和测试样例，但它不能从不存在的需求中推导出可信权限、设备安全状态和物理约束。工程师必须补齐系统保证，并通过测试、仿真、故障注入和现场安全流程验证。

复习时应保留 AI 初稿、审查清单、修正后的代码与测试结果。这个过程比只展示“我用 AI 很快写完接口”更能证明新时代的工程能力。

面向当前职业转型的复习投入

以下比例针对“资深 Web/full-stack 工程师转向机器人系统软件、工业可视化和机器视觉”的当前阶段，不是所有读者的通用公式。

非求职冲刺期

这个配置的理由是：Web 基础已经是优势，继续投入会提高熟练度，却不能单独证明机器人系统能力。当前更稀缺的是把 Web3D、实时数据、设备状态、视觉和安全约束整合成可检查的作品。

投递前六至八周

把通用编程基础、编码练习和模拟面试提高到总时间的 30% 至 35%，依据目标职位描述补齐算法、网络、数据库和目标语言；同时继续维护项目 Demo 和证据，不要在面试前把作品开发完全停掉。

目标职位不同，权重也应调整：

• 工业数字孪生 / 机器人可视化：Vue/Three.js、实时数据、性能和坐标系权重更高。
• 机器人应用软件 / 系统集成：Node/Python/C++、设备协议、状态机、异常恢复权重更高。
• 机器视觉应用：OpenCV、标定、PnP、手眼标定、部署与光学条件权重更高。
• 纯 VLA、强化学习或控制算法岗位：当前仍是高跨度方向，不能用 Web 八股或 Agent 使用经验替代数学、训练和机器人实验能力。

分级自测题

A. 基础事实：能否直接回答

• Vue 的 computed 与 watch 分别解决什么问题？
• key 为什么首先是身份问题？
• TypeScript 的 unknown 和 any 有什么差别？
• 类型断言为什么不能校验 JSON？
• Node.js 所谓“单线程”的边界是什么？
• writable.write() 返回 false 表示什么？

B. 机制解释：能否继续追问两层

• 一次 Vue 响应式写入怎样到达 DOM 更新？
• 为什么从 reactive 直接解构可能丢失响应性？
• 判别联合怎样与 never 形成穷尽性检查？
• Promise 微任务、process.nextTick() 和事件循环是什么关系？
• pipeline() 比连续 pipe() 多承担了什么职责？

C. 工程决策：能否比较方案

• 高频设备数据应该每条更新 UI、定时批量发布还是只保留最新值？不同消息是否应使用相同策略？
• 什么情况下选择 Worker Threads，什么情况下选择独立进程或服务？
• 运行时校验使用手写 guard 还是 schema 库，怎样决定？
• Vue 页面缓存解决了什么，数据缓存又由谁负责？

D. 故障诊断：能否提供证据

• 页面运行两小时后内存持续增长，先看哪些指标和 profile？
• Node.js 平均延迟正常但 p99 很高，怎样分层定位？
• 路由参数切换后偶尔显示上一台设备，原因和修复是什么？
• AI 生成的重试逻辑导致设备动作重复，系统缺了哪些保证？

E. 项目证据：能否诚实说明边界

• 哪个项目证明你处理过实时数据、背压或高频渲染？
• 哪个故障是你亲自定位的，最初假设与最终原因是否一致？
• 哪些结论只有个人实验，还没有生产验证？
• 如果重新设计当前项目，会保留和改变什么？

可执行复习清单

机制卡

• [ ] 每张卡只覆盖一个机制，包含结论、运行过程、失败边界和最小实验。
• [ ] 能闭卷讲五分钟，并回答至少两层“为什么”和“什么时候失效”。
• [ ] 面试价值：通过基础筛选，避免资深候选人在核心原理上失速。

项目深挖包

• [ ] 准备架构图、关键接口、状态流、技术取舍、性能数据、测试和一次失败复盘。
• [ ] 能围绕同一项目接受二十分钟追问，而不是只复述产品功能。
• [ ] 面试价值：证明系统集成、正确性、可靠性、可观测性和交付能力。

AI 代码审查记录

• [ ] 保留原始提示和生成代码，不把修正后的结果伪装成人工一次写成。
• [ ] 标出接受、拒绝和需要实验确认的部分，并写明证据。
• [ ] 用测试、静态检查、基准或故障注入验证修正结果。
• [ ] 面试价值：证明会使用 AI，同时保留工程判断和结果责任。

职业转型证据

• [ ] 至少完成一个机器人调试台、数字孪生或视觉引导工作流的可运行作品。
• [ ] README 说明数据流、状态、坐标系、安全边界、故障处理和测试策略。
• [ ] 明确区分已有 Web 工程证据、正在形成的机器人软件能力和尚未证明的算法能力。
• [ ] 面试价值：把“11 年 Web 工程师想转机器人”变成“能够交付面向机器人的软件系统，并持续补强机器人深度”。

更新时间：2026-07

核心观点：Codebase Memory MCP 的本质不是 AI Coding Agent，而是 AI 的 Context Layer，用于帮助 AI 快速理解大型代码库。

一、项目简介

Codebase Memory MCP 是一个 MCP（Model Context Protocol）Server。它能够扫描整个代码仓库，构建代码知识图谱（Knowledge Graph），供 Claude Code、Codex、Cursor 等 AI Agent 查询。

它的目标不是代替 IDE，也不是直接生成代码，而是解决 AI Coding 中最重要的问题之一：

Context Retrieval（上下文获取）

对于大型项目，AI 往往需要花费大量时间：

• grep 搜索
• 查找文件
• 阅读大量无关源码
• 建立调用关系

Codebase Memory MCP 将这些工作提前完成，并保存成结构化知识图谱，使 AI 能够通过查询而不是全文扫描获取上下文。

二、它解决的问题

传统 AI Coding 工作方式大致是：

1
2
3
4
5
6
7
8
9
10
11

Agent
  ↓
grep
  ↓
read file
  ↓
grep
  ↓
read file
  ↓
...

大型仓库中，大量 Token 会消耗在“寻找代码”上。

使用 Codebase Memory MCP 后，工作方式变成：

1
2
3
4
5
6
7

Agent
  ↓
MCP Tool Call
  ↓
Knowledge Graph
  ↓
返回结构化关系

例如，Agent 想知道：

1

谁调用了 ProcessOrder()？

无需 grep 整个仓库，可以直接返回类似关系：

1
2
3
4
5
6
7

CheckoutController
        ↓
OrderApplication
        ↓
OrderService
        ↓
ProcessOrder()

三、核心思想

1. 它不是文档

很多人容易把它理解成：

1
2
3
4
5

Repository
    ↓
生成 Markdown
    ↓
AI 阅读 Markdown

这个理解不准确。

更接近真实的流程是：

1
2
3
4
5
6
7

AI Agent
    │
MCP Tool
    │
Knowledge Graph(SQLite)
    │
返回 JSON

例如返回：

1
2
3
4
5
6
7

{
  "symbol": "OrderService",
  "called_by": [
    "CheckoutController",
    "RetryWorker"
  ]
}

AI 获取的是结构化数据，而不是一大段索引文本。

2. 它更像数据库

Codebase Memory 保存的对象可能包括：

• Class
• Function
• Method
• Package
• Interface
• Module
• Route
• Resource

保存的关系可能包括：

• CALLS
• IMPORTS
• IMPLEMENTS
• HTTP_CALLS
• DATA_FLOW
• TESTS
• CONFIGURES

因此它更像：

1
2
3
4
5

Neo4j
+
SQLite
+
MCP Server

而不是：

1
2
3

Markdown
+
RAG

四、整体工作流程

第一步：扫描代码仓库

启动 Codebase Memory MCP 后，它会解析整个代码仓库：

1
2
3
4
5

Repository
    ↓
Tree-sitter
    ↓
AST

第二步：构建知识图谱

从 AST 和工程结构中抽取：

• Class
• Method
• Symbol
• Import
• Call Graph
• Dependency

最终形成：

1

Knowledge Graph(SQLite)

例如：

1
2
3
4
5
6
7

OrderController
      ↓
OrderService
      ↓
Redis
      ↓
MySQL

第三步：启动 MCP Server

MCP Server 向 AI 提供一系列 Tool，例如：

• find_symbol
• search_graph
• trace_call_path
• architecture_summary
• impact_analysis
• query_graph

Agent 不需要知道数据库位置，只需要调用 Tool。

第四步：AI Coding

例如用户提出需求：

1

增加订单取消功能

Agent 可能执行：

1
2
3
4
5
6
7
8
9
10
11

architecture_summary()
        ↓
find_symbol()
        ↓
trace_call_path()
        ↓
impact_analysis()
        ↓
read_file()
        ↓
edit_file()

注意：Graph 并不会替代阅读源码。

它只是帮助 Agent 找到应该阅读哪些源码。

第五步：修改后更新图谱

修改代码后，通常有两种方式更新图谱。

方式一：全量重新扫描

1
2
3

Repository
    ↓
重新 Index

优点是简单，缺点是大型项目速度较慢。

方式二：增量更新（推荐）

监听文件变化：

1
2
3
4
5

OrderService.java
    ↓
重新解析 AST
    ↓
更新 Graph

这样无需重新扫描整个仓库。

五、完整工作流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

Git Repository
        │
        ▼
Codebase Memory MCP
        │
        ▼
Tree-sitter AST
        │
        ▼
Knowledge Graph(SQLite)
        │
        ▼
MCP Server
        │
        ├──────────────┐
        ▼              ▼
Claude Code      Codex CLI
        │              │
        └────Tool Calls┘
               │
        查询调用关系
        查询依赖
        查询影响范围
               │
               ▼
         精准定位源码
               │
               ▼
           修改代码
               │
               ▼
     文件监听 / 增量更新 Graph

六、与传统 RAG 的区别

传统 RAG 流程：

1
2
3
4
5
6
7
8
9

Question
    ↓
Embedding
    ↓
Vector Search
    ↓
Chunk
    ↓
LLM

返回结果通常是文本片段。

Knowledge Graph 流程：

1
2
3
4
5
6
7
8
9

Question
    ↓
Graph Query
    ↓
Node
    ↓
Relation
    ↓
LLM

返回结果通常是结构化关系，例如：

1
2
3
4
5

Controller
    ↓
Service
    ↓
Repository

它返回的不是“几段可能相关的文本”，而是“实体和实体之间的关系”。

七、与 Sourcegraph 的区别

Sourcegraph 更像：

Google Search

Codebase Memory 更像：

Neo4j + MCP

八、优势

1. 快速理解大型仓库

减少：

• grep
• find
• read

提升 AI 获取上下文的效率。

2. Call Graph 查询

例如：

1

谁调用 ProcessOrder()？

无需扫描整个仓库。

3. Impact Analysis

例如：

1

修改 User 实体，会影响哪些模块？

Graph 可以帮助分析：

• Controller
• Service
• Repository
• Test
• API

4. Architecture Summary

快速生成：

• Layer
• Module
• Entry Point
• Hotspot

帮助 Agent 理解整体架构。

5. 支持大型仓库

相比反复读取源码，Graph 查询通常：

• 更快
• Token 更少
• 上下文更稳定

九、局限性

1. 不理解设计原因

Graph 能回答：

• What
• How

但不能直接回答：

• Why

例如：

1

为什么这里不用 RabbitMQ？

这类问题仍然需要：

• ADR
• Design Doc
• PR
• Issue
• Wiki

2. 动态语言分析有限

对于以下场景，静态图谱无法完全覆盖：

• Reflection
• 动态导入
• Plugin
• Runtime Dispatch

3. 不能替代 IDE

IDE 仍负责：

• Rename
• Refactor
• Diagnostics
• Go To Definition

Knowledge Graph 负责帮助 AI 理解项目。两者定位不同。

十、个人评价

技术价值

评分：★★★★★（9.5/10）

非常适合：

• 大型代码仓库
• AI Coding
• MCP Agent
• 企业级项目

成熟度

评分：★★★★☆（约 7-8/10）

项目发展很快，但仍需注意：

• API 可能持续变化
• 生态仍在快速演进
• 不同语言和框架的图谱质量会有差异

是否值得学习

值得。

真正值得学习的不是某个具体 API，而是背后的工程思想：

Context Engineering（上下文工程）

未来 AI Coding 的竞争力，不再只是更强的大模型，而是更完整的上下文系统：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Git
  ↓
Code Graph
  ↓
ADR
  ↓
Design Docs
  ↓
Issue
  ↓
Runtime Trace
  ↓
Knowledge Base
  ↓
AI Agent

LLM 只是最上层。真正决定 AI Agent 能力上限的是：

Context Layer（上下文层）

十一、我的总结

Codebase Memory MCP 并不是一个“更会写代码”的 AI Agent，而是 AI Agent 的“导航系统”。

它通过预先构建代码知识图谱，让 Agent 从“反复扫描源码”转变为“查询结构化上下文”，从而显著降低 Token 消耗，提高定位效率，并增强对大型代码仓库的理解能力。

对于 AI Native 开发而言，它代表了一种重要方向：

AI 的能力不仅取决于模型本身，更取决于提供给模型的上下文质量。

随着 AI 开发工具的发展，代码知识图谱很可能会与 ADR、设计文档、Issue、运行时监控、测试结果等信息融合，形成完整的 Context Layer，成为未来 AI Native 软件工程的重要基础设施。

后续可继续研究的问题

这篇笔记后续可以继续补充：

• Codebase Memory MCP 与 LSP 的边界
• Tree-sitter 对不同语言的解析能力差异
• 静态图谱与运行时 Trace 如何融合
• 如何把 ADR、Issue、PR、测试结果纳入 Context Layer
• 如何评估一个 Code Graph 对 AI Coding 的真实帮助

涉及项目最新状态、API、安装方式或版本兼容性时，需要重新查证。

背景

在机器人数字孪生、运动规划可视化、碰撞检测和工业仿真中，经常会遇到两类模型：

• 机器人结构模型：描述机器人有哪些连杆、关节、坐标系、关节轴、关节限位。
• CAD 几何模型：描述零件真实形状、曲面、实体、装配关系、STEP/IGES 等工程格式。

URDF 和 OCCT 分别对应这两类问题。

一句话理解：

URDF 负责“机器人是什么结构、怎么运动”；OCCT 负责“几何实体是什么形状、如何加工和转换”。

在机器人软件里，URDF 更偏机器人语义，OCCT 更偏 CAD 几何内核。

一、URDF 是什么

URDF 全称是 Unified Robot Description Format，是 ROS 生态中常用的机器人描述格式，本质是一个 XML 文件。

它主要描述：

• 机器人由哪些 link 组成。
• link 之间通过哪些 joint 连接。
• 每个关节的类型、轴向、限位和初始变换。
• 每个 link 的视觉模型、碰撞模型和惯性参数。

URDF 不是三维建模软件，也不是运动控制算法。它更像机器人系统中的“结构配置文件”。

二、URDF 的核心元素

1. robot

robot 是根节点，表示一个机器人模型。

1
2
3