LangGraph 06：循环重试与缓存

发表于 更新于 分类于
用条件边或 Command 做工具循环，结合 RemainingSteps、RetryPolicy 与 CachePolicy 控制递归与开销。

前言

Agent 调工具时常要「模型 → 工具 → 模型」多轮往返，直到不再需要 tool_calls。
循环若不设上限会撞上递归限制；偶发失败需要节点级重试；纯函数节点则适合结果缓存。
本文覆盖工具循环（条件边与 Command/goto）、RemainingSteps、RetryPolicy，以及 CachePolicy + InMemoryCache。
聊天模型对接 火山方舟 Coding Plan；重试与缓存示例可不调 LLM。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖（版本请按项目实际调整）。

1
2
3
4
uv init langgraph-loops-retry
cd langgraph-loops-retry
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langchain-core langgraph python-dotenv rich

在项目根目录创建 .env：

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

不要把 .env 提交进 Git。
后续示例用 uv run python xxx.py 运行。

实现

工具循环边

最常见写法：llm_node 后用条件边；有 tool_calls 进 tool_node，否则进 output_node；工具结果再回到 llm_node。
下面用假天气/新闻工具，并故意以约 60% 概率失败，迫使模型按系统提示重试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import os
import random
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


@tool
def get_weather(city: str = "上海") -> str:
    """查询指定城市当日天气。"""
    return f"{city}的天气是晴朗的"


@tool
def get_news(domain: Literal["AI", "食品安全"]) -> str:
    """查询特定领域的当日热点。"""
    if domain == "AI":
        return "某厂商发布了新一代编码模型。"
    return "某食品抽检项目公布最新通报。"


tools = [get_weather, get_news]
model_with_tools = model.bind_tools(tools)
FAIL_PROB = 6  # 0-9 中小于 6 则失败，约 60%


class OverAllState(MessagesState):
    user_input: str
    final_output: str


def input_node(state: OverAllState) -> OverAllState:
    return {"messages": [HumanMessage(state["user_input"])]}


def llm_node(state: OverAllState) -> OverAllState:
    return {"messages": [model_with_tools.invoke(state["messages"])]}


def tool_node(state: OverAllState) -> OverAllState:
    messages = list(state["messages"])
    ai_msg = messages[-1]
    for call in ai_msg.tool_calls:
        if random.randint(0, 9) < FAIL_PROB:
            messages.append(
                ToolMessage(
                    content="网络波动,调用失败,请重试",
                    tool_call_id=call["id"],
                )
            )
            continue
        fn = get_weather if call["name"] == "get_weather" else get_news
        messages.append(fn.invoke(call))
    return {"messages": messages}


def output_node(state: OverAllState) -> OverAllState:
    return {"final_output": state["messages"][-1].content}


def router(state: OverAllState) -> Literal["tool_node", "output_node"]:
    last = state["messages"][-1]
    return "tool_node" if getattr(last, "tool_calls", None) else "output_node"


builder = StateGraph(state_schema=OverAllState)
builder.add_node("input_node", input_node)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "input_node")
builder.add_edge("input_node", "llm_node")
builder.add_conditional_edges("llm_node", router)
builder.add_edge("tool_node", "llm_node")
builder.add_edge("output_node", END)

graph = builder.compile()
result = graph.invoke(
    {
        "user_input": "查询今天的上海天气和AI新闻热点",
        "messages": [
            SystemMessage("如果工具调用失败,必须重新调用直到成功为止")
        ],
    }
)
rprint(result["final_output"])

边循环清晰，适合先学图结构。
生产里更常见 ToolNode；这里手写是为了演示失败重试语义。

Command 循环

同一逻辑也可用 Command：在 llm_node 里根据是否有 tool_calls 设置 goto。
此时不必再挂 llm_node 的条件边，工具边仍回到 llm_node。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import os
import random
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.types import Command
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


@tool
def get_weather(city: str = "上海") -> str:
    """查询指定城市当日天气。"""
    return f"{city}的天气是晴朗的"


@tool
def get_news(domain: Literal["AI", "食品安全"]) -> str:
    """查询特定领域的当日热点。"""
    return "热点摘要占位。"


tools = [get_weather, get_news]
model_with_tools = model.bind_tools(tools)


class OverAllState(MessagesState):
    user_input: str
    final_output: str


def input_node(state: OverAllState) -> OverAllState:
    return {"messages": [HumanMessage(state["user_input"])]}


def llm_node(
    state: OverAllState,
) -> Command[Literal["tool_node", "output_node"]]:
    ai_msg = model_with_tools.invoke(state["messages"])
    goto = "tool_node" if ai_msg.tool_calls else "output_node"
    return Command(update={"messages": [ai_msg]}, goto=goto)


def tool_node(state: OverAllState) -> OverAllState:
    messages = list(state["messages"])
    ai_msg = messages[-1]
    for call in ai_msg.tool_calls:
        if random.randint(0, 9) < 6:
            messages.append(
                ToolMessage("网络波动,调用失败,请重试", tool_call_id=call["id"])
            )
        else:
            fn = get_weather if call["name"] == "get_weather" else get_news
            messages.append(fn.invoke(call))
    return {"messages": messages}


def output_node(state: OverAllState) -> OverAllState:
    return {"final_output": state["messages"][-1].content}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("input_node", input_node)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "input_node")
builder.add_edge("input_node", "llm_node")
builder.add_edge("tool_node", "llm_node")
builder.add_edge("output_node", END)

graph = builder.compile()
rprint(
    graph.invoke(
        {
            "user_input": "查询今天的上海天气和AI新闻热点",
            "messages": [
                SystemMessage("如果工具调用失败,必须重新调用直到成功为止")
            ],
        }
    )["final_output"]
)

Command 把「写消息 + 选下一跳」收进一个节点，图上的条件边更少。
两种循环语义等价，按团队习惯选型即可。

RemainingSteps

RemainingSteps 是托管字段：表示在 recursion_limit 约束下还剩多少超步。
可在路由里提前收尾，避免硬撞 GraphRecursionError。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import Literal, TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from langgraph.managed import RemainingSteps
from rich import print as rprint


class OverAllState(TypedDict):
    remaining_steps: RemainingSteps


def loop_node(state: OverAllState, config: RunnableConfig) -> OverAllState:
    step = config["metadata"]["langgraph_step"]
    rprint(f"loop_node step={step} remaining={state['remaining_steps']}")
    return {}


def router(state: OverAllState) -> Literal["loop_node", "__end__"]:
    if state["remaining_steps"] < 3:
        rprint("剩余超步不足 3，结束循环")
        return END
    return "loop_node"


builder = StateGraph(state_schema=OverAllState)
builder.add_node("loop_node", loop_node)
builder.add_edge(START, "loop_node")
builder.add_conditional_edges("loop_node", router)

graph = builder.compile()
graph.invoke({}, config={"recursion_limit": 10})

不必手动赋值 remaining_steps；运行时会注入并递减。
阈值可按业务改；示例在不足 3 步时主动 END。

递归上限

若节点边写回自身且从不退出，最终会触发 GraphRecursionError。
可用 config={"recursion_limit": N} 明确上限并捕获异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class EmptyState(TypedDict):
    pass


def loop_node(state: EmptyState, config: RunnableConfig) -> EmptyState:
    rprint(f"step={config['metadata']['langgraph_step']}")
    return {}


builder = StateGraph(state_schema=EmptyState)
builder.add_node("loop_node", loop_node)
builder.add_edge(START, "loop_node")
builder.add_edge("loop_node", "loop_node")

graph = builder.compile()
try:
    graph.invoke({}, config={"recursion_limit": 10})
except GraphRecursionError as exc:
    rprint(f"超步耗尽: {exc}")

工具循环务必保证「无 tool_calls → 出口」或依赖 RemainingSteps。
仅靠提高 recursion_limit 不是根治方案。

RetryPolicy

节点级 RetryPolicy 在节点抛错时自动重试，适合瞬时网络/HTTP 失败。
下面人为抛出 HTTPError，观察最多尝试 3 次后仍失败。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy
from requests import HTTPError
from rich import print as rprint


class EmptyState(TypedDict):
    pass


def node_a(state: EmptyState) -> EmptyState:
    rprint("node_a 运行中，准备失败")
    raise HTTPError("simulated")


builder = StateGraph(state_schema=EmptyState)
builder.add_node(
    "node_a",
    node_a,
    retry_policy=RetryPolicy(max_attempts=3, jitter=False),
)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

graph = builder.compile()
try:
    graph.invoke({})
except HTTPError as exc:
    rprint(f"重试耗尽: {exc}")

jitter=False 便于演示；生产可打开抖动减轻雪崩。
业务可预期的失败（如鉴权错误）通常不应无限重试。

节点缓存

对相同输入可缓存节点输出：编译时传入 cache=InMemoryCache()，节点上挂 CachePolicy(ttl=...)。
下面用 sleep 模拟耗时；第二次相同输入应明显更快。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import time
from operator import add
from typing import Annotated, TypedDict

from langgraph.cache.memory import InMemoryCache
from langgraph.graph import END, START, StateGraph
from langgraph.types import CachePolicy
from rich import print as rprint


class OverAllState(TypedDict):
    user: str
    invoke_counts: Annotated[int, add]


def node_a(state: OverAllState) -> OverAllState:
    rprint(f"node_a 执行 user={state['user']}")
    time.sleep(2)
    return {"invoke_counts": 1}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a, cache_policy=CachePolicy(ttl=10))
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

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

rprint("首次", graph.invoke({"user": "小明", "invoke_counts": 0}))
rprint("命中缓存", graph.invoke({"user": "小明", "invoke_counts": 0}))
rprint("不同输入", graph.invoke({"user": "小花", "invoke_counts": 0}))
time.sleep(11)
rprint("TTL 过期后", graph.invoke({"user": "小花", "invoke_counts": 0}))

缓存键与节点输入相关；换 user 会未命中。
ttl 到期后会重新执行节点；进程内 InMemoryCache 重启即清空。

验证

  1. 工具循环：最终 final_output 应综合天气与新闻，中间可能出现失败 ToolMessage。
  2. RemainingSteps：在剩余不足 3 时打印结束日志并正常返回。
  3. 缓存：同输入第二次几乎无 sleep；换用户或等 TTL 后再慢一次。

总结

  1. 循环：条件边或 Command(goto=...) 都能做工具往返。
  2. 收尾：用 RemainingSteps 主动退出，或捕获 GraphRecursionError。
  3. 韧性：瞬时错误用 RetryPolicy；纯计算节点用 CachePolicy 降本。