一、为什么 Agent 需要多种记忆？认知科学启示

1.1 人类记忆系统的类比

人类记忆类型	特点	Agent 记忆对应
感觉记忆	毫秒级暂存感官输入	输入预处理缓冲区
短期记忆 (STM)	容量 7±2 项，持续 15-30 秒	Short-Term Memory
工作记忆 (WM)	STM + 中央执行系统	Agent 的上下文管理
长期记忆 (LTM)	几乎无限容量，持久存储	Long-Term / External Memory
情景记忆	基于时间/事件的记忆	Conversation History
语义记忆	事实/概念知识	Vector Database

💡 关键洞察：
Agent 的"智能" = 短期记忆的高效利用 + 长期记忆的精准检索

1.2 LangChain 记忆架构全景

---

二、Short-Term Memory：对话上下文的生命线

2.1 本质与作用

• 定义：存储 当前对话轮次 的上下文信息
• 载体：LLM 的 Context Window（如 GPT-4o 的 128K tokens）
• 核心功能：维持对话连贯性，避免"失忆"

2.2 实现方式：ConversationBufferMemory

from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# 初始化短时记忆
memory = ConversationBufferMemory()
llm = ChatOpenAI(model="gpt-4o")
chain = ConversationChain(llm=llm, memory=memory)

# 对话示例
response1 = chain.invoke("我叫张三")
response2 = chain.invoke("我叫什么名字？")  # 正确回答"张三"

2.3 关键特性

特性	说明	限制
自动维护	自动保存用户/Agent 的对话历史	无
Token 感知	自动截断超长历史（需配合 TokenWindowMemory）	默认不截断
序列化支持	可保存/加载对话历史	需手动实现

2.4 性能瓶颈与优化

问题：Context Window 溢出

# 当对话过长时，旧消息被丢弃
for i in range(100):
    chain.invoke(f"消息 {i}")
# 此时询问"第一条消息是什么？" → LLM 无法回答

解决方案1：TokenWindowMemory（自动截断）

from langchain.memory import ConversationBufferWindowMemory

# 仅保留最近 5 轮对话
memory = ConversationBufferWindowMemory(k=5)

解决方案2：SummaryMemory（摘要压缩）

from langchain.memory import ConversationSummaryMemory

# 用 LLM 生成对话摘要
memory = ConversationSummaryMemory(llm=llm)

⚠️ 代价权衡：

• BufferWindow：丢失早期细节
• Summary：消耗额外 tokens，可能丢失关键信息

---

三、Long-Term Memory：持久化对话历史

3.1 本质与作用

• 定义：跨会话持久存储用户交互历史
• 载体：本地文件、数据库、云存储
• 核心功能：实现 个性化体验（如记住用户偏好）

3.2 实现方式：Filesystem Persisted Memory

import json
from langchain.memory import ConversationBufferMemory

class PersistentMemory:
    def __init__(self, user_id: str, file_path: str = "memories/"):
        self.user_id = user_id
        self.file_path = f"{file_path}{user_id}.json"
        self.memory = self.load_memory()
    
    def load_memory(self) -> ConversationBufferMemory:
        try:
            with open(self.file_path, 'r') as f:
                messages = json.load(f)
                memory = ConversationBufferMemory()
                for msg in messages:
                    if msg['type'] == 'human':
                        memory.chat_memory.add_user_message(msg['content'])
                    else:
                        memory.chat_memory.add_ai_message(msg['content'])
                return memory
        except FileNotFoundError:
            return ConversationBufferMemory()
    
    def save_memory(self):
        messages = []
        for msg in self.memory.chat_memory.messages:
            messages.append({
                'type': 'human' if msg.type == 'human' else 'ai',
                'content': msg.content
            })
        with open(self.file_path, 'w') as f:
            json.dump(messages, f, ensure_ascii=False, indent=2)

# 使用示例
pm = PersistentMemory("user_123")
chain = ConversationChain(llm=llm, memory=pm.memory)
chain.invoke("我喜欢咖啡")
pm.save_memory()  # 下次会话仍记得偏好

3.3 数据库集成（生产级方案）

from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class ConversationHistory(Base):
    __tablename__ = 'conversations'
    id = Column(Integer, primary_key=True)
    user_id = Column(String(50))
    message_type = Column(String(10))  # 'human' or 'ai'
    content = Column(Text)
    timestamp = Column(String(30))

class DatabaseMemory:
    def __init__(self, database_url: str, user_id: str):
        self.engine = create_engine(database_url)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.user_id = user_id
    
    def load_memory(self) -> ConversationBufferMemory:
        session = self.Session()
        records = session.query(ConversationHistory)\
                        .filter_by(user_id=self.user_id)\
                        .order_by(ConversationHistory.id)\
                        .all()
        memory = ConversationBufferMemory()
        for record in records:
            if record.message_type == 'human':
                memory.chat_memory.add_user_message(record.content)
            else:
                memory.chat_memory.add_ai_message(record.content)
        session.close()
        return memory
    
    def save_message(self, message_type: str, content: str):
        session = self.Session()
        record = ConversationHistory(
            user_id=self.user_id,
            message_type=message_type,
            content=content,
            timestamp=datetime.now().isoformat()
        )
        session.add(record)
        session.commit()
        session.close()

3.4 隐私与安全考量

• 数据加密：敏感信息需 AES 加密存储
• GDPR 合规：提供用户数据删除接口
• 最小化存储：仅保存必要对话片段

from cryptography.fernet import Fernet

class EncryptedMemory(PersistentMemory):
    def __init__(self, user_id: str, key: bytes):
        self.cipher = Fernet(key)
        super().__init__(user_id)
    
    def save_memory(self):
        # 加密后存储
        raw_data = json.dumps(messages)
        encrypted = self.cipher.encrypt(raw_data.encode())
        with open(self.file_path, 'wb') as f:
            f.write(encrypted)

---

四、External Memory：连接外部知识源

4.1 本质与作用

• 定义：实时查询 外部系统 获取信息
• 载体：API、数据库、文件系统、企业知识库
• 核心功能：突破 LLM 的 知识截止限制

4.2 典型应用场景

场景	外部源	工具实现
实时股价	Yahoo Finance API	yfinance 工具
订单状态	企业 ERP 数据库	SQL 查询工具
最新新闻	RSS/新闻 API	Web 搜索工具
用户资料	CRM 系统	REST API 工具

4.3 实战：动态股票查询工具

import yfinance as yf
from langchain_core.tools import tool

@tool
def get_stock_price(symbol: str) -> str:
    """Get current stock price for a given symbol"""
    try:
        ticker = yf.Ticker(symbol)
        hist = ticker.history(period="1d")
        price = hist['Close'].iloc[-1]
        return f"{symbol} price: ${price:.2f}"
    except Exception as e:
        return f"Error fetching {symbol}: {str(e)}"

# 集成到 Agent
from langchain import hub
from langchain.agents import create_tool_calling_agent

prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_tool_calling_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[get_stock_price],
    prompt=prompt
)

# 用户问："苹果公司股价多少？"
# Agent 自动调用 get_stock_price("AAPL")

4.4 外部记忆 vs Vector 记忆

维度	External Memory	Vector Memory
数据新鲜度	实时（秒级）	静态（需定期更新）
查询精度	精确匹配（如 ID 查询）	语义相似度
延迟	高（依赖外部 API）	低（本地向量检索）
成本	按 API 调用计费	一次性嵌入成本
适用场景	结构化数据、实时数据	非结构化文档、知识库

💡 黄金法则：
“精确查询用 External，模糊检索用 Vector”

---

五、Vector Memory：语义知识库的核心

5.1 本质与作用

• 定义：将文本转换为 向量 embeddings，通过相似度检索
• 载体：Chroma、Pinecone、Weaviate 等向量数据库
• 核心功能：实现 语义级知识召回

5.2 工作原理

5.3 实战：构建企业知识库

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 1. 加载文档
loader = PyPDFLoader("company_handbook.pdf")
docs = loader.load()

# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(docs)

# 3. 创建向量库
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

# 4. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 5. 集成到 Agent
from langchain.tools.retriever import create_retriever_tool

handbook_tool = create_retriever_tool(
    retriever,
    name="company_handbook",
    description="Search company policies and procedures"
)

agent = create_tool_calling_agent(
    llm=ChatOpenAI(),
    tools=[handbook_tool],
    prompt=prompt
)

5.4 高级优化技巧

技巧1：混合检索（Hybrid Search）

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# 关键词检索 + 向量检索
bm25_retriever = BM25Retriever.from_documents(splits)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]
)

技巧2：元数据过滤

# 仅检索特定部门的文档
filtered_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 3,
        "filter": {"department": "HR"}
    }
)

技巧3：重排序（Rerank）

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

compressor = CohereRerank(top_n=3)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

---

六、四类记忆的协同作战：真实案例解析

6.1 案例：智能客服 Agent

class CustomerServiceAgent:
    def __init__(self, user_id: str):
        # Short-Term: 当前对话上下文
        self.short_memory = ConversationBufferMemory()
        
        # Long-Term: 用户历史记录
        self.long_memory = PersistentMemory(user_id)
        
        # External: 订单系统 API
        self.order_tool = create_order_tool()
        
        # Vector: 产品知识库
        self.knowledge_tool = create_knowledge_tool()
        
        # 组合所有工具
        tools = [self.order_tool, self.knowledge_tool]
        self.agent = create_tool_calling_agent(
            llm=ChatOpenAI(),
            tools=tools,
            prompt=self.build_prompt()
        )
    
    def build_prompt(self):
        # 动态注入长期记忆
        user_history = self.long_memory.get_summary()
        base_prompt = hub.pull("customer-service-prompt")
        return base_prompt.partial(user_context=user_history)
    
    def handle_query(self, query: str):
        # 合并短时+长时记忆
        full_memory = self.combine_memories()
        
        response = self.agent.invoke({
            "input": query,
            "chat_history": full_memory
        })
        
        # 更新所有记忆
        self.short_memory.save_context({"input": query}, {"output": response})
        self.long_memory.save_message("human", query)
        self.long_memory.save_message("ai", response)
        self.long_memory.save()
        
        return response

6.2 记忆调用优先级策略

用户查询类型	优先使用的记忆	示例
上下文相关	Short-Term	“刚才说的那个方案…”
个性化需求	Long-Term	“像上次一样订咖啡”
实时数据	External	“我的订单到哪了？”
知识问答	Vector	“退货政策是什么？”

6.3 性能监控指标

class MemoryMonitor:
    def __init__(self):
        self.metrics = {
            "short_term_tokens": 0,
            "long_term_hits": 0,
            "external_calls": 0,
            "vector_retrievals": 0
        }
    
    def log_short_term(self, tokens: int):
        self.metrics["short_term_tokens"] += tokens
    
    def log_external_call(self, tool_name: str):
        self.metrics["external_calls"] += 1
        print(f"Called external tool: {tool_name}")

---

七、避坑指南：常见错误与解决方案

7.1 错误1：Short-Term 记忆爆炸

现象：对话越长，响应越慢，最终超 token 限制
解决方案：

• 使用 ConversationSummaryBufferMemory（摘要+窗口）
• 设置最大 token 限制：

  from langchain.memory import MaxTokenMemory
  
  memory = MaxTokenMemory(
      llm=llm,
      max_token_limit=2000,
      memory_key="chat_history"
  )
  

7.2 错误2：Vector 记忆检索噪声

现象：返回不相关文档，导致 LLM 生成幻觉
解决方案：

• 调整相似度阈值：

  retriever = vectorstore.as_retriever(
      search_kwargs={"score_threshold": 0.7}
  )
  

• 添加查询扩展：

  from langchain.retrievers.multi_query import MultiQueryRetriever
  
  multi_retriever = MultiQueryRetriever.from_llm(
      retriever=retriever,
      llm=ChatOpenAI()
  )
  

7.3 错误3：External 记忆权限泄露

现象：Agent 暴露了内部 API 密钥或敏感数据
解决方案：

• 工具函数中过滤敏感字段：

  def safe_order_lookup(order_id: str):
      data = internal_api.get_order(order_id)
      # 仅返回必要字段
      return {
          "status": data.status,
          "estimated_delivery": data.delivery_date
      }
  

• 使用沙箱环境执行工具

7.4 错误4：Long-Term 记忆隐私违规

现象：存储了用户身份证号、银行卡等 PII 信息
解决方案：

• 预处理对话历史，移除 PII：

  import re
  
  def remove_pii(text: str) -> str:
      # 移除身份证号
      text = re.sub(r'\d{17}[\dXx]', '[REDACTED]', text)
      # 移除银行卡号
      text = re.
      
  # 在保存前清洗
  clean_msg = remove_pii(user_message)
  memory.save(clean_msg)
  

---

八、未来展望：记忆系统的演进方向

8.1 神经符号系统（Neuro-Symbolic）

• 结合：LLM 的泛化能力 + 符号系统的精确推理
• 应用：用图数据库存储实体关系，LLM 进行路径推理

8.2 记忆压缩技术

• 方法：使用小型模型对对话历史进行摘要
• 优势：减少 90% 的 token 消耗，保持关键信息

8.3 主动记忆管理

• 机制：Agent 自主决定何时存储/遗忘信息
• 示例：

  if "important" in user_message:
      long_memory.save_with_priority(message, priority=HIGH)
  

8.4 跨 Agent 记忆共享

• 场景：多个 Agent 协作时共享知识库
• 实现：分布式向量数据库 + 访问控制

---

九、总结：记忆选型决策树

终极建议：
“从 Short-Term 开始，按需叠加其他记忆类型。过度设计记忆系统是 Agent 开发的最大陷阱！”

---

附录 A：记忆类型速查表

记忆类型	存储位置	持久性	延迟	适用场景
Short-Term	LLM Context	会话内	极低	对话连贯性
Long-Term	文件/DB	跨会话	低	用户画像
External	外部系统	实时	高	动态数据
Vector	向量库	静态	中	知识检索

附录 B：LangChain 记忆官方文档

• Memory Types
• Vector Stores
• Agents with Memory