多轮对话长上下文-向量检索和混合召回示例-编程阁

在处理多轮对话的上下文管理时，理论往往很美，但工程落地全是坑。

之前探索了多轮对话长上下文增量摘要和结构化摘要示

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

这里进一步探索向量检索和混合召回，所用示例参考和修改自网络资料。

1 检索和找回

向量检索不要只关注到对原文的摘要，更细粒度的相关历史原文对完善对话更重要。

这里通过向量检索和混合召回，尝试示例如何还原这些更细粒度的信息。

1.1 向量检索

这里向量数据库相当于上下文回收站。指将历史对话存入向量数据库。

即使截断了旧消息，也要存储在向量数据库中。

这里的向量检索，不是指经典RAG式的纯向量检索，而是指指代词触发检索。

当用户提到"上次"、"之前"等高熵指代词时，自动检索相关历史原文并融合到当前上下文中。

场景示例如下

场景：用户说：“帮我改成刚才说的那个地址。”
处理：识别到“刚才”这类高熵指代词时，立即去向量库搜[刚才, 地址]，把原文找回来临时拼接到 Context 里。

1.2 混合召回

这里指摘要+原始片段的混合召回。

1）不要只给摘要，不给证据

给模型的最终 Prompt 结构建议示例如下

[System Prompt]
[长期记忆摘要：用户叫张三，想买红毛衣...]
[相关历史原文片段：用户 5 分钟前说：'只要 XL 号，L 号有点紧']
[当前提问：还有货吗？]

2）实现路径

将未被摘要覆盖的、但被检索召回的历史对话原文作为 RAG 的 Context 插入。

这是因为摘要处理宏观叙事，原文片段处理微观指代，如尺寸、颜色代码。

在电商或更细粒度场景，微观尺度的细粒度信息更重要。

2 代码示例

2.1 场景说明

这是一个个人助理场景，示例如何将历史对话存入向量数据库。

当用户提到"上次"、"之前"等高熵指代词时，自动检索相关历史原文并融合到当前上下文中。

2.2 代码示例

个人助理场景中，向量检索+混合召回的实现代码示例如下。

1）环境配置

sentence-transformer从hf拉取向量模型，所以这里需要配置hf镜像HF_ENDPOINT。

这里所用大模型调用选用openai格式，需要配置api key和base url。

示例代码如下所示。

import os os.environ['HF_ENDPOINT'] = "https://hf-mirror.com" 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

2）向量检索&混合召回

向量检索&混合召回的具体说明瑞啊

"""
场景：个人助理 - 向量检索 + 混合召回
演示：
- 所有对话存入 ChromaDB 向量数据库
- 当用户问题包含"上次"、"之前"、"刚才"等指代词时，触发向量检索
- 检索到的相关历史原文片段与当前对话融合
- 摘要保留宏观信息，检索提供微观细节
- 使用 sentence-transformers/all-MiniLM-L6-v2 模型生成本地嵌入向量
- ChromaDB 集合自动适配嵌入维度（384维）
"""

这里选用all-MiniLM-L6-v2，因为内存占用仅80MB，适合轻量级部署。

若对中文效果有更高要求，可更换为paraphrase-multilingual-MiniLM-L12-v2。

向量检索&混合召回代码示例如下

import os import json import uuid from datetime import datetime from openai import OpenAI import chromadb from sentence_transformers import SentenceTransformer class PersonalAssistant: def __init__( self, api_key: str = None, model: str = "gpt-4o-mini", embedding_model: str = "all-MiniLM-L6-v2" ): # 初始化 OpenAI 客户端（仅用于对话生成） self.client = OpenAI() self.model = model # 初始化本地 Sentence-Transformers 模型 print(f"[初始化] 加载嵌入模型: {embedding_model}") self.embedder = SentenceTransformer(embedding_model) self.embedding_dim = self.embedder.get_sentence_embedding_dimension() # 会话标识 self.session_id = str(uuid.uuid4())[:8] # 初始化 ChromaDB（持久化存储） self.chroma_client = chromadb.PersistentClient(path="./assistant_memory") # 创建或获取集合时指定嵌入函数为 None，因为我们手动计算向量 self.collection = self.chroma_client.get_or_create_collection( name=f"conversation_{self.session_id}", metadata={"hnsw:space": "cosine"} ) # 近期对话历史（保留最近 10 条用于快速访问） self.recent_messages = [] # 结构化摘要 self.global_summary = { "user_name": None, "preferences": [], "important_dates": [], "ongoing_tasks": [] } # 指代词触发词 self.trigger_words = ["上次", "之前", "刚才", "刚刚", "前面", "刚刚说", "之前说", "刚才说"] def _get_embedding(self, text: str) -> list: """使用本地 Sentence-Transformers 模型计算嵌入向量""" # 返回 list 类型，ChromaDB 要求 float 列表 embedding = self.embedder.encode(text, normalize_embeddings=True) return embedding.tolist() def _should_retrieve(self, user_input: str) -> bool: """判断是否需要触发向量检索""" return any(word in user_input for word in self.trigger_words) def _store_conversation(self, user_msg: str, assistant_msg: str): """将一轮对话存入向量数据库""" turn_id = str(uuid.uuid4()) combined_text = f"用户: {user_msg}\n助手: {assistant_msg}" embedding = self._get_embedding(combined_text) self.collection.add( ids=[turn_id], embeddings=[embedding], metadatas=[{ "user_msg": user_msg, "assistant_msg": assistant_msg, "timestamp": datetime.now().isoformat(), "turn_type": "full_conversation" }], documents=[combined_text] ) def _retrieve_relevant_history(self, query: str, top_k: int = 3) -> list: """检索与当前查询相关的历史对话片段""" query_embedding = self._get_embedding(query) results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k, include=["documents", "metadatas", "distances"] ) retrieved = [] if results["documents"] and results["documents"][0]: for i, doc in enumerate(results["documents"][0]): distance = results["distances"][0][i] if results["distances"] else None meta = results["metadatas"][0][i] if results["metadatas"] else {} retrieved.append({ "content": doc, "similarity": 1 - distance if distance else None, "metadata": meta }) return retrieved def _update_global_summary(self, user_input: str, ai_response: str): """更新全局摘要（简单规则 + 可扩展为 LLM 提取）""" if "我叫" in user_input: name = user_input.split("我叫")[-1].strip().split()[0] self.global_summary["user_name"] = name if "喜欢" in user_input or "偏好" in user_input: self.global_summary["preferences"].append(user_input[:50]) def _build_context(self, user_input: str) -> list: """构建包含摘要和检索片段的混合上下文""" self.recent_messages.append({"role": "user", "content": user_input}) system_prompt = { "role": "system", "content": "你是一位贴心的个人助理，能够记住用户的偏好和之前的对话内容。" } context = [system_prompt] # 1. 注入全局摘要 if self.global_summary["user_name"]: summary_text = f"[用户信息] 姓名: {self.global_summary['user_name']}" if self.global_summary["preferences"]: summary_text += f", 偏好: {', '.join(self.global_summary['preferences'][-3:])}" context.append({"role": "system", "content": summary_text}) # 2. 如果触发指代词，进行向量检索 if self._should_retrieve(user_input): print("\n[调试] 检测到指代词，触发向量检索...") retrieved = self._retrieve_relevant_history(user_input, top_k=3) if retrieved: retrieved_text = "[相关历史对话片段]（来自之前的对话）:\n" for i, item in enumerate(retrieved): retrieved_text += f"片段{i+1}: {item['content']}\n" print(f"[调试] 检索到片段: {item['content'][:80]}...") context.append({"role": "system", "content": retrieved_text}) # 3. 添加最近对话 context.extend(self.recent_messages[-10:]) return context def chat(self, user_input: str) -> str: context = self._build_context(user_input) response = self.client.chat.completions.create( model=self.model, messages=context, temperature=0.7, ) ai_response = response.choices[0].message.content self.recent_messages.append({"role": "assistant", "content": ai_response}) self._store_conversation(user_input, ai_response) self._update_global_summary(user_input, ai_response) return ai_response

3）运行测试

以下是运行测试的示例

# ========== 测试运行 ========== if __name__ == "__main__": assistant = PersonalAssistant(model=model_name) print("=== 个人助理测试 (向量检索 + 混合召回) ===") # 第一段对话：建立记忆 print("\n--- 第一阶段：建立记忆 ---") queries_phase1 = [ "你好，我叫张三，我喜欢喝咖啡和爬山。", "我下周要去杭州出差三天。", "帮我查一下杭州下周的天气。", ] for q in queries_phase1: print(f"\n👤 用户: {q}") resp = assistant.chat(q) print(f"🤖 助手: {resp[:100]}..." if len(resp) > 100 else f"🤖 助手: {resp}") # 第二段对话：测试指代词检索 print("\n\n--- 第二阶段：测试指代词检索 ---") queries_phase2 = [ "我刚才说我喜欢喝什么来着？", "上次说的出差是去哪里？", "之前提到的杭州，有什么推荐的咖啡店吗？", ] for q in queries_phase2: print(f"\n👤 用户: {q}") resp = assistant.chat(q) print(f"🤖 助手: {resp}")

输出示例如下所示

[初始化] 加载嵌入模型: all-MiniLM-L6-v2
=== 个人助理测试 (向量检索 + 混合召回) ===
--- 第一阶段：建立记忆 ---
👤 用户: 你好，我叫张三，我喜欢喝咖啡和爬山。
🤖 助手: 你好，张三！很高兴认识你。☕️⛰️
我已经记住了你喜欢**喝咖啡**和**爬山**，这些都是非常棒的生活爱好呢！喝咖啡能提神醒脑，爬山能亲近自然、放松身心。
以后如果你想寻找不错的咖啡馆、挑选咖啡...
👤 用户: 我下周要去杭州出差三天。
🤖 助手: 收到，张三！杭州是个非常棒的城市，特别适合结合你的爱好来安排行程。☕️⛰️
既然你是去出差，时间可能比较紧凑，我为你整理了一些结合**咖啡**和**爬山**的灵感，供你参考：
📍 **关于爬山（亲...
👤 用户: 帮我查一下杭州下周的天气。
🤖 助手: 张三，关于杭州下周的具体天气预报，因为我目前无法直接联网获取实时的气象数据，为了信息的准确性，建议你出发前通过手机天气应用或网站确认一下最新的预报哦。📱
不过，根据杭州以往的气候特点，我有几个小贴士...

--- 第二阶段：测试指代词检索 ---
👤 用户: 我刚才说我喜欢喝什么来着？
[调试] 检测到指代词，触发向量检索...
[调试] 检索到片段: 用户: 你好，我叫张三，我喜欢喝咖啡和爬山。
助手: 你好，张三！很高兴认识你。☕️⛰️
我已经记住了你喜欢**喝咖啡**和**爬山**，这些都是非常棒的生活...
[调试] 检索到片段: 用户: 帮我查一下杭州下周的天气。
助手: 张三，关于杭州下周的具体天气预报，因为我目前无法直接联网获取实时的气象数据，为了信息的准确性，建议你出发前通过手机天...
[调试] 检索到片段: 用户: 我下周要去杭州出差三天。
助手: 收到，张三！杭州是个非常棒的城市，特别适合结合你的爱好来安排行程。☕️⛰️
既然你是去出差，时间可能比较紧凑，我为你...
🤖 助手: 张三，你刚才说过你喜欢**喝咖啡**呀！☕️
当然，你还提到过喜欢**爬山**。⛰️ 这两个爱好我都牢牢记住啦，毕竟这可是为你规划杭州行程的重要线索呢。
怎么突然考我啦？是不是想让我根据这个喜好，推荐一些杭州不错的咖啡店了？😄
👤 用户: 上次说的出差是去哪里？
[调试] 检测到指代词，触发向量检索...
[调试] 检索到片段: 用户: 你好，我叫张三，我喜欢喝咖啡和爬山。
助手: 你好，张三！很高兴认识你。☕️⛰️
我已经记住了你喜欢**喝咖啡**和**爬山**，这些都是非常棒的生活...
[调试] 检索到片段: 用户: 我刚才说我喜欢喝什么来着？
助手: 张三，你刚才说过你喜欢**喝咖啡**呀！☕️
当然，你还提到过喜欢**爬山**。⛰️ 这两个爱好我都牢牢记住啦，毕...
[调试] 检索到片段: 用户: 帮我查一下杭州下周的天气。
助手: 张三，关于杭州下周的具体天气预报，因为我目前无法直接联网获取实时的气象数据，为了信息的准确性，建议你出发前通过手机天...
🤖 助手: 张三，你上次说的是要去**杭州**出差呀！🏙️
你提到过是**下周**出发，行程大概是**三天**。我们之前还聊到了杭州的天气、适合爬山的路线（比如宝石山、龙井村），以及不错的咖啡店推荐呢。☕️⛰️
是对行程还有什么顾虑吗？还是需要我帮你再细化一下这三天的安排？😄
👤 用户: 之前提到的杭州，有什么推荐的咖啡店吗？
[调试] 检测到指代词，触发向量检索...
[调试] 检索到片段: 用户: 上次说的出差是去哪里？
助手: 张三，你上次说的是要去**杭州**出差呀！🏙️
你提到过是**下周**出发，行程大概是**三天**。我们之前还聊到了杭...
[调试] 检索到片段: 用户: 我刚才说我喜欢喝什么来着？
助手: 张三，你刚才说过你喜欢**喝咖啡**呀！☕️
当然，你还提到过喜欢**爬山**。⛰️ 这两个爱好我都牢牢记住啦，毕...
[调试] 检索到片段: 用户: 你好，我叫张三，我喜欢喝咖啡和爬山。
助手: 你好，张三！很高兴认识你。☕️⛰️
我已经记住了你喜欢**喝咖啡**和**爬山**，这些都是非常棒的生活...
🤖 助手: 张三，既然你那么喜欢**喝咖啡**，又是去**杭州**出差，我当然要好好为你挑选几家既能满足咖啡瘾，又适合你行程的店啦！☕️
结合你**三天出差**的时间以及喜欢**爬山**的爱好，我把推荐分成了三类，方便你根据工作安排灵活选择：
### 1. 🏞️ **爬山后的放松站（景区附近）**
既然你计划去**宝石山**或**龙井村**爬山，这些地方附近就有不错的咖啡点：
* **北山街沿线咖啡馆：** 爬完宝石山下来，沿着北山街走，有很多面朝西湖的咖啡馆。在这里点一杯手冲，看着湖景放松，非常惬意。
* **龙井村/满觉陇：** 虽然这里以茶闻名，但现在也有很多“茶咖”结合的小店。空气极好，适合爬山后休息，顺便体验一下杭州的特色风味。
### 2. 💼 **出差党友好型（适合办公/洽谈）**
如果你需要在行程中找个安静地方处理工作或见客户：
* **天目里（Oōeli）：** 这里是杭州的文化地标，有很多设计感极强的咖啡店（比如 % Arabica 等）。环境安静、审美在线，非常适合商务人士，离市区也不算远。
* **湖滨银泰附近：** 交通便利，选择多，适合利用碎片时间喝一杯。
### 3. 🏆 **精品咖啡爱好者必打卡**
杭州的精品咖啡氛围很浓，如果你想去专门品鉴一下：
* **金属手 (Metal Hands)：** 如果你追求咖啡品质，这家连锁精品咖啡在杭州的口碑很不错。
* **网格咖啡 (Grid Coffee)：** 也是专注于精品豆的选择，适合对风味有要求的你。
💡 **小建议：**
* 因为你是**下周**出发，建议你在去之前通过地图软件确认一下营业状态，避免跑空。
* 如果时间紧张，我可以帮你把**咖啡店**和**爬山路线**串起来，规划一个“高效摸鱼”路线，让你出差之余也能享受爱好。
你想先了解哪一类的具体位置？或者需要我帮你把它们放进你的三天行程里吗？😄