返回顶部
l

langgraph-architectureLangGraph架构指南

Guides architectural decisions for LangGraph applications. Use when deciding between LangGraph vs alternatives, choosing state management strategies, designing multi-agent systems, or selecting persistence and streaming approaches.

作者: admin | 来源: ClawHub
源自
ClawHub
版本
V 1.0.0
安全检测
已通过
136
下载量
免费
免费
0
收藏
概述
安装方式
版本历史

langgraph-architecture

LangGraph架构决策

何时使用LangGraph

在以下情况下使用LangGraph:

  • - 有状态对话 - 具有记忆的多轮交互
  • 人在回路中 - 审批关卡、修正、干预
  • 复杂控制流 - 循环、分支、条件路由
  • 多智能体协调 - 多个LLM协同工作
  • 持久化 - 从检查点恢复、时间旅行调试
  • 流式传输 - 实时令牌流式传输、进度更新
  • 可靠性 - 重试、错误恢复、持久性保证

考虑替代方案的情况:

场景替代方案原因
单次LLM调用直接API调用开销不合理
线性流水线
LangChain LCEL | 更简单的抽象 | | 无状态工具使用 | 函数调用 | 无需持久化 | | 简单RAG | LangChain检索器 | 内置模式 | | 批处理 | 异步任务 | 不同的执行模型 |

状态模式决策

TypedDict vs Pydantic

TypedDictPydantic
轻量级,更快运行时验证
字典式访问
属性访问 | | 无验证开销 | 类型强制转换 | | 更简单的序列化 | 复杂嵌套模型 |

建议:大多数情况下使用TypedDict。需要验证或复杂嵌套结构时使用Pydantic。

归约器选择

用例归约器示例
聊天消息add_messages处理ID、RemoveMessage
简单追加
operator.add | Annotated[list, operator.add] | | 保留最新 | 无(LastValue) | field: str | | 自定义合并 | Lambda | Annotated[list, lambda a, b: ...] | | 覆盖列表 | Overwrite | 绕过归约器 |

状态大小考虑

python

小状态(< 1MB)- 放入状态


class State(TypedDict):
messages: Annotated[list, add_messages]
context: str

大数据 - 使用存储

class State(TypedDict): messages: Annotated[list, add_messages] document_ref: str # 存储引用

def node(state, *, store: BaseStore):
doc = store.get(namespace, state[document_ref])
# 处理而不使检查点膨胀

图结构决策

单图 vs 子图

单图适用于:

  • - 所有节点共享相同状态模式
  • 简单线性或分支流程
  • 节点数 < 10

子图适用于:

  • - 需要不同状态模式
  • 跨图可复用组件
  • 团队关注点分离
  • 复杂分层工作流

条件边 vs 命令

条件边命令
基于状态的路由路由 + 状态更新
独立路由函数
节点内决策 | | 更清晰的可视化 | 更灵活 | | 标准模式 | 动态目标 |

python

条件边 - 当路由是重点时


def router(state) -> Literal[a, b]:
return a if condition else b
builder.addconditionaledges(node, router)

命令 - 当结合路由与更新时

def node(state) -> Command: return Command(goto=next, update={step: state[step] + 1})

静态 vs 动态路由

静态边(add_edge):

  • - 构建时已知的固定流程
  • 更清晰的图可视化
  • 更容易推理

动态路由(addconditionaledges、Command、Send):

  • - 基于状态的运行时决策
  • 智能体驱动的导航
  • 扇出模式

持久化策略

检查点选择

检查点用例特性
InMemorySaver仅测试重启后丢失
SqliteSaver
开发 | 单文件,本地 | | PostgresSaver | 生产 | 可扩展,并发 | | 自定义 | 特殊需求 | 实现BaseCheckpointSaver |

检查点范围

python

完全持久化(默认)


graph = builder.compile(checkpointer=checkpointer)

子图选项

subgraph = sub_builder.compile( checkpointer=None, # 从父图继承 checkpointer=True, # 独立检查点 checkpointer=False, # 无检查点(原子运行) )

何时禁用检查点

  • - 应原子运行的短生命周期子图
  • 状态模式不兼容的子图
  • 无需恢复的性能关键路径

多智能体架构

监督者模式

最适合:

  • - 清晰层级
  • 集中决策
  • 不同智能体专业化

┌─────────────┐
│ 监督者 │
└──────┬──────┘
┌────────┬───┴───┬────────┐
▼ ▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│智能体1│ │智能体2│ │智能体3│ │智能体4│
└──────┘ └──────┘ └──────┘ └──────┘

对等模式

最适合:

  • - 协作智能体
  • 无清晰层级
  • 灵活通信

┌──────┐ ┌──────┐
│智能体1│◄───►│智能体2│
└──┬───┘ └───┬──┘
│ │
▼ ▼
┌──────┐ ┌──────┐
│智能体3│◄───►│智能体4│
└──────┘ └──────┘

交接模式

最适合:

  • - 顺序专业化
  • 清晰阶段转换
  • 每阶段不同能力

┌────────┐ ┌────────┐ ┌────────┐
│研究 │───►│规划 │───►│执行 │
└────────┘ └────────┘ └────────┘

流式传输策略

流模式选择

模式用例数据
updatesUI更新仅节点输出
values
状态检查 | 每一步的完整状态 | | messages | 聊天用户体验 | LLM令牌 | | custom | 进度/日志 | 通过StreamWriter的自定义数据 | | debug | 调试 | 任务 + 检查点 |

子图流式传输

python

从子图流式传输


async for chunk in graph.astream(
input,
stream_mode=updates,
subgraphs=True # 包含子图事件
):
namespace, data = chunk # namespace表示深度

人在回路中设计

中断位置

策略用例
interruptbefore操作前审批
interruptafter
完成后审查 | | 节点内interrupt() | 动态、上下文暂停 |

恢复模式

python

简单恢复(同一线程)


graph.invoke(None, config)

带值恢复

graph.invoke(Command(resume=approved), config)

恢复特定中断

graph.invoke(Command(resume={interrupt_id: value}), config)

修改状态并恢复

graph.updatestate(config, {field: newvalue}) graph.invoke(None, config)

错误处理策略

重试配置

python

每节点重试


RetryPolicy(
initial_interval=0.5,
backoff_factor=2.0,
max_interval=60.0,
max_attempts=3,
retry_on=lambda e: isinstance(e, (APIError, TimeoutError))
)

多个策略(首个匹配获胜)

builder.addnode(node, fn, retrypolicy=[ RetryPolicy(retryon=RateLimitError, maxattempts=5), RetryPolicy(retryon=Exception, maxattempts=2), ])

回退模式

python
def nodewithfallback(state):
try:
return primary_operation(state)
except PrimaryError:
return fallback_operation(state)

或使用条件边进行复杂回退路由

def routeonerror

标签

skill ai

通过对话安装

该技能支持在以下平台通过对话安装:

OpenClaw WorkBuddy QClaw Kimi Claude

方式一:安装 SkillHub 和技能

帮我安装 SkillHub 和 langgraph-architecture-1776067800 技能

方式二:设置 SkillHub 为优先技能安装源

设置 SkillHub 为我的优先技能安装源,然后帮我安装 langgraph-architecture-1776067800 技能

通过命令行安装

skillhub install langgraph-architecture-1776067800

下载

⬇ 下载 langgraph-architecture v1.0.0(免费)

文件大小: 4.31 KB | 发布时间: 2026-4-17 15:11

v1.0.0 最新 2026-4-17 15:11
langgraph-architecture 1.0.0

- Initial release providing architectural guidance for LangGraph applications.
- Covers state management, reducers, and persistence strategies.
- Outlines choices for graph structure, multi-agent design patterns, and streaming options.
- Details human-in-the-loop architectures, error handling, and scaling practices.
- Helps compare LangGraph to alternatives for common scenarios.

Archiver·手机版·闲社网·闲社论坛·羊毛社区· 多链控股集团有限公司 · 苏ICP备2025199260号-1

Powered by Discuz! X5.0   © 2024-2025 闲社网·线报更新论坛·羊毛分享社区·http://xianshe.com

p2p_official_large
返回顶部