LangGraph入门到精通第二期：基础API（一）

前言

前面我们陆续聊了 LangGraph 的几个基础能力：Reducers、Send、Interrupt、Command。今天接着挖一挖剩下的那些实用功能。说实话，LangGraph 的这套设计理念挺有意思——把状态管理、记忆、子图这些概念揉进图结构里，既保留了流程图的直观，又解决了传统 DAG 框架不擅长的对话记忆问题。

checkpoint

Checkpoints 是 LangGraph 在执行过程中——确切地说是在每个超级步骤（Super-step）结束时——自动拍下的状态快照。每个 Checkpoint 其实就是一个 StateSnapshot 对象，里面装着当前所有状态通道的数据，比如图中的 foo、bar 字段值。同时它还记录了下一步该执行哪个节点（next 字段），如果为空说明流程已经跑完了。配置信息（config）里藏着唯一标识符 thread_id 和 checkpoint_id，元数据里则是执行来源、修改记录、步骤编号这些。另外还有创建时间和父检查点配置，用来构建完整的执行链路。

那谁来负责保存和管理这些快照呢？Checkpointer。LangGraph 提供了几种现成的实现：

• InMemorySa ver：存在内存里，适合开发和测试，机器一重启就没了
• SqliteSa ver：存在 SQLite 文件里，小型生产环境或本地开发够用了
• PostgresSa ver：存在 PostgreSQL 里，生产环境标配，支持并发

常用的方法就两个：get_state 拿当前快照，get_state_history 拉取所有历史快照——包括每一步的执行情况，这对于调试和回放非常有用。

记忆绑定

想让 graph 带上快照功能，编译的时候必须传入 checkpointer 参数，就像这样：

# 内存记忆器：保存每一步执行状态，支持回溯checkpointer = InMemorySa ver()
# 编译并绑定记忆graph = workflow.compile(checkpointer=checkpointer)

会话标识

不同对话或执行上下文怎么区分？靠 thread_id。配置一个独一无二的 ID 即可：

config = {"configurable": {"thread_id": "thread-1"}}

快照获取

# 获取快照snapshot = graph.get_state(config)
# 获取全部历史快照history = list(graph.get_state_history(config))

图调用

graph 本身不存快照，只负责跑流程。快存在 CheckPointer 里。所以在调用 graph 时得带上 config。背后的逻辑是：graph 根据 config 中的 thread_id 和 checkpoint 绑定，取快照时再拿着 thread_id 去 checkpoint 里拉数据。调用方式：

# 执行图result = graph.invoke(initial_state, config)

更多细节可以直接看官方文档，这里不赘述。

Context Memory

之前用 LangChain 的 ChatModel 时，输入通常是一个 Message 对象列表，HumanMessage 或 AIMessage。在 LangGraph 里，State 也可以实现类似 BaseMessage 的自动合并功能，让对话历史自然累积。

自动合并消息的 State

class ChatState(TypedDict):
    # messages 自动合并消息的历史记录
    # Annotated：附加信息
    # Sequence：有序列表
    messages: Annotated[Sequence[BaseMessage], operator.add]

关键就在于 operator.add 这个 reducer，它会把节点返回的消息追加到已有列表里，而不是覆盖。

用户输入

def handle_user_input(state: ChatState):
    """处理用户输入节点（增强健壮性）"""
    try:
        user_input = input("n用户输入（输入'Q'结束）: ").strip()
        if user_input.lower() == "Q":
            logger.info(f"正在结束对话...")
            return END
        # 保留历史记录并追加新消息
        return {"messages": [HumanMessage(content=user_input)]}
    except Exception :
        return END

这里有个小细节：节点返回的是 {"messages": [HumanMessage(...)]}，由于 reducer 是加法，新消息会追加到历史记录后面，而不是替换整条列表。

AI 回复

def generate_ai_response(state: ChatState):
    """生成AI响应节点（增加错误处理）"""
    try:
        # 使用最近6条消息保持上下文连贯性
        recent_history = state["messages"][-6:]
        """做个摘要"""
        response = model.invoke(recent_history)
        return {"messages": [response]}
    except Exception as e:
        error_msg = f"系统暂时无法响应，请稍后再试（错误代码：{str(e)[:30]})"
        return {"messages": [AIMessage(content=error_msg)]}

注意这里只取了最近 6 条消息，避免上下文太长导致 token 浪费。错误处理也做了兜底，不会让整个图崩溃。

节点注册

# 节点注册
# RunnableLambda 可不写，add_node 内部会自动包装成 RunnableLambda
builder.add_node("user_input", handle_user_input)
builder.add_node("ai_response", RunnableLambda(generate_ai_response))

实际上 RunnableLambda 不是必须的，add_node 内部会自动把普通函数包装成 RunnableLambda，所以直接传函数名就行。

对话循环

while True:
    try:
        # 执行对话流程
        result = conversation.invoke(state)
        if result is None or "messages" not in result:
            break
        # 提取最新交互记录
        new_messages = result["messages"][len(state["messages"]):]
        # 打印AI响应
        for msg in new_messages:
            if isinstance(msg, AIMessage):
                logger.info(f"n【AI响应】n{msg.content}n")
        # 更新对话状态
        state = result
        # 检查退出条件
        if any(isinstance(m, HumanMessage) and m.content.lower() == "退出" for m in state["messages"]):
            break
    except Exception as e:
        logger.info(f"n系统异常：{str(e)}")
        break

这个循环的作用就是持续对话，每次只取出新生成的消息（通过 len(state["messages"]) 截取），然后打印 AI 回复。退出条件有两个：用户输入了 "退出" 或者发生了异常。整体设计相当健壮。

Subgraphs

子图的概念很好理解——就像面向对象里的父类/子类关系。LangGraph 允许你把复杂的工作流拆成更小、更易管理的子图，然后把这些子图作为 Node 挂到主图上。它提供了两种模式：

• 共享模式：父图和子图拥有共享的状态键。共享的键（比如 foo）会在父图和子图之间自动同步，子图可以直接读写。这就像父类里的 public 变量。
• 不同状态模式：父图和子图的状态结构完全不同。这时候就得在父图节点里手动做“输入映射 → 子图调用 → 输出映射”三步转换。

先看共享状态模式的代码：

class SharedParentState(TypedDict):
    text: str
    # 父图与子图共享的键
    result: str
    # 父图私有键（可选）

class SharedSubgraphState(TypedDict):
    text: str
    # 与父图共享的键（必须同名）
    processed: str
    # 子图私有键（仅子图可见）

def subgraph_node(state: SharedSubgraphState) -> SharedSubgraphState:
    """子图节点：将共享键"text"转大写，存到私有键"processed"""
    return {
        "text": state["text"].upper(),
        # 修改共享键（自动同步到父图）
        "processed": f"Processed: {state['text'].upper()}"
        # 子图私有数据
    }

def parent_node(state: SharedParentState) -> SharedParentState:
    """父图节点：直接调用子图（共享键自动传递）"""
    return state
    # 子图会修改"text"键，父图后续可直接用

注意这里子图修改了共享的 text 键（变成大写），这个变更会自动同步回父图，父图后续的节点拿到的 text 就是大写后的值。

不同状态模式则麻烦一些，需要在父图包装节点里做手动映射：

def parent_wrapper_node(state: DifferentParentState) -> DifferentParentState:
    """父图包装节点：【核心：手动状态映射】
    因为父子图状态完全独立，必须手动：
    1. 父图状态 → 子图输入
    2. 调用子图
    3. 子图输出 → 父图状态
    """
    # ---------------------- 输入映射 ----------------------
    # 父图键：user_query → 子图键：task
    subgraph_input = {"task": state["user_query"]}
    # ---------------------- 调用独立子图 ----------------------
    # 子图与父图状态隔离，不会互相污染
    subgraph_output = compiled_different_subgraph.invoke(subgraph_input)
    # ---------------------- 输出映射 ----------------------
    # 子图键：result → 父图键：answer
    return {
        "answer": subgraph_output["result"],
        "user_query": state["user_query"]  # 保留父图原始输入
    }

这里 subgraph_input = {"task": state["user_query"]} 就是输入映射，把父图的 user_query 映射成子图的 task。子图内部只认 task 字段：

def different_subgraph_node(state: DifferentSubgraphState) -> DifferentSubgraphState:
    # 处理逻辑：给 task 增加子图前缀
    return {"result": f"[Subgraph] Processed: {state['task']}"}

输出映射把子图的 result 写回父图的 answer。这样两个状态结构完全不同的图也能协同工作了。

更多细节可以翻官方文档。

visualization

LangGraph 还自带可视化能力，编译后的 graph 可以直接画成流程图。一行代码就能搞定：

graph.get_graph().draw_mermaid_png(output_file_path=sa ve_file_path)

这玩意儿特别方便，调试时一眼就能看清节点之间的依赖关系和执行顺序。具体支持哪些格式、怎么定制，官方文档里都有。