LangGraph定义节点（Node）

未完待续

1.Node概述

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'

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

微信支付

支付宝

LangGraph

#原创 #AI #大模型 #框架 #Agent #智能体 #LangGraph

LangGraph定义节点（Node）

https://blog.liuzijian.com/post/2026/06/28/langgraph-node/

作者

Liu Zijian

发布于

2026年6月28日

许可协议

使用tun2socks实现全局代理下一篇