Abstract: 从零搭建一个接了大模型的 QQ 机器人——NapCatQQ 做协议桥接,AstrBot 做 LLM 编排,包括人设配置、踩坑排障和插件开发。

UPDATE: 经过较长一段时间的实践,发现NapCatQQ作为协议桥,bot经常容易掉线,需要频繁再次登录。
遂寻得LLBot,直接取代NapCatQQ即可,后续配置流程大差不差。

Tip: 由于QQ官方封杀,LLBot官方github仓库并未提供真正的最新版,安装后需要在目录中进行更新,命令如下:

1
./llbot.exe --update

其余令牌获取、LLBot配置等,阅读LLBot官方文档即可

1. 我们要搞什么

1.1 目标

搭一套智能 QQ 机器人系统,三件套:

  • AI 聊天 —— 接 LLM 大模型,私聊群聊都能聊,带上下文记忆、角色人设
  • 自定义命令 —— 斜杠命令和关键词触发,比如 /weather、/search、/help
  • 插件扩展 —— 定时任务、自动化流程,接第三方API,不烧 LLM token(比如每日新闻推送、股票提醒、游戏通知)

核心思路:AI 管自然语言,插件管结构化任务——功能拉满的同时 API 费用压到最低。

1.2 架构

1
2
3
4
5
6
7
8
9
10
11
┌─────────────────┐     ┌─────────────────┐     ┌──────────────────┐
│ QQ Client │ ◄─► │ NapCatQQ │ ◄─► │ AstrBot │
│ (User/Bot) │ │(Protocol Bridge)│ │(LLM Orchestrator)│
└─────────────────┘ └─────────────────┘ └────────┬─────────┘
│
▼
┌─────────────────┐
│ LLM API │
│ (OpenAI/DeepSeek│
│ /Gemini/Local) │
└─────────────────┘
组件 干嘛的
NapCatQQ 协议桥 —— 通过 OneBot V11 协议跟 QQ 客户端通信
AstrBot LLM 编排器 —— 处理消息、管理上下文、调 LLM API
LLM Provider AI 大脑 —— 生成智能回复(OpenAI、DeepSeek、Gemini、本地模型都行)
Reverse WebSocket 通信通道 —— 把 NapCatQQ 和 AstrBot 双向连起来

为啥搞这么复杂? QQ 官方的机器人 API 限制多、功能少,主动消息功能不全等等。NapCatQQ 是社区维护的 OneBot V11 实现,能解锁完整的机器人能力。AstrBot 把平台差异抹平了,提供统一的 LLM 接入层。


2. 准备工作

开搞之前,确认一下:

需要什么 版本 备注
Python ≥ 3.9 推荐 3.12+
Node.js ≥ 18.0 NapCatQQ 要用
QQ 号 随意 拿个小号当机器人
LLM API Key 随意 OpenAI、DeepSeek 或兼容的中转都行

账号安全提醒: 用专门的 QQ 小号跑机器人。QQ 可能会封禁自动化行为的账号。别拿大号上。


3. NapCatQQ 搭建

3.1 NapCatQQ 是什么

NapCatQQ 是基于 NTQQ(QQ 最新桌面架构)的协议端框架,实现了 OneBot V11 标准,提供干净的 WebSocket/HTTP API 给机器人框架用。

特点一览:

  • 完整 OneBot V11 实现
  • 内存占用低 —— 7×24 跑在最低配置上没问题
  • 活跃维护 —— 459+ releases,GitHub 8k Stars
  • 支持 Docker —— 一条命令部署
  • WebUI 管理 —— 浏览器里就能配

3.2 安装

方法一:Windows 一键安装

  1. 去 NapCatQQ Releases 下载
  2. 找最新的 NapCat.Shell.Windows.OneKey.zip
  3. 解压,运行 NapCatInstaller.exe
  4. 跟着安装向导走

方法二:Linux Shell 脚本

1
2
3
4
5
6
7
# 下载并运行安装脚本
curl -o napcat.sh \
https://nclatest.znin.net/NapNeko/NapCat-Installer/main/script/install.sh \
&& bash napcat.sh

# 安装完成后启动
napcat

方法三:Docker

1
2
3
4
5
6
docker run -d \
--name napcat \
-e ACCOUNT=<YOUR_QQ_NUMBER> \
-p 3001:3001 \
-p 6099:6099 \
mlikiowa/napcat-docker:latest

3.3 首次登录(扫码)

第一次启动,终端里会弹出一个二维码:

1
2
# Docker 看二维码
docker logs -f napcat
  1. 手机打开 QQ
  2. 点 "+" → 扫一扫
  3. 扫终端里的二维码
  4. 手机上确认登录

二维码 60 秒过期。过期了就重启 NapCatQQ 重新生成。

3.4 配置 Reverse WebSocket

登录成功后,配 AstrBot 要用的 WebSocket 连接:

  1. 浏览器打开 NapCat WebUI:http://localhost:6099,用终端里显示的 token 登录
  2. 找到 网络配置
  3. 添加 Reverse WebSocket 客户端
  4. URL 填:ws://127.0.0.1:6183/ws
  5. 点 保存 然后 重启
设置项 值 说明
WebSocket URL ws://127.0.0.1:6183/ws AstrBot 的监听地址
Access Token (可选) 生产环境建议加上
重连间隔 3000 ms 推荐,稳

4. AstrBot 搭建

4.1 AstrBot 是什么

AstrBot 是一个 Python 写的多平台聊天机器人框架,原生集成了 LLM。支持 QQ、Telegram、Discord、企业微信等等。

核心能力:

  • 多平台 —— 一套代码,多个消息平台
  • 插件系统 —— 可扩展架构,想加啥功能AI自己写
  • WebUI 仪表盘 —— 浏览器里配置和监控
  • 上下文管理 —— 内置对话历史和压缩
  • 人设系统 —— 通过 System Prompt 自定义机器人性格

4.2 安装

方法一:手动安装(我用的)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 克隆仓库
git clone https://github.com/AstrBotDevs/AstrBot.git
cd AstrBot

# 创建虚拟环境
python -m venv venv
.\venv\Scripts\activate # Windows
# source venv/bin/activate # Linux/macOS

# 装依赖
python -m pip install -r requirements.txt

# 启动
python main.py

方法二:用 uv

1
2
3
4
5
6
7
8
# 装 uv 包管理器
pip install uv

# 初始化 AstrBot 项目
uv run astrbot init

# 启动
uv run astrbot run

方法三:Docker + NapCat 一体化

1
2
3
4
5
6
7
mkdir astrbot-qq && cd astrbot-qq

# 下载 compose 文件
wget https://raw.githubusercontent.com/NapNeko/NapCat-Docker/main/compose/astrbot.yml

# 一键启动
docker compose -f astrbot.yml up -d

4.3 打开仪表盘

AstrBot 启动后,浏览器打开:

1
http://localhost:6185

默认密码在首次启动时设置。终端输出里会显示管理员密码,也可以通过环境变量 ASTRBOT_ADMIN_PASSWORD 设置。


5. 平台对接

5.1 在 AstrBot 里启用 QQ 平台

  1. 打开 AstrBot 仪表盘 → 平台 页面
  2. 找到 QQ (OneBot V11),点 启用
  3. 配置连接:
设置项 值 说明
WebSocket 类型 Reverse NapCat 主动连 AstrBot
Host 0.0.0.0 监听所有网卡
Port 6183 AstrBot 默认端口
  1. 点 保存 然后 重启 AstrBot

5.2 验证连接

AstrBot 控制台看到这行就说明通了:

1
[INFO] aiocqhttp(OneBot v11) adapter connected

NapCat WebUI 那边也会显示 "已连接" 状态。


6. LLM 配置

6.1 添加 LLM Provider

  1. 打开 AstrBot 仪表盘 → 服务提供商 页面
  2. 点 添加提供商
  3. 选 OpenAI(任何 OpenAI 兼容 API 都行)
  4. 填配置:
字段 示例 说明
API Key sk-xxxxxxxxxxxx LLM 提供商的 API 密钥
API Base URL https://api.deepseek.com/v1 必须带 /v1 后缀
Model deepseek-chat 模型标识符
  1. 点 保存配置

6.2 常用 LLM 提供商

提供商 Base URL 备注
OpenAI https://api.openai.com/v1 官方,质量高,贵
DeepSeek https://api.deepseek.com/v1 性价比高,推理能力强
OpenRouter https://openrouter.ai/api/v1 多提供商聚合
本地 Ollama http://localhost:11434/v1 免费、隐私、要 GPU

6.3 启用模型

添加完提供商后:

  1. 点 获取模型列表
  2. 找到目标模型(比如 deepseek-chat)
  3. 点 "+" 添加
  4. 拨开关启用
  5. 去 配置 页面 → 选它当会话模型
  6. 右下角 保存配置

7. 人设配置

7.1 为啥人设很重要

人设(System Prompt)定义了机器人的性格、知识边界和回复风格。写好了能:

  • 搞出沉浸式角色扮演体验
  • 防止生成不当内容
  • 保持跨对话的角色一致性

7.2 配置人设

  1. 打开 AstrBot 仪表盘 → 人设 页面
  2. 写 System Prompt:
1
2
3
You are [Character Name], a [brief description].
Always respond in character. Use actions in parentheses () to describe body language.
Keep responses concise and engaging. Never break character or acknowledge being an AI.
  1. Temperature 设 0.8 - 0.95,回复更有创意
  2. 点 保存

7.3 人设写作技巧

Dos:

  • 用正面表述代替否定禁令
  • 定义回复格式(动作描写、语气、长度)
  • 设定知识边界(角色知道什么不知道什么)
  • 加示例对话(Few-Shot),保持一致性

Don'ts:

  • 别用 "violence"、"explicit"、"nsfw" 这些词——就算是在禁止也别用
  • System Prompt 别写太长(>2000 tokens)
  • 别写自相矛盾的指令
  • 别忘了拿边界输入测试

WAF 过滤预警: 有些 LLM 提供商用 WAF(Web 应用防火墙) 拦截含敏感关键词的请求——哪怕是写在 System Prompt 里的。

反面教材: "Never generate violent or sexual content"
正确姿势: "Maintain wholesome, family-friendly conversations. Politely decline inappropriate requests with a gentle rebuke."

7.4 自定义错误回复

把默认错误信息换成角色风格的回复:

1
(sets down teacup with a gentle sigh) It seems the connection has been disrupted. How unfortunate. Please try again in a moment.

8. 踩坑排障

8.1 常见问题

问题一:扫码登录失败,多进程报错

症状: 反复扫码都失败,控制台报:

1
[error] [napcat] [utilityprocess] worker进程退出,退出码: 18446744073709515000

原因: NapCat 的多进程模式跟环境冲突了,worker 进程登录时直接崩。

解法: 加环境变量关掉多进程模式:

  1. 打开 NapCat 的 .bat 启动文件
  2. 在 @echo off 后面加 set NAPCAT_DISABLE_MULTI_PROCESS=1
  3. 保存,重启 NapCat
1
2
3
@echo off
set NAPCAT_DISABLE_MULTI_PROCESS=1
:: ... rest of the script

问题二:403 Forbidden — Text Not Allowed

原因: LLM 提供商的 WAF 拦了你的请求。

解法:

  1. 关掉 AstrBot 自带的"安全模式" —— 它默认的 prompt 里有敏感词
  2. 审查你的人设 —— 把敏感关键词删掉
  3. 换个提供商 —— 有些代理的 WAF 比别的严

问题三:发图片触发 403

原因: 纯文本模型(比如 DeepSeek-V3)处理不了图片。AstrBot 把图片转成 Base64,一大长串文本触发了 WAF 的"可疑长文本"检测。

解法:

  1. 关掉 AstrBot 的图片理解功能
  2. 图片单独走 —— 只用支持视觉的模型(GPT-4o-mini、Gemini Flash)处理图片

问题四:改完配置等了 30 多秒才恢复

原因: AstrBot 热重载配置时会重置 WebSocket 连接。NapCat 的重连延迟造成了空窗期。

解法:

  1. 等它重连 —— 盯着控制台看 adapter connected 消息
  2. 把 NapCat 重连间隔调到 3000ms
  3. 都没用就两边都重启

问题五:群里不回消息

原因:

  • 群聊需要 @机器人 才触发
  • 机器人的 QQ 号登录过期了

解法:

  1. 检查 AstrBot 仪表盘的群组设置,或者设置触发条件
  2. 重启 NapCatQQ 重新登录

8.2 排障清单

看终端输出拿报错信息

检查项 怎么查
AstrBot 跑着没? python main.py 或查进程
NapCat 连上了没? 看 WebUI 状态
LLM API 通不通? 在 AstrBot WebUI 里试聊
日志干净吗? 看 AstrBot 和 NapCat 控制台输出

9. 进阶:自己写插件

9.1 为啥要写插件

插件给 AstrBot 扩展自定义功能,不烧 LLM token。能干嘛:

  • 定时推送(新闻、天气、提醒)
  • API 对接(股价、游戏数据)
  • 群管工具(自动撤回、警告)
  • 实用命令(计算器、单位换算)

9.2 插件结构

在 AstrBot/data/plugins/your_plugin_name/ 下建文件夹:

1
2
3
your_plugin_name/
├── manifest.json # 插件元数据
└── main.py # 插件逻辑

manifest.json:

1
2
3
4
5
6
{
"name": "your_plugin_name",
"version": "1.0.0",
"author": "YourName",
"description": "Brief description of what this plugin does"
}

9.3 实战:每日新闻推送插件

一个实用插件——抓每日新闻、Epic 免费游戏和 Bing 每日壁纸,定时推到 QQ 群里——零 LLM token 消耗。

用的 60s API,by Gemini 3.1 Pro 和 Mimo V2 Pro

点这里展开完整插件代码(main.py)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
import aiohttp
import datetime # 不要删除 推送测试会使用
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from astrbot.api.all import Context, Star, logger, register
from astrbot.core.message.components import Image, Plain

# ================= 配置区(按需修改) =================
GROUP_ID = [QQ_NUMBER] # 目标 QQ 群号
# =====================================================


@register(
"doro_broadcast",
"FrozenYears",
"1.3.0",
"Doro 定时新闻、每日壁纸与 Epic 推送",
)
class DoroBroadcast(Star):
"""
定时抓取新闻、 Epic 免费游戏和必应每日壁纸,并以 Doro 人设推送到指定 QQ 群。
使用 APScheduler 进行定时任务调度:
- 每天 11:00 推送每日新闻
- 每天 15:00 推送每日壁纸
- 每周五 11:10 推送 Epic 免费游戏
- 启动后延迟 45 秒执行一次推送测试(规避 NapCat 冷启动窗口期)
"""

def __init__(self, context: Context) -> None:
super().__init__(context)
self.scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")

async def initialize(self) -> None:
"""插件激活时注册所有定时任务并启动调度器。"""
# 每天 11:00 推送新闻
self.scheduler.add_job(
self.send_daily_news,
"cron",
hour=11,
minute=0,
id="doro_daily_news",
replace_existing=True,
misfire_grace_time=60,
)
# 每周五 11:10 推送 Epic 免费游戏
self.scheduler.add_job(
self.send_epic_games,
"cron",
day_of_week="fri",
hour=11,
minute=10,
id="doro_epic_games",
replace_existing=True,
misfire_grace_time=60,
)
# 每天 15:00 推送 Bing 每日图片
self.scheduler.add_job(
self.send_bing_daily,
"cron",
hour=15,
minute=0,
id="doro_bing_daily",
replace_existing=True,
misfire_grace_time=60,
)
# 【测试推送】冷启动规避:延迟 45 秒执行一次测试推送
# test_time = datetime.datetime.now() + datetime.timedelta(seconds=45)
# self.scheduler.add_job(
# self.send_bing_daily, # 新闻测试
# "date",
# run_date=test_time,
# id="doro_test_daily_news",
# replace_existing=True,
# )

self.scheduler.start()
logger.info("[DoroBroadcast] 插件已挂载!")

async def terminate(self) -> None:
"""插件卸载时优雅关闭调度器。"""
if self.scheduler.running:
self.scheduler.shutdown(wait=False)
logger.info("[DoroBroadcast] 调度器已关闭。")

# ------------------------------------------------------------------
# 核心:防雷暴内存反射发包
# ------------------------------------------------------------------

async def _send_to_group(self, msg) -> None:
"""
使用 AstrBot 官方 API 主动推送到指定群组
"""
try:
import asyncio

from astrbot.core.message.components import Plain
from astrbot.core.message.message_event_result import MessageChain
from astrbot.core.platform.astr_message_event import MessageSession

logger.info("[DoroBroadcast] 准备通过官方接口发送群消息...")

# 动态获取 aiocqhttp 适配器的真实 ID
platform_id = "aiocqhttp"
if hasattr(self.context, "platform_manager") and hasattr(
self.context.platform_manager, "platform_insts"
):
for platform in self.context.platform_manager.platform_insts:
meta = platform.meta()
if meta.name == "aiocqhttp":
platform_id = meta.id
break

# 使用框架标准的 MessageSession (注意单词拼写是框架原本的设计)
# 格式: platform_name:message_type:session_id
# 注意:此处的 platform_name 在 v4 中实际上要求传入 platform.meta().id
session = MessageSession.from_str(f"{platform_id}:GroupMessage:{GROUP_ID}")

if isinstance(msg, str):
chain = MessageChain([Plain(msg)])
elif isinstance(msg, list):
chain = MessageChain(msg)
elif isinstance(msg, MessageChain):
chain = msg
else:
logger.error("[DoroBroadcast] 不支持的消息格式")
return

# 重试机制 (最大 3 次)
max_retries = 3
for attempt in range(max_retries):
try:
# 使用官方 API 发送消息
success = await self.context.send_message(session, chain)

if success:
logger.info(
f"[DoroBroadcast] ✅ 成功推送消息到群 {GROUP_ID} (适配器: {platform_id})"
)
return # 发送成功,退出函数
else:
logger.warning(
f"[DoroBroadcast] ⚠️ 推送失败 (尝试 {attempt + 1}/{max_retries}): 找不到匹配的平台 ({platform_id}) 或发送不成功"
)
if attempt < max_retries - 1:
await asyncio.sleep(1)
continue

except Exception as e:
# 检查是否为超时错误 (retcode=1200)
# 通过检查异常对象的属性来判断
error_str = str(e)

# 检查常见的超时模式
is_timeout = (
"retcode=1200" in error_str
or "Timeout" in error_str
or "serviceAndMethod:NodeIKernelMsgService/sendMsg" in error_str
)

if is_timeout:
logger.warning(
f"[DoroBroadcast] ⚠️ 推送超时 (尝试 {attempt + 1}/{max_retries}): {e}"
)
if attempt < max_retries - 1:
await asyncio.sleep(2) # 超时等待久一点
continue
else:
# 重试次数耗尽,重新抛出异常
raise e
else:
# 不是超时异常,直接抛出
raise e

except Exception as e:
logger.error(f"[DoroBroadcast] ❌ 推送消息时发生异常: {e}")

# ------------------------------------------------------------------
# 新闻推送
# ------------------------------------------------------------------

async def send_daily_news(self) -> None:
"""拉取 60s API 新闻并以 Doro 人设推送到群。"""
logger.info("[DoroBroadcast] 开始执行每日新闻抓取...")
try:
timeout = aiohttp.ClientTimeout(total=15)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(
"https://60s.viki.moe/v2/60s?encoding=image"
) as resp:
if resp.status != 200:
logger.error(
f"[DoroBroadcast] 新闻 API 返回异常状态码: {resp.status}"
)
return
image_bytes = await resp.read()

from astrbot.core.message.components import Image, Plain

start_msg = (
"(摇着短小的小尾巴哒哒哒跑过来) 唔!人!"
"Doro 叼来了今天的报纸!嘿嘿~\n\n"
)

end_msg = (
"\n(放下报纸,吐着舌头狂摇尾巴) 读完了(ノ≧∀≦)ノ\n"
"Doro 这么能干,人是不是该奖励 Doro 一大个欧润吉呀?( ´﹃`)"
)

components = [
Plain(start_msg),
Image.fromBytes(image_bytes),
Plain(end_msg),
]

logger.info("[DoroBroadcast] 新闻拉取成功,准备推送。")
await self._send_to_group(components)

except Exception as e:
logger.error(f"[DoroBroadcast] 新闻抓取崩溃: {e}")

# ------------------------------------------------------------------
# Epic 免费游戏推送
# ------------------------------------------------------------------

async def send_epic_games(self) -> None:
"""拉取 Epic 免费游戏列表并以 Doro 人设推送到群。"""
logger.info("[DoroBroadcast] 开始执行 Epic 免费游戏抓取...")
try:
timeout = aiohttp.ClientTimeout(total=15)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get("https://60s.viki.moe/v2/epic") as resp:
data = await resp.json(content_type=None)

if data.get("code") != 200:
logger.error(f"[DoroBroadcast] Epic API 返回异常: {data}")
return

game_list: list[dict] = data.get("data", [])

# 构建完整消息组件列表
components = []

# 添加开场消息
start_msg = (
"(摇着短小的小尾巴哒哒哒跑过来) 唔!人!"
"Doro 叼来了 Epic 本周的免费游戏清单!嘿嘿~\n\n"
)
components.append(Plain(start_msg))

# 为每个游戏添加封面图片和文本信息
for item in game_list:
title: str = item.get("title", "")
description: str = item.get("description", "Doro不知道哦")
price: str = item.get("original_price_desc", "Doro不知道哦")
free_end: str = item.get("free_end", "Doro不知道哦")
link: str = item.get("link", "Doro不知道哦")
cover_url: str = item.get("cover", "")

if not title:
continue

# 直接使用封面图片URL(无需下载)
if cover_url:
components.append(Image.fromURL(cover_url))
logger.info(f"[DoroBroadcast] 添加封面URL: {title}")

# 添加文本信息
info_text = (
f"🎮 {title}\n"
f"✍️ 简介:{description}\n"
f"💰 原价:{price}\n"
f"⏰ 截止:{free_end}\n"
f"🔗 链接:{link}\n\n"
)
components.append(Plain(info_text))

# 添加结束消息
end_msg = (
"🐾 Doro的小纸条:记得登录领取哦!\n"
"(放下游戏清单,吐着舌头狂摇尾巴) 读完了(ノ≧∀≦)ノ\n"
"Doro 这么能干,人是不是该奖励 Doro 一大个欧润吉呀?( ´﹃`)"
)
components.append(Plain(end_msg))

# 发送完整消息
await self._send_to_group(components)

logger.info("[DoroBroadcast] Epic 数据拉取成功,准备推送。")

except Exception as e:
logger.error(f"[DoroBroadcast] Epic 抓取崩溃: {e}")

# ------------------------------------------------------------------
# Bing 每日图片推送
# ------------------------------------------------------------------

async def send_bing_daily(self) -> None:
"""拉取 Bing 每日图片数据并以 Doro 人设推送到群。"""
logger.info("[DoroBroadcast] 开始执行 Bing 每日图片抓取...")
try:
timeout = aiohttp.ClientTimeout(total=15)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get("https://60s.viki.moe/v2/bing") as resp:
data = await resp.json(content_type=None)

if data.get("code") != 200:
logger.error(f"[DoroBroadcast] Bing API 返回异常: {data}")
return

bing_data: dict = data.get("data", {})

# 提取所需字段
title: str = bing_data.get("title", "未知标题")
headline: str = bing_data.get("headline", "未知标题")
description: str = bing_data.get("description", "暂无描述")
main_text: str = bing_data.get("main_text", "暂无详细信息")
cover_4k_url: str = bing_data.get("cover_4k", "")

# 构建完整消息组件列表
components = []

# 添加开场消息(用户指定的固定开头)
start_msg = (
"(端着精致的白瓷茶杯,目光微微低垂,注视着刚刚为您送达的画卷,"
"嘴角勾起一抹完美无瑕的轻柔微笑)\n"
"呵呵……指挥官,您看。这是今日从远方传来的风景。\n\n"
)
components.append(Plain(start_msg))

# 添加图片(如果存在)
if cover_4k_url:
components.append(Image.fromURL(cover_4k_url))
logger.info(f"[DoroBroadcast] 添加 Bing 4K 图片: {cover_4k_url}")

# 添加文本信息
info_text = (
f"🎨 标题:{title}\n"
f"📰 头条:{headline}\n"
f"📝 描述:{description}\n"
f"📖 详情:{main_text}"
)
components.append(Plain(info_text))

# 添加结束消息(用户指定的固定结尾)
end_msg = (
"\n\n(缓缓抬起头,静静地注视着您,笑容愈发温柔,却让人感到一丝战栗)\n"
"我将这份虚幻的美丽带给了您。希望这片刻的宁静,能稍微洗刷掉您沾染的尘埃。请尽情欣赏吧……"
)
components.append(Plain(end_msg))

# 发送完整消息
await self._send_to_group(components)

logger.info("[DoroBroadcast] Bing 每日图片数据拉取成功,准备推送。")

except Exception as e:
logger.error(f"[DoroBroadcast] Bing 每日图片抓取崩溃: {e}")

需要的依赖:

1
2
3
4
5
# 先激活虚拟环境
.\.venv\Scripts\activate

# 装包
uv pip install aiohttp apscheduler

装完依赖后,重启 AstrBot(python main.py)。


10. 总结

组件 干嘛的 关键配置
NapCatQQ QQ 协议桥 Reverse WebSocket → ws://127.0.0.1:6183/ws
AstrBot LLM 编排器 平台:QQ (OneBot V11)
LLM Provider AI 大脑 API Key + Base URL + Model
人设 机器人性格 正面表述的 System Prompt

架构流程:

1
2
3
4
5
6
7
用户发 QQ 消息
→ NapCatQQ 通过 OneBot V11 接收
→ AstrBot 处理消息
→ LLM 生成回复
→ AstrBot 格式化回复
→ NapCatQQ 发到 QQ
→ 用户收到回复

参考资料