LangGraph Node 节点

1.概述

LangGraph中，节点（Node）就是一个python函数，可以是同步的也可以是异步的，接受以下参数：

  • State 图的状态
  • Config 一个RunnableConfig对象，包含诸如thread_id之类的配置信息以及诸如tags之类的跟踪信息
  • Runtime: 一个Runtime对象，包含运行时context以及其他信息，如store和stream_writer

定义好节点后，通过add_node()将节点指定名称并添加进图中，如果未指定名称，系统默认分配一个和函数同名的默认名称。

节点涉及应该符合以下原则：

  • 单一职责原则：每个节点只负责一件事，避免功能过于复杂
  • 无状态：节点本身不保存状态，所有状态由State保存
  • 幂等性：相同输入应当保证相同输出
  • 可测试性：节点逻辑应当易于单元测试

2.定义Node

node就是一个干活的函数，除了传入State还支持传入其他参数

from functools import partial
from typing import TypedDict

from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END


class DemoState(TypedDict):
    user_input: str
    resp: str
    count: int
    process_data: dict

def node1(state: DemoState, param1: int, param2: str) -> dict:
    return {"resp":"sin(x)"}


if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)

    process_with_param = partial(node1, param1=1, param2='hello')

    graph.add_node('node1', process_with_param)

    graph.add_edge(START, 'node1')
    graph.add_edge('node1', END)

    app = graph.compile()

    res = app.invoke({
        'user_input': '什么是正弦函数'
    })
    print(res)


{'user_input': '什么是正弦函数', 'resp': 'sin(x)'}

3.节点缓存

LangGraph支持基于节点输入对节点进行缓存，为节点指定缓存策略，每个缓存策略支持：

.venv\Lib\site-packages\langgraph\types.py

@dataclass(**_DC_KWARGS)
class CachePolicy(Generic[KeyFuncT]):
    """Configuration for caching nodes."""

    key_func: KeyFuncT = default_cache_key  # type: ignore[assignment]
    """Function to generate a cache key from the node's input.
    Defaults to hashing the input with pickle."""

    ttl: int | None = None
    """Time to live for the cache entry in seconds. If `None`, the entry never expires."""

  • key_func

    用于根据节点的输入生成缓存键，默认情况下是使用pickle对输入进行hash运算的结果，当一个节点开始执行时，系统会使用其配置的key_func根据当前节点的输入数据生成一个唯一的键，LangGraph会检查缓存中是否存在这个键。如果存在（缓存命中），则直接返回之前存储的结果，跳过该节点的实际执行，如果不存在（缓存未命中），则正常执行节点函数，并将结果与缓存键关联后存入缓存。

  • ttl

    即缓存的生存时间（以秒为单位），如果未指定，缓存将永不过期，ttl参数能控制缓存的有效期。例如，对于依赖实时数据的天气查询节点，可以设置较短的ttl（如60秒）。而对于处理静态信息或变化不频繁数据的节点，则可以设置较长的 ttl甚至不设置（None），让缓存永久有效，直到手动清除

例：节点缓存

设置节点结果缓存10秒钟：cache_policy=CachePolicy(ttl=10)
采用内存进行缓存：graph.compile(cache=InMemoryCache())

未超时，访问节点不会触发重新计算，但是超时了就会触发节点重新计算

import time
from functools import partial
from typing import TypedDict

from langgraph.cache.memory import InMemoryCache
from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END
from langgraph.types import RetryPolicy, CachePolicy
from requests import RequestException, Timeout

class DemoState(TypedDict):
    count: int

def node1(state: DemoState) -> dict:
    print("节点计算：state['count'] * 2")
    time.sleep(3)
    return {'count': state['count'] * 2}


if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)


    graph.add_node(node='node1', action=node1, cache_policy=CachePolicy(ttl=10))

    graph.add_edge(START, 'node1')
    graph.add_edge('node1', END)

    app = graph.compile(cache=InMemoryCache())

    print('*' * 30)
    res = app.invoke({'count': 2})
    print('执行1', res)
    res = app.invoke({'count': 2})
    print('执行2', res)

    time.sleep(10) #模拟超时，触发重新计算
    res = app.invoke({'count': 2})
    print('执行3', res)



******************************
节点计算：state['count'] * 2
执行1 {'count': 4}
执行2 {'count': 4}
节点计算：state['count'] * 2
执行3 {'count': 4}

4.异常重试机制

LangGraph提供了错误处理和重试机制来指定重试次数，重试间隔，重试异常等，通过add_node(, retry_policy=RetryPolicy() )设置异常重试

RetryPolicy有以下几个参数：

  • max_attempts 最大重试次数

  • initial_interval 初始间隔

  • jitter 抖动(True/False)

  • backoff_factor 退避乘数，每次重试时间间隔的增长倍数

  • retry_on 默认值是default_retry_on默认会对所有除下方以外的异常进行重试，还可以是数组[], 只重试固定的异常，还可以自定义指定。

    ValueError,
TypeError,
ArithmeticError,
ImportError,
LookupError,
NameError,
SyntaxError,
RuntimeError,
ReferenceError,
StopIteration,
StopAsyncIteration,
OSError

例：对异常Exception('custom error')重试三次，模拟API联网调用失败

from functools import partial
from typing import TypedDict

from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.constants import END
from langgraph.types import RetryPolicy
from requests import RequestException, Timeout

class DemoState(TypedDict):
    age: int

total_count = 0

def node1(state: DemoState) -> dict:
    global total_count
    if total_count < 3:
        total_count += 1
        print('发生异常了， total_count：', total_count)
        raise Exception('custom error')

    print('返回')
    return state


retry_policy = RetryPolicy(
    max_attempts=3,    #最大重试次数
)

if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)
    graph.add_node('node1', node1, retry_policy=retry_policy)

    graph.add_edge(START, 'node1')
    graph.add_edge('node1', END)

    app = graph.compile()
    res = app.invoke({})
    print(res)
发生异常了， total_count： 1
发生异常了， total_count： 2
发生异常了， total_count： 3
Traceback (most recent call last):
  ......
Exception: custom error
During task with name 'node1' and id 'cdbe84ca-75b7-6c24-54c8-f494ace679af'

5.命令（Command）

命令（Command），是一个控制图执行的多功能原语，是一种节点返回的特殊形式，不同于之前的node函数直接返回State对象，Command把更新状态和控制流程跳转合并在一个对象同时完成，可以更新State的同时将流程路由到一个节点。

Command接受四个参数:

  • update 应用状态更新
  • goto 导航到特定节点
  • graph 从子图导航时以父图作为目标
  • resume 提供一个值，以便在中断后恢复执行

Command用于三种场景:

  1. 从节点返回，使用update/goto/graph将状态更新和流程控制结合起来
  2. 作为invoke或stream的输入，使用resume在中断后继续执行
  3. 从工具返回，类似从节点返回，在工具内部结合状态更新和控制流程

普通dict返回和Command返回的区别：

普通dict返回 Command返回
更新状态 支持 通过update={}
控制跳转 不支持，依赖外部add_conditional_edges() 通过goto=或goto=END
动态路由 需要单独的路由函数 节点内部直接决定下一步

例：

from typing import Annotated, TypedDict

from langgraph.graph import StateGraph
from langgraph.types import Command
from langgraph.constants import START
from langgraph.constants import END

# 全局常量：统一递归限制，便于维护
RECURSION_LIMIT = 50

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]  # 自动合并消息
    current_agent: str
    task_completed: bool

# 决策节点
def decision_agent(state: AgentState) -> Command[AgentState]:
    print("决策节点: decision_agent => ", state)

    last_message = state['messages'][-1][1]

    # 先判断流程是否早已经结束了
    if state["task_completed"]:
        print("decision_agent 任务已完成，终止流程")
        return Command(
            update={"messages": [("system", "所有任务处理完成，流程正常结束")]},
            goto=END
        )

    if '数学' in last_message:
        print("跳到数学解题节点")
        return Command(
            update={
                "messages": [("system","路由到模拟数学计算")] ,
                "current_agent": "math_agent"
            },
            goto='math_agent'
        )
    elif '翻译' in last_message:
        print("跳转翻译节点")
        return Command(
            update={
                "messages": [("system","路由到模拟英语翻译")] ,
                "current_agent": "translation_agent"
            },
            goto='translation_agent'
        )
    else:
        print("未识别任务类型，标记任务完成并终止")
        return Command(
            update={
                "messages": [("system", "标记任务完成并终止")],
                "task_completed": True
            },
            goto=END
        )



def math_agent(state: AgentState) -> Command[AgentState]:
    print("执行节点: math_agent => ", state)
    return Command(
        update={
            "current_agent": "math_agent",
            "messages": [("assistant", "模拟数学计算完成")],
            "task_completed": True
        },
        goto='decision_agent'
    )




def translation_agent(state: AgentState) -> Command[AgentState]:
    print("执行节点: translation_agent => ", state)
    return Command(
        update={
            "current_agent": "translation_agent",
            "messages": [("assistant", "模拟英语翻译完成")],
            "task_completed": True
        },
        goto='decision_agent'
    )





if __name__ == '__main__':
    graph = StateGraph(AgentState)


    graph.add_node('decision_agent', decision_agent)
    graph.add_node('math_agent', math_agent)
    graph.add_node('translation_agent', translation_agent)

    graph.add_edge(START, 'decision_agent')
    graph.add_edge('decision_agent', END)

    graph.add_edge('math_agent', 'decision_agent')
    graph.add_edge('translation_agent', 'decision_agent')

    app = graph.compile()

    app.get_graph().print_ascii()

    initial_state = {
        "messages": [("user", "我需要翻译")],
        "current_agent": "user",
        "task_completed": False
    }
    
    cfg  = {
        "recursion_limit": RECURSION_LIMIT
    }

    print("初始状态:", initial_state)
    result = app.invoke(initial_state, config=cfg)
    print('最终状态: ', result)

  +-----------+    
  | __start__ |    
  +-----------+    
         *         
         *         
         *         
+----------------+ 
| decision_agent | 
+----------------+ 
         *         
         *         
         *         
    +---------+    
    | __end__ |    
    +---------+    
初始状态: {'messages': [('user', '我需要翻译')], 'current_agent': 'user', 'task_completed': False}
决策节点: decision_agent =>  {'messages': [('user', '我需要翻译')], 'current_agent': 'user', 'task_completed': False}
跳转翻译节点
执行节点: translation_agent =>  {'messages': [('user', '我需要翻译'), ('system', '路由到模拟英语翻译')], 'current_agent': 'translation_agent', 'task_completed': False}
决策节点: decision_agent =>  {'messages': [('user', '我需要翻译'), ('system', '路由到模拟英语翻译'), ('assistant', '模拟英语翻译完成')], 'current_agent': 'translation_agent', 'task_completed': True}
decision_agent 任务已完成，终止流程
最终状态:  {'messages': [('user', '我需要翻译'), ('system', '路由到模拟英语翻译'), ('assistant', '模拟英语翻译完成'), ('system', '所有任务处理完成，流程正常结束')], 'current_agent': 'translation_agent', 'task_completed': True}

"如果文章对您有帮助，可以请作者喝杯咖啡吗？"

微信二维码

微信支付

支付宝二维码

支付宝


LangGraph
#原创 #AI #大模型 #框架 #Agent #智能体 #LangGraph
LangGraph Node 节点
https://blog.liuzijian.com/post/2026/06/28/langgraph-node/
作者
Liu Zijian
发布于
2026年6月28日
许可协议