记忆系统：短期与长期记忆

为什么记忆如此重要

LLM 本身是无状态的——每次 API 调用都是独立的。要让 Agent 具备"记忆"，需要在应用层管理和传递上下文。记忆系统的设计直接影响 Agent 的：

• 连贯性：是否能记住之前的对话和任务进展
• 成本：历史越长，每次调用的 Token 消耗越大
• 准确性：过长的上下文可能导致 LLM 注意力分散
• 持久化：应用重启后是否能恢复会话

短期记忆：对话历史管理

最简单的记忆形式是直接在 State 中维护消息列表。LangGraph 的 add_messages reducer 自动处理消息追加：

from langchain_core.messages import (
    BaseMessage, HumanMessage, AIMessage,
    SystemMessage, trim_messages
)
from langgraph.graph.message import add_messages
from typing import Annotated

# ── 基础：完整消息历史 ────────────────────────────────────
class BufferMemoryState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# ── 高级：滑动窗口 - 只保留最近 N 条消息 ─────────────────
def sliding_window_reducer(existing: list, new: list) -> list:
    """只保留最近 20 条消息（10轮对话）。"""
    combined = add_messages(existing, new)
    return combined[-20:]

class WindowMemoryState(TypedDict):
    messages: Annotated[list[BaseMessage], sliding_window_reducer]

# ── Token 感知裁剪（LangChain 内置）─────────────────────
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def agent_with_trim(state: BufferMemoryState):
    # trim_messages：按 Token 数裁剪，保留 system + 最近的对话
    trimmed = trim_messages(
        state["messages"],
        max_tokens=4000,
        strategy="last",         # 保留最新的
        token_counter=llm,
        include_system=True,    # 保留 SystemMessage
        allow_partial=False,    # 不允许截断单条消息
        start_on="human"        # 从 Human 消息开始
    )
    response = llm.invoke(trimmed)
    return {"messages": [response]}

摘要记忆：ConversationSummaryMemory

当对话变长时，将历史压缩为摘要是平衡 Token 成本与上下文保留的最佳策略：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class SummaryMemoryState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    summary: str          # 历史对话的压缩摘要
    message_count: int    # 消息计数，用于触发压缩

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是对话摘要助手。将以下对话历史压缩为简洁摘要。
如果已有摘要，请将新对话融入其中更新摘要。
摘要应保留：用户目标、关键决策、重要信息、未完成任务。
现有摘要：{existing_summary}"""),
    ("human", "请摘要以下新对话：\n{new_messages}")
])

summarizer = SUMMARY_PROMPT | ChatOpenAI(model="gpt-4o-mini")

def summarize_if_needed(state: SummaryMemoryState):
    """当消息超过阈值时，压缩历史消息。"""
    messages = state["messages"]

    # 每累计 10 条新消息触发一次压缩
    if len(messages) < 10:
        return {}  # 无需压缩

    # 保留最新 4 条（2轮对话），压缩其余
    to_summarize = messages[:-4]
    to_keep = messages[-4:]

    new_summary = summarizer.invoke({
        "existing_summary": state.get("summary", ""),
        "new_messages": "\n".join(
            f"{m.__class__.__name__}: {m.content}"
            for m in to_summarize
        )
    })

    return {
        "summary": new_summary.content,
        "messages": to_keep  # 替换为压缩后的短列表
    }

def agent_with_summary(state: SummaryMemoryState):
    """Agent 节点：将摘要注入为 System Message。"""
    system_content = "你是智能助手。"
    if state.get("summary"):
        system_content += f"\n\n【历史对话摘要】\n{state['summary']}"

    messages = [SystemMessage(content=system_content)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

向量存储记忆：长期知识库

向量存储记忆允许 Agent 检索与当前问题语义相关的历史信息，突破上下文窗口限制：

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
import uuid

# ── 初始化向量存储 ────────────────────────────────────────
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(
    collection_name="agent_memory",
    embedding_function=embeddings,
    persist_directory="./agent_memory_db"
)

# ── 写入记忆：将重要信息存入向量库 ───────────────────────
def save_to_memory(content: str, metadata: dict = None):
    doc = Document(
        page_content=content,
        metadata={
            "id": str(uuid.uuid4()),
            "timestamp": datetime.now().isoformat(),
            **(metadata or {})
        }
    )
    vector_store.add_documents([doc])

# ── 检索记忆：语义相似性搜索 ─────────────────────────────
def recall_memories(query: str, k: int = 3) -> str:
    results = vector_store.similarity_search_with_score(query, k=k)
    if not results:
        return "没有找到相关历史记录。"

    memories = []
    for doc, score in results:
        if score < 0.8:  # 过滤低相关性结果
            memories.append(
                f"[{doc.metadata.get('timestamp', '未知时间')[:10]}] "
                f"{doc.page_content}"
            )
    return "\n".join(memories) if memories else "无高相关历史记录。"

# ── 在 Agent 节点中使用长期记忆 ──────────────────────────
@tool
def search_my_memory(query: str) -> str:
    """搜索用户的历史对话和偏好记录。适用于需要参考过去信息的问题。"""
    return recall_memories(query)

def memory_write_node(state):
    """在 Agent 完成任务后，将重要信息写入长期记忆。"""
    last_exchange = state["messages"][-2:]  # 最后一轮对话
    summary = summarizer.invoke({
        "existing_summary": "",
        "new_messages": str(last_exchange)
    })
    save_to_memory(summary.content, {"type": "conversation"})
    return {}

LangGraph 检查点：跨会话持久化

LangGraph 内置的检查点机制是最简洁的短期记忆持久化方案，无需手动管理消息列表：

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
import sqlite3

# 开发/测试：SQLite 检查点
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

# 生产：PostgreSQL 检查点
# checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)

graph = graph_builder.compile(checkpointer=checkpointer)

# 同一 thread_id = 自动加载历史，继续上次会话
config = {"configurable": {"thread_id": "user_alice_001"}}

# 第一次对话
graph.invoke({"messages": [HumanMessage(content="我喜欢 Python 编程")]}, config)

# 应用重启后，第二次对话仍能记住
graph.invoke({"messages": [HumanMessage(content="推荐我一个适合的框架")]}, config)
# Agent 会记住"我喜欢 Python"，推荐 LangChain/FastAPI 等

# 查看检查点历史（时间旅行）
for checkpoint in graph.get_state_history(config):
    print(checkpoint.metadata["step"], len(checkpoint.values["messages"]))

记忆管理策略对比