把 DeepSeek 接入 QQ 机器人看起来只是把 API 地址和 Key 换一换实际做起来才会发现一次完整的“拟人化聊天”要同时处理消息协议、多轮上下文、回复节奏和多段发送逻辑。尤其“多段回复”这个需求很多人一开始不理解模型明明一次就能生成整段回复为什么还要拆成好几条发出去等真正把机器人拉进群里看到一整屏又硬又长的公告式回答才会明白分段不只是形式问题而是拟人化体验的一部分。下面会从零搭一个可运行的 DeepSeek QQ 聊天机器人先跑通 API再接入 OneBot 11 协议的消息端接着实现人设、上下文和回复节奏最后重点讲多段回复的拆分策略、发送控制和排错方法。整个项目不依赖付费网关也不需要复杂的分布式架构代码量控制在几百行内适合想理解完整链路而不是只会套模板的开发者。1. 先理清 QQ 机器人接入 DeepSeek 的完整链路1.1 一条消息从 QQ 群到 DeepSeek 再回来经过哪些节点写代码之前要先画清楚链路。用户在 QQ 群里发一句“你好”这句消息并不会直接到达 DeepSeek。它要经过这样一条路径用户通过 QQ 客户端把消息发到 QQ 服务器。运行在你本机或服务器上的 OneBot 协议实现以机器人账号的身份登录 QQ并把收到的消息转换成标准事件。你的 Python 进程通过 WebSocket 连接到这个协议实现接收消息事件。Python 进程把用户消息和该会话的历史记录一起组装成 messages 列表调用 DeepSeek API。DeepSeek 返回回复文本或者以流式方式逐步返回文本增量。Python 进程把完整回复按句子边界切分成多段逐条通过 OneBot 操作接口发送到原群。这里最关键的一点是DeepSeek 只负责“文本到文本”它不知道 QQ、不知道群号、不知道消息 ID。所有协议处理、会话隔离、发送节奏都必须由你自己的服务完成。很多初学项目把代码写得像是“调一次接口就结束”结果上线后各种奇怪现象回得慢、回得生硬、多人群里上下文串台、消息发不出去本质都是链路里少了某一环。1.2 谁负责“拟人化”拟人化不是一个模型开关而是三个环节共同作用的结果系统提示词负责定义语气、句式和边界让模型不要一开口就是客服腔。会话管理负责记住上下文让机器人不会每句话都像失忆一样重新开始。回复节奏负责控制“什么时候说、说多长、分几条说”让输出看起来像人在打字。模型本身只负责生成文本。它生成的文本可能已经比较口语化但如果你的程序一次性把 500 个字甩到群里再口语化的内容也会显得像一个公告。这就是多段回复逻辑存在的意义它把模型的输出重新组织成符合聊天场景的节奏。1.3 三种接入姿势怎么选在动手前先明确要采用哪种开发方式。下面三种方式都能做差别在控制力、开发速度和排错成本。接入方式适合场景优点要注意的问题纯手写 OneBot 客户端本文采用想理解协议和完整链路代码可控排错方便没有黑盒协议细节需要自己处理NoneBot2 框架快速开发功能机器人插件生态丰富异步模型成熟底层被封装协议问题不好定位现成第三方机器人项目只想立刻跑起来上手最快依赖不确定扩展和排查空间小本文选择纯手写方式不是因为框架不好而是因为“保姆级教学”的目标是把链路讲清楚。手写一遍之后你再去用 NoneBot2 或任何封装框架都会知道它内部在做什么。2. 环境准备先对齐账号、依赖和目录再写代码2.1 前置条件清单开始安装之前先把下面这些条件准备好缺一项都会在中途卡住。前置项说明DeepSeek API Key在 DeepSeek 开放平台注册账号后创建注意账户余额和接口限流机器人 QQ 账号建议使用小号或测试号避免影响日常账号Python 3.10本文代码基于 asyncio 和较新的 openai SDKOneBot 11 协议实现例如 NapCat、LLOneBot、Lagrange 等任选一个即可运行环境本机可用于学习和测试生产部署需要服务器并保证进程常驻注意机器人账号和日常账号分离是底线。开发过程中难免出现消息错发、循环调用、异常刷屏等情况用小号测试可以把风险隔离在可控范围内。2.2 安装 Python 依赖核心依赖只有两个openai和websockets。前者用于调用 DeepSeek 的 OpenAI 兼容接口后者用于连接 OneBot 协议实现的 WebSocket 服务端。python -m venv .venv source .venv/bin/activate pip install openai websocketsWindows 下激活虚拟环境的命令是.venv\Scripts\activate。安装完成后可以把版本固定到requirements.txtopenai1.30.0 websockets12.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_keysk-your-key-here, base_urlhttps://api.deepseek.com, ) response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个群聊助手回答简洁自然不超过100字。}, {role: user, content: 你好简单介绍一下自己。}, ], temperature0.9, max_tokens200, ) print(response.choices[0].message.content)这段代码解决的是“能不能调通”的问题。执行后控制台会打印模型的回复文本。如果没有报错说明 Key、base_url、模型名和网络连通性都没有问题。3.2 返回结构里哪些字段有用DeepSeek 兼容 OpenAI 的返回结构核心字段如下字段含义用途choices[0].message.content模型生成的回复正文直接发送给用户choices[0].message.reasoning_content推理模型的思考内容多轮对话时需要特殊处理usage.prompt_tokens输入消耗的 token 数统计成本和排查超长问题usage.completion_tokens输出消耗的 token 数统计成本和排查超长问题一个典型的响应 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( modeldeepseek-chat, messages[ {role: system, content: 你是一个群聊助手。}, {role: user, content: 给我讲讲怎么做番茄炒蛋。}, ], streamTrue, ) 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 建议用fgroup:{group_id}这样不同群互不干扰。如果希望更细粒度可以用fgroup:{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 为什么要分段先解决“为什么”。模型一次调用能生成完整回复不需要你费劲拆分。但实际聊天场景里一次性发超长文本有三个问题体验问题一大段文字像公告不像聊天。长度问题QQ 对超长文本有限制超过一定长度可能发送失败或显示异常。节奏问题逐句发送并带间隔能模拟“正在打字”的节奏让互动更像真人。这里要澄清一个常见误解分段不是把一段文字机械切成几块而是在语义完整的句子边界处切分并且每段之间有合理的发送间隔。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_keyconfig.DEEPSEEK_API_KEY, base_urlconfig.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( modeldeepseek-chat, messagesmessages, streamTrue, temperature0.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.pyDEEPSEEK_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 15BOT_SELF_ID需要填成机器人账号的实际 QQ 号。不填或填 0机器人就可能把自己发的消息当成用户消息形成自我对话死循环。7.2 主程序 main.pyimport asyncio import json import random import websockets from openai import AsyncOpenAI import config deepseek AsyncOpenAI( api_keyconfig.DEEPSEEK_API_KEY, base_urlconfig.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( modelconfig.DEEPSEEK_MODEL, messagesmessages, streamTrue, temperature0.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 fgroup:{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_sizeNone) 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 启动和验证步骤按下面的顺序执行每步都确认结果启动协议实现并登录机器人账号确认 WebSocket 服务在3001端口监听。运行python main.py日志显示“连接成功”。在群里发“ping”确认机器人回“pong”。在群里发一句正常聊天内容观察 DeepSeek 回复是否被分成多条发送。连续发两条消息确认第二条会在第一条完整发送完后才被处理。7.4 预期日志正常情况下日志大致如下[INFO] 正在连接 OneBot WebSocket: ws://127.0.0.1:3001 [INFO] 连接成功 [INFO] 收到群 10001 消息: 你好 [INFO] 分段发送: 你好呀我是群里的聊天机器人。 [INFO] 等待 0.9s [INFO] 分段发送: 有什么想聊的可以直接说。如果只看到“收到消息”而没有后续问题在 DeepSeek 调用环节。如果看到“分段发送”但群里没消息问题在发送环节。日志就是用来做这个阶段判断的。8. 从现象找原因这一套最容易踩的坑8.1 收不到消息现象常见原因检查方式处理建议群里发消息程序没反应WebSocket 没连上看启动日志是否显示“连接成功”确认协议实现的端口和 URL 一致连接成功但收不到消息事件类型没勾选 message在协议实现里打开事件日志勾选 message 上报程序收到但没输出post_type过滤错误打印完整事件 JSON检查事件字段名和大小写记住一个判断原则先看协议实现里能不能看到消息事件。能看到问题在 Python 程序看不到问题在协议实现配置。8.2 消息发不出去现象常见原因检查方式处理建议action 没有响应action 名称拼错对照 OneBot 11 规范确认群消息用send_group_msg发送后显示发送失败消息体结构不对打印实际发送的 JSON检查 params 里字段名是否拼错单条内容太长没有做分段或段长过大看发送的文本长度调低MIN_SPLIT_LEN和分段上限账号出现异常提示发送频率过高查看协议实现日志降低频率增加间隔内容保持合规发送失败时第一件事是打印你真正发出去的 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把思考内容丢掉了下一轮调用会被接口拒绝。处理方式按顺序排查聊天机器人优先使用deepseek-chat这类非思考模式不涉及该字段最简单。如果必须使用思考模式保存历史时不要把 assistant 消息拆字段把完整的 assistant 消息对象存下来下一轮原样放回。如果当前 session 的历史已经混用了普通模式和思考模式清空该会话历史再试。如果你的请求经过第三方网关或中转服务先直连官方接口复现确认是官方拒绝还是中间层改写消息体导致。注意遇到 400 时先把链路拆开直连 DeepSeek 官方接口是否正常如果正常再检查自己的消息构造和上下文保存逻辑如果有中间网关最后检查网关是否改写了消息体。8.4 多段回复被合并或像刷屏现象原因处理建议几段内容几乎同时到达分段之间没有延时或延时太短每段发送后 sleep 0.6 到 1.2 秒一段特别长长句里缺少标点没触发切分增加硬切逻辑控制段长上限回应很快但内容碎短句被单独切出调大MIN_SPLIT_LEN过短残句并入上一段消息被平台拦截连续发送条数过多降低长回复频率限制单次回复的最大段数多段回复的调参没有标准答案跟模型输出风格强相关。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 学习环境到生产环境的差距本文的完整代码可以在本机跑通但它是一个学习版本不是生产版本。两者差距很大维度学习环境生产环境会话存储内存 dict重启即丢Redis 或数据库多实例共享配置管理写在 config.py环境变量或配置中心密钥隔离日志print结构化日志记录消息 ID、耗时、token监控无延迟、错误率、token 消耗、队列积压稳定性手动重启systemd 或 Docker 守护断线自动重连内容安全只有系统提示词关键词过滤、敏感内容识别、频率限制账号风险测试小号与正式业务隔离遵守平台规则生产环境至少要补上 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 跑通。