LangChain与LangGraph实现AI智能体记忆机制详解-平芜编程栈

1. 为什么需要带记忆的AI智能体

在传统对话系统中，每次交互都被视为独立事件。这种"健忘症"设计导致用户体验割裂——用户不得不反复解释相同需求，就像每次走进咖啡店都要重新介绍自己喜好一样低效。我们以客服场景为例：

无记忆系统：用户第5次询问"我的订单状态"时，机器人依然要求提供订单号
有记忆系统：能自动关联用户历史订单，直接显示最新物流信息

LangChain与LangGraph提供的记忆机制，本质上是在对话流中构建了状态感知层。这个设计借鉴了人类工作记忆（Working Memory）的认知模型：

短期记忆：相当于大脑的临时记事本，保存当前对话的上下文（如最近3轮问答）
长期记忆：类似经验库，存储跨会话的重要信息（如用户偏好、历史行为模式）

关键区别：短期记忆随对话结束而清空，长期记忆会持久化到数据库

2. 环境搭建与工具链配置

2.1 基础环境准备

推荐使用Python 3.10+和虚拟环境隔离依赖：

# 创建并激活虚拟环境 python -m venv langchain_env source langchain_env/bin/activate # Linux/Mac langchain_env\Scripts\activate # Windows # 安装核心库 pip install langchain langgraph openai tiktoken

2.2 记忆存储方案选型

根据数据敏感性选择存储后端：

存储类型	适用场景	安装命令
InMemoryStore	开发测试	内置无需安装
Redis	生产环境高频访问	`pip install redis`
PostgreSQL	结构化记忆数据	`pip install psycopg2-binary`
Chroma	向量化记忆检索	`pip install chromadb`

对于长期记忆，建议采用分层存储策略：

from langgraph.store import RedisStore, InMemoryStore short_term_memory = InMemoryStore() # 短期记忆 long_term_memory = RedisStore( # 长期记忆 host="redis.example.com", port=6379, db=0 )

3. 对话记忆的实现细节

3.1 短期记忆管理实战

LangGraph通过StateGraph维护对话状态，典型实现包含三个关键组件：

记忆注入器：将历史对话注入当前上下文

from langchain_core.messages import HumanMessage, AIMessage def inject_memory(state): # 保留最近5轮对话作为短期记忆 kept_messages = state["messages"][-5*2:] # 假设每轮包含用户+AI各1条消息 return {"messages": kept_messages}

状态检查点：自动持久化对话状态

from langgraph.graph import StateGraph workflow = StateGraph(Annotated[dict, "state"]) workflow.add_node("chat", chat_agent) workflow.set_entry_point("chat") app = workflow.compile(checkpointer=FileSystemCheckpointer("./checkpoints"))

上下文窗口优化：采用滑动窗口+摘要技术

from langchain.chains import LLMChain from langchain.prompts import PromptTemplate summary_prompt = PromptTemplate.from_template( "将以下对话压缩为3句话摘要:\n{dialogue}" ) def summarize_history(messages): dialogue = "\n".join([f"{m.type}: {m.content}" for m in messages]) return LLMChain(llm=llm, prompt=summary_prompt).run(dialogue)

3.2 长期记忆的四种模式

根据LangGraph文档，长期记忆可分为四种类型：

语义记忆：结构化事实存储

# 存储用户偏好 long_term_memory.put( ("user_prefs", user_id), "coffee_preference", {"type": "latte", "temp": "hot", "size": "grande"} )

情景记忆：对话事件记录

# 记录服务调用历史 long_term_memory.put( ("service_logs", user_id), str(uuid.uuid4()), { "timestamp": datetime.now(), "service": "order_status", "params": {"order_id": "12345"} } )

程序性记忆：操作流程模板

# 保存客服话术模板 long_term_memory.put( ("procedural", "cs_scripts"), "refund_process", { "steps": [ "确认订单信息", "解释退款政策", "提供解决方案选项" ], "phrases": { "empathy": "理解您的不便...", "resolution": "我们可以提供..." } } )

混合记忆：复合查询示例

# 跨类型记忆检索 def get_user_context(user_id): prefs = long_term_memory.get(("user_prefs", user_id)) logs = long_term_memory.search(("service_logs", user_id), limit=3) return {"preferences": prefs, "history": logs}

4. 典型问题与调优策略

4.1 记忆污染问题

当记忆库积累无效数据时会出现"幻觉响应"，解决方案：

记忆清洗机制：

def clean_memory(user_id): # 自动过期30天前的记录 old_logs = long_term_memory.search( ("service_logs", user_id), filter={"timestamp": {"$lt": datetime.now()-timedelta(days=30)}} ) for log in old_logs: long_term_memory.delete(("service_logs", user_id), log.key) # 低频偏好自动降级 prefs = long_term_memory.get(("user_prefs", user_id)) for k, v in prefs.items(): if v["last_used"] < datetime.now()-timedelta(days=90): long_term_memory.put( ("archive", user_id), k, {"value": v, "reason": "inactive"} ) long_term_memory.delete(("user_prefs", user_id), k)

人工审核接口：

@app.route("/review_memories/<user_id>") def review_memories(user_id): flagged = detect_sensitive_content( long_term_memory.search(("user_prefs", user_id)) ) return render_template("memory_review.html", items=flagged)

4.2 性能优化方案

针对高频访问场景的优化技巧：

记忆缓存策略：

from functools import lru_cache @lru_cache(maxsize=1000) def get_cached_memory(namespace, key): return long_term_memory.get(namespace, key)

批量异步写入：

import asyncio async def batch_update(memories): semaphore = asyncio.Semaphore(10) # 控制并发量 async def process_one(item): async with semaphore: await long_term_memory.aput(*item) await asyncio.gather(*[process_one(item) for item in memories])

向量索引加速：

from langchain.embeddings import OpenAIEmbeddings # 为记忆建立向量索引 embeddings = OpenAIEmbeddings() long_term_memory.create_index( "semantic_search", dims=1536, # OpenAI向量维度 embedding_fn=embeddings.embed_query )

5. 完整实现案例：电商客服智能体

5.1 架构设计

graph TD A[用户提问] --> B{记忆查询} B -->|新用户| C[基础应答流程] B -->|老用户| D[个性化应答] C --> E[记录到短期记忆] D --> F[调用长期记忆] F --> G[生成定制响应] G --> H[更新记忆] H --> E

5.2 核心代码实现

from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], "对话历史"] user_id: str def retrieve_memory(state: AgentState): # 获取长期记忆 memories = long_term_memory.search( ("user_history", state["user_id"]), query=state["messages"][-1].content ) return {"memories": memories} def generate_response(state: AgentState): # 结合记忆生成响应 context = { "history": state["messages"], "memories": state.get("memories", []) } response = chat_agent.invoke(context) # 保存重要信息到长期记忆 if should_remember(response): long_term_memory.put( ("user_history", state["user_id"]), str(uuid.uuid4()), { "content": response.content, "metadata": extract_entities(response) } ) return {"messages": [response]} workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve_memory) workflow.add_node("respond", generate_response) workflow.add_edge("retrieve", "respond") workflow.set_entry_point("retrieve") app = workflow.compile()