RAG 系列 9：上下文感知 RAG 与多轮对话记忆

上一篇文章已经实现了一条基础 RAG 链：用户提出问题，Retriever 从 Chroma 中找到相关文档，本地 Qwen3 再根据文档回答。

这条链可以回答“什么是任务分解？”，但是连续对话时还会遇到一个问题：

用户第一轮问“什么是任务分解？”，第二轮只问“它有哪些常见实现方式？”

第二句话里的“它”没有明确指向。单独把这句话交给向量数据库，Retriever 并不知道“它”表示任务分解，很可能召回错误内容。本篇在基础 RAG 上增加聊天历史和问题改写，让检索链能够理解这类省略上下文的追问。

1. 基础 RAG 为什么不理解追问

向量检索比较的是当前查询和文档向量之间的相似度。基础 Retriever 收到的只有：

它有哪些常见实现方式？

这句话没有“任务分解”四个字，语义信息不足。虽然聊天模型看到前文后可以理解“它”，但向量检索发生在回答生成之前，Retriever 此时还没有看到聊天历史。

解决办法不是把全部聊天记录直接拼到检索词中，而是增加一个问题改写步骤：

聊天历史：
用户：什么是任务分解？
助手：任务分解是把复杂任务拆成更小的子任务。

最新问题：
它有哪些常见实现方式？

改写结果：
任务分解有哪些常见的实现方式？

改写后的问题可以独立理解，再交给 Retriever 才能稳定召回任务分解相关文档。

下面的流程图同时画出了“无历史直接检索”和“有历史先改写再检索”两条路径，并标出了回答完成后的历史写回过程。

2. 上下文感知 RAG 的执行流程

完整流程分为两个阶段：

1. 生成检索问题：有聊天历史时，先让 LLM 把追问改写成独立问题。
2. 生成最终回答：Retriever 使用独立问题检索 Chroma，再让 LLM 根据召回文档、原始问题和聊天历史回答。

因此，有历史的问答通常会调用模型两次。第一次只负责改写问题，第二次才负责回答。没有历史时，create_history_aware_retriever() 会直接把原始问题交给 Retriever，不会额外调用模型改写。

这里还要区分两类数据：

• Chroma 知识库保存网页切分后的外部知识。
• 聊天历史保存某个会话中用户和模型已经说过的话。

create_history_aware_retriever() 不会把聊天记录写入 Chroma，也不会对聊天记录执行向量搜索。聊天历史只用于补全当前问题。

改写得到的独立问题只用于检索，不应替换用户原始问题。最终回答提示词仍然接收原始问题、聊天历史和召回文档，这样既能提高检索准确性，也不会因为改写措辞而改变用户真正想问的内容。

3. 准备运行环境

本篇继续使用上一篇构建的 Chroma 知识库，因此先确认已经执行：本地 Qwen3 服务仍由主虚拟环境启动：

source .venv/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --host 127.0.0.1 \
  --port 18080 \
  --chat-template-args '{"enable_thinking": false}'

RAG 代码使用独立环境：

source .venv_rag/bin/activate

先检查模型服务：

curl http://127.0.0.1:18080/v1/models

4. 先单独验证问题改写

代码:

"""演示：用聊天历史把省略上下文的追问改写成独立问题。

背景：
  多轮对话里用户常说「它有哪些方式？」，这句话单独拿去向量检索，
  「它」指什么模型不知道，检索结果会很差。

本脚本只做「改写」这一小步，不涉及检索和作答。
后续脚本（02～05）会在检索前自动调用相同的改写逻辑。

示例：
  历史：「什么是任务分解？」→「任务分解是...」
  追问：「它有哪些常见实现方式？」
  改写：「任务分解有哪些常见实现方式？」
"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from langchain_core.messages import AIMessage, HumanMessage

from rag.p09_context_aware_rag.common import build_rewrite_chain, create_llm

# 改写只需短输出，max_tokens=96 足够。
llm = create_llm(max_tokens=96)
rewrite_chain = build_rewrite_chain(llm)

# --- 模拟上一轮问答 ---
# HumanMessage / AIMessage 是 LangChain 的标准消息类型，
# 与 ChatOpenAI 接口一致，后续会存入 InMemory / SQLite 历史。
chat_history = [
    HumanMessage(content="什么是任务分解？"),
    AIMessage(content="任务分解是把复杂任务拆成更小、更容易处理的子任务。"),
]
# 追问里的「它」指代上文「任务分解」，缺少历史则语义不完整。
follow_up = "它有哪些常见实现方式？"

# 调用改写链：模板会把 chat_history 插入 MessagesPlaceholder 位置。
standalone_question = rewrite_chain.invoke(
    {"input": follow_up, "chat_history": chat_history}
)

print(f"原始追问：{follow_up}")
print(f"独立问题：{standalone_question}")

执行:

python rag/p08_rag_knowledge_base/01_build_agent_knowledge_base.py

在组合完整 RAG 之前，先只测试“聊天历史 + 最新问题 → 独立问题”。

MessagesPlaceholder 会把 chat_history 中的完整消息列表插入提示词。StrOutputParser() 则把模型返回的 AIMessage 转成普通字符串，便于直接交给 Retriever。

结果:

原始追问：它有哪些常见实现方式？
独立问题：任务分解有哪些常见的实现方式？

模型没有回答问题，只补全了缺失的主语，说明改写提示词达到了预期。

5. 构建上下文感知 RAG

问题改写验证通过后，再把它接到 Chroma Retriever 前面。

代码:

"""演示：完整的上下文感知 RAG（历史由代码手动传入，不自动保存）。

与 p08/03_basic_rag.py 的对比：
  p08：问题 → 检索 → 作答（单轮）
  p09：历史 + 追问 → 改写 → 检索 → 作答（多轮）

本脚本与 03 的区别：
  这里 chat_history 由代码手动构造和传入；
  03 使用 RunnableWithMessageHistory 自动维护历史。
"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from langchain_core.messages import AIMessage, HumanMessage

from rag.p09_context_aware_rag.common import (
    build_answer_chain,
    build_rewrite_chain,
    context_aware_ask,
    create_llm,
    create_retriever,
)

# --- 初始化组件（配置均来自 common.py）---
llm = create_llm()
retriever = create_retriever()
rewrite_chain = build_rewrite_chain(llm)
answer_chain = build_answer_chain(llm)

# --- 手动构造聊天历史 ---
# 实际应用中这些消息来自上一轮对话；这里硬编码模拟多轮场景。
chat_history = [
    HumanMessage(content="什么是任务分解？"),
    AIMessage(content="任务分解是把复杂任务拆成更小、更容易处理的子任务。"),
]
follow_up = "它有哪些常见实现方式？"

# --- 三步 RAG（详见 common.context_aware_ask）---
result = context_aware_ask(
    rewrite_chain,
    answer_chain,
    retriever,
    chat_history,
    follow_up,
)

print(f"改写后的问题：{result['standalone_question']}")
print(f"回答：{result['answer']}")
print("召回文档：")
for index, document in enumerate(result["context"], start=1):
    content = " ".join(document.page_content.split())
    print(f"[{index}] {content[:220]}")

代码中有三条关键链：

• history_aware_retriever：根据历史决定直接检索还是先改写问题。
• document_chain：把召回文档填入 {context}，再让模型回答。
• rag_chain：把检索和回答组合成完整 RAG。

Retriever 继续使用 similarity_score_threshold，阈值为 0.4。这个值来自当前知识库与本地 Qwen3 Embedding 的实际测试：已知问题的相关度高于阈值，无关问题低于阈值。更换知识库或 Embedding 模型后需要重新测量，不能把 0.4 当成通用值。

执行结果:

改写后的问题：任务分解有哪些常见的实现方式？
回答：任务分解的常见实现方式包括使用LLM进行简单提示、使用特定任务的指令，以及结合外部经典规划器（如PDDL）进行长期规划。此外，还有通过思维链（CoT）和思维树（ToT）等方法进行分解。
召回文档：
[1] Task decomposition can be done (1) by LLM with simple prompting like "Steps for XYZ.\n1.", "What are the subgoals for achieving XYZ?", (2) by using task-specific instructions; e.g. "Write a story outline." for writing a
[2] Component One: Planning# A complicated task usually involves many steps. An agent needs to know what they are and plan ahead. Task Decomposition# Chain of thought (CoT; Wei et al. 2022) has become a standard prompting te

追问中没有出现“任务分解”，但召回文档已经明确落在 Task Decomposition 小节，说明问题改写确实参与了检索。

6. 自动保存内存历史

上一个示例仍然由代码手动传入 chat_history。真实聊天程序通常只传当前输入和 session_id，历史应当由程序自动读取和保存。

LangChain 提供的 RunnableWithMessageHistory 可以包装现有 RAG 链。它不会改变 RAG 的检索逻辑，只负责在调用前读取历史，并在调用后写入本轮用户消息和模型回答。

代码:

"""演示：用 RunnableWithMessageHistory 自动保存内存中的聊天历史。

与 02 的区别：
  02 需要手动维护 chat_history 列表；
  本脚本每次 invoke 后，框架自动把「用户问题 + AI 回答」追加到历史。

RunnableWithMessageHistory 关键参数：
  input_messages_key="input"       用户输入字段名
  history_messages_key="chat_history"  历史注入到 rag_step 的这个字段
  output_messages_key="answer"     从返回值中取回答写入历史

注意：InMemoryChatMessageHistory 存在 Python 进程内存中，进程结束即丢失。
04/05 会改为 SQLite 持久化。
"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.history import RunnableWithMessageHistory

from rag.p09_context_aware_rag.common import (
    build_answer_chain,
    build_rewrite_chain,
    context_aware_ask,
    create_llm,
    create_retriever,
)

llm = create_llm()
retriever = create_retriever()
rewrite_chain = build_rewrite_chain(llm)
answer_chain = build_answer_chain(llm)


def rag_step(inputs: dict) -> dict:
    """单次 RAG 调用的入口，供 RunnableWithMessageHistory 包装。

    inputs 由框架组装：
      input：用户当前问题（我们传入）
      chat_history：该 session 之前的消息（框架自动注入）
    """
    return context_aware_ask(
        rewrite_chain,
        answer_chain,
        retriever,
        inputs.get("chat_history", []),
        inputs["input"],
    )


# RunnableLambda：把普通 Python 函数包装成 LangChain Runnable。
rag_runnable = RunnableLambda(rag_step)

# 全局字典：session_id → 该会话的消息列表。
history_store: dict[str, InMemoryChatMessageHistory] = {}


def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    """RunnableWithMessageHistory 的回调：按 session_id 取/建历史对象。"""
    if session_id not in history_store:
        history_store[session_id] = InMemoryChatMessageHistory()
    return history_store[session_id]


chatbot = RunnableWithMessageHistory(
    rag_runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

# config 里的 session_id 决定读写哪一份历史。
main_config = {"configurable": {"session_id": "rag-main"}}
other_config = {"configurable": {"session_id": "rag-other"}}

# --- 主会话第 1 轮：历史为空，相当于 p08 单轮 RAG ---
first = chatbot.invoke({"input": "什么是任务分解？"}, config=main_config)
print(f"主会话第 1 轮：{first['answer']}")

# --- 主会话第 2 轮：框架自动注入第 1 轮的历史 ---
# 「它」可被改写成「任务分解有哪些...」，检索质量显著提升。
second = chatbot.invoke(
    {"input": "它有哪些常见实现方式？"}, config=main_config
)
print(f"主会话第 2 轮：{second['answer']}")

# --- 新会话对比：没有历史，「它」无法被理解 ---
print(f"新会话调用前消息数：{len(get_session_history('rag-other').messages)}")
other = chatbot.invoke(
    {"input": "它有哪些常见实现方式？"}, config=other_config
)
print(f"新会话回答：{other['answer']}")
# 主会话 2 轮 × 2 条消息 = 4 条；新会话 1 轮 = 2 条。
print(f"主会话消息数：{len(get_session_history('rag-main').messages)}")
print(f"新会话消息数：{len(get_session_history('rag-other').messages)}")

三个参数必须与被包装链的字段一致：

• input_messages_key=”input”：当前用户消息的字段名。
• history_messages_key=”chat_history”：历史消息的字段名。
• output_messages_key=”answer”：需要保存为 AI 消息的输出字段名。

session_id 只用于选择哪一份历史记录，不会自动写入 Prompt，也不会暴露给模型。真正进入模型上下文的是该会话对应的 HumanMessage 和 AIMessage。

执行结果:

主会话第 1 轮：任务分解是将复杂任务拆解为多个较小、更易管理的子任务的过程。它有助于简化问题解决过程，使模型或代理能够逐步处理每个子任务。常见的方法包括使用提示词引导模型分解任务，或借助外部规划工具（如PDDL）进行规划。
主会话第 2 轮：任务分解的常见实现方式包括：1）使用LLM通过简单提示引导分解，如“步骤为XYZ”或“实现XYZ的子目标是什么”；2）使用特定任务的指令，如“写一个故事大纲”；3）借助外部经典规划器（如PDDL）进行长期规划，将问题转化为PDDL描述并生成计划；4）通过思维链（CoT）或思维树（ToT）等方法，探索多个可能的推理路径。
新会话调用前消息数：0

No relevant docs were retrieved using the relevance score threshold 0.4
新会话回答：我不知道。
主会话消息数：4
新会话消息数：2

同一个 rag-main 会话能理解第二轮的“它”。新的 rag-other 会话没有第一轮历史，因此无法确定“它”指什么，并按照提示词回答“我不知道”。

每轮对话会保存一条 HumanMessage 和一条 AIMessage，所以主会话两轮后有 4 条消息。

7. 内存历史的限制

InMemoryChatMessageHistory 使用 Python 字典保存消息，代码简单，适合学习和临时测试。但是进程结束后字典会消失，服务重启后不能继续之前的会话。

如果需要在新进程中恢复聊天记录，可以把 RunnableWithMessageHistory 的历史实现替换为 SQLChatMessageHistory。RAG 链本身不需要改变。

8. 使用 SQLite 保存第一轮对话

第一个脚本重新创建本文数据库，并写入第一轮问答。

代码:

"""演示：把上下文感知 RAG 的第一轮问答持久化到 SQLite。

与 03 的区别：
  03 用 InMemoryChatMessageHistory，进程结束历史消失；
  本脚本用 SQLChatMessageHistory，历史写入 chat_history.db 文件。

与 05 的配合：
  04 写入第一轮问答 → 05 在新进程中读取同一 session_id 的历史，继续追问。

每次运行会删除并重建 chat_history.db，保证测试可重复。
"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.history import RunnableWithMessageHistory

from rag.p09_context_aware_rag.common import (
    PROJECT_ROOT,
    SQLITE_PATH,
    build_answer_chain,
    build_rewrite_chain,
    context_aware_ask,
    create_llm,
    create_retriever,
)

# 04 和 05 必须使用相同的 SESSION_ID，才能读写同一会话。
SESSION_ID = "task-decomposition-demo"

# 重建 SQLite 文件，避免多次运行导致历史累积干扰测试。
SQLITE_PATH.parent.mkdir(parents=True, exist_ok=True)
SQLITE_PATH.unlink(missing_ok=True)

llm = create_llm()
retriever = create_retriever()
rewrite_chain = build_rewrite_chain(llm)
answer_chain = build_answer_chain(llm)


def rag_step(inputs: dict) -> dict:
    return context_aware_ask(
        rewrite_chain,
        answer_chain,
        retriever,
        inputs.get("chat_history", []),
        inputs["input"],
    )


def get_session_history(session_id: str) -> SQLChatMessageHistory:
    """从 SQLite 读写指定 session 的消息。

    table_name：消息存储的表名，04/05 必须一致。
    connection：sqlite:/// 前缀表示本地文件数据库。
    """
    return SQLChatMessageHistory(
        session_id=session_id,
        connection=f"sqlite:///{SQLITE_PATH}",
        table_name="rag_message_store",
    )


chatbot = RunnableWithMessageHistory(
    RunnableLambda(rag_step),
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

config = {"configurable": {"session_id": SESSION_ID}}
result = chatbot.invoke({"input": "什么是任务分解？"}, config=config)

print(f"第一轮回答：{result['answer']}")
# 一轮对话产生 2 条消息：HumanMessage + AIMessage。
print(f"SQLite 消息数：{len(get_session_history(SESSION_ID).messages)}")
print(f"数据库文件：{SQLITE_PATH.relative_to(PROJECT_ROOT)}")

运行结果:

第一轮回答：任务分解是将复杂任务拆解为多个较小、更易处理的子任务或步骤的过程。它有助于简化问题解决过程，使模型或代理能够逐步完成任务。常用的方法包括使用提示让大语言模型生成步骤，或借助外部规划工具（如PDDL）进行规划。
SQLite 消息数：2
数据库文件：data/p09_context_aware_rag/chat_history.db

数据库中保存了一条用户消息和一条模型消息。session_id 为 task-decomposition-demo，后续必须使用相同值才能恢复这段历史。

9. 在新进程中恢复历史

退出第一个脚本后，再单独运行恢复脚本。它不会重新写入第一轮问题，而是先读取 SQLite，再继续提出“它有哪些常见实现方式？”。

代码:

"""演示：在新进程中读取 SQLite 历史，继续省略上下文的追问。

使用场景：
  用户关闭浏览器后再打开，或后端进程重启，仍能从数据库恢复对话上下文。

运行顺序：
  1. 04_write_sqlite_history.py  → 写入第一轮问答到 SQLite
  2. 本脚本（可另开终端/进程）  → 读取历史，处理追问

关键：SESSION_ID 与 04 相同，才能读到上一轮的消息。
"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.history import RunnableWithMessageHistory

from rag.p09_context_aware_rag.common import (
    SQLITE_PATH,
    build_answer_chain,
    build_rewrite_chain,
    context_aware_ask,
    create_llm,
    create_retriever,
)

SESSION_ID = "task-decomposition-demo"

if not SQLITE_PATH.exists():
    raise RuntimeError("历史数据库不存在，请先运行 04_write_sqlite_history.py。")

llm = create_llm()
retriever = create_retriever()
rewrite_chain = build_rewrite_chain(llm)
answer_chain = build_answer_chain(llm)


def rag_step(inputs: dict) -> dict:
    return context_aware_ask(
        rewrite_chain,
        answer_chain,
        retriever,
        inputs.get("chat_history", []),
        inputs["input"],
    )


def get_session_history(session_id: str) -> SQLChatMessageHistory:
    # 连接参数与 04 完全一致，读写同一个 db 文件和表。
    return SQLChatMessageHistory(
        session_id=session_id,
        connection=f"sqlite:///{SQLITE_PATH}",
        table_name="rag_message_store",
    )


chatbot = RunnableWithMessageHistory(
    RunnableLambda(rag_step),
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

# 启动前先检查：04 是否已写入历史。
history = get_session_history(SESSION_ID)
if not history.messages:
    raise RuntimeError("目标会话没有历史，请先运行 04_write_sqlite_history.py。")

print(f"恢复前消息数：{len(history.messages)}")

# 追问「它有哪些...」；框架从 SQLite 加载历史，改写时可理解「它」= 任务分解。
config = {"configurable": {"session_id": SESSION_ID}}
result = chatbot.invoke(
    {"input": "它有哪些常见实现方式？"},
    config=config,
)

print(f"改写后的问题：{result['standalone_question']}")
print(f"恢复后的追问回答：{result['answer']}")
# 恢复前 2 条 + 本轮 2 条 = 4 条。
print(f"回答后消息数：{len(get_session_history(SESSION_ID).messages)}")

结果:

恢复前消息数：2
恢复后的追问回答：任务分解的常见实现方式包括：1）通过提示让大语言模型生成步骤，如“步骤为XYZ”；2）使用特定任务的指令，如“写一个故事大纲”；3）结合外部经典规划器（如PDDL）进行长期规划。
回答后消息数：4

这是两个独立 Python 进程。恢复脚本启动时已经读取到 2 条消息，回答追问后变成 4 条，证明 SQLite 历史可以跨进程恢复。

10. Chroma 与 SQLite 分别保存什么

这个案例同时出现 Chroma 和 SQLite，但它们承担的职责完全不同。

Chroma 中保存的是从 Agent 博客提取的文档块及其向量。无论哪个用户提问，知识库内容都相同。

SQLite 中保存的是某个 session_id 的用户消息和模型回答。它用于补全追问中的省略信息，不作为知识文档参与相似度检索。

因此，本篇实现的是“持久化的会话历史”，仍然属于会话范围的短期记忆。它不是把历史消息向量化后长期检索，也不等于向量数据库长期记忆。