05 Langgraph笔记
01 Node缓存
langgraph的node函数如果比较费时间，可以将node的计算结果缓存起来，避免多次重复计算； 启用方式：
1. 在compiel的时候传入cache对象
2. 在node传入cache_policy
示例：
from langgraph.cache.memory import InMemoryCache
def my_keyfunc(state):
return hash(some_field, ...)
def my_keyfunc(state, config):
return hash(some_state, some_config...)
builder.add_node(
"expensive_node",
expensive_node,
cache_policy=CachePolicy(ttl=3, key_func=my_keyfunc)
)
builder.compile(cache=InMemoryCache())
备注
缓存默认只根据node的入参state进行hash，不同的thread_id只要参数相同就会复用。 要区分需要在key_func中传入config，并且将thread_id作为hash计算的入参； key_func的定义时可以只传入state,也可以传入(state,config)
02 Edge和Route
静态路由： 通过add_edge实现；
动态路由： 通过在node中返回Command(goto=...)
动态并发： 【Send(node_a, {"channe1": value}) for value in some_list]
条件路由：
通过静态路由通过add_conditional_edge实现;
在node中动态判断根据不同的代码返回不同的Command
2.1 通过add_contional_edge实现条件分支
from langgraph.types import Send
def condition_check_func(state: AgentState, config: Optional[Config]):
"""
直接返回一个字符串：
1. 直接返回需要跳转的node的名字
2. 直接返回某个字符串，但是在add_conditional_edge的时候需要添加一个当前函数返回的字符串到node名字的映射
3. 返回一个包含Send的列表，用于支持动态并发
"""
if condition1 is True:
return "node_a"
else:
return "node_b"
def condition_check_func2(state: OverallState):
"""
根据上一层的返回数据，动态的开启并发。
"""
return [Send("generate_joke", {"subject": s}) for s in state['subjects']]
graph.add_conditional_edges("node_a", condition_check_func)
graph.add_conditional_edges("node_b", condition_check_func2)