langgraph子图

子图本质上就是将一个完整的“图Graph”作为一个“节点Node”来使用。在 LangGraph 的底层设计中子图并不是一个脱离于图之外的独立概念它其实就是 StateGraph 的一个普通节点——只不过这个节点本身也是一张完整的图。当您把子图作为一个节点添加到父图中时子图就被编译成了一个封闭的执行单元。这就像您把 CPU 插在主板上主板父图只知道“这个节点接收某种输入、产生某种输出”它不知道 CPU 内部有几个核心、几条总线内部有几个节点、几条边。这在工程上完全等同于微服务架构中“服务间只通过 API 通信不共享内部实现”的原则。from typing_extensions import TypedDict from langgraph.graph import StateGraph, START # 1. 定义子图状态和逻辑 class SubState(TypedDict): sub_input: str # 子图输入 sub_output: str # 子图输出 # 2 子图的节点:处理逻辑 def sub_process(state: SubState) - SubState: 子图核心处理逻辑 return {sub_output: f子图处理结果{state[sub_input].upper()}} # 3 构建并编译子图 sub_builder StateGraph(SubState) sub_builder.add_node(sub_process, sub_process) sub_builder.add_edge(START, sub_process) subgraph sub_builder.compile() #################### 子图的构建################################# # 1 定义父图状态和调用节点 class ParentState(TypedDict): parent_input: str # 父图输入 parent_output: str # 父图输出 # 2 父图节点--自己使用代码控制,调用子图 def call_subgraph_node(state: ParentState) - ParentState: 父图节点调用子图核心桥接逻辑 # 步骤1父状态 → 子状态自定义转换 sub_input_state {sub_input: state[parent_input]} # 步骤2调用子图节点内调用图的核心操作 sub_output_state subgraph.invoke(sub_input_state) # 步骤3子状态 → 父状态自定义转换 return {parent_output: sub_output_state[sub_output]} # 构建并编译父图 parent_builder StateGraph(ParentState) parent_builder.add_node(call_subgraph, call_subgraph_node) parent_builder.add_edge(START, call_subgraph) parent_graph parent_builder.compile() png_data parent_graph.get_graph(xrayTrue).draw_mermaid_png() with open(从节点调用子图.png, wb) as f: f.write(png_data) print(从节点调用子图状态图已经保存) # 3. 执行测试 result parent_graph.invoke({parent_input: hello langgraph}) print(最终结果, result[parent_output]) # 输出最终结果子图处理结果HELLO LANGGRAPH call_subgraph_node 是父图的普通节点但内部通过 subgraph.invoke() 调用了完整的子图 状态转换完全自定义灵活适配父子图不同的字段命名parent_input → sub_input 这种方式的优势是可控性强可在调用前后添加日志、异常处理、数据校验等逻辑。 所以call_subgraph_node 这个函数本质上就是一个“外包的翻译官”步骤 1翻译官把父图的包裹拆开挑出子图需要的东西parent_input装进子图的盒子里sub_input。步骤 2翻译官把盒子递给子图子图在里面一顿操作然后递回一个结果盒子sub_output_state。步骤 3翻译官把子图的结果盒子拆开挑出父图需要的东西sub_output重新贴上父图的标签parent_output交还给父图的下一个流程。def call_subgraph_node(state: ParentState) - ParentState: 父图节点调用子图核心桥接逻辑 # 步骤1父状态 → 子状态自定义转换 sub_input_state {sub_input: state[parent_input]} # 步骤2调用子图节点内调用图的核心操作 sub_output_state subgraph.invoke(sub_input_state) # 步骤3子状态 → 父状态自定义转换 return {parent_output: sub_output_state[sub_output]}call_subgraph_node这段函数实际上就是在做两件事先调用一下子图。把子图干完的结果塞回给父图。连redis短期记忆案例from langgraph.graph import StateGraph, MessagesState, START #pip install langgraph-checkpoint-redis from langgraph.checkpoint.redis import RedisSaver from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() import os # 1 连接大模型 model ChatOpenAI( modelos.getenv(MODEL_NAME), api_keyos.getenv(API_KEY), base_urlos.getenv(BASE_URL), ) # 2 创建 RedisSaver--使用上下文管理器 DB_URI redis://127.0.0.1:6380 # 使用docker跑一个 with RedisSaver.from_conn_string(DB_URI) as checkpointer: # with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: checkpointer.setup() # 在第一次使用 Redis 检查点时调用 checkpointer.setup()多次执行也没问题 # 3 节点 def call_model(state: MessagesState): response model.invoke(state[messages]) return {messages: response} # 3 构建图 builder StateGraph(MessagesState) builder.add_node(call_model) builder.add_edge(START, call_model) # 3.1 传入短期记忆---》放在redis中 graph builder.compile(checkpointercheckpointer) config { configurable: { thread_id: 1 } } # for chunk in graph.stream( # {messages: [{role: user, content: 你好我是lqz}]}, # config, # stream_modevalues # ): # chunk[messages][-1].pretty_print() for chunk in graph.stream( {messages: [{role: user, content: 我是谁?}]}, config, stream_modevalues ): chunk[messages][-1].pretty_print()mysql连接案例# pip install langgraph-checkpoint-mysql[pymysql] from langgraph.graph import StateGraph, MessagesState, START from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver from langchain_openai import ChatOpenAI from dotenv import load_dotenv import os import pymysql load_dotenv() # 1 连接大模型 model ChatOpenAI( modelos.getenv(MODEL_NAME), api_keyos.getenv(API_KEY), base_urlos.getenv(BASE_URL), ) # 2 创建mysql的 短期记忆 # 方式一 connection pymysql.connect(hostlocalhost, userroot, passwordlqz123?, databaselanggraph01, autocommitTrue) checkpointer PyMySQLSaver(connection) checkpointer.setup() def call_model(state: MessagesState): response model.invoke(state[messages]) return {messages: response} builder StateGraph(MessagesState) builder.add_node(call_model) builder.add_edge(START, call_model) graph builder.compile(checkpointercheckpointer) config { configurable: { thread_id: 1 } } for chunk in graph.stream( {messages: [{role: user, content: 你好我是lqz}]}, config, stream_modevalues ): chunk[messages][-1].pretty_print() for chunk in graph.stream( {messages: [{role: user, content: 我是谁?}]}, config, stream_modevalues ): chunk[messages][-1].pretty_print() # 方式二 # DB_URI fmysql://root:lqz123127.0.0.1:3306/langgraph01 # with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer: # checkpointer.setup() # def call_model(state: MessagesState): # response model.invoke(state[messages]) # return {messages: response} # # builder StateGraph(MessagesState) # builder.add_node(call_model) # builder.add_edge(START, call_model) # # graph builder.compile(checkpointercheckpointer) # # config { # configurable: { # thread_id: 1 # } # } # # for chunk in graph.stream( # {messages: [{role: user, content: 你好我是lqz}]}, # config, # stream_modevalues # ): # chunk[messages][-1].pretty_print() # # for chunk in graph.stream( # {messages: [{role: user, content: 我是谁?}]}, # config, # stream_modevalues # ): # chunk[messages][-1].pretty_print()