记忆的持久化与多用户

💾 如何使用 PostgresChatMessageHistory 将对话历史存入 PostgreSQL？写出关键配置。

要将对话历史持久化到 PostgreSQL，LangChain 提供了开箱即用的 PostgresChatMessageHistory。它基于 psycopg2 或 asyncpg，将每一条聊天消息作为一行数据存储在数据库表中。

关键步骤与配置：

1. 安装依赖 需要安装 langchain-community 以及 PostgreSQL 驱动（通常为 psycopg2-binary）。

pip install langchain-community psycopg2-binary

1. 准备数据库表

LangChain 不会自动创建表，你需要预先执行以下 SQL：

CREATE TABLE chat_history (
    id SERIAL PRIMARY KEY,
    session_id VARCHAR(255) NOT NULL,
    message JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_session_id ON chat_history(session_id);

表结构可以根据需要调整，但 session_id 和 message 是必须的。message 字段存储 LangChain 消息对象的 JSON 序列化形式。

1. 配置连接 使用连接字符串（DSN）创建 PostgresChatMessageHistory 实例。

from langchain_community.chat_message_histories import PostgresChatMessageHistory

history = PostgresChatMessageHistory(
    connection_string="postgresql://user:password@localhost:5432/mydb",
    session_id="user_123_session_001",
    table_name="chat_history"
)

1. 集成到 Memory 中 将 history 对象传递给 ConversationBufferMemory 等记忆组件。

高级配置与技巧：

• 自定义表名：通过 table_name 参数指定。

• 使用异步驱动：如果使用 asyncpg，可以换成 PostgresChatMessageHistory 的异步版本（或直接使用 AsyncPostgresChatMessageHistory，具体取决于版本）。

• 处理连接池：在高并发下，建议使用连接池（如 SQLAlchemy 配合 QueuePool）来管理数据库连接，而不是每次创建新的 psycopg2 连接。你可以通过自定义引擎传入：

from sqlalchemy import create_engine
engine = create_engine("postgresql://...", pool_size=10)
history = PostgresChatMessageHistory(engine=engine, session_id="...")

• 从 LangChain 0.0.30+ 开始支持直接传入 SQLAlchemy 引擎。

• 索引优化：确保 session_id 上有索引，因为每次加载记忆都是按 session_id 查询。

内部实现细节： add_message 方法会将 LangChain 的消息对象（如 HumanMessage）序列化成 JSON 格式存入 message 列。加载时，它执行 SELECT * FROM chat_history WHERE session_id = %s ORDER BY id，然后反序列化 JSON 得到消息列表。

踩过的坑：

• 如果消息中包含特殊字符，JSON 序列化可能会失败？LangChain 的消息对象都实现了 .json() 方法，一般没问题。

• 在高并发下，多个请求可能同时写入同一个 session，需要确保数据库的隔离级别和连接管理得当，避免死锁。

---

🏷️ 持久化记忆时，如何同时存储用户 ID、会话 ID 等元信息？

在很多场景下，你不仅需要存储对话内容，还需要关联用户 ID、租户 ID、会话标题等元数据，以便进行筛选、审计或个性化。

方案一：扩展数据库表，增加元数据列 在创建 chat_history 表时，增加 user_id、tenant_id、session_title 等字段。然后在每次插入消息时，手动设置这些列的值。 LangChain 原生的 PostgresChatMessageHistory 不支持传入额外元数据，因此你需要自定义一个子类，重写 add_message 方法。

from langchain_community.chat_message_histories import PostgresChatMessageHistory

class ExtendedPostgresHistory(PostgresChatMessageHistory):
    def __init__(self, user_id: str, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.user_id = user_id

    def add_message(self, message):
        # 先调用父类方法，但需要额外插入 user_id
        # 父类实现是执行 INSERT，我们可以覆盖整个方法
        import json
        from psycopg2 import sql
        msg_json = json.dumps(message.dict())
        with self.engine.connect() as conn:
            conn.execute(
                sql.SQL("INSERT INTO {} (session_id, message, user_id) VALUES (%s, %s, %s)").format(sql.Identifier(self.table_name)),
                (self.session_id, msg_json, self.user_id)
            )

使用时，你就可以为不同用户创建不同的 ExtendedPostgresHistory 实例。

方案二：将元数据编码到 session_id 中 一种更简单但不太规范的方法是把元数据（如 user_id）嵌入 session_id 本身。例如，采用 user_123:session_456 作为 session ID。这样你可以通过字符串解析来提取用户 ID，但缺点是无法利用数据库索引高效查询某个用户的所有会话。

方案三：使用支持元数据的 BaseChatMessageHistory 实现 LangChain 的 BaseChatMessageHistory 是一个非常灵活的基类。你可以从头实现一个，使用 SQLAlchemy 模型，包含你所需要的所有元数据字段。这能获得最大的控制权，但需要编写更多代码。

推荐做法：在大多数项目中，扩展数据库表并自定义历史记录类是最佳平衡点。它保持了与 LangChain 生态的兼容性，又提供了所需的元数据存储能力。

---

📄 如果你需要将会话历史导出为 OpenAI 格式的 JSON，LangChain 的消息对象有对应方法吗？

是的，LangChain 的消息对象可以方便地转换为 OpenAI 兼容的 JSON 格式。这对于迁移数据、分析或使用其他工具非常有用。

消息对象的转换方法：

• 每个消息对象（HumanMessage, AIMessage 等）都有 .dict() 和 .json() 方法，可以序列化为 Python 字典或 JSON 字符串。

• LangChain 也提供了工具函数 convert_messages_to_openai_format（具体名称可能变化），将消息列表转换为 OpenAI API 所需的格式，即包含 role 和 content 的字典列表。

示例：

from langchain.schema import HumanMessage, AIMessage
from langchain.adapters import openai as openai_adapter

messages = [
    HumanMessage(content="你好"),
    AIMessage(content="你好！有什么可以帮助你的？")
]

# 方法1：使用内置转换
openai_format = openai_adapter.convert_messages_to_openai_format(messages)
# 结果: [{'role': 'user', 'content': '你好'}, {'role': 'assistant', 'content': '你好！...'}]

# 方法2：手动转换
def to_openai_format(messages):
    return [{"role": "user" if isinstance(m, HumanMessage) else "assistant", "content": m.content} for m in messages]

导出为 JSON 文件：

import json

with open("chat_export.json", "w") as f:
    json.dump(openai_format, f, ensure_ascii=False, indent=2)

LangChain 的 HumanMessage 映射为 user，AIMessage 映射为 assistant，SystemMessage 映射为 system，FunctionMessage 映射为 function。这种直接转换使得无缝切换到 OpenAI 的 API 或任何兼容 OpenAI 格式的工具成为可能。

---

🔒 在多用户场景下，你如何保证用户 A 不能访问用户 B 的对话历史？

多用户隔离是安全性的基本要求。实现隔离的关键在于访问控制和数据隔离两个层面。

1. 应用层访问控制 在 API 层面，你需要对用户身份进行认证（例如 JWT token），然后在每次请求时验证该用户是否有权访问目标 session_id。绝不能仅依赖前端传过来的 session_id，必须将 session_id 与已验证的 user_id 绑定。

典型的流程：

• 用户登录后，后端生成或接收一个 session_id，并在服务器端（如数据库或缓存）建立 session_id -> user_id 的映射。

• 当用户发出聊天请求时，后端从认证上下文获取 user_id，然后检查请求中的 session_id 是否属于该 user_id。如果不匹配，返回 403。

• 数据层隔离

在数据库或存储中，确保对话数据与用户强关联。有两种常见做法：

• 基于 user_id 命名 session_id：将 user_id 作为 session_id 的前缀或命名空间。例如 session_id = f"user_{user_id}:session_{session_key}"。这样仅凭 session_id 就能区分用户。但这不够安全，因为理论上其他用户可能猜测。

• 在查询时增加 user_id 过滤：在 PostgresChatMessageHistory 中，自定义查询逻辑，始终附加 WHERE user_id = ? 条件。这需要你扩展历史记录类，并在加载消息时根据已验证的 user_id 过滤，而不是仅凭 session_id。

• 利用 LangChain 的 BaseChatMessageHistory 扩展 你可以在自定义历史记录类的 messages 属性中，根据当前的 user_id 和 session_id 构建查询，确保只返回该用户在此会话下的消息。这样即使 session_id 被泄露，其他用户也无法直接通过它查询到数据（因为数据库查询会强加 user_id 条件）。

最佳实践：

• 永远不要信任客户端：所有敏感操作都在服务端执行。

• 避免使用自增 ID 作为会话标识：使用 UUID 或加密随机字符串。

• 对所有数据库查询都施加 user_id 约束。

---

📛 使用 Redis 存储会话时，你会如何设计 key 的命名规则？

在 Redis 中，key 的命名既影响可读性，也影响维护和集群分片。一个好的命名规则应该清晰、可扩展且易于管理。

推荐命名方案：{命名空间}:{资源类型}:{标识符}

对于聊天历史，常见的方式是：

• 按用户和会话隔离：chat:session:{user_id}:{session_id} 例如 chat:session:1001:d8f3a2b0。

• 仅按会话隔离（不显式包含用户 ID）：chat:session:{session_id}，但你需要保证 session_id 是全局唯一的，且在业务层处理用户隔离。

• 如果需要按时间清理，可以在 key 中加入日期：chat:session:2024-01-15:{session_id}，但这会使跨天会话管理复杂化。

设计考量：

• 避免 key 过大：过长的 key 会占用更多内存，影响性能。

• 使用冒号分隔：冒号是 Redis 的常用层级分隔符，在 Redis Desktop Manager 等工具中会以文件夹形式展示，便于浏览。

• 加入版本号：如果未来可能改变存储结构，可以在命名空间中加入版本，如 chat:v2:session:...。

• 便于批量操作：比如你想清理某个用户的所有会话，可以用 SCAN 命令扫描 chat:session:1001:* 并删除。

示例：

session_id = f"chat:session:{user_id}:{session_uuid}"
history = RedisChatMessageHistory(session_id=session_id, url="redis://...")

这样的命名清晰明了，方便监控、调试和运维。

---

🌐 当你使用 LangServe 部署一个带记忆的链时，如何自动管理 session？

LangServe 是 LangChain 的服务化框架，它可以自动将你的链部署为 REST API。对于带记忆的链，LangServe 内置了自动会话管理功能。

工作原理：

• 当客户端调用 API 时，需要在请求头中传递 X-Session-Id（或请求体中的 session_id 字段，取决于配置）。

• LangServe 会自动解析该会话 ID，并使用它来初始化或加载对应的 Memory。

• 你不需要在每个请求中手动创建 Memory，只需在定义链时提供 Memory 的工厂函数或配置。

具体步骤：

1. 定义链时，使用 RunnableWithMessageHistory 包装你的链。这个包装器会为每个 session_id 自动管理历史记录。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.memory import ChatMessageHistory

def get_session_history(session_id: str):
    # 根据 session_id 返回一个 ChatMessageHistory 实例（可以是内存或持久化）
    return ChatMessageHistory()

chain_with_history = RunnableWithMessageHistory(
    your_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

1. 使用 LangServe 的 add_routes 注册。LangServe 会自动处理 session_id 的传递。

from langserve import add_routes
add_routes(app, chain_with_history, path="/chat")

1. 客户端调用：在 POST 请求的 JSON body 中包含 input 和 session_id（或通过 header X-Session-Id）。LangServe 会自动从请求中提取 session_id，并调用 get_session_history 获取对应的历史记录，注入到链中。

优势：

• 完全自动化，开发者无需在链中手动管理 Memory。

• 支持多种后端（内存、Redis、Postgres等），只需在 get_session_history 中返回相应的 ChatMessageHistory 实例。

• 同一个链可以同时服务多个用户，自动隔离。

踩过的坑：确保 get_session_history 是线程安全的，并且在高并发下能正确返回对应会话的历史记录。如果使用 Redis 等共享存储，注意连接池配置。

---

🔄 服务重启后，之前存储在内存中的记忆会丢失吗？如何避免？

如果使用基于内存的 ChatMessageHistory（如默认的 ChatMessageHistory 或 InMemoryStore），服务重启后所有对话历史将完全丢失。因为数据只存在于进程内存中。

避免丢失的方法：使用持久化的 ChatMessageHistory 后端。

LangChain 提供了多种持久化后端，你可以根据需求选择：

• 数据库：PostgresChatMessageHistory, SQLChatMessageHistory。数据存储在磁盘，重启后依然存在。

• 缓存/消息队列：RedisChatMessageHistory。Redis 支持持久化（RDB/AOF），重启后数据可以恢复，但取决于配置。

• 云服务：DynamoDBChatMessageHistory (AWS), CosmosDBChatMessageHistory (Azure) 等。

示例：切换到 Redis

from langchain.memory.chat_message_histories import RedisChatMessageHistory

history = RedisChatMessageHistory(session_id="...", url="redis://localhost:6379")
memory = ConversationBufferMemory(chat_memory=history)

这样即使服务重启，对话历史仍然存储在 Redis 中，不会丢失。

如果必须使用内存（比如开发环境），可以考虑使用外部会话粘性（Sticky Sessions）和共享内存，但这在生产中不推荐。

注意：如果你的链使用了 ConversationSummaryMemory 等包含摘要的记忆，摘要数据本身也存储在 ChatMessageHistory 中，因此持久化后一并保存。

---

⏪ 你如何实现一个“记忆回滚”功能？即回到某个历史状态。

记忆回滚功能在某些场景下很有用，比如用户想撤销最后一次交互，或者回到对话的某个分支点。

实现思路：

• 维护消息列表的版本历史：在 ChatMessageHistory 中，保存每一次对话轮次后的消息列表快照。可以使用数据库的事务日志或单独的版本表。

• 提供回滚接口：根据时间戳或轮次 ID，将当前对话的消息列表替换为历史快照。

基于 PostgresChatMessageHistory 的回滚方案： 你可以扩展 PostgresChatMessageHistory，增加 rollback_to 方法。

1. 在 chat_history 表中增加 turn_id 字段，表示对话轮次。

2. 保存上下文时，同时记录轮次。

3. 回滚时，执行 DELETE FROM chat_history WHERE session_id = ? AND turn_id > ?，删除指定轮次之后的所有消息。

基于 Redis 的回滚方案： Redis 的 List 不支持直接按索引删除，但你可以：

• 存储每个轮次的消息在独立的 List 中，例如 chat:session:xxx:turn:1, turn:2。

• 主历史记录是一个指向当前轮次的引用。

• 回滚时，将引用指针回拨，并丢弃后面的轮次数据。

示例代码（概念）：

class RollbackableHistory(PostgresChatMessageHistory):
    def rollback_to_turn(self, turn_id):
        # 执行删除操作
        with self.engine.connect() as conn:
            conn.execute(
                f"DELETE FROM {self.table_name} WHERE session_id = %s AND turn_id > %s",
                (self.session_id, turn_id)
            )
        # 清除缓存（如果有）

用户体验设计：

• 通常与“编辑消息”或“重试”功能结合。在聊天界面，用户点击“撤销”后，前端显示回滚前的对话状态，后端同步清理记忆。

注意：回滚可能会破坏记忆中的摘要（如果使用了摘要记忆），需要重新生成摘要。

---

🐌 如果一个对话非常长，持久化存储的读写会成为瓶颈吗？如何优化？

是的，随着对话历史增长，从数据库加载整个历史可能变得缓慢，并且占用大量内存。持久化存储的读写瓶颈主要在两方面：数据库查询和网络传输。

优化策略：

1. 分页加载与懒加载：不必一次性加载全部历史。LangChain 的 Memory 目前主要是全量加载，但你可以自定义 load_memory_variables 方法，只加载最近的 N 轮或 N 个 token 的历史。例如，在 SQL 查询中使用 ORDER BY id DESC LIMIT 100，然后反转顺序。

2. 数据库索引优化：确保 session_id 和排序字段（如 id 或 created_at）上有合适的索引。对于极长的历史，考虑使用分区表（如按时间分区），让查询只扫描近期分区。

3. 使用摘要记忆：ConversationSummaryMemory 或 ConversationSummaryBufferMemory 会在服务端将早期对话压缩为摘要，只保留精炼的信息，从而减少每次加载的数据量。这是最有效的长期优化。

4. 缓存热点数据：对于活跃会话，可以将最近几轮对话缓存在 Redis 中，数据库作为冷存储。LangChain 的 RedisChatMessageHistory 本身就非常适合作为高速缓存。

5. 异步 I/O：使用异步版本的 ChatMessageHistory（如 AsyncRedisChatMessageHistory），避免 I/O 阻塞事件循环。

6. 减少网络往返：将多条消息的写入合并为批量操作。有些历史记录后端支持 add_messages 方法（复数），一次插入多条。

7. 数据库读写分离：对于读多写少的场景，使用只读副本进行历史读取。

实际经验：在绝大多数聊天应用中，很少有单会话超过几千条消息的。真正导致瓶颈的往往是没有限制地全量加载历史。因此，在 Memory 层面或 Prompt 层面限制历史长度是第一步。

---

🌍 在分布式部署中，Redis 作为记忆存储有哪些优势和需要注意的问题？

优势：

• 高性能：数据存储在内存，读写速度极快，延迟在微秒级，非常适合作为会话存储。

• 数据结构丰富：List 天然适合存储有序的消息列表，支持范围查询（LRANGE）和修剪（LTRIM）。

• 持久化选项：通过 RDB 或 AOF 实现数据持久化，重启后可恢复。

• 分布式支持：Redis Cluster 或 Sentinel 可以实现高可用和水平扩展。

• 共享状态：所有服务实例共享同一份记忆数据，天然支持无状态服务的水平扩展。

• 内建 TTL：轻松设置会话过期时间，自动清理旧数据。

需要注意的问题：

• 内存成本：所有数据存储在内存，当会话数量庞大或历史很长时，内存消耗可能很高。需要合理设置 maxmemory 和淘汰策略（如 volatile-lru），避免内存耗尽。

• 持久化配置：默认配置下 Redis 重启会丢失数据。必须开启 AOF 或 RDB，并测试恢复流程。

• 网络延迟与连接管理：在分布式环境中，Redis 通常作为独立服务部署。需要确保网络延迟低，并使用连接池（如 redis-py 的 ConnectionPool）防止连接数爆炸。

• 单点故障：如果使用单实例 Redis，宕机会导致所有会话丢失。应采用 Sentinel 或 Cluster 高可用方案。

• 原子性与事务：Redis 的 List 操作并非完全事务性的。在并发写入同一会话时，可能会出现顺序混乱？通常对话是串行的，但如果存在多个请求同时写，需要应用层加锁或使用 Redis 事务（MULTI/EXEC）。

• 数据一致性：如果使用 AOF 持久化，默认每秒同步一次，宕机可能丢失最后一秒的数据。根据可靠性要求调整 appendfsync。

最佳实践：

• 设置合理的 TTL 过期时间。

• 使用 RedisChatMessageHistory 时，指定 ttl 参数，每次写入自动刷新过期时间。

• 监控 Redis 的内存使用和命中率。

• 对于超长会话，定期用 LTRIM 修剪早期消息，保持 List 长度可控。

---

🆘 如何为记忆存储设置备份和恢复机制？

备份和恢复是生产环境必不可少的环节。方案取决于使用的存储后端。

1. PostgreSQL

使用数据库自带的备份工具：

• 逻辑备份：pg_dump 导出特定表的数据。

• 物理备份：pg_basebackup 或 WAL 归档。

• 可以设置定时任务（cron）每天导出 chat_history 表为 SQL 文件，并上传到对象存储（S3）。

• Redis

• 开启 AOF 或 RDB 持久化，定期将持久化文件复制到异地。

• 使用 redis-cli --rdb dump.rdb 或 BGSAVE 命令手动触发快照。

• 自动化脚本：定期执行 BGSAVE，并将生成的 dump.rdb 备份到安全位置。

• 应用层备份 你可以实现一个自定义的 ChatMessageHistory，在每次 add_message 时将消息同时写入一个备用存储（如 S3、日志文件）作为冷备。这样即使主存储损坏，也可以从备份中恢复。

恢复机制：

• 对于 PostgreSQL，使用 psql 或 pg_restore 恢复数据。

• 对于 Redis，停止服务，用备份的 dump.rdb 替换当前文件，重启即可。

• 恢复后，需要确保 session_id 对应的历史记录重新加载到内存或缓存中。

测试：定期进行恢复演练，确保备份文件可用。

---

⚠️ 你是否遇到过记忆数据与对话不同步的问题？是什么导致的？

是的，在复杂场景下，记忆与对话不同步的问题时有发生，主要源于并发写入、异常处理不当或消息顺序错误。

常见原因：

1. 并发请求导致消息交错：如果用户在上一轮回复尚未完成时又发送了新消息，两个请求可能同时向同一个 session_id 写入消息，导致 HumanMessage 和 AIMessage 的顺序混乱。因为 LangChain 的 add_message 是独立的数据库写入操作，没有应用层锁。

2. 异常时未正确保存状态：当链运行过程中发生异常（如 LLM 超时），可能已经执行了部分工具调用或生成了部分回复，但并未全部完成。此时如果仍调用 save_context，就会将不完整的对话存入记忆。或者，如果开发者忘记保存，就会丢失轮次。

3. 使用流式输出时更新时机错误：如果在流式生成过程中，每收到一个 chunk 就将其存入记忆，会导致记忆中出现不完整的句子。应该等完整回复生成后再一次性保存。

4. 摘要记忆更新冲突：ConversationSummaryMemory 在每次新对话后更新摘要。如果多个请求同时触发摘要更新，可能导致摘要基于过时或混乱的历史生成，进而丢失信息。

5. 消息对象序列化/反序列化错误：持久化时，如果自定义消息类型没有正确注册，加载时可能解析失败，导致部分历史丢失。

如何避免：

• 实现请求串行化：对于同一个 session_id，在应用层使用锁（如 asyncio.Lock）确保同一时刻只有一个请求在处理该会话。

• 使用原子操作：将多条消息的写入合并到一个事务中。例如，自定义 add_messages 方法，一次性写入整个轮次。

• 完善异常处理：确保在 finally 块中正确保存或回滚记忆状态。

• 流式处理规范：只在 astream 完全结束后才调用 memory.save_context。

• 版本化摘要：在摘要更新时，先读取当前历史，生成新摘要，然后使用 CAS（Compare-And-Swap）写入，避免冲突。

---

🧹 如何给记忆数据增加“遗忘”机制？比如自动清理超过 30 天未活动的会话。

遗忘机制对于隐私合规（如 GDPR 的“被遗忘权”）和资源管理至关重要。

实现方式：

1. 利用 Redis 的 TTL RedisChatMessageHistory 支持 ttl 参数。当会话在 ttl 秒内没有活动，整个 key 会被自动删除。这相当于“自动遗忘”。

history = RedisChatMessageHistory(session_id="...", ttl=2592000)  # 30天

1. PostgreSQL 定时清理任务 在数据库中，通过 created_at 字段和 session_id 分组，编写定时任务（如 pg_cron 或外部 cron 脚本）定期删除超过 30 天未更新的会话数据。

DELETE FROM chat_history
WHERE session_id IN (
    SELECT session_id FROM chat_history
    GROUP BY session_id
    HAVING MAX(created_at) < NOW() - INTERVAL '30 days'
);

1. 应用层主动遗忘 当用户请求删除账号或特定会话时，调用 memory.clear() 或直接操作存储后端删除对应的 session 记录。这属于“主动遗忘”。

2. 冷热数据分离

将超过一定时间的会话数据转移到廉价的归档存储（如对象存储），并从主存储中删除。需要时可以从归档恢复，但日常查询不会加载它们。

1. 摘要替代遗忘

如果不想完全删除，可以定期将旧会话压缩为一段摘要，然后删除原始消息，用摘要作为该会话的长期记忆。

设计要点：确保遗忘操作记录在审计日志中，以证明合规性。

---

📬 LangChain 中的 Message 对象有哪些类型？HumanMessage、AIMessage、FunctionMessage 等分别在何时使用？

LangChain 的消息系统为多角色对话提供了丰富的类型。这些类型帮助 LLM 区分不同来源的信息，从而更准确地理解上下文。

消息类型	使用场景	内容示例
HumanMessage	代表用户发送的消息。在聊天链中，用户的输入会被自动封装为此类型。	HumanMessage(content="你好")
AIMessage	代表 AI（LLM）的回复。如果 LLM 只是普通文本生成，返回此类。	AIMessage(content="你好！有什么可以帮你的？")
SystemMessage	系统提示词，用于设定对话背景或角色。通常放在消息列表的开头。	SystemMessage(content="你是一个乐于助人的助手。")
FunctionMessage	表示调用某个函数的请求。OpenAI 函数调用时，LLM 返回此类（包含函数名和参数）。	FunctionMessage(name="search", content='{"query": "天气"}')
ToolMessage	工具执行的结果。在 LangChain 的 Tool 调用后，将工具返回的内容封装为此类，反馈给 LLM。	ToolMessage(content="搜索结果：...", tool_call_id="123")
ChatMessage	通用消息，可指定任意角色（role 参数）。用于兼容非标准角色。	ChatMessage(role="moderator", content="请保持礼貌")
AIMessageChunk	AIMessage 的分块版本，用于流式输出。	流式生成时逐步返回。

何时使用：

• 在对话链中，框架自动创建 HumanMessage 和 AIMessage。

• 当你在构建 Prompt 模板时，可以手动创建 SystemMessage 来设定规则。

• 在使用 Agent 或工具时，会用到 ToolMessage / FunctionMessage 来传递工具执行结果。

• ChatMessage 提供了最大的灵活性，你可以自定义角色，但通常不建议滥用，因为 LLM 可能无法理解自定义角色。

这些消息类型确保了 LangChain 与不同 LLM 提供商（OpenAI、Anthropic等）之间的兼容性，LangChain 会在底层将它们转换为对应 API 需要的格式。

---

🧩 在将记忆传递给模型时，LangChain 是如何把消息列表转换为 Prompt 的字符串或消息数组的？

LangChain 的 Memory 在将历史注入 Prompt 时，会根据 Prompt 模板的需求，提供两种格式：字符串（String） 和 消息列表（List of Messages）。

1. 转换为字符串（传统模式） 当 Prompt 模板中包含 {chat_history} 占位符且该模板是纯字符串模板时，Memory 会将消息列表序列化为一个文本块。

2. 对于 ConversationBufferMemory，它默认使用 get_buffer_string() 方法，将消息转换为如下格式：

Human: 你好
AI: 你好！有什么可以帮你的？

• 这种格式是预定义的，但你可以在创建 Memory 时通过 human_prefix 和 ai_prefix 参数自定义前缀。

• 对于 ConversationSummaryMemory，它直接输出摘要文本（字符串）。

• 转换为消息列表（现代模式） 当 Memory 的 return_messages=True 时，它不会将历史转换为字符串，而是直接返回一个 List[BaseMessage]。这通常用于配合聊天模型（如 ChatOpenAI）的 Prompt 模板，因为这些模板接收消息数组而不是字符串。 LangChain 的 MessagesPlaceholder 可以在 Prompt 模板中放置一个变量名，该变量在运行时填充为消息列表。例如：

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

当传入 chat_history 为消息列表时，这些消息会被直接拼接在 Prompt 中。ChatOpenAI 原生支持消息数组，LangChain 会将其直接传递给 API。

内部转换细节：

• 如果你使用了 return_messages=False（默认），记忆会调用 get_buffer_string 生成字符串，然后通过常规的模板替换填入 Prompt。

• 如果 return_messages=True，记忆的 load_memory_variables 返回的字典中，history 键对应的值是一个消息列表，LangChain 会识别并将其注入到 MessagesPlaceholder 所在位置。

优点：消息数组更结构化，避免了因格式错误（如分隔符冲突）导致的问题，并且更好地利用了聊天模型的训练格式。

踩过的坑：当 Prompt 模板混用字符串和消息数组时可能会出错。务必保持一致性：如果使用聊天模型，建议采用消息列表模式。

通过这种灵活的转换机制，LangChain 使得记忆可以无缝适配不同类型的 LLM 和 Prompt 风格。

---