基础图与状态
  → 条件路由与 Checkpoint
  → Tool Calling / ReAct
  → Human-in-the-loop / Command
  → Streaming / Retry / Error Handling
  → Parallel / Reducer / Structured Output
  → Runtime Context / Subgraph / Send
  → Store / Functional API
  → State History / Testing / Observability
  → Durable Persistence / Async / Safe Tools

课程目录

课程	文件	核心内容	是否需要模型 API
01	`01_basic_agent_no_llm.py`	`StateGraph`、State、Node、Edge、`START/END`	否
02	`02_chat_graph.py`	Chat Model 节点、消息 Reducer	是
03	`03_conditional_edge.py`	条件边与路由函数	否
04	`04_checkpointer.py`	Checkpointer、`thread_id`、多轮状态	是
05	`05_tool_call.py`	`bind_tools`、`ToolNode`、ReAct 工具循环	是
06	`06_human_in_the_loop.py`	`interrupt()`、人工审批、暂停与恢复	是
07	`07_command.py`	`Command(update/goto/resume)`	否
08	`08_streaming.py`	`updates`、`messages`、多模式 Streaming	是
09	`09_retry_and_errors.py`	`RetryPolicy`、`error_handler`、循环保护	否
10	`10_parallel_and_reducers.py`	并行 Super-step、Reducer、更新冲突	否
11	`11_structured_output.py`	Pydantic、`with_structured_output`	是
12	`12_runtime_context.py`	State / Config / Runtime Context、依赖注入	否
13	`13_subgraphs.py`	子图作为父图节点、子图 Streaming	否
14	`14_map_reduce_send.py`	`Send`、动态并行 Map-Reduce	否
15	`15_long_term_store.py`	Checkpointer 与 Store、跨线程长期记忆	否
16	`16_functional_api.py`	`@entrypoint`、`@task`、Future、`previous`	否
17	`17_state_history.py`	状态快照、历史、Replay、Fork、Time Travel	否
18	`18_testing.py`	节点、路由和整图的确定性单元测试	否
19	`19_observability.py`	Tags、Metadata、自定义事件、可选 LangSmith	否
20	`20_persistent_memory.py`	SQLite 持久化 Checkpointer/Store、进程间恢复	否
21	`21_async_graph.py`	`ainvoke/astream`、异步并行、重试和节点超时	否
22	`22_safe_tools.py`	参数校验、权限、审批、幂等与审计	否

环境要求

Python 3.11+（当前在 Python 3.12 环境验证）
LangGraph 1.2.9+
模型课程还需要 langchain-openai 和 python-dotenv
课程 20 的 SQLite 持久化后端是可选依赖

使用 Conda

conda create -n langgraph python=3.12 -y
conda activate langgraph
python -m pip install -U "langgraph>=1.2.9" langchain-openai python-dotenv pydantic

SQLite 持久化课程额外安装：

python -m pip install "langgraph-checkpoint-sqlite>=3,<4"

使用 uv

uv venv
.venv\Scripts\activate
uv pip install "langgraph>=1.2.9" langchain-openai python-dotenv pydantic

模型配置

需要模型 API 的课程会从 .env 读取配置。创建本地 .env：

OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://your-openai-compatible-endpoint/v1
OPENAI_MODEL=your-model-name

.env 已被 .gitignore 排除，不要提交真实密钥。

使用官方 OpenAI 时，可以根据服务要求省略 OPENAI_BASE_URL；本仓库部分早期示例带有兼容服务的默认地址，运行前请检查模型名和地址是否与自己的服务匹配。

运行示例

运行单个课程：

python .\01_basic_agent_no_llm.py
python .\10_parallel_and_reducers.py
python .\17_state_history.py

运行测试课程：

python .\18_testing.py

运行持久化课程：

# 真正写入本地 SQLite，重复运行可观察跨进程持久化
python .\20_persistent_memory.py

# 无需可选依赖的非持久化 fallback
python .\20_persistent_memory.py --fallback

默认 SQLite 文件是：

persistent_memory.sqlite3

可指定其他路径：

python .\20_persistent_memory.py --db .\data\memory.sqlite3

运行异步课程：

python .\21_async_graph.py

17～22 课程概要

17：State History / Time Travel

演示：

graph.get_state(config) 获取最新快照
graph.get_state_history(config) 查看历史
使用历史 checkpoint config 重放
graph.update_state() 从历史状态创建新分支
验证 Fork 不会修改原历史快照

18：Testing

使用标准库 unittest，不调用真实模型或网络：

普通节点单元测试
路由函数测试
编译图的集成测试
确定性输入和断言

19：Observability

演示：

RunnableConfig 中的 tags 和 metadata
get_stream_writer() 发送自定义进度事件
stream_mode=["custom", "updates"]
可选启用 LangSmith，不设置密钥也能本地运行

启用 LangSmith 时，在 .env 中配置：

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=your-langsmith-key
LANGSMITH_PROJECT=langgraph-learning

20：Persistent Memory

演示两类持久化：

Checkpointer：保存 thread_id 对应的执行状态
Store：保存按用户 Namespace 组织的跨线程长期记忆

生产环境还应考虑连接池、迁移、备份、加密、访问控制和数据保留策略。SQLite 适合本地学习和单进程应用；多实例服务通常使用数据库后端。

21：Async Graph

演示：

async def 节点
await graph.ainvoke(...)
async for ... in graph.astream(...)
并行异步 I/O
异步节点 RetryPolicy
LangGraph 1.2+ 节点 timeout
NodeTimeoutError

22：Safe Tools

使用完全模拟的“转账”流程演示：

Pydantic 严格参数校验
Runtime Context 权限检查
interrupt() 敏感操作审批
幂等键避免重复副作用
审计日志
密钥和依赖不进入 State

该课程不会连接支付系统，也没有真实外部副作用。

重要概念速查

State、Context 与 Config

State   = 工作流知道和产生的数据，会随节点更新并可进入 Checkpoint
Context = 本次运行所需的只读依赖，如用户身份、服务、权限
Config  = LangGraph 执行配置，如 thread_id、recursion_limit、tags

Checkpointer 与 Store

Checkpointer = 一条 thread 的执行状态、暂停位置和历史
Store        = 跨 thread 的用户记忆或长期业务数据

Graph API 与 Functional API

Graph API      = 显式节点和边，适合可视化、复杂编排
Functional API = @entrypoint + @task + 普通 Python 控制流

安全原则

不将 API Key、密码或 Token 写入 State
敏感工具执行前进行权限检查和人工审批
外部副作用必须支持幂等
参数错误不要盲目重试
recursion_limit 是安全网，不是业务终止条件
生产环境不要使用 InMemorySaver / InMemoryStore 代替持久化后端

官方文档

LangGraph Overview：https://docs.langchain.com/oss/python/langgraph/overview
Graph API：https://docs.langchain.com/oss/python/langgraph/graph-api
Functional API：https://docs.langchain.com/oss/python/langgraph/functional-api
Persistence：https://docs.langchain.com/oss/python/langgraph/persistence
Streaming：https://docs.langchain.com/oss/python/langgraph/streaming
Fault Tolerance：https://docs.langchain.com/oss/python/langgraph/fault-tolerance

README.md Unescape Escape

LangGraph 学习示例

学习路线