pytest tests/test_retrieval_and_prompt.py tests/test_memory_history.py -q

用户：入职流程有哪些步骤？
AI：入职流程包括：1. 提交材料 2. 签订合同 3. 部门审批 4. 领取工位...

用户：那审批需要多久？          ← 这是一个追问
      ↑ "那审批" 指的是什么审批？不结合历史无法理解

"那审批呢" → "那" 指代的是 入职流程中的审批步骤
"费用呢"   → "费用" 需要结合上文确定是 入职费用 还是 培训费用
"还有吗"   → 需要结合上文确定在问什么"还有"

def rewrite_query_if_needed(query, history_messages, should_rewrite):
    if not should_rewrite or not history_messages:
        return query  # 直接返回原问题

# qa_core/intent/classifier.py — 触发 requires_rewrite 的场景

# 场景 1：追问规则命中
if history and (FOLLOW_UP_HINTS.search(query.strip()) or len(query.strip()) <= 8):
    return IntentResult(
        intent="FOLLOW_UP",
        requires_rewrite=True,  # ← 标记需要改写
        ...
    )

# 场景 2：LLM 判断需要改写
return IntentResult(
    intent=decision.intent,
    requires_rewrite=decision.requires_rewrite or decision.intent == "FOLLOW_UP",
    ...
)

# qa_core/pipeline/rewrite.py
from langchain_core.messages import HumanMessage, SystemMessage

def rewrite_query_if_needed(query, history_messages, should_rewrite):
    if not should_rewrite or not history_messages:
        return query

    # 只取最近 8 条历史 — 聚焦当前话题
    history_text = format_messages(history_messages[-8:])

    llm = get_chat_model(streaming=False)  # 非流式，需要完整结果
    response = llm.invoke([
        SystemMessage(content=REWRITE_SYSTEM_PROMPT),
        HumanMessage(
            content=f"对话历史：\n{history_text}\n\n"
                    f"当前问题：{query}\n\n"
                    f"改写后的检索问题："
        ),
    ])

    rewritten = str(response.content).strip()
    if not rewritten:
        raise RuntimeError("查询改写返回空结果，无法生成独立检索问题。")
    return rewritten

REWRITE_SYSTEM_PROMPT = """
你是一个查询改写助手。你的任务是把用户的追问问题改写成独立的检索问题。

规则：
1. 如果问题已经可以独立理解，直接返回原问题
2. 如果问题包含代词(那、这个、它、他们)或依赖上文，请结合对话历史补全
3. 补全后的检索问题应该包含具体的实体和动作
4. 只返回改写后的问题，不要加任何解释

示例：
对话历史：
用户：入职流程有哪些步骤
AI：入职流程包括提交材料、签订合同、部门审批...

当前问题：那审批呢
改写后的检索问题：入职流程中的审批步骤是什么

对话历史：
用户：API 限流怎么处理
AI：API 限流可以通过升级套餐或联系技术支持处理...

当前问题：升级套餐多少钱
改写后的检索问题：API 限流相关套餐的升级费用是多少
"""

history_text = format_messages(history_messages[-8:])  # 只取最近 8 条

if not should_rewrite or not history_messages:
    return query  # 直接返回原问题，不做任何修改

用户问：新人入职当天要带什么
知识库写：入职报到需提交的材料清单

原始问题："新人入职当天要带什么"

查询变体：
  1. "新人入职当天要带什么"        ← 原问题
  2. "入职报到需要提交哪些材料"    ← 正式表述
  3. "入职当天需要准备什么文件"    ← 另一种问法
  4. "入职需要携带的证件和材料"    ← 更具体的表述

# qa_core/pipeline/query_variants.py
def generate_query_variants(query: str, *, enabled: bool) -> list[str]:
    """返回"原问题 + 少量同义检索表达"。

    该函数是召回增强，不是问题改写。返回值同时传给 FAQ 和文档集合的 search_many()。

    生成策略分 3 层：
    1. enabled=False 或配置关闭 → 直接返回 [原问题]
    2. 短结构化问题 → 跳过扩展(已经足够清晰，扩展无收益)
    3. 本地启发式命中 → 用关键词替换规则生成变体(快、稳定、不用 LLM)
    4. 以上都不满足 → 调用 LLM 结构化输出生成变体
    """
    settings = get_settings()
    cleaned = query.strip()
    if not enabled or not cleaned or settings.retrieval_variant_max <= 0:
        return [cleaned]

    if _looks_like_short_structured_question(cleaned):
        return [cleaned]

    heuristic_variants = _heuristic_variants(cleaned, settings.retrieval_variant_max)
    if len(heuristic_variants) > 1:
        return heuristic_variants

    variants = [cleaned]
    model = get_chat_model(streaming=False).with_structured_output(QueryVariants)
    result = model.invoke([
        SystemMessage(content=QUERY_VARIANT_SYSTEM_PROMPT),
        HumanMessage(content=f"原问题：{cleaned}\n最多生成 {settings.retrieval_variant_max} 条检索表达。"),
    ])
    for item in result.queries:
        candidate = str(item).strip()
        if candidate and candidate not in variants:
            variants.append(candidate)
        if len(variants) >= settings.retrieval_variant_max + 1:
            break
    return variants

def _heuristic_variants(query: str, max_extra: int) -> list[str]:
    """为高频业务知识/FAQ 说法生成低延迟的本地变体。"""
    variants = [query]
    normalized = query.lower()

    def add(candidate: str) -> None:
        """在保持顺序和上限的前提下，追加非空变体。"""
        candidate = candidate.strip()
        if candidate and candidate not in variants and len(variants) < max_extra + 1:
            variants.append(candidate)

    if "安装" in query or "失败" in query or "报错" in query:
        add(query.replace("失败", "报错"))
        add(query.replace("报错", "失败"))
    if "流程" in query:
        add(query.replace("流程", "SOP"))
        add(query.replace("流程", "处理步骤"))
    if "发票" in query:
        add(query.replace("发票", "开票"))
        add(query.replace("发票", "账单"))
    if "开票" in query:
        add(query.replace("开票", "发票"))
    if "告警" in query:
        add(query.replace("告警", "报警"))
        add(query.replace("告警", "异常"))
    if "webhook" in normalized:
        add(query.replace("webhook", "回调").replace("Webhook", "回调"))
        add(query.replace("Webhook", "Webhook 回调").replace("webhook", "Webhook 回调"))
    if "资料" in query:
        add(query.replace("资料", "材料"))
        add(query.replace("资料", "记录"))
    if "材料" in query:
        add(query.replace("材料", "资料"))
    if "流程" in query and "怎么走" in query:
        add(query.replace("怎么走", "有哪些步骤"))
        add(query.replace("流程", "办理流程"))
    if "怎么排查" in query:
        add(query.replace("怎么排查", "如何处理"))
        add(query.replace("怎么排查", "处理步骤"))
    if "能不能" in query:
        add(query.replace("能不能", "是否可以"))
    if "可以吗" in query:
        add(query.replace("可以吗", "是否可以"))
    return variants

# qa_core/pipeline/query_variants.py
def _looks_like_short_structured_question(query: str) -> bool:
    """判断一个问题是否已经足够结构化，不值得再调用 LLM 扩展检索表达。

    使用场景：
    - 问题已经短且清晰；
    - 本地启发式没有命中额外高价值同义词；
    - 继续调用 LLM 只会增加尾延迟，而不会明显提高召回。

    这类问题在当前项目里主要是：
    - 流程/资料/材料/SOP 类标准问法；
    - "怎么排查""怎么处理""能不能""可以吗"这类业务 FAQ 句式；
    - 已经带有明确业务主语的短句。
    """
    compact = query.strip()
    if not compact or len(compact) > 24:
        return False
    return any(
        marker in compact
        for marker in (
            "怎么走",
            "资料",
            "材料",
            "怎么排查",
            "怎么处理",
            "需要哪些",
            "能不能",
            "可以吗",
            "是什么",
            "要看什么",
        )
    )

    # 本地启发式未命中时，使用 Pydantic 结构化输出让 LLM 生成变体
    # 注意：该逻辑是 generate_query_variants() 内部的内联实现，不是独立函数
    if not heuristic_hit:
        model = get_chat_model(streaming=False).with_structured_output(QueryVariants)
        result = model.invoke([
            SystemMessage(content=QUERY_VARIANT_SYSTEM_PROMPT),
            HumanMessage(content=f"原问题：{cleaned}\n最多生成 {settings.retrieval_variant_max} 条检索表达。"),
        ])
        for item in result.queries:
            ...  # 去重、上限控制后追加到 variants 列表

# RetrievalPlan 是 frozen dataclass，use_query_variants 在 build_retrieval_plan() 构造时设定
# qa_core/retrieval/strategy.py
plan = RetrievalPlan(
    ...
    use_query_variants=intent.intent in {"KNOWLEDGE_QUERY", "FOLLOW_UP"},
    ...
)

# qa_core/memory/history.py — ChatHistoryStore 类
def get_context_messages(self, session_id: str):
    # 1. 读取已有会话摘要(从 MySQL 摘要表加载)
    summary = self.get_summary(session_id)

    # 2. 读取最近 8 条消息(由 settings.history_recent_messages 控制)
    recent = self.get_messages(session_id, limit=self.settings.history_recent_messages)

    # 3. 拼接：摘要作为 SystemMessage 放在前面
    if summary:
        return [SystemMessage(content=f"历史摘要：{summary}")] + recent
    return recent

# qa_core/memory/history.py — ChatHistoryStore 类

def get_summary(self, session_id: str) -> str:
    """从 MySQL 摘要表加载当前会话的对话摘要。"""
    if not self.settings.history_summary_enabled:
        return ""
    self.ensure_summary_table()
    with self.engine.begin() as conn:
        row = conn.execute(
            text(f"SELECT summary FROM {self.settings.chat_summary_table_name} "
                 f"WHERE session_id=:session_id"),
            {"session_id": session_id},
        ).fetchone()
    return str(row[0]) if row and row[0] else ""

def refresh_summary_if_needed(self, session_id: str) -> None:
    """当消息数超过阈值时，用 LLM 生成/更新摘要并写入 MySQL。"""
    if not self.settings.history_summary_enabled:
        return
    messages = self.get_messages(session_id)
    if len(messages) < self.settings.history_summary_after_messages:
        return

    # 只总结"较早历史"，最近 N 条继续原文保留
    older_messages = messages[: -self.settings.history_recent_messages]
    if not older_messages:
        return

    current_summary = self.get_summary(session_id) or "无"
    history_text = format_messages(older_messages)
    llm = get_chat_model(streaming=False)
    response = llm.invoke([
        SystemMessage(content=HISTORY_SUMMARY_SYSTEM_PROMPT),
        HumanMessage(content=(
            f"已有摘要：\n{current_summary}\n\n"
            f"新增历史：\n{history_text}\n\n"
            f"请输出不超过 {self.settings.history_summary_max_chars} 字的更新摘要。"
        )),
    ])
    summary = str(response.content).strip()[: self.settings.history_summary_max_chars]
    if summary:
        self.save_summary(session_id, summary)

整个会话的上下文管理策略：

时间轴 ──────────────────────────────────────────────>

轮次 1-8：全部保留在最近消息中
轮次 9-20：前 14 轮压缩为一段摘要 + 最近 8 轮完整保留
轮次 15+：摘要逐步更新 + 始终保留最近 8 轮

发给 LLM 的内容：
┌────────────────────┐
│ 会话摘要(如果有)  │  ← 200-1200 字符
├────────────────────┤
│ 最近 8 条消息      │  ← 完整对话文本
└────────────────────┘

[System Prompt]
   ↓
[会话摘要] ← 如果有(200-1200 字符，概括前十几轮的要点)
   ↓
[最近 8 条消息] ← 完整原文(保留措辞细节和指代关系)
   ↓
[当前问题] ← 用户刚发的问题

# qa_core/memory/history.py — ChatHistoryStore 类

def get_context_messages(self, session_id: str) -> list[BaseMessage]:
    """对应上图右半部分：拼给 LLM 的最终上下文。"""
    # 读取已持久化的最近 N 条完整消息(默认 8 条)
    recent = self.get_messages(session_id, limit=self.settings.history_recent_messages)
    # 从 MySQL 摘要表读取已有摘要
    summary = self.get_summary(session_id)
    if summary:
        # 摘要作为 SystemMessage 放在前面，让 rewrite 和 answer prompt
        # 都能把它当作背景事实读取
        return [SystemMessage(content=f"历史摘要：{summary}")] + recent
    return recent

def refresh_summary_if_needed(self, session_id: str) -> None:
    """对应上图左半部分：第 15+ 轮时触发摘要压缩。"""
    if not self.settings.history_summary_enabled:
        return
    messages = self.get_messages(session_id)
    # 判断是否需要压缩：消息数 < 阈值则跳过
    if len(messages) < self.settings.history_summary_after_messages:
        return

    # 只压缩"较早历史"，最近 N 条继续原文保留
    older = messages[: -self.settings.history_recent_messages]
    if not older:
        return

    current = self.get_summary(session_id) or "无"
    llm = get_chat_model(streaming=False)       # 非流式，后台执行
    response = llm.invoke([
        SystemMessage(content=HISTORY_SUMMARY_SYSTEM_PROMPT),
        HumanMessage(content=(
            f"已有摘要：\n{current}\n\n"
            f"新增历史：\n{format_messages(older)}\n\n"
            f"请输出不超过 {self.settings.history_summary_max_chars} 字的更新摘要。"
        )),
    ])
    summary = str(response.content).strip()[: self.settings.history_summary_max_chars]
    if summary:
        self.save_summary(session_id, summary)   # upsert 到 MySQL 摘要表

# qa_core/retrieval/store.py — MilvusHybridStore 类
def search_many(
    self,
    queries: list[str],
    *,
    k: int,
    source_filter: str | None,
    kb_version: str | None = None,
    valid_sources: list[str] | None = None,
    data_scope: DataScope | None = None,
    source_type: Literal["faq", "doc"],
    rerank: bool = True,
) -> RetrievalResult:
    """对多个查询变体分别检索，合并重复 chunk 后统一重排。

    核心流程：
    1. 清洗查询变体 → 逐个变体做轻量检索(先不 rerank)
    2. 按 chunk_id/faq_id 合并重复命中，只保留最高分
    3. 按分数排序 → 取候选上限 → CrossEncoder 统一重排
    """
    merged: dict[str, RetrievalHit] = {}
    for q in normalize_queries(queries):
        result = self.search(q, k=k, ..., rerank=False)
        merge_hits_by_document(merged, result.hits)   # 原地更新 dict

    hits = sort_hits_by_score(merged.values())
    if rerank and hits:
        hits = self._rerank(queries[0], hits)         # 原问题作为 rerank query
    return RetrievalResult(hits=hits[:k], ...)

def rewrite_query_if_needed(query: str, history_messages, should_rewrite: bool) -> str:
    if not should_rewrite or not history_messages:
        return query

    history_text = format_messages(history_messages[-8:])
    llm = get_chat_model(streaming=False)
    response = llm.invoke([
        SystemMessage(content=REWRITE_SYSTEM_PROMPT),
        HumanMessage(content=f"对话历史：\n{history_text}\n\n当前问题：{query}\n\n改写后的检索问题："),
    ])
    rewritten = str(response.content).strip()
    if not rewritten:
        raise RuntimeError("查询改写返回空结果，无法生成独立检索问题。")
    return rewritten

messages = [
    SystemMessage(content="把用户追问改写成独立检索问题，不要回答问题。"),
    HumanMessage(content=f"历史：{format_messages(history)}\n当前追问：{query}"),
]
rewritten = get_chat_model(streaming=False).invoke(messages).content.strip()

def generate_query_variants(query: str, *, enabled: bool) -> list[str]:
    settings = get_settings()
    cleaned = query.strip()

    # 第 1 层：未启用、空问题、或配置上限为 0，只返回原问题
    if not enabled or not cleaned or settings.retrieval_variant_max <= 0:
        return [cleaned]

    # 第 2 层：短结构化问题不扩展，避免多余 LLM 调用和语义漂移
    if _looks_like_short_structured_question(cleaned):
        return [cleaned]

    # 第 3 层：本地启发式优先，覆盖流程/SOP、发票/开票、告警/报警等高频同义词
    heuristic_variants = _heuristic_variants(cleaned, settings.retrieval_variant_max)
    if len(heuristic_variants) > 1:
        return heuristic_variants

    # 第 4 层：规则没覆盖的新表达，再调用 LLM 结构化生成
    variants = [cleaned]
    model = get_chat_model(streaming=False).with_structured_output(QueryVariants)
    result = model.invoke([...])
    for item in result.queries:
        candidate = str(item).strip()
        if candidate and candidate not in variants:
            variants.append(candidate)
        if len(variants) >= settings.retrieval_variant_max + 1:
            break
    return variants

context_messages = history_store.get_context_messages(session_id)
history_store.refresh_summary_if_needed(session_id)

python -m pytest tests/test_memory_history.py tests/test_retrieval_and_prompt.py -q