企业官网建设流程全解析

本文详细介绍了如何在大模型开发中建立和使用子图，将复杂流程封装为独立节点。通过客服知识检索与回复生成的实例，阐述了子图的拆分依据、状态契约、通信方式及持久化范围。文章强调子图设计需关注输入输出独立性、可测试性和内部流程，并提供两种通信方式供选择。此外，还讨论了子图状态的保存策略和嵌套流式观察方法，帮助开发者更好地理解和应用子图技术，提升代码组织和运行效率。

本章以“客服知识检索与回复生成”为主线，依次说明子图的拆分依据、状态契约、两种通信方式、持久化范围以及嵌套流式。

环境基线：Python 3.13、langchain>=1.3.2、LangGraph 1.x。示例使用确定性节点，便于观察状态变化；将节点内部实现替换为模型或检索器后，子图结构不需要改变。

一、何时拆成子图

假设客服流程包含以下步骤：

规范化问题 → 改写检索词 → 检索知识 → 生成回复 → 质量检查 → 必要时改写

将六个节点直接放入父图可以正常运行，但检索逻辑、回复逻辑和全局编排会共享同一个 State。后续增加查询扩展、混合检索或多轮质检时，父图会持续膨胀。

更合理的结构是保留简单的规范化节点，并将具有内部流程的能力拆成两个子图：

父图 ├── normalize_question 普通节点 ├── retrieve_knowledge 检索子图 │ ├── rewrite_query │ └── retrieve_documents └── generate_answer 回复子图 ├── draft_answer ├── quality_check └── revise_answer

判断一段流程是否适合拆成子图，可以检查三个条件：

1. 具有明确的输入输出。 调用方不需要了解内部节点，只需要提供输入并接收结果。

2. 能够独立测试或替换。 例如检索子图可以从关键词检索替换为向量检索，而父图接口保持不变。

3. 内部存在有意义的流程。 多个步骤共享私有状态、重试或路由逻辑时，子图可以形成稳定的模块边界。

只有一个简单计算节点时，通常没有必要再增加一层子图。子图层级越深，检查点命名空间、Trace 和流式路径也会越长。模块化的目标是降低耦合，而不是增加结构层次。

二、设计子图接口

子图设计的核心不是add_node，而是父图与子图之间的状态契约。要先确定哪些字段属于全局业务状态，哪些字段只在模块内部使用。

本例的父图只维护三个公共字段：

from typing_extensions import TypedDict class ParentState(TypedDict): question: str evidence: list[str] answer: str

检索子图使用自己的 Schema：

class 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、日志列表等追加型字段，两层应明确使用一致的合并语义。否则子图返回列表后，父图可能执行覆盖，也可能再次追加。对于不需要暴露的内部消息，更稳妥的做法是只向父图返回摘要或最终结果。

三、父子图通信

假设你有一个父图（主流程）和一个子图（检索模块）。把子图接入父图有两种方式：

比较如下：

核心区别是数据转换由谁来做：

• 方式一：我们自己做，在适配节点里显式转换。
• 方式二：框架自动做，通过共享状态键自动映射。

3. 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 条）

这种方式边界清晰，父子图完全解耦，字段名随便改。但是缺点是每个接入点都要写样板代码，比较麻烦。

3. 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. 3 两者对比

选型建议：

• 子图会被多个不同父图复用，且各父图的字段命名不统一 → 方式一（适配节点）
• 子图专为当前父图设计，字段名已经对齐 → 方式二（直接节点）
• 需要在输入输出时做校验、裁剪、摘要 → 方式一（适配节点）
• 追求代码简洁，不需要额外处理 → 方式二（直接节点）

四、子图状态能够保留多久

子图可以有自己的 Checkpointer，独立于父图。compile(checkpointer=...)有三种模式：

如下：

4. 1 None：默认模式

不传checkpointer参数时，子图使用默认模式。

本例中的检索和回复子图都是“一次性任务”，每次根据当前问题重新执行，不需要记住上一次调用的内部状态：

retrieval_graph = retrieval_builder.compile() # 等价于 checkpointer=None answer_graph = answer_builder.compile()

默认模式下，子图不会把上一次调用产生的私有状态带入下一次调用。但有一个例外：如果父图配置了 Checkpointer，子图的单次执行仍可通过父图参与中断恢复（例如父图 HITL 暂停后，子图未完成的部分可以在恢复时继续执行）。

适用场景：一次性查询、独立计算任务、任何“每次重新开始”的子任务。

4. 2 True：按线程保存内部状态

传入checkpointer=True时，子图在同一thread_id的多次调用之间累积内部状态。

answer_graph = answer_builder.compile(checkpointer=True)

比如一个多轮研究助手，子图需要在多次用户提问之间保留“已经搜过哪些关键词”“已经分析过哪些文档”，这些信息属于子图内部私有状态，父图不需要知道细节，但子图自己必须记住。

适用场景：多轮对话子图、编码助手（保留文件修改历史）、持续任务跟踪器。

⚠️ 重要限制：同一个子图实例不能在同一步中并行调用多次。

原因是多个并行调用会写入同一个检查点命名空间（checkpoint_ns），导致写入冲突。如果你需要对不同数据做并行处理，有两种方案：

1. 使用默认模式（None），每次调用独立，不跨调用共享状态

2. 使用不同的子图节点实例，每个有自己的命名空间

简单理解：checkpointer=True的子图就像一个“有状态的服务”，同一时间只能处理一个请求。

4. 3 False：不保存状态

传入checkpointer=False时，子图完全不保存任何状态。

pure_graph = pure_builder.compile(checkpointer=False)

这个模式最轻量，但代价是：如果子图内部有interrupt()（HITL 中断），暂停后无法恢复，因为状态根本没存。所以它只适合“进去→计算→出来”的纯函数式子图，没有任何需要中断或恢复的逻辑。

适用场景：数据格式转换、校验、纯计算模块，不需要中断、恢复和内部状态检查的任务。

与None的区别：None模式在父图有 Checkpointer 时，单次执行内的中断是可恢复的；False模式彻底不存，中断即丢失。

4. 4 父图 Checkpointer 与子图的关系

无论子图配置什么模式，子图的实际持久化能力都依赖父图提供 Saver：

from langgraph.checkpoint.memory import InMemorySaver # 父图配置 Checkpointer graph = parent_builder.compile(checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": "customer-1001"}}

如果父图没有配置 Checkpointer，子图的持久化链路不完整，checkpointer=True也无法正常工作。生产环境请将InMemorySaver替换为 SQLite、Postgres 等持久化后端。

五、如何观察子图内部执行

子图在父图中表现为一个节点。只观察父图更新时，可以看到generate_answer完成，却无法区分内部的草拟、质量检查和改写步骤。嵌套流式用于把这些内部执行暴露给调试工具、日志系统或前端服务。

5. 1 使用 v3 事件流观察子图

应用代码优先使用stream_events(..., version="v3")。它提供消息、状态、子图和最终输出等类型化投影，不需要业务代码自行拆解底层事件元组。

stream = graph.stream_events( { "question": "退款多久到账？", "evidence": [], "answer": "", }, config={"configurable": {"thread_id": "stream-v3"}}, version="v3", ) # 每个 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 的原始命名空间，否则图结构或节点名称调整后，前端协议也需要同步修改。

5. 2 使用 v2 原始流进行底层排障

当需要查看 Pregel 的原始updates、values或tasks时，可以使用stream(..., version="v2")。设置subgraphs=True后，父图和子图事件采用统一的StreamPart结构：

for chunk in graph.stream( { "question": "退款多久到账？", "evidence": [], "answer": "", }, config={"configurable": {"thread_id": "stream-v2"}}, ) stream_mode="updates", subgraphs=True, version="v2", ): 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 使用subgraphs=True并根据ns判断事件来源。

5. 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": [ f"FAQ 命中：{state['rewritten_query']}", "退款原路返回，通常需要 1～3 个工作日。", ] } 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(checkpointer=InMemorySaver()) def main() -> None: graph = build_parent_graph() config = {"configurable": {"thread_id": "ch25-demo"}} payload = { "question": " 退款多久到账？ ", "evidence": [], "answer": "", } result = graph.invoke(payload, config=config) 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": ["退款需要 1～3 个工作日。"], "answer": "", "quality_score": 0, } ) assert "1～3 个工作日" 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

最终验收应覆盖以下内容：

1. 父图 State 只保留question、evidence和answer。

2. 检索子图可以独立执行，父图通过适配节点完成字段转换。

3. 回复子图可以直接挂载，私有的quality_score不进入父图结果。

4. v3 事件流能够识别检索和回复两个嵌套执行。

5. v2 开启subgraphs=True后，子图事件具有非空ns。

6. 修改任一子图的内部实现时，只要状态契约不变，父图无需调整。

如何学习大模型 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大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

以上全套大模型资料如何领取？