LangGraph 1.2.7异步模式下中断消息未存入对话历史求助

问题原因 #

你当前的代码直接修改传入的state对象后触发interrupt，这种方式在异步astream模式下无法将状态更新持久化到checkpointer。LangGraph框架依赖节点返回的状态更新值来同步和保存状态，而非直接修改传入的state对象，中断会提前终止节点执行，导致手动修改的状态未被提交。

解决方案 #

让节点函数返回包含待添加消息的状态更新，同时返回中断指令，由LangGraph自动处理状态更新和中断逻辑，确保状态被正确持久化。

修改后的代码示例 #

import asyncio
from typing import Annotated, TypedDict

from langchain.messages import AnyMessage, HumanMessage
from langchain_core.messages import AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import interrupt


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


def ask_user(state: State):
    """返回状态更新和中断指令，由框架处理状态持久化"""
    message = AIMessage(content="Need more info")
    # 返回要添加的消息，以及中断指令
    return {
        "messages": [message],
        "__interrupt__": interrupt(message.content)
    }


workflow = StateGraph(State)
workflow.add_node("ask_user", ask_user)
workflow.add_edge(START, "ask_user")
workflow.add_edge("ask_user", END)
graph = workflow.compile(checkpointer=MemorySaver())


async def main():
    config = RunnableConfig(configurable={"thread_id": "1"})
    response = graph.astream(input={"messages": [HumanMessage(content="Hello")]}, config=config)

    async for event in response:
        print(event)

    hist = await graph.aget_state(config=config)
    print(hist.values)


asyncio.run(main())

执行结果 #

运行后，hist.values中的messages列表会包含两条消息：初始的HumanMessage("Hello")和中断时的AIMessage("Need more info")，符合异步模式下的需求。

补充说明 #

如果偏好更明确的命令式写法，也可以使用Command类组合状态更新和中断：

from langgraph.types import Command

def ask_user(state: State):
    message = AIMessage(content="Need more info")
    return Command(
        updates={"messages": [message]},
        interrupt=interrupt(message.content)
    )

效果与直接返回字典一致，只是写法更结构化。

内容的提问来源于stack exchange，提问作者lcs_godoy