【KnowPivot 开源实战 02】手把手实现 QQ 群智能客服：KnowPivot + Python 桥接搭建全自动问答机器人-171主机测评

本文是 KnowPivot 开源实战系列的第 2 篇。上一篇我们介绍了 KnowPivot 的开源发布，这一篇直接上干货——用 KnowPivot 搭建一个 QQ 群全自动智能客服机器人，实现群内提问、秒级回复。

一、场景与目标

1.1 实际需求

很多团队都有这样的场景：

公司内部系统（OA、ERP、项目管理系统等）的操作问题，新员工反复问老员工
IT 运维群，每天被"VPN 怎么连"“打印机怎么配”"账号怎么重置"刷屏
客服群，大量重复性问题占用人力

能不能把这些系统的操作文档丢进知识库，然后在 QQ 群里 @机器人就能自动回答？

这就是本文要实现的目标。

1.2 最终效果

在这里插入图片描述

[群成员A] @客服小K 项目编号怎么生成？
[客服小K] 项目编号（统一编号）在项目上报后由系统自动生成，
生成规则为：年份(4位) + 流水号(5位)。
[引用：问答对 – 统一编号何时生成？]

[群成员B] @客服小K 临时用地审批流程是什么？
[客服小K] 临时用地审批流程如下：
1. 用地单位提交申请 → 2. 所在乡镇初审 → 3. 县局审查 …
[引用：文档 – 临时用地管理办法.pdf 第3页]

二、技术方案选型

2.1 两个核心组件 + 一个桥接层

组件职责选择理由

KnowPivot	RAG 知识库 + 智能问答	全本地、零成本、四层检索增强
QQ 机器人框架	QQ 协议适配	基于 OneBot 11 标准协议，生态成熟
Python 桥接脚本	消息中转	连接 QQ 机器人和 KnowPivot，约 150 行代码

说明：出于保护原因，本文未提及实际使用的具体程序名称，仅以类同方式描述实现思路。文中的桥接代码基于标准 OneBot 11 协议，可适配任意兼容框架。如需获取本套完整落地程序（含具体工具选型、配置模板、完整部署脚本），可联系本人交流。

2.2 为什么不用现成的集成方案？

市面上有不少集成了大模型能力的机器人框架，比如支持 GPT 的微信机器人、AI 对话机器人等。这些框架功能很强，但往往：

内置了固定的大模型接口（OpenAI、DeepSeek 等），接入私有知识库需要额外开发
走的是官方 API 通道，部分平台只能接收 @机器人的消息，无法监听群内所有对话
配置复杂，二开成本高

KnowPivot 本身就是一个标准化的 REST API 服务，只要机器人框架能对外推送消息事件，就能通过简单的 HTTP 调用接入。这就是本文的核心思路：

不依赖任何特定的大模型机器人框架，用一段通用的 Python 桥接代码，把 QQ 消息事件和 KnowPivot 的 RAG API 串起来。

2.3 整体架构

消息流转非常清晰：

QQ 机器人框架收到 QQ 群消息，通过 WebSocket 推送过来

桥接脚本判断是否需要回复（@机器人 / 关键词触发 / 全部回复）

调用 KnowPivot 的 Agent 对话 API，获取 RAG 回答

将回答通过 QQ 机器人发送回 QQ 群

核心优势：桥接脚本完全不依赖具体的机器人框架。只要对方遵循 OneBot 11 协议，代码一行不用改。在这里插入图片描述

2.4 OneBot 11 协议简介

OneBot 11 是一个标准化的机器人协议，定义了消息事件的 JSON 格式和 API 调用方式。简单说，你的 QQ 机器人把收到的消息封装成这样的 JSON：

{
"post_type": "message",
"message_type": "group",
"group_id": "123456789",
"user_id": "987654321",
"message": "@机器人 VPN 怎么连接？"
}

桥接脚本解析这个 JSON，提取用户问题和群号，然后调用 KnowPivot API，最后把回答通过同样的协议发回去。

三、环境准备

3.1 硬件与软件要求

项目要求

操作系统	Windows / Linux 均可
Python	3.10 ~ 3.12
QQ 机器人框架	支持 OneBot 11 反向 WebSocket 的任意框架
KnowPivot	已部署运行（参考第 1 篇文章）
Ollama	已安装并拉取模型（如 qwen3:7b）

3.2 本文环境

操作系统：Windows 11
Python：3.10
QQ 机器人：OneBot 11 兼容框架
KnowPivot：运行在 http://127.0.0.1:8001
Ollama：qwen3:7b 本地模型

四、部署 QQ 机器人框架

4.1 选型与安装

选择一款支持 OneBot 11 反向 WebSocket 的 QQ 机器人框架。目前主流的选择包括：

基于 NTQQ 的框架：直接对接 QQ NT 底层协议，支持正向/反向 WebSocket
基于官方 API 的框架：走腾讯开放平台接口，需要注册开发者账号

本文推荐基于 NTQQ 的方案，因为：

无需官方审核，扫码即用
支持正向/反向 WebSocket，对接灵活
可以监听群内所有消息（按需过滤）

前往对应项目的 GitHub Releases 页面下载最新版本，按官方文档完成安装和扫码登录。

4.2 配置反向 WebSocket

登录机器人管理面板后，进入网络配置，添加一个反向 WebSocket 连接：

配置项值说明

连接类型	反向 WebSocket	机器人主动连接我们的 Python 脚本
地址	ws://127.0.0.1:8765	桥接脚本监听的 WebSocket 地址
Token	your_token_here	自定义一个 Token，用于鉴权
重连间隔	5000ms	断线自动重连

保存配置后，机器人会尝试连接 ws://127.0.0.1:8765。此时桥接脚本还没启动，连接会失败，这是正常的——脚本启动后会自动连上。

4.3 获取机器人 QQ 号

在管理面板中可以看到当前登录的 QQ 号，记下来，后面要用。

五、编写桥接机器人

5.1 项目结构

在 KnowPivot 项目旁边创建一个新目录：

qq_bot/
├── bot.py # 机器人主程序（约 150 行）
├── requirements.txt # 依赖
└── .env # 配置（从 .env.example 复制）

5.2 安装依赖

cd qq_bot
pip install websockets requests pydantic-settings

requirements.txt：

websockets>=12.0
requests>=2.31
pydantic-settings>=2.0

5.3 配置文件

.env.example：

# ─── WebSocket 连接配置 ────────────────────────
# WebSocket 监听端口（机器人反向 WS 连接目标）
WS_PORT=8765
# OneBot Access Token（与机器人配置一致）
WS_TOKEN=your_token_here

# ─── KnowPivot 配置 ────────────────────────────
# KnowPivot 服务地址
KNOWPIVOT_BASE_URL=http://127.0.0.1:8001
# Agent ID（在 KnowPivot 管理界面创建 Agent 后获取）
AGENT_ID=your_agent_id_here

# ─── 机器人行为配置 ────────────────────────────
# 机器人 QQ 号
BOT_QQ=123456789
# 触发模式：at（仅@机器人回复）/ all（群内所有消息都回复）
TRIGGER_MODE=at
# 回复前缀（可选）
REPLY_PREFIX=

5.4 核心代码：bot.py

"""
QQ 群智能客服桥接机器人
QQ 机器人 (OneBot 11) ←→ KnowPivot (RAG API)
"""
import asyncio
import json
import os
import logging
import re
import requests
import websockets
from pydantic_settings import BaseSettings, SettingsConfigDict

# ─── 配置 ───────────────────────────────────────────────────
class BotConfig(BaseSettings):
model_config = SettingsConfigDict(
env_file=os.path.join(os.path.dirname(__file__), ".env"),
env_file_encoding="utf-8",
env_file_safe=True,
extra="ignore",
)

ws_port: int = 8765
ws_token: str = ""
knowpivot_base_url: str = "http://127.0.0.1:8001"
agent_id: str = ""
bot_qq: str = ""
trigger_mode: str = "at" # at / all
reply_prefix: str = ""

config = BotConfig()

# ─── 日志 ───────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("qq_bot")

# ─── 知识库检索（社交消息预判 + KnowPivot API） ──────────────
# 简单的社交消息正则，匹配问候/感谢/告别等，直接回复不走 RAG
SOCIAL_PATTERNS = [
r"^(你好|hi|hello|嗨|在吗|在不在)\\s*[？?!!。]*$",
r"^(谢谢|感谢|thanks|thx|3q)\\s*[！!。.]*$",
r"^(好的|收到|了解|明白|ok|OK)\\s*[！!。.]*$",
r"^(早上好|下午好|晚上好|早安|晚安)\\s*[！!。.]*$",
]

SOCIAL_REPLIES = [
"你好呀！有什么可以帮助你的吗？直接 @我提问就行~",
"不客气！还有其他问题随时 @我哦~",
"收到！如果后续有问题，随时 @我~",
"你好！我是智能客服小K，有任何系统操作问题都可以 @我询问~",
]

def is_social_message(text: str) –> str | None:
"""判断是否为社交消息，返回回复内容或 None"""
cleaned = text.strip().lower()
for pattern in SOCIAL_PATTERNS:
if re.match(pattern, cleaned):
import random
return random.choice(SOCIAL_REPLIES)
return None

def call_knowpivot(question: str, history: list = None) –> dict:
"""调用 KnowPivot Agent 对话 API"""
url = f"{config.knowpivot_base_url.rstrip('/')}/api/agent/chat"
payload = {
"agent_id": config.agent_id,
"question": question,
"history": history or [],
}
try:
resp = requests.post(url, json=payload, timeout=60)
data = resp.json()
if data.get("code") == 0:
return data["data"]
else:
logger.warning(f"KnowPivot 返回错误: {data.get('msg')}")
return {"answer": f"知识库查询失败：{data.get('msg', '未知错误')}"}
except requests.exceptions.Timeout:
return {"answer": "知识库响应超时，请稍后再试~"}
except Exception as e:
logger.error(f"调用 KnowPivot 异常: {e}")
return {"answer": "知识库暂时不可用，请稍后再试~"}

# ─── OneBot 消息处理 ────────────────────────────────────────
def extract_text(message: str | list | dict) –> str:
"""从 OneBot 消息段中提取纯文本"""
if isinstance(message, str):
return message
if isinstance(message, list):
parts = []
for seg in message:
if isinstance(seg, dict) and seg.get("type") == "text":
parts.append(seg.get("data", {}).get("text", ""))
return "".join(parts).strip()
return ""

def should_reply(post_type: str, message_type: str,
sub_type: str, user_id: str, message: str,
at_me: bool) –> bool:
"""判断是否需要回复此消息"""
# 只处理群消息
if post_type != "message" or message_type != "group":
return False
# 不回复自己发的消息
if str(user_id) == config.bot_qq:
return False
# 不回复系统消息
if sub_type == "anonymous":
return False

if config.trigger_mode == "at":
return at_me
else: # all
return True

def remove_at_mention(message: str) –> str:
"""移除消息中的 @机器人部分"""
# CQ 码格式: [CQ:at,qq=123456]
cleaned = re.sub(r"\\[CQ:at,qq=\\d+\\]", "", message).strip()
return cleaned

def build_reply_message(answer: str, references: list = None) –> str:
"""构建回复消息文本"""
reply = ""
if config.reply_prefix:
reply += config.reply_prefix

reply += answer

# 可选：附加引用来源
if references:
reply += "\\n\\n—\\n📋 参考来源："
for i, ref in enumerate(references[:3], 1):
ref_type = ref.get("type", "")
if ref_type == "qa_pair":
reply += f"\\n {i}. 问答对：{ref.get('question', '')[:30]}"
elif ref_type == "auto":
reply += f"\\n {i}. 文档片段"
else:
reply += f"\\n {i}. 知识库片段"

return reply

# ─── WebSocket 服务端 ──────────────────────────────────────
async def handle_connection(websocket):
"""处理单个 WebSocket 连接"""
remote = websocket.remote_address
logger.info(f"新连接: {remote}")

try:
async for raw_message in websocket:
try:
event = json.loads(raw_message)
except json.JSONDecodeError:
continue

post_type = event.get("post_type")
if post_type != "message":
# 处理心跳等事件
if event.get("meta_event_type") == "lifecycle":
logger.info(f"生命周期事件: {event.get('sub_type')}")
continue

# ── 解析消息 ──
message_type = event.get("message_type")
sub_type = event.get("sub_type", "")
user_id = str(event.get("user_id", ""))
group_id = str(event.get("group_id", ""))
raw_msg = event.get("message", "")
sender = event.get("sender", {})

text = extract_text(raw_msg)
if not text:
continue

# 检查是否 @了机器人
at_me = False
if isinstance(raw_msg, list):
for seg in raw_msg:
if (isinstance(seg, dict)
and seg.get("type") == "at"
and str(seg.get("data", {}).get("qq", "")) == config.bot_qq):
at_me = True
break

# 判断是否需要回复
if not should_reply(post_type, message_type, sub_type,
user_id, text, at_me):
continue

# 移除 @机器人
question = remove_at_mention(text) if at_me else text
if not question:
continue

nickname = sender.get("card") or sender.get("nickname", "未知")
logger.info(f"[群{group_id}] {nickname}({user_id}): {question}")

# ── 社交消息预判 ──
social_reply = is_social_message(question)
if social_reply:
await send_group_message(websocket, group_id, social_reply)
continue

# ── 调用 KnowPivot RAG ──
logger.info(f"调用 KnowPivot: {question[:50]}…")
result = call_knowpivot(question)

answer = result.get("answer", "知识库暂时无法回答此问题。")
references = result.get("references", [])

# ── 构建并发送回复 ──
reply_text = build_reply_message(answer, references)
await send_group_message(websocket, group_id, reply_text)
logger.info(f"已回复 [群{group_id}]: {answer[:80]}…")

except websockets.exceptions.ConnectionClosed:
logger.warning(f"连接断开: {remote}")
except Exception as e:
logger.error(f"连接异常: {e}")

async def send_group_message(websocket, group_id: str, message: str):
"""通过 OneBot API 发送群消息"""
api_payload = {
"action": "send_group_msg",
"params": {
"group_id": int(group_id),
"message": message,
},
"echo": "qq_bot_reply",
}
await websocket.send(json.dumps(api_payload))

async def main():
logger.info("=" * 50)
logger.info("QQ 群智能客服桥接机器人")
logger.info(f"WebSocket 监听端口: {config.ws_port}")
logger.info(f"KnowPivot 地址: {config.knowpivot_base_url}")
logger.info(f"Agent ID: {config.agent_id}")
logger.info(f"触发模式: {config.trigger_mode}")
logger.info("=" * 50)

async with websockets.serve(
handle_connection,
"0.0.0.0",
config.ws_port,
):
logger.info(f"服务已启动，等待机器人连接 ws://0.0.0.0:{config.ws_port}")
await asyncio.Future() # 永不退出

if __name__ == "__main__":
asyncio.run(main())

5.5 代码说明

整个桥接脚本大约 150 行，核心逻辑只有这几个函数：

函数作用

handle_connection	WebSocket 服务端主循环，接收机器人事件
should_reply	判断是否需要回复（@模式 / 全部模式）
is_social_message	社交消息预判（问候/感谢等，不走 RAG）
call_knowpivot	调用 KnowPivot Agent API，获取 RAG 回答
build_reply_message	构建回复文本（可选附加引用来源）
send_group_message	通过 OneBot API 发送群消息

六、配置 KnowPivot Agent

6.1 创建知识库并导入资料

在 KnowPivot 管理界面（http://127.0.0.1:8001/）：

创建知识库：命名如"系统操作手册"，绑定 Embedding 模型

上传文档：将系统操作手册、FAQ 文档等 PDF/TXT/MD 文件上传

录入问答对：将常见问题整理为标准问答对，检索优先级最高

示例问答对：
Q: VPN 怎么连接？
A: 1. 下载 VPN 客户端 → 2. 安装后打开 → 3. 服务器地址填写 vpn.company.com → 4. 输入工号和密码 → 5. 点击连接

Q: 密码忘了怎么办？
A: 1. 访问 http://password.company.com → 2. 输入工号 → 3. 通过手机验证码验证身份 → 4. 设置新密码

6.2 创建 Agent

进入「Agent 管理」→「新建 Agent」：

配置项建议值说明

Agent 名称	系统客服小K	群里显示的机器人身份
绑定知识库	系统操作手册	刚才创建的知识库
绑定 LLM 模型	Qwen3-7B	Ollama 本地模型
系统提示词	见下方	定义机器人角色和回答风格
Top-K	4~6	检索条数
Temperature	0.3	低温度，回答更保守准确
相似度阈值	1.0	适中，不过严也不过松

系统提示词示例：

你是公司内部系统的智能客服助手。你的职责是回答用户关于系统操作的问题。

回答规则：
1. 基于知识库中的参考资料回答，不要编造知识库之外的内容
2. 如果知识库中没有相关资料，直接回复"抱歉，知识库中暂无相关内容，建议联系 IT 支持群（群号 xxxxx）"
3. 回答要简洁、分步骤、可操作
4. 涉及具体操作时，列出步骤编号
5. 语气友好专业

创建完成后，记下 Agent ID（UUID 格式），填入桥接脚本的 .env 配置中。

6.3 在 Agent 详情页测试对话

在测试对话中输入几个问题，确认回答效果满意后再接入 QQ。

七、启动与联调

7.1 启动顺序

# 终端 1：启动 KnowPivot
cd KnowPivot/mini_rag_backend
python main.py

# 终端 2：启动 QQ 机器人框架（如果没在运行）
# 按对应框架的启动方式运行

# 终端 3：启动桥接机器人
cd qq_bot
python bot.py

7.2 验证连接

启动桥接机器人后，控制台应该看到：

2026-07-20 16:30:00 [INFO] ==================================================
2026-07-20 16:30:00 [INFO] QQ 群智能客服桥接机器人
2026-07-20 16:30:00 [INFO] WebSocket 监听端口: 8765
2026-07-20 16:30:00 [INFO] KnowPivot 地址: http://127.0.0.1:8001
2026-07-20 16:30:00 [INFO] Agent ID: c2c169ad-4d11-4258-acbb-f9d752c38e9b
2026-07-20 16:30:00 [INFO] 触发模式: at
2026-07-20 16:30:00 [INFO] ==================================================
2026-07-20 16:30:00 [INFO] 服务已启动，等待机器人连接 ws://0.0.0.0:8765

# 机器人连接成功后：
2026-07-20 16:30:05 [INFO] 新连接: ('127.0.0.1', 54321)

7.3 在 QQ 群测试

在目标 QQ 群中 @机器人提问：

[你] @客服小K VPN 怎么连接？

桥接机器人控制台输出：

2026-07-20 16:31:00 [INFO] [群987654321] 张三(111222333): VPN 怎么连接？
2026-07-20 16:31:00 [INFO] 调用 KnowPivot: VPN 怎么连接？…
2026-07-20 16:31:03 [INFO] 已回复 [群987654321]: 1. 下载 VPN 客户端 → 2. 安装后打开 → …

QQ 群中收到机器人回复，测试通过。

八、进阶优化

8.1 多轮对话支持

桥接脚本目前是单轮对话（每次调用 KnowPivot 不传 history）。如果需要支持多轮上下文，可以加一个简单的内存缓存：

# 在 bot.py 顶部添加
from collections import defaultdict
import time

# 用户最近 N 条对话记录
conversation_cache: dict[str, list] = defaultdict(list)
MAX_HISTORY = 6
CACHE_TTL = 300 # 5 分钟过期

def get_history(user_id: str) –> list:
"""获取用户历史对话"""
now = time.time()
conversations = conversation_cache[user_id]
# 清理过期记录
conversation_cache[user_id] = [
c for c in conversations if now – c["time"] < CACHE_TTL
]
return [
{"role": c["role"], "content": c["content"]}
for c in conversation_cache[–MAX_HISTORY:]
]

def add_to_history(user_id: str, role: str, content: str):
"""添加对话记录"""
conversation_cache[user_id].append({
"role": role,
"content": content,
"time": time.time(),
})

在调用 call_knowpivot 前后分别调用 get_history 和 add_to_history 即可。

8.2 限流防刷

防止群内刷屏影响 KnowPivot 服务：

from collections import defaultdict
import time

last_reply_time: dict[str, float] = defaultdict(float)
COOLDOWN = 10 # 同一用户 10 秒内只回复一次

def check_rate_limit(user_id: str) –> bool:
now = time.time()
if now – last_reply_time[user_id] < COOLDOWN:
return False
last_reply_time[user_id] = now
return True

8.3 关键词触发（非 @模式）

如果不想每次都 @机器人，可以设置关键词触发：

TRIGGER_KEYWORDS = ["怎么", "如何", "什么是", "为什么", "？", "?", " help", "帮助"]

def should_reply_by_keyword(text: str) –> bool:
return any(kw in text for kw in TRIGGER_KEYWORDS)

8.4 后台运行与自动重启

Windows（使用 nssm 注册为服务）：

nssm install QQBot "C:\\path\\to\\venv\\Scripts\\python.exe" "C:\\path\\to\\qq_bot\\bot.py"
nssm start QQBot

Linux（使用 systemd）：

# /etc/systemd/system/qq-bot.service
[Unit]
Description=QQ Group Bot Bridge
After=network.target

[Service]
Type=simple
User=bot
WorkingDirectory=/opt/qq_bot
ExecStart=/opt/qq_bot/venv/bin/python bot.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl enable qq-bot
sudo systemctl start qq-bot

九、常见问题

Q1：机器人框架启动后无法扫码登录？

检查 QQ 客户端版本是否符合框架要求。部分框架要求 QQ 版本在 9.9.26 以上。如果已安装 QQ 但框架找不到，尝试重新安装指定版本的 QQ。

Q2：桥接脚本启动了但机器人没有连上来？

检查机器人管理面板中反向 WebSocket 地址是否正确（ws://127.0.0.1:8765）

检查 Token 是否一致

确认桥接脚本先于机器人启动，或机器人配置了断线重连

Q3：机器人回复"知识库暂时无法回答"？

检查知识库是否有资料

检查 Agent ID 是否正确

检查 KnowPivot 是否正常运行（访问 /api/health）

适当调大相似度阈值

Q4：如何让机器人回复带图片或 Markdown？

QQ 群文本消息不支持 Markdown 渲染。如果需要发送图片，可以使用 OneBot 的图片消息段：

# 发送图片消息
api_payload = {
"action": "send_group_msg",
"params": {
"group_id": int(group_id),
"message": [
{"type": "text", "data": {"text": "请参考以下截图："}},
{"type": "image", "data": {"file": "https://example.com/screenshot.png"}},
],
},
}

十、总结

本文从零搭建了一个完整的 QQ 群智能客服系统，核心链路清晰：

QQ 机器人 (OneBot 11) → WebSocket → Python 桥接（150行）→ KnowPivot API（RAG）→ 回复

核心优势：

零成本：全部本地部署，无 API 费用
数据安全：文档和对话数据不出内网
灵活可控：150 行桥接代码，想改什么随时改
通用性强：基于标准 OneBot 11 协议，不绑定任何特定机器人框架
检索增强：四层 RAG 检索，回答准确有据可查

项目地址：https://github.com/mzsoft-site/KnowPivot.git

gitcode(国内用户友好)：https://gitcode.com/mzsoft/KnowPivot.git

上一篇：【KnowPivot 开源实战 01】KnowPivot 开源发布：零成本搭建私有化 RAG 知识库下一篇预告：《混合检索 + 重排序 + Multi-Query：四层 RAG 检索链路实战》，我会深入拆解 KnowPivot 的 RAG 检索引擎，对比各层策略的效果差异，敬请期待~

本文涉及的所有代码均已开源，欢迎 Star 和 Fork！