Liam 项目 LangGraph 状态管理实战从 Annotation 状态 Schema 到生产级 Checkpoint 持久化【免费下载链接】liamAutomatically generates beautiful and easy-to-read ER diagrams from your database.项目地址: https://gitcode.com/GitHub_Trending/li/liam本文以docs/langgraph/state-management.md为核心骨架结合 liam 开源仓库一个自动从数据库生成美观易读 ER 图的工具中frontend/internal-packages/agent的真实源码系统讲解 LangGraphJS 的状态定义、Schema 分离、私有状态传递、reducer 合并语义与 Checkpoint 持久化并给出可直接复制的 TypeScript 示例与生产级落地参考。导读状态State是 LangGraph 图的血液——每个节点读写状态、每条边传递状态、每次编译与恢复都围绕状态展开。本文围绕docs/langgraph/state-management.md中State Schema DesignState Updates and MergingState Persistence Patterns三大主题先讲清Annotation定义状态、输入/输出 Schema 分离、节点间私有状态传递这三种核心模式再深入 reducer 合并语义与状态校验最后从MemorySaver到 PostgreSQL 再到 liam 项目自研的SupabaseCheckpointSaver完整覆盖从原型到生产的状态持久化路径。读完你可以直接在自己的 LangGraphJS 项目中落地一套类型安全、可持久化、可恢复的多 Agent 工作流状态方案。1. State Schema Design用 Annotation 定义图状态1.1 Annotation.Root状态 Schema 的唯一入口LangGraphJS 中图的状态结构通过Annotation函数声明式定义。Annotation.Root创建顶层状态对象其中的每个字段都由AnnotationT()描述其类型、合并策略reducer与默认值。这也是 liam 项目所有 Agent 图lead-agent、pm-agent、db-agent、qa-agent统一采用的状态定义方式。最小示例来自 state-management.mdimport { BaseMessage } from langchain/core/messages; import { Annotation } from langchain/langgraph; const GraphAnnotation Annotation.Root({ messages: AnnotationBaseMessage[]({ reducer: (currentState, updateValue) currentState.concat(updateValue), default: () [], }), });要点拆解reducer合并函数决定当多个节点或一个节点多次运行向同一字段写入时如何合并。上面的concat是消息列表的经典写法新消息追加到历史消息之后。default默认值工厂字段在状态中尚不存在时的初始值注意它是一个返回值的函数避免所有实例共享同一个可变引用。TypeScript 泛型AnnotationBaseMessage[]使每个字段在节点签名中拥有精确类型配合StateGraph的泛型推导编译器能在图构建期拦截字段拼写错误与类型不匹配。1.2 liam 项目中的真实状态 Schema打开 workflowAnnotation.tsfrontend/internal-packages/agent/src/workflowAnnotation.ts可以看到一个更完整、更贴近生产的状态定义它演示了 LangGraphJS 官方推荐的MessagesAnnotation展开写法以及reducer 覆盖写和默认值 END两个进阶技巧import { Annotation, END, MessagesAnnotation } from langchain/langgraph import type { Schema } from liam-hq/schema import type { AnalyzedRequirements } from ./schemas/analyzedRequirements import { workflowSchemaIssuesAnnotation } from ./workflowSchemaIssuesAnnotation export const workflowAnnotation Annotation.Root({ ...MessagesAnnotation.spec, analyzedRequirements: AnnotationAnalyzedRequirements({ reducer: (x, y) y ?? x, default: () ({ goal: , testcases: {}, }), }), schemaData: AnnotationSchema, organizationId: Annotationstring, userId: Annotationstring, designSessionId: Annotationstring, schemaIssues: workflowSchemaIssuesAnnotation, next: Annotationstring({ reducer: (x, y) y ?? x ?? END, default: () END, }), })从源码结构可以提炼出几个高频设计决策...MessagesAnnotation.spec展开MessagesAnnotation是 LangGraphJS 内置的、自带reducer的消息通道。通过展开其spec可以把它作为基座再叠加业务字段避免手写消息 reducer。reducer: (x, y) y ?? x这是一个覆盖式reducer——新值非空时直接替换旧值否则保留旧值。它适用于analyzedRequirements这种整块替换而非逐条追加的业务对象。reducer: (x, y) y ?? x ?? END路由字段next的合并策略同时承担默认值END。配合addConditionalEdges(leadAgent, (state) state.next, ...)实现基于状态的动态路由见 createGraph.ts。default: () ({ goal: , testcases: {} })业务对象必须给出确定的初始形状保证第一个节点读取时不会遇到undefined。值得注意liam 团队把 reducer 语义写进了 workflowSchemaIssuesAnnotation.ts 的注释里——同一字段在不同图中可能采用不同的合并策略这正是理解状态合并的关键。1.3 reducer 语义的同字段不同策略对比在 liam 中schemaIssuesSchema 问题列表在工作流级与QA Agent 级使用了完全相反的 reducer 策略这是一个非常有教学价值的对照实验场景reducer语义原因工作流级workflowSchemaIssuesAnnotation(prev, next) next ?? prev替换新值优先工作流需要在 DB Agent 处理后清空问题列表schemaIssues: []若用 concat[].concat(prev)等于旧值永远清不掉QA Agent 级concat追加并行收集QA 阶段需要把多个测试用例产生的问题累积起来再统一处理注释中明确记录了这一决策workflowSchemaIssuesAnnotation.ts/** * Uses a replacement reducer instead of concat because: * - Workflow needs to clear issues after DB agent processing * - Setting schemaIssues: [] should actually clear the array * - With concat reducer, [] would be concatenated (prev.concat([]) prev) * * This is different from QA agents annotation which uses concat * for parallel issue collection. */ export const workflowSchemaIssuesAnnotation AnnotationArraySchemaIssue({ reducer: (prev, next) next ?? prev, default: () [], })工程启示不要盲目对数组一律使用concat。先问这个字段被谁写、需要追加还是替换、是否会被显式清空再决定 reducer 的语义。concat 适合消息历史这类只增不改的数据替换式 reducer 适合整块重算或需要支持清空的数据。2. Input/Output Schema 分离让图的边界更清晰2.1 基础写法LangGraphJS 允许为同一张图定义独立的输入与输出 Schema外部调用者只向输入 Schema 字段传参图的最终输出只暴露输出 Schema 字段中间字段对调用者完全不可见。这样既提升了类型安全也保护了内部实现细节。来自 state-management.md 的示例const InputAnnotation Annotation.Root({ question: Annotationstring(), }); const OutputAnnotation Annotation.Root({ answer: Annotationstring(), }); const GraphAnnotation Annotation.Root({ ...InputAnnotation.spec, ...OutputAnnotation.spec, });将输入字段与输出字段分别定义为独立的Annotation.Root对象再用展开语法合并为图的完整状态。StateGraph(GraphAnnotation)之后调用graph.invoke({ question: ... })时编译器只允许你传输入字段读取result时也只提示输出字段。2.2 输入 Schema 的实战作用控制子图入参在 liam 中Input/Output Schema 分离的思想还被用于精确控制子图节点的入参范围。在 createGraph.ts 中主工作流调用 pm-agent 子图时只挑选必要字段传入const callPmAgent async (state: WorkflowState, config: RunnableConfig) { const output await pmAgentSubgraph.invoke( { messages: state.messages, analyzedRequirements: state.analyzedRequirements, designSessionId: state.designSessionId, schemaData: state.schemaData, analyzedRequirementsRetryCount: 0, }, config, ) return { ...state, ...output } }同样调用 db-agent 与 qa-agent 时分别构造modifiedState清空messages、注入prompt、清空schemaIssues。这种调用侧显式挑选字段 子图内部独立 Annotation的组合本质上是把 Input/Output Schema 分离原则应用到了子图边界父图只暴露子图需要的输入子图只返回父图关心的输出再通过return { ...state, ...output }合并回主状态。3. Private State Between Nodes节点间私有状态传递有些中间结果如搜索 query、检索到的文档、临时的 SQL 片段只在少数节点之间流转不需要出现在图的输入/输出 Schema 中。LangGraphJS 提供了两种私有状态方案节点级input注解与子图隔离。3.1 节点级 input 注解RAG 管道示例原文档给出了一个非常经典的 RAG 管道示例——生成 query → 检索文档 → 生成答案三个阶段每个中间节点的输出只被下游节点消费最终输出只有answerimport { Annotation, StateGraph } from langchain/langgraph; // 图的整体状态只暴露 question 与 answer const OverallStateAnnotation Annotation.Root({ question: Annotationstring, answer: Annotationstring, }); // 生成 query 的节点输出 const QueryOutputAnnotation Annotation.Root({ query: Annotationstring, }); // 检索文档的节点输出 const DocumentOutputAnnotation Annotation.Root({ docs: Annotationstring[], }); // 生成答案的节点输入 整体状态 文档 const GenerateOutputAnnotation Annotation.Root({ ...OverallStateAnnotation.spec, ...DocumentOutputAnnotation.spec }); // 节点 1生成 query const generateQuery async (state: typeof OverallStateAnnotation.State) { return { query: state.question rephrased as a query!, }; }; // 节点 2检索文档输入仅 QueryOutputAnnotation const retrieveDocuments async (state: typeof QueryOutputAnnotation.State) { return { docs: [state.query, some random document], }; }; // 节点 3生成答案 const generate async (state: typeof GenerateOutputAnnotation.State) { return { answer: state.docs.concat([state.question]).join(\n\n), }; }; const graph new StateGraph(OverallStateAnnotation) .addNode(generate_query, generateQuery) .addNode(retrieve_documents, retrieveDocuments, { input: QueryOutputAnnotation }) .addNode(generate, generate, { input: GenerateOutputAnnotation }) .addEdge(__start__, generate_query) .addEdge(generate_query, retrieve_documents) .addEdge(retrieve_documents, generate) .compile();关键点addNode(name, node, { input: SomeAnnotation })声明该节点的输入范围LangGraph 运行时会自动把图中对应的字段注入节点参数query、docs不会出现在最终输出里——intermediate states populated by the input annotations are not present in the final output主 Schema 保持干净state-management.md每个节点的参数类型由各自的 Annotation 推导比如retrieveDocuments只能访问query无法误读answer。3.2 子图作为私有状态容器当中间状态规模较大、需要自己的节点编排甚至持久化时更优雅的做法是把它们放进子图。liam 的主工作流正是这一模式的规模化实践createGraph.ts 中四个 Agent 各自拥有独立的Annotation.Root与独立的StateGraphconst leadAgentSubgraph createLeadAgentGraph() const pmAgentSubgraph createPmAgentGraph() const dbAgentSubgraph createDbAgentGraph() const qaAgentSubgraph createQaAgentGraph()子图内部的状态字段如 QA 的测试用例、DB 的生成 SQL天然不会泄漏到父图 Schema父图只通过invoke时的显式传参与返回值交换数据。从源码结构可以推断这种父图负责编排、子图负责领域逻辑与私有状态的层次划分是管理多 Agent 复杂状态的关键架构决策。4. State Updates and Merging更新合并与状态校验4.1 状态校验把 Annotation 交给 StateGraphAnnotation.Root产出的对象同时携带类型信息TypeScript 层面与运行时校验能力LangGraph 内部会依据 Annotation 对节点返回值做合并。将 Annotation 传入构造函数即可获得全程类型安全import { StateGraph } from langchain/langgraph; const workflow new StateGraph(GraphAnnotation);在 liam 中createGraph的完整构建链展示了Annotation 贯穿始终的写法createGraph.tsconst graph new StateGraph(workflowAnnotation) .addNode(validateInitialSchema, validateInitialSchemaNode) .addNode(leadAgent, leadAgentSubgraph) .addNode(pmAgent, callPmAgent, { subgraphs: [pmAgentSubgraph] }) .addNode(dbAgent, callDbAgent, { subgraphs: [dbAgentSubgraph] }) .addNode(qaAgent, callQaAgent, { subgraphs: [qaAgentSubgraph] }) .addConditionalEdges(START, (state) { ... }) .addConditionalEdges(leadAgent, (state) state.next, { ... }) ... .compile()4.2 合并时机与清空陷阱理解 reducer 的合并时机是排查状态 bug 的钥匙。LangGraph 的合并发生在每个节点返回之后节点返回值会按照该字段的 reducer 与当前状态合并形成新的 channel 值并触发一次 checkpoint若配置了 checkpointer。前面 1.3 节提到的schemaIssues: []清空问题就是一个典型的合并时机 reducer 语义共同决定的边界情况——如果使用 concatprev.concat([]) prev数组永远清不掉工作流会在 DB Agent 处理完后带着陈旧问题反复循环换成替换式 reducer 后[]才能真正清空数组从而终止循环createGraph.ts 中schemaIssues: []的注释明确写着 Clear schemaIssues after DB agent processing to prevent infinite loops。5. State Persistence Patterns从内存到生产级持久化持久化是 LangGraph 状态管理的最后一块拼图。通过checkpointer把每个节点执行后的状态快照checkpoint落盘图就可以跨多次invoke恢复对话thread 级持久化、支持时间旅行、支持断点续跑。5.1 Thread-level Persistence内存版原文档给出的最小实现适合原型验证import { MemorySaver } from langchain/langgraph; const checkpointer new MemorySaver(); const graph workflow.compile({ checkpointer });调用时通过config.configurable.thread_id标识会话await graph.invoke(input, { configurable: { thread_id: thread-1 } });同一个thread_id的后续调用会自动从最新 checkpoint 继续。5.2 PostgreSQL Persistence生产首选官方PostgresSaver一行接入import { PostgresSaver } from langchain/langgraph-checkpoint-postgres; const checkpointer PostgresSaver.fromConnString(postgresql://...); const graph workflow.compile({ checkpointer });liam 的createGraph也把 checkpointer 设计为可选注入参数默认不持久化、传入即启用createGraph.tsexport const createGraph (checkpointer?: BaseCheckpointSaver) { // ...构建图... return checkpointer ? graph.compile({ checkpointer }) : graph.compile() }5.3 生产级实战SupabaseCheckpointSaverliam 没有直接使用官方 PostgresSaver而是基于langchain/langgraph-checkpoint的BaseCheckpointSavernumber自研了 SupabaseCheckpointSaver其核心设计与踩坑点非常值得借鉴1多租户数据隔离构造时注入organizationId所有查询都带.eq(organization_id, this.organizationId)过滤如getTuple、list中对三张表的一致过滤保证不同组织的 checkpoint 互不可见。type SupabaseCheckpointSaverOptions { organizationId: string // Data isolation for multi-tenancy }2三表存储模型checkpoint 本体存checkpoints表channel_versions、versions_seen、metadata、parent_checkpoint_idchannel 值按版本拆分存checkpoint_blobs待写入的 task 结果存checkpoint_writes。读取时需分别查询三张表再组装CheckpointTuple写入时通过 RPCput_checkpoint原子插入 checkpoint 与 blobsSupabaseCheckpointSaver.ts。3BYTEA 双重编码Supabase REST API 对 BYTEA 列有特殊行为——写入前需把序列化数据转 Base64读取时 Supabase 返回的是 PostgreSQL hex 格式如\x6465...。因此实现里用hexToUint8Array/uint8ArrayToBase64做双向转换SupabaseCheckpointSaver.ts注释明确记录了数据流Write: JSON → serialize → Base64 → BYTEARead: BYTEA (hex) → decode hex → Base64 → deserialize → JSON。4兼容 LangGraph 0.4.x 的deleteThread批量删除checkpoints/checkpoint_writes/checkpoint_blobs三表数据SupabaseCheckpointSaver.ts用于会话清理。5pending sends 迁移_maybeMigratePendingSends会把旧版本 checkpoint 中位于checkpoint_writes的 pending sends 恢复到channel_values[TASKS]保证老 checkpoint 仍可被新版本运行时正确恢复SupabaseCheckpointSaver.ts。该实现有配套的单测与集成测试SupabaseCheckpointSaver.test.ts 覆盖三表读写、BYTEA 编解码、deleteThread等SupabaseCheckpointSaver.integration.test.ts 在真实数据库上验证线程级持久化与恢复。工程启示如果要在生产环境接入 Supabase/REST 类后端直接使用官方 PostgresSaver 会因 BYTEA 编码差异而失败。参考 liam 的做法——继承BaseCheckpointSaver自定义getTuple/list/put/putWrites四个核心方法并针对数据源的二进制列编码做适配层。5.4 持久化与 Map-Reduce 的协同持久化不只服务对话续接还支撑 liam QA Agent 的map-reduce 并行模式。在 createQaAgentGraph.ts 中START通过SendAPI 并行分发多个testcaseGeneration子任务任务结果写入 checkpoint 后汇总到applyGeneratedSqls。这正是 LangGraph checkpoint 的底层能力——每个并行任务的状态写入都独立持久化失败时可精确恢复到未完成任务。6. 实践清单与常见陷阱围绕本文内容给出可直接对照的落地清单定义状态一律用Annotation.Root业务字段显式标注类型、reducer、default需要消息通道时直接展开MessagesAnnotation.spec。选择 reducer 语义只增不改用concat整块替换用(x, y) y ?? x需要支持清空的数组绝不能只用 concat。分离输入输出把输入字段与输出字段拆成独立Annotation.Root用展开语法合并子图边界同样只传必要字段。私有状态节点间临时数据用addNode(..., { input: PrivateAnnotation })限定输入范围或直接下沉到子图保持主 Schema 干净。状态校验把完整 Annotation 传给new StateGraph(annotation)让编译器与运行时同时把关。持久化原型用MemorySaver生产优先 PostgreSQL 系 checkpointer接入 Supabase 等 REST 后端时注意 BYTEA 双重编码与多租户隔离可参考SupabaseCheckpointSaver。线程续接invoke时始终携带{ configurable: { thread_id } }同一个 thread 内多次调用自动基于最新 checkpoint 继续。高频陷阱提醒数组字段一律concat导致清不掉liam 在workflowSchemaIssuesAnnotation注释中记录了此坑default直接写对象字面量而非工厂函数导致多实例共享引用忘记展开MessagesAnnotation.spec而手写消息 reducer容易丢掉内置的 message 去重与类型支持在 Supabase REST 场景直接套官方 PostgresSaverBYTEA 读取失败。7. 延伸阅读关联文档state-management.md本文的原始骨架LangGraph 系列文档README.md、core-concepts.md、control-flow.md、advanced-features.md、multi-agent.md、streaming.md、tool-calling.md状态 Schema 源码workflowAnnotation.ts、workflowSchemaIssuesAnnotation.ts图编排源码createGraph.ts、createQaAgentGraph.ts持久化源码与测试SupabaseCheckpointSaver.ts、SupabaseCheckpointSaver.test.ts、SupabaseCheckpointSaver.integration.test.ts【免费下载链接】liamAutomatically generates beautiful and easy-to-read ER diagrams from your database.项目地址: https://gitcode.com/GitHub_Trending/li/liam创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考