LangGraph 子图 Subgraph 工程化拆分|gent 业务如何拆成可复用流水线

一个 State 定义膨胀到几十个字段节点之间边线纵横交错想改一个功能却不敢动——因为牵一发而动全身。更头疼的是做另一个项目时发现很多逻辑几乎一样却只能再写一遍。这不是你的错。是单体图架构的问题。当一个 Agent 图超过 10 个节点状态就会开始爆炸。字段相互耦合边界条件纠缠调试时要在巨大的状态对象中追踪字段变化——像在毛线团里找线头。假设你有一个智能客服 Agent需要做以下工作查询知识库调用天气 API搜索最新资讯核实信息来源生成回答检查回答质量发送到不同渠道如果全部写在一个 StateGraph 里class HugeState(TypedDict): query: str kb_results: list[str] weather_data: dict search_results: list[str] verified_claims: list[str] draft_answer: str quality_score: float channel_config: dict audit_log: list[str] # ... 越来越长致命问题状态爆炸每个节点都能读写所有字段改一个字段就得审视所有节无法复用知识库查询逻辑在其他项目也需要却只能再写一遍团队协作困难多人同时修改同一个图的 edges合并冲突是噩梦子图的解决思路很直接像微服务拆解单体应用一样把大图按职责拆成独立的子图。子图的工作原理子图本质上就是一个被当作节点使用的已编译图。当父图调用子图时LangGraph 做了三件事状态映射把父图的指定字段映射到子图的输入独立执行子图在自己的状态空间内运行到结束结果回传把子图的输出字段映射回父图的对应字段最关键的特性子图的状态是私有的。父图只能看到你显式暴露出来的字段。这意味着不同子图可以有完全独立的字段定义同一个字段名在不同子图中各不相干修改子图内部逻辑不影响其他图LangGraph v1.2.7 支持两种模式取决于父图和子图是否共享状态键。通信模式不同状态——节点内调用适用于父图和子图的状态模式完全不同。你需要写一个包装函数来做状态转换from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START, END # 子图独立的内部状态 class SubgraphState(TypedDict): raw_query: str results: list[str] def search_kb(state: SubgraphState): # 子图内部逻辑完全不知道父图的存在 return {results: [f找到关于 {state[raw_query]} 的结果]} def rerank(state: SubgraphState): # 子图内部的第二个节点 return {results: sorted(state[results])} subgraph_builder StateGraph(SubgraphState) subgraph_builder.add_node(search, search_kb) subgraph_builder.add_node(rerank, rerank) subgraph_builder.add_edge(START, search) subgraph_builder.add_edge(search, rerank) subgraph_builder.add_edge(rerank, END) kb_subgraph subgraph_builder.compile() # 父图完全不同的状态 class ParentState(TypedDict): user_input: str final_answer: str def call_kb_subgraph(state: ParentState) - ParentState: # 包装函数父→子→父的状态转换 subgraph_input {raw_query: state[user_input]} subgraph_output kb_subgraph.invoke(subgraph_input) return {final_answer: \n.join(subgraph_output[results])} builder StateGraph(ParentState) builder.add_node(knowledge_base, call_kb_subgraph) builder.add_edge(START, knowledge_base) builder.add_edge(knowledge_base, END) graph builder.compile()这种模式的好处是隔离性最强。子图可以独立测试、独立部署只要接口签名不变内部随便重构。共享状态当父图和子图共享部分状态键时可以直接把编译后的子图传给 add_nodeLangGraph 自动处理状态映射from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START, END # 共享状态 class SharedState(TypedDict): query: str search_results: list[str] # 子图 def sub_search(state: SharedState): return {search_results: [f结果1: {state[query]}, f结果2: {state[query]}]} sub_builder StateGraph(SharedState) sub_builder.add_node(sub_search, sub_search) sub_builder.add_edge(START, sub_search) sub_builder.add_edge(sub_search, END) subgraph sub_builder.compile() # 父图 builder StateGraph(SharedState) builder.add_node(parent_node, lambda s: s) builder.add_node(child, subgraph) # 直接传入子图无需包装函数 builder.add_edge(START, parent_node) builder.add_edge(parent_node, child) builder.add_edge(child, END) graph builder.compile()我们用真实的业务场景来演示。假设我们要构建一个内容分析 Agent需要搜索相关资讯分析文章情感提取关键实体生成摘要报告子图 A搜索模块from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class SearchState(TypedDict): query: str sources: list[str] results: list[dict] def search_web(state: SearchState) - SearchState: # 模拟搜索 return {results: [ {title: AI 最新突破 2026, content: ...}, {title: 大模型应用落地实践, content: ...}, ]} def filter_relevant(state: SearchState) - SearchState: # 过滤低质量结果 state[results] [r for r in state[results] if r.get(title)] return state search_builder StateGraph(SearchState) search_builder.add_node(search, search_web) search_builder.add_node(filter, filter_relevant) search_builder.add_edge(START, search) search_builder.add_edge(search, filter) search_builder.add_edge(filter, END) search_graph search_builder.compile()子图 B分析模块class AnalysisState(TypedDict): content: str sentiment: str entities: list[str] summary: str def analyze_sentiment(state: AnalysisState) - AnalysisState: # 模拟情感分析 return {sentiment: 正面} def extract_entities(state: AnalysisState) - AnalysisState: # 模拟实体提取 return {entities: [AI, 大模型, LangGraph]} def generate_summary(state: AnalysisState) - AnalysisState: summary ( f情感: {state[sentiment]}, f实体: {, .join(state[entities])} ) return {summary: summary} analysis_builder StateGraph(AnalysisState) analysis_builder.add_node(sentiment, analyze_sentiment) analysis_builder.add_node(entities, extract_entities) analysis_builder.add_node(summarize, generate_summary) analysis_builder.add_edge(START, sentiment) analysis_builder.add_edge(START, entities) analysis_builder.add_edge(sentiment, summarize) analysis_builder.add_edge(entities, summarize) analysis_builder.add_edge(summarize, END) analysis_graph analysis_builder.compile()父图编排调度class ReportState(TypedDict): topic: str raw_articles: list[dict] sentiment_result: str entity_list: list[str] report_summary: str final_report: str def call_search(state: ReportState) - ReportState: output search_graph.invoke({query: state[topic]}) return {raw_articles: output[results]} def call_analysis(state: ReportState) - ReportState: content_text \n.join( [a.get(content, ) for a in state[raw_articles]] ) output analysis_graph.invoke({content: content_text}) return { sentiment_result: output[sentiment], entity_list: output[entities], report_summary: output[summary], } def generate_report(state: ReportState) - ReportState: report ( f# {state[topic]} 分析报告\n\n f情感倾向{state[sentiment_result]}\n f关键实体{, .join(state[entity_list])}\n f摘要{state[report_summary]}\n ) return {final_report: report} builder StateGraph(ReportState) builder.add_node(search_phase, call_search) builder.add_node(analysis_phase, call_analysis) builder.add_node(report_phase, generate_report) builder.add_edge(START, search_phase) builder.add_edge(search_phase, analysis_phase) builder.add_edge(analysis_phase, report_phase) builder.add_edge(report_phase, END) agent builder.compile(checkpointerMemorySaver()) # 执行 result agent.invoke( {topic: 2026年大模型行业应用趋势}, {configurable: {thread_id: report-001}} ) print(result[final_report])子图持久化配置# 模式一每次调用默认推荐 # 每次调用子图都重新开始内部分支状态不保留 subgraph sub_builder.compile() # checkpointerNone # 模式二每线程跨调用记忆 # 同一线程内子图状态在多次调用之间累积 subgraph sub_builder.compile(checkpointerTrue) # 模式三无状态纯函数 # 没有检查点不支持中断像普通函数 subgraph sub_builder.compile(checkpointerFalse)完整代码from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 子图1网页搜索模块私有独立状态 class SearchSubState(TypedDict): query: str # 子图独有输入 source_links: list[str] # 子图内部字段父图不可感知 article_results: list[dict] def node_web_search(state: SearchSubState) - SearchSubState: 模拟全网搜索接口 mock_data [ {title: 2026大模型落地行业白皮书, content: 企业AI Agent规模化落地趋势明显}, {title: 多智能体子图架构最佳实践, content: LangGraph子图拆分解决状态爆炸问题}, {title: 垂直行业知识库检索优化方案, content: 检索重排提升问答准确率} ] return {article_results: mock_data} def node_filter_useless(state: SearchSubState) - SearchSubState: 过滤空内容低质量文章 valid_articles [item for item in state[article_results] if item.get(content)] return {article_results: valid_articles} # 构建并编译搜索子图 search_graph_builder StateGraph(SearchSubState) search_graph_builder.add_node(web_search, node_web_search) search_graph_builder.add_node(filter, node_filter_useless) search_graph_builder.add_edge(START, web_search) search_graph_builder.add_edge(web_search, filter) search_graph_builder.add_edge(filter, END) # 子图无独立记忆每次调用全新上下文 search_subgraph search_graph_builder.compile(checkpointerNone) # 子图2内容情感实体分析模块私有独立状态 class AnalysisSubState(TypedDict): full_text: str sentiment_label: str key_entities: list[str] short_summary: str def node_sentiment_analysis(state: AnalysisSubState) - AnalysisSubState: 模拟情感分析大模型调用 if 落地 in state[full_text] or 优化 in state[full_text]: return {sentiment_label: 积极乐观} return {sentiment_label: 中性客观} def node_entity_extract(state: AnalysisSubState) - AnalysisSubState: 模拟实体抽取 entities [大模型, LangGraph, AI Agent, 子图架构] return {key_entities: entities} def node_text_summarize(state: AnalysisSubState) - AnalysisSubState: 生成文本摘要 summary f文本情感{state[sentiment_label]}核心实体{,.join(state[key_entities])} return {short_summary: summary} # 并行执行情感实体再汇总摘要 analysis_graph_builder StateGraph(AnalysisSubState) analysis_graph_builder.add_node(sentiment, node_sentiment_analysis) analysis_graph_builder.add_node(entity_extract, node_entity_extract) analysis_graph_builder.add_node(summary, node_text_summarize) analysis_graph_builder.add_edge(START, sentiment) analysis_graph_builder.add_edge(START, entity_extract) analysis_graph_builder.add_edge(sentiment, summary) analysis_graph_builder.add_edge(entity_extract, summary) analysis_graph_builder.add_edge(summary, END) analysis_subgraph analysis_graph_builder.compile(checkpointerNone) # 父图全局状态仅对外暴露业务字段 class ReportParentState(TypedDict): topic: str # 用户输入主题 raw_articles: list[dict] # 接收搜索子图输出 sentiment: str # 接收分析子图输出 entities: list[str] text_summary: str final_markdown_report: str # 最终产出 # ---------------------- 父图节点调用搜索子图状态转换包装函数 ---------------------- def invoke_search_subgraph(state: ReportParentState) - ReportParentState: # 父状态映射到子图输入 sub_input {query: state[topic]} sub_out search_subgraph.invoke(sub_input) # 子图输出回填父状态 return {raw_articles: sub_out[article_results]} # ---------------------- 父图节点调用分析子图 ---------------------- def invoke_analysis_subgraph(state: ReportParentState) - ReportParentState: # 合并所有文章文本作为子图输入 combine_text \n.join([art[content] for art in state[raw_articles]]) sub_input {full_text: combine_text} sub_out analysis_subgraph.invoke(sub_input) return { sentiment: sub_out[sentiment_label], entities: sub_out[key_entities], text_summary: sub_out[short_summary] } # ---------------------- 父图节点组装最终报告 ---------------------- def build_final_report(state: ReportParentState) - ReportParentState: md f# 行业分析报告{state[topic]} ## 一、基础信息 文本整体情感{state[sentiment]} ## 二、核心实体 {, .join(state[entities])} ## 三、内容摘要 {state[text_summary]} ## 四、原始素材列表 for idx, article in enumerate(state[raw_articles], 1): md f{idx}. {article[title]}\n return {final_markdown_report: md} # 组装父图并全局持久化记忆 parent_builder StateGraph(ReportParentState) parent_builder.add_node(search_stage, invoke_search_subgraph) parent_builder.add_node(analysis_stage, invoke_analysis_subgraph) parent_builder.add_node(report_stage, build_final_report) # 流程图链路 parent_builder.add_edge(START, search_stage) parent_builder.add_edge(search_stage, analysis_stage) parent_builder.add_edge(analysis_stage, report_stage) parent_builder.add_edge(report_stage, END) # 全局记忆持久化支持多线程会话 global_checkpointer MemorySaver() main_agent parent_builder.compile(checkpointerglobal_checkpointer) # 执行入口 if __name__ __main__: # 会话唯一ID区分不同用户/任务 thread_config {configurable: {thread_id: task-2026-model-report-001}} # 启动Agent run_result main_agent.invoke( input{topic: 2026年大模型智能体子图架构落地趋势}, configthread_config ) # 打印最终报告 print( * 60) print(run_result[final_markdown_report])学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】