---
title: "智能体架构 —— 情景记忆与语义记忆栈，让智能体真正“长脑子”"
source: wechat
url: https://mp.weixin.qq.com/s/eyhtnbnN-3ocGKvslf2u8A
ingest_date: 2026-07-04
vxc: 64
stars: 4
sha256: 6e79d07c0da5395a3136358576813959f1baa4f92325d445a246fbee227c4123
---

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

**来源**: 数据STUDIO

**发布日期**: 2026-04-02

**原文链接**: https://mp.weixin.qq.com/s/eyhtnbnN-3ocGKvslf2u8A

---

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

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

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

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

## 什么是“双栈记忆”？

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

- 情景记忆
  ：就像你写日记，记录下某个时间、地点发生的事件。它回答“发生了什么？”（例如，“上周，用户问我英伟达股价，并提到自己是保守型投资者。”）。我们会用
  向量数据库
  来存储，以便后续根据语义相似性快速找到相关过往对话。

- 语义记忆
  ：就像你从日记中提炼出的知识卡片——张三喜欢科技股，李四对新能源感兴趣。它回答“我知道什么？”（例如，“用户Alex是保守型投资者”“Alex对科技股感兴趣”）。这些信息之间存在关联，因此我们用
  图数据库
  （如Neo4j）来管理，因为它擅长处理复杂的“实体-关系”查询。

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

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

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

- 回忆
   ：收到用户消息后，同时查询两个记忆系统——在情景向量库中找相似的过往对话，在语义图库中找与当前话题相关的事实。

- 生成
   ：将检索到的记忆拼接到提示词中，让大模型生成一个“知情”的、个性化的回复。

- 编码
   ：对话结束后，后台进程分析本次对话，生成摘要（新的情景记忆），并提取出关键实体和关系（新的语义记忆），分别存入库中。

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

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

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

### 阶段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 not in 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 模型 ---  
# 定义我们想要提取的知识的结构。  
class Node(BaseModel):  
    id: str = Field(description="节点的唯一标识符，可以是人名、公司代码或概念。")  
    type: str = Field(description="节点的类型（例如，'用户'、'公司'、'投资理念'）。")  
    properties: Dict[str, Any] = Field(description="节点的属性字典。")  

class Relationship(BaseModel):  
    source: Node = Field(description="关系的源节点。")  
    target: Node = Field(description="关系的目标节点。")  
    type: str = Field(description="关系的类型（例如，'是'、'对...感兴趣'）。")  
    properties: Dict[str, Any] = Field(description="关系的属性字典。")  

class KnowledgeGraph(BaseModel):  
    """表示从对话中提取的结构化知识。"""  
    relationships: List[Relationship] = Field(description="要添加到知识图谱的关系列表。")  

# --- 4. “记忆制造者”智能体 ---  
def create_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 智能体定义状态  
class AgentState(TypedDict):  
    user_input: str  
    retrieved_memories: Optional[str]  
    generation: str  

# 定义图的节点  

def retrieve_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}  

def generate_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}  

def update_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：演示与检查

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

def run_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: "苹果公司"})  等。这些结构化知识正是智能体能够“记住”用户的关键。

## 写在最后

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

核心回顾

- 两种记忆
   ：情景记忆记录“过去发生过什么”（用向量库），语义记忆提取“我知道什么”（用图库），双栈结合，全面存储知识。

- 工作流程
   ：回忆（检索）→ 生成（回答）→ 编码（更新记忆），形成一个持续学习的闭环。

- 避坑指南
   ：记忆会随着时间膨胀，需要设计策略进行总结、修剪或遗忘，避免存储爆炸；同时，在语义检索时，最好先抽取实体再查询，否则可能漏掉关键信息。

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

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

🏴‍☠️宝藏级🏴‍☠️ 原创公众号『 数据STUDIO 』内容超级硬核。公众号以Python为核心语言，垂直于数据科学领域，包括 可戳 👉 Python ｜ MySQL ｜ 数据分析 ｜ 数据可视化 ｜ 机器学习与数据挖掘 ｜ 爬虫 等，从入门到进阶！

长按👇关注- 数据STUDIO -设为星标，干货速递