函数式 API 总览 (Functional API overview)
LangGraph 函数式 API 允许开发者以极简代码侵入度，在常规函数中直接赋予应用程序持久化执行、记忆系统、人在回路审核与流式传输能力。
函数式 API（Functional API）旨在将 LangGraph 的四大核心基础设施能力——持久化执行 (Persistence)、记忆系统 (Memory)、人在回路 (Human-in-the-loop)以及流式传输 (Streaming)——以最低的侵入度融入已有代码中。
它专门面向使用标准编程语言控制流（如 if 分支条件判断、for 循环与常规异步函数调用）的业务系统。与许多强制要求将代码重构为显式流水线或刚性 DAG 的编排框架不同，函数式 API 允许开发者自由编写原生代码逻辑，同时天然获得工业级的工作流治理能力。
函数式 API 构建在两大基石原语之上：
entrypoint：工作流入口。它封装顶层工作流逻辑并负责整个执行流的生命周期调度，管理长耗时任务的检查点快照与人在回路中断挂起。task：任务单元。代表一个离散的工作执行步骤（例如外部 API 调用或特定数据清洗逻辑）。Task 产生类似 Future 的延迟对象，其执行结果会被自动持久化缓存；当流程从中断中恢复重跑时，已完成的 Task 会直接从检查点读取缓存结果，避免重复执行造成副作用或高昂开销。
Functional API 与 Graph API 的对比
| 对比维度 | 函数式 API (Functional API) | 图 API (Graph API) |
|---|---|---|
| 控制流表达 | 原生代码命令式：使用常规的 if/else、for 循环与函数调用 | 声明式状态图：使用节点（Node）与流转边（Edge/ConditionalEdge） |
| 状态维护机制 | 函数作用域局部变量，无需定义显式 State 或 Reducer 聚合函数 | 严格定义全局 StateSchema，通过通道与 Reducer 管理跨节点状态更新 |
| 检查点机制 | Task 执行完成后将其结果动态增量追加至当前 Entrypoint 的检查点中 | 图中每个超步（Super-step）执行完毕后生成一个完整的全局快照检查点 |
| 架构可视化 | 运行时动态调度，不生成静态图架构图 | 天然支持将整个工作流编译为拓扑图，便于可视化展示、追踪与分享 |
| 适用场景 | 快速改造已有业务函数、串行为主或轻量交互的智能体 | 复杂的循环重试状态机、需要图拓扑可视化审查的生产级工作流 |
快速实战示例 (Example)
以下示例演示一个自动撰写文章、触发人在回路审批中断并根据人类审核结果继续流转的工作流：
typescript
import { MemorySaver, entrypoint, task, interrupt, Command } from "@langchain/langgraph";
// 1. 定义耗时任务单元
const writeEssay = task("writeEssay", async (topic: string) => {
// 模拟耗时生成操作
await new Promise((resolve) => setTimeout(resolve, 1000));
return `An essay about topic: ${topic}`;
});
// 2. 定义工作流入口，挂载检查点持久化
const workflow = entrypoint(
{ checkpointer: new MemorySaver(), name: "workflow" },
async (topic: string) => {
// 执行任务并缓存结果
const essay = await writeEssay(topic);
// 触发人在回路中断挂起，等待人工审核
const isApproved = interrupt({
essay,
action: "请审核该文章是否可以发布 (approve/reject)",
});
return {
essay,
isApproved,
};
}
);深入执行过程解析
- 初始触发与中断挂起： 通过配置会话
thread_id启动工作流。智能体执行writeEssay任务并持久化文章结果。随后代码执行到interrupt()，整个流程被安全挂起，返回中断信号并等待人工介入。 - 断点恢复重跑与结果缓存复用： 当审核人员在管理后台或客户端做出决策后，传入
new Command({ resume: true })恢复执行。此时工作流函数从头部重新进入，但由于writeEssay的执行结果此前已被写入检查点，该任务会瞬间直接命中缓存返回，而绝不会重复发起二次模型调用。interrupt()接收到人工传入的true并向下返回最终产物。
工作流入口 (Entrypoint)
使用 entrypoint 可以将任意普通异步函数升级为具备状态机治理能力的工作流实例：
定义规范
传入配置对象（包含名称与检查点实现）以及一个业务执行函数。业务函数必须仅接收一个形参作为工作流输入（如有多个参数，请打包为一个对象传入）。
typescript
import { entrypoint } from "@langchain/langgraph";
const myWorkflow = entrypoint(
{ checkpointer, name: "myWorkflow" },
async (input: { query: string; userId: string }) => {
// 业务编排逻辑...
return result;
}
);序列化要求
为了支持状态检查点与跨服务断点恢复，Entrypoint 的输入参数与输出返回值必须是 JSON 可序列化的数据结构。
运行与流式调用
与 Graph API 一致，编译后的 Entrypoint 实例提供标准的 invoke 与 streamEvents 方法：
typescript
const config = {
configurable: {
thread_id: "thread_123"
}
};
// 同步等待执行终态
const result = await myWorkflow.invoke({ query: "LangChain" }, config);
// 流式消费事件
const stream = await myWorkflow.streamEvents({ query: "LangChain" }, config, { version: "v3" });
for await (const message of stream.messages) {
for await (const token of message.text) {
process.stdout.write(token);
}
}断点恢复 (Resuming)
当流程遭遇 interrupt 挂起后，向 Command 传入 resume 载荷唤醒工作流：
typescript
import { Command } from "@langchain/langgraph";
const resumeConfig = {
configurable: {
thread_id: "thread_123"
}
};
const finalResult = await myWorkflow.invoke(
new Command({ resume: { approved: true, reviewerNotes: "通过" } }),
resumeConfig
);