一、从一次聊天记录泄露说起

大家有没有想过,你辛辛苦苦做的聊天机器人,后台把用户和它的对话一股脑存进数据库,某天数据库被拖库,或者日志被人扒了,那些对话里包含的身份证号、银行卡、家庭住址,全成了别人的“金矿”。这不是吓唬人,而是真实发生过的事。很多开发者在做 LangChain 应用时,只顾着让对话历史“记住”上下文,却忘了给这些记忆加把锁。今天咱们就聊聊,怎么在配置对话历史存储的时候,把敏感数据的安全防护做扎实,让坏人即使拿到数据,也读不出有用的东西。

我见过不少项目,一开始为了图省事,直接用最原始的列表或者普通文件存对话记录。结果呢?一是数据明文躺在那,二是没有权限控制,三是连最基本的过期删除都没有。这就像把家里的钥匙放在门口的垫子下面,看似方便,其实危机四伏。

要解决这个问题,不能只靠某一个环节的单点防护,而是要从存储、读写、加密、脱敏、访问控制几个层面同时下手。下面我会用生活中的例子,配合实际的代码,一步步带大家把防护措施配置起来。

二、敏感数据到底藏在哪里

在动手之前,咱们先得知道哪些东西算敏感。用户和机器人聊“我想查一下我尾号1234的银行卡余额”,这里“尾号1234”是敏感数据;聊“我住在幸福小区3号楼”,这个住址是敏感数据;甚至有时候用户会无意中说出自己的手机号、邮箱、生日。这些信息如果原样存在对话历史里,一旦泄露,后果非常严重。

另外,还要注意间接敏感数据。比如用户问“我上周五做了个阑尾炎手术,现在能喝酒吗”,从这句话里能推断出用户的健康状态,这属于医疗隐私。再比如,用户说“我在某某银行办了贷款”,这反映了财务信息。这些间接数据同样需要保护。

那我们在配置存储时,就得先做一次“敏感数据盘点”,清楚知道哪些字段需要特别处理。LangChain 的对话历史记录通常是一组消息对象,每条消息包含 type(消息类型,比如 human 或 ai)、content(内容)、还有额外的 metadata(元数据)。敏感数据大多藏在 content 和 metadata 里。

三、LangChain对话历史存储是怎么回事

LangChain 为我们提供了多种对话历史存储方案,常见的有 InMemoryChatMessageHistory(内存版)、FileChatMessageHistory(文件版),还有配合 SQLChatMessageHistory(数据库版)这样的组件。它们的作用就是把聊天消息保存下来,让机器人能“记得”之前的对话内容。

比如,我们用一个字典结构来临时存放多 session 的聊天记录,这在 demo 里很常见。但一旦上了生产环境,这种临时存储就不够看了。这时很多人会换成数据库,比如 MySQL、PostgreSQL,或者用 Redis 这种内存数据库。但不管用哪种,敏感数据的安全防护逻辑几乎是通用的。

咱们可以这样理解:对话历史存储就像一个“笔记本”,机器人靠它记性变好,但笔记本里记的内容如果太隐私,就需要用密码本(加密)、涂黑(脱敏)、加锁(权限控制)来保护。而 LangChain 的扩展能力允许我们自定义存储方式,或者自定义消息的预处理逻辑,这就给了我们做防护的空间。

四、安全防护配置实战(Python技术栈)

下面我们就进入真正的实战环节。为了照顾大部分开发者,我们统一使用 Python 技术栈,配合 LangChain 社区版来做演示。在示例中,我会重点展示三个防护动作:

  1. 对敏感字段做数据脱敏,比如把身份证号中间几位替换成星号;
  2. 对写入存储前的消息内容做加密,让明文变成密文;
  3. 设置访问控制,只有持有正确密钥的程序才能读取。

4.1 先搭一个简单的对话历史存储

我们先写一个最原始的例子,看看数据是怎么被明文存储的。这里我们用一个自定义的存储类,内部用一个文件保存 JSON 数据,方便观察。

# 技术栈:Python 3.10 + LangChain 0.2.x
# 文件:simple_history.py
import json
from pathlib import Path

class SimpleFileHistory:
    """
    对话历史存储的简单实现
    注意:这个版本没有安全防护,仅用于演示“裸奔”的效果
    """
    def __init__(self, session_id: str, file_path: str = "./history.json"):
        self.session_id = session_id
        self.file_path = Path(file_path)
        self.messages = self._load()

    def _load(self):
        """从文件加载历史消息"""
        if self.file_path.exists():
            data = json.loads(self.file_path.read_text(encoding="utf-8"))
            return data.get(self.session_id, [])
        return []

    def add_message(self, role: str, content: str):
        """添加一条消息,注意:直接明文存储"""
        self.messages.append({
            "role": role,
            "content": content  # 敏感数据原样保存
        })
        self._save()

    def _save(self):
        """保存到文件"""
        all_data = {}
        if self.file_path.exists():
            all_data = json.loads(self.file_path.read_text(encoding="utf-8"))
        all_data[self.session_id] = self.messages
        self.file_path.write_text(
            json.dumps(all_data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

    def get_messages(self):
        """返回历史消息"""
        return self.messages


# 使用示例:看,明文写入
if __name__ == "__main__":
    history = SimpleFileHistory(session_id="user_001")
    history.add_message("human", "我的身份证号是110101199001011234")
    history.add_message("ai", "好的,我已经记住了你的身份证号。")
    print(history.get_messages())
    # 此时打开 history.json,能看到明文身份证号,非常危险

看到没有,就这么简单的一段代码,就把用户身份证号明文写进文件了。万一这个文件被同步到某个公共仓库,那用户的信息就全裸奔了。所以我们得着手改造。

4.2 加一层脱敏:让关键信息打码

脱敏是成本最低、见效最快的防护手段。我们可以在消息进入历史存储之前,把身份证、手机号、银行卡这类固定格式的敏感数据替换成星号。注意,脱敏会丢失部分原始信息,但很多场景下我们并不需要完整的敏感数据,比如只是用来做语义理解时,“身份证号是110***********1234”已经足够。

我们用 Python 的正则表达式实现一个脱敏函数,然后封装一个新的历史存储类。

# 技术栈:Python 3.10 + LangChain 0.2.x
# 文件:masked_history.py
import json
import re
from pathlib import Path

# 手机号:1开头的11位数字,脱敏后保留前3位和后4位
def mask_phone(content: str) -> str:
    """把手机号中间4位替换为星号"""
    # 1开头 + 10位数字,总长11位
    return re.sub(r'(1\d{2})\d{4}(\d{4})', r'\1****\2', content)

# 身份证号:18位,其中第7-14位是出生日期,比较敏感
def mask_id_card(content: str) -> str:
    """把身份证号中的出生日期部分替换为星号"""
    # 简单匹配18位身份证,前6位 + 8位日期 + 4位校验位
    return re.sub(r'(\d{6})\d{8}(\d{4})', r'\1********\2', content)

# 银行卡号:一般是16到19位数字,保留后4位
def mask_bank_card(content: str) -> str:
    """把银行卡号除了后4位以外的数字替换为星号"""
    # 这里用更精确的匹配:至少15位连续数字
    return re.sub(r'(\d{12,15})\d{4}', r'\1****', content)

class MaskedFileHistory:
    """
    带脱敏能力的对话历史存储
    """
    def __init__(self, session_id: str, file_path: str = "./masked_history.json"):
        self.session_id = session_id
        self.file_path = Path(file_path)
        self.messages = self._load()

    def _load(self):
        if self.file_path.exists():
            data = json.loads(self.file_path.read_text(encoding="utf-8"))
            return data.get(self.session_id, [])
        return []

    def add_message(self, role: str, content: str):
        """添加消息,先脱敏再保存"""
        safe_content = mask_phone(content)
        safe_content = mask_id_card(safe_content)
        safe_content = mask_bank_card(safe_content)

        self.messages.append({
            "role": role,
            "content": safe_content  # 保存的是脱敏后的内容
        })
        self._save()

    def _save(self):
        all_data = {}
        if self.file_path.exists():
            all_data = json.loads(self.file_path.read_text(encoding="utf-8"))
        all_data[self.session_id] = self.messages
        self.file_path.write_text(
            json.dumps(all_data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

    def get_messages(self):
        return self.messages


# 使用示例
if __name__ == "__main__":
    history = MaskedFileHistory(session_id="user_002")
    history.add_message("human", "我的手机号是13812345678,身份证是110101199001011234")
    history.add_message("ai", "收到,你的手机尾号是5678,对吧?")
    print(json.dumps(history.get_messages(), ensure_ascii=False, indent=2))
    # 输出内容中的手机号和身份证号都已经被打码

注意,上面代码里的脱敏逻辑比较朴素,实际项目中还可以用更完善的脱敏框架,比如 presidio 这类能识别自定义实体的库。不过思路是相通的:在写入之前过滤敏感信息。

4.3 再上一层加密:明文变密文

脱敏虽然好,但它破坏了原始数据。如果某些业务场景必须要完整的身份证号才能完成核验,那脱敏就不合适了。这时候就需要加密。加密之后,存储层的数据全是密文,就算文件泄露,没有密钥也解不开。

我们使用 Python 的 cryptography 库来做对称加密。加密和解密使用同一个密钥(或者由同一密钥派生)。这个密钥必须安全地存放在环境变量或密钥管理服务中,绝不能硬编码在代码里。

# 技术栈:Python 3.10 + LangChain 0.2.x + cryptography
# 文件:encrypted_history.py
import json
import base64
import os
from pathlib import Path
from cryptography.fernet import Fernet

class EncryptedFileHistory:
    """
    使用 Fernet 对称加密的对话历史存储
    """
    def __init__(self, session_id: str, key: bytes, file_path: str = "./encrypted_history.enc"):
        # 传入的 key 是 Fernet 密钥,一般通过环境变量获取
        self.cipher = Fernet(key)
        self.session_id = session_id
        self.file_path = Path(file_path)
        self.messages = self._load()

    def _encrypt_text(self, plain_text: str) -> str:
        """加密普通文本,返回 base64 编码的密文字符串"""
        token = self.cipher.encrypt(plain_text.encode("utf-8"))
        return base64.b64encode(token).decode("utf-8")

    def _decrypt_text(self, encrypted_text: str) -> str:
        """解密从存储读出的密文"""
        token = base64.b64decode(encrypted_text.encode("utf-8"))
        return self.cipher.decrypt(token).decode("utf-8")

    def _load(self):
        """从文件加载密文消息,并解密到内存中"""
        if self.file_path.exists():
            raw_data = json.loads(self.file_path.read_text(encoding="utf-8"))
            session_messages = raw_data.get(self.session_id, [])
            decoded_messages = []
            for msg in session_messages:
                decoded_messages.append({
                    "role": msg["role"],
                    # 注意:content 存的是密文,这里要解出来
                    "content": self._decrypt_text(msg["content"])
                })
            return decoded_messages
        return []

    def add_message(self, role: str, content: str):
        """添加消息,加密后再保存"""
        encrypted_content = self._encrypt_text(content)
        self.messages.append({
            "role": role,
            "content": content  # 内存中保留明文,方便程序使用
        })
        self._save(encrypted_content)

    def _save(self, encrypted_content: str):
        """把密文写入文件,注意这里不能把明文直接写入"""
        # 读取已有文件中的所有 session 数据,把密文更新进去
        all_data = {}
        if self.file_path.exists():
            all_data = json.loads(self.file_path.read_text(encoding="utf-8"))

        # 把该 session 现有的消息逐条加密(这里假设都是新的,简单处理)
        # 更严谨的做法是保存前统一加密所有消息
        all_data[self.session_id] = [
            {
                "role": m["role"],
                "content": self._encrypt_text(m["content"])  # 用当前消息重新加密
            }
            for m in self.messages
        ]
        self.file_path.write_text(
            json.dumps(all_data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

    def get_messages(self):
        """获取明文消息"""
        return self.messages


# 密钥生成工具:如果还没有密钥,可以运行这段生成
def generate_key() -> bytes:
    """生成一个 Fernet 密钥,保存到环境变量里"""
    key = Fernet.generate_key()
    print(f"请把这个密钥保存到环境变量: {key.decode()}")
    return key


if __name__ == "__main__":
    # 实践中应该从环境变量里读取,比如 os.environ.get("HISTORY_ENCRYPT_KEY")
    secret_key = generate_key()
    history = EncryptedFileHistory(
        session_id="user_003",
        key=secret_key,  # 这里演示直接传入
        file_path="./encrypted_history.json"
    )
    history.add_message("human", "我的住址是幸福小区3号楼202室")
    history.add_message("ai", "好的,你的住址已记录。")
    print(history.get_messages())

    # 直接查看文件内容,会发现都是密文,比如类似 gAAAAAB...
    print("--- 文件内容预览 ---")
    from pathlib import Path
    content = Path("./encrypted_history.json").read_text(encoding="utf-8")
    print(content)

注意,上面这个示例为了简短,做了一些简化。实际使用时,应该把“加载时解密全部消息”和“保存时加密全部消息”这两个动作分开,避免重复加密。另外,密钥的轮换、加密库的升级都需要纳入运维计划。

4.4 在 LangChain 的对话链里集成

LangChain 做对话应用时,通常会用到 ConversationChain,然后给它配一个 memory 对象。这个 memory 对象决定历史消息如何保存。我们可以把上面自定义的加密存储包装成 LangChain 的 ChatMessageHistory 接口,这样就能无缝接入。

LangChain 提供了一个抽象基类 BaseChatMessageHistory,我们只要实现 add_message 和 clear 方法即可。我们用刚才的 EncryptedFileHistory 改造一下。

# 技术栈:Python 3.10 + LangChain 0.2.x
# 文件:langchain_secure_history.py
import os
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.schema import HumanMessage, AIMessage
from langchain_core.chat_history import BaseChatMessageHistory
from cryptography.fernet import Fernet
# 注意:实际当中可以使用之前的 EncryptedFileHistory 作为底层存储
# 这里为了演示,直接内嵌加密逻辑

class EncryptedChatHistory(BaseChatMessageHistory):
    """
    把加密历史存储适配成 LangChain 的 ChatMessageHistory
    """
    def __init__(self, session_id: str, key_str: str, file_path: str):
        self.cipher = Fernet(key_str.encode() if isinstance(key_str, str) else key_str)
        self.session_id = session_id
        self.file_path = file_path
        self._messages = self._load_all()

    def _encrypt(self, text: str) -> str:
        return self.cipher.encrypt(text.encode()).decode()

    def _decrypt(self, token: str) -> str:
        return self.cipher.decrypt(token.encode()).decode()

    def _load_all(self):
        """从文件读取密文并解密为消息对象列表"""
        import json
        from pathlib import Path
        if Path(self.file_path).exists():
            data = json.loads(Path(self.file_path).read_text(encoding="utf-8"))
            raw_msgs = data.get(self.session_id, [])
        else:
            raw_msgs = []
        messages = []
        for item in raw_msgs:
            content = self._decrypt(item["content"])
            if item["type"] == "human":
                messages.append(HumanMessage(content=content))
            elif item["type"] == "ai":
                messages.append(AIMessage(content=content))
        return messages

    def add_message(self, message):
        """LangChain 会调用这个方法添加消息"""
        # 加密内容
        if isinstance(message, HumanMessage):
            msg_type = "human"
        elif isinstance(message, AIMessage):
            msg_type = "ai"
        else:
            msg_type = "system"
        encrypted_content = self._encrypt(message.content)
        # 为了简单,这里把消息追加到内存和文件
        self._messages.append(message)
        self._save_to_file(msg_type, encrypted_content)

    def _save_to_file(self, msg_type: str, encrypted_content: str):
        import json
        from pathlib import Path
        all_data = {}
        if Path(self.file_path).exists():
            all_data = json.loads(Path(self.file_path).read_text(encoding="utf-8"))
        session_msgs = all_data.get(self.session_id, [])
        session_msgs.append({
            "type": msg_type,
            "content": encrypted_content
        })
        all_data[self.session_id] = session_msgs
        Path(self.file_path).write_text(
            json.dumps(all_data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

    def clear(self):
        """清空历史"""
        self._messages = []
        import json
        from pathlib import Path
        if Path(self.file_path).exists():
            data = json.loads(Path(self.file_path).read_text(encoding="utf-8"))
            data.pop(self.session_id, None)
            Path(self.file_path).write_text(
                json.dumps(data, ensure_ascii=False, indent=2),
                encoding="utf-8"
            )


# 与 ConversationBufferMemory 结合使用
def build_memory(session_id: str, key: str, file_path: str):
    # 创建我们的加密历史存储
    chat_history = EncryptedChatHistory(
        session_id=session_id,
        key_str=key,
        file_path=file_path
    )
    # 用 ConversationBufferMemory 包装起来
    memory = ConversationBufferMemory(
        chat_memory=chat_history,
        return_messages=True
    )
    return memory


if __name__ == "__main__":
    # 真实环境一定要从环境变量读取,这里演示写死
    os.environ["HISTORY_KEY"] = Fernet.generate_key().decode()
    key = os.environ["HISTORY_KEY"]

    memory = build_memory(
        session_id="langchain_user_01",
        key=key,
        file_path="./langchain_encrypted_history.json"
    )

    # 模拟 LangChain 写入对话历史
    memory.chat_memory.add_message(HumanMessage(content="我的工资卡号是6222020200112233445"))
    memory.chat_memory.add_message(AIMessage(content="好的,卡号已保存。"))

    # 读取历史(解密后)
    for msg in memory.chat_memory.messages:
        print(f"{msg.type}: {msg.content}")

    # 查看存储文件,确认是密文
    print(Path("./langchain_encrypted_history.json").read_text(encoding="utf-8"))

通过适配 BaseChatMessageHistory,我们就能够把加密能力平滑地嵌入 LangChain 的对话流程,不需要改动业务代码。这种“适配器”模式在真实项目里非常实用。

4.5 访问控制和过期清理

除了加密和脱敏,我们还要限制谁能访问这些历史文件。如果是在 Linux 服务器上,可以通过文件权限 chmod 600 history.enc 让只有当前用户能读写。如果是数据库,则要设置独立的数据库账号,只给读写特定表的权限。在应用层,还可以使用 API Key 或者身份认证来限制调用。

另外,敏感数据不应该永久保留。假设我们的聊天机器人的对话只保留 30 天,30 天之后就应该自动清理这些加密记录。我们可以写一个简单的调度任务,扫描存储文件,删除超过保质期的 session。

下面演示一个带过期时间的加密历史存储,我们会给每一条消息记录一个时间戳,并在读取时跳过过期数据。

# 技术栈:Python 3.10 + LangChain 0.2.x
# 文件:expiring_history.py
import json
import time
from pathlib import Path
from cryptography.fernet import Fernet

class ExpiringEncryptedHistory:
    """
    带过期时间和加密的对话历史存储
    """
    def __init__(self, session_id: str, key: str, file_path: str, expire_seconds: int = 2592000):
        # expire_seconds 默认30天
        self.cipher = Fernet(key.encode())
        self.session_id = session_id
        self.file_path = Path(file_path)
        self.expire_seconds = expire_seconds

    def _encrypt(self, text: str) -> str:
        return self.cipher.encrypt(text.encode()).decode()

    def _decrypt(self, token: str) -> str:
        return self.cipher.decrypt(token.encode()).decode()

    def add_message(self, role: str, content: str):
        """添加一条带时间戳的加密消息"""
        data = self._load_all_data()
        session_msgs = data.get(self.session_id, [])
        session_msgs.append({
            "role": role,
            "content": self._encrypt(content),
            "timestamp": time.time()
        })
        data[self.session_id] = session_msgs
        self._save_all_data(data)

    def get_active_messages(self):
        """只返回没过期且解密后的消息"""
        data = self._load_all_data()
        session_msgs = data.get(self.session_id, [])
        now = time.time()
        active_msgs = []
        for msg in session_msgs:
            if now - msg["timestamp"] <= self.expire_seconds:
                active_msgs.append({
                    "role": msg["role"],
                    "content": self._decrypt(msg["content"]),
                    "timestamp": msg["timestamp"]
                })
            # 如果消息过期,就不返回(实际可以顺手清理)
        return active_msgs

    def cleanup_expired(self):
        """清理过期消息,并且把文件重新写回"""
        data = self._load_all_data()
        now = time.time()
        new_data = {}
        for sid, msgs in data.items():
            new_msgs = []
            for msg in msgs:
                if now - msg["timestamp"] <= self.expire_seconds:
                    new_msgs.append(msg)
            if new_msgs:
                new_data[sid] = new_msgs
        self._save_all_data(new_data)

    def _load_all_data(self):
        if self.file_path.exists():
            try:
                return json.loads(self.file_path.read_text(encoding="utf-8"))
            except json.JSONDecodeError:
                return {}
        return {}

    def _save_all_data(self, data: dict):
        self.file_path.write_text(
            json.dumps(data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )


if __name__ == "__main__":
    key = Fernet.generate_key().decode()
    hist = ExpiringEncryptedHistory(
        session_id="user_expire",
        key=key,
        file_path="./expiring_history.json",
        expire_seconds=86400  # 1天过期,方便测试
    )
    hist.add_message("human", "我的邮箱是test@example.com")
    hist.add_message("ai", "邮箱已记录")

    # 读取有效消息
    print(hist.get_active_messages())

    # 模拟等待超过1秒,然后清理(这里为了演示,可以把 expire_seconds 设成1秒)
    import time
    time.sleep(2)
    hist.cleanup_expired()
    print(hist.get_active_messages())  # 返回空列表

这种“时间戳 + 惰性清理”的模式,能保证即便存储介质泄露,攻击者拿到的也只是过期失效的密文。如果再配合提前销毁密钥,那就算密文还在,也永远无法解密。

五、应用场景分析

哪些场景特别需要这套安全防护配置呢?

第一个典型场景是金融客服机器人。用户会询问转账记录、信用卡账单,甚至可能直接贴上卡号。如果对话历史被明文存储,一旦被爬虫或者内部员工非法访问,风险极大。我们的脱敏和加密配置在这里能起到决定性作用。

第二个场景是医疗咨询机器人。患者会描述病症、用药情况,这些健康信息属于法律严格保护的隐私。这时候,最好做到“数据可用但不可见”,也就是程序内部可以用明文处理,但落到磁盘时必须密文。

第三个场景是教育领域,尤其是面向未成年人的学习助手。未成年人很容易无意识地透露姓名、学校、家庭住址。我们不仅需要脱敏,还要限制数据保留时间,避免长期沉淀导致隐私累积。

第四个场景是企业内部的文档问答助手。员工在提问时可能会粘贴包含商业机密的代码或合同条款。我们可以针对这种场景做更细粒度的脱敏,比如识别密钥、API Token 等字段并直接丢弃。

六、技术优缺点

先说说这种防护方式的优点。加密存储使得数据在静止状态下是安全的,即使硬盘被拿走、文件被复制,没有密钥也无法读取。脱敏操作简单高效,能快速降低敏感数据的暴露面,而且不需要改动存储结构。时间戳过期机制则保证了数据“保质期”,符合数据最小化原则。

但它也有缺点。加密会带来一定的性能开销,每一次读写都需要加解密操作,在大量对话记录时可能会成为瓶颈。脱敏则可能丢失原始信息,在某些需要完整数据的应用场景里会受限。另外,密钥管理本身是个复杂话题。如果密钥保护得不好,比如硬编码在代码里,那等于明文存储。还有,加密后的数据无法直接做全文搜索,除非引入加密搜索技术,这又大大增加了系统复杂度。

权衡之下,我的建议是:根据数据敏感程度分级处理。纯日常闲聊可以脱敏处理;涉及身份信息和财务信息,必须加密;敏感度最高的场景,还要配合权限管理、审计日志和定期密钥轮换。

七、注意事项

第一,密钥绝对不能写死在代码里。要用环境变量或者专门的密钥管理服务,比如 AWS KMS、Vault。并且要定期轮换密钥。一旦发现密钥泄露,要立即更换,并利用旧密钥重新加密数据。

第二,写入日志的时候要小心。很多应用在调试时会打印完整对话内容,这不经意间就把敏感信息暴露在日志系统里。建议记录消息 ID 和角色,而不是内容,或者对内容做摘要。

第三,要考虑备份文件的安全性。很多人只加密了主存储文件,但备份文件是明文,或者备份文件没有同样加密。这等于前门锁了,后门大开。备份时也要沿用同样的加密策略。

第四,LangChain 的 memory 不止一种。比如 ConversationSummaryMemory 会生成摘要并存储,摘要中可能也包含敏感信息。不要只关注原始消息,摘要同样需要脱敏或加密。

第五,内存中的明文数据也要防范。如果有人能 dump 内存,一样能得到敏感信息。虽然这不是配置能完全解决的,但可以定期清理不用的历史数据,并尽量缩短明文在内存中的生命周期。

第六,权限控制不能少。无论是文件还是数据库,都要遵循最小权限原则。应用进程用独立的系统账号,数据库账号只授权必要的表操作,这样即使被入侵,攻击者也无法读写其他数据。

八、文章总结

今天我们用唠家常的方式,聊了聊 LangChain 对话历史存储中敏感数据泄露的防护问题。我们从“风险从哪里来”说起,分析了敏感数据的藏身之处,然后一步步演示了如何对对话历史进行脱敏、加密、过期清理,以及如何把这些能力适配到 LangChain 的 memory 体系中。防护不是一刀切,要有层次。脱敏解决“不该看”的问题,加密解决“看了也白看”的问题,过期清理解决“不需要存这么久”的问题。

代码里用的例子都比较简化,但思路是通用的。希望你做项目时,在给机器人“增强记忆”的同时,也记得给它戴上一副“防护面具”。数据安全无小事,尤其是涉及用户隐私的时候。多花一点点时间配置好这些防护措施,就能避免未来可能的大麻烦。