LangGraph 10：静态断点与 ToolNode

发表于 分类于
讲解 interrupt_before/after 静态断点，对比手写工具循环与 ToolNode，并简述 wrap_tool_call 重试与工具缓存。

前言

上一篇的 interrupt() 是动态人机协同：停在哪儿由代码运行时决定。
还有一类静态断点：编译图时声明「进入某节点前 / 离开某节点后」一律暂停，适合调试或固定审批闸门。
工具侧则不必每次手写 tool_calls 循环；ToolNode 可直接当节点，并用 wrap_tool_call 做重试或缓存。
本文覆盖 interrupt_before / interrupt_after、手写工具节点与 ToolNode 对照，以及包装器要点。
示例对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。

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

在项目根目录配置 Coding Plan 的 .env（字段与上一篇相同）。
静态断点同样需要 checkpointer 与稳定的 thread_id。

静态断点

在 compile 时传入节点名列表。
首次 invoke 会在第一个断点停下；之后用 invoke(None, config=...) 推进到下一闸门，直到跑完。

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
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

class OverAllState(TypedDict):
    final_res: str

def node_a(state: OverAllState) -> OverAllState:
    return {"final_res": "node_a"}

def node_b(state: OverAllState) -> OverAllState:
    return {"final_res": "node_b"}

def node_c(state: OverAllState) -> OverAllState:
    return {"final_res": "node_c"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"],
)
config = {"configurable": {"thread_id": "static-1"}}

rprint(graph.invoke({}, config=config))
rprint(graph.invoke(None, config=config))
rprint(graph.invoke(None, config=config))
rprint(graph.invoke(None, config=config))

也可在单次 invoke(..., interrupt_before=..., interrupt_after=...) 里临时指定断点，不必写进 compile。
注意：临时列表与编译期配置混用时，某次调用若省略断点参数，后续行为可能与预期不一致；联调时固定一种写法更稳妥。
并行图上声明多个静态断点时，各分支会各自卡在闸门，恢复顺序仍按 invoke(None) 逐步推进。

手写工具循环

经典 Agent 环：LLM 节点 → 条件路由 → 手写 tool_node 解析 tool_calls → 再回 LLM。
下面用 Coding Plan 绑定两个本地工具。

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
import os
from typing import Literal

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

load_dotenv()

@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"

@tool(parse_docstring=True)
def get_news(home_or_abroad: bool) -> str:
    """查询国内外新闻。

    Args:
        home_or_abroad: True 国内，False 国外
    """
    return "国内要闻摘要" if home_or_abroad else "国际要闻摘要"

tools = [get_weather, get_news]
tools_by_name = {t.name: t for t in tools}

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

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

def tool_node(state: MessagesState) -> MessagesState:
    last = state["messages"][-1]
    outs: list[ToolMessage] = []
    for call in last.tool_calls:
        result = tools_by_name[call["name"]].invoke(call["args"])
        outs.append(
            ToolMessage(name=call["name"], content=result, tool_call_id=call["id"])
        )
    return {"messages": outs}

def router(state: MessagesState) -> Literal["tool_node", "__end__"]:
    if state["messages"][-1].tool_calls:
        return "tool_node"
    return END

builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "llm_node")
graph = builder.compile()

res = graph.invoke(
    {"messages": [HumanMessage("今天北京天气怎么样？国内有什么新闻？")]}
)
for msg in res["messages"]:
    msg.pretty_print()

手写循环灵活，但要自己处理并行 tool_calls、异常与消息字段，容易漏。

ToolNode

把上面的 tool_node 换成预构建 ToolNode(tools=...)，图结构不变。
ToolNode 会批量执行调用并生成 ToolMessage。

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
import os
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv()

@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"

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

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

def router(state: MessagesState) -> Literal["tool_node", "__end__"]:
    return "tool_node" if state["messages"][-1].tool_calls else END

builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", ToolNode(tools=tools))
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "llm_node")
graph = builder.compile()

graph.invoke({"messages": [HumanMessage("北京今天天气？")]})

工具函数还可注入 ToolRuntime，返回 Command(update=...) 同时改自定义状态字段与 messages，适合把天气结果写到独立键里。

wrap_tool_call

ToolNode(..., wrap_tool_call=fn) 可在真正 execute(request) 前后插入逻辑。
常见用途：有限次重试、统一打日志、按参数做结果缓存。

下面演示失败重试：工具随机抛 ConnectionError，包装器最多尝试 context.max_attempts 次。

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
import os
import random
from dataclasses import dataclass
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv()

@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    if random.randint(1, 10) < 8:
        raise ConnectionError("网络波动")
    return f"{city} 天气晴朗，微风"

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

@dataclass
class UserContext:
    max_attempts: int

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

def router(state: MessagesState) -> Literal["tool_node", "__end__"]:
    return "tool_node" if state["messages"][-1].tool_calls else END

def wrap_tool_call(request, execute):
    max_attempts = request.runtime.context.max_attempts
    tool_call_id = request.runtime.tool_call_id
    tool_msg = None
    for _ in range(max_attempts):
        try:
            tool_msg = execute(request)
            break
        except ConnectionError:
            continue
    if tool_msg is None:
        tool_msg = ToolMessage(
            tool_call_id=tool_call_id,
            content="调用次数达到上限，调用失败",
        )
    return tool_msg

builder = StateGraph(state_schema=MessagesState, context_schema=UserContext)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", ToolNode(tools=tools, wrap_tool_call=wrap_tool_call))
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "llm_node")
graph = builder.compile()

graph.invoke(
    {"messages": [HumanMessage("今天北京天气怎么样？")]},
    context=UserContext(max_attempts=3),
)

缓存版思路相同：用 (tool_name, json.dumps(args)) 做键，命中则直接构造 ToolMessage，未命中再 execute(request) 并写入进程内 dict。
注意缓存只适合幂等、短时效工具；跨进程要换 Redis 等外部存储。

总结

  1. 静态断点：interrupt_before / interrupt_after + checkpointer；恢复用 invoke(None, config=...)。
  2. 动态 interrupt() 适合业务审批；静态断点适合固定闸门与调试。
  3. 优先用 ToolNode 替代手写工具循环，除非你要完全自定义消息形状。
  4. wrap_tool_call 适合重试、观测与简单工具缓存，勿缓存有副作用的写操作。