x-rw-graphengineering-g3 — 确定性图编排引擎

定位: LoopEngine（想清楚）→ GraphEngine（可靠做）

> 📋 1 分钟判断：G3 适合你吗？
>
> 适合用 G3 的场景（G3 是正确答案）：
> - ✅ 你有结构化的多步骤任务（如内容流水线、数据处理管道、审批流），想可靠自动执行
> - ✅ 你需要人工审批干预点 + 自动恢复执行（HITL 中断/恢复）
> - ✅ 你需要条件路由、执行验证、运行时监控、目标漂移检测等治理能力
>
> 不适合用 G3 的场景（G3 不是正确答案）：
> - ❌ 你只需要一个简单的单步脚本/函数调用 → 用 bash/Python 直接写
> - ❌ 你只需要自动化的 CI/CD 流水线 → 用 Jenkins/GitHub Actions
> - ❌ 你只需要 LLM 对话/单次 Prompt 调用 → 用 TRAE 直接对话
>
> 30 秒决策：
> > 你有一个任务想自动化？ > ├─ 是单步骤？ → 直接写脚本，不需要 G3 > └─ 是多步骤流程？ > ├─ 步骤间简单串行，不需人工审批？ → 考虑 bash/Makefile > └─ 需要条件分支/审批点/监控/重试？ > ├─ 需要人工审批介入？ → ✅ G3 hitl_gate 节点 > ├─ 需要条件路由？ → ✅ G3 llm_gate 节点 > └─ 需要执行验证/目标对齐？ → ✅ G3 validator/monitor/grounding >

> <small>🟢 新手 · 🟡 普通用户 · 🔴 进阶用户</small>

用户旅程导航

🟢 架构速览（新手必读）

用户请求 → Plan (JSON) → StateGraph 编译 → 节点拓扑执行 → 结果输出
                              │
        ┌─────────────────────┼─────────────────────┐
        ↓                     ↓                     ↓
    deterministic         llm_narrow           tool_call
    hitl_gate ← HITL 中断 → resume 恢复
    llm_gate  ← LLM 动态路由 → __next 输出
    validator ← 断言操作符验证 → pass/fail/warn
    monitor   ← 声明式订阅 → log/veto/degrade
    grounding ← 目标漂移检测 → block/escalate

核心模块

🟢 快速开始

前置: Python 3.8+，零外部依赖。

# 1. 验证 Plan 格式
python3 cli/cli_bridge.py validate --input demo/blog-pipeline.json
# 输出: "验证通过！共 7 个任务"

# 2. 执行计划（遇 HITL 自动中断）
python3 cli/cli_bridge.py run --input demo/blog-pipeline.json --thread-id sess-001
# 输出: "执行完成，状态: interrupted"（HITL 节点处中断）

# 3. Shell Tool 模式（JSON 字符串直接传入，无需文件）
python3 cli/execute.py run '{"version":"1.0","plan":{"title":"测试","tasks":[{"id":"t1","title":"任务","type":"deterministic","risk_level":"low","config":{"operation":"data_transform","input_data":{"x":1}}}]}}'
# 输出: {"status":"completed"}

# 4. 从 HITL 中断点恢复
python3 cli/execute.py resume <cp_id> '{"decision":"approve"}' '{"plan":{...}}'

# 5. 查看执行状态
python3 cli/cli_bridge.py status --thread-id sess-001

> 验证点: 执行 python3 cli/cli_bridge.py validate --input demo/blog-pipeline.json，应返回 0 并输出"验证通过"。

🟢 Plan JSON 格式

{
  "version": "1.0",
  "source": "loopengine-v10.3",
  "session_id": "sess-001",
  "goal_anchor": "sha256:...",
  "trust_level": 3,
  "hitl_policy": {
    "risk_high": "interrupt_before",
    "risk_medium": "auto_confirm",
    "risk_low": "auto_confirm"
  },
  "plan": {
    "title": "计划标题",
    "tasks": [
      {"id": "t1", "title": "任务1", "type": "deterministic", "risk_level": "low"}
    ],
    "transitions": [
      {"from": "t1", "to": "t2"},
      {"from": "t2", "type": "conditional", "condition": "context.get('status')=='ok'", "routes": {"True":"t3","False":"__end__"}}
    ],
    "dependencies": [],
    "conditional_edges": [],
    "parallel_groups": []
  }
}

完整 Schema 见 docs/specs/graph_input_schema.json。

🟢 端到端示例（完整 7 节点流水线）

参考 demo/blog-pipeline.json:

draft → word-count → screenshot → review(HITL) ──(条件: approve?)──→ format-check → publish → notify

# 执行到 HITL 中断点
python3 cli/cli_bridge.py run --input demo/blog-pipeline.json --thread-id blog-test --debug

# 审批通过后恢复
python3 cli/cli_bridge.py resume --thread-id blog-test --checkpoint-id <id> --plan demo/blog-pipeline.json --value '{"decision":"approve"}'

🟡 节点类型

决策树

节点执行的目标是什么？
├── 确定性逻辑操作 → deterministic  [IMPLEMENTED]
│   ├── 写文件 → operation: file_write
│   ├── 模板渲染 → operation: template_render
│   └── 数据转换 → operation: data_transform
├── 单次 LLM 调用 → llm_narrow  [IMPLEMENTED]
├── 执行命令/脚本 → tool_call  [IMPLEMENTED]
├── 人工审批中断 → hitl_gate  [IMPLEMENTED]
├── LLM 动态路由 → llm_gate  [IMPLEMENTED]
├── 上游输出验证 → validator  [IMPLEMENTED]
├── 实时监控告警 → monitor  [IMPLEMENTED]
├── 目标漂移检测 → grounding  [IMPLEMENTED]
├── 多路径汇合同步 → barrier  [IMPLEMENTED]
└── 多源数据合并 → merge  [IMPLEMENTED]

> 节点状态说明：[IMPLEMENTED] = 有独立实现代码，可直接使用。[ALIAS: X] = 别名，运行时等价于节点 X，语义不同但行为一致。[PLACEHOLDER] = 已注册名称，行为映射至 core 节点，已有 tool_call+curl 替代方案。

deterministic — 确定性节点

llm_narrow — 窄 LLM 节点

tool_call — 工具调用节点

hitl_gate — 人工审批节点

llm_gate — LLM 路由节点（v0.3.0）

输出: {"__next": selected_id, "__reason": reason, "output": {...}}

validator — 验证节点（v0.3.0）

8 种断言操作符: exists type eq ne in range len_gt not_empty

monitor — 监控节点（v0.3.0）

grounding — 接地节点（v0.3.0）

🟡 边类型

条件边语法:

condition = "context.get('resume_value', {}).get('decision') == 'approve'"
routes = {"True": "next_node", "False": "__end__"}

🟡 CLI 命令

> v0.5.1 兼容说明：
> - O2 v10.4.1 输入：transitions 的 type: "SEQUENCE"/"data_flow" 被精确映射为对应 EdgeType；config.retry_max / config.timeout_seconds 自动映射到 max_retries / timeout_seconds；upstream_anchor 透传到输出（锚定链）。
> - C0 V3.x 输出契约：run --report <path> 生成 execution_report.json，含 status + node_results(字典) + summary{total_nodes, completed, failed}，被 C0 check_output.check_graph() 与 sync_metrics_from_report() 消费。
> - 向后兼容：无 type 的旧版 transitions 仍按 NORMAL 处理；原 outputs/ 输出保持不变。

Shell Tool 模式

# 从 JSON 字符串直接执行
python3 cli/execute.py run '{"version":"1.0","plan":{...}}'

# 从文件执行
python3 cli/execute.py run plan.json

# 恢复执行（返回 JSON）
python3 cli/execute.py resume <cp_id> '{"decision":"approve"}' '{"plan":{...}}'

# 调试模式（显示每步执行详情）
python3 cli/execute.py run '...' --debug

返回格式: {"status":"completed"} / {"status":"interrupted","checkpoint_id":"..."} / {"status":"error","error":"..."}

🟡 检查点与缓存

节点结果缓存（v0.2.0）

同 thread 下重复运行，deterministic/llm_narrow 节点自动跳过已完成的节点。

LLM 响应缓存（v0.2.0）

temperature=0 时自动缓存 (model + system_prompt + user_prompt + output_schema) → 响应。同入参不重复调用。

# 清除缓存
python3 -c "from graphengine import SqliteCheckpointer; SqliteCheckpointer().clear_llm_cache()"

检查点自动保存

每个节点执行后自动保存到 SQLite（默认 ./checkpoints/graphengine.db）。支持崩溃恢复。

审计日志

事件类型: node_enter / node_exit / llm_route / hitl_decision / node_interrupt / node_cache_hit

🟡 治理体系

HITL 审批

执行 → hitl_gate → 保存检查点 → 抛出 HITLInterrupt → 中断 → resume({'decision':'approve'}) → 继续
                                                                         ↓ reject → 条件路由到拒绝分支

GovernancePermissions

from graphengine import GovernancePermissions
GovernancePermissions.from_trust_level(1)  # {只读, max_tokens=5K, max_time=10s}
GovernancePermissions.from_trust_level(2)  # {可写, max_tokens=20K, max_time=60s}
GovernancePermissions.from_trust_level(3)  # {全权限, max_tokens=100K, max_time=300s}

GraphState 治理字段

• governance_log — 治理事件链，通过 add_governance_event() 记录
• metrics — 运行时指标，由 monitor 节点采集
• drift_score — 目标漂移分数 (0.0~1.0)，由 grounding 节点更新

🔴 自愈回环（v0.5.0 Self-Healing）

根因分类器

三层失败分类：
- EXEC — 执行级失败（超时、环境、API错误）→ 同层重试
- PLAN — 规划级失败（依赖缺失、条件模糊）→ 触发 Loop 重规划
- STRAT — 战略级失败（假设违背、架构冲突）→ 升级 TOGAF + 人工介入

from graphengine.healing import classify_failure
result = classify_failure(TimeoutError(), "n1", "tool_call", {}, {}, 0)
# -> FailureClassification(category="EXEC", confidence=0.9)

全局熔断

from graphengine.healing import CircuitBreaker
cb = CircuitBreaker()
cb.record_node_retry("n1", 6)
tripped, reason = cb.check()  # -> (True, "CB1: 节点 n1 重试 6 次 > 阈值 5")

5 条熔断规则：
| ID | 条件 | 默认阈值 |
|:--:|------|:--------:|
| CB1 | 单节点重试次数 | > 5 次 |
| CB2 | 整图重规划次数 | > 2 次 |
| CB3 | 连续 STRAT 分类 | >= 3 个 |
| CB4 | 执行时长占比 | > 预估 300% |
| CB5 | Token 成本占比 | > 预算 150% |

Auto-Pause 条件触发器

编译期自动注入 hitl_gate 节点的 5 条规则：
| ID | 条件 | 默认 |
|:--:|------|:----:|
| R1 | 外部 API 调用 | 启用 |
| R2 | 数据写入/删除 | 启用 |
| R3 | LLM 调用成本 > 阈值 | 启用 |
| R4 | 节点置信度 < 0.6 | 关闭（POC） |
| R5 | 高风险 + interrupt_before | 启用 |

PLAN 级自愈回调

from graphengine.healing import HealingManager

def my_callback(node_id, subgraph, classification, snapshot):
    return {"plan": {"tasks": [...]}}  # 新 Plan JSON

hm = HealingManager(plan_callback=my_callback)
new_plan = hm.request_replan("n3", "依赖缺失", state)

G3→O2→G3 辩论闭环（v0.5.1 / O2 v10.4.2 debate-poll 对齐）

当 PLAN 级失败且无自定义回调时，G3 默认启用 O2 辩论桥（adapters/o2_debate_bridge.py），协议与 O2 debate-poll / C0 loop_control.py 完全对齐：

G3 运行时 PLAN 失败
  → debate/debate_req_{session_id}.json（O2DebateRequest）
  → O2 决策（loop_engine_helper.py debate-poll，确定性规则无需 LLM）
  → debate/debate_resp_{session_id}.json（O2DebateResponse, status=debated + new_plan）
  → G3 invoke 时自动加载 → 消费 new_plan 重建图 → 按新 Plan 继续执行

协议字段（O2DebateRequest → O2DebateResponse）：
| 请求 | 响应 |
|------|------|
| session_id / failed_node_id / category / reason / confidence / error_type / retry_count | session_id / status / new_plan / verdict / recommended_action / confidence / debated_by / response_ts |

恢复条件：status == "debated" 且 new_plan 为非空合法 G3 Plan JSON（invoke() 消费时经 build_graph_from_plan 重建图）。

辩论目录约定（跨目录运行关键）：
- G3 默认写 <项目根>/debate/；O2/C0 默认从 ~/.trae-cn/skills/x-rw-graphengineering-g3/debate/ 扫描
- 三者均支持 G3_DEBATE_DIR 环境变量覆盖 —— 跨目录运行时（G3 以工作区项目方式运行 + O2/C0 以技能方式运行）必须显式设置 G3_DEBATE_DIR 为同一目录，否则 G3 写入的请求 O2 扫描不到
- C0 loop_control v2.2 支持 enter-review → sync_from_g3() → review_request.json 与 exit-review → write_g3_response() 双写 decision.json + debate_resp_{sid}.json，实现 G3→C0→O2→G3 全自动闭环

🟡 异常体系

GraphEngineError
├── CompileError
│   ├── GraphCycleError — 循环依赖
│   ├── NodeNotFoundError — 节点不存在
│   └── EdgeValidationError — 边验证错误
├── ExecutionError
│   ├── ConditionEvalError — 条件表达式错误
│   └── MaxRetriesExceededError — 重试耗尽
├── CheckpointError
└── TimeoutError

🔴 Python API

from graphengine import StateGraph, GraphState, SqliteCheckpointer, get_node_func

g = StateGraph(name="工作流")
g.add_node("n1", "节点1", "deterministic",
           func=get_node_func("deterministic"),
           config={"operation": "data_transform", "input_data": {"x": 1}})
g.set_entry_point("n1")
g.set_checkpointer(SqliteCheckpointer())
result = g.compile().invoke(GraphState(thread_id="sess-001", goal="测试"))
print(result.status)

输入规范化

from adapters.normalize import normalize_plan
plan = normalize_plan({"title":"计划","tasks":[{"type":"确定性","risk_level":"低"}]})
# → type 自动转为 "deterministic", risk_level 自动转为 "low"

🔴 与 LoopEngine 集成

LoopEngine 规划完成后输出 ─── 执行就绪 ─── JSON，直接喂给 GraphEngine：

python3 cli/execute.py run '<graph_plan_json>'

集成指南: docs/loop-graph-integration-guide.md

🟡 测试

python3 tests/unit/test_state.py              # 状态管理（5 项）
python3 tests/unit/test_llm_gate.py           # llm_gate+validator（10 项）
python3 tests/e2e/test_hitl_resume.py          # HITL 中断恢复（2 项）
python3 tests/e2e/test_loop_graph_integration.py  # LoopEngine 集成（4 项）
python3 -m pytest tests/ -v                    # 全量

🔴 深入阅读

🟢 版本

v0.5.1 C0/O2 适配 — 测试 100% 通过，零外部依赖。

关键演进: v0.1.0 MVP(4节点/2边) → v0.2.0 Caching(节点+LLM缓存) → v0.3.0 Extensions(llm_gate/validator/monitor/grounding+治理边) → v0.4.0 Governance(子图/权限/对抗指标/13节点/8边型) → v0.5.0 Self-Healing(自愈回环/三级根因/全局熔断/PLAN回调) → v0.5.1 C0/O2 适配(execution_report 契约 + SEQUENCE 边映射 + retry 映射 + 锚定链)