Triangulate 🔺 — 三视角共识决策引擎 V1.0.0

将拜占庭多视角共识架构泛化为通用思考/执行框架。

架构概览

┌──────────────────────────────────────────────┐
│  Layer 6: renderer.py       输出渲染层        │
│  Layer 5: orchestrator.py   编排层 + 状态机    │
│           unit_of_work.py   事务边界 (UoW)    │
│           context_bus.py    上下文总线         │
│           middleware.py     横切关注点         │
│           pipeline_steps.py 阶段函数           │
│  Layer 4: consensus.py      共识引擎          │
│  Layer 3: dispatcher.py     执行调度层         │
│  Layer 2: decomposer.py     管理层拆解         │
│  Layer 1: adapters/         输入适配层         │
├──────────────────────────────────────────────┤
│  schemas.py         数据契约 (Pydantic v2)    │
│  pipeline.py        管线上下文 (不可变)        │
│  state_machine.py   显式状态机 (10 状态)      │
│  circuit_breaker.py 可执行熔断器              │
│  idempotency.py     幂等性保证               │
│  side_effects.py    副作用收集器              │
│  saga.py            事务回滚 (外部集成)        │
│  timeout_manager.py 统一超时管理              │
│  pipeline_guard.py  阶段间契约守卫            │
└──────────────────────────────────────────────┘

模块索引

快速使用

from orchestrator import TriangulateOrchestrator
from adapters import InputAdapter
from decomposer import TaskDecomposer
from dispatcher import ExecutionDispatcher
from consensus import ConsensusEngine
from renderer import ReportRenderer

# 完整配置（Saga 回滚 + 幂等性缓存 + 熔断保护）
orchestrator = TriangulateOrchestrator(
    input_adapter=InputAdapter(strict_mode=True),
    decomposer=TaskDecomposer(enforce_parallel=True),
    dispatcher=ExecutionDispatcher(global_timeout=300),
    consensus_engine=ConsensusEngine(),
    renderer=ReportRenderer(),
    enable_saga=True,              # Saga 事务回滚
    enable_idempotency=True,       # SHA-256 幂等性缓存
    enable_circuit_breaker=True,   # 6 条可执行熔断规则
)

# 支持多种输入格式
report = orchestrator.run("分析数据库选型")
report = orchestrator.run({"task_description": "分析数据库选型", "importance": 4})

# 查看工作流进度（含 Saga/熔断/幂等性状态）
progress = orchestrator.get_progress()
print(f"Phase: {progress['phase']}, Degraded: {progress['degraded']}")
print(f"Breaker: {progress['breaker']}, Idempotency: {progress['idempotency']}")

# Markdown 渲染
renderer = ReportRenderer()
print(renderer.render_markdown(report))

# 多次 run 之间需 reset
orchestrator.reset()
report2 = orchestrator.run("另一个任务")

# 部署时注册 sessions_spawn 创建的 session ID（Saga 回滚用）
orchestrator._auto_register_session("session-abc", "strategy")
orchestrator._auto_register_session("session-def", "execution")

核心思想

一个视角有盲区，两个视角有偏差，三个独立视角 ≥ 2/3 一致则可信。

决策层 (3个独立agent/视角)
    ↑ 任务指令           ↓ 共识结果
管理层 (任务拆解 + 调度 + 报告汇总)
    ↑ 可并行子任务       ↓ 执行报告
执行层 (N个并发 sessions_spawn)

三层各司其职：

• 决策层 — 定方向、审结果（什么值得做，什么结论可信）
• 管理层 — 拆任务、管调度（怎么做，谁做什么，怎么收拢）
• 执行层 — 干活出结果（并行跑子任务）

何时触发（仅显式调用）

本技能 不自动激活。仅在用户明确请求三视角分析时触发。

以下情况绝不触发：

• 通用"帮我决策"或"哪个方案好" → 正常回答，不 spawn 子代理
• "从多个角度分析" → 常规多角度思考，不使用共识引擎
• 任何未明确提及 triangulate/三角验证/三视角的请求

配置档次

选择： 简单问答→不触发 / 2-5步→轻量或平衡 / ≥5步或重要性≥4→全量 / 需复用已有技能→子技能调度

子代理 Prompt 标准：所有 sessions_spawn 的 task 参数必须遵循 AGENTS.md 五段模板（角色定义 → 上下文 → 任务 → 约束 → 输出格式）。以下示例已对齐此标准。

工作流（5 阶段）

阶段一：决策层定策略

1.1 决策者A（主视角）

任务类型：[参考视角模板库]
任务描述：[一句话]
重要性：[1-5]
推荐配置：[全量/平衡/轻量]
需管理层拆任务：[是/否]
需执行层：[是/否]
推荐视角模板：[模板编码，如T-01，或空着用默认]

1.2 决策者B（批判/安全视角）

task: "[System: 角色定义] 你是一位严格的批判型分析者，专门评估方案的盲区和风险，不受现有共识影响。

[Context: 上下文状态] 当前正在对以下任务进行三视角共识分析的第一轮评估：
[任务描述]

[Task: 具体任务] 基于你的视角，独立评估以下方面并输出统一 DecisionResult JSON：
① 值得投入资源吗？（1-5分，说明理由）→ importance
② 推荐什么配置档次？→ config: "全量"/"平衡"/"轻量"/"子技能调度"
③ 有什么被忽略的风险或盲区？→ risks[]
④ 你推荐从什么角度切入？→ top_findings[]

[Constraints: 约束条件] 你的评估必须严格、批判、与主视角独立。不要受已有共识影响。严格遵循自己的判断。

[Output: 输出格式] 输出以下统一 JSON 格式：
{
  \"agent_id\": \"B\",
  \"importance\": <1-5>,
  \"config\": \"全量\",
  \"reasoning\": \"<≥10字的推理说明>\",
  \"risks\": [\"<风险1>\", \"<风险2>\"],
  \"top_findings\": [\"<结论1>\", \"<结论2>\"],
  \"confidence\": <0.0-1.0>
}"

1.3 决策者C（实用视角）

task: "[System: 角色定义] 你是一位务实型分析者，专注于评估方案的实际产出价值和落地可行性。

[Context: 上下文状态] 当前正在对以下任务进行三视角共识分析的第一轮评估：
[任务描述]

[Task: 具体任务] 基于你的视角，独立评估并输出统一 DecisionResult JSON：
① 实际产出价值多大？（1-5分）→ importance
② 推荐什么配置档次？→ config
③ 最小可行方案是什么？→ top_findings[]
④ 如果只能产出核心结论 → top_findings[]

[Constraints: 约束条件] 你的评估必须务实、实事求是，关注最小可行方案。不受其他视角影响。

[Output: 输出格式] 输出以下统一 JSON 格式：
{
  \"agent_id\": \"C\",
  \"importance\": <1-5>,
  \"config\": \"平衡\",
  \"reasoning\": \"<≥10字的推理说明>\",
  \"risks\": [],
  \"top_findings\": [\"<结论1>\", \"<结论2>\"],
  \"confidence\": <0.0-1.0>
}"

1.4 共识检查

由 consensus.py 执行：

✅ ≥2/3 维度一致 → CONSENSUS → 采纳
⚠️ 分歧 → DIVERGENCE → 自动重试（最多 2 轮）
❌ 连续 2 轮分歧 → FALLBACK_TO_USER → 转交用户全权裁定

阶段二：管理层拆解（可选）

由 decomposer.py 执行，自动拆解为 TaskDAG + 循环依赖检测。

task: "[System: 角色定义] 你是一个管理型AI agent，负责将复杂任务拆解为可并行执行的子任务，并分配资源和约束。

[Context: 上下文状态] 以下是决策层第一阶段共识结果，需要拆解为可执行的工作包：
主任务：[阶段一共识结果]
配置档次：[全量/平衡/轻量]

[Task: 具体任务] 将主任务拆解为 2-5 个可并行执行的子任务。
每个子任务需包含：
- 子任务编号
- 明确的子目标（可验证的完成标准）
- 输出格式要求
- 最大 token 预算

[Constraints: 约束条件]
- 子任务之间不能有数据依赖（必须可并行）
- 每个子任务规模相当
- 总量不超过配置档次的上限

[Output: 输出格式] 为每个子任务输出一条完整的 `sessions_spawn` 指令，每条指令遵循五段模板"

阶段三：执行层并行

由 dispatcher.py 执行，支持并发 + 退避重试 + 熔断。

每个执行器一个 sessions_spawn（context: isolated），用 sessions_yield 等待。

管理层汇总：去重、归类、标注质量评分（1-5）。

阶段四：决策层审阅

决策者B、C独立审阅汇总报告，各自筛选最佳结论（≤3条）。 共识检查后进入输出。

阶段五：输出

由 renderer.py 渲染（保留原始格式）：

═══════════════════════════════════════
🔺 Triangulate 分析报告
───────────────────────────────────────

📌 任务：[描述]
⚙️ 配置：[档次] | 执行器×N | 决策者×N
🧩 视角：[模板编码]

📊 核心结论
① [来自共识]
② [来自共识]
③ [来自共识]

🔍 分歧点（如有）
- [说明]

⚠️ 不确定性
- [局限与不确定因素]
═══════════════════════════════════════

预置视角模板库

按领域分类，失效用。优先在库中查找，找不到则用「视角自创建」规则实时生成。

🏗️ 技术 & 工程

🔬 分析 & 决策

📈 商业 & 战略

👥 人际 & 组织

🎯 个人 & 成长

🏅 体育 & 娱乐

🎨 创作 & 内容

⚡ 高频专业实践（新增）

子技能调度模式（Skill Dispatch）

源自 codex 的「主技能派发子技能」模式。适用于需要多个已有技能协作完成的任务。

何时使用

• 任务可以拆分为多个独立子技能能分别处理的子任务
• 已有现成的 SKILL.md 可以复用
• 替代「把子任务写死在主技能里」——让子技能自己负责自己的逻辑

工作流

管理层识别子任务 → 映射到外部 SKILL.md
    ↓
执行层为每个子技能创建一个 sessions_spawn
    ↓
每个子任务 task 中传入 skill_path 参数
    ↓
子 agent 自行读取 SKILL.md 执行
    ↓
管理层合并所有子技能输出

task 模板

task: "你是 Claw 的子任务执行器。
请基于以下技能路径完成任务：

技能路径: skills/<子技能目录>/SKILL.md
任务范围: [具体说明这个子技能的 scope]

步骤：
1. 读取 SKILL.md 了解完整流程
2. 按技能定义执行任务
3. 输出结果（格式不限，但必须完整）

限制：
- 不修改技能文件
- 不跨越 scope 边界
- 完成后返回摘要"

与决策层的关系

子技能调度模式可跳过或结合决策层：

• 跳过决策层：子任务明确，直接分派
• 结合决策层：先让决策层定「哪些子技能参与」，再让执行层调度

退出条件

• 所有子技能返回结果 → 进入管理层合并
• 任一子技能失败 → 记录失败原因，其他照常合并
• 用户中途打断 → 立即返回已有结果

视角自创建

如果库中找不到合适模板，用以下规则即时生成：

给任意任务X生成三个视角：
A = 主流视角（最直接的、最可能的、主流的答案）
B = 批判视角（质疑前提、考虑失败、找反例）
C = 第三维视角（跨域类比、长期影响、非常规角度）

示例 — 评估第三方支付风险：

自创建视角：
A — 技术：集成复杂度、API质量、文档、SLA
B — 业务：费用结构、用户影响、迁移成本
C — 运维：监控告警、故障恢复、技术支持

也可在预置模板上微调 — 如把T-01的A视角改成「可维护性」，就适配了不同侧重的选型。

视角自动匹配速查

多个匹配时按「具体>通用」选。都不匹配→视角自创建。

与书内化技能的关系

book-full-dimension-internalization 是 triangulate 的特化实例：

Triangulate
  └─ book-full-dimension-internalization
       ├ 视角固定为知识学习的三个维度
       ├ 输出格式固定为知识文件
       └ 添加熔断：预算上限 + 余额告警

熔断条件

由 circuit_breaker.py 提供可执行实现：

限制

1. 需 sessions_spawn / sessions_yield 工具可用，否则退化为单线程
2. 全量配置消耗 tokens 较多 — 按档次选择
3. 适合慢思考类任务，不适合快反应类任务

安全：高危模板需要显式确认

以下模板涉及爬虫/渗透/逆向等高风险领域，不会自动推荐：

这些模板仅提供分析框架，不执行任何实际操作。具体执行由用户自行完成。

触发短语

仅以下关键词触发，其余请求不激活本技能：

幂等性保证

通过 idempotency.py 提供：

from idempotency import IdempotencyGuard

guard = IdempotencyGuard(ttl_seconds=1800)
result, is_cache_hit = guard.execute_or_cache(
    user_input,
    executor_fn=lambda: orchestrator.run(user_input),
)
# is_cache_hit=True → 直接返回缓存，不重复执行副作用

状态机状态

Agent 执行指令

职责边界

标准工作流

1. 用户提出任务
2. sessions_spawn 3个决策者agent（视角A/B/C）
3. 收集结果 -> 共识检查 >=2/3
4. sessions_spawn 管理层agent拆解任务
5. sessions_spawn N个执行器agent并行执行
6. 收集执行结果 -> 报告聚合
7. 交付用户

错误恢复

• 决策不一致（<2/3）-> 列出分歧点请用户裁决
• 子agent失败 -> 重新spawn该子任务（最多2次）
• 执行器超时 -> 跳过该子任务，标注未完成