智能体架构 —— 情景记忆与语义记忆栈，让智能体真正“长脑子”

你是否有过这样的经历？和一个AI助手聊了好几次，它还是记不住你的名字，更别提你反复强调的“我是保守型投资者，偏好成熟科技股”。每次对话都像第一次见面，翻来覆去问你同样的问题，感觉像在对着一只“金鱼”聊天——记忆只有几秒。

市面上大多数智能体（Agent）确实如此。它们的记忆是短暂的，只存在于当前会话中。一旦窗口关闭，之前聊过的内容、用户偏好、积累的知识都烟消云散。但真正的智能体，应该能随着与用户的互动不断学习和成长，成为你的长期伙伴。

今天，我们就要解决这个痛点。我们将实现一个结构化记忆架构，它模仿人类认知，结合了情景记忆和语义记忆。简单来说，智能体既能记住“过去聊过什么”，又能从这些对话中提取出“我了解用户什么”。这样，无论多久后再对话，它都能回忆起你的喜好，给出真正个性化的答案。

废话不多说，直接上干货。

什么是“双栈记忆”？

在正式动手前，我们先搞清楚这两个概念。

• 情景记忆：就像你写日记，记录下某个时间、地点发生的事件。它回答“发生了什么？”（例如，“上周，用户问我英伟达股价，并提到自己是保守型投资者。”）。我们会用向量数据库来存储，以便后续根据语义相似性快速找到相关过往对话。
• 语义记忆：就像你从日记中提炼出的知识卡片——张三喜欢科技股，李四对新能源感兴趣。它回答“我知道什么？”（例如，“用户Alex是保守型投资者”“Alex对科技股感兴趣”）。这些信息之间存在关联，因此我们用图数据库（如Neo4j）来管理，因为它擅长处理复杂的“实体-关系”查询。

两者结合，智能体既能回顾往事，又能构建一个关于用户和世界的、丰富互联的知识库。这种“双栈记忆”是实现深度个性化和上下文感知交互的关键。

架构工作流：从“过目就忘”到“过目不忘”

我们设计的智能体，在每次对话中会经历三个核心阶段：

1. 回忆：收到用户消息后，同时查询两个记忆系统——在情景向量库中找相似的过往对话，在语义图库中找与当前话题相关的事实。
2. 生成：将检索到的记忆拼接到提示词中，让大模型生成一个“知情”的、个性化的回复。
3. 编码：对话结束后，后台进程分析本次对话，生成摘要（新的情景记忆），并提取出关键实体和关系（新的语义记忆），分别存入库中。

这种设计让智能体能够无限期地积累用户知识，而不会被上下文窗口的长度所限制。

动手实现：搭建“双栈记忆”智能体

我们一步步来，先安装依赖、配置环境，再构建记忆组件，最后组装成智能体。

阶段0：基础与设置

首先，安装必要的库，包括向量库（FAISS）、图数据库驱动（neo4j）以及LangChain生态的相关组件。

# !pip install -q -U langchain-nebius langchain langgraph rich python-dotenv langchain_community langchain-openai neo4j faiss-cpu tiktoken

接着，导入模块并加载API密钥。这里我们使用Nebius提供的LLM和Embedding服务，并配置LangSmith追踪。注意，图数据库Neo4j需要单独安装并运行（本地或云服务），我们通过环境变量配置连接信息。

import os
import uuid
from typing import List, Dict, Any, Optional, Tuple
from dotenv import load_dotenv
# Pydantic 用于数据建模
from pydantic import BaseModel, Field
# LangChain 组件
from langchain_nebius import ChatNebius, NebiusEmbeddings
from langchain_community.graphs import Neo4jGraph
from langchain_community.vectorstores import FAISS
from langchain.docstore.document import Document
from langchain_core.prompts import ChatPromptTemplate
# LangGraph 组件
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
# 用于美化打印
from rich.console import Console
from rich.markdown import Markdown
# --- API 密钥和追踪设置 ---
load_dotenv()
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Agentic Architecture - Memory Stack (Nebius)"
# 检查所需的环境变量
required_vars = ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "NEO4J_URI", "NEO4J_USERNAME", "NEO4J_PASSWORD"]
for var in required_vars:
if var notin os.environ:
        print(f"警告：环境变量 {var} 未设置。")
print("环境变量已加载，追踪已设置。")

阶段1：构建记忆组件

这是核心部分。我们将初始化向量存储和图数据库，并定义“记忆制造者”——一个专门负责从对话中生成摘要和提取结构化知识的智能体。

console = Console()
llm = ChatNebius(model="mistralai/Mixtral-8x22B-Instruct-v0.1", temperature=0)
embeddings = NebiusEmbeddings()
# --- 1. 用于情景记忆的向量存储 ---
# 实际项目中应该持久化（如保存到磁盘），这里我们仅做演示，存储在内存中。
try:
    episodic_vector_store = FAISS.from_texts(["用于启动存储的初始文档"], embeddings)
except ImportError:
    console.print("[bold red]未安装 FAISS。请运行 `pip install faiss-cpu`。[/bold red]")
    episodic_vector_store = None
# --- 2. 用于语义记忆的图数据库 ---
try:
    graph = Neo4jGraph(
        url=os.environ.get("NEO4J_URI"),
        username=os.environ.get("NEO4J_USERNAME"),
        password=os.environ.get("NEO4J_PASSWORD")
    )
# 为了演示干净，清空数据库（生产环境请勿轻易使用！）
    graph.query("MATCH (n) DETACH DELETE n")
except Exception as e:
    console.print(f"[bold red]连接到 Neo4j 失败：{e}。请检查您的凭据和连接。[/bold red]")
    graph = None
# --- 3. “记忆制造者”的 Pydantic 模型 ---
# 定义我们想要提取的知识的结构。
classNode(BaseModel):
    id: str = Field(description="节点的唯一标识符，可以是人名、公司代码或概念。")
    type: str = Field(description="节点的类型（例如，'用户'、'公司'、'投资理念'）。")
    properties: Dict[str, Any] = Field(description="节点的属性字典。")
classRelationship(BaseModel):
    source: Node = Field(description="关系的源节点。")
    target: Node = Field(description="关系的目标节点。")
    type: str = Field(description="关系的类型（例如，'是'、'对...感兴趣'）。")
    properties: Dict[str, Any] = Field(description="关系的属性字典。")
classKnowledgeGraph(BaseModel):
"""表示从对话中提取的结构化知识。"""
    relationships: List[Relationship] = Field(description="要添加到知识图谱的关系列表。")
# --- 4. “记忆制造者”智能体 ---
defcreate_memories(user_input: str, assistant_output: str):
    conversation = f"用户：{user_input}\n助手：{assistant_output}"

# 4a. 创建情景记忆（摘要）
    console.print("--- 创建情景记忆（摘要） ---")
    summary_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一名摘要专家。为以下的用户-助手互动创建一个简洁的单句摘要。此摘要将作为记忆用于未来的回忆。"),
        ("human", "互动：\n{interaction}")
    ])
    summarizer = summary_prompt | llm
    episodic_summary = summarizer.invoke({"interaction": conversation}).content

    new_doc = Document(page_content=episodic_summary, metadata={"created_at": uuid.uuid4().hex})
    episodic_vector_store.add_documents([new_doc])
    console.print(f"[green]情景记忆已创建：[/green]'{episodic_summary}'")

# 4b. 创建语义记忆（事实提取）
    console.print("--- 创建语义记忆（图） ---")
    extraction_llm = llm.with_structured_output(KnowledgeGraph)
    extraction_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一名知识提取专家。你的任务是从对话中识别关键实体及其关系，并将其建模为图。重点关注用户的偏好、目标和陈述的事实。"),
        ("human", "从以下互动中提取所有关系：\n{interaction}")
    ])
    extractor = extraction_prompt | extraction_llm
try:
        kg_data = extractor.invoke({"interaction": conversation})
if kg_data.relationships:
for rel in kg_data.relationships:
                graph.add_graph_documents([rel], include_source=True)
            console.print(f"[green]语义记忆已创建：[/green]向图数据库添加了 {len(kg_data.relationships)} 个关系。")
else:
            console.print("[yellow]本次互动中未识别出新的语义记忆。[/yellow]")
except Exception as e:
        console.print(f"[red]无法提取或保存语义记忆：{e}[/red]")
if episodic_vector_store and graph:
    print("记忆组件初始化成功。")

阶段2：构建记忆增强型智能体

现在，我们将使用LangGraph构建一个完整的对话流程，包含检索→生成→更新三个步骤。

# 为我们的 LangGraph 智能体定义状态
classAgentState(TypedDict):
    user_input: str
    retrieved_memories: Optional[str]
    generation: str
# 定义图的节点
defretrieve_memory(state: AgentState) -> Dict[str, Any]:
"""从情景记忆和语义记忆中检索记忆的节点。"""
    console.print("--- 检索记忆 ---")
    user_input = state['user_input']

# 从情景记忆中检索
    retrieved_docs = episodic_vector_store.similarity_search(user_input, k=2)
    episodic_memories = "\n".join([doc.page_content for doc in retrieved_docs])

# 从语义记忆中检索（简单示例，实际应先用实体抽取）
try:
# 使用全文索引检索相关节点及其关系
        semantic_memories = str(graph.query("""
            UNWIND $keywords AS keyword
            CALL db.index.fulltext.queryNodes("entity", keyword) YIELD node, score
            MATCH (node)-[r]-(related_node)
            RETURN node, r, related_node LIMIT 5
            """, {'keywords': user_input.split()}))
except Exception as e:
        semantic_memories = f"无法查询图数据库：{e}"

    retrieved_content = f"相关过往对话（情景记忆）：\n{episodic_memories}\n\n相关事实（语义记忆）：\n{semantic_memories}"
    console.print(f"[cyan]检索到的上下文：\n{retrieved_content}[/cyan]")

return {"retrieved_memories": retrieved_content}
defgenerate_response(state: AgentState) -> Dict[str, Any]:
"""使用检索到的记忆生成响应的节点。"""
    console.print("--- 生成响应 ---")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个乐于助人且个性化的金融助手。使用检索到的记忆来为你的响应提供信息，并使其贴合用户。如果记忆表明用户的偏好（例如，他们是保守型投资者），你必须尊重这一点。"),
        ("human", "我的问题是：{user_input}\n\n以下是一些可能相关的记忆：\n{retrieved_memories}")
    ])
    generator = prompt | llm
    generation = generator.invoke(state).content
    console.print(f"[green]生成的响应：\n{generation}[/green]")
return {"generation": generation}
defupdate_memory(state: AgentState) -> Dict[str, Any]:
"""用最新的互动更新记忆的节点。"""
    console.print("--- 更新记忆 ---")
    create_memories(state['user_input'], state['generation'])
return {}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_memory)
workflow.add_node("generate", generate_response)
workflow.add_node("update", update_memory)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "update")
workflow.add_edge("update", END)
memory_agent = workflow.compile()
print("记忆增强型智能体图编译成功。")

阶段3：演示与检查

我们来模拟一个多轮对话。前两轮给智能体“喂”一些记忆，第三轮测试它是否能利用这些记忆提供个性化回答。

defrun_interaction(query: str):
    result = memory_agent.invoke({"user_input": query})
return result['generation']
console.print("\n--- 💬 互动 1：播种记忆 ---")
run_interaction("嗨，我叫Alex。我是一名保守型投资者，主要对成熟的科技公司感兴趣。")
console.print("\n--- 💬 互动 2：问一个具体问题 ---")
run_interaction("你觉得苹果公司（AAPL）怎么样？")
console.print("\n--- 🧠 互动 3：记忆测试 ---")
run_interaction("根据我的目标，那只股票有什么好的替代选择吗？")

运行后，你会看到智能体在第三轮回答时，提到了Alex的保守目标和过往讨论的苹果公司，并给出了微软、谷歌等更成熟的替代选择，并且解释了为什么这些更符合他的偏好。

检查记忆存储

我们可以直接查询内部存储，验证记忆是否真的被保存下来。

console.print("--- 🔍 检查情景记忆（向量存储）---")
retrieved_docs = episodic_vector_store.similarity_search("用户的投资策略", k=3)
for i, doc in enumerate(retrieved_docs):
    print(f"{i+1}. {doc.page_content}")
console.print("\n--- 🕸️ 检查语义记忆（图数据库）---")
print(f"图模式：\n{graph.get_schema}")
query_result = graph.query("MATCH (n:User)-[r:INTERESTED_IN|HAS_GOAL]->(m) RETURN n, r, m")
print(f"图中的关系：\n{query_result}")

你会发现，图中已经存在了类似 (User {id: "Alex"}) -[:HAS_GOAL]-> (Goal {value: "保守"}) 的关系，以及 (User) -[:INTERESTED_IN]-> (Company {name: "苹果公司"}) 等。这些结构化知识正是智能体能够“记住”用户的关键。

写在最后

通过这个实战，我们亲手搭建了一个拥有长期记忆的智能体。它不再是一只“金鱼”，而是一个能随着与你的互动不断成长、越来越懂你的伙伴。

核心回顾

1. 两种记忆：情景记忆记录“过去发生过什么”（用向量库），语义记忆提取“我知道什么”（用图库），双栈结合，全面存储知识。
2. 工作流程：回忆（检索）→ 生成（回答）→ 编码（更新记忆），形成一个持续学习的闭环。
3. 避坑指南：记忆会随着时间膨胀，需要设计策略进行总结、修剪或遗忘，避免存储爆炸；同时，在语义检索时，最好先抽取实体再查询，否则可能漏掉关键信息。

构建有记忆的智能体，是通往真正个性化、通用人工智能的必经之路。虽然管理大规模记忆会带来新的挑战（如一致性、隐私保护等），但我们今天打下的基础，已经让你有能力迈出这一步。

你在实际项目中，有没有遇到过需要智能体长期记忆的场景？比如个性化的客服、学习助手等。你会怎么设计记忆的存储和检索策略？欢迎在评论区分享你的想法！