十分钟上手 LangGraph：构建一个带记忆的聊天机器人

本篇手把手完成第一个 LangGraph 应用：一个能连续对话、流式输出的命令行聊天机器人。重点搞懂两件事——状态怎么累加（add_messages）和图怎么搭（StateGraph）。全文代码加起来不到 40 行。

直接调大模型，差在哪？

很多人写的第一个"聊天机器人"是这样的：

while True:
    user_input = input("User: ")
    reply = llm.invoke(user_input)
    print("Assistant:", reply)

跑起来没问题，但有两个致命缺陷：

1. 没有记忆：每轮对话都只把当前这句话发给模型，上一轮说过"我叫小林"，这一轮再问"我叫什么"它照样答不上来；
2. 没有流程：所有逻辑挤在循环里。往后要加工具调用、加意图识别、加人工审核，代码会缠成一团。

LangGraph 的解法：把对话建模成一张图，把对话历史建模成图的状态。

一、先看全局：一次请求的完整旅程

图中只有两个角色：状态负责记住一切，节点负责干活。下面逐块实现。

二、第一步：定义状态——add_messages 的累加魔法

状态定义了整张图"共享内存"的形状。聊天机器人只需要一个字段：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

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

这短短三行是 LangGraph 状态设计的核心语法，拆开看：

• TypedDict：用类型注解约束状态的结构。messages 是必填字段、类型是 list，写错键名（比如 message）IDE 会直接提示——比裸 dict 安全得多；
• Annotated[list, add_messages]：给 messages 字段挂上一条合并规则。规则的内容是：节点返回的新消息追加到已有历史后面，而不是覆盖。

add_messages 的合并逻辑可以简化为下面这段伪代码：

def add_messages(existing, updates):
    current = existing.copy() if existing is not None else []
    if updates is None:
        return current
    if isinstance(updates, list):
        return current + updates
    return current + [updates]

关键差异在一个对比表里：

写法	节点返回新消息后	最终状态里的 messages
messages: list（普通字段）	直接覆盖	只剩新消息，历史丢失
messages: Annotated[list, add_messages]	触发追加规则	旧历史 + 新消息，上下文完整

这个机制是"有记忆"的地基：对话历史存在 messages 列表里，每轮都追加而不是替换，模型永远能看到完整的上下文。

三、第二步：搭图——四个动作

有了状态，就可以画图了：

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START
import os

# 1. 创建绑定了状态结构的图构建器（后续所有装配都发生在这上面）
graph_builder = StateGraph(State)

# 2. 初始化模型（这里以 DeepSeek 为例，任何 OpenAI 兼容接口都行）
llm = init_chat_model("deepseek-chat", api_key=os.environ.get("DEEPSEEK_API_KEY"))

# 3. 定义节点函数：输入状态，返回"部分更新"
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 4. 注册节点 + 连边 + 编译
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

注意节点函数的契约：输入是完整状态，输出只需要返回你想改的字段。chatbot 只返回了 {"messages": [...]}，框架会把它交给 add_messages 追加进状态——节点不必关心"怎么合并"，只管"我要产出什么"。

四、第三步：流式输出与交互循环

编译后的图实例有两个执行入口：

方法	行为	适用场景
graph.invoke(状态)	同步阻塞，跑完返回最终状态	批量任务，如批量生成报告
graph.stream(状态)	返回事件迭代器，边执行边产出	聊天机器人等交互场景

流式版本长这样：

def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

graph.stream 返回的每个 event 是一个字典：键是节点名，值是该节点的最新状态。value["messages"][-1] 取最新一条消息——因为 add_messages 一直在追加，最后一条永远是刚生成的片段，于是屏幕上就有了逐段实时打印的效果。

最后包一层命令行循环：

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

运行效果：

User: who are you?
Assistant: I'm DeepSeek Chat, your AI assistant.

五、最小骨架清单

到目前为止，一个 LangGraph 应用的全部构成只有六件事：

组成	作用	对应 API
状态结构	定义共享数据的形状与合并规则	TypedDict + Annotated[..., add_messages]
图构建器	绑定状态结构的构建器	StateGraph(State)
节点	干活的执行单元	add_node("名字", 函数)
边	定义"下一步去哪"	add_edge(START, "节点名")
编译	校验并生成可执行实例	compile()
执行	传入初始状态启动流程	invoke() / stream()

小结

这个机器人虽然小，五脏俱全：状态存上下文，节点干实事，边定流程。往后要加记忆持久化、工具调用、条件分支，都是往这张图上继续加东西，骨架不用动。

下一篇，我们给这个机器人接上工具——让它在回答"明天天气怎么样"时，真正去调用一次天气工具，而不是瞎编。