LangChain Memory模块:对话历史管理最佳实践
引言
去年冬天,我帮一家做智能客服的团队做架构评审。他们的系统上线三个月,用户投诉量却一直降不下来。问题很有意思:用户在第一轮对话里说了"我买的是Pro版订阅",聊到第五轮时问"我的套餐能不能升级",机器人回复:"请问您购买的是哪个版本?"——用户当场就炸了。
技术团队排查后发现,问题不在模型,而在对话历史管理。他们用的是最朴素的方案:把每轮对话直接拼成一个不断增长的字符串塞进prompt。聊到十几轮后,要么超了上下文窗口被截断(把关键信息截掉了),要么token成本高得离谱,要么就是模型开始"遗忘"早期内容。
这不是个例。Memory(记忆)是LLM应用里最容易被低估、却最容易翻车的模块。模型本身是无状态的——它就像一个每天失忆的服务员,你每次点单都得重新自我介绍一遍。而Memory模块要做的,就是给这个失忆的服务员配一个"点单记录本",还要决定:记什么、记多久、怎么翻。
LangChain的Memory模块演进到今天,已经从早期的 ConversationBufferMemory 一路走到了与LangGraph深度整合的 checkpointer 体系。这篇文章我会带你从源码级别拆解这套机制,讲清楚为什么很多人的Memory用错了,以及在生产环境里到底该怎么设计。
核心概念:把Memory想象成一家餐厅的前台
我们先做个生活化类比。
想象一家高档餐厅,每位客人进门时,前台都要记录信息。这里有几种策略:
策略一:全记(Buffer Memory)
前台拿一个超长的本子,客人说的每句话都一字不落记下来。好处是信息完整,坏处是客人聊三小时,本子厚得翻不动,服务员每次上菜前都要从头读一遍——慢且贵。
策略二:只记最近几条(Window Memory)
前台只记最近5句话。翻页快,但客人开头说的"我对花生过敏"可能就被冲掉了——这是致命的。
策略三:记摘要(Summary Memory)
前台不动原话,每聊一段就写一句总结:"客人点了牛排,要求七分熟,忌花生。"本子薄了,但细节丢了,而且写总结本身也要花时间(调一次LLM)。
策略四:混合(Summary Buffer)
最近几轮记原话,更早的记摘要。这是生产环境里最常用的折中方案。
在LangChain的语境里,这些策略对应到几个核心抽象:
| 概念 | 作用 | 对应类 |
|---|---|---|
BaseChatMemory |
记忆的抽象基类,负责读写历史 | 所有Memory的父类 |
ChatMessageHistory |
底层存储,管理消息列表 | InMemoryChatMessageHistory 等 |
BaseChatMessageHistory |
存储接口,可对接Redis/DB | 需自行实现或使用集成包 |
RunnableWithMessageHistory |
现代LCEL范式下的记忆包装器 | 推荐用法 |
Checkpointer |
LangGraph的状态持久化机制 | MemorySaver / SqliteSaver |
关键认知:Memory本质上是"在调用LLM前,把历史消息注入prompt;在调用后,把新消息写回存储" 这个循环的封装。理解这一点,你就理解了所有Memory实现的本质。
源码深度分析:从 Buffer 到 Checkpointer 的演进
1. 老式Memory的读写循环(以 ConversationBufferMemory 为例)
先看最经典实现的骨架。LangChain早期版本中,BaseChatMemory 的核心逻辑大致是这样:
class BaseChatMemory(BaseMemory):
chat_memory: BaseChatMessageHistory
def save_context(self, inputs: Dict, outputs: Dict) -> None:
# 把用户输入和AI输出分别写入历史
input_str = self._get_input_output(inputs, "input")
output_str = self._get_input_output(outputs, "output")
self.chat_memory.add_user_message(input_str)
self.chat_memory.add_ai_message(output_str)
def load_memory_variables(self, inputs: Dict) -> Dict:
# 读取历史,按 memory_key 返回给prompt
return {self.memory_key: self.chat_memory.messages}这里的 memory_key 就是注入prompt的变量名。比如你设 memory_key="history",prompt模板里就得有 {history} 占位符。
这里藏着第一个大坑:load_memory_variables 是无状态的——它每次都把整个 chat_memory.messages 全量返回。也就是说,Buffer Memory的"记忆"完全靠 ChatMessageHistory 里的列表长度撑着。聊到第50轮,这个列表就有100条消息,全塞进prompt,token直接爆掉。
2. Window Memory 的滑动窗口逻辑
ConversationBufferWindowMemory 的源码关键在于它重写了 load_memory_variables:
class ConversationBufferWindowMemory(BaseChatMemory):
k: int = 5 # 只保留最近k轮
@property
def buffer(self) -> List[BaseMessage]:
# 核心:切片取最后 2*k 条消息
# 因为一轮对话 = user + ai 两条
return self.chat_memory.messages[-self.k * 2:] \
if self.k > 0 else []
def load_memory_variables(self, inputs: Dict) -> Dict:
# 注意:这里只影响"读",不影响"写"
# chat_memory里依然是全量存储
return {self.memory_key: self.buffer}注意这个设计:Window Memory只截断"读",chat_memory 里依然存了全量历史。这既是优点(历史可回查)也是缺点(内存会持续增长,除非你用外部存储)。
3. Summary Memory 的异步摘要陷阱
ConversationSummaryMemory 的机制更微妙。它在 save_context 里会调用一次LLM来更新摘要:
def save_context(self, inputs, outputs):
input_str = self._get_input_output(inputs, "input")
output_str = self._get_input_output(outputs, "output")
# 取出当前摘要
buffered = self.buffer # 字符串形式的摘要
# 调用LLM生成新摘要 —— 这里是一次额外的LLM调用!
new_summary = self.predict_new_summary(
self.chat_memory.messages, buffered
)
# 清空历史,只存摘要
self.chat_memory.clear()
self.chat_memory.add_message(SystemMessage(content=new_summary))这里有两个生产级问题:
- 每次 save 都多一次LLM调用,延迟和成本直接翻倍。
ConversationSummaryBufferMemory缓解了这个问题(攒够一定token才摘要),但本质还在。 - 摘要是有损压缩。上面那个客服案例里,"Pro版订阅"这种实体信息,很可能在摘要时被模糊成"用户询问了套餐相关",然后关键信息就丢了。
4. 现代范式:RunnableWithMessageHistory 与 LCEL
LangChain 0.1+ 之后,官方主推 LCEL(LangChain Expression Language)范式,Memory的用法变成了这样:
from langchain_core.runnables.history import RunnableWithMessageHistory
chain_with_history = RunnableWithMessageHistory(
chain, # 你的LCEL链
get_session_history, # 一个返回ChatMessageHistory的函数
input_messages_key="input", # 输入消息的key
history_messages_key="history", # 历史注入的key
)它的源码核心在 _enter_history 和 _exit_history 两个方法:
def _enter_history(self, input, config):
hist = self.get_session_history(config) # 按session_id取历史
messages = hist.messages.copy()
# 把历史消息合并进输入
return {**input, self.history_messages_key: messages}
def _exit_history(self, output, config, input):
hist = self.get_session_history(config)
# 把本轮输入输出写回
hist.add_user_message(input[self.input_messages_key])
hist.add_ai_message(output)这个范式的最大价值:get_session_history 是按 session_id 隔离的。多用户并发时,每个用户有独立的历史,不会串。这是老式全局 Memory 做不到的。
5. LangGraph Checkpointer:Memory 的终极形态
到了LangGraph,Memory的概念被彻底重构为 State + Checkpointer。整个对话是一个不断演进的 State 对象,而 Checkpointer 负责在每个节点执行后把 State 快照持久化。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
# 调用时传入 thread_id,即会话ID
config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [HumanMessage("你好")]}, config)为什么这是终极形态? 因为它解决了老Memory的三个根本缺陷:
- State是结构化的,不只是消息列表,可以包含用户画像、工具调用结果、中间推理状态。
- Checkpointer可按节点持久化,支持"时光旅行"(回滚到任意历史节点)、断点续跑、Human-in-the-loop。
- thread_id 天然隔离会话,且可对接 Postgres/SQLite 做真正的持久化。
下面是这几种方案的架构对比:
全量消息列表] B -->|Window| D[滑动窗口
只读最近K轮] B -->|Summary| E[LLM摘要
额外调用+有损] B -->|LangGraph| F[StateGraph State
+ Checkpointer] C --> G[注入Prompt] D --> G E --> G F --> H[按thread_id持久化
结构化State] G --> I[调用LLM] H --> I I --> J[写回存储] J --> C J --> F style F fill:#e1f5ff style H fill:#e1f5ff
从图中可以看到,前三者都是"消息列表 + 读写循环"的变体,而LangGraph把Memory上升到了状态管理的层面。
实战代码:三个可直接运行的示例
示例1:基于Redis的会话隔离Memory(生产级基础版)
这是最实用的方案——用Redis做外部存储,按session_id隔离,避免多用户串话。
# pip install langchain langchain-openai redis
import json
import redis
from typing import List, Sequence
from langchain_core.messages import BaseMessage, messages_from_dict, message_to_dict
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# ========== 1. 实现基于Redis的历史存储 ==========
class RedisChatMessageHistory(BaseChatMessageHistory):
"""将消息持久化到Redis List,每个session一个key"""
def __init__(self, session_id: str, redis_client: redis.Redis,
ttl: int = 3600, max_messages: int = 100):
self.session_id = session_id
self.redis = redis_client
self.ttl = ttl # 会话过期时间(秒)
self.max_messages = max_messages # 防止无限增长
@property
def key(self) -> str:
return f"chat:history:{self.session_id}"
@property
def messages(self) -> List[BaseMessage]:
# 从Redis读出所有消息并反序列化
raw = self.redis.lrange(self.key, 0, -1)
return [messages_from_dict(json.loads(item)) for item in raw]
def add_message(self, message: BaseMessage) -> None:
# 序列化单条消息写入
self.redis.rpush(self.key, json.dumps(message_to_dict(message)))
# 修剪:只保留最近 max_messages 条
self.redis.ltrim(self.key, -self.max_messages, -1)
# 刷新过期时间
self.redis.expire(self.key, self.ttl)
def clear(self) -> None:
self.redis.delete(self.key)
# ========== 2. 组装带记忆的链 ==========
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的技术顾问,回答简洁准确。"),
MessagesPlaceholder(variable_name="history"), # 历史注入点
("human", "{input}"),
])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
chain = prompt | llm
# 关键:按session_id返回对应的历史存储
def get_session_history(session_id: str) -> BaseChatMessageHistory:
return RedisChatMessageHistory(session_id, redis_client)
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
# ========== 3. 使用:不同用户天然隔离 ==========
config_user_a = {"configurable": {"session_id": "user-A"}}
config_user_b = {"configurable": {"session_id": "user-B"}}
r1 = chain_with_history.invoke({"input": "我叫张三,在做RAG系统"}, config_user_a)
print("A:", r1.content)
r2 = chain_with_history.invoke({"input": "我叫什么?在做什么?"}, config_user_a)
print("A回忆:", r2.content) # 会正确回答"张三、RAG"
r3 = chain_with_history.invoke({"input": "我叫什么?"}, config_user_b)
print("B回忆:", r3.content) # 会说"不知道",因为B是独立会话示例2:智能摘要 + 关键实体提取的混合Memory
这个方案解决"摘要丢关键信息"的问题——在摘要之外,额外维护一个结构化实体槽位。
# pip install langchain langchain-openai pydantic
from typing import List, Dict
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ========== 1. 用结构化输出提取关键实体 ==========
class UserFacts(BaseModel):
"""用户事实槽位,用于对抗摘要的信息损失"""
name: str = Field(default="", description="用户姓名")
subscription: str = Field(default="", description="订阅版本,如Pro/Free")
preferences: List[str] = Field(default_factory=list, description="偏好或禁忌")
goal: str = Field(default="", description="用户的核心诉求")
extract_prompt = ChatPromptTemplate.from_messages([
("system", "从对话中提取用户事实。若某字段对话中未提及,保持原值不变。"),
("human", "已有事实:{existing}\n\n最新对话:\n{dialogue}")
])
extractor = extract_prompt | llm.with_structured_output(UserFacts)
# ========== 2. 混合Memory实现 ==========
class HybridMemory:
"""
结构:
- recent_messages: 最近N轮原文(保证细节)
- summary: 更早对话的摘要(压缩历史)
- facts: 结构化用户事实(抗损失)
"""
def __init__(self, recent_window: int = 6, summary_trigger: int = 10):
self.recent_messages: List[BaseMessage] = []
self.summary: str = ""
self.facts: UserFacts = UserFacts()
self.recent_window = recent_window # 保留最近几轮原文
self.summary_trigger = summary_trigger # 超过多少条触发摘要
def add_turn(self, user_msg: str, ai_msg: str) -> None:
self.recent_messages.append(HumanMessage(content=user_msg))
self.recent_messages.append(AIMessage(content=ai_msg))
# 更新结构化事实
dialogue = f"用户:{user_msg}\n助手:{ai_msg}"
self.facts = extractor.invoke({
"existing": self.facts.model_dump_json(),
"dialogue": dialogue
})
# 超出窗口时触发摘要压缩
if len(self.recent_messages) > self.summary_trigger:
self._compress()
def _compress(self) -> None:
# 把超出窗口的早期消息压缩进摘要
to_compress = self.recent_messages[:-self.recent_window]
keep = self.recent_messages[-self.recent_window:]
old_text = "\n".join(f"{m.type}: {m.content}" for m in to_compress)
summary_prompt = ChatPromptTemplate.from_messages([
("system", "把对话压缩成简洁摘要,保留关键决策和事实。"),
("human", "已有摘要:{old_summary}\n\n新增对话:\n{new_text}")
])
chain = summary_prompt | llm
result = chain.invoke({"old_summary": self.summary, "new_text": old_text})
self.summary = result.content
self.recent_messages = keep
def build_prompt_messages(self) -> List[BaseMessage]:
# 组装最终注入prompt的消息
ctx = (
f"【用户档案】\n"
f"姓名:{self.facts.name}\n"
f"订阅:{self.facts.subscription}\n"
f"偏好:{', '.join(self.facts.preferences) or '无'}\n"
f"诉求:{self.facts.goal}\n\n"
f"【历史摘要】\n{self.summary or '(无)'}"
)
return [SystemMessage(content=ctx)] + self.recent_messages
# ========== 3. 使用演示 ==========
memory = HybridMemory(recent_window=4, summary_trigger=6)
conversation = [
("你好,我是李四,买的是Pro版订阅", "你好李四,Pro版支持API调用,请问有什么可以帮您?"),
("我对花生过敏,点餐时请注意", "已记录,我会提醒相关服务。"),
("我想问下我的套餐能否升级", "Pro版可以升级到企业版,需要我介绍吗?"),
("先不用,我就想知道现在能干嘛", "Pro版支持每月10万次API调用和优先支持。"),
("对了,我叫什么来着,买的什么版", "您是李四,购买的是Pro版订阅。"),
]
for user, ai in conversation:
memory.add_turn(user, ai)
# 关键验证:即使经过压缩,实体信息依然保留
print("事实槽位:", memory.facts.model_dump())
print("\n最终prompt上下文:")
for m in memory.build_prompt_messages():
print(f" [{m.type}] {m.content[:80]}...")这个方案的精髓:摘要负责"压缩上下文长度",结构化事实负责"对抗信息损失"。两者互补,这才是生产级Memory该有的样子。
示例3:基于LangGraph的持久化对话(推荐架构)
这是2026年的主流做法。用 StateGraph 管理对话状态,用 SqliteSaver 做持久化,支持多轮、可回滚、可恢复。
# pip install langgraph langchain-openai langgraph-checkpoint-sqlite
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, SystemMessage
# ========== 1. 定义状态结构 ==========
class AgentState(TypedDict):
# add_messages 是reducer:新消息会追加而非覆盖
messages: Annotated[list[BaseMessage], add_messages]
# 可以扩展自定义字段,如用户画像
user_profile: dict
# ========== 2. 构建图 ==========
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
def chatbot_node(state: AgentState) -> dict:
"""核心对话节点:读取历史,生成回复"""
system = SystemMessage(
content=f"你是技术顾问。用户档案:{state.get('user_profile', {})}"
)
# state["messages"] 里已经包含了完整历史
response = llm.invoke([system] + state["messages"])
return {"messages": [response]} # 追加到状态
builder = StateGraph(AgentState)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
# ========== 3. 挂载持久化Checkpointer ==========
# 生产环境建议用 PostgresSaver,这里用SQLite演示
with SqliteSaver.from_conn_string("chat_history.db") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "session-001"}}
# 第一轮
r1 = graph.invoke(
{"messages": [("user", "我是王五,在做多模态Agent")],
"user_profile": {"name": "王五", "domain": "多模态Agent"}},
config
)
print("AI:", r1["messages"][-1].content)
# 第二轮 —— 注意:无需手动传历史,checkpointer自动恢复
r2 = graph.invoke(
{"messages": [("user", "我刚才说我做什么来着?")]},
config
)
print("AI:", r2["messages"][-1].content) # 会正确回忆
# ========== 4. 时光旅行:查看历史状态 ==========
history = list(graph.get_state_history(config))
print(f"\n共有 {len(history)} 个状态快照")
for snapshot in history[:3]:
print(f" checkpoint_id: {snapshot.config['configurable']['checkpoint_id']}")
# ========== 5. 从任意历史点分叉 ==========
# 取倒数第二个checkpoint,重新走一条不同的路
if len(history) >= 2:
fork_config = history[1].config
forked = graph.invoke(
{"messages": [("user", "换个话题,讲讲RAG的chunking策略")]},
fork_config
)
print("\n分叉后:", forked["messages"][-1].content)LangGraph方案的核心优势在这段代码里体现得淋漓尽致:
- 无需手动管理历史:
add_messagesreducer 自动追加,checkpointer 自动加载。
- thread_id 隔离:换个 id 就是全新会话。
- 时光旅行:
get_state_history能拿到所有快照,可以从任意点分叉重跑。
- 持久化:进程重启后,
thread_id不变就能恢复对话。
方案对比:LangChain Memory vs 其他业界方案
| 维度 | LangChain老式Memory | LangGraph Checkpointer | 纯手写 | LlamaIndex Memory |
|---|---|---|---|---|
| 会话隔离 | 需自行实现 | thread_id原生 | 需自行实现 | 支持 |
| 持久化 | 需对接外部存储 | 内置多种Saver | 完全自定义 | 内置 |
| 状态结构 | 仅消息列表 | 任意结构化State | 任意 | 消息+记忆块 |
| 时光旅行 | 不支持 | 支持 | 需自建 | 部分支持 |
| 摘要压缩 | 内置(有损) | 需自建节点 | 灵活 | 内置 |
| 学习成本 | 低 | 中高 | 高 | 中 |
| 生产推荐度 | 不推荐新项目 | ⭐⭐⭐⭐⭐ | 特定场景 | ⭐⭐⭐⭐ |
选型建议:
- 新项目直接用 LangGraph。它的 State + Checkpointer 范式是当前最成熟的方案,尤其是需要多轮、需要持久化、需要人机协作的场景。
- 已有LCEL项目用
RunnableWithMessageHistory平滑迁移,成本最低。
- 手写只在你需要极致控制(比如自定义的向量化记忆、图数据库记忆)时考虑。
- LlamaIndex 的
ChatMemoryBuffer+VectorMemory在检索式记忆上做得更细,如果你的核心诉求是"从海量历史里检索相关片段",值得一看。
一个容易被忽略的点:向量化记忆(Vector Memory)。当历史特别长(比如客服场景积累了几百轮),可以把历史消息embedding后存向量库,每轮只检索最相关的K条注入。LangChain有 VectorStoreRetrieverMemory,但它本质是RAG思路在Memory上的应用——注意,它不保证时间连续性,可能检索到语义相关但时间错乱的内容,慎用于强时序场景。
最佳实践与避坑指南
坑1:把Memory当"无限存储"用
现象:用 ConversationBufferMemory 上线,跑了几天发现token成本暴涨、响应变慢。
根因:Buffer Memory 每次把全量历史塞进prompt,token随轮次线性增长(实际是超线性,因为每轮都要重发之前所有内容)。
解法:永远给Memory设上限。要么用Window,要么用摘要,要么用LangGraph在节点里做裁剪。
坑2:摘要丢关键信息
现象:用户说过的具体数字、版本号、订单号,在几轮后"消失"了。
根因:摘要是有损压缩,LLM会优先保留"语义"而丢弃"实体"。
解法:像示例2那样,摘要 + 结构化槽位双轨并行。实体信息进槽位,语义信息进摘要。
坑3:多用户串话
现象:A用户的问题,B用户收到了相关回复。
根因:用了全局单例Memory,或者session_id生成逻辑有bug。
解法:强制每个请求携带唯一session_id/thread_id,且隔离存储。示例1的Redis方案就是标准做法。
坑4:忽略消息顺序与角色
现象:模型偶尔"角色混乱",把AI的话当用户说的。
根因:手动拼接历史时把 HumanMessage / AIMessage 搞混了。
解法:永远用 BaseMessage 子类,别用裸字符串。LCEL和LangGraph天然保证了这一点。
坑5:Memory写入时机错误
现象:工具调用失败时,错误信息被写进了历史,污染后续对话。
解法:在LangGraph里,把Memory写入放在成功节点之后,或者用条件边控制。老式Memory的 save_context 是"先写后判",要小心。
最佳实践清单
- 会话ID必带:每个请求都要有唯一标识,别用全局Memory。
- 设上限:无论哪种Memory,都要有token或条数上限。
- 结构化优先:能用结构化槽位存的,别指望摘要。
- 持久化到外部:生产环境别用
InMemory,用Redis/Postgres。 - 可观测:记录每轮注入的token数、历史条数,方便调优。
- 新项目上LangGraph:State + Checkpointer 是当下最优解。
- 定期清理:会话过期策略(TTL)能省下大量存储成本。
总结
回到开头那个客服案例。如果他们当初用的是"结构化事实槽位 + 摘要 + 最近窗口"的混合方案,或者直接上LangGraph的Checkpointer,"Pro版订阅"这个信息就永远不会丢。
我们把这篇文章的要点串一下:
- Memory的本质是"调用前注入历史,调用后写回"的循环封装。
- 演进路径:Buffer → Window → Summary → RunnableWithMessageHistory → LangGraph Checkpointer,每一步都在解决前一步的缺陷。
- 源码层面:老式Memory靠
load_memory_variables/save_context两个钩子,LangGraph靠State reducer + Checkpointer。
- 生产方案:结构化事实对抗摘要损失,thread_id隔离会话,外部存储保证持久化。
- 三条铁律:会话必隔离、历史必设限、关键信息必结构化。
最后留一个延伸思考:Memory的边界正在模糊。当上下文窗口越来越大(百万token级别),"压缩历史"这件事的价值在下降;但"检索相关历史"和"结构化用户状态"的价值在上升。未来的Memory模块,可能会从"管理对话记录"演变成"管理用户长期记忆"——那时候,它离真正的Agent记忆系统就不远了。
而LangGraph的State + Checkpointer架构,正是朝这个方向迈出的关键一步。
*参考:LangChain官方文档 langchain_core.runnables.history、langgraph.checkpoint 模块源码;LangGraph 0.2.x API。*