共计 2667 个字符,预计需要花费 7 分钟才能阅读完成。
一、为什么我们需要上下文工程?
最近在开发客服对话机器人时,发现当对话轮次超过 10 轮后,AI 就开始出现前言不搭后语的情况。比如用户刚说完 ” 我想订周五的机票 ”,过几轮再问 ” 那天的天气如何 ” 时,AI 已经完全忘记 ” 周五 ” 这个关键信息。这种上下文丢失问题严重影响用户体验。

经过排查发现,主流大语言模型的上下文窗口就像一块 ” 白板 ”,新内容会不断覆盖旧内容。更麻烦的是,重要信息和非重点信息混杂在一起,导致关键记忆难以留存。
二、技术方案选型指南
2.1 常见方案对比
- 纯 Prompt 方案 :在每轮对话中重复关键信息
- 优点:实现简单
- 缺点:浪费 token,超过窗口大小仍会丢失
- 完整记忆库 :保存全部对话历史
- 优点:信息完整
- 缺点:检索效率低,噪声干扰严重
- 向量记忆库 :将信息编码为向量存储
- 优点:支持语义检索,节省空间
- 缺点:需要额外基础设施
实测发现,混合使用分层 Prompt 和向量记忆库的方案性价比最高。下面分享我的实现方法。
三、实战四步走方案
3.1 分层提示词设计
def build_prompt(system_msg, memory, user_input):
"""
构建分层提示模板
:param system_msg: 系统级指令
:param memory: 相关记忆片段
:param user_input: 当前用户输入
"""return f"""
[系统指令] {system_msg}
[相关记忆] {memory}
[当前对话] {user_input}
"""
关键技巧:
- 系统指令保持稳定,定义 AI 的角色和能力
- 记忆部分动态注入,只保留相关性高的内容
- 当前对话保持简洁,避免冗余
3.2 记忆管理系统实现
使用 FAISS+LangChain 搭建记忆库:
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
import logging
class MemoryManager:
def __init__(self):
self.embeddings = OpenAIEmbeddings()
self.db = FAISS.from_texts([""], self.embeddings) # 初始化空数据库
def add_memory(self, text: str, metadata: dict):
"""添加记忆片段"""
try:
self.db.add_texts([text], [metadata])
logging.info(f"Memory added: {text[:50]}...")
except Exception as e:
logging.error(f"Add memory failed: {str(e)}")
def retrieve_memory(self, query: str, k=3) -> list:
"""检索相关记忆"""
try:
docs = self.db.similarity_search(query, k=k)
return [doc.page_content for doc in docs]
except Exception as e:
logging.error(f"Retrieve memory failed: {str(e)}")
return []
3.3 对话状态机设计
用有限状态机管理对话流程:
graph LR
A[初始状态] -->| 问候 | B[信息收集]
B -->| 完成必填项 | C[服务提供]
C -->| 用户提问 | D[QA 模式]
D -->| 返回 | C
对应代码实现:
class DialogStateMachine:
def __init__(self):
self.current_state = "INIT"
def transition(self, user_input):
if self.current_state == "INIT" and "你好" in user_input:
self.current_state = "INFO_COLLECT"
return "请问需要什么帮助?"
# 其他状态转移逻辑...
四、性能优化技巧
4.1 记忆检索优化
- 相似度算法选择:
- 余弦相似度:适合常规语义匹配
- 内积相似度:对短文本效果更好
- 混合检索策略:
def hybrid_retrieve(self, query): # 先按时间筛选最近 10 条 recent = self.db.search(query, filter={"time": {"$gt": time.time()-3600}}, k=10) # 再语义检索 semantic = self.db.similarity_search(query, k=3) return recent + semantic
4.2 上下文窗口管理
实现滑动缓存策略:
def update_context_window(self, new_msg):
# 保留最近 5 轮对话
if len(self.context_window) >= 5:
self.context_window.pop(0)
self.context_window.append(new_msg)
# 将移出窗口的内容存入长期记忆
if len(self.context_window) == 5:
self.memory_manager.add_memory(self.context_window[0])
五、避坑指南
5.1 避免记忆污染
- 设置记忆有效期:
# 添加记忆时记录时间戳 metadata = {"create_time": time.time(), "expire_in": 3600} - 定期清理:
def clean_expired_memories(self): expired = self.db.search("", filter={"create_time": {"$lt": time.time()-3600}}) self.db.delete(expired)
5.2 敏感信息处理
实现记忆擦除功能:
def erase_sensitive_memory(self, keyword):
targets = self.db.search(keyword)
for doc in targets:
if "credit_card" in doc.page_content:
self.db.delete([doc.id])
六、进阶思考方向
在实现基础功能后,我开始思考更深入的问题:
1. 如何对记忆进行压缩摘要,避免信息膨胀?
2. 多模态记忆(图片、语音等)该如何存储和检索?
3. 不同记忆之间如何建立关联网络?
这些问题的解决,或许能让我们打造出更像人类的 AI 记忆系统。你在实践中遇到过哪些记忆管理的挑战?欢迎一起探讨。
正文完
