程序员小白必看：轻松掌握子图构建与使用，提升大模型开发效率

本文详细介绍了如何在大模型开发中建立和使用子图将复杂流程封装为独立节点。通过客服知识检索与回复生成的实例阐述了子图的拆分依据、状态契约、通信方式及持久化范围。文章强调子图设计需关注输入输出独立性、可测试性和内部流程并提供两种通信方式供选择。此外还讨论了子图状态的保存策略和嵌套流式观察方法帮助开发者更好地理解和应用子图技术提升代码组织和运行效率。本章以“客服知识检索与回复生成”为主线依次说明子图的拆分依据、状态契约、两种通信方式、持久化范围以及嵌套流式。环境基线Python 3.13、langchain1.3.2、LangGraph 1.x。示例使用确定性节点便于观察状态变化将节点内部实现替换为模型或检索器后子图结构不需要改变。一、何时拆成子图假设客服流程包含以下步骤规范化问题 → 改写检索词 → 检索知识 → 生成回复 → 质量检查 → 必要时改写将六个节点直接放入父图可以正常运行但检索逻辑、回复逻辑和全局编排会共享同一个 State。后续增加查询扩展、混合检索或多轮质检时父图会持续膨胀。更合理的结构是保留简单的规范化节点并将具有内部流程的能力拆成两个子图父图 ├── normalize_question 普通节点 ├── retrieve_knowledge 检索子图 │ ├── rewrite_query │ └── retrieve_documents └── generate_answer 回复子图 ├── draft_answer ├── quality_check └── revise_answer判断一段流程是否适合拆成子图可以检查三个条件具有明确的输入输出。 调用方不需要了解内部节点只需要提供输入并接收结果。能够独立测试或替换。 例如检索子图可以从关键词检索替换为向量检索而父图接口保持不变。内部存在有意义的流程。 多个步骤共享私有状态、重试或路由逻辑时子图可以形成稳定的模块边界。只有一个简单计算节点时通常没有必要再增加一层子图。子图层级越深检查点命名空间、Trace 和流式路径也会越长。模块化的目标是降低耦合而不是增加结构层次。二、设计子图接口子图设计的核心不是add_node而是父图与子图之间的状态契约。要先确定哪些字段属于全局业务状态哪些字段只在模块内部使用。本例的父图只维护三个公共字段from typing_extensions import TypedDict class ParentState(TypedDict): question: str evidence: list[str] answer: str检索子图使用自己的 Schemaclass RetrievalState(TypedDict): query: str rewritten_query: str documents: list[str]其中rewritten_query是检索模块的中间结果documents是模块输出。父图使用的字段名分别是question和evidence两者并不相同因此调用时需要显式映射。回复子图则与父图共享question、evidence和answer同时增加一个私有质量分数class AnswerState(TypedDict): question: str evidence: list[str] answer: str quality_score: int # 仅供回复子图内部使用公共字段构成子图接口私有字段保存实现细节。这样设计有两个好处父图不需要承载查询改写结果、质量分数等临时数据。子图可以调整内部节点和私有字段只要公共输入输出保持兼容父图就不需要修改。还需要注意 Reducer。若父图与子图共享messages、日志列表等追加型字段两层应明确使用一致的合并语义。否则子图返回列表后父图可能执行覆盖也可能再次追加。对于不需要暴露的内部消息更稳妥的做法是只向父图返回摘要或最终结果。三、父子图通信假设你有一个父图主流程和一个子图检索模块。把子图接入父图有两种方式比较如下方式怎么接数据怎么传方式一在父节点里调用subgraph.invoke把子图调用包在一个适配节点里手动 从父State取值 → 组装成子图需要的输入 → 调用子图 → 取子图输出 → 手动写入父State方式二把编译好的子图直接当节点用parent_builder.add_node(xxx, subgraph)自动 LangGraph 把父State里同名字段自动传给子图子图输出自动合并进父State核心区别是数据转换由谁来做方式一我们自己做在适配节点里显式转换。方式二框架自动做通过共享状态键自动映射。1 不同Schema当父图和子图的状态字段名完全不同或者需要对输入输出做额外处理如裁剪、校验、格式转换则使用这种模式。例如父图用question子图用query父图想要evidence子图返回documents字段名对不上需要翻译。一般这种都是子图先建立或独立建立独立模块不知道自己会被哪个父图调用# 子图有自己的 State 定义 class RetrievalState(TypedDict): query: str # 输入待检索问题 rewritten_query: str # 内部改写后的查询 documents: list[str] # 输出检索结果 retrieval_graph ... # 编译好的子图然后在父图里建一个适配节点负责把父State翻译成子图能吃的格式再把子图输出翻译回父State能用的格式def retrieve_knowledge(state: ParentState) - dict: # 1. 输入转换父图 question → 子图 query result retrieval_graph.invoke({ query: state[question], rewritten_query: , documents: [], }) # 2. 输出转换子图 documents → 父图 evidence return {evidence: result[documents]}适配节点是明确的“边界”。你在这一层可以做字段重命名question→query输入裁剪只取需要的字段不把整个父State倒过去数据校验输出摘要子图返回 100 条只取前 5 条这种方式边界清晰父子图完全解耦字段名随便改。但是缺点是每个接入点都要写样板代码比较麻烦。2 共享状态键父图和子图有公共字段且字段名完全一致不需要做转换适合直接使用这种。例如父State和子State都有question、evidence、answer三个字段父图把子图当“一个步骤”插入流程。这样的话我们建子图时 State 里必须包含和父图共用的字段class AnswerState(TypedDict): question: str # 父图也有的字段 evidence: list[str] # 父图也有的字段 answer: str # 父图也有的字段 quality_score: int # 子图私有字段父图没有子图写完后编译成answer_graph。然后直接把它当节点添加到父图里不需要适配函数parent_builder.add_node(generate_answer, answer_graph)LangGraph 在运行时自动做这件事父图进入generate_answer节点时自动把父State里的question和evidence传给子图因为字段名相同子图执行完后自动把子State里的answer合并回父State子图的quality_score不属于父State被自动忽略不会污染父State为什么需要共享字段因为子图执行时它的输入数据必须从父State来。如果父State里没有question和evidence子图就拿不到值。所以“共享状态键”本质是用同名字段作为数据通道。3 两者对比维度方式一适配节点方式二直接节点父子State是否有同名字段不需要全靠手动转换必须 有同名字段框架靠它自动传数据数据如何传递手动subgraph.invoke()框架自动映射同名键私有字段处理适配节点里只传需要的字段子图私有字段父图没有自动忽略代码量较多每个子图要写适配函数较少一行add_node解耦程度高字段名随便改低字段名必须对齐牵一发动全身适用场景子图被多个父图复用、字段名不一致、需要额外处理子图专为当前父图设计、字段名天然一致选型建议子图会被多个不同父图复用且各父图的字段命名不统一 → 方式一适配节点子图专为当前父图设计字段名已经对齐 → 方式二直接节点需要在输入输出时做校验、裁剪、摘要 → 方式一适配节点追求代码简洁不需要额外处理 → 方式二直接节点四、子图状态能够保留多久子图可以有自己的 Checkpointer独立于父图。compile(checkpointer...)有三种模式如下子图配置持久化范围适用场景checkpointerNone默认单次调用内部大多数一次性子任务单次执行中可继承父图 CheckpointercheckpointerTrue同一thread_id的多次调用之间子图需要跨轮次保留内部状态如多轮研究助手checkpointerFalse不保存任何状态纯计算模块不需要中断、恢复或状态检查1 None默认模式不传checkpointer参数时子图使用默认模式。本例中的检索和回复子图都是“一次性任务”每次根据当前问题重新执行不需要记住上一次调用的内部状态retrieval_graph retrieval_builder.compile() # 等价于 checkpointerNone answer_graph answer_builder.compile()默认模式下子图不会把上一次调用产生的私有状态带入下一次调用。但有一个例外如果父图配置了 Checkpointer子图的单次执行仍可通过父图参与中断恢复例如父图 HITL 暂停后子图未完成的部分可以在恢复时继续执行。适用场景一次性查询、独立计算任务、任何“每次重新开始”的子任务。2 True按线程保存内部状态传入checkpointerTrue时子图在同一thread_id的多次调用之间累积内部状态。answer_graph answer_builder.compile(checkpointerTrue)比如一个多轮研究助手子图需要在多次用户提问之间保留“已经搜过哪些关键词”“已经分析过哪些文档”这些信息属于子图内部私有状态父图不需要知道细节但子图自己必须记住。适用场景多轮对话子图、编码助手保留文件修改历史、持续任务跟踪器。⚠️ 重要限制同一个子图实例不能在同一步中并行调用多次。原因是多个并行调用会写入同一个检查点命名空间checkpoint_ns导致写入冲突。如果你需要对不同数据做并行处理有两种方案使用默认模式None每次调用独立不跨调用共享状态使用不同的子图节点实例每个有自己的命名空间简单理解checkpointerTrue的子图就像一个“有状态的服务”同一时间只能处理一个请求。3 False不保存状态传入checkpointerFalse时子图完全不保存任何状态。pure_graph pure_builder.compile(checkpointerFalse)这个模式最轻量但代价是如果子图内部有interrupt()HITL 中断暂停后无法恢复因为状态根本没存。所以它只适合“进去→计算→出来”的纯函数式子图没有任何需要中断或恢复的逻辑。适用场景数据格式转换、校验、纯计算模块不需要中断、恢复和内部状态检查的任务。与None的区别None模式在父图有 Checkpointer 时单次执行内的中断是可恢复的False模式彻底不存中断即丢失。4 父图 Checkpointer 与子图的关系无论子图配置什么模式子图的实际持久化能力都依赖父图提供 Saverfrom langgraph.checkpoint.memory import InMemorySaver # 父图配置 Checkpointer graph parent_builder.compile(checkpointerInMemorySaver()) config {configurable: {thread_id: customer-1001}}如果父图没有配置 Checkpointer子图的持久化链路不完整checkpointerTrue也无法正常工作。生产环境请将InMemorySaver替换为 SQLite、Postgres 等持久化后端。五、如何观察子图内部执行子图在父图中表现为一个节点。只观察父图更新时可以看到generate_answer完成却无法区分内部的草拟、质量检查和改写步骤。嵌套流式用于把这些内部执行暴露给调试工具、日志系统或前端服务。1 使用 v3 事件流观察子图应用代码优先使用stream_events(..., versionv3)。它提供消息、状态、子图和最终输出等类型化投影不需要业务代码自行拆解底层事件元组。stream graph.stream_events( { question: 退款多久到账, evidence: [], answer: , }, config{configurable: {thread_id: stream-v3}}, versionv3, ) # 每个 subgraph 对象表示一次嵌套图执行 for subgraph in stream.subgraphs: print(subgraph:, subgraph.graph_name) print(path:, subgraph.path) # 本例没有调用模型因此观察子图状态变化 for value in subgraph.values: print(state:, value) final_state stream.output print(final_state[answer])如果子图内部调用聊天模型可以遍历subgraph.messages获取消息增量。path表示从根图到当前嵌套执行的路径可用于区分检索子图、回复子图及更深层的节点。服务端可以依据path和事件类型转换为稳定的业务协议例如{type: progress, module: generate_answer, stage: quality_check}不建议前端直接依赖包含运行时 ID 的原始命名空间否则图结构或节点名称调整后前端协议也需要同步修改。2 使用 v2 原始流进行底层排障当需要查看 Pregel 的原始updates、values或tasks时可以使用stream(..., versionv2)。设置subgraphsTrue后父图和子图事件采用统一的StreamPart结构for chunk in graph.stream( { question: 退款多久到账, evidence: [], answer: , }, config{configurable: {thread_id: stream-v2}}, ) stream_modeupdates, subgraphsTrue, versionv2, ): if chunk[type] ! updates: continue source parent if not chunk[ns] else chunk[ns] print(source, chunk[data])其中type表示流模式例如updates。ns为空元组时表示父图非空时表示子图路径。data保存当前事件的数据。第 6 篇已经介绍了updates、values、messages和custom等流模式本篇不再重复。嵌套场景只需记住v3 使用stream.subgraphs读取类型化投影v2 使用subgraphsTrue并根据ns判断事件来源。3 自定义进度只传递业务信息长时间运行的检索或批处理节点可以通过get_stream_writer()写入业务进度from langgraph.config import get_stream_writer def retrieve_large_index(state: RetrievalState) - dict: writer get_stream_writer() writer({stage: retrieve, progress: 0.2}) # 执行实际检索 documents [命中的知识条目] writer({stage: retrieve, progress: 1.0}) return {documents: documents}get_stream_writer()只能在图运行上下文中使用。为了便于单元测试应将检索计算与进度上报拆开避免业务函数必须依赖流式运行环境。六、完整装配、测试与验收前面的代码片段属于同一套客服流程。完整装配时父图只负责三个模块的顺序不需要了解检索查询如何改写也不需要保存回复质量分数客服知识检索与回复生成子图完整示例。 from __future__ import annotations from typing_extensions import TypedDict from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph class ParentState(TypedDict): question: str evidence: list[str] answer: str class RetrievalState(TypedDict): query: str rewritten_query: str documents: list[str] class AnswerState(TypedDict): question: str evidence: list[str] answer: str quality_score: int def rewrite_query(state: RetrievalState) - dict: query state[query].strip().replace(, ) return {rewritten_query: f客服政策 {query}} def retrieve_documents(state: RetrievalState) - dict: return { documents: [ fFAQ 命中{state[rewritten_query]}, 退款原路返回通常需要 13 个工作日。, ] } def build_retrieval_graph(): builder StateGraph(RetrievalState) builder.add_node(rewrite_query, rewrite_query) builder.add_node(retrieve_documents, retrieve_documents) builder.add_edge(START, rewrite_query) builder.add_edge(rewrite_query, retrieve_documents) builder.add_edge(retrieve_documents, END) return builder.compile() def draft_answer(state: AnswerState) - dict: evidence .join(state[evidence]) return {answer: f关于“{state[question]}”{evidence}} def quality_check(state: AnswerState) - dict: score 90 if 工作日 in state[answer] else 50 return {quality_score: score} def revise_answer(state: AnswerState) - dict: if state[quality_score] 80: return {answer: state[answer] 当前信息不足请转人工确认。} return {answer: state[answer] 请以支付渠道实际到账时间为准。} def build_answer_graph(): builder StateGraph(AnswerState) builder.add_node(draft_answer, draft_answer) builder.add_node(quality_check, quality_check) builder.add_node(revise_answer, revise_answer) builder.add_edge(START, draft_answer) builder.add_edge(draft_answer, quality_check) builder.add_edge(quality_check, revise_answer) builder.add_edge(revise_answer, END) return builder.compile() retrieval_graph build_retrieval_graph() answer_graph build_answer_graph() def normalize_question(state: ParentState) - dict: return {question: state[question].strip()} def retrieve_knowledge(state: ParentState) - dict: 适配父图与检索子图的不同 Schema。 result retrieval_graph.invoke( { query: state[question], rewritten_query: , documents: [], } ) return {evidence: result[documents]} def build_parent_graph(): builder StateGraph(ParentState) builder.add_node(normalize_question, normalize_question) builder.add_node(retrieve_knowledge, retrieve_knowledge) builder.add_node(generate_answer, answer_graph) builder.add_edge(START, normalize_question) builder.add_edge(normalize_question, retrieve_knowledge) builder.add_edge(retrieve_knowledge, generate_answer) builder.add_edge(generate_answer, END) return builder.compile(checkpointerInMemorySaver()) def main() - None: graph build_parent_graph() config {configurable: {thread_id: ch25-demo}} payload { question: 退款多久到账 , evidence: [], answer: , } result graph.invoke(payload, configconfig) print(result[question]) print(result[evidence]) print(result[answer]) if __name__ __main__: main()模块化测试应分为三层。第一层验证检索子图的输入输出契约def test_retrieval_graph_contract(): result retrieval_graph.invoke( {query: 退款时间, rewritten_query: , documents: []} ) assert result[rewritten_query] assert result[documents]第二层验证回复子图不依赖父图和检索实现def test_answer_graph_uses_evidence(): result answer_graph.invoke( { question: 退款需要多久, evidence: [退款需要 13 个工作日。], answer: , quality_score: 0, } ) assert 13 个工作日 in result[answer] assert result[quality_score] 80第三层验证父图编排和字段映射def test_parent_graph_end_to_end(): graph build_parent_graph() result graph.invoke( {question: 退款多久到账 , evidence: [], answer: }, config{configurable: {thread_id: test-parent}}, ) assert result[question] 退款多久到账 assert result[evidence] assert 工作日 in result[answer] assert quality_score not in result # 子图私有字段不进入父 State最终验收应覆盖以下内容父图 State 只保留question、evidence和answer。检索子图可以独立执行父图通过适配节点完成字段转换。回复子图可以直接挂载私有的quality_score不进入父图结果。v3 事件流能够识别检索和回复两个嵌套执行。v2 开启subgraphsTrue后子图事件具有非空ns。修改任一子图的内部实现时只要状态契约不变父图无需调整。如何学习大模型 AI 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取