微信机器人接入知识库 RAG，做企业专属智能问答

前言

企业内部沉淀的文档、FAQ、产品手册，往往散落在飞书、Confluence、本地目录里，员工每次遇到问题都要翻半天，或者直接在微信群问同事——同样的问题反复占用人力。

把这些文档接入 RAG（检索增强生成）管道，再把入口放到微信，是一个低摩擦的落地路径：员工不需要学新工具，直接在熟悉的微信私聊或群聊里提问，机器人秒回。本文从 RAG 的基本原理出发，逐步拆解如何用 Python 把文档库、向量检索、大模型回答、微信消息收发串成一条完整链路，附完整可运行示例。

一、RAG 基本原理与企业知识问答的适配逻辑

1.1 RAG 是什么

RAG 全称 Retrieval-Augmented Generation，核心思路是：先检索、再生成。

普通 LLM 直接用模型参数里的知识回答，无法感知你的私有文档，也无法避免"幻觉"。RAG 在生成前插入一个检索步骤——把用户问题向量化，到文档向量库里找最相关的段落，把这些段落塞进 Prompt，让模型基于真实文档回答。

用户提问
  │
  ▼
问题向量化（Embedding Model）
  │
  ▼
向量库检索（Top-K 相关段落）
  │
  ▼
组装 Prompt（System + 检索段落 + 用户问题）
  │
  ▼
LLM 生成回答
  │
  ▼
返回答案（可附来源段落）

1.2 为什么微信是好入口

• 员工日均打开微信频率远高于内部系统；
• 群聊场景下，@机器人提问，答案对所有人可见，减少重复提问；
• 无需额外 App、无需培训，开箱即用。

二、整体架构设计

┌──────────────────────────────────────────────────────┐
│  离线阶段（文档入库）                                  │
│  企业文档 → 分块(Chunking) → Embedding → 向量库       │
└──────────────────────────────────────────────────────┘
                        │
                        │ 索引构建完成
                        ▼
┌──────────────────────────────────────────────────────┐
│  在线阶段（实时问答）                                  │
│                                                      │
│  微信消息 → 回调服务 → 问题理解 → 向量检索             │
│                          │                           │
│                    Prompt 组装 → LLM 生成 → 微信回复  │
└──────────────────────────────────────────────────────┘

核心组件：

三、文档入库：分块与向量化

3.1 文档加载与分块

企业文档通常是 PDF、Word 或 Markdown。以下示例统一处理成纯文本后按字数分块：

python# doc_indexer.py
# 代码为示例，具体依赖版本以实际环境为准
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_text(file_path: str) -> str:
    """简单示例：仅处理 .txt / .md，其他格式可用对应解析库"""
    return Path(file_path).read_text(encoding="utf-8")

def split_text(text: str, chunk_size=500, chunk_overlap=50) -> list[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", " "]
    )
    return splitter.split_text(text)

分块策略要点：

• chunk_size 推荐 400～600 汉字，太短上下文不足，太长占 Token 多；
• chunk_overlap 保留 10% 左右重叠，避免关键信息被截断在块边界；
• 如果文档有明确章节结构，优先按标题切块，语义更完整；
• 对于表格、代码片段等结构化内容，建议单独处理，不与正文混切，否则向量化后语义噪声较大；
• 分块完成后，建议打印几个样本人工检查边界是否合理，发现问题及时调整分隔符优先级。

3.2 向量化入库

python# 以 Chroma + 本地 Embedding 为例
# 注意：实际 Embedding 模型选型请参考官方文档

import chromadb
from chromadb.utils import embedding_functions

EMBED_MODEL = "你的Embedding模型名称"  # 以实际部署为准

def build_index(chunks: list[str], collection_name: str = "kb"):
    client = chromadb.PersistentClient(path="./chroma_db")
    ef = embedding_functions.SentenceTransformerEmbeddingFunction(
        model_name=EMBED_MODEL
    )
    collection = client.get_or_create_collection(
        name=collection_name,
        embedding_function=ef
    )
    ids = [f"chunk_{i}" for i in range(len(chunks))]
    collection.add(documents=chunks, ids=ids)
    print(f"入库完成，共 {len(chunks)} 个分块")
    return collection

入库是一次性操作，文档更新时重新运行即可。生产环境建议给每个分块加元数据（来源文件名、页码），方便回答时附上出处。此外，若文档量较大（万级分块以上），建议切换到 Milvus 或 Weaviate 等支持分布式的向量库，Chroma 在大规模场景下查询延迟会明显上升。

四、在线检索与 Prompt 组装

4.1 检索相关段落

pythondef retrieve(query: str, collection, top_k: int = 5) -> list[str]:
    results = collection.query(
        query_texts=[query],
        n_results=top_k
    )
    # results["documents"] 是 [[chunk1, chunk2, ...]]
    return results["documents"][0]

top_k 一般取 3～6，越多提供的上下文越丰富，但也占更多 Token。实践中可以先用相似度分数做一层过滤，低于阈值的段落即使在 Top-K 内也不传给 LLM，避免无关内容干扰回答。

4.2 组装 Prompt 并调用 LLM

pythondef build_prompt(query: str, contexts: list[str]) -> str:
    context_text = "\n\n".join(
        [f"【参考{i+1}】\n{c}" for i, c in enumerate(contexts)]
    )
    return f"""你是企业内部知识库助手，只根据以下参考资料回答问题，若资料中没有相关信息请明确说"暂无相关记录"，不要编造。

{context_text}

用户问题：{query}

请给出简洁、准确的回答："""

# 调用 LLM（以 OpenAI 风格接口为例，实际以所用模型 SDK 为准）
import openai

def ask_llm(prompt: str, model: str = "你的模型名称") -> str:
    client = openai.OpenAI(
        api_key="你的LLM_API_KEY",  # 以官方文档为准
        base_url="https://你的LLM接口地址"
    )
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,   # 知识问答偏低温，减少发散
        max_tokens=1024
    )
    return resp.choices[0].message.content.strip()

Prompt 设计注意事项：明确要求模型"只依据提供的参考资料"，是防止幻觉的关键约束。如果企业文档覆盖不全，宁可让模型回答"暂无记录"，也不要让它凭空发挥，否则会损害用户信任。

五、微信消息收发：接入 HTTP 接口

知识问答逻辑写好后，需要把微信消息作为触发器。微信个人号的消息收发可以通过 HTTP REST 接口实现：WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可，以官方文档为准。

5.1 接口基础配置

python# wechat_client.py
# 代码为示例，具体接口/字段以官方文档为准

BASE  = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"              # 扫码登录后获得
HEADERS = {"token": TOKEN}       # 鉴权字段名以官方文档为准

import requests

def send_text(to_wxid: str, content: str) -> dict:
    """发送文本消息"""
    url = f"{BASE}/message/postText"
    body = {
        "appId": APPID,
        "toWxid": to_wxid,
        "content": content
    }
    resp = requests.post(url, json=body, headers=HEADERS, timeout=10)
    return resp.json()

返回结构示例：{"ret": 200, "msg": "操作成功", "data": {...}}，ret == 200 表示成功。

5.2 配置消息回调

接收微信消息需要先用 setCallback 接口注册一个公网可访问的回调地址，平台会把消息 POST 到该地址：

pythondef set_callback(callback_url: str) -> dict:
    url = f"{BASE}/login/setCallback"
    body = {"appId": APPID, "callbackUrl": callback_url}
    resp = requests.post(url, json=body, headers=HEADERS, timeout=10)
    return resp.json()

回调 payload 示例（字段以官方文档为准）：

json{
  "appId": "你的appId",
  "fromWxid": "发件人微信ID",
  "toWxid": "收件人微信ID",
  "type": 1,
  "content": "用户的提问内容",
  "msgId": "消息ID",
  "createTime": 1700000000
}

5.3 回调服务整合 RAG

python# callback_server.py
# 依赖：pip install flask chromadb sentence-transformers openai requests
# 代码为示例，具体接口/字段以官方文档为准

from flask import Flask, request, jsonify
import chromadb
from chromadb.utils import embedding_functions

app = Flask(__name__)

EMBED_MODEL = "你的Embedding模型名称"
ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name=EMBED_MODEL)
client_db = chromadb.PersistentClient(path="./chroma_db")
collection = client_db.get_collection(name="kb", embedding_function=ef)

@app.route("/wechat/callback", methods=["POST"])
def wechat_callback():
    data = request.json or {}
    msg_type = data.get("type")
    content  = data.get("content", "").strip()
    from_id  = data.get("fromWxid", "")

    # 只处理文本消息（type==1 为示例，以官方文档为准）
    if msg_type != 1 or not content:
        return jsonify({"code": 200})

    # RAG 流程
    contexts = retrieve(content, collection)
    prompt   = build_prompt(content, contexts)
    answer   = ask_llm(prompt)

    # 回复
    send_text(to_wxid=from_id, content=answer)
    return jsonify({"code": 200})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

回调服务必须公网可访问（可用 Nginx 反代或内网穿透工具），且响应需在合理时间内返回 200，否则平台会重发。如果 RAG 检索 + LLM 生成耗时较长，建议采用异步方案：回调接口立即返回 200，把实际处理任务投入消息队列，处理完成后再调用发送接口主动推送答案。

六、群聊场景：@触发与消息过滤

在群聊中，通常只希望 @机器人 才触发 RAG，避免处理所有群消息。回调字段里会携带 atWxidList 或消息内容包含 @昵称，可据此过滤：

pythondef is_at_me(data: dict, my_wxid: str) -> bool:
    """判断消息是否 @了机器人"""
    # 字段名以官方文档为准
    at_list = data.get("atWxidList", [])
    if my_wxid in at_list:
        return True
    # 部分场景通过内容判断
    content = data.get("content", "")
    return f"@" in content  # 可结合昵称做精确匹配

@app.route("/wechat/callback", methods=["POST"])
def wechat_callback_group():
    data    = request.json or {}
    my_id   = APPID   # 或单独维护机器人微信ID
    content = data.get("content", "").strip()
    from_id = data.get("fromWxid", "")
    room_id = data.get("roomId", "")   # 群ID，字段以官方文档为准

    # 群消息需 @触发
    if room_id and not is_at_me(data, my_id):
        return jsonify({"code": 200})

    # 去掉 @部分，提取纯问题
    clean_query = content.replace(f"@机器人昵称", "").strip()

    contexts = retrieve(clean_query, collection)
    prompt   = build_prompt(clean_query, contexts)
    answer   = ask_llm(prompt)

    # 群聊回复时 ats 参数可 @回提问者（字段以官方文档为准）
    reply_to = room_id if room_id else from_id
    send_text(to_wxid=reply_to, content=answer)
    return jsonify({"code": 200})

群聊场景额外注意：群成员发送的 @昵称 文本在不同客户端版本下格式可能略有差异，建议同时支持"@昵称"和"@微信ID"两种判断方式。另外，若机器人在多个群同时活跃，建议用群 ID 维度做独立的频率限制，防止某一群的高频提问影响其他群的响应速度。

七、知识库运维与优化建议

7.1 文档更新策略

文档更新时容易踩的坑：若用文件名作为 chunk 的元数据 ID 前缀，文件改名会导致旧 chunk 无法被精准删除，建议用文件的哈希值或业务侧的文档 ID 作为唯一标识，彻底避免这个问题。

7.2 回答质量优化

• Rerank：Top-K 检索后用交叉编码器对候选段落重排序，准确率提升明显；
• Hybrid Search：向量检索 + 关键词检索（BM25）混合，补足语义检索对专有名词的不足；
• 多轮对话：在 Prompt 里携带最近 2～3 轮历史，让模型能理解上下文追问；
• 答案校验：对回答附上参考段落出处，方便用户核实，也提升可信度；
• 冷启动评估：上线前准备 20～50 条典型问答对，逐一跑一遍，用作基准，后续迭代时对比基准分数，确保改动没有引入回退。

7.3 防止滥用

• 对单个微信 ID 的请求频率做限流（如每分钟不超过 10 次），防止无效刷量；
• 敏感问题可在 Prompt 中加入"不回答与公司业务无关的问题"指令；
• 定期审查回答日志，发现质量问题及时补充文档或调整分块策略；
• 对于涉及薪资、合同、人事等高度敏感类文档，建议单独建库并做权限隔离，只对有权限的员工微信 ID 开放查询入口，防止越权访问。

总结

把 RAG 和微信消息打通，核心链路并不复杂：文档分块入向量库、用户提问触发检索、检索结果喂给 LLM、回答通过 HTTP 接口推送回微信。难点在于分块策略的精细调整、群聊消息过滤逻辑、以及知识库的持续运维。

实操层面有几个常见问题值得提前关注：一是分块粒度要反复测试，过大或过小都会直接拉低回答质量；二是 Embedding 模型和 LLM 要尽量选用同一语言体系的组合，中文场景下混用英文 Embedding 和中文 LLM 容易出现语义对齐偏差；三是回调服务的稳定性和响应时延要优先保障，LLM 调用超时应有兜底回复，避免用户长时间无响应。

按本文的结构搭建起基础版本后，再根据实际业务反馈逐步迭代 Rerank、混合检索等高级能力，是更务实的演进路径。知识库的价值最终体现在"文档质量 × 检索精度 × 模型能力"的乘积上，任何一环的短板都会被放大，持续的运维投入不可或缺。