一、先搞懂：什么是子图（Subgraph）？
1. 一句话定义
子图 = 把一个完整的小 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(q=state["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. 子图 2：RAG 检索子图
# 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(c=state["context"], q=state["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(model="glm-4", temperature=0) # 模拟向量库（你可替换成真实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 系统必须用子图 + 模块化
