从API到多段回复：手把手搭建DeepSeek QQ机器人完整链路

把 DeepSeek 接入 QQ 机器人，看起来只是把 API 地址和 Key 换一换，实际做起来才会发现，一次完整的“拟人化聊天”要同时处理消息协议、多轮上下文、回复节奏和多段发送逻辑。尤其“多段回复”这个需求，很多人一开始不理解：模型明明一次就能生成整段回复，为什么还要拆成好几条发出去？等真正把机器人拉进群里，看到一整屏又硬又长的公告式回答，才会明白分段不只是形式问题，而是拟人化体验的一部分。下面会从零搭一个可运行的 DeepSeek QQ 聊天机器人：先跑通 API，再接入 OneBot 11 协议的消息端，接着实现人设、上下文和回复节奏，最后重点讲多段回复的拆分策略、发送控制和排错方法。整个项目不依赖付费网关，也不需要复杂的分布式架构，代码量控制在几百行内，适合想理解完整链路而不是只会套模板的开发者。

1. 先理清 QQ 机器人接入 DeepSeek 的完整链路

1.1 一条消息从 QQ 群到 DeepSeek 再回来，经过哪些节点

写代码之前要先画清楚链路。用户在 QQ 群里发一句“你好”，这句消息并不会直接到达 DeepSeek。它要经过这样一条路径：

1. 用户通过 QQ 客户端把消息发到 QQ 服务器。
2. 运行在你本机或服务器上的 OneBot 协议实现，以机器人账号的身份登录 QQ，并把收到的消息转换成标准事件。
3. 你的 Python 进程通过 WebSocket 连接到这个协议实现，接收消息事件。
4. Python 进程把用户消息和该会话的历史记录一起组装成 messages 列表，调用 DeepSeek API。
5. DeepSeek 返回回复文本，或者以流式方式逐步返回文本增量。
6. Python 进程把完整回复按句子边界切分成多段，逐条通过 OneBot 操作接口发送到原群。

这里最关键的一点是：DeepSeek 只负责“文本到文本”，它不知道 QQ、不知道群号、不知道消息 ID。所有协议处理、会话隔离、发送节奏，都必须由你自己的服务完成。很多初学项目把代码写得像是“调一次接口就结束”，结果上线后各种奇怪现象：回得慢、回得生硬、多人群里上下文串台、消息发不出去，本质都是链路里少了某一环。

1.2 谁负责“拟人化”

拟人化不是一个模型开关，而是三个环节共同作用的结果：

• 系统提示词负责定义语气、句式和边界，让模型不要一开口就是客服腔。
• 会话管理负责记住上下文，让机器人不会每句话都像失忆一样重新开始。
• 回复节奏负责控制“什么时候说、说多长、分几条说”，让输出看起来像人在打字。

模型本身只负责生成文本。它生成的文本可能已经比较口语化，但如果你的程序一次性把 500 个字甩到群里，再口语化的内容也会显得像一个公告。这就是多段回复逻辑存在的意义：它把模型的输出重新组织成符合聊天场景的节奏。

1.3 三种接入姿势怎么选

在动手前，先明确要采用哪种开发方式。下面三种方式都能做，差别在控制力、开发速度和排错成本。

本文选择纯手写方式，不是因为框架不好，而是因为“保姆级教学”的目标是把链路讲清楚。手写一遍之后，你再去用 NoneBot2 或任何封装框架，都会知道它内部在做什么。

2. 环境准备：先对齐账号、依赖和目录再写代码

2.1 前置条件清单

开始安装之前，先把下面这些条件准备好，缺一项都会在中途卡住。

注意：机器人账号和日常账号分离是底线。开发过程中难免出现消息错发、循环调用、异常刷屏等情况，用小号测试可以把风险隔离在可控范围内。

2.2 安装 Python 依赖

核心依赖只有两个：openai和websockets。前者用于调用 DeepSeek 的 OpenAI 兼容接口，后者用于连接 OneBot 协议实现的 WebSocket 服务端。

python -m venv .venv source .venv/bin/activate pip install openai websockets

Windows 下激活虚拟环境的命令是.venv\Scripts\activate。安装完成后，可以把版本固定到requirements.txt：

openai>=1.30.0 websockets>=12.0

实际安装时以你当前环境支持的版本为准。这里不锁死具体版本号，是因为 openai SDK 迭代比较快，接口细节可能有小差异，落地前先确认一下版本更稳妥。

2.3 项目目录结构

qq-deepseek-bot/ ├── config.py # 配置项集中管理 ├── main.py # 主程序：WebSocket 连接与事件处理 ├── requirements.txt # Python 依赖 └── README.md # 运行说明

这个结构足够小，但已经体现了“配置和逻辑分离”的原则。不要把 API Key 直接散落在代码里，后面生产环境还要把配置挪到环境变量或配置中心。

3. 先用最小代码跑通 DeepSeek 聊天接口

3.1 同步调用最小示例

先不碰 QQ，用最小代码验证 DeepSeek API 能正常返回。DeepSeek 提供 OpenAI 兼容接口，所以直接使用openai库，把base_url指向 DeepSeek 的接口地址即可。

from openai import OpenAI client = OpenAI( api_key="sk-your-key-here", base_url="https://api.deepseek.com", ) response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是一个群聊助手，回答简洁自然，不超过100字。"}, {"role": "user", "content": "你好，简单介绍一下自己。"}, ], temperature=0.9, max_tokens=200, ) print(response.choices[0].message.content)

这段代码解决的是“能不能调通”的问题。执行后，控制台会打印模型的回复文本。如果没有报错，说明 Key、base_url、模型名和网络连通性都没有问题。

3.2 返回结构里哪些字段有用

DeepSeek 兼容 OpenAI 的返回结构，核心字段如下：

一个典型的响应 JSON 长这样：

{ "id": "chatcmpl-xxx", "choices": [ { "finish_reason": "stop", "message": { "content": "你好呀，我是群里的聊天机器人。", "reasoning_content": null, "role": "assistant" } } ], "usage": { "prompt_tokens": 42, "completion_tokens": 18, "total_tokens": 60 } }

注意reasoning_content这个字段。使用deepseek-reasoner或在思考模式下调用时，它可能不为空。后面排查章节会专门讲它引起的 400 报错。

3.3 流式输出：多段回复的起点

流式输出的价值不是省时间，而是让你能在文本生成的中间过程拿到增量内容。多段回复的核心思路就是：生成到句子边界就发一段，再继续生成下一句。

stream = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是一个群聊助手。"}, {"role": "user", "content": "给我讲讲怎么做番茄炒蛋。"}, ], stream=True, ) for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="")

这里chunk.choices[0].delta.content是当前这一小段增量文本，通常只有几个字。把增量累加起来，就是完整回复。

3.4 先在命令行验证上下文，再进 QQ

注意：先跑通命令行版，再接入 QQ。否则一旦出问题，你无法判断是 API 的问题还是消息协议的问题。

建议先用一个简单的 while 循环在命令行里做多轮对话，确认 system、user、assistant 三种消息交替时上下文不会乱。这一步完成后，再进入消息端接入。

4. 消息端接入：OneBot 11 协议下的收发

4.1 协议实现的选择与 WebSocket 配置

以 NapCat、LLOneBot、Lagrange 这类常见项目为例，它们在登录方式和界面上有差异，但对外都能提供接近 OneBot 11 规范的事件和操作接口。在你的协议实现里找到 WebSocket 相关配置，一般需要确认以下几点：

• 开启 WebSocket 服务端，记录它监听的端口，例如3001。
• 上报数据类型选择 JSON。
• 事件订阅里勾选message类型，否则程序收不到消息。

不同项目对“正向 WebSocket”和“反向 WebSocket”的叫法并不统一。本文采用的做法是：协议实现启动一个 WebSocket 服务端，你的 Python 程序作为客户端主动连接。如果你的协议实现只能作为客户端去连接别人的服务端，那 Python 端就要改成 WebSocket 服务端，代码方向反过来。先确认清楚这个连接方向，能少踩很多坑。

4.2 接收 message 事件

OneBot 11 的事件是 JSON。收到群消息时，典型事件长这样：

{ "post_type": "message", "message_type": "group", "group_id": 10001, "user_id": 20001, "message_id": 30001, "message": "你好", "raw_message": "你好" }

程序里按post_type和message_type分流即可：

async def handle_event(ws, event: dict): if event.get("post_type") != "message": return if event.get("message_type") == "group": group_id = event["group_id"] text = extract_text(event.get("message", "")) if text: await on_group_message(ws, group_id, text)

需要特别注意message字段有时是字符串，有时是消息段数组。从数组里提取纯文本要这样处理：

def extract_text(message) -> str: if isinstance(message, str): return message.strip() 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 ""

消息段格式是 OneBot 11 的核心概念。{"type": "text", "data": {"text": "你好"}}是纯文本，{"type": "at", "data": {"qq": "123"}}是 @ 某人。后面要支持 @ 触发、图片、表情，都是在message数组里解析对应 type。

4.3 发送消息

OneBot 11 通过发送 JSON 操作来执行发消息动作。发送群消息的操作是send_group_msg，私聊是send_private_msg。

import asyncio import json send_lock = asyncio.Lock() async def send_onebot_action(ws, action: str, params: dict): payload = { "action": action, "params": params, } async with send_lock: await ws.send(json.dumps(payload)) async def send_group_msg(ws, group_id: int, text: str): await send_onebot_action(ws, "send_group_msg", { "group_id": group_id, "message": text, })

这里加send_lock是因为后面会使用asyncio.create_task并发处理多个群的消息。多个任务同时向同一个 WebSocket 发送 JSON，不加锁会出现消息交错，接收端无法正确解析。

4.4 先做 ping/pong 最小闭环

接入 DeepSeek 之前，先让机器人收到“ping”回“pong”：

async def on_group_message(ws, group_id: int, text: str): if text == "ping": await send_group_msg(ws, group_id, "pong")

在群里发一条“ping”，机器人回复“pong”，说明消息收发链路已经通。此时才进入下一步，把 DeepSeek 接进来。如果这一步就不通，问题几乎都在协议实现配置或 WebSocket 连接上，和 AI 相关代码无关。

5. 拟人化聊天的三块基石：人设、上下文和回复节奏

5.1 系统提示词：拟人的底层边界

拟人不等于让模型随便说，而是给它一套稳定的表达约束。系统提示词写得好，回复语气会明显不同。

SYSTEM_PROMPT = """你是群聊成员“小深”，不是客服，也不是搜索引擎。 规则： 1. 回答口语化，避免“首先/其次/最后”式排比和书面腔。 2. 别人问什么就答什么，不主动长篇大论。 3. 用词自然，偶尔带一点语气词，但不要每句都带。 4. 不确定的事直接说不知道，不编造。 5. 回答一般不超过150字。 """

关键点在于“约束表达形式”，而不是“约束知识范围”。你可以把“口语化”“简短”“不说套话”写进去，也可以加上“不在回答里重复用户的问题”这类针对聊天场景的规则。但不建议把人设写成一长段小说式背景，模型很可能记不住，还会占用大量上下文 token。

5.2 多轮上下文：按群隔离并按轮次截断

拟人聊天的第二个基石是记忆。没有上下文的机器人，每句话都是重新开始，聊不了三轮就会露馅。比较简单的做法是用内存字典保存每个群的会话历史：

MAX_HISTORY = 10 histories: dict[str, list[dict]] = {} def get_history(key: str) -> list[dict]: return histories.setdefault(key, []) def append_message(key: str, role: str, content: str): history = get_history(key) history.append({"role": role, "content": content}) if len(history) > MAX_HISTORY: del history[: len(history) - MAX_HISTORY]

群聊场景的 key 建议用f"group:{group_id}"，这样不同群互不干扰。如果希望更细粒度，可以用f"group:{group_id}:user:{user_id}"按人隔离，但代价是上下文会被切得很碎，记忆效果反而不如按群聚合。这里需要根据产品定位取舍。

MAX_HISTORY控制的是保存的消息条数，不是 token 数。实际调用 DeepSeek 时，历史越多，prompt_tokens越高，成本和延迟都会上升。聊天场景先按 10 到 20 条历史控住，后续要精细控制时再改成按 token 估算截断。

5.3 回复延迟：拟人不等于秒回

真人打字需要时间。如果机器人 0.1 秒就回复 200 字，用户第一反应不是“智能”，而是“这是脚本”。所以回复前要根据文本长度模拟一个思考加打字的过程：

import random async def human_delay(text: str): seconds = min(0.8 + len(text) / 40, 3.0) await asyncio.sleep(random.uniform(seconds * 0.6, seconds * 1.4))

长度越长，延迟越长，并且延迟要在一定范围内随机浮动，避免每次都一样。这里不要写死固定时间，固定延迟很容易被用户识别出机械感。

5.4 抢话和重复回复怎么避免

群里同时有多人说话时，机器人最容易犯的错是“抢话”和“自我对话”。三个基本防护：

• 忽略机器人自己发的消息。在事件处理里判断user_id，如果等于机器人账号，直接返回。
• 忽略非文本消息。表情、图片、系统通知都会触发事件，先过滤掉。
• 同一会话加锁。一条回复还没发送完，又来一条新消息，不要让两个回复任务同时往同一个群发文本，否则输出会交错。

会话锁的实现放在下一章多段回复里一起讲，因为它和分段发送是配套的。

6. 多段回复逻辑：为什么难、怎么拆、怎么发

6.1 为什么要分段

先解决“为什么”。模型一次调用能生成完整回复，不需要你费劲拆分。但实际聊天场景里，一次性发超长文本有三个问题：

1. 体验问题：一大段文字像公告，不像聊天。
2. 长度问题：QQ 对超长文本有限制，超过一定长度可能发送失败或显示异常。
3. 节奏问题：逐句发送并带间隔，能模拟“正在打字”的节奏，让互动更像真人。

这里要澄清一个常见误解：分段不是把一段文字机械切成几块，而是在语义完整的句子边界处切分，并且每段之间有合理的发送间隔。

6.2 分段策略对比

固定长度硬切最简单，但不推荐，因为很可能把“我不喜欢吃”切成“我不喜”和“欢吃”。标点切分是性价比最高的方案。流式按句切分效果最好，也是本文推荐的做法。

6.3 非流式：按标点拆，保留标点并控制段长

如果暂时不想用流式，可以先拿到完整回复，再在本地切分。核心是使用正则，在中文句号、感叹号、问号和换行之后切分，并且让标点保留在前一段末尾：

import re def split_sentences(text: str, max_len: int = 200) -> list[str]: parts = re.split(r'(?<=[。！？!?；;\n])', text.strip()) sentences = [p for p in (s.strip() for s in parts) if p] segments = [] current = "" for sent in sentences: if len(current) + len(sent) <= max_len: current += sent else: if current: segments.append(current) if len(sent) > max_len: while len(sent) > max_len: segments.append(sent[:max_len]) sent = sent[max_len:] current = sent if current: segments.append(current) return segments

正则里的(?<=[。！？!?；;\n])是零宽正向后行断言，表示只在“后面是结尾标点或换行”的位置切分，切分时标点不会丢失。max_len用来兜底：遇到一个超长句子，临时按字符硬切，避免出现一段几百字的情况。

这个方案的缺点是没法做到“边生成边发”，必须等模型完整生成，首段延迟会高一些。

6.4 流式：句子边界一到就发送

流式方案改进了首段延迟。模型逐块吐字，只要缓冲区末尾出现句子结束符并且长度足够，就把这一段发出去，然后清空缓冲区继续累计。

from openai import AsyncOpenAI deepseek = AsyncOpenAI( api_key=config.DEEPSEEK_API_KEY, base_url=config.DEEPSEEK_BASE_URL, ) async def reply_with_streaming(session_key: str, user_text: str, send_text): messages = build_messages(session_key, user_text) stream = await deepseek.chat.completions.create( model="deepseek-chat", messages=messages, stream=True, temperature=0.9, ) buffer = "" async for chunk in stream: if not chunk.choices: continue delta = chunk.choices[0].delta.content if not delta: continue buffer += delta if buffer and buffer[-1] in "。！？!?；;\n" and len(buffer) >= 15: await send_text(buffer) buffer = "" await asyncio.sleep(random.uniform(0.5, 1.2)) if buffer: await send_text(buffer)

send_text是一个回调，在 OneBot 场景里就是send_group_msg的封装。每发送一段后 sleep 0.5 到 1.2 秒，是为了模拟打字停顿。流结束后缓冲区里剩余的内容作为最后一段发送。

注意：分段发送最重要的是节奏，不是数量。每段之间保持 0.5 到 1.5 秒的随机间隔，比拼命缩短首字延迟更能提升“像真人”的体验。

6.5 分段发送的边界处理与频率控制

分段发送看着简单，实际有不少边界情况需要处理。

第一个是代码块。如果模型回复里带 Markdown 代码块，只按标点切分会把代码块切碎。简单做法是：缓冲区中反引号数量是奇数时不切分：

def should_split(buffer: str, min_len: int = 15) -> bool: if len(buffer) < min_len: return False if buffer.count("```") % 2 == 1: return False if "http://" in buffer[-50:] or "https://" in buffer[-50:]: return False return buffer[-1] in "。！？!?；;\n"

第二条规则是保护 URL。句号是 URL 的常见字符，按句号切分会把链接切断，所以最后 50 个字符里出现 URL 时，暂不切分。

第二个是过短残句。如果最后一段只有几个字，比如“嗯嗯”，单独发一条会显得很碎。可以在流结束后判断，如果剩余 buffer 长度小于阈值，就把上一段和它合并。这个逻辑在完整代码里可以按需加上。

第三个是频率控制。即使算法正确，连续快速发 10 条消息也很容易被平台判定为异常行为。控制发消息的总频率，不要长时间高频输出。如果模型要输出很长，可以在累计到一定条数后加大间隔，或者在回复开头就引导模型控制长度。

6.6 多段发送期间的状态锁

分段发送需要时间。在发送过程中，如果同一会话又来新消息，必须保证两个回复任务不会交错输出。按会话加锁是最直接的做法：

session_locks: dict[str, asyncio.Lock] = {} async def process_conversation(ws, session_key: str, text: str, group_id: int): lock = session_locks.setdefault(session_key, asyncio.Lock()) async with lock: async def send_text(part: str): await send_group_msg(ws, group_id, part) await reply_with_streaming(session_key, text, send_text)

加锁后，同一群的新消息会排队等待，等当前回复全部发完再处理。这里也可以换一种策略：发现锁被占用时直接忽略新消息，或者提示“我正在打字，稍等”。两种策略各有取舍，产品要求不同，实现也不同。至少不能让两个任务同时往一个群发内容。

7. 完整代码整合与运行验证

7.1 配置文件

把 Key、端口、模型名和分段参数集中到config.py：

DEEPSEEK_API_KEY = "sk-your-key-here" DEEPSEEK_BASE_URL = "https://api.deepseek.com" DEEPSEEK_MODEL = "deepseek-chat" ONEBOT_WS_URL = "ws://127.0.0.1:3001" BOT_SELF_ID = 0 # 机器人账号，用于过滤自己发的消息 MAX_HISTORY = 10 MIN_SPLIT_LEN = 15

BOT_SELF_ID需要填成机器人账号的实际 QQ 号。不填或填 0，机器人就可能把自己发的消息当成用户消息，形成自我对话死循环。

7.2 主程序 main.py

import asyncio import json import random import websockets from openai import AsyncOpenAI import config deepseek = AsyncOpenAI( api_key=config.DEEPSEEK_API_KEY, base_url=config.DEEPSEEK_BASE_URL, ) send_lock = asyncio.Lock() session_locks: dict[str, asyncio.Lock] = {} histories: dict[str, list[dict]] = {} SYSTEM_PROMPT = ( f"你是群聊成员“小深”，说话自然口语化，" "回答简洁，不写排比句，不堆术语，" "不确定的事直接说不知道，回答一般不超过150字。" ) def get_history(key: str) -> list[dict]: return histories.setdefault(key, []) def extract_text(message) -> str: if isinstance(message, str): return message.strip() 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 build_messages(key: str, user_text: str) -> list[dict]: history = get_history(key) history.append({"role": "user", "content": user_text}) if len(history) > config.MAX_HISTORY: del history[: len(history) - config.MAX_HISTORY] return [{"role": "system", "content": SYSTEM_PROMPT}] + history def should_split(buffer: str) -> bool: if len(buffer) < config.MIN_SPLIT_LEN: return False if buffer.count("```") % 2 == 1: return False return buffer[-1] in "。！？!?；;\n" async def send_onebot_action(ws, action: str, params: dict): payload = {"action": action, "params": params} async with send_lock: await ws.send(json.dumps(payload)) async def send_group_msg(ws, group_id: int, text: str): await send_onebot_action(ws, "send_group_msg", { "group_id": group_id, "message": text, }) async def reply_stream(ws, key: str, user_text: str, group_id: int): messages = build_messages(key, user_text) full_text = "" buffer = "" stream = await deepseek.chat.completions.create( model=config.DEEPSEEK_MODEL, messages=messages, stream=True, temperature=0.9, ) async for chunk in stream: if not chunk.choices: continue delta = chunk.choices[0].delta.content if not delta: continue buffer += delta full_text += delta if should_split(buffer): await send_group_msg(ws, group_id, buffer) buffer = "" await asyncio.sleep(random.uniform(0.6, 1.3)) if buffer: await send_group_msg(ws, group_id, buffer) if full_text: get_history(key).append({"role": "assistant", "content": full_text}) async def handle_event(ws, event: dict): if event.get("post_type") != "message": return if event.get("user_id") == config.BOT_SELF_ID: return if event.get("message_type") == "group": group_id = event["group_id"] text = extract_text(event.get("message", "")) if not text: return key = f"group:{group_id}" lock = session_locks.setdefault(key, asyncio.Lock()) async with lock: try: await reply_stream(ws, key, text, group_id) except Exception as exc: print("[ERROR]", exc) await send_group_msg(ws, group_id, "我刚走神了，再说一次？") async def main(): print("[INFO] 正在连接 OneBot WebSocket:", config.ONEBOT_WS_URL) async with websockets.connect(config.ONEBOT_WS_URL, max_size=None) as ws: print("[INFO] 连接成功") async for raw in ws: try: event = json.loads(raw) except json.JSONDecodeError: continue asyncio.create_task(handle_event(ws, event)) if __name__ == "__main__": asyncio.run(main())

代码里有两个地方需要重点理解。

第一个是build_messages的副作用。它在构造请求时就把 user 消息写进了历史。如果 API 调用失败，这条 user 消息会残留到下一轮。严谨做法是先用临时列表拼请求，成功后再把 user 消息和 assistant 回复一起写入历史。教学版本为了清晰先这样写，生产环境必须改成成功后再提交。

第二个是asyncio.create_task(handle_event(...))。每个事件都被放进独立任务，同一群的消息会排队等待锁，不同群的消息可以并发处理。这是异步机器人最基本的并发模型。

7.3 启动和验证步骤

按下面的顺序执行，每步都确认结果：

1. 启动协议实现并登录机器人账号，确认 WebSocket 服务在3001端口监听。
2. 运行python main.py，日志显示“连接成功”。
3. 在群里发“ping”，确认机器人回“pong”。
4. 在群里发一句正常聊天内容，观察 DeepSeek 回复是否被分成多条发送。
5. 连续发两条消息，确认第二条会在第一条完整发送完后才被处理。

7.4 预期日志

正常情况下，日志大致如下：

[INFO] 正在连接 OneBot WebSocket: ws://127.0.0.1:3001 [INFO] 连接成功 [INFO] 收到群 10001 消息: 你好 [INFO] 分段发送: 你好呀，我是群里的聊天机器人。 [INFO] 等待 0.9s [INFO] 分段发送: 有什么想聊的可以直接说。

如果只看到“收到消息”而没有后续，问题在 DeepSeek 调用环节。如果看到“分段发送”但群里没消息，问题在发送环节。日志就是用来做这个阶段判断的。

8. 从现象找原因：这一套最容易踩的坑

8.1 收不到消息

记住一个判断原则：先看协议实现里能不能看到消息事件。能看到，问题在 Python 程序；看不到，问题在协议实现配置。

8.2 消息发不出去

发送失败时第一件事是打印你真正发出去的 JSON，而不是盯着代码猜。绝大多数问题都能在 payload 里直接看到。

8.3 deepseek-reasoner 多轮报 reasoning_content 错误

如果在使用推理模型或思考模式时，连续多轮对话出现下面这种报错：

upstream_status: http 400 cause: the `reasoning_content` in the thinking mode must be passed back to the api.

原因是：思考模式的多轮会话要求把上一轮 assistant 消息里的reasoning_content一起回传。如果你在保存上下文时只保留了content，把思考内容丢掉了，下一轮调用会被接口拒绝。

处理方式按顺序排查：

1. 聊天机器人优先使用deepseek-chat这类非思考模式，不涉及该字段，最简单。
2. 如果必须使用思考模式，保存历史时不要把 assistant 消息拆字段，把完整的 assistant 消息对象存下来，下一轮原样放回。
3. 如果当前 session 的历史已经混用了普通模式和思考模式，清空该会话历史再试。
4. 如果你的请求经过第三方网关或中转服务，先直连官方接口复现，确认是官方拒绝还是中间层改写消息体导致。

注意：遇到 400 时，先把链路拆开：直连 DeepSeek 官方接口是否正常；如果正常，再检查自己的消息构造和上下文保存逻辑；如果有中间网关，最后检查网关是否改写了消息体。

8.4 多段回复被合并或像刷屏

多段回复的调参没有标准答案，跟模型输出风格强相关。deepseek-chat在不同的 temperature 下，句子长度差异很大。建议先记录几组真实回复，再根据实际分布调整MIN_SPLIT_LEN和MAX_SEGMENT_LEN。

8.5 上下文超长和并发重复回复

上下文超长的表现是：请求越来越慢，甚至返回类似 context length exceeded 的报错。原因基本只有一个：历史列表无限增长。检查usage.prompt_tokens的变化趋势，然后把MAX_HISTORY调低，或者改成按 token 估算截断。

并发重复回复的表现是：用户发一条消息，机器人回了两次，或者两段回复交错发送。原因是同一个会话没有加锁，多个事件任务同时执行。检查点：

• 是否在handle_event里处理了所有post_type为 message 的事件。
• 是否过滤了机器人自己的消息。
• 是否每个会话 key 都有独立的asyncio.Lock。
• 是否多个进程或实例同时连了同一个 WebSocket。

只要session_locks是以 group_id 为 key 的独立锁，并且所有回复都在锁内发送，交错问题就不会出现。

9. 生产环境最佳实践与扩展方向

9.1 学习环境到生产环境的差距

本文的完整代码可以在本机跑通，但它是一个学习版本，不是生产版本。两者差距很大：

生产环境至少要补上 WebSocket 断线重连。当前main.py如果连接断开，整个程序会退出。简单做法是在main外层包一层重试循环，捕获ConnectionClosed后等待几秒重新连接。

9.2 发布前检查清单

每个机器人上线前，都建议按这个清单过一遍：

发布前检查清单 - [ ] API Key 已从代码移到环境变量，未提交到代码仓库 - [ ] WebSocket 断线重连逻辑已实现 - [ ] 上下文按群隔离并有轮次上限 - [ ] 多段发送每段之间有随机延时 - [ ] 机器人不会处理自己发的消息 - [ ] 所有异常都有日志，不会静默失败 - [ ] 单条消息长度有上限，不会发送超长文本 - [ ] API 调用失败时，历史记录不会残留脏数据 - [ ] token 消耗有统计，每天记录调用次数和成本 - [ ] 内容合规：系统提示词已约束，必要时增加过滤层

清单里的每一项都能对应到具体代码或配置，不是空泛口号。上线前逐条检查，能避免大多数低级事故。

9.3 扩展方向

这个项目跑通之后，可以在几个方向上继续深入：

• 支持 @ 触发：解析message数组里的 at 段，只有 @ 机器人才回复。
• 支持图片和 CQ 码：用 OneBot 的消息段发送图片、表情和合并转发。
• 多账号接入：同一套逻辑同时连接多个 WebSocket，按账号隔离会话。
• 模型路由：闲聊用deepseek-chat，复杂问题切到deepseek-reasoner，控制成本和响应速度。
• 知识库增强：把群聊高频问题的答案写入向量库，先检索再生成。
• 迁移到 NoneBot2：如果后续要做的功能越来越多，可以基于本文的理解迁移到插件化框架。

注意：任何聊天机器人都要遵守目标平台的规则。控制发送频率、避免刷屏、不过度自动化打扰用户，既是对账号的保护，也是对平台秩序的尊重。

回到最初的问题：DeepSeek 接入 QQ 机器人，真正有价值的不是那一次 API 调用，而是你如何组织消息链路、会话状态和输出节奏。多段回复看起来只是把长文本切成几句，但它牵出的流式处理、句子边界、频率控制和并发锁，恰恰是聊天类应用最常见的工程问题。建议拿到代码后不要只跑通就结束，先做三件事：把配置全部外置，给机器人加上断线重连，然后统计每天的 token 消耗。做完这三件事，你再去接微信、飞书或 Telegram，会发现核心逻辑几乎可以直接复用。第一步，先把 ping/pong 跑通。