中断

动态中断

基本介绍

LangGraph 的动态中断机制允许用户在图执行过程中设置暂停点，等待外部输入后再继续执行，提供了人机交互接口
当中断触发时，LangGraph 会通过持久化机制保存当前图状态，并无限期等待直到用户恢复执行
中断通过在任意图节点中调用 interrupt 函数实现，该函数可以接收任何可 JSON 序列化的值，并将该值暴露给调用方
中断触发后，调用方根据 interrupt 暴露的信息生成反馈，重新调用图时通过 Command 将反馈传递给计算图，恢复图的运行
恢复运行后，调用方通过 Command 传递的反馈会作为 interrupt 函数的返回值，参与后续计算
启用中断
（1）配置检查点存储器
（2）设置 thread_id，如果要确保中断后可以恢复执行，就需要保存运行图的完整状态，所以必须启用可恢复执行
（3）在需要中断的位置调用 interrupt
恢复中断
满足以下要求（1）基于相同的配置再次调用计算图（2）将输入替换为 Command 实例即可恢复运行，LangGraph 会从中断节点继续运行，该节点会被再次执行
通过 Command 实例的 resume 属性将用户反馈传递给计算图，中断节点重新运行时，传递给 resume 属性的值将会作为 interrupt 函数的返回值

HITL 模式

调用 interrupt 可以暂停状态图的执行，并等待外部输入后恢复运行，因此它是实现 HITL 的核心机制
不过，仅仅调用 interrupt 并不一定构成 HITL，只有当中断后的决策、输入或修改确实由人类完成时，才能称为 HITL
如果恢复数据完全由程序自动提供，本质上仍然是普通的中断与恢复机制

python

from typing import TypedDict

from langgraph.graph import StateGraph,START,END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver


#1. 声明状态
class OverAllState(TypedDict):
    username:str

#2. 声明节点
def node_a(state:OverAllState)->OverAllState:
    username = interrupt("请输入您的姓名")
    return {
        "username":username
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a",node_a)
builder.add_edge(START,"node_a")
builder.add_edge("node_a",END)

#4. 想要使用中断 => 必须配置检查点
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

config = {"configurable":{"thread_id":"123"}}
interrupt_res = graph.invoke({},config=config)
print(interrupt_res)

prompt = interrupt_res['__interrupt__'][0].value
print(prompt)
username = input(prompt)
# 5. 恢复图的中断执行
resume_res = graph.invoke(Command(resume=username),config=config)
print(resume_res)

多个并行中断

python

from typing import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver

class OverAllState(TypedDict):
    username: str
    age: int

def node_a(state: OverAllState) -> OverAllState:
    username = interrupt("请输入您的姓名")
    return {
        "username": username
    }

def node_b(state: OverAllState) -> OverAllState:
    age = interrupt("请输入您的年龄")
    return {
        "age": age
    }

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

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

config = {"configurable": {"thread_id": "123"}}
interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)

# 构建 中断ID -> 恢复指令的映射
resume_map = {}
for i in interrupted_res['__interrupt__']:
    user_input = input(f"{i.value}: ")
    if "年龄" in i.value:
        resume_map[i.id] = int(user_input)
    else:
        resume_map[i.id] = user_input

resumed_res = graph.invoke(Command(resume=resume_map), config=config)
print('=' * 30, '-> resumed_res <-', '=' * 30)
print(resumed_res)

审批模式

python

from time import sleep
from typing import TypedDict, Literal

from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph,START,END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek

from dotenv import load_dotenv
load_dotenv(override=True)

model =ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking":{
            "type":"disabled"
        }
    }
)

#1. 声明状态
class OverAllState(TypedDict):
    topic:str
    poem:str
    is_approved:bool

#2. 声明节点
def approve_node(state:OverAllState) -> Command[Literal["llm_node","default_node"]]:
    is_approved = interrupt("是否同意调用模型?")
    goto = "llm_node" if is_approved else "default_node"
    return Command(
        goto=goto,
        update={"is_approved":is_approved}
    )

def llm_node(state:OverAllState) -> OverAllState:
    topic = state["topic"]
    res = model.invoke([HumanMessage(content=f"帮我写一首关于{topic}主题的七言绝句,只写诗句,不需要赏析")]).content

    return {
        "poem":res
    }

def default_node(state:OverAllState) -> OverAllState:
    return {
        "poem":"请求被拒绝"
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState)

builder.add_node("approve_node",approve_node)
builder.add_node("llm_node",llm_node)
builder.add_node("default_node",default_node)
builder.add_edge(START,"approve_node")
builder.add_edge("llm_node",END)
builder.add_edge("default_node",END)

#4. 添加检查点后端
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

#5. 首次执行 -> 触发中断
config = {"configurable":{"thread_id":"234"}}
interrupt_res = graph.invoke({"topic":"菊花"},config=config)
print(interrupt_res)
#6. 人工审核
user_approved = input("是否同意调用模型?(y/n)").strip().lower() == 'y'
# 重新调用图
approved_res = graph.invoke(Command(resume=user_approved),config=config)
print(approved_res)

#7. 审核不通过
config1 = {"configurable":{"thread_id":"456"}}
interrupt_res = graph.invoke({"topic":"牡丹花"},config=config1)
print(interrupt_res)
user_approved = input("是否同意调用模型?(y/n)").strip().lower() == 'y'
# 重新调用图
approved_res = graph.invoke(Command(resume=user_approved),config=config1)
print(approved_res)

审批与编辑模式

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    reviewed_poem: str

def llm_node(state: OverAllState) -> OverAllState:
    topic = state['topic']

    res = model.invoke([HumanMessage(content=f"帮我写一首关于 {topic} 的七言绝句，只给出诗句，不要赏析")]).content
    return {
        "poem": res
    }

def review_node(state: OverAllState) -> OverAllState:
    reviewed_poem = interrupt({
        "instruction": "请审核并修改大模型生成的七言绝句",
        "poem": state['poem']
    })
    return {
        "reviewed_poem": reviewed_poem
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("review_node", review_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "review_node")
builder.add_edge("review_node", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

config = {"configurable": {"thread_id": "review_test"}}
interrupted_res = graph.invoke({"topic": "布偶猫"}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)

print("原始诗句：")
print(interrupted_res['__interrupt__'][0].value['poem'])
user_review = input("请审核并修改诗句（直接回车保留原诗）: ")
if not user_review.strip():
    user_review = interrupted_res['__interrupt__'][0].value['poem']
reviewed_res = graph.invoke(Command(resume=user_review), config=config)
print('=' * 30, '-> reviewed_res <-', '=' * 30)
print(reviewed_res)

工具执行与审批模式

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

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

    Args:
        city: 城市名称
    """
    is_approved = interrupt({
        "action": "get_weather",
        "question": "是否同意查询天气？"
    })

    logger.info("is_approved: {}", is_approved)
    if is_approved:
        return f"{city} 今天天气不错"
    else:
        return "用户拒绝查询天气"

tools_by_name = {"get_weather": get_weather}

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)
model_with_tools = model.bind_tools([get_weather])

def llm_node(state: MessagesState) -> MessagesState:
    messages = state['messages']
    response = model_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

def tool_node(state: MessagesState) -> MessagesState:
    last_msg = state['messages'][-1]

    tool_msgs = []
    for tool_call in last_msg.tool_calls:
        tool = tools_by_name[tool_call["name"]]
        logger.info("工具 {} 被调用, 对应的 tool_call: {}", tool_call["name"], tool_call)
        tool_res = tool.invoke(tool_call["args"])
        tool_msg = ToolMessage(
            name = tool_call["name"],
            content = tool_res,
            tool_call_id = tool_call["id"]
        )
        tool_msgs.append(tool_msg)

    return {
        "messages": tool_msgs
    }

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")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

config = {"configurable": {"thread_id": "tool_test"}}
interrupted_res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？")]}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
for msg in interrupted_res['messages']:
    msg.pretty_print()
print('=' * 30, '-> interrupt_info <-', '=' * 30)
print(interrupted_res['__interrupt__'])

user_approved = input("是否同意查询天气？(y/n): ").strip().lower() == 'y'
approved_res = graph.invoke(Command(resume=user_approved), config=config)
print('=' * 30, '-> approved_res <-', '=' * 30)
for msg in approved_res['messages']:
    msg.pretty_print()

单节点串行中断模式

中断恢复时，整个被中断的节点函数都会重新运行
单个节点中串行地多次调用 interrupt 函数时，检查点存储器会记录历史的 resume 信息，LangGraph 运行时会读取这些信息，并在 interrupt 函数中维护索引，按照节点内调用 interrupt 函数的顺序，逐个取出历史 resume 的值，并将它们作为 interrupt 函数的返回值
所以，已经被恢复的 interrupt 不会被重复触发，并且，由此可以推断，我们要保证历史 resume 可以被正确应用，就应保证中断恢复前后的多次 interrupt 相对顺序保持不变
当历史 resume 耗尽后，本次恢复运行时传入的 resume 会作为本次中断的返回值，然后节点函数继续运行
所有中断都触发并恢复后，计算图正常结束

python

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class OverAllState(TypedDict):
    username: str # 姓名
    age: int # 年龄
    gender: Literal["male", "female"] # 性别

def get_info_node(state: OverAllState) -> OverAllState:
    username = interrupt("请输入您的用户名：")
    age = interrupt("请输入您的年龄：")
    gender = interrupt("请输入您的性别：(male/female)")

    return {
        "username": username,
        "age": age,
        "gender": gender
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("get_info_node", get_info_node)
builder.add_edge(START, "get_info_node")
builder.add_edge("get_info_node", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "seq_interrupt_test"}}
username_interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> username_interrupted_res <-', '=' * 30)
print(username_interrupted_res)

user_name = input("请输入您的用户名：")
age_interrupted_res = graph.invoke(Command(resume=user_name), config=config)
print('=' * 30, '-> age_interrupted_res <-', '=' * 30)
print(age_interrupted_res)

user_age = input("请输入您的年龄：")
gender_interrupted_res = graph.invoke(Command(resume=int(user_age)), config=config)
print('=' * 30, '-> gender_interrupted_res <-', '=' * 30)
print(gender_interrupted_res)

user_gender = input("请输入您的性别：(male/female): ")
resumed_res = graph.invoke(Command(resume=user_gender), config=config)
print('=' * 30, '-> resumed_res <-', '=' * 30)
print(resumed_res)

使用规范

不要用 try/catch 包裹 interrupt 调用：中断的触发是通过抛出 GraphInterrupt 异常实现的，如果用 try/catch 包裹，则底层运行时无法感知断点，不会中断计算图
不要更改单个节点内 interrupt 的调用顺序：（1）恢复运行时整个节点函数都会重新运行，而非精确地从断点继续（2）同一节点中存在多个断点时：历史恢复记录被记录在检查点中，并在恢复运行时按顺序加载；一旦中断恢复前后的断点顺序、数量不完全一致，将会导致语义混乱，程序执行结果无法满足预期，这条规范就是为了避免这样的情况
不要在 interrupt 中传递复杂类型：LangGraph 运行时会将 interrupt 函数接收到的参数经过 JSON 序列化之后传递给调用者，如果传递不支持 JSON 序列化的复杂类型，如函数，将会抛出异常
断点之前的副作用操作必须是幂等的：中断恢复时，断点所在的函数会被重复执行，所以，如果断点之前存在不满足幂等性的副作用操作，将会导致多次调用结果不一致

静态断点

基本介绍

LangGraph 提供了用于调试的静态断点，它是在状态图编译或调用时设置的，不会在运行时动态触发，不属于业务逻辑，所以称为静态断点
动态断点提供了接收人类反馈的接口，而静态断点只是暂停计算图的运行，不能向调用者传递信息，也不能接收调用者的反馈

用法说明

（1）静态断点也需要基于检查点恢复，所以必须配置检查点，启用可恢复运行机制
（2）计算图会在 interrupt_before 指定的节点执行之前产生中断，暂停计算
（3）计算图会在 interrupt_after 指定的节点执行之后产生中断，暂停计算
（4）计算图运行到断点位置会中断，和动态断点不同的是，它会在超步边界而非内部中断，断点前的超步：三个阶段全部完成；断点后的超步：三个阶段都没有开始
（5）静态断点不会返回任何中断信息，只会将当前最新的状态返回，因此，我们可以用静态断点查看每个超步边界的中间状态
（6）传入相同的配置并将 None 作为输入，再次调用计算图，会从断点位置继续运行
（7）支持在两个阶段设置静态断点，断点都是在运行时生效，计算图调用时设置的断点优先级更高，如果调用时传入的断点列表不为空，则会覆盖编译时配置

状态图编译时

python

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)

计算图调用时

python

first_res = graph.invoke(
    {},
    config=config,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)

编译时设置断点

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

from loguru import logger

class OverAllState(TypedDict):
    final_res: str

def node_a(state: OverAllState) -> OverAllState:
    logger.info("node_a 被执行")
    return {
        "final_res": "node_a 运行的中间结果"
    }

def node_b(state: OverAllState) -> OverAllState:
    logger.info("node_b 被执行")
    return {
        "final_res": "node_b 运行的中间结果"
    }

def node_c(state: OverAllState) -> OverAllState:
    logger.info("node_c 被执行")
    return {
        "final_res": "node_c 运行的中间结果"
    }

def node_d(state: OverAllState) -> OverAllState:
    logger.info("node_d 被执行")
    return {}

def node_e(state: OverAllState) -> OverAllState:
    logger.info("node_e 被执行")
    return {}

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_node("node_d", node_d)
builder.add_node("node_e", node_e)

builder.add_edge(START, "node_a")
builder.add_edge(START, "node_d")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_d", "node_e")
builder.add_edge(["node_b", "node_e"], "node_c")
builder.add_edge("node_c", END)

checkpointer = InMemorySaver()
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)

from IPython.display import display
display(graph)

config = {"configurable": {"thread_id": "123"}}

logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke({}, config=config)
logger.info("第一次执行结果：{}", first_res)

logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(None, config=config)
logger.info("第二次执行结果：{}", second_res)

logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(None, config=config)
logger.info("第三次执行结果：{}", third_res)

logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(None, config=config)
logger.info("第四次执行结果：{}", final_res)

logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)

调式时设置断点

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

from loguru import logger

class OverAllState(TypedDict):
    final_res: str

def node_a(state: OverAllState) -> OverAllState:
    logger.info("node_a 被执行")
    return {
        "final_res": "node_a 运行的中间结果"
    }

def node_b(state: OverAllState) -> OverAllState:
    logger.info("node_b 被执行")
    return {
        "final_res": "node_b 运行的中间结果"
    }

def node_c(state: OverAllState) -> OverAllState:
    logger.info("node_c 被执行")
    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)

checkpointer = InMemorySaver()
graph = builder.compile(
    checkpointer=checkpointer
)

from IPython.display import display
display(graph)

config = {"configurable": {"thread_id": "123"}}

logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke(
    {},
    config=config,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)
logger.info("第一次执行结果：{}", first_res)

logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(
    None,
    config=config,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)
logger.info("第二次执行结果：{}", second_res)

logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(
    None,
    config=config,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)
logger.info("第三次执行结果：{}", third_res)

logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(
    None,
    config=config,
    interrupt_before=["node_a", "node_b"],
    interrupt_after=["node_a", "node_b"]
)
logger.info("第四次执行结果：{}", final_res)

logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)

工具调用节点

ToolNode

手动实现有助于理解工具调用的基本流程：模型生成工具调用请求，工具节点根据名称查找并执行工具，再将执行结果封装成与原工具调用对应的 ToolMessage
不过，实际项目还需要处理参数校验、异常处理、运行时注入、并行执行以及 Command 传播等问题
ToolNode 对这些通用逻辑进行了封装，适合在需要自定义图结构和工具调用流程的场景中使用

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.graph.message import MessagesState
from langchain.messages import HumanMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek

from dotenv import load_dotenv
load_dotenv(override=True)

@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：国外新闻
    """
    if home_or_abroad:
        return "Kimi 新模型发布"
    return "Anthropic 暂停新模型访问"

tools = [get_weather, get_news]

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)
model_with_tools = model.bind_tools(tools=tools)

def llm_node(state: MessagesState) -> MessagesState:
    messages = state['messages']
    response = model_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

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", 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()

from IPython.display import display
display(graph)

res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？国内有哪些新闻？")]})
for msg in res['messages']:
    msg.pretty_print()

ToolRuntime

ToolRuntime 是专门面向工具调用的运行时对象
按照官方规范，工具函数中存在名为 runtime、并且类型标注为 ToolRuntime 的参数时，底层运行时会在调用工具前自动注入 ToolRuntime 实例
根据源码实现，某些不规范的写法也能注入运行时实例，但并不规范，不推荐
需要注意，ToolRuntime 与图节点中使用的 langgraph.runtime.Runtime 并不是同一个类型，前者额外提供了当前图状态、运行配置和工具调用 ID 等工具调用专属信息
当前版本，ToolRuntime 的核心定义如下

python

@dataclass
class ToolRuntime(_DirectlyInjectedToolArg, Generic[ContextT, StateT]):
    state: StateT
    context: ContextT
    config: RunnableConfig
    stream_writer: StreamWriter
    tool_call_id: str | None
    store: BaseStore | None

（1）state：图状态，短期记忆
（2）context：运行时上下文
（3）config：运行时配置，包含元数据
（4）stream_writer：自定义流式输出写入器
（5）tool_call_id：工具调用 ID
（6）store：长期记忆
借助 runtime，工具可以访问上述所有资源

工具中更新状态

工具不仅可以返回普通结果，还可以返回 Command(update=...)，将业务结果写入图状态
当工具由模型发起调用时，消息历史中的 AIMessage.tool_calls 必须紧跟与之匹配的 ToolMessage
当工具返回普通结果时，ToolNode 会帮我们完成运行结果到 ToolMessage 的转换，并追加到消息列表
但工具返回 Command 时，上述操作需要开发者完成，此时还应在 Command.update 的消息字段中写入对应的 ToolMessage
如果同一轮并行执行的多个工具可能更新同一个普通状态字段，需要为该字段配置合适的归并函数；否则可能出现并发更新冲突，本例中的两个工具分别更新 weather_res 和 news_res，因此不存在该问题

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt.tool_node import ToolNode, ToolRuntime
from langgraph.types import Command
from langgraph.graph.message import MessagesState

from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek

from dotenv import load_dotenv
load_dotenv(override=True)

class OverAllState(MessagesState):
    weather_res: str
    news_res: str

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

    Args:
        city: 城市名称
    """
    res = f"{city} 今天天气不错"
    tool_call_id = runtime.tool_call_id
    tool_msg = ToolMessage(tool_call_id=tool_call_id, content=res)
    return Command(
        update = {
            "weather_res": res,
            "messages": [tool_msg]
        }
    )

@tool(parse_docstring=True)
def get_news(home_or_abroad: bool, runtime: ToolRuntime) -> Command:
    """
    查询国内外新闻

    Args:
        home_or_abroad: 查询国内还是国外新闻，True: 国内新闻，False：国外新闻
    """
    if home_or_abroad:
        res = "Kimi 新模型发布"
    else:
        res = "Anthropic 暂停新模型访问"
    tool_call_id = runtime.tool_call_id
    tool_msg = ToolMessage(tool_call_id=tool_call_id, content=res)
    return Command(
        update = {
            "news_res": res,
            "messages": [tool_msg]
        }
    )

tools = [get_weather, get_news]

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)
model_with_tools = model.bind_tools(tools=tools)

def llm_node(state: OverAllState) -> OverAllState:
    messages = state['messages']
    response = model_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

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

builder = StateGraph(state_schema=OverAllState)
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()

from IPython.display import display
display(graph)

res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？国内有哪些新闻？")]})
print('=' * 30, '-> messages <-', '=' * 30)
for msg in res.pop('messages'):
    msg.pretty_print()

print('=' * 30, '-> res without messages <-', '=' * 30)
print(res)

实现重试机制

ToolNode 提供同步工具调用包装器 wrap_tool_call，用于在工具执行前后插入自定义逻辑
它与 LangChain Agent 中间件 wrap_tool_call 的核心机制相同：接收当前工具调用请求 request 和真正执行工具的回调 execute，可以选择不调用、调用一次或多次 execute（request），并最终返回 ToolMessage 或 Command
因此，wrap_tool_call 可以用于重试、缓存、请求改写、短路返回和自定义控制流等场景
异步工具链路还可以使用对应的 wrap_tool_call，此处通过重试和缓存机制的实现

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.graph.message import MessagesState

from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek

from dataclasses import dataclass
from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

import random

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

    Args:
        city: 城市名称
    """
    # 70% 概率因网络波动而调用失败
    rand_int = random.randint(1,10)
    if rand_int < 8:
        raise ConnectionError("网络波动失败")
    return f"{city} 今天天气不错"

tools = [get_weather]

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)
model_with_tools = model.bind_tools(tools=tools)

@dataclass
class UserContext:
    max_attempts: int

def llm_node(state: MessagesState) -> MessagesState:
    messages = state['messages']
    response = model_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

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

def wrap_tool_call(request, execute):
    max_attempts = request.runtime.context.max_attempts
    tool_call_id = request.runtime.tool_call_id

    tool_msg = ""
    for i in range(max_attempts):
        try:
            tool_msg = execute(request)
            break
        except ConnectionError as e:
            logger.info("工具调用失败，当前调用次数: {}, 调用次数上限: {}, 异常信息: {}", i + 1, max_attempts, e)

    if not tool_msg:
        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()

from IPython.display import display
display(graph)

res = graph.invoke(
    {"messages": [HumanMessage("今天北京天气如何？")]},
    context = UserContext(max_attempts=3))
for msg in res['messages']:
    msg.pretty_print()

实现缓存机制

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.graph.message import MessagesState

from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek

from loguru import logger
import json
from dotenv import load_dotenv
load_dotenv(override=True)

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

    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"

tools = [get_weather]

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)
model_with_tools = model.bind_tools(tools=tools)

def llm_node(state: MessagesState) -> MessagesState:
    messages = state['messages']
    response = model_with_tools.invoke(messages)

    return {
        "messages": [response]
    }

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

global_cache = dict()
def wrap_tool_call(request, execute):
    tool_name = request.tool_call["name"]
    tool_args = json.dumps(request.tool_call["args"])
    tool_call_id = request.runtime.tool_call_id

    cache_key = (tool_name, tool_args)
    cache = global_cache.get(cache_key)

    if cache:
        logger.info("{} 调用命中缓存", tool_name)
        tool_msg = ToolMessage(
            tool_call_id = tool_call_id,
            content = cache
        )
    else:
        tool_msg = execute(request)
        logger.info("将 {} 的调用结果写入缓存", tool_name)
        global_cache[cache_key] = tool_msg.content

    return tool_msg

builder = StateGraph(state_schema=MessagesState)
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()

from IPython.display import display
display(graph)

print('=' * 30, '-> 第一次调用 - 北京 <-', '=' * 30)
res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？")]})
for msg in res['messages']:
    msg.pretty_print()

print('=' * 30, '-> 第二次调用 - 北京 <-', '=' * 30)
res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？")]})
for msg in res['messages']:
    msg.pretty_print()

print('=' * 30, '-> 第三次调用 - 杭州 <-', '=' * 30)
res = graph.invoke({"messages": [HumanMessage("今天杭州天气如何？")]})
for msg in res['messages']:
    msg.pretty_print()

中断

动态中断 ​

基本介绍 ​

LangGraph 的动态中断机制允许用户在图执行过程中设置暂停点，等待外部输入后再继续执行，提供了人机交互接口 ​

当中断触发时，LangGraph 会通过持久化机制保存当前图状态，并无限期等待直到用户恢复执行 ​

中断通过在任意图节点中调用 interrupt 函数实现，该函数可以接收任何可 JSON 序列化的值，并将该值暴露给调用方 ​

中断触发后，调用方根据 interrupt 暴露的信息生成反馈，重新调用图时通过 Command 将反馈传递给计算图，恢复图的运行 ​

恢复运行后，调用方通过 Command 传递的反馈会作为 interrupt 函数的返回值，参与后续计算 ​

启用中断 ​

（1）配置检查点存储器 ​

（2）设置 thread_id，如果要确保中断后可以恢复执行，就需要保存运行图的完整状态，所以必须启用可恢复执行 ​

（3）在需要中断的位置调用 interrupt ​

恢复中断 ​

满足以下要求（1）基于相同的配置再次调用计算图（2）将输入替换为 Command 实例即可恢复运行，LangGraph 会从中断节点继续运行，该节点会被再次执行 ​

通过 Command 实例的 resume 属性将用户反馈传递给计算图，中断节点重新运行时，传递给 resume 属性的值将会作为 interrupt 函数的返回值 ​

HITL 模式 ​

调用 interrupt 可以暂停状态图的执行，并等待外部输入后恢复运行，因此它是实现 HITL 的核心机制 ​

不过，仅仅调用 interrupt 并不一定构成 HITL，只有当中断后的决策、输入或修改确实由人类完成时，才能称为 HITL ​

如果恢复数据完全由程序自动提供，本质上仍然是普通的中断与恢复机制 ​

多个并行中断 ​

审批模式 ​

审批与编辑模式 ​

工具执行与审批模式 ​

单节点串行中断模式 ​

中断恢复时，整个被中断的节点函数都会重新运行 ​

所以，已经被恢复的 interrupt 不会被重复触发，并且，由此可以推断，我们要保证历史 resume 可以被正确应用，就应保证中断恢复前后的多次 interrupt 相对顺序保持不变 ​

当历史 resume 耗尽后，本次恢复运行时传入的 resume 会作为本次中断的返回值，然后节点函数继续运行 ​

所有中断都触发并恢复后，计算图正常结束 ​

使用规范 ​

不要用 try/catch 包裹 interrupt 调用：中断的触发是通过抛出 GraphInterrupt 异常实现的，如果用 try/catch 包裹，则底层运行时无法感知断点，不会中断计算图 ​

不要在 interrupt 中传递复杂类型：LangGraph 运行时会将 interrupt 函数接收到的参数经过 JSON 序列化之后传递给调用者，如果传递不支持 JSON 序列化的复杂类型，如函数，将会抛出异常 ​

断点之前的副作用操作必须是幂等的：中断恢复时，断点所在的函数会被重复执行，所以，如果断点之前存在不满足幂等性的副作用操作，将会导致多次调用结果不一致 ​

静态断点 ​

基本介绍 ​

LangGraph 提供了用于调试的静态断点，它是在状态图编译或调用时设置的，不会在运行时动态触发，不属于业务逻辑，所以称为静态断点 ​

动态断点提供了接收人类反馈的接口，而静态断点只是暂停计算图的运行，不能向调用者传递信息，也不能接收调用者的反馈 ​

用法说明 ​

（1）静态断点也需要基于检查点恢复，所以必须配置检查点，启用可恢复运行机制 ​

（2）计算图会在 interrupt_before 指定的节点执行之前产生中断，暂停计算 ​

（3）计算图会在 interrupt_after 指定的节点执行之后产生中断，暂停计算 ​

（4）计算图运行到断点位置会中断，和动态断点不同的是，它会在超步边界而非内部中断，断点前的超步：三个阶段全部完成；断点后的超步：三个阶段都没有开始 ​

（5）静态断点不会返回任何中断信息，只会将当前最新的状态返回，因此，我们可以用静态断点查看每个超步边界的中间状态 ​

（6）传入相同的配置并将 None 作为输入，再次调用计算图，会从断点位置继续运行 ​

（7）支持在两个阶段设置静态断点，断点都是在运行时生效，计算图调用时设置的断点优先级更高，如果调用时传入的断点列表不为空，则会覆盖编译时配置 ​

状态图编译时 ​

计算图调用时 ​

编译时设置断点 ​

调式时设置断点 ​

工具调用节点 ​

ToolNode ​

手动实现有助于理解工具调用的基本流程：模型生成工具调用请求，工具节点根据名称查找并执行工具，再将执行结果封装成与原工具调用对应的 ToolMessage ​

不过，实际项目还需要处理参数校验、异常处理、运行时注入、并行执行以及 Command 传播等问题 ​

ToolNode 对这些通用逻辑进行了封装，适合在需要自定义图结构和工具调用流程的场景中使用 ​

ToolRuntime ​

ToolRuntime 是专门面向工具调用的运行时对象 ​

按照官方规范，工具函数中存在名为 runtime、并且类型标注为 ToolRuntime 的参数时，底层运行时会在调用工具前自动注入 ToolRuntime 实例 ​

根据源码实现，某些不规范的写法也能注入运行时实例，但并不规范，不推荐 ​

需要注意，ToolRuntime 与图节点中使用的 langgraph.runtime.Runtime 并不是同一个类型，前者额外提供了当前图状态、运行配置和工具调用 ID 等工具调用专属信息 ​

当前版本，ToolRuntime 的核心定义如下 ​

（1）state：图状态，短期记忆 ​

（2）context：运行时上下文 ​

（3）config：运行时配置，包含元数据 ​

（4）stream_writer：自定义流式输出写入器 ​

（5）tool_call_id：工具调用 ID ​

（6）store：长期记忆 ​

借助 runtime，工具可以访问上述所有资源 ​

工具中更新状态 ​

工具不仅可以返回普通结果，还可以返回 Command(update=...)，将业务结果写入图状态 ​

当工具由模型发起调用时，消息历史中的 AIMessage.tool_calls 必须紧跟与之匹配的 ToolMessage ​

当工具返回普通结果时，ToolNode 会帮我们完成运行结果到 ToolMessage 的转换，并追加到消息列表 ​

但工具返回 Command 时，上述操作需要开发者完成，此时还应在 Command.update 的消息字段中写入对应的 ToolMessage ​

如果同一轮并行执行的多个工具可能更新同一个普通状态字段，需要为该字段配置合适的归并函数；否则可能出现并发更新冲突，本例中的两个工具分别更新 weather_res 和 news_res，因此不存在该问题 ​

实现重试机制 ​

ToolNode 提供同步工具调用包装器 wrap_tool_call，用于在工具执行前后插入自定义逻辑 ​

它与 LangChain Agent 中间件 wrap_tool_call 的核心机制相同：接收当前工具调用请求 request 和真正执行工具的回调 execute，可以选择不调用、调用一次或多次 execute（request），并最终返回 ToolMessage 或 Command ​

因此，wrap_tool_call 可以用于重试、缓存、请求改写、短路返回和自定义控制流等场景 ​

异步工具链路还可以使用对应的 wrap_tool_call，此处通过重试和缓存机制的实现 ​

实现缓存机制 ​

动态中断

基本介绍

LangGraph 的动态中断机制允许用户在图执行过程中设置暂停点，等待外部输入后再继续执行，提供了人机交互接口

当中断触发时，LangGraph 会通过持久化机制保存当前图状态，并无限期等待直到用户恢复执行

中断通过在任意图节点中调用 interrupt 函数实现，该函数可以接收任何可 JSON 序列化的值，并将该值暴露给调用方

中断触发后，调用方根据 interrupt 暴露的信息生成反馈，重新调用图时通过 Command 将反馈传递给计算图，恢复图的运行

恢复运行后，调用方通过 Command 传递的反馈会作为 interrupt 函数的返回值，参与后续计算

启用中断

（1）配置检查点存储器

（2）设置 thread_id，如果要确保中断后可以恢复执行，就需要保存运行图的完整状态，所以必须启用可恢复执行

（3）在需要中断的位置调用 interrupt

恢复中断

满足以下要求（1）基于相同的配置再次调用计算图（2）将输入替换为 Command 实例即可恢复运行，LangGraph 会从中断节点继续运行，该节点会被再次执行

通过 Command 实例的 resume 属性将用户反馈传递给计算图，中断节点重新运行时，传递给 resume 属性的值将会作为 interrupt 函数的返回值

HITL 模式

调用 interrupt 可以暂停状态图的执行，并等待外部输入后恢复运行，因此它是实现 HITL 的核心机制

不过，仅仅调用 interrupt 并不一定构成 HITL，只有当中断后的决策、输入或修改确实由人类完成时，才能称为 HITL

如果恢复数据完全由程序自动提供，本质上仍然是普通的中断与恢复机制

多个并行中断

审批模式

审批与编辑模式

工具执行与审批模式

单节点串行中断模式

中断恢复时，整个被中断的节点函数都会重新运行

所以，已经被恢复的 interrupt 不会被重复触发，并且，由此可以推断，我们要保证历史 resume 可以被正确应用，就应保证中断恢复前后的多次 interrupt 相对顺序保持不变

当历史 resume 耗尽后，本次恢复运行时传入的 resume 会作为本次中断的返回值，然后节点函数继续运行

所有中断都触发并恢复后，计算图正常结束

使用规范

不要用 try/catch 包裹 interrupt 调用：中断的触发是通过抛出 GraphInterrupt 异常实现的，如果用 try/catch 包裹，则底层运行时无法感知断点，不会中断计算图

不要在 interrupt 中传递复杂类型：LangGraph 运行时会将 interrupt 函数接收到的参数经过 JSON 序列化之后传递给调用者，如果传递不支持 JSON 序列化的复杂类型，如函数，将会抛出异常

断点之前的副作用操作必须是幂等的：中断恢复时，断点所在的函数会被重复执行，所以，如果断点之前存在不满足幂等性的副作用操作，将会导致多次调用结果不一致

静态断点

基本介绍

LangGraph 提供了用于调试的静态断点，它是在状态图编译或调用时设置的，不会在运行时动态触发，不属于业务逻辑，所以称为静态断点

动态断点提供了接收人类反馈的接口，而静态断点只是暂停计算图的运行，不能向调用者传递信息，也不能接收调用者的反馈

用法说明

（1）静态断点也需要基于检查点恢复，所以必须配置检查点，启用可恢复运行机制

（2）计算图会在 interrupt_before 指定的节点执行之前产生中断，暂停计算

（3）计算图会在 interrupt_after 指定的节点执行之后产生中断，暂停计算

（4）计算图运行到断点位置会中断，和动态断点不同的是，它会在超步边界而非内部中断，断点前的超步：三个阶段全部完成；断点后的超步：三个阶段都没有开始

（5）静态断点不会返回任何中断信息，只会将当前最新的状态返回，因此，我们可以用静态断点查看每个超步边界的中间状态

（6）传入相同的配置并将 None 作为输入，再次调用计算图，会从断点位置继续运行

（7）支持在两个阶段设置静态断点，断点都是在运行时生效，计算图调用时设置的断点优先级更高，如果调用时传入的断点列表不为空，则会覆盖编译时配置

状态图编译时

计算图调用时

编译时设置断点

调式时设置断点

工具调用节点

ToolNode

手动实现有助于理解工具调用的基本流程：模型生成工具调用请求，工具节点根据名称查找并执行工具，再将执行结果封装成与原工具调用对应的 ToolMessage

不过，实际项目还需要处理参数校验、异常处理、运行时注入、并行执行以及 Command 传播等问题

ToolNode 对这些通用逻辑进行了封装，适合在需要自定义图结构和工具调用流程的场景中使用

ToolRuntime

ToolRuntime 是专门面向工具调用的运行时对象

按照官方规范，工具函数中存在名为 runtime、并且类型标注为 ToolRuntime 的参数时，底层运行时会在调用工具前自动注入 ToolRuntime 实例

根据源码实现，某些不规范的写法也能注入运行时实例，但并不规范，不推荐

需要注意，ToolRuntime 与图节点中使用的 langgraph.runtime.Runtime 并不是同一个类型，前者额外提供了当前图状态、运行配置和工具调用 ID 等工具调用专属信息

当前版本，ToolRuntime 的核心定义如下

（1）state：图状态，短期记忆

（2）context：运行时上下文

（3）config：运行时配置，包含元数据

（4）stream_writer：自定义流式输出写入器

（5）tool_call_id：工具调用 ID

（6）store：长期记忆

借助 runtime，工具可以访问上述所有资源

工具中更新状态

工具不仅可以返回普通结果，还可以返回 Command(update=...)，将业务结果写入图状态

当工具由模型发起调用时，消息历史中的 AIMessage.tool_calls 必须紧跟与之匹配的 ToolMessage

当工具返回普通结果时，ToolNode 会帮我们完成运行结果到 ToolMessage 的转换，并追加到消息列表

但工具返回 Command 时，上述操作需要开发者完成，此时还应在 Command.update 的消息字段中写入对应的 ToolMessage

如果同一轮并行执行的多个工具可能更新同一个普通状态字段，需要为该字段配置合适的归并函数；否则可能出现并发更新冲突，本例中的两个工具分别更新 weather_res 和 news_res，因此不存在该问题

实现重试机制

ToolNode 提供同步工具调用包装器 wrap_tool_call，用于在工具执行前后插入自定义逻辑

它与 LangChain Agent 中间件 wrap_tool_call 的核心机制相同：接收当前工具调用请求 request 和真正执行工具的回调 execute，可以选择不调用、调用一次或多次 execute（request），并最终返回 ToolMessage 或 Command

因此，wrap_tool_call 可以用于重试、缓存、请求改写、短路返回和自定义控制流等场景

异步工具链路还可以使用对应的 wrap_tool_call，此处通过重试和缓存机制的实现

实现缓存机制