---
type: study
created: 2026-08-11
updated: 2026-08-11
sensitivity: private
status: active
tags:
  - study
  - langchain
  - langgraph
  - python
  - agent-architecture
topic: LangGraph Python 基础、运行时与 API 选择
sources:
  - https://docs.langchain.com/oss/python/langgraph
  - https://docs.langchain.com/oss/python/langgraph/overview
  - https://docs.langchain.com/oss/python/langgraph/install
  - https://docs.langchain.com/oss/python/langgraph/quickstart
  - https://docs.langchain.com/oss/python/langgraph/thinking-in-langgraph
  - https://docs.langchain.com/oss/python/langgraph/workflows-agents
  - https://docs.langchain.com/oss/python/langgraph/choosing-apis
  - https://docs.langchain.com/oss/python/langgraph/functional-api
  - https://docs.langchain.com/oss/python/langgraph/use-functional-api
  - https://docs.langchain.com/oss/python/langgraph/pregel
last_verified: 2026-08-11
---
# LangGraph Python 基础、运行时与 API 选择

> `/oss/python/langgraph` 是 overview 的规范入口别名，浏览器最终落到 `/overview`。

## 核心结论

LangGraph 是面向长时间运行、有状态 agent 的低层编排框架与运行时。它不替你抽象提示词、模型或固定 agent 架构，而是解决更底层的问题：

- 在一个流程中混合确定性代码与 LLM 决策；
- 显式管理状态、分支、循环和并行；
- 在节点边界持久化并从失败或人工暂停处恢复；
- 持续流式暴露执行过程；
- 把人工审批、长期任务和可追踪状态作为一等能力。

如果只是常规“模型调用工具直到回答”的 agent，优先从 LangChain 的 create_agent 开始；当需要显式状态机、复杂恢复语义、确定性控制或定制编排时再使用 LangGraph。Deep Agents 和 LangChain agent 都构建在 LangGraph 之上。

## 安装与最低认知

当前 Python 栈以 Python 3.10+ 为基线：

    pip install -U langgraph

文档示例通常同时使用 LangChain 的模型、消息和工具接口，因此实际项目常见安装组合是：

    pip install -U langgraph langchain

模型提供商包需要单独安装。LangGraph 本身不要求必须使用 LangChain，也不绑定具体模型提供商。

在开始前至少要理解：

1. 模型如何接收消息并返回 AI 消息；
2. 工具 schema、工具调用与 ToolMessage 的配对；
3. 节点副作用为什么需要幂等；
4. thread、checkpoint 与恢复不是普通函数调用语义。

## “用 LangGraph 思考”的五步法

### 1. 先画真实业务流程

不要先画一个“agent”盒子。先把待自动化过程拆为离散步骤，每个步骤只做一类事情，例如：

- LLM 步骤：分类、推理、生成；
- 数据步骤：搜索、读取数据库、加载客户资料；
- 动作步骤：发邮件、写数据库、创建工单；
- 人工步骤：审批、补充信息、纠正输出。

每个节点应有清晰输入、输出、失败模式和是否允许重试的定义。

### 2. 决定哪些路径由代码控制，哪些由模型决定

- 固定顺序、权限、金额阈值、审计要求：用确定性边或代码判断；
- 意图识别、工具选择、未知子任务拆分：可以交给模型；
- 不应让模型控制的内容仍由服务端控制，例如权限、租户、真实资源 ID 和最终写操作。

### 3. 设计共享 State

State 是节点间共享的原始业务数据，而不是拼好的提示词。只存：

- 后续步骤无法重建的输入；
- 多个下游节点都要使用的结果；
- 重新获取成本高的数据；
- 恢复、审计或人工复核需要的数据。

能推导出来的内容在节点内按需计算。提示词应在调用模型时格式化，不要写入 State。这样提示词调整不会破坏已有 checkpoint，调试时也能看清真实数据。

### 4. 按失败责任设计节点

| 错误类型 | 处理者 | 推荐策略 |
|---|---|---|
| 网络抖动、限流、暂时性 5xx | 系统 | RetryPolicy，指数退避与抖动 |
| 工具失败、模型可修正的格式错误 | 模型 | 把错误写回状态并重新规划 |
| 缺少账号、审批或澄清信息 | 用户 | interrupt 暂停并等待输入 |
| 重试耗尽但可补偿 | 工作流 | error_handler 进入补偿分支 |
| 未知程序错误 | 开发者 | 向外抛出，保留 trace 与 checkpoint |

节点越小，checkpoint 越细，失败后重复工作越少，单元测试和观测也越容易；但过度拆分会增加状态接口与图复杂度。外部服务、不同重试策略和需要单独观测的步骤通常值得拆开。

### 5. 再连接图

节点完成工作，边决定下一步。动态路由可以放在 conditional edge 中，也可以由节点返回 Command 同时更新状态与决定去向。人工暂停要求父图配置 checkpointer，并在调用时提供稳定的 thread_id。

## Workflow 与 Agent 的边界

- **Workflow**：路径由代码预先规定，强调可预测、可测试和稳定顺序。
- **Agent**：模型动态决定工具与下一步，适合问题与解法都不可完全预知的场景。

实际生产系统通常是混合体：外层是确定性的工作流边界，局部节点内部允许 agent 自主选择工具。

### 常见编排模式

| 模式 | 结构 | 适用场景 | 主要风险 |
|---|---|---|---|
| Prompt chaining | 上一步输出进入下一步 | 可拆分且每步可验证的任务 | 上游错误逐步放大 |
| Parallelization | 多个独立分支并行后汇总 | 降低延迟、多视角评分 | 合并顺序与并发限额 |
| Routing | 先分类再进入专用路径 | 售前、退款、技术支持等分流 | 分类错误导致错误能力边界 |
| Orchestrator-worker | 动态拆任务、派发 worker、汇总 | 子任务数量事先未知 | 成本失控、重复任务、汇总偏差 |
| Evaluator-optimizer | 生成、评价、反馈、迭代 | 有明确质量标准但需多轮改进 | 无终止条件形成死循环 |
| ReAct agent | 模型与工具持续反馈循环 | 解法不可预先确定 | 工具权限、循环上限、提示注入 |

Orchestrator-worker 的动态 fan-out 使用 Send：每个 worker 获得自己的输入状态，结果通过 reducer 聚合到共享键。ToolNode 是常用的预构建工具执行节点，负责并行工具调用、错误处理和状态注入；工具只能看到传给 ToolNode 的状态。

## Graph API 与 Functional API

两套 API 共享同一个 Pregel 运行时，可以在同一应用里互相调用，也都支持 persistence、streaming、HITL 和 memory。差异主要是表达方式，不是能力等级。

| 维度 | Graph API | Functional API |
|---|---|---|
| 控制流 | 显式 nodes、edges、shared state | Python if、for、函数调用 |
| 状态 | 显式 schema 与 reducer | 函数作用域状态，previous 保存跨调用值 |
| checkpoint | 每个 super-step 形成新 checkpoint | task 结果写入 entrypoint 对应 checkpoint |
| 可视化 | 静态结构可直接画图 | 运行时动态生成，不支持完整静态图 |
| 并行与汇合 | 分支、Send、reducer 表达清晰 | 并发启动 task future 后统一等待 |
| 改造成本 | 需要把流程重构为图 | 对既有过程式代码改动较小 |
| 团队协作 | 节点接口和图结构适合多人协作 | 适合局部、线性或快速原型 |

### 何时选 Graph API

- 多个决策点与复杂条件分支；
- 并行路径需要汇合；
- 多组件共享状态；
- 需要可视化、审计或团队共同维护；
- 希望节点级别配置超时、重试、缓存与错误处理。

### 何时选 Functional API

- 给既有过程式代码增加持久化、流式和 HITL；
- 线性流程或少量简单分支；
- 状态天然局限在函数内；
- 快速验证想法，避免先定义完整图 schema。

### 组合与迁移

- Graph 节点可以调用 Functional entrypoint；
- Functional entrypoint 可以调用已编译 graph；
- 当过程式流程出现多处分支、共享状态或可视化需求时，从 Functional 拆为 StateGraph；
- 当一个图只是线性包装且节点划分没有独立恢复价值时，可收敛为 Functional API。

## Functional API 的执行契约

### 两个原语

- **@entrypoint**：工作流入口，只接受一个位置参数；多参数应包装为字典。通常传入 checkpointer 才能启用持久化和 HITL。
- **@task**：可并发、可重试、结果可 checkpoint 的离散工作单元。task 只能从 entrypoint、其他 task 或 StateGraph node 内调用，不能从应用主代码直接调用。

task 调用立即返回 future；同步代码用 result，异步代码用 await 获取结果。

### 哪些工作必须放进 task

- API、数据库写入、发邮件等副作用；
- 当前时间、随机数等非确定性输入；
- 长耗时且恢复时不应重复计算的操作；
- 希望并行、独立重试或单独观测的操作；
- interrupt 前后会受 replay 影响的工作。

entrypoint、task 的输入输出必须可序列化；以 JSON 兼容的字典、列表、字符串、数字和布尔值为主。

### replay 不是从暂停代码行继续

恢复时 Functional API 会从 entrypoint 开头重放，但已完成的 task 和 subgraph 结果从 checkpoint 读取，不重新执行。由此得到三个硬约束：

1. task 与 interrupt 的顺序不能随意调整；
2. 非确定性控制流必须由输入或已 checkpoint 的 task 结果决定；
3. 一个未完成的 task 可能再次执行，写操作仍必须具备幂等键或查重逻辑。

@entrypoint.final 可以把“返回给调用者的值”和“保存到 previous、供下次调用使用的值”分开。

### Functional API 已覆盖的能力

- task 并行执行；
- 调用 graph 或其他 entrypoint；
- 自定义 streaming；
- retry、async-only timeout 和 task cache；
- 从错误恢复；
- interrupt、工具调用审批；
- previous 形式的短期记忆；
- Store 形式的跨 thread 长期记忆。

## Pregel 运行时模型

StateGraph.compile 与 @entrypoint 最终都会生成 Pregel 实例。运行时把 actor 与 channel 组合起来，采用 Bulk Synchronous Parallel 模型。

每个 super-step 有三个阶段：

1. **Plan**：找出本轮因输入或上轮 channel 更新而被激活的 actor；
2. **Execution**：并行执行本轮 actor；本轮写入对同轮其他 actor 不可见；
3. **Update**：统一把写入应用到 channel，然后规划下一轮。

当没有 actor 被激活，或达到最大 step 数时结束。这解释了：

- 同一 super-step 的节点天然并行；
- reducer 在汇合并行写入时决定如何合并；
- checkpoint、time travel 和恢复都以 super-step 边界为核心；
- recursion_limit 统计的是 super-step，不是单个节点调用次数。

### Channel 类型

| Channel | 行为 | 典型用途 |
|---|---|---|
| LastValue | 新值覆盖旧值 | 普通输入、输出与步间传值 |
| Topic | 发布订阅，可累积或去重 | 多生产者、多值传播 |
| BinaryOperatorAggregate | 用二元操作持续聚合 | 计数、总和、累积结果 |
| DeltaChannel | 每步只存增量，读取时重建 | 长对话消息等持续增长字段 |

DeltaChannel 从 LangGraph 1.2 起为 beta。其 bulk reducer 必须满足结合律且为纯函数；稳定 ID 应在写入前生成，不能在重建 reducer 中生成。设置 snapshot_frequency 可以用额外完整快照换取有界读取延迟。已有 thread 写入 DeltaChannel 格式后，旧版本运行时无法直接读取，回滚前必须迁移或丢弃相关 checkpoint。

## Quickstart 背后的最小闭环

官方计算器示例的 Graph API 结构是：

1. messages 与调用次数构成 State；
2. model node 决定回答还是发起 tool call；
3. tool node 执行工具并返回 ToolMessage；
4. conditional edge 在 tool node 与 END 之间选择；
5. 工具结果返回 model node，形成循环。

Functional API 实现同一闭环时，把工具和模型调用包成 task，在 entrypoint 内使用普通 while 与 if 控制循环。二者行为相同，差别只在结构表达。

## 选型速查

1. 能用 LangChain create_agent 满足吗？能则先用高层 API。
2. 是否需要显式图、并行汇合、多处条件分支或节点级恢复？是则选 Graph API。
3. 是否是在既有 Python 流程上补 durable execution？是则先选 Functional API。
4. 是否存在外部写操作、人工暂停或恢复？无论 API，都先设计幂等与 checkpoint。
5. 是否需要长对话、高频写入大字段？评估 DeltaChannel，但把 beta 格式与回滚成本写进发布方案。

## 关联笔记

- [[02-Graph-API状态与控制流]]
- [[03-持久化记忆与容错]]
- [[04-流式HITL与子图]]
- [[05-生产化测试迁移与案例]]

## 来源

- [LangGraph overview](https://docs.langchain.com/oss/python/langgraph/overview)
- [Install LangGraph](https://docs.langchain.com/oss/python/langgraph/install)
- [Quickstart](https://docs.langchain.com/oss/python/langgraph/quickstart)
- [Thinking in LangGraph](https://docs.langchain.com/oss/python/langgraph/thinking-in-langgraph)
- [Workflows and agents](https://docs.langchain.com/oss/python/langgraph/workflows-agents)
- [Choosing between Graph and Functional APIs](https://docs.langchain.com/oss/python/langgraph/choosing-apis)
- [Functional API overview](https://docs.langchain.com/oss/python/langgraph/functional-api)
- [Use the functional API](https://docs.langchain.com/oss/python/langgraph/use-functional-api)
- [LangGraph runtime](https://docs.langchain.com/oss/python/langgraph/pregel)