分析日期: 2026-06-02
分析范围: LangGraph (独立框架) vs 祖龙 ZULONG (具身机器人系统)
聚焦维度: 图记忆、任务编排、状态管理、工具调用、持久化、认知架构
LangGraph 与祖龙是两个独立设计的图驱动智能体系统,在概念层面存在诸多"趋同进化"的相似点,但在架构深度、领域适配和设计哲学上有本质差异。祖龙并非基于 LangGraph 构建,而是在 LangGraph 流行之前就独立设计了自己完整的图记忆+图编排体系。
| 维度 | LangGraph | 祖龙 (ZULONG) |
|---|---|---|
| 定位 | 通用 Agent 编排框架 (SDK) | 具身机器人四层操作系统 |
| 图记忆 | 无持久跨运行记忆,仅有 checkpointer (单运行内保存/恢复状态) | MemoryGraph — NetworkX/igraph 异构图,14种节点,14种边,BFS扩散+赫布学习+艾宾浩斯衰减+全生命周期 |
| 图编排 | StateGraph — 有向图,节点+边+条件边,单图执行 | TaskGraph (树+DAG) + FC Loop (while 循环) 双层架构 |
| 持久化 | SQLite checkpointer (按 thread_id 存状态快照) | LMDB + FAISS + 分片 Hybrid 存储 + 原子持久化 |
| 工具调用 | ToolNode + tool_choice 绑定 | L1-B 工具预判 -> L2 单次决策 -> FC 循环执行 |
| 人机协作 | interrupt() + Command(resume=...) |
EventBus 紧急中断 + L1-B Gatekeeper + 多模式审批 |
| 流式输出 | stream() / astream_events() |
OutputGateway 流式代理 + 毫秒级 STOP/PAUSE/RESUME |
LangGraph 由 LangChain 团队开发,是一个低层级 Agent 编排框架。其核心思想是将 Agent 逻辑建模为有向图 (StateGraph),其中:
- 节点 (Node): 计算步骤 — 可以是 LLM 调用、工具调用、或任意 Python 函数
- 边 (Edge): 定义节点间的流转关系
- 条件边 (Conditional Edge): 基于当前 State 动态决定下一个节点
- State: 贯穿整个执行的带类型状态对象,节点通过返回部分 State 来增量更新
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
# 1. 定义 State schema
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # add_messages = reducer
# 2. 构建图
builder = StateGraph(AgentState)
builder.add_node("agent", call_model) # LLM 节点
builder.add_node("tools", ToolNode(tools)) # 工具节点
builder.add_edge(START, "agent") # start -> agent
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent") # tools -> agent (循环)
# 3. 编译
graph = builder.compile(checkpointer=MemorySaver())State 是 LangGraph 的核心抽象,有以下特性:
- 有类型: 通过 Pydantic BaseModel 或 TypedDict 定义 schema
- Reducers: 每个字段可以指定合并策略。例如
add_messages将新消息追加到消息列表(而非覆盖),实现了对话历史的自然累积 - 自动合并: 每个节点返回部分 State,框架自动按 reducer 规则合并到全局 State
- 不可变更新: State 在每次 Superstep 后生成新版本,旧版本在 checkpointer 中保留
# Reducer 机制示例
class State(TypedDict):
messages: Annotated[list, add_messages] # 追加
score: Annotated[int, operator.add] # 累加
name: str # 覆盖(默认 reducer)这是理解 LangGraph 与祖龙差异的关键。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
# 内存模式(开发/测试)
checkpointer = MemorySaver()
# SQLite 持久化模式(生产)
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
# 编译时注入
graph = builder.compile(checkpointer=checkpointer)
# 调用时指定 thread_id
config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [{"role": "user", "content": "你好"}]}, config=config)每次 Superstep (一个节点执行完成) 后:
1. 当前完整 State 被序列化
2. 以 (thread_id, checkpoint_id) 为键存入 checkpointer
3. checkpoint_id 自动递增
恢复时:
1. 传入相同的 thread_id
2. LangGraph 自动加载最新的 checkpoint State
3. 从该 State 继续执行
| 特性 | 支持情况 | 说明 |
|---|---|---|
| 运行内恢复 | ✅ | 同一个 thread_id 内,暂停后可恢复 |
| 跨运行恢复 | ✅ | 同一个 thread_id,关闭程序后重新打开可恢复 |
| 跨会话记忆 | ❌ | 不同 thread_id 之间完全隔离,无任何共享 |
| 语义检索 | ❌ | 无法"回忆"历史对话中的相似内容 |
| 知识演化 | ❌ | 无衰减、巩固、关联学习等机制 |
| 图谱可视化 | ❌ | 纯状态快照,无图结构 |
本质定义: LangGraph 的 checkpointer 是**"对话的草稿纸"** — 它能让你暂停对话、稍后继续,但它不知道你昨天和上个星期说了什么。每次新 thread_id 都是一个全新的世界。
LangChain 文档中提到的 "Long-term Memory" 指的是在 State 中手动添加一个 memories 字段,每次调用时由开发者自行从外部存储(如向量数据库)加载相关记忆并写入 State。这不是框架内置能力,而是留给开发者的扩展点。
# 典型的 LangGraph "长期记忆" 模式(需手动实现)
class State(TypedDict):
messages: Annotated[list, add_messages]
memories: list # 手动从外部加载的长期记忆
def agent(state: State):
# 1. 手动从向量数据库检索相关记忆
relevant = vector_store.search(state["messages"][-1])
# 2. 注入到 prompt
# 3. LLM 推理
return {"messages": [response]}LangGraph 中不存在显式的 "while loop" — 循环通过图的反馈边隐式定义:
START -> agent -> tools -> agent -> tools -> ... -> END
^___________________|
(反馈边, 形成循环)
图编译时,LangGraph 内部维护一个 recursion_limit(默认 25),超过此限制抛出 GraphRecursionError 以防止无限循环。这是 LangGraph 唯一的"死循环防护"机制。
from langgraph.prebuilt import ToolNode
tools = [search_tool, calculator_tool]
tool_node = ToolNode(tools)
builder.add_node("agent", call_model.bind_tools(tools))
builder.add_node("tools", tool_node)
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})ToolNode自动解析 LLM 返回的tool_calls并执行对应工具- 工具执行结果通过
ToolMessage写回 State should_continue检查最后一条消息是否包含tool_calls
# 在某节点中暂停
def approval_node(state):
if needs_approval(state):
return interrupt({"message": "请审核此操作", "data": state})
return state
# 外部恢复
graph.invoke(Command(resume={"approved": True}), config=config)interrupt()暂停图执行并返回当前 StateCommand(resume=...)从外部恢复执行- 暂停期间 State 已通过 checkpointer 持久化
# 模式 1: 值流
for event in graph.stream(input, config):
print(event)
# 模式 2: 事件流(更细粒度)
async for event in graph.astream_events(input, config):
if event["event"] == "on_chat_model_stream":
print(event["data"]["chunk"])# 并行执行 (Send API)
def continue_to_jokes(state):
return [Send("tell_joke", {"topic": t}) for t in state["topics"]]
# 子图
parent_builder.add_node("sub_agent", sub_graph.compile())祖龙有三张彼此协同的"图",构成完整的认知执行闭环:
┌──────────────────────────────────────────────────────────┐
│ 祖龙三图架构 │
│ │
│ MemoryGraph (持久异构图) │
│ ├── 14种节点 + 14种边 │
│ ├── BFS扩散激活 + 赫布学习 + 艾宾浩斯衰减 │
│ ├── 双路检索 (热路径BFS + 冷路径FAISS) │
│ └── 九阶段全生命周期 │
│ ↕ 双向同步 │
│ TaskGraph (会话任务DAG) │
│ ├── 树形分解 (HIERARCHY) + DAG依赖 (DEPENDENCY) │
│ ├── Kahn拓扑排序 + 并行执行组 │
│ └── 递归深度类型推导 (depth->type) │
│ ↕ 驱动 │
│ FC Loop (执行编排循环) │
│ ├── check -> call_model -> exec_tools -> eval_response │
│ ├── 工具预算硬控 + 6信号死循环检测 │
│ └── 远程/内部工具分离 + 异步暂停/恢复 │
└──────────────────────────────────────────────────────────┘
文件: zulong/memory/memory_graph.py (~3825 行)
存储后端: zulong/memory/storage_hybrid/sharded_memory_graph.py (igraph + LMDB)
适配器: zulong/memory/graph_adapters.py (8 个适配器, ~86KB)
进化引擎: zulong/memory/memory_evolution.py
- 底层: NetworkX
DiGraph(兼容层) / igraph + LMDB (生产分片 Hybrid 后端) - 14 种节点类型:
| # | 类型 | 来源 | 用途 |
|---|---|---|---|
| 1 | TASK | TaskGraph | 任务执行节点 |
| 2 | DIALOGUE | ShortTermMemory | 对话轮次 |
| 3 | KNOWLEDGE | KnowledgeGraph | 实体事实 |
| 4 | EXPERIENCE | ExperienceRAG | 学习经验 |
| 5 | EPISODE | EpisodicMemory | 摘要情节 |
| 6 | FILE | TaskNode.files | 引用文件 |
| 7 | CONCEPT | KG CONCEPT | 抽象概念 |
| 8 | PERSON | PersonProfile/KG | 人物身份 |
| 9 | DOCUMENT | (预留) | 文档切片 |
| 10 | CODE_SYMBOL | Tree-sitter AST | 函数/类/方法 |
| 11 | MODULE | 文件系统 | 目录/包/模块 |
| 12 | TOOL_CALL | L2 执行 | 工具调用记录 |
| 13 | TOOL_RESULT | L2 执行 | 工具输出/错误 |
| 14 | APPROVAL | L2 执行 | 审批决策 |
- 14 种边类型:
| # | 类型 | 结构边? | 用途 |
|---|---|---|---|
| 1 | HIERARCHY | 是 | 父子关系 (永不移除) |
| 2 | DEPENDENCY | 是 | 数据依赖 (永不移除) |
| 3 | TEMPORAL | 是 | 时间序列 (永不移除) |
| 4 | REFERENCE | 否 | 跨类型引用 |
| 5 | SEMANTIC | 否 | 向量余弦相似度 |
| 6 | CAUSAL | 否 | 因果关系 |
| 7 | ASSOCIATION | 否 | 赫布共激活创建 |
| 8 | DERIVED_FROM | 否 | 经验来源 |
| 9 | APPLIES_TO | 否 | 经验适用性 |
| 10 | SIMILAR_TO | 否 | 经验间相似度 |
| 11 | CORRECTS | 否 | 纠正性经验 |
| 12 | DEPENDS_ON | 否 | 经验依赖 |
| 13 | CONTRADICTS | 否 | 冲突检测 |
| 14 | FAILED_THEN_SUCCEEDED | 否 | 错误恢复追踪 |
a) BFS 扩散激活 (compute_activations)
# 核心公式
propagated = current_activation * edge_weight * decay^depth
# 参数
max_depth = 3 # 静态模式
decay = 0.5 # 每跳衰减 50%
min_activation = 0.01 # 低于此值停止传播对于对话窗口大小自适应 (compute_activations_dynamic):
- 大窗口 (>64K) + 低使用率 (<50%): max_depth=5, 更广泛传播
- 小窗口 (<32K) 或高使用率 (>80%): max_depth=2, 减少噪声
b) 赫布学习 (hebbian_strengthen)
# 边权重增强
eta = 0.1
new_weight = old_weight + eta * (1.0 - old_weight) # 渐近逼近 1.0
# 共激活关联边创建
if coactivation_count >= 3: # 阈值
创建 ASSOCIATION 边,权重 = 0.3节点对共激活计数器跟踪,达到阈值 3 时自动创建 ASSOCIATION 边。每节点最多 10 条 ASSOCIATION 出边,总共最多 5000 个共激活对。
c) 艾宾浩斯衰减 (decay_and_prune)
# 指数衰减公式
decayed_weight = weight * exp(-elapsed_hours * ln(2) / half_life)
# 基于重要度的半衰期
_TRIVIAL: 6 小时
_NORMAL: 24 小时 (默认)
_IDENTITY: 720 小时 (30 天)
_FACT: 360 小时 (15 天)
_IMPORTANT: 168 小时 (7 天)
_MUST_REMEMBER: 无穷 (永不衰减)剪枝阈值: weight < 0.05 直接移除; 0.05 < weight < 0.15 提交 LLM 审查。孤儿节点(度=0)根据重要度有 6h-7d 的容忍期后清除。
d) 双路并行检索 (retrieve_context)
# 并行执行
hot_task = _retrieve_hot(query, window_seconds, session_id) # BFS + 关键词
cold_task = _retrieve_cold(query, top_k, window_seconds) # FAISS 向量
hot_results, cold_results = await asyncio.gather(hot_task, cold_task)- 热路径 (<50ms): session boost(2.0) + 元记忆检测 + 关键词匹配(0.8) + 语义相似度(0.4) + 1跳 BFS 扩展 + 焦点路径 boost(0.2) + 重要度 boost
- 冷路径 (<200ms): FAISS 向量搜索(权重 0.7) + 关键词搜索(权重 0.3) + BFS 深度搜索(drill-down) + 重要度 boost
- L1 摘要导航: SQLite 摘要 + FAISS 向量,混合并行检索
| 阶段 | 方法/触发器 | 行为 |
|---|---|---|
| Bootstrap | add_node() | 创建节点, HOT 温度, NORMAL 重要度, base_strength=1.0 |
| Activate | compute_activations() | BFS 扩散, 更新种子节点 access_count, 记录共激活边 |
| Hebb | hebbian_strengthen() | 增强共激活边权重, 创建新 ASSOCIATION 边 |
| Auto-embed | _auto_embed_node() | 后台生成向量嵌入, 发现语义邻居 |
| Consolidate | run_importance_review() | access_count>=3 触发 NORMAL->IMPORTANT 升级 |
| Decay | decay_and_prune() | 指数衰减, 移除弱边 |
| Review | submit_prune_review() | 弱边提交 LLM 审查 (KEEP/DISCARD/COMPRESS/PROMOTE/MERGE) |
| Forget | cleanup_orphan_nodes() | 移除孤立节点和语义隔离节点 |
| Persist | 自动保存 | 30 分钟周期 (积压时自适应缩短至 5 分钟) |
- Temperature: HOT (<1h) / WARM (1-24h) / COLD (>24h) — 动态计算,决定热/冷路径
- Importance: TRIVIAL(0) < NORMAL(1) < IDENTITY(2) < FACT(3) < IMPORTANT(4) < MUST_REMEMBER(5) — 决定衰减半衰期
- TimeScope: RECENT / NON_RECENT — 按可配置窗口 (默认 30 分钟) 判定
适配器是将后端存储器投影到统一 MemoryGraph 的桥梁:
| 适配器 | 源系统 | 生成节点 | 关键边 |
|---|---|---|---|
| TaskGraphAdapter | pipeline.TaskGraph | TASK, FILE, CODE_SYMBOL | HIERARCHY, DEPENDENCY, REFERENCE |
| KnowledgeGraphAdapter | KnowledgeGraph | KNOWLEDGE, CONCEPT, PERSON | CAUSAL, REFERENCE |
| DialogueAdapter | ShortTermMemory | DIALOGUE | TEMPORAL, REFERENCE |
| EpisodeAdapter | EpisodicMemory | EPISODE | REFERENCE, DERIVED_FROM |
| PersonProfileAdapter | PersonProfile | PERSON | REFERENCE |
| ExperienceAdapter | ExperienceRAG | EXPERIENCE | APPLIES_TO, DERIVED_FROM, SIMILAR_TO 等 6 种 |
| CodeGraphAdapter | Tree-sitter AST | CODE_SYMBOL, MODULE | HIERARCHY, REFERENCE |
文件: zulong/l2/task_graph.py (~1342 行)
调度器: TaskScheduler (同文件内)
TaskGraph 是一个混合双图结构——在分层树上叠加有向依赖图:
TaskGraph:
├── _nodes: Dict[str, TaskNode] # 扁平节点映射
├── _h_edges: List[(parent, child)] # 层级边 -> 定义一棵树
│ _h_edge_set: set() # 去重索引
├── _d_edges: List[DependencyEdge] # 依赖边 -> 定义叶子节点间的 DAG
└── parallel_groups: List[List[str]] # Kahn 算法计算出的并行组depth_to_type(0) = "requirement" # 根需求
depth_to_type(1) = "analysis" # 分析
depth_to_type(2) = "outline" # 大纲
depth_to_type(3) = "task" # 任务
depth_to_type(4) = "subtask" # 子任务
depth_to_type(5) = "micro_task" # 微任务
depth_to_type(6+) = "step_6" ... # 永不收敛示例分解树:
"坦克大战游戏" (req, depth=0)
├── "阶段0: 需求分析" (analysis, depth=1)
│ ├── "创建项目结构" (outline, depth=2)
│ │ ├── "创建HTML骨架" (task, depth=3)
│ │ │ ├── "编写<!DOCTYPE html>" (subtask, depth=4)
│ │ │ │ └── "设置viewport meta" (micro_task, depth=5)
convert_leaf_to_parent(node_id, children):
"""将叶子节点转为中间节点,挂载子节点(增量拆分)"""
depth = parent_depth + 1
# 自动编号: node_id_1, node_id_2, ...
# 自动类型: depth_to_type(depth)父节点状态从子节点自动聚合:
- 所有子节点 completed -> parent = completed
- 任一子节点 in_progress -> parent = in_progress
- 任一子节点 blocked -> parent = blocked
# DFS 环检测 (validate_dependencies)
# 在依赖图中运行 DFS,检测后向边
# Kahn 算法拓扑排序 (compute_execution_tiers)
# 计算分层执行顺序,同层节点可并行执行
tiers = [[tier0_ids], [tier1_ids], ...]
# 下一批可执行节点 (get_next_executable)
# 所有前置依赖都 completed/skipped 的 pending/blocked 叶子节点状态: pending -> in_progress -> completed
-> blocked (可通过 rollback 回到 pending)
-> needs_adjust (需要修订)
-> skipped (依赖 blocked, 自动跳过)
completed -> needs_adjust (质量回检失败)每次状态变更自动同步:
_sync_node_to_memory_graph(node_id, status):
# 1. 在 MemoryGraph 中查找对应 TASK 节点
# 2. 更新 task_status / task_result / activation
# 3. completed -> activation=0.1 / in_progress -> 0.9 / blocked -> 0.5TaskGraphAdapter().sync(mg, tg) 在 BFS 激活时进行完整结构同步。
文件: zulong/ide/ide_fc_runner.py (~5600 行, IDEFCRunner)
引擎: zulong/l2/inference_engine.py (~4500 行)
while True:
1. _check(state) # Gatekeeper: 中断标志/轮数限制/重复工具检测
2. _call_model(state) # LLM API 调用 (流式)
3a. 若有 tool_calls:
- 应用工具预算限制
- _exec_tools(state) # 内部/远程分离执行
- 若有远程工具 -> 暂停等待
- 继续循环
3b. 若无 tool_calls:
- _eval_response(state) # 7 层安全网
- 返回 "done" / "continue" / "cb_force"
4. 若 "done" -> _finalize() -> 任务完成- CircuitBreaker 收敛检查: 循环 > 20 轮次 + cb_force 信号
- 信息缺口检测: InformationGapDetector 检查
- 规则守护: RuleGuardian 违规检查
- 漂移检测: DriftDetector 离题检查
- 回填检测: 文本响应匹配到 TaskGraph 节点
- 未完成任务守卫: >30% 叶子未完成 -> 强制重试
- 最终确认: 检查是否真的完成
| 信号 | 检测条件 | 触发动作 |
|---|---|---|
| tool_streak | 同一工具连续调用 >= 5 次 | 移除该工具定义 |
| loop_count | 总轮次 >= 软限制 | 注入收敛引导 |
| response_length | 响应长度持续增长 | 长度截断 |
| no_progress | 3 轮无实质进展 | cb_force 信号 |
| empty_response | 连续空响应 | 直接终止 |
| duplicate_tool | 重复调用相同工具+参数 3 次 | 跳过执行 |
# 用户显式要求 "不要工具/最多N个工具/不超过N次调用"时:
if tool_calls_used >= tool_call_budget:
# 硬控: 移除 tools 和 tool_choice 从 API 参数
kwargs.pop("tools", None)
kwargs.pop("tool_choice", None)
# 注入强制收敛提示
messages.append("已达到工具调用上限, 请基于现有上下文生成总结")L1-B ToolPredictor
└── 推理前预判: 基于用户输入预判所需工具清单
↓
L2 单次决策
└── LLM 在预判清单基础上做一次工具选择
↓
FC 循环执行
├── Internal 工具 (Python 线程池执行)
└── Remote 工具 (VS Code WebSocket 异步执行)
└── 暂停 -> 等待 tool_result -> 恢复
| 能力维度 | LangGraph | 祖龙 MemoryGraph |
|---|---|---|
| 记忆模型 | 线性状态快照 | 异构有向图 (14节点 x 14边) |
| 跨会话记忆 | ❌ (不同 thread_id 完全隔离) | ✅ (核心能力) |
| 激活传播 | 无 | BFS 3跳加权传播 + 动态深度自适应 |
| 关联学习 | 无 | 赫布共激活增强 + 自动关联边创建 |
| 记忆衰减 | 无 | 艾宾浩斯指数衰减 (6h~∞半衰期) |
| 语义检索 | 需手动集成外部向量数据库 | 内置双路并行检索 (BFS + FAISS) |
| 记忆审查 | 无 | LLM 异步审查 (KEEP/DISCARD/COMPRESS/PROMOTE/MERGE) |
| 注意力标记 | 无 | Temperature x Importance x TimeScope 三维标签 |
| 生命周期 | 无 (State 无概念) | 九阶段完整生命周期 |
| 持久化 | SQLite (checkpointer) | LMDB + igraph + FAISS + 分片 + 原子写入 |
| 可视化 | 无内置 | BFS 扩散实时激活路径 + 注意力三态 + Canvas 2D 动画 |
本质差异: LangGraph 的 checkpointer 是"对话的草稿纸",祖龙 MemoryGraph 是"机器人的终身大脑"。两者在记忆维度上不是同一量级的设计。
| 能力维度 | LangGraph StateGraph | 祖龙 TaskGraph + FC Loop |
|---|---|---|
| 编排模型 | 隐式 (图的拓扑定义循环) | 显式 (while 循环驱动图变更) |
| 图结构 | 单层有向图 (节点+边+条件边) | 双层: 树形分解 + DAG 依赖 |
| 任务分解 | 需手动设计子图 | depth_to_type() 自动 6 层类型推导 |
| 执行顺序 | 图边定义(拓扑顺序) | Kahn 算法拓扑排序 + 并行执行组 |
| 状态 Schema | Pydantic/TypedDict + Reducer 自动合并 | 手动字典 + 显式字段更新 |
| 状态持久化 | Checkpointer 自动全量保存 | 手动序列化 + 防抖自动保存 + 崩溃恢复 |
| 死循环防护 | recursion_limit (单信号) | 6 信号 CircuitBreaker |
| 工具预算 | 无 | 硬控 (移除 tools + 注入收敛提示) |
| 任务回滚 | 无 | rollback_node + 经验修正 |
| 任务归档 | 无 | 自动完成 + 自动归档 |
| 跨会话恢复 | ✅ (同 thread_id) | ✅ (反序列化 + 重建) |
| 多任务并发 | 需手动设计子图 | 并行执行组 (同层节点可并行) |
| 能力维度 | LangGraph | 祖龙 |
|---|---|---|
| 工具定义 | ToolNode 绑定 | IDEToolRegistry 注册 |
| 工具选择 | LLM 自主决策 | L1-B 预判 -> L2 决策 -> FC 执行 |
| 工具分类 | 无分类 | Internal (Python) / Remote (VS Code) |
| 远程执行 | 同步阻塞 | 异步暂停/恢复 + WebSocket 双向通信 |
| 预算控制 | 无 | 硬控 (tool_call_budget) |
| 参数验证 | Pydantic schema | 别名映射 + 默认值 + 路径修正 + 必填验证 |
LangGraph:
┌────────────────────────┐
│ StateGraph (扁平) │
│ agent <-> tools │
└────────────────────────┘
无层级, 单图执行
祖龙:
┌───────────────────────────────────────┐
│ L3 专家层: DualBrain 热备 + 专家池 │
├───────────────────────────────────────┤
│ L2 认知层: InferenceEngine + FC循环 │
│ + TaskGraph + CircuitBreaker │
├───────────────────────────────────────┤
│ L1 感知层: L1-A 反射 / L1-B 调度 │
│ L1-C 视觉 / L1-D 听觉 │
│ L1-E 安全 │
├───────────────────────────────────────┤
│ L0 设备层: 摄像头/麦克风/扬声器/传感器 │
└───────────────────────────────────────┘
四层严格分层, 层间 EventBus 解耦
LangGraph 的 checkpointer 解决的是运行内可靠性问题 — "Agent 执行到一半挂了,能从断点恢复"。哲学: "Stateless between runs."
祖龙的 MemoryGraph 解决的是机器人终身学习问题 — "机器人需要记住用户是谁、做过什么任务、犯过什么错误"。哲学: "Every interaction enriches a persistent knowledge fabric."
这与两者的定位完全吻合: LangGraph 面向"帮我写个客服 Agent"的单次任务; 祖龙面向"与用户共同生活数年"的持久关系。
LangGraph 的编排是**"图即循环"** — agent 和 tools 节点通过反馈边形成隐式循环。修改 Agent 行为 = 修改图拓扑。
祖龙的编排是**"循环驱动图"** — FC Loop 是一个显式 while 循环,在循环内部操作 TaskGraph 并同步 MemoryGraph。TaskGraph 是数据(可持久化、版本化、跨会话恢复),FC Loop 是控制流。修改 Agent 行为 = 修改循环中的策略函数。
| 维度 | LangGraph | 祖龙 |
|---|---|---|
| 环境感知 | 无 | L1-C 四层视觉 + L1-D 三层音频 |
| 紧急中断 | interrupt() (手动调用) | EventBus 毫秒级穿透 |
| 重评估 | 无 | 环境快照对比 (ABORT/REPLAN/RESUME/MERGE) |
| 语音交互 | 无 | 唤醒词 + ASR + TTS 全链路 |
| 安全反射 | 无 | L1-A 50ms 紧急响应 |
| 电源管理 | 无 | L1-B 动态加载/卸载 L2 模型 |
-
自动 Checkpointer 模式: 将当前手动的
_save_runner_state()改为每次 Superstep 自动持久化,使暂停/恢复更加透明和可靠 -
声明式条件边:
add_conditional_edges的声明式路由比 if-else 分支更具可读性和可测试性 -
Pydantic State Schema: 强类型 State 定义 + automatic reducer 合并,减少手动字段操作错误
-
子图组合:
add_subgraph可用于更模块化地封装 L3 专家调用链路 -
统一流式事件:
astream_events()的统一事件流抽象与祖龙的 EventBus 可互为参考
- 真正长期记忆: 这是 LangGraph 社区最频繁提及的痛点,祖龙 MemoryGraph 提供完整方案
- 多层认知架构: 反射层/调度层/认知层的层次化设计
- 多维死循环防护: 6 信号 vs 单信号 recursion_limit
- 任务生命周期: 创建->分解->执行->级联->归档 完整闭环
- 环境感知与重评估: 环境快照对比 + 智能决策
- 三维记忆标签: Temperature x Importance x TimeScope
可以将 LangGraph 作为祖龙 L2 层内部的编排子引擎:
# 概念示意 (非实际实现)
class LangGraphOrchestratedTask:
"""用 LangGraph 替代 FC Loop 中的 call_model->exec_tools 微循环"""
def build_graph(self, tools, task_context):
builder = StateGraph(TaskState)
builder.add_node("think", self.call_llm)
builder.add_node("act", ToolNode(tools))
builder.add_conditional_edges("think", self.should_act, ...)
builder.add_edge("act", "think")
# 注入祖龙能力
builder.add_node("remember", self.sync_to_memory_graph)
builder.add_node("check_budget", self.enforce_tool_budget)
builder.add_node("circuit_break", self.check_circuit_breaker)
return builder.compile()但这会引入不必要的依赖和复杂度。祖龙当前的 FC Loop 已经工作良好,且拥有 LangGraph 不具备的深层次能力(工具预算硬控、6 信号断路器、七层安全网、远程工具异步暂停等)。
| 问题 | 答案 |
|---|---|
| 祖龙是否基于 LangGraph? | 否。完全独立设计 |
| 两者概念层面是否相似? | 是。都认可"图"是 Agent 编排的最佳抽象 |
| 祖龙能否用 LangGraph 替代? | 不能。LangGraph 无法满足祖龙的终身记忆、四层认知、硬实时约束 |
| LangGraph 能否受益于祖龙经验? | 是。长期记忆是多 Agent 框架的普遍痛点 |
LangGraph 和祖龙是两条独立演化的河流,因共同的"图驱动智能体"理念产生相似的表层形态,但在深层 — 记忆的持久性、认知的层次化、硬实时的约束 — 祖龙走得更远。LangGraph 为"跑完的脚本"提供优雅的编排抽象,祖龙为"活着的机器人"构建完整的多层认知操作系统。
| 文件 | 用途 |
|---|---|
TSD/祖龙 (ZULONG) 机器人系统技术规格说明书 (TSD)1.7.txt |
系统最高指导文档 (v2.9.10) |
zulong/memory/memory_graph.py |
异构图记忆 (~3825 行) |
zulong/memory/graph_adapters.py |
八大图适配器 (~86KB) |
zulong/memory/memory_evolution.py |
记忆进化引擎 |
zulong/memory/storage_hybrid/sharded_memory_graph.py |
分片 Hybrid 存储后端 |
zulong/l2/task_graph.py |
递归 DAG 任务图 (~1342 行) |
zulong/ide/ide_fc_runner.py |
FC 循环执行器 (~5600 行) |
zulong/l2/inference_engine.py |
L2 推理引擎 (~4500 行) |
zulong/l2/circuit_breaker.py |
6 信号死循环检测器 (~521 行) |
zulong/core/event_bus.py |
事件总线 (单例, 优先级队列) |
zulong/l1b/scheduler_gatekeeper.py |
L1-B 调度与意图守门 (~2652 行) |