客服 RAG 不好用的常见原因，不是检索算法不行，而是知识库里根本没有能用的知识。历史聊天记录里其实藏着大量真实问答对，但直接灌进去会带来噪声、过期答案和互相矛盾的版本。这篇讲怎么把它洗干净再入库。

一、从聊天里挖 QA 对

不是每条消息都能变成知识。可用的是成对出现的问答：客户问、客服答，且答完之后客户没有继续追问（追问通常意味着答得不对）。用这三条筛：

• 问句有明确问题词（怎么、能不能、多少钱、多久）

• 答句来自客服侧（非客户发言），且长度大于 10 字

• 答完 30 分钟内客户没有继续发同类问题

筛出来的对子存成结构化记录：问题、答案、来源会话、时间。这比让运营凭空写 FAQ 快得多，也更贴近客户真实说法。

二、清洗：去掉过期和矛盾的

聊天记录最大的问题是时效。三年前的"活动价 199"今天还挂在知识库里，客户一问就会答错。所以每条 QA 必须带有效期标记：涉及价格、活动、库存的默认 30 天，涉及流程和规则的长一些。

矛盾处理靠聚类：把语义相近的问题归到一组，同组里取时间最新的答案作为准，其余标为历史版本不参与检索。不做这一步，知识库越大答案越乱。数据拉取通道见 Eyun 消息接口，按会话维度批量取比逐条拉高效。

三、切分与增量更新

一条 QA 不要太长——超过 300 字就按要点拆成多条，检索时命中率高。入库时同时存原始问题和几个同义改写（"怎么退"和"退款流程是什么"指向同一条），召回率会明显好一些。

更新用增量：每天新增会话跑一遍抽取，只处理上次水位之后的数据。已入库的 QA 如果来源会话被撤回或删除，要把对应条目下架——知识库不做下架，迟早会答出一条客户已经否认过的信息。

知识库构建对照

环节	做法	关键点	常见坑
挖掘	问+答+无追问	答方是客服侧	把闲聊当知识
清洗	时效标记+聚类去重	同组留最新	过期价格误导
切分	单条 ≤300 字	存同义改写	一条太长召不回
更新	按水位增量+下架	撤回要同步删	只加不删越答越错

QA 抽取与入库实现

import time, collections

class ChatKnowledgeBuilder:
    QA_MAX_LEN = 300
    Q_WORDS = ("怎么", "能不能", "多少钱", "多久", "可以吗", "是否")
    PRICE_WORDS = ("价格", "活动", "优惠", "库存")
    VALID_DAYS = {"price": 30, "rule": 365}

    def __init__(self, api, store):
        self.api = api
        self.store = store

    def build(self, session_id):
        since = self.store.watermark(session_id)
        msgs = self.api.get("/message/history", params={
            "session_id": session_id,
            "since": since}).json()["items"]

        pairs = []
        for i, m in enumerate(msgs):
            if not self._is_question(m): continue
            # 找后续第一条客服回复
            answer = self._find_answer(msgs, i)
            if not answer: continue
            # 30 分钟内无追问才算有效
            if self._followed_up(msgs, i, answer): continue
            pairs.append((m, answer))

        for q, a in pairs:
            self._upsert_qa(session_id, q, a)

        self.store.set_watermark(session_id,
                                 msgs[-1]["ts"] if msgs else since)
        return len(pairs)

    def _is_question(self, m):
        return any(w in m["content"] for w in self.Q_WORDS)

    def _find_answer(self, msgs, idx):
        q = msgs[idx]
        for m in msgs[idx + 1:]:
            if m["from_me"] and len(m["content"]) >= 10:
                if m["ts"] - q["ts"] <= 3600:
                    return m
                break
        return None

    def _followed_up(self, msgs, q_idx, answer):
        for m in msgs[q_idx + 1:]:
            if m["ts"] > answer["ts"] + 1800: break
            if not m["from_me"] and self._is_question(m):
                return True
        return False

    def _upsert_qa(self, session_id, q, a):
        question = q["content"][:120]
        answer = a["content"]
        # 超长按句拆
        chunks = ([answer] if len(answer) <= self.QA_MAX_LEN
                  else [s for s in answer.split("。") if s][:3])

        # 时效：涉价格/活动给短期限
        kind = "price" if any(w in answer for w in self.PRICE_WORDS) else "rule"
        expire = int(time.time()) + self.VALID_DAYS[kind] * 86400

        group_key = self.store.cluster_key(question)   # 语义聚类键
        existing = self.store.find_latest(group_key)
        if existing and a["ts"] < existing["ts"]:
            # 旧版本，标历史不参与检索
            self.store.save_qa(session_id, question, answer,
                               groups=chunks, expire=expire,
                               state="history")
            return

        self.store.save_qa(session_id, question, answer,
                           groups=chunks, expire=expire, state="active",
                           source_msg_id=a["msg_id"])
        # 同组旧答案降级
        self.store.demote_older(group_key, keep_ts=a["ts"])

    def revoke_source(self, msg_id):
        # 来源被撤回 → 知识条目下架
        self.store.retire_by_source(msg_id)

落地建议

聊天记录变知识库，重点在"选得准"和"删得掉"。只留客户问、客服答、且客户没追问的三段式对子；每条 QA 带有效期，价格类 30 天到期自动失效；相近问题聚成一组合并，只留最新答案当准，旧的降级为历史版本。增量更新按水位跑，来源消息撤回时要把知识条目一起下架——只往里加不往外拿，这个库迟早会答错。Eyun 平台负责把历史会话稳定地吐出来，怎么切成知识、留哪条、删哪条，由你的清洗规则和调度策略决定。