一、先搞懂什么是子图Subgraph1. 一句话定义子图 把一个完整的小 Agent当成另一个大图里的一个节点来用。就像一个大项目 总系统主图里面的「检索模块」 子图 A里面的「写作模块」 子图 B里面的「反思校验模块」 子图 C子图让代码彻底解耦、可复用、可独立测试、可独立维护。2. 为什么要用子图核心价值✅ 代码结构清晰大 Agent 不会写成几千行代码而是拆成多个小文件、小模块。✅ 可复用一个 RAG 子图可以被任何主图调用不用重复写。✅ 易维护修改检索逻辑 → 只改检索子图修改写作逻辑 → 只改写作子图互不影响✅ 适合复杂系统多智能体系统、规划执行系统、RAG 工具混合系统必须用子图。3. 子图 vs 普通节点 的区别表格类型功能复杂度复用性普通节点一个函数低差子图完整小 Agent多节点 路由 状态高极强二、LangGraph 子图核心 API必须记住1. 创建子图builder StateGraph(SubState) # 添加节点、边 subgraph builder.compile()2. 把子图变成主图的一个节点main_builder.add_node(my_subgraph, subgraph)3. 连接子图main_builder.add_edge(start, my_subgraph)就这么简单三、模块化 子图 最佳工程架构我直接给你企业级标准结构agent_system/ ├── main.py # 主图总调度 ├── subgraphs/ │ ├── __init__.py │ ├── planner.py # 规划子图 │ ├── rag.py # RAG检索子图 │ ├── writer.py # 写作子图 │ └── validator.py # 校验子图 ├── state.py # 全局状态定义 ├── settings.py # 配置 └── utils.py # 工具每个子图独立文件、独立状态、独立逻辑。四、实战开发一个 主图 4 个子图 的超级 Agent我直接写完整可运行代码结构超级清晰功能介绍主图调度大脑负责调用规划子图→ 拆步骤调用RAG 子图→ 查文档调用写作子图→ 生成答案调用校验子图→ 检查答案质量完整代码可直接复制运行1. 全局状态共用from typing import TypedDict, Annotated, List from langchain_core.messages import BaseMessage from langgraph.graph import add_messages import operator class GlobalState(TypedDict): user_query: str plan: List[str] # 规划结果 context: str # RAG检索结果 content: str # 写作结果 is_valid: bool # 校验结果 messages: Annotated[List[BaseMessage], add_messages] error: str | None2. 子图 1任务规划子图# subgraphs/planner.py from langgraph.graph import StateGraph from langchain_core.prompts import ChatPromptTemplate def create_planner_subgraph(llm): def plan_node(state): prompt ChatPromptTemplate.from_template( 把用户问题拆成3步 1. 理解问题 2. 检索资料 3. 生成答案 用户问题{q} 只返回步骤用逗号分隔 ) res llm.invoke(prompt.format(qstate[user_query])) steps res.content.strip().split(,) return {plan: steps} builder StateGraph(GlobalState) builder.add_node(plan, plan_node) builder.set_entry_point(plan) return builder.compile()3. 子图 2RAG 检索子图# subgraphs/rag.py from langgraph.graph import StateGraph def create_rag_subgraph(vector_store): def retrieve_node(state): docs vector_store.similarity_search(state[user_query]) context \n.join([d.page_content for d in docs]) return {context: context} builder StateGraph(GlobalState) builder.add_node(retrieve, retrieve_node) builder.set_entry_point(retrieve) return builder.compile()4. 子图 3写作生成子图# subgraphs/writer.py from langgraph.graph import StateGraph from langchain_core.prompts import ChatPromptTemplate def create_writer_subgraph(llm): def write_node(state): prompt ChatPromptTemplate.from_template( 用资料回答问题 资料{c} 问题{q} ) res llm.invoke(prompt.format(cstate[context], qstate[user_query])) return {content: res.content} builder StateGraph(GlobalState) builder.add_node(write, write_node) builder.set_entry_point(write) return builder.compile()5. 子图 4校验子图# subgraphs/validator.py from langgraph.graph import StateGraph def create_validator_subgraph(llm): def validate_node(state): content state[content] is_valid len(content) 20 # 简单判断 return {is_valid: is_valid} builder StateGraph(GlobalState) builder.add_node(validate, validate_node) builder.set_entry_point(validate) return builder.compile()6. 主图把所有子图组装起来# main.py from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from state import GlobalState from subgraphs.planner import create_planner_subgraph from subgraphs.rag import create_rag_subgraph from subgraphs.writer import create_writer_subgraph from subgraphs.validator import create_validator_subgraph # 初始化LLM llm ChatOpenAI(modelglm-4, temperature0) # 模拟向量库你可替换成真实RAG class MockVectorStore: def similarity_search(self, query): return [type(obj, (object,), {page_content: 文档内容})] vector_store MockVectorStore() # 创建子图 planner create_planner_subgraph(llm) rag create_rag_subgraph(vector_store) writer create_writer_subgraph(llm) validator create_validator_subgraph(llm) # 主图 builder StateGraph(GlobalState) # 把子图变成主图的节点 builder.add_node(planner, planner) builder.add_node(rag, rag) builder.add_node(writer, writer) builder.add_node(validator, validator) # 流程 builder.add_edge(START, planner) builder.add_edge(planner, rag) builder.add_edge(rag, writer) builder.add_edge(writer, validator) builder.add_edge(validator, END) # 编译 agent builder.compile() # 测试 if __name__ __main__: result agent.invoke({ user_query: 介绍Dubbo核心功能, messages: [] }) print(最终答案, result[content])五、这就是模块化 子图的终极威力✅ 每个子图可以独立测试✅ 每个子图可以独立更新✅ 每个子图可以被其他项目复用✅ 代码结构超级清晰✅ 支持无限扩展加子图即可六、我帮你总结最核心的 3 句话子图 小 Agent子图可以像普通节点一样加入主图大型 Agent 系统必须用子图 模块化
LangGraph 子图 + 模块化设计 全解
一、先搞懂什么是子图Subgraph1. 一句话定义子图 把一个完整的小 Agent当成另一个大图里的一个节点来用。就像一个大项目 总系统主图里面的「检索模块」 子图 A里面的「写作模块」 子图 B里面的「反思校验模块」 子图 C子图让代码彻底解耦、可复用、可独立测试、可独立维护。2. 为什么要用子图核心价值✅ 代码结构清晰大 Agent 不会写成几千行代码而是拆成多个小文件、小模块。✅ 可复用一个 RAG 子图可以被任何主图调用不用重复写。✅ 易维护修改检索逻辑 → 只改检索子图修改写作逻辑 → 只改写作子图互不影响✅ 适合复杂系统多智能体系统、规划执行系统、RAG 工具混合系统必须用子图。3. 子图 vs 普通节点 的区别表格类型功能复杂度复用性普通节点一个函数低差子图完整小 Agent多节点 路由 状态高极强二、LangGraph 子图核心 API必须记住1. 创建子图builder StateGraph(SubState) # 添加节点、边 subgraph builder.compile()2. 把子图变成主图的一个节点main_builder.add_node(my_subgraph, subgraph)3. 连接子图main_builder.add_edge(start, my_subgraph)就这么简单三、模块化 子图 最佳工程架构我直接给你企业级标准结构agent_system/ ├── main.py # 主图总调度 ├── subgraphs/ │ ├── __init__.py │ ├── planner.py # 规划子图 │ ├── rag.py # RAG检索子图 │ ├── writer.py # 写作子图 │ └── validator.py # 校验子图 ├── state.py # 全局状态定义 ├── settings.py # 配置 └── utils.py # 工具每个子图独立文件、独立状态、独立逻辑。四、实战开发一个 主图 4 个子图 的超级 Agent我直接写完整可运行代码结构超级清晰功能介绍主图调度大脑负责调用规划子图→ 拆步骤调用RAG 子图→ 查文档调用写作子图→ 生成答案调用校验子图→ 检查答案质量完整代码可直接复制运行1. 全局状态共用from typing import TypedDict, Annotated, List from langchain_core.messages import BaseMessage from langgraph.graph import add_messages import operator class GlobalState(TypedDict): user_query: str plan: List[str] # 规划结果 context: str # RAG检索结果 content: str # 写作结果 is_valid: bool # 校验结果 messages: Annotated[List[BaseMessage], add_messages] error: str | None2. 子图 1任务规划子图# subgraphs/planner.py from langgraph.graph import StateGraph from langchain_core.prompts import ChatPromptTemplate def create_planner_subgraph(llm): def plan_node(state): prompt ChatPromptTemplate.from_template( 把用户问题拆成3步 1. 理解问题 2. 检索资料 3. 生成答案 用户问题{q} 只返回步骤用逗号分隔 ) res llm.invoke(prompt.format(qstate[user_query])) steps res.content.strip().split(,) return {plan: steps} builder StateGraph(GlobalState) builder.add_node(plan, plan_node) builder.set_entry_point(plan) return builder.compile()3. 子图 2RAG 检索子图# subgraphs/rag.py from langgraph.graph import StateGraph def create_rag_subgraph(vector_store): def retrieve_node(state): docs vector_store.similarity_search(state[user_query]) context \n.join([d.page_content for d in docs]) return {context: context} builder StateGraph(GlobalState) builder.add_node(retrieve, retrieve_node) builder.set_entry_point(retrieve) return builder.compile()4. 子图 3写作生成子图# subgraphs/writer.py from langgraph.graph import StateGraph from langchain_core.prompts import ChatPromptTemplate def create_writer_subgraph(llm): def write_node(state): prompt ChatPromptTemplate.from_template( 用资料回答问题 资料{c} 问题{q} ) res llm.invoke(prompt.format(cstate[context], qstate[user_query])) return {content: res.content} builder StateGraph(GlobalState) builder.add_node(write, write_node) builder.set_entry_point(write) return builder.compile()5. 子图 4校验子图# subgraphs/validator.py from langgraph.graph import StateGraph def create_validator_subgraph(llm): def validate_node(state): content state[content] is_valid len(content) 20 # 简单判断 return {is_valid: is_valid} builder StateGraph(GlobalState) builder.add_node(validate, validate_node) builder.set_entry_point(validate) return builder.compile()6. 主图把所有子图组装起来# main.py from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from state import GlobalState from subgraphs.planner import create_planner_subgraph from subgraphs.rag import create_rag_subgraph from subgraphs.writer import create_writer_subgraph from subgraphs.validator import create_validator_subgraph # 初始化LLM llm ChatOpenAI(modelglm-4, temperature0) # 模拟向量库你可替换成真实RAG class MockVectorStore: def similarity_search(self, query): return [type(obj, (object,), {page_content: 文档内容})] vector_store MockVectorStore() # 创建子图 planner create_planner_subgraph(llm) rag create_rag_subgraph(vector_store) writer create_writer_subgraph(llm) validator create_validator_subgraph(llm) # 主图 builder StateGraph(GlobalState) # 把子图变成主图的节点 builder.add_node(planner, planner) builder.add_node(rag, rag) builder.add_node(writer, writer) builder.add_node(validator, validator) # 流程 builder.add_edge(START, planner) builder.add_edge(planner, rag) builder.add_edge(rag, writer) builder.add_edge(writer, validator) builder.add_edge(validator, END) # 编译 agent builder.compile() # 测试 if __name__ __main__: result agent.invoke({ user_query: 介绍Dubbo核心功能, messages: [] }) print(最终答案, result[content])五、这就是模块化 子图的终极威力✅ 每个子图可以独立测试✅ 每个子图可以独立更新✅ 每个子图可以被其他项目复用✅ 代码结构超级清晰✅ 支持无限扩展加子图即可六、我帮你总结最核心的 3 句话子图 小 Agent子图可以像普通节点一样加入主图大型 Agent 系统必须用子图 模块化
