LangGraph知识

• 目录

  一、基础概念

  1.状态

  （1）State特性-Reducer规约器

  2.节点

  3.边

  二、LangGraph的高级特性

  1.Send机制：[send(节点,参数)]

  2.Command节点跳转

  3.checkpointer持久化

  3.1持久化redis的使用

  4.Threads 会话id

  5.Configuration节点内部配置

  6.interupt中断

  7.Subgraph子图

  8.Graphviz图可视化

  9.ToolNode工具节点

  三、对图的细粒度控制

  1.人机交互

  2.retry=RetryPolicy节点重试策略

  3.同一用户所有消息的持久化存储

  4.checkpoint持久化存储到数据库

  5.消息的摘要和删除RemoveMessage处理

  6.toolNode工具调用失败处理

  7.toolNode工具调用失败，转到新的工具节点调用

  7.将图state状态注入给工具函数

  8.将图第三方存储和配置注入给工具函数

  9.工具函数中更新state

一、基础概念

• 核心：围绕更新图的State

1.状态

• 定义的变量类
• 全局共享的数据容器,独立的数据库

（1）Reducer规约器

• 把每次节点return的消息，都追加到消息列表
• 消息存储只在当前的图中有效
• 对当前新的值进行返回，并合并之后的值，类似消息记录
• 自定义规约器

operator.add add_messages MessagesState class ChatState(TypedDict): messages: Annotated[Sequence[AnyMessage], operator.add]

2.节点

• 执行业务逻辑，更改状态的可执行函数

3.边

• 连接节点间的流转规则（可以为函数）

（1）正常边

（2）条件边：决定走向流程（函数控制）

return “事件”

根据“事件”：“节点”跳转

from langchain_core.messages import AnyMessage from langchain_core.runnables import RunnableConfig from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END #定义State class State(TypedDict): messages: list[AnyMessage] extra_field: int #定义Node def process_input_node(state: State, config: RunnableConfig): print(f"process_input_node :{state}") user_id = config.get("configurable", {}).get("user_id", "default_user") print(f"Node 'process_input_node' processing for user: {user_id}") return {"process_input": state["extra_field"]+2} #定义另一个Node def another_node(state: State): # 这个节点没有 config 参数 print(f"another_node :{state}") state["extra_field"] = 10 return {"some_other_data": "done"} def node_a(state: State): print("Called A") print(f"node_a :{state}") return {"extra_field": state["extra_field"] + 2} def node_b(state: State): print("Called B") print(f"node_b :{state}") return {"extra_field": state["extra_field"] + 1} # 路由选择 def route_tools(state: State): # 判断是否为偶数（最后一位二进制位为0表示偶数） if state["extra_field"] & 1 == 0: return "tj_1" else: return "tj_2" graph_builder = StateGraph(State) graph_builder.add_node("processor", process_input_node) graph_builder.add_node("finalizer", another_node) graph_builder.add_node("node_a", node_a) graph_builder.add_node("node_b", node_b) graph_builder.add_edge(START, "processor") #条件边:事件，节点 graph_builder.add_conditional_edges('processor', route_tools, {"tj_1": "node_b", "tj_2": "node_a"}) graph_builder.add_edge("node_a", END) graph_builder.add_edge("node_b", "finalizer") # 从业务节点连接到结束节点 graph_builder.add_edge("finalizer", END) # 编译 graph = graph_builder.compile() #打印{'messages': ['My', 'name'], 'extra_field': 1}，中间的业务节点对状态并没有贡献 print(graph.invoke({"messages": ["My", "name"], "extra_field": 2}))

二、LangGraph的高级特性

Schema数据结构

reducer归约器

一直叠加消息，

1.Send节点跳转

根据条件跳转到节点：[send(节点别名,参数)]

• Map-Reduce（扇出）。将一个列表拆分成多个任务并行执行
• 并行。同时启动多个节点（或同一节点的多个实例）。
• 必须放在add_conditional_edges的返回结果中。

import operator from typing import Annotated from typing import TypedDict from langgraph.graph import StateGraph from langgraph.types import Send from langgraph.graph import END, START class OverallState(TypedDict): # 属性 subjects: list[str] # 保存消息 jokes: Annotated[list[str], operator.add] # 条件节点 def continue_to_jokes(state: OverallState): # [ # Send("generate_joke", {"subject": "cats"}), # Send("generate_joke", {"subject": "dogs"}) # ] # send("节点别名", {"临时参数": "cats"}) return [Send("generate_joke", {"linshi_v": s}) for s in state['subjects']] def generate_joke(state: OverallState): v = state['linshi_v'] res = f"Joke about {v}" return {"jokes": [res]} workflows = StateGraph(OverallState) workflows.add_node("generate_joke",generate_joke) # 条件边 workflows.add_conditional_edges(START, continue_to_jokes) workflows.add_edge("generate_joke", END) graph = workflows.compile() # {"subjects": ["cats", "dogs"]}是因为state里面定义了这个属性字段 result = graph.invoke({"subjects": ["cats", "dogs"]}) # 最终返回的结果也是定义的状态的内容 # {'subjects': ['cats', 'dogs'], # 'jokes': ['Joke about cats', 'Joke about dogs']} print(result)

2.Command节点跳转

• 动态路由。在节点内部决定下一步去哪、更新什么状态。
• 串行/跳转。决定下一个执行的单一节点。
• 直接更新全局状态（update字段）。
• 直接放在节点函数的return中。

• **Command(resume=xxx)

  = 专门用来恢复interrupt()暂停的流程 **

graph=Command.PARENT, # 指定跳转的父图 goto=goto, # 跳转到子图相应节点 resume= # 给被暂停的interrupt节点传递数据， update={"foo": value}, # 更新值的内容,value是占位符

config

1.state = graph.get_state(config)到底拿到了什么？

当你执行这行代码时，返回的state对象（通常是StateSnapshot类型）包含几个核心字段：

• state.values: 当前全局状态字典（即你的foo、user_feedback等数据）。

• state.next:最关键的字段。它是一个元组（Tuple），记录了接下来要执行的节点名称。

• state.tasks: 当前正在处理或被中断的任务列表。

• state.config: 当前状态对应的配置（包含checkpoint_id等）。

3.checkpointer持久化

• 保存图中state的状态，state更新一次，保存一次（编译的时候传入启用）
• invoke调用的时候，必须结合 config使用
• 父图设置了检查点，子图一样具有检查点
• 参数：

config:配置

metadata:与此检查点相关的元数据

values:此时状态的通道

next:图中下一个要执行的节点名称元组

tasks:包含有关要执行的下个任务的RregelTask对象的元组

3.1持久化在redis中的使用

4.Threads_id 和session_id

• session_id:LangChain 核心生态（全组件通用）,用户 / 客户端级「长期会话标识」
• thread_id:LangGraph 专属，单次图执行级，图的State的持久化

• 典型业务场景：客服机器人的工作流

• 单个用户（session_id="customer_789"）发起多次咨询，每次咨询触发一次 LangGraph 工作流执行；

# 配置双层标识：session_id（用户级） + thread_id（单次图执行级） config: RunnableConfig = { "configurable": { "session_id": "user_001", # 手动指定：用户001的唯一会话标识 "thread_id": "joke_graph_run_001" # 手动指定：本次笑话生成的图执行标识 } }

5.Configuration节点内部配置

• 特殊保留参数，可用于切换某个大模型
• 与图中传入的配置不同，这个手动处理接收

def _call_model(state: AgentState, config: RunnableConfig): print(graph.invoke({"messages": [HumanMessage(content="你是谁？")]}, config=config))

6.interupt中断

• 需要结合checkpointer+config+command使用
• 写了interupt，就得写两次调用，第一次运行到节点等待，第二次传递值得到数据
• 在图中的某个节点，使用interupt运行到该节点，操作暂停，调用这个图的时候，需要通过command的resume传递一个消息给暂停的节点，让图继续运行

**Command(resume=xxx)

专门用来恢复interrupt()暂停的流程 **

7.Subgraph子图

• 父图把子图当做一个节点用，一般有共享属性字段

• builder.add_node("node_2", subgraph)

• 在父图里面调用子图使用，一般没有共享属性字段

• response = Subgraph.invoke({"subBar": state["ParFoo"]})

8.Graphviz图可视化

from IPython.display import Image, display """可视化图""" #方式一：需要单独安装graphviz # 方法：从官网下载 # 访问 Graphviz官网 https://graphviz.org/download/ # 下载Windows版本的安装包 # 运行安装程序并按照提示完成安装 # 将Graphviz的bin目录添加到系统PATH环境变量中（通常是C:\Program Files\Graphviz\bin） # 并且请用try块包裹下面的语句，因为执行会报错，但是不影响图片的生成 try: display(Image(app.get_graph().draw_png(output_file_path='./可视化图1.png'))) except: pass #方式二：不需要额外安装软件，但是访问网址mermaid.ink非常容易失败，开启科学上网比较容易成功 display(Image(app.get_graph().draw_mermaid_png(output_file_path='./可视化图2.png')))

9.ToolNode工具节点

• 对工具进行实际操作使用
• toolnode+llm.bind_tools=agent
• 当工具操作完后，还需要判断是否要继续调用工具

# 做实际操作的 tool_node = ToolNode([multiply,add]) # 这个是让模型知道有哪些工具可以用 model_with_tools = llm.bind_tools([multiply,add])

from typing import TypedDict from gradio import Image from langchain_core.prompts import PromptTemplate from langchain_core.tools import tool from langgraph.constants import START, END from langgraph.graph import MessagesState, StateGraph from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI import os from dotenv import load_dotenv load_dotenv() def get_client(): api_key = os.getenv("ali_api_key") model="qwen-plus" base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # langchain调用访问大模型 client = ChatOpenAI(api_key = api_key,base_url= base_url,model=model) return client llm = get_client() @tool def add(a:int,b:int): """计算两数相加""" return a+b @tool def mul(a:int,b:int): """两数相乘""" return a*b # 定义工具节点，调用工具 toolNode = ToolNode([add,mul],name="工具") # 让大模型知道有哪些工具使用 model_with_tools = llm.bind_tools([add,mul]) def call_model(state: MessagesState): messages = state["messages"] response = model_with_tools.invoke(messages) return {"messages": [response]} def use_tool(state: MessagesState): messages = state["messages"] last_ai_message = messages[-1] print("last_ai_message",last_ai_message) if last_ai_message.tool_calls: return "toolNode" return END builder = StateGraph(MessagesState) builder.add_node("call_model", call_model) builder.add_node("toolNode", toolNode) builder.add_edge(START, "call_model") # 条件边,判断是否需要一直调用工具 builder.add_conditional_edges("call_model", use_tool) # 如果调用了工具，把调用工具后的结果再给大模型，判断是否还需要调用工具 builder.add_edge("toolNode", "call_model") graph=builder.compile() res = graph.invoke({"messages": [{"role": "user", "content": "10+2x2=?"}]}) # print(res) # from IPython.display import Image, display # try: # display(Image(graph.get_graph().draw_png(output_file_path='./toolNode.png'))) # except: # pass

9.1 ToolNode工具调用失败后的处理

• 工具可能失败的原因：无效的参数，外部api不可用、工具内部逻辑问题等
• 如果不处理这些错误，程序直接报错，处理了，会根据代码里面的内容，模型返回友好提示
• 通过handle_tool_errors= True实现

# 当输入长沙，工具调用就失败了 @tool def get_weather(location: str): """获取当前天气.""" print('location：', location) if location == "SH": raise ValueError("输入查询必须是专有名词") elif location == "上海": return "气温23度，有雾." else: # 为什么不用return，因为工具需要返回正确的答案 raise ValueError("无效输入.") # tool_node = ToolNode([get_weather]) tool_node = ToolNode([get_weather],handle_tool_errors= True)

9.2.toolNode工具调用失败，转而使用新的工具节点

• 当第一个工具调用失败后，转而使用新的工具节点
• 要求第一个工具节点里面设定规则，比如调用错误后在toolmessage里面设置错误，第二个工具匹配到这个错误，开始使用工具

10.retry=RetryPolicy节点重试策略

• 当节点里面执行异常后，进行重试
• 可以自定义重试的异常，也可以所有异常都重试

builder.add_node("query_database",query_database,retry=RetryPolicy(retry_on=sqlite3.OperationalError)) builder.add_node("model", call_model, retry=RetryPolicy(max_attempts=5)) def query_database(state): try: # 数据库逻辑，执行出错会「自然抛出异常」 conn = sqlite3.connect("test.db") conn.execute("SELECT * FROM table") return {"messages": [{"role": "assistant", "content": "查询成功"}]} except Exception as e: # 这个except的触发时机：框架完成所有重试（3次）后，仍抛出异常时 # 此时重试已全部失败，捕获后「无需重抛」，自定义处理即可 error_info = f"数据库查询失败（已重试{2}次），原因：{str(e)}" # 返回失败状态，让图继续执行后续节点，不终止流程 return {"messages": [{"role": "assistant", "content": error_info}]} # 重试策略正常配置，会完整执行 builder.add_node( "query_database", query_database, retry=RetryPolicy(retry_on=Exception, max_attempts=3) # 1次原始+2次重试 )

11.消息的摘要和删除RemoveMessage处理

• 摘要发生在“把消息送进 LLM 之前”这一步。

• 删除消息是根据删除消息列表里面对应id实现的

图的方法调用

12.将图state状态注入给工具函数

• 注入状态：InjectedState，
• 注入后工具函数就能正常的取值使用
• Annotated是 Python 3.9+ 自带的类型注解增强器
• Annotated[真正类型, 元数据1, 元数据2, ...]
• InjectedState 是 LangGraph 提供的一个标记常量，含义： “这个参数不需要 LLM 填，由框架把当前图的 state 整字典注入进来。

# 存储方式 class State(AgentState): docs: List[str] @tool def get_context(question: str, state: Annotated[dict, InjectedState]): """获取回答问题的相关背景.""" return "\n\n".join(doc for doc in state["docs"])

13.将图第三方存储和配置注入给工具函数

• InjectedStore()+RunnableConfig
• 每次app.stream/app.invoke时传的config字典：在图内部会被自动封装成 RunnableConfig 实例

# 数据库方式 def get_context( question: str, config: RunnableConfig, # 参数1：注入运行时配置 store: Annotated[BaseStore, InjectedStore()], # 参数2：注入文档存储 ) -> Tuple[str, List[Document]]: """获取回答问题的相关背景.""" # 从运行时配置中获取 user_id user_id = config.get("configurable", {}).get("user_id") # 从注入的 store 中根据 user_id 搜索文档 docs = [item.value["doc"] for item in store.search(("documents", user_id))] return "\n\n".join(doc for doc in docs) # 返回拼接后的文档字符串 doc_store = InMemoryStore() graph = create_agent(model, tools, checkpointer=checkpointer, store=doc_store)

14.工具函数中更新state

工具函数不是节点，不能直接更新state

通过工具id和return command实现

# 自动注入tool_call的id tool_call_id: Annotated[str, InjectedToolCallId], return Command( update={ "user_info": user_info, "messages": [ ToolMessage( "成功查询用户信息", tool_call_id=tool_call_id ) ], } )

stream_mode的参数

15.大量工具如何选择调用

RAG思想

16.Milddle ware中间件

1.langchain中间件的作用

• 控制和自定义Agent执行的每一步，包括日志记录，工具选择，备选方案等

2.预定义中间件

Summarization：摘要
Human-in-the-loop：人机交互
PII detection：处理个人身份信息

3.自定义中间件：

基于装饰器的中间件

@before_agent / @after_agent:整个智能体（Agent）开始/结束执行一次任务前/后，仅运行一次

@before_model / @after_model:在每次调用大模型之前都会运行 / 在每次拿到大模型的回复之后都会运行

@wrap_model_call- 包裹住了整个模型调用过程，可以完全控制调用。
@wrap_tool_call- 包裹住了每次工具（如查询数据库、调用API）的执行过程。

@dynamic_prompt- 通常结合 @before_model或 @wrap_model_call使用，在模型调用前生效。