大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



之前探索了LLM滑动窗口与记忆机制的应用探索

https://blog.csdn.net/liliang199/article/details/

其中，提到多种Memory形态，工作记忆 (Working Memory)、长期记忆 (Long-Term Memory)。

这里进一步在Agent系统中通过示例的方式探索这些记忆形态，所用示例参考和修改自网络资料。

在工业界，智能体面临三个核心矛盾：

1）实时性（响应延迟<100ms）

2）容量限制（LLM上下文窗口有限）

3）长期一致性（跨会话保留知识）。

分层记忆正是解决这些矛盾的标准范式。

1.1 分层记忆

以下从存储、生命周期、数据特征、容量等多个层面分析分层记忆。

1.2 关键设计模式

1）Write-through cache

短期记忆优点类似于Write-through cache。

短期记忆实时写如，但是需要定期压缩/摘要后迁入长期记忆

2）Retrieval-augmented generation (RAG)

长期以及优点类似于RAG，检索长期记忆后注入Prompt。

3）实体消歧

由于数据存在多个副本，所以需要使用关系库维护唯一实体ID，避免重复

1.3 核心数据流

以下是本智能体记忆系统的核心数据流。

用户输入

→ [感知] 接收原始消息 + 会话ID

→ [提取] 实体识别 (NER) + 语义摘要 (LLM/小模型)

→ [存储]

├─ 短期: Redis List LPUSH (保留最近20条)

├─ 工作: Redis Hash (更新任务状态)

├─ 实体: PostgreSQL INSERT ... ON CONFLICT UPDATE

└─ 长期: Chroma upsert (embedding + metadata)

→ [检索]

├─ 短期: 读取当前会话上下文

├─ 工作: 读取任务进度

├─ 实体: SQL 查询用户偏好/已知实体

└─ 长期: 向量相似性搜索 (top-k)

→ [注入] 构造增强Prompt: System + 检索记忆 + 对话历史 + 用户输入

→ [生成] LLM响应

这里实现一个完整的面向智能客服助理的记忆系统，支持：

1）跨会话记住用户姓名、偏好颜色

2）记住用户曾问过的问题（长期语义检索）

3）维护当前对话工作记忆（如未完成订单）

4）实体记忆去重更新​​​​​​​

各层记忆主要采用如下实现方式

1. 短期记忆：Redis存储最近10轮对话
2. 工作记忆：Redis存储当前会话的待处理任务
3. 长期记忆：Chroma向量库存储历史事实摘要
4. 实体记忆：SQLite存储用户结构化信息

2.1 工具安装

在实际运行程序前，需要安装记忆系统依赖的工具。

比如Redis、ChromaDB、SQLite、sentence-transformers。

1）chromadb

pip install chromadb

2）sentence-transformers

pip install sentence-transformers

3）安装redis

这里由于是mac，所以采用brew安装redis-server。

命令如下

brew install redis

brew 启动redis

brew services start redis

检查redis启动状态，示例如下，说明redis已经启动。

% ps aux | grep redis

user 56882 0.4 0.1 9356 ?? S 12:03下午 0:00.05 /usr/local/opt/redis/bin/redis-server 127.0.0.1:6379

user 56897 0.0 0.0 1600 s003 S+ 12:03下午 0:00.01 grep redis

其他工具安装类似，不再一一罗列。

2.2 代码示例

以下是智能客服助理Agent的记忆系统的代码示例和解读。

1）LLM配置

由于涉及到向量模型，需要从hf下载，这里配置hf-mirror国内站点。

同时配置openai的api key和base url，示例代码如下所示。

 
  
    
    
 
     
import os 
 
     
os.environ[‘HF_ENDPOINT’] = "https://hf-mirror.com"
 
     
import config
 
     
model_name = gpt_model_name # LLM名称,比如deepseek-r1, qwen3.5-8b os.environ[‘OPENAI_API_KEY’] = gpt_api_key # LLM供应商提供的api key os.environ[‘OPENAI_BASE_URL’] = gpt_api_url # LLM供应商提供llm访问api的url
 
     
import openai
 
     
client = openai.OpenAI()
 
    

 
2）Agent 存储配置
 

这里进一步配置Agent的存储系统，比如

redis，配置host、port、ttl等。

chromedb，配置persist_dir、collection name。

sentence-transformer，配置向量模型，这里对比测试后采用bce-embedding-base_v1

sql db，由于是示例系统，这里采用sqlite。

示例代码如下。

 
  
    
    
 
     
# ========== 1. 配置层 ========== 
 
     
class Config:
 
     
REDIS_HOST = "localhost" REDIS_PORT = 6379 REDIS_DB = 0 SESSION_TTL = 3600 # 会话记忆过期时间(秒) WORK_MEMORY_TTL = 1800 # 工作记忆过期时间 # 向量数据库 CHROMA_PERSIST_DIR = "./chroma_data_v3" COLLECTION_NAME = "long_term_memory_v3" # EMBEDDING_MODEL = "all-MiniLM-L6-v2" # 轻量级384维 EMBEDDING_MODEL = "maidalun1020/bce-embedding-base_v1" # 关系数据库 (生产用PostgreSQL, 演示用SQLite) SQLITE_PATH = "./entity_memory_v3.db" # LLM (示例使用mock, 可替换为真实OpenAI) USE_REAL_LLM = True # 设为True并配置OPENAI_API_KEY
 
    
 
3）Redis配置
 

这里构建redis连接，用于管理短期记忆、工作记忆的相关内容。

 
  
    
    
 
     
import redis 
 
     
========== 2. 存储客户端初始化 ==========
 
     
Redis
 
     
redis_client = redis.Redis(
 
     
host=Config.REDIS_HOST, port=Config.REDIS_PORT, db=Config.REDIS_DB, decode_responses=True 
 
     
)
 
    

 
4）向量模型
 

这里采用transformer-sentence向量工具，示例代码如下所示。

采用的向量模型为maidalun1020/bce-embedding-base_v1

 
  
    
    
 
     
import chromadb 
 
     
from chromadb.config import Settings from sentence_transformers import SentenceTransformer
 
     
向量模型 & Chroma
 
     
emb_model = Config.EMBEDDING_MODEL print(emb_model) embedder = SentenceTransformer(emb_model) chroma_client = chromadb.PersistentClient(path=Config.CHROMA_PERSIST_DIR) long_term_collection = chroma_client.get_or_create_collection(
 
     
name=Config.COLLECTION_NAME, metadata={"hnsw:space": "cosine"} 
 
     
)
 
    

 
以下是对向量模型和chromadb系统的一个简单测试。
 
  
    
    
 
     
collection_cosine = chroma_client.get_or_create_collection( name="test_v2", metadata={"hnsw:space": "cosine"} 
 
     
) content_list = ["记住我每周三下午有空接电话", "你好", "购买", "体验不好"] embedding_array = [embedder.encode(content).tolist() for content in content_list]
 
     
添加一些示例向量
 
     
collection_cosine.add(
 
     
ids=[f"id_{i}" for i in range(len(embedding_array))], embeddings=embedding_array 
 
     
)
 
     
query = "什么是否接电话方便" query = "周三下午有空接电话" query_embedding = embedder.encode(query).tolist()
 
     
查询并返回距离分数
 
     
results = collection_cosine.query(
 
     
query_embeddings=[query_embedding], n_results=4, include=["distances"] # 返回距离分数 
 
     
) print(results)
 
    

 
输出如下，可见语义越接近，向量之间的距离越近。
 

{‘ids’: [[‘id_0’, ‘id_2’, ‘id_1’, ‘id_3’]], ‘embeddings’: None, ‘documents’: None, ‘uris’: None, ‘included’: [‘distances’], ‘data’: None, ‘metadatas’: None, ‘distances’: [[0., 0.87817, 0.79089, 0.79089]]}

这与chromadb的向量距离的定义一致，具体参考如下链接。

https://blog.csdn.net/liliang199/article/details/

5）sqlite配置

这里配置sqlite，用于agent系统的实体记忆的相关内容。

 
  
    
    
 
     
from datetime import datetime 
 
     
import sqlalchemy as sa from sqlalchemy import create_engine, Column, String, Integer, DateTime, Float from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker
 
     
关系库 (实体记忆)
 
     
engine = create_engine(f"sqlite:///{Config.SQLITE_PATH}", echo=False) Base = declarative_base()
 
     
class UserEntity(Base):
 
     
__tablename__ = "user_entities" id = Column(String, primary_key=True) name = Column(String, default=None) preferred_color = Column(String, default=None) phone = Column(String, default=None) last_active = Column(DateTime, default=datetime.utcnow) total_interactions = Column(Integer, default=0) # database default 
 
     
class EntityRelation(Base):
 
     
__tablename__ = "entity_relations" id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) subject_id = Column(String) # 用户ID predicate = Column(String) # 关系类型，如 "likes", "purchased" object_value = Column(String) # 对象值或ID created_at = Column(DateTime, default=datetime.utcnow) 
 
     
Base.metadata.create_all(engine) SessionLocal = sessionmaker(bind=engine)
 
    

 
6）Agent Memory
 

这里基于以上redis、chromadb、sqlite，构建Agent Memory系统。

包括短期以及，工作记忆、长期记忆、以及实体记忆的数据管理的实现。

 
  
    
    
 
     
""" 
 
     
Agent Memory 工业级实现 - 智能客服助理 特性：
 
     
 
      
短期记忆：Redis存储最近10轮对话
 
      
工作记忆：Redis存储当前会话的待处理任务
 
      
长期记忆：Chroma向量库存储历史事实摘要
 
      
实体记忆：SQLite存储用户结构化信息 """
 
     
 
     
import json import hashlib import uuid from typing import List, Dict, Any, Optional
 
     
========== 3. 记忆管理核心类 ==========
 
     
class AgentMemory:
 
     
def __init__(self, session_id: str, user_id: Optional[str] = None): self.session_id = session_id self.user_id = user_id or f"anon_{session_id[:8]}" # ---------- 短期记忆 (对话流水) ---------- def add_short_term(self, role: str, content: str): """存储单条对话记录到Redis列表，保留最近20条""" key = f"short_term:{self.session_id}" entry = json.dumps({ "role": role, "content": content, "timestamp": datetime.utcnow().isoformat() }) redis_client.lpush(key, entry) redis_client.ltrim(key, 0, 19) # 只保留20条 redis_client.expire(key, Config.SESSION_TTL) def get_short_term(self, n: int = 10) -> List[Dict]: """获取最近n条对话历史""" key = f"short_term:{self.session_id}" entries = redis_client.lrange(key, 0, n-1) return [json.loads(e) for e in entries][::-1] # 时间正序 # ---------- 工作记忆 (当前任务状态) ---------- def set_work_memory(self, key: str, value: Any): """存储工作记忆，例如当前正在处理的订单号""" redis_key = f"work:{self.session_id}:{key}" redis_client.setex(redis_key, Config.WORK_MEMORY_TTL, json.dumps(value)) def get_work_memory(self, key: str) -> Optional[Any]: redis_key = f"work:{self.session_id}:{key}" val = redis_client.get(redis_key) return json.loads(val) if val else None def clear_work_memory(self): """清空当前会话的所有工作记忆""" keys = redis_client.keys(f"work:{self.session_id}:*") if keys: redis_client.delete(*keys) # ---------- 长期记忆 (语义向量库) ---------- def add_long_term(self, content: str, category: str = "fact", importance: float = 0.4): """将重要信息嵌入并存入向量库""" # 生成唯一ID (基于内容哈希) doc_id = hashlib.md5(f"{self.user_id}:{content}".encode()).hexdigest() embedding = embedder.encode(content).tolist() metadata = { "user_id": self.user_id, "category": category, "importance": importance, "timestamp": datetime.utcnow().isoformat() } print("add content {content} to long term memory!") long_term_collection.upsert( ids=[doc_id], embeddings=[embedding], metadatas=[metadata], documents=[content] ) return doc_id def retrieve_long_term(self, query: str, top_k: int = 5, min_relevance: float = 0.5) -> List[Dict]: """语义检索长期记忆""" query_embedding = embedder.encode(query).tolist() results = long_term_collection.query( query_embeddings=[query_embedding], n_results=top_k, where={"user_id": self.user_id} # 仅检索当前用户 ) docs = results.get("documents", [[]])[0] metas = results.get("metadatas", [[]])[0] distances = results.get("distances", [[]])[0] # 过滤低相关性 (cosine距离越小越相似，阈值0.3对应相似度~0.7) print(f"min_relevance: {min_relevance}") filtered = [] for doc, meta, dist in zip(docs, metas, distances): if dist <= min_relevance: filtered.append({"content": doc, "metadata": meta, "relevance": 1-dist}) return filtered # ---------- 实体记忆 (结构化关系库) ---------- # def upsert_user_entity(self, kwargs): # """更新用户实体信息 (如姓名、偏好颜色)""" # session = SessionLocal() # entity = session.query(UserEntity).filter_by(id=self.user_id).first() # if not entity: # entity = UserEntity(id=self.user_id) # session.add(entity) # for key, value in kwargs.items(): # if hasattr(entity, key) and value is not None: # setattr(entity, key, value) # entity.last_active = datetime.utcnow() # entity.total_interactions += 1 # session.commit() # session.close() def upsert_user_entity(self, kwargs): """Update or create user entity with structured information.""" session = SessionLocal() entity = session.query(UserEntity).filter_by(id=self.user_id).first() if not entity: # Explicitly set total_interactions to 0 to avoid None entity = UserEntity(id=self.user_id, total_interactions=0) session.add(entity) # Update dynamic fields from kwargs (e.g., name, preferred_color) for key, value in kwargs.items(): if hasattr(entity, key) and value is not None: setattr(entity, key, value) # Update timestamps and interaction counter entity.last_active = datetime.utcnow() if entity.total_interactions is None: entity.total_interactions = 0 entity.total_interactions += 1 session.commit() session.close() def get_user_entity(self) -> Dict: """Retrieve user entity as a plain dictionary.""" session = SessionLocal() entity = session.query(UserEntity).filter_by(id=self.user_id).first() session.close() if entity: # Convert SQLAlchemy model to dictionary manually return return {} def add_entity_relation(self, predicate: str, object_value: str): """存储关系: user_id likes "red" / purchased "product_A" """ session = SessionLocal() rel = EntityRelation( subject_id=self.user_id, predicate=predicate, object_value=object_value ) session.add(rel) session.commit() session.close() def get_relations(self, predicate: Optional[str] = None) -> List[Dict]: session = SessionLocal() query = session.query(EntityRelation).filter_by(subject_id=self.user_id) if predicate: query = query.filter_by(predicate=predicate) results = query.all() session.close() return [{"predicate": r.predicate, "object": r.object_value} for r in results] # ---------- 综合检索：聚合所有记忆层 ---------- def retrieve_all_memories(self, user_query: str) -> Dict: """为生成响应检索所有相关记忆""" return , "long_term": self.retrieve_long_term(user_query, top_k=3), "user_entity": self.get_user_entity(), "user_relations": self.get_relations() }
 
    
 
7）核心逻辑
 

这里基于AgentMemory，构建Agent智能体的核心逻辑，包括llm调用、各种记忆的协同。

 
  
    
    
 
     
# ========== 4. 智能体核心逻辑 ========== 
 
     
class CustomerServiceAgent:
 
     
def __init__(self, session_id: str, user_id: Optional[str] = None): self.memory = AgentMemory(session_id, user_id) def _call_llm_for_extraction(self, user_message: str) -> Dict: """ Use OpenAI to extract structured entities from user message. Returns a dict with keys: name, preferred_color, order_id, complaint, etc. """ if not Config.USE_REAL_LLM: return {} # fallback to rule-based # openai.api_key = Config.OPENAI_API_KEY prompt = f""" 
 
     
Extract the following entities from the user’s message. Return ONLY a JSON object. Possible entities: name (person’s name), preferred_color (color they like), order_id (order number), complaint (boolean, true if they complain), fact_to_remember (any statement they ask to remember). If an entity is not present, omit it or set to null.
 
     
User message: "{user_message}"
 
     
Example output: {{"name": "John", "preferred_color": "blue", "order_id": "ORD123", "complaint": false}} """
 
     
 try: response = client.chat.completions.create( model=model_name, messages=[{"role": "user", "content": prompt}], temperature=0.0, max_tokens=3000, extra_body={ "top_k": 20, "chat_template_kwargs": {"enable_thinking": False}, } ) print(response.choices[0].message) content = response.choices[0].message.content.strip() # Extract JSON from response (might have markdown) if "json" in content: content = content.split("json")[1].split("")[0] elif "" in content: content = content.split("")[1].split("")[0] extracted = json.loads(content) return extracted except Exception as e: print(f"LLM extraction failed: {e}, falling back to rule-based") return {} def _extract_entities_and_facts(self, user_message: str) -> Dict: """ Extract entities using LLM (preferred) or rule-based fallback. """ # Try LLM first if enabled if Config.USE_REAL_LLM: extracted = self._call_llm_for_extraction(user_message) if extracted: # Convert complaint boolean to string for consistency with original logic if extracted.get("complaint") is True: extracted["complaint"] = "true" return extracted # Fallback to original rule-based extraction (kept intact) extracted = {} # Extract name if "我是" in user_message: name = user_message.split("我是")[-1].split()[0] extracted["name"] = name elif "我叫" in user_message: name = user_message.split("我叫")[-1].split()[0] extracted["name"] = name # Extract color preference if "喜欢" in user_message and "颜色" in user_message: color = user_message.split("喜欢")[-1].split("颜色")[0].strip() extracted["preferred_color"] = color elif "偏爱" in user_message: color = user_message.split("偏爱")[-1].split()[0] extracted["preferred_color"] = color # Extract order ID if "订单" in user_message and "号" in user_message: import re order_match = re.search(r"订单(d+)", user_message) if order_match: extracted["order_id"] = order_match.group(1) return extracted def _update_memories(self, user_message: str, response: str): """根据交互更新各层记忆""" # 1. 存储对话到短期记忆 self.memory.add_short_term("user", user_message) self.memory.add_short_term("assistant", response) # 2. 提取实体并存入关系库 entities = self._extract_entities_and_facts(user_message) if entities.get("name"): self.memory.upsert_user_entity(name=entities["name"]) if entities.get("preferred_color"): self.memory.upsert_user_entity(preferred_color=entities["preferred_color"]) self.memory.add_entity_relation("likes_color", entities["preferred_color"]) # 3. 重要事实存入长期记忆 (例如用户抱怨、特定需求) if "投诉" in user_message or "不满意" in user_message: self.memory.add_long_term( f"用户表达不满: {user_message}", category="complaint", importance=0.9 ) elif "记住" in user_message: # 显式要求记住的内容 fact = user_message.replace("记住", "").strip() self.memory.add_long_term(fact, category="explicit", importance=0.8) # 4. 工作记忆更新: 如果有订单号，保存为当前任务 if entities.get("order_id"): self.memory.set_work_memory("current_order", entities["order_id"]) def _build_prompt_with_memories(self, user_input: str) -> str: """构建增强Prompt，注入检索到的记忆""" memories = self.memory.retrieve_all_memories(user_input) # print(f"user_input: {user_input}, memories: {memories}") prompt_parts = [] prompt_parts.append("你是一个智能客服助理，记住以下信息帮助用户：") # 用户实体 (姓名/偏好) entity = memories["user_entity"] if entity.get("name"): prompt_parts.append(f"用户姓名：{entity['name']}") if entity.get("preferred_color"): prompt_parts.append(f"用户偏好颜色：{entity['preferred_color']}") # 关系记忆 relations = memories["user_relations"] if relations: rel_str = ", ".join([f"{r['predicate']}:{r['object']}" for r in relations]) prompt_parts.append(f"用户历史关系：{rel_str}") # 长期记忆中的相关事实 long_term = memories["long_term"] if long_term: facts = " 
 
     
".join([f"- {item[‘content’]} (相关度{item[‘relevance’]:.2f})" for item in long_term])
 
     
 prompt_parts.append(f"历史记住的事实： 
 
     
{facts}")
 
     
 # 工作记忆 if memories["work_memory"].get("current_order"): prompt_parts.append(f"用户当前正在处理订单：{memories['work_memory']['current_order']}") # 短期对话历史 (最近3轮) history = memories["short_term"][-3:] if memories["short_term"] else [] if history: hist_str = " 
 
     
".join([f"{h[‘role’]}: {h[‘content’]}" for h in history])
 
     
 prompt_parts.append(f"最近对话： 
 
     
{hist_str}")
 
     
 prompt_parts.append(f"用户最新问题：{user_input}") prompt_parts.append("请基于以上信息，友好、准确地回答。") # print(prompt_parts) return " 
 
     
".join(prompt_parts)
 
     
def _call_llm(self, prompt: str) -> str: """调用LLM生成响应 (此处用模拟，可替换真实OpenAI)""" if Config.USE_REAL_LLM: # import openai # openai.api_key = Config.OPENAI_API_KEY response = client.chat.completions.create( model=model_name, messages=[{"role": "user", "content": prompt}], temperature=0.7 ) return response.choices[0].message.content else: # 模拟智能响应 (演示用) if "推荐" in prompt and "偏好颜色" in prompt: color = prompt.split("偏好颜色：")[-1].split(" 
 
     
")[0] if "偏好颜色" in prompt else "经典"
 
     
 return f"根据您喜欢的{color}颜色，我推荐我们的{color}系列产品，非常受欢迎。" elif "订单" in prompt: return "您的订单正在处理中，预计3天内送达。需要查询具体进度吗？" elif "姓名" in prompt: name = prompt.split("用户姓名：")[-1].split(" 
 
     
")[0] if "用户姓名" in prompt else "用户"
 
     
 return f"您好{name}，很高兴再次为您服务！" else: return "感谢您的咨询。请问还有其他可以帮助您的吗？" def chat(self, user_input: str) -> str: """主入口：感知→检索→注入→生成→更新记忆""" # 感知 & 检索 & 注入 enhanced_prompt = self._build_prompt_with_memories(user_input) # 生成响应 response = self._call_llm(enhanced_prompt) # 更新所有记忆层 self._update_memories(user_input, response) return response
 
    
 
8）运行示例
 

这里通过实际客服对话，验证Agent存储系统的工作逻辑。

示例代码如下

 
  
    
    
 
     
# ========== 5. 测试与演示 (工业级场景) ========== 
 
     
def run_demo():
 
     
""" 模拟真实用户跨会话交互，验证记忆系统功能 场景：用户第一次询问产品，留下姓名和偏好；第二次询问推荐；第三次查询订单；第四次跨会话重新连接 """ print("="*60) print("智能客服助理记忆系统演示") print("="*60) # 模拟用户ID (生产环境从登录态获取) user_id = "user_v3" # 第一次会话 print(" 
 
     
【第一次会话】")
 
     
session1 = CustomerServiceAgent(session_id="sess_v3_001", user_id=user_id) msg1 = "你好，我是李明，我喜欢蓝色，你们有蓝色产品吗？" print(f"用户: {msg1}") resp1 = session1.chat(msg1) print(f"客服: {resp1}") msg2 = "帮我推荐一款蓝色产品" print(f"用户: {msg2}") resp2 = session1.chat(msg2) print(f"客服: {resp2}") # 显示第一次会话后的记忆状态 print(" 
 
     
— 第一次会话结束，记忆状态 —")
 
     
mem = session1.memory print(f"用户实体: ") print(f"实体关系: ") long = mem.retrieve_long_term("蓝色产品", top_k=2) print(f"长期记忆检索'蓝色产品': {[l['content'] for l in long]}") work = mem.get_work_memory("current_order") print(f"工作记忆: {work}") # 第二次会话 (新session，但同一user_id) print(" 
 
     
【第二次会话 - 跨会话记忆恢复】")
 
     
session2 = CustomerServiceAgent(session_id="sess_v3_002", user_id=user_id) msg3 = "我之前问过蓝色产品，现在想下单" print(f"用户: {msg3}") resp3 = session2.chat(msg3) print(f"客服: {resp3}") msg4 = "我的订单号是ORD12345，帮我查一下" print(f"用户: {msg4}") resp4 = session2.chat(msg4) print(f"客服: {resp4}") # 验证工作记忆是否保存了订单号 print(" 
 
     
— 第二次会话后工作记忆 —")
 
     
print(f"当前订单: ") # 第三次会话：显示长期记忆中的投诉处理 print(" 
 
     
【第三次会话 - 投诉记忆】")
 
     
session3 = CustomerServiceAgent(session_id="sess_v3_003", user_id=user_id) msg5 = "我上次投诉过产品质量，你们有改进吗？" print(f"用户: {msg5}") resp5 = session3.chat(msg5) print(f"客服: {resp5}") # 显式要求记住事实 msg6 = "记住我每周三下午有空接电话" print(f"用户: {msg6}") resp6 = session3.chat(msg6) print(f"客服: {resp6}") # 验证长期记忆检索 print(" 
 
     
— 检索’接电话时间’ —")
 
     
retrieved = session3.memory.retrieve_long_term("用户什么时间方便接电话") for item in retrieved: print(f" -> {item['content']} (相关度{item['relevance']:.2f})") print(" 
 
     
" + "="*60)
 
     
print("演示完成。记忆系统成功保留实体、关系、长期事实和任务状态。") print("="*60) 
 
     
if name == "main":
 
     
# 确保Redis服务已启动 (若未启动则跳过Redis部分，示例会报错，可改为Mock) try: redis_client.ping() except redis.ConnectionError: print("警告: Redis未运行，将使用内存字典模拟 (生产环境请启动Redis)") # 此处可替换为简单的dict模拟，为简洁不展开 pass run_demo()
 
    
 
输出如下所示。
 

============================================================

智能客服助理记忆系统演示

============================================================

【第一次会话】

用户: 你好，我是李明，我喜欢蓝色，你们有蓝色产品吗？

min_relevance: 0.5

ChatCompletionMessage(content=‘{"name": "李明", "preferred_color": "蓝色", "complaint": false}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

客服:

您好，李明先生！很高兴为您服务。

我已经记住了您对蓝色的偏好。我们有很多产品都提供蓝色选项，这也是非常经典且受欢迎的颜色。

为了给您更精准的推荐，请问您具体想了解哪一类的产品呢？比如是服装、数码配件还是家居用品？一旦您告知具体类别，我可以为您详细介绍现有的蓝色款式。

期待您的回复，祝您生活愉快！

用户: 帮我推荐一款蓝色产品

min_relevance: 0.5

ChatCompletionMessage(content=‘{"preferred_color": "blue"}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

客服:

您好，李明先生！没问题，既然您想直接看看产品，那我就结合您对蓝色的偏好，为您挑选几款不同类别中口碑极佳的热门好物供您参考：

1. 数码类：深海蓝无线降噪耳机

* 特点：配色沉稳高级，音质清晰，是目前非常受欢迎的爆款。

2. 服饰类：天空蓝休闲衬衫

* 特点：面料舒适透气，颜色清新自然，非常适合日常穿着。

3. 家居类：靛蓝色陶瓷保温杯

* 特点：设计简约大方，手感极佳，实用又美观。

您可以看看这几款中是否有您心仪的？如果您更倾向于某一个特定类别（比如只想看数码或服装），也欢迎随时告诉我，我再为您做更详细的介绍和筛选！

期待您的反馈，祝您购物愉快！

— 第一次会话结束，记忆状态 —

用户实体: {‘id’: ‘user_v3’, ‘name’: ‘李明’, ‘preferred_color’: ‘blue’, ‘phone’: None, ‘last_active’: datetime.datetime(2026, 4, 8, 8, 52, 0, 99649), ‘total_interactions’: 3}

实体关系: [{‘predicate’: ‘likes_color’, ‘object’: ‘蓝色’}, {‘predicate’: ‘likes_color’, ‘object’: ‘blue’}]

min_relevance: 0.5

长期记忆检索’蓝色产品’: []

工作记忆: None

【第二次会话 - 跨会话记忆恢复】

用户: 我之前问过蓝色产品，现在想下单

min_relevance: 0.5

ChatCompletionMessage(content=‘{"preferred_color": "blue", "complaint": false}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

客服:

您好，李明先生！很高兴再次为您服务。😊

我注意到您之前咨询过蓝色的产品，而且了解到您一直对蓝色情有独钟。蓝色确实是个很经典且舒适的选择！

没问题，我现在就可以协助您完成下单。请问您具体想订购哪一款蓝色产品呢？是之前看过的那款，还是有新的心仪选择？确认好后，我会立刻帮您办理后续流程。

随时待命，期待您的回复！

用户: 我的订单号是ORD12345，帮我查一下

min_relevance: 0.5

ChatCompletionMessage(content=‘{"order_id": "ORD12345"}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

客服:

您好，李明先生！👋

收到，订单号 ORD12345 已记录，我立刻为您查询最新状态。🔍

看来您已经成功下单了！结合您之前对蓝色产品的青睐，我会特别帮您核对一下订单中的商品颜色是否符合您的偏好。

请问您主要想确认的是物流进度，还是订单的具体商品明细呢？请告诉我，我马上为您提供详细信息！

随时为您服务！😊

— 第二次会话后工作记忆 —

当前订单: ORD12345

【第三次会话 - 投诉记忆】

用户: 我上次投诉过产品质量，你们有改进吗？

min_relevance: 0.5

ChatCompletionMessage(content=‘{"complaint": true}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

add content {content} to long term memory!

客服:

李明先生，您好！我是您的智能客服助理。

非常抱歉之前的产品问题给您带来了不便，我完全理解您对产品质量的担忧。我知道您一直钟意我们的蓝色系产品，是咱们的忠实用户，正因为如此，我们更希望您在享受喜爱的颜色时，也能拥有无可挑剔的质量体验。

关于您关心的"是否有改进"这一问题，我们非常重视每一位用户的投诉反馈，并会据此推动产品优化。但由于具体的改进措施可能涉及不同的生产批次和整改方案，为了确保给您的信息准确无误，我需要为您查询具体的投诉记录。

请问您方便提供一下之前的订单号或投诉单号吗？或者我可以立即为您转接专属人工客服，让他们为您详细说明针对该问题的具体改进情况和后续保障。

感谢您的信任与反馈，我们一定会给您一个满意的答复！

用户: 记住我每周三下午有空接电话

min_relevance: 0.5

ChatCompletionMessage(content=‘{"fact_to_remember": "我每周三下午有空接电话"}’, refusal=None, role=‘assistant’, annotations=None, audio=None, function_call=None, tool_calls=[], reasoning=None)

add content {content} to long term memory!

客服:

李明先生，您好！

没问题，我已经为您特别备注好了：每周三下午是您方便接听电话的时间段。

这样后续关于您反馈的产品质量问题的跟进处理，或者咱们蓝色系产品的相关服务通知，我们都会优先安排在这个时间段联系您，确保既能及时为您解决问题，又不会打扰到您的日常工作。

请您放心，之前提到的质量改进查询我正在为您协调中。一旦有确切消息，我们会严格按照您的时间偏好与您沟通。感谢您的信任与配合，祝您本周愉快！

— 检索’接电话时间’ —

min_relevance: 0.5

-> 我每周三下午有空接电话 (相关度0.58)

============================================================

演示完成。记忆系统成功保留实体、关系、长期事实和任务状态。

============================================================

Agent有效存储上下文信息到到短期、工作、长期以及实体记忆，较好应用到与用户的对话中。

---

LLM滑动窗口与记忆机制的应用探索

https://blog.csdn.net/liliang199/article/details/

ChromaDB距离计算公式示例

https://blog.csdn.net/liliang199/article/details/

redis-server conda install

https://anaconda.org/channels/anaconda/packages/redis-server/overview

mac中dyld[5999]: Library not loaded: libssl.3.dylib解决方法

https://blog.csdn.net/weixin_/article/details/

BCEmbedding: Bilingual and Crosslingual Embedding for RAG

https://hf-mirror.com/maidalun1020/bce-embedding-base_v1