1.节点

在 LangGraph 中,节点(Node)是图里执行具体工作的基本单元。结合状态与 Pregel 算法,节点就是那个读取状态、执行逻辑、然后返回状态更新的“执行者”。

核心定义与签名

从代码层面看,节点的本质是一个函数(或可调用对象),其签名约定为:接收当前状态,返回部分更新。

  • 输入:节点拿到的是当前步骤的状态快照。
  • 输出:节点不需要返回完整的状态,只需返回它想要修改的字段(Partial Update)。LangGraph 会负责将这些更新合并回全局状态中。

节点的常见类型与实现形式

你可以把任何逻辑封装成节点,常见的实现形式包括:

  • 普通函数:最直接的方式。你可以写一个 Python 或 JS 函数,在里面调用大模型、查询数据库或运行工具,然后返回结果。
  • LangChain Runnable:任何遵循 LangChain Runnable 接口的对象(如 Chain、Retriever)都可以直接作为节点使用。
  • 可调用类:包含 __call__ 方法的类实例也可以作为节点,适合需要维护内部状态或复杂初始化的场景。
  • 异步节点:支持 async def 定义的异步函数,用于处理并发的 I/O 操作(如并行调用多个 API)。

特殊节点与高级特性

除了用户自定义的普通节点,LangGraph 还提供了一些内置或特殊类型的节点:

  • 预构建节点:LangGraph 提供了开箱即用的节点,例如 ToolNode。它专门用于自动执行工具调用,你无需自己编写工具调用的循环逻辑,直接将其加入图中即可。
  • 子图作为节点:一个编译后的图(本身也是一个 Pregel 实例)可以被嵌套在另一个图的节点中运行。这允许你将复杂的流程模块化,构建分层级的图结构。
  • 特殊路由节点(Command):虽然通常由边来决定路由,但节点也可以通过返回 Command 对象来同时更新状态并决定下一步跳转到哪个节点,从而将控制流逻辑内聚在节点内部。

与 Pregel 算法的关系

对于 Pregel 超级步模型:节点就是那个在 “执行阶段” 被调度的对象。在一个超级步中,被选中的所有节点会并行执行;它们各自基于同一个状态快照进行计算;执行完毕后,它们返回的更新会在 “更新阶段” 被统一提交。

2.节点的输入输出

2.1 节点输入

– 在LangGraph中,一般来讲,节点都是 Python 函数(可以是同步的,也可以是异步的),它们接受以下参数:

  • state:图的状态,代表了具体的业务数据;
  • config:一个RunnableConfig对象,包含诸如thread_id之类的配置信息,在调用图时,也可以传递用户定义的其他配置;
  • runtime:一个Runtime对象,包含运行时context(可自定context,在调用图时传入即可)以及其他信息,如store和stream_writer等;2

– 以上参数,会在运行过程中,自动被LangGraph运行时注入。
– 节点定义config和runtime入参如下所示,在运行时,这两个参数LangGraph名称会通过关键字传参方式进行传参,因此两个参数位置可以调换。

具体示例如下:

# 导入time模块,用来模拟处理耗时
import time
# TypedDict定义状态结构,Any任意类型,List列表注解
from typing import TypedDict, Any, List
# RunnableConfig:LangGraph标准的配置类型
from langchain_core.runnables import RunnableConfig
# StateGraph构建状态图,START起点,END终点
from langgraph.graph import StateGraph, START, END
# Runtime:LangGraph运行时对象,用来传递context上下文、流式输出writer
from langgraph.runtime import Runtime

# 模拟大模型类,代替真实LLM
class MockLLM:
    # invoke方法,接收prompt,直接返回拼接好的模拟回答
    def invoke(self, prompt: str):
        return f"AI Generated: Answer for '{prompt}'"

# 模拟数据库类
class MockDatabase:
    # 根据user_id获取用户信息,简单判断是否是vip用户
    def get_user_info(self, user_id: str):
        return {"id": user_id, "role": "vip" if "vip" in user_id else "standard"}

# 定义客服流程的状态结构
class CustomerSupportState(TypedDict):
    query: str        # 用户原始提问
    response: str      # 最终给用户的客服回复
    log: List[str]     # 流程处理日志,记录关键步骤

# 客服业务节点函数
# state:状态;config:配置(放user_id等);runtime:运行时对象,承载context和stream writer
def node_customer_service(state: CustomerSupportState, config: RunnableConfig, runtime: Runtime) -> dict:
    """
    客服节点:展示如何从 config 中获取注入的依赖 (LLM, DB) 并使用 runtime 推送进度。
    """
    # 取出用户问题
    user_query = state["query"]
    
    # 1、从runtime当中获取context对象,context就是我们外部注入进来的依赖
    llm = runtime.context['llm']
    db = runtime.context['db']

    # 取出config里的configurable配置块
    configurable = config.get("configurable", {})
    # 获取用户id,取不到就默认是guest访客
    user_id = configurable.get("user_id", "guest")
    
    print(f"\n[Node] 开始处理,User ID: {user_id}")
    
    # 2、验证依赖是否存在,做简单容错
    if not llm or not db:
        return {"response": "System Error: Dependencies not injected!", "log": ["Error"]}

    # 3、使用db对象查看用户角色
    user_info = db.get_user_info(user_id)
    user_tier = user_info.get("role")
    print(f"[Node] 从 DB 获取用户角色: {user_tier}")

    # 4、使用 runtime.writer 向图外输出自定义流式数据(可以前端展示进度、状态)
    writer = runtime.stream_writer
    writer({"status": "thinking", "message": f"正在调用 LLM 为 {user_tier} 用户生成回复..."})
    # 休眠0.5秒模拟LLM思考耗时
    time.sleep(0.5)

    # 5、根据用户角色构建不同的 Prompt,并模拟 LLM 调用
    prompt = f"User({user_tier}) asks: {user_query}"
    llm_response = llm.invoke(prompt)
    
    # 返回字典,更新state中的response和log
    return {
        "response": llm_response,
        "log": [f"Processed by {llm.__class__.__name__}"]
    }

    # 这个return永远不会执行,属于冗余代码,可以删掉
    return None

# 运行演示的入口函数
def run_demo():
    # 创建图构造器,绑定状态结构
    builder = StateGraph(CustomerSupportState)
    # 注册客服节点
    builder.add_node("customer_service", node_customer_service)
    # 连线:起点到客服节点
    builder.add_edge(START, "customer_service")
    # 客服节点执行完直接结束
    builder.add_edge("customer_service", END)
    # 编译图
    graph = builder.compile()
    
    # 初始化外部依赖对象:模拟LLM、模拟DB
    my_llm = MockLLM()
    my_db = MockDatabase()
    
    print("==================================================")
    print("场景: 依赖注入演示")
    print("演示目标: 将 LLM 和 DB 对象通过 runtime.context 注入到节点中")
    print("==================================================")
    
    # 准备初始状态:用户的提问
    initial_state = {"query": "如何升级会员?"}
    
    # user_id为一个配置信息,将其放入到config当中
    config = {
        "configurable": {
            "user_id": "vip_user_999",
        }
    }
    
    print("[System] 开始运行图,并注入依赖对象...")
    
    # 组装要注入的上下文依赖,会放进runtime.context给节点使用
    context = {
        "llm": my_llm,
        "db": my_db
    }
    # 执行图,额外传入context参数实现依赖注入
    result = graph.invoke(initial_state,config=config,context=context)
    # 打印最终返回的状态结果
    print(result)
    

if __name__ == "__main__":
    run_demo()

运行结果 :
在这里插入图片描述

2.2 节点输出

节点的输出为当前节点对状态的增量更新,而不能将接收到的整个状态实例返回出去。
LangGraph 底层会将所有节点输出的状态,都作为增量状态,并尝试和当前的全局状态做一次合并操作。如果将整个状态都输出,对于没有配置任何 reducer 的状态键,langgraph 底层无法完成合并,会抛出异常;对于配置了 reducer 的状态键,如果节点输出了不属于该节点更新的状态,也会导致数据产生问题。

以下演示错误返回了整个状态的问题:

# 导入类型注解工具 TypedDict,用来定义状态结构
from typing import TypedDict
# 导入LangGraph内置常量:START 图起点、END 图终点
from langgraph.constants import START, END
# 导入状态图核心类 StateGraph
from langgraph.graph import StateGraph


# 定义图的状态结构体 MyState,继承 TypedDict
# 状态里保存整个流程流转过程中所有需要共享的数据
class MyState(TypedDict):
    query: str         # 用户原始提问
    file_result: str   # 本地文件检索结果
    web_result: str    # 联网搜索结果
    final_answer: str  # LLM 融合信息后的最终回答


def query_web(state: MyState) -> dict:
    """
    【节点函数】做网络搜索,返回搜索结果
    ⚠️ 错误演示写法:直接return整个state,而非只返回需要增量修改的字段
    LangGraph 的设计理念:节点只返回【要更新的部分状态】,而不是把完整state返回
    """
    # 从传入的共享状态中取出用户问题
    query = state['query']
    # 修改state里的web_result字段
    state['web_result'] = f'{query}的网络搜索结果'
    # 把整个state字典返回(不推荐的写法,仅用来演示)
    return state


def query_file(state: MyState) -> dict:
    """
    【节点函数】做本地文件检索,返回文件内检索结果
    ⚠️ 和上面一样:直接返回完整 state,属于不推荐写法
    """
    # 获取用户查询问题
    query = state['query']
    # 填充文件检索结果到状态
    state['file_result'] = f'{query}的文件搜索结果'
    # 返回完整state
    return state


def answer(state: MyState) -> dict:
    """
    【节点函数】汇总web和文件两路结果,交给LLM生成最终答案
    这个节点会在 query_web、query_file 两个节点都跑完之后才执行
    """
    # 读取两路检索结果
    web_result = state['web_result']
    file_result = state['file_result']
    # 拼接生成最终回答
    final_answer = f'LLM基于{web_result},{file_result} 的最终结果'
    # 写入状态
    state['final_answer'] = final_answer
    return state


# 实例化状态图,指定状态结构为 MyState
graph = StateGraph(state_schema=MyState)

# 向图中注册3个节点:节点名默认和函数名一致
graph.add_node(answer)
graph.add_node(query_web)
graph.add_node(query_file)

# 【并行分支】START起点同时指向 query_web 和 query_file
# LangGraph 这里会并行执行这两个节点(双分支并行检索)
graph.add_edge(START, 'query_web')
graph.add_edge(START, 'query_file')

# 两个分支都执行完成之后,才进入 answer 节点(汇聚节点)
graph.add_edge('query_web', 'answer')
graph.add_edge('query_file', 'answer')

# 编译图,生成可执行的runnable对象
compiled_graph = graph.compile()

# 调用图,传入初始状态:只有query字段
res = compiled_graph.invoke({"query": "什么是Langgraph"})

# 打印最终回答
print(res['final_answer'])

3.特殊节点

START 和 END 节点都是 LangGraph 当中的特殊节点。
START 它代表着将用户输入发送到图中的节点。引用此节点的主要目的是确定应首先调用哪些节点。
END节点是一个特殊节点,代表终止节点。当想表示哪些边在完成后没有动作时,会引用这个节点。
START 和 END 节点本质上是一个字符串,如下所示:

END = sys.intern("__end__")
"""The last (maybe virtual) node in graph-style Pregel."""
START = sys.intern("__start__")
"""The first (maybe virtual) node in graph-style Pregel."""

START 作为一个特殊 节点,在图执行完START节点之后,也会创建一个 checkpoint

4 节点缓存

LangGraph 支持基于节点输入对节点进行缓存。对于配置了缓存的节点,且缓存结果没有过期,以相同的输入再次调用节点时,可直接从缓存当中读取结果,不需要再进行节点计算。
使用缓存的方法如下:

  • (1)编译图时指定缓存存储后端:langgraph.cache 包当中提供了内存级别的缓存后端,Redis 缓存后端,和 Sqlite 缓存后端的具体实现,需要构造相关实例,并在图编译时传入。
  • (2)为节点指定缓存策略。每个缓存策略需配置:
    • key_func:用于根据节点的输入生成缓存键,默认情况下是使用 pickle 对输入进行 hash 运算的结果。
    • ttl:缓存的生存时间(以秒为单位)。如果未指定,缓存将永不过期。
import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache

from langgraph.types import CachePolicy
from langgraph.constants import START
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    x: int
    result: int

builder = StateGraph(State)
checkpointer = InMemorySaver()

# 1、定义节点:模拟一些有耗时计算的节点
def expensive_node(state: State) -> dict[str, int]:
    # expensive computation
    print(f"expensive_node 被调用")
    time.sleep(5)
    print(f"expensive_node 计算完成")
    return {"result": state["x"] * 2}

# 2、添加节点时,为节点配置缓存策略,这里设置为10秒缓存,缓存键的配置同样使用默认的方式,可以在CachePolicy当中传入key_func配置不同的缓存键生成策略
builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=10))
builder.add_edge(START, "expensive_node")

# 3、图编译时,传入缓存器,此处使用InMemoryCache,也可以使用RedisCache等
graph = builder.compile(cache=InMemoryCache(),checkpointer=checkpointer)

# 4、第一次调用,传入的状态为5
print(graph.invoke({"x": 5},config={"configurable":{"thread_id":"1"}}))

# 5、第二次调用,传入相同的状态,由于缓存策略,会直接从缓存中返回结果,而不会重新调用节点
print(graph.invoke({"x": 5},config={"configurable":{"thread_id":"2"}}))

5.节点重试

– 在很多使用场景中,有些节点由于客观原因限制,导致其执行过程是不稳定的。因此,我们可能希望节点拥有自定义的重试策略,例如在调用API、查询数据库或调用大语言模型(LLM)等情况下。
– 为节点添加重试策略,需要在 add_node 中设置 retry_policy 参数。retry_policy 参数接受一个 RetryPolicy 命名元组对象。
– RetryPolicy 对象有两个属性:max_attempts 和 retry_on参数,前者定义了总共重试次数,后者定义了对于哪些异常类型进行重试。

示例代码如下:

import random
from typing import Dict, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

# ===================== 定义图状态 =====================
# State:节点之间传递的状态结构,只有一个 result 字段保存接口返回结果
class State(TypedDict):
    result: str

# 全局计数器,用来记录 unstable_api_call 已经尝试调用多少次
attempt_counter = 0


def unstable_api_call(state: State) -> Dict[str, Any]:
    """
    模拟一个不稳定的API调用,有概率抛出异常失败,用来测试重试机制
    :param state: 当前图状态
    :return: 增量更新的状态字典(LangGraph规范写法,只返回要更新的字段)
    """
    global attempt_counter
    attempt_counter += 1
    print(f"尝试调用API,这是第 {attempt_counter} 次尝试")

    # 业务逻辑:前2次调用主动抛出异常模拟失败,第3次调用成功
    if attempt_counter < 3:
        raise Exception(f"模拟API调用失败 (尝试 {attempt_counter})")
    else:
        # 第三次尝试成功,返回需要更新的状态字段
        return {
            "result": f"API调用成功,经过 {attempt_counter} 次尝试"
        }


def value_error_call(state: State) -> Dict[str, Any]:
    """
    模拟抛出 ValueError 的节点
    ✅ 重点:LangGraph 默认重试策略【不会重试ValueError】这类内置异常
    """
    print("调用会抛出 ValueError 的节点")
    raise ValueError("模拟 ValueError 异常")


def run_demo():
    print("=== LangGraph 节点重试策略演示 ===\n")

    # 重置全局计数器,保证每个演示之间互不干扰
    global attempt_counter
    attempt_counter = 0

    # ---------------------- 演示1:默认重试策略 ----------------------
    print("1. 使用默认重试策略:")
    print("   默认策略会对除特定异常外的所有异常进行重试")
    print("   不会重试的异常包括: ValueError, TypeError, ArithmeticError, ImportError,")
    print("                     LookupError, NameError, SyntaxError, RuntimeError,")
    print("                     ReferenceError, StopIteration, StopAsyncIteration, OSError\n")

    builder1 = StateGraph(State)

    # 注册节点,绑定重试策略
    # retry_policy=RetryPolicy(max_attempts=5):最多总共尝试5次
    builder1.add_node(
        "unstable_call",
        unstable_api_call,
        retry_policy=RetryPolicy(max_attempts=5)
    )

    # 连线:起点 -> unstable_call -> 结束
    builder1.add_edge(START, "unstable_call")
    builder1.add_edge("unstable_call", END)

    # 编译图
    graph1 = builder1.compile()

    print("测试默认重试策略:")
    try:
        # 执行图,传入初始状态
        result = graph1.invoke({"result": ""})
        print(f"最终结果: {result}\n")
    except Exception as e:
        print(f"最终失败: {type(e).__name__}: {e}\n")

    # ---------------------- 演示2 预留位置:自定义重试策略 ----------------------
    print("2. 使用自定义重试策略:")
    print("   自定义策略只对特定错误进行重试\n")
    # 重置计数器(这里代码里演示2逻辑是空的,你后续可以在这里补充)
    attempt_counter = 0

    # ---------------------- 演示3:测试默认不重试的异常(ValueError) ----------------------
    print("3. 测试不会重试的异常类型:")

    builder3 = StateGraph(State)

    # 注册抛ValueError的节点,同样配置最多尝试3次
    builder3.add_node(
        "value_error_call",
        value_error_call,
        retry_policy=RetryPolicy(max_attempts=3)
    )

    # 连线
    builder3.add_edge(START, "value_error_call")
    builder3.add_edge("value_error_call", END)

    graph3 = builder3.compile()

    print("测试 ValueError(默认策略不会重试):")
    try:
        result = graph3.invoke({"result": ""})
        print(f"最终结果: {result}\n")
    except Exception as e:
        # 只会抛一次异常,不会自动重试
        print(f"最终失败: {type(e).__name__}: {e}\n")


if __name__ == "__main__":
    # 程序入口,运行整套重试demo
    run_demo()

补充

  1. RetryPolicy 参数
    • max_attempts:最大尝试次数(包含第一次原始调用)
    • retry_on:自定义要捕获并重试的异常类型列表,就是演示 2 要写的核心参数
    • backoff:退避策略(重试间隔,指数退避等)
  2. 默认规则:
    • 普通 Exception 子类(网络异常、自定义异常)默认会重试
    • ValueError / TypeError / SyntaxError 这类内置语法 / 参数类异常,LangGraph 默认判定为代码问题,不是临时故障,不重试

6.图内外数据传递

– 默认情况下,通过 graph.invoke 调用图时,仅在整个图的执行过程都结束之后,我们才能够拿到最终的状态,那么如果想要在图的执行过程当中,想要获取到图内部所产生的数据,应该如何实现?
– 考虑 Agent 的一个场景:首先,我们希望在LLM 输出时,就能够拿到 LLM 所产生的 token,在前端进行展示;其次,如果 Agent 的流程较长,我们希望能够拿到,当前正在执行的节点或者流程是什么。
– 由于 LangGraph 实现了 langchain_core 当中的 Runnable 接口,其为我们提供了 stream 和 astream 方法,也即流式输出方法,通过流式输出,就能解决前面所说到的问题。
Stream方法提供了多种不同的模式,如下表所示:

模式描述
values每一个执行后,流式输出完整的状态
updates图执行过程中,每一步执行后流式输出增量更新。如果在同一个步当中产生了多个增量更新,这些增量更新会分别流式输出。
custom流式输出节点内部的自定义数据。
messages在任何调用了LLM的节点当中,流式输出两元组数据:(LLM Token,metadata)
debug流式输出所有能输出的信息
混合模式流模式传入列表,在列表当中添加多种不同的模式,可以得到多种流式输出

7.人工审核节点

Agent 在完成用户设定的任务时,有时候我们希望用户参与当中部分重要决策过程。LangGraph 为此提供了一个非常方便的原语:interrupt。
其原理如下图所示:
在这里插入图片描述
可以在节点内部任意位置引入 interrupt,当节点执行到interrupt 所在位置时,就会停止执行,并可以得到从图内传递出来的,需要用户审核的数据,在用户进行相关审核完成后,又可通过 graph.invoke,让图继续往下执行。

需要注意的是,使用 interrupt 时,在第二次调用 graph.invoke(command) 继续执行时,有 interrupt 的函数,会从函数起点往下执行,如果在函数当中,有更新数据、调用 API 接口等相关操作时,会造成多次重复执行的情况。因此,一定要保证这种类型操作的幂等性,或者是将其封装到一个单独的节点当中,避免在恢复时再次执行。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐