blog_v1.1_从能跑到稳定

我把昨天那个 AI Agent，今天升级成了"每天自己跑"的版本

接上一篇：我用一周时间从零做了一个 AI 信息采集 Agent

如果说 v1.0 是"能跑给你看"，v1.1 就是"它每天自己跑给我看"。
这中间隔的不只是几行代码，是工程思维的彻底升级。

楔子：从"demo"到"系统"的距离

昨天 v1.0 发布的时候，我拍着胸脯说"项目跑通了"。

但今天早上一打开 logs，发现 v1.0 的"跑通"其实是这样的：

✅ 1 条数据入库（Rickroll MV） ⚠️ Monitor 扫 2 个 UP 主，1 个 412 失败 ⚠️ 那 1 条入库的数据，up_name 字段是 null ⚠️ 数据库里有 1 条 FAILED 任务，永远停在那 ⚠️ 想跑必须我手动 python main.py

这是"demo"。这不是"系统"。

demo 和系统的区别，我后来想了想，是这 5 件事：

今天一整天我做的事情，就是把项目从左边推到右边。

目录

• 一、开工前的"诊断"
• 二、阶段 1：让 LLM 不再因为 JSON 截断挂掉
• 三、阶段 2：B 站反爬的真正根因
• 四、阶段 2.5：意外修复的 async bug
• 五、阶段 3：失败任务的自动复活
• 六、阶段 4：不用我管它，每天自己跑
• 七、阶段 5：发布 v1.1
• 八、复盘：今天到底学到了什么

一、开工前的"诊断"

写代码之前，我做的第一件事是跑一次 v1.0 的 main.py，把所有问题摆在桌面上。

这是工程师跟"急于动手"的人最大的区别：先看清楚，再下刀。

跑出来的真实日志：

[Stage 1] Monitoring targets for new content... [Monitor] Scanning UID 285286947 ... Found 0 latest videos [Monitor] Scanning UID 1333131174 ... Error: 412 Precondition Failed [Stage 2] No pending tasks. [Stage 3] No collected tasks waiting for LLM. >>> Pipeline completed in 0.5s New URLs: 0 Collected: 0 Processed: 0

数据库现状：

COMPLETED: 1 ← 昨天的 Rickroll MV FAILED: 1 ← 不知道为什么挂的

诊断完毕，问题清单：

1. 🔴Monitor 完全失灵（412 反爬）
2. 🔴Collector 选择器疑似过期（v1.0 那条数据 up_name 是 null）
3. 🟡LLM JSON 截断（max_tokens=2000 隐患还在）
4. 🟡FAILED 任务没人管（永远停在那）

重要的一步：我把这份诊断报告先发给我自己，作为今天工作的 baseline。改完后再跑一次同样的命令，对比前后变化——这是衡量"今天有没有真正改善"的唯一客观标准。

💡 教训：不要凭感觉判断"项目变好了"。用数字说话。

二、阶段 1：让 LLM 不再因为 JSON 截断挂掉

第一个修的是 LLM。原因很简单——这是最容易修的，最先做能立刻找回信心。

v1.0 的 processor.py 长这样：

response=client.chat.completions.create(model=self.model,messages=[...],temperature=0.2,max_tokens=2000# ⚠️ 长内容会被截断)raw_output=response.choices[0].message.content.strip()try:parsed=json.loads(raw_output)# ⚠️ 截断了就 raisereturnjson.dumps(parsed)exceptjson.JSONDecodeErrorase:print(f"[WARN] LLM returned invalid JSON:{e}")# ⚠️ print 不是 loggerreturnNone# ⚠️ 失败的原始输出没保存，无法复盘

我数了一下，这段代码里有 4 个不同层次的问题：

我重写了一版，把这 4 件事一起解决：

classLLMProcessor:MAX_TOKENS=4000# 翻倍INPUT_CHAR_LIMIT=8000# 输入也加大LLM_TIMEOUT_SEC=90# ⚠️ 后面踩坑后加的@staticmethoddef_safe_json_parse(raw_output):"""两段式解析，应对 LLM 各种抽风"""text=raw_output.strip()# 去掉 ```json 包裹iftext.startswith("```"):lines=text.splitlines()[1:-1]text="\n".join(lines).strip()# 第一次尝试：直接解析try:returnjson.loads(text),Noneexceptjson.JSONDecodeErrorase:first_err=str(e)# 第二次尝试：截取最外层大括号start,end=text.find("{"),text.rfind("}")ifstart!=-1andend>start:try:returnjson.loads(text[start:end+1]),Noneexceptjson.JSONDecodeErrorase:returnNone,f"first:{first_err}| bracket:{e}"returnNone,f"first:{first_err}| no balanced braces"def_dump_failure(self,raw_output,error_message):"""失败原文落盘，方便复盘"""ts=datetime.now().strftime("%Y%m%d_%H%M%S_%f")path=self.FAILURE_LOG_DIR/f"llm_fail_{ts}.txt"path.write_text(f"{error_message}\n---\n{raw_output}")

然后用 6 个边界测试验证_safe_json_parse：

# 1. 干净 JSON → ✅# 2. ```json 代码块包裹 → ✅# 3. 前后带解释文字"好的，结果是" → ✅# 4. 截断的 JSON → ✅ 优雅返回 None + error# 5. 空字符串 → ✅# 6. 代码块 + 尾部换行 → ✅

💡 教训：重写不是只改坏的那一行。每次重写都问自己——“这段代码隐含了什么假设？这些假设以后会不会变？”

比如这个 LLM 调用，原代码假设了 a) 输出永远是合法 JSON、b) 输出永远不超过 max_tokens、c) 失败信息不需要保存。把这些假设全部颠倒过来重新设计，就是 v1.1。

三、阶段 2：B 站反爬的真正根因

这是今天最难的部分。也是收获最大的部分。

v1.0 时我以为 412 是简单的"反爬"，加点 headers 就好了。但当我用 Playwright 打开浏览器，带着真实的 cookie 去访问那个 API，结果让我吓了一跳：

HTTP 200 { "code": -799, "message": "请先登录...", "data": null }

不是 HTTP 错误，是业务错误码 -799。

查了半天才发现：B 站 2024 年起对/x/space/arc/search接口加了WBI 签名机制——必须在请求里带上w_rid和wts两个参数，参数值由复杂的 MD5 + 二次混淆算法生成。光有 cookie 不够，必须有签名。

这是一个真正的硬技术墙：

传统反爬： HTTP 412 / 403 / 429 → 加 headers / 加 cookie 就能过 WBI 签名： HTTP 200 + code=-799 → 必须实现签名算法，cookie 完全无效

我在这里做了一个重要的工程决策：

不去硬磕签名算法。

因为 a) 算法会变，b) 维护成本高，c)我有更稳的路径——Playwright。

于是 monitor.py 的设计变成了"双路径 + 智能 fallback"：

deffetch_bilibili_urls(self,uid):# 第一路径：requests 试一下（快但脆弱）urls=self._fetch_via_requests(uid)ifurlsisnotNone:returnurls# 第二路径：Playwright fallback（慢但稳）returnself._fetch_via_playwright(uid)

关键是怎么判断要切换路径。我没用粗暴的 try/except，而是精确识别 B 站的业务错误码：

NEED_FALLBACK_CODES={-799,-412,-403,-509}ifdata.get("code")inNEED_FALLBACK_CODES:returnNone# 触发 fallbackifdata.get("code")==-111:return[]# UP 主真的不存在，不要 fallback

这种"精确识别 + 智能降级"的设计，是这阶段我学到最重要的事：

💡 教训：反爬不是一道墙，是一棵决策树。

HTTP 错误码、业务错误码、空响应、超时——每一种"失败"都对应不同的根因和不同的修复策略。粗暴的 try/except 会把这些信号全部抹平。

写一个错误码识别表，比写 100 行重试逻辑值钱。

实地探测的时候我还顺手发现了选择器全部过期：

.video-desc ❌ 已下线（v1.0 一直在用） .desc-content ❌ 已下线 .video-desc-container ✅ 新主选择器 #v_desc ✅ 新主选择器

collector.py 整个重写，每个字段都用 3-5 个备选选择器降级。

四、阶段 2.5：意外修复的 async bug

写完阶段 2 跑起来，cookie 注入成功，monitor 拿到 30 个新视频——我以为搞定了。

然后端到端跑main.py，崩了：

RuntimeError: This event loop is already running

定位发现：v1.0 时 main.py 是用asyncio.run(pipeline.run())启动的，但 stage1_monitor 是同步函数——v1.0 时不调用 Playwright fallback 所以无所谓，但 v1.1 我加的 Playwright fallback 内部用了 asyncio.run()：

# v1.1 我写的同步包装，错的def_fetch_via_playwright(self,uid):returnasyncio.run(self._fetch_via_playwright_async(uid))# ↑ 已经在 asyncio.run() 里了，再调一次 → 撞车

这就是著名的“event loop already running”反模式。

这个 bug 在 v1.0 不存在，是因为 v1.0 的 monitor 永远不会进 Playwright 路径。v1.1 引入 fallback 才让它显形。

修复方案是把 monitor 的 API 拆成同步 + 异步两个版本：

# 同步版（独立脚本调用）deffetch_bilibili_urls(self,uid):...returnself._fetch_via_playwright(uid)# 内部 asyncio.run# 异步版（main.py 这种 async 上下文调用）asyncdeffetch_bilibili_urls_async(self,uid):...returnawaitself._fetch_via_playwright_async(uid)# 直接 await

main.py 里改一行：

# Beforeadded=self.monitor.sync_targets(self.TARGET_UIDS)# Afteradded=awaitself.monitor.sync_targets_async(self.TARGET_UIDS)

💡 教训：新功能可能会让旧代码的潜在 bug 显形。

这个 async bug 在 v1.0 时一直存在，但因为永远不触发那条路径，无事发生。v1.1 触发了路径，bug 就来敲门了。

写代码的时候，永远要意识到你"没走过的路"——可能未来某天有人会走，那时候就不一定是你来修。

五、阶段 3：失败任务的自动复活

跑批的时候发现，火山方舟 kimi-k2.6 偶发会出现90 秒以上的长尾延迟——单条 LLM 调用卡住，整条流水线没法继续。

修了一个 timeout：

self.client.chat.completions.create(...timeout=90,)

但更深的问题是：失败的任务永远停在 FAILED 状态，没人管。

v1.0 的失败处理：

# 失败了 → 标记 FAILED → 然后呢？self.db.update_task_status(url,"FAILED")

没有"然后"。FAILED 任务永远不会被重新尝试。

我设计了一个智能重试机制。关键洞察是：

失败要分类。LLM 阶段失败和 collector 阶段失败需要不同的恢复策略。

defrequeue_failed(self,max_retry=3):"""智能恢复 FAILED 任务"""# 策略 1：已经有 raw_contents → LLM 阶段失败# 不用重新爬，直接把状态回到 COLLECTED 让 LLM 重试UPDATE task_queue SET status='COLLECTED'WHERE status='FAILED'AND retry_count<? AND url IN(SELECT url FROM raw_contents)# 策略 2：没有 raw_contents → collector 阶段失败# 需要重新爬，回到 PENDINGUPDATE task_queue SET status='PENDING'WHERE status='FAILED'AND retry_count<?# 策略 3：retry_count >= 3 → 放弃，保留 FAILED# （不做任何 UPDATE）

main.py启动时自动调用：

asyncdefrun(self):# v1.1：每次启动先回滚 FAILEDself.db.requeue_failed()# 然后正常跑流水线new_urls=awaitself.stage1_monitor()...

这个机制的真实威力：v1.0 时代留下的那条 FAILED 任务（昨天 LLM 截断的那条），今天我什么都没做，就被自动复活并处理成功了：

[Stage 3] (1/2) Cleaning: https://www.bilibili.com/video/BV15f4y1A7Hi [Processor] LLM extracted OK. Title: 【中英字幕】《baby》- Justin Bieber 贾斯汀比伯 1080P超高清MV [Stage 3] OK

那一刻系统第一次有了"自愈能力"。

💡 教训：让系统会自我修复，比让代码不出错更重要。

错误是必然的（网络抖、LLM 抽风、DOM 改版）。承认它必然，设计一个能自愈的机制——比花 10 倍时间追求"零错误"实际多了。

六、阶段 4：不用我管它，每天自己跑

到这里，项目已经能稳定运行了。但还需要我手动python main.py——这就还是个工具，不是系统。

我设计了双轨调度：

轨道 A：传统 crontab（项目自带，作品集价值）

#!/bin/bash# run.shcd"$(dirname"$0")"source../.venv/bin/activateif["$1"="--batch"];thenpython run_batch.py"$2"elsepython main.pyfi

挂到 crontab：

30 9 * * * /path/to/ai_collector_project/run.sh

任何用户 clone 我的仓库，都能马上用。

轨道 B：Hermes Agent 调度（私人优化）

我用的工具是 Hermes Agent，自带 cron 调度器。但有个问题——main.py 输出大量调试日志，直接发到飞书会把消息流刷爆。

写了一个专用的 cron entrypoint，只输出 8 行精简摘要：

# ai_collector_cron.pydefmain():# 跑前后对比 DB 状态db=DBManager()before=db.get_run_summary()# 静默跑（吞掉详细日志）withredirect_stdout(silenced):asyncio.run(AIPipeline().run())after=db.get_run_summary()delta_completed=after['COMPLETED']-before['COMPLETED']# 寂静策略：什么都没发生 → 不打扰ifdelta_completed==0and...:print("[SILENT]")return# 输出精简摘要（≤ 8 行）print("📡 AI Collector 每日报告")print(f"✅ 本次处理：{delta_completed}条")print(f"📊 数据库：COMPLETED{after['COMPLETED']}/ PENDING{after['PENDING']}")print("📰 最近入库：")fortitleinget_latest_titles(3):print(f" •{title}")

最后挂到 Hermes cron：

cronjob.create(name="AI Collector daily run",schedule="30 9 * * *",no_agent=True,script="ai_collector_daily.sh",deliver="feishu")

第二天早上 9:30，飞书自动收到：

📡 AI Collector 每日报告 ✅ 本次处理：3 条 📊 数据库：COMPLETED 18 / PENDING 24 / FAILED 0 📰 最近入库： • 智谱发布 GLM-5.2 模型 • Anthropic 推出 Claude Fable 5 • Google 发布 DiffusionGemma 📁 详情见 logs/pipeline.log

项目从此完成"自治"——我什么都不用做，它每天自己跑，每天给我送早报。

💡 教训：自动化的最后一公里是"通知"。

写完调度代码不算完。摘要怎么呈现、什么时候保持沉默、错误怎么提醒——这些细节决定了用户是"享受系统"还是"被消息淹没"。

我特别喜欢"寂静策略"这个设计——没有新数据时，绝对不打扰你。

七、阶段 5：发布 v1.1

最后一步：把所有改动整理成一个正式的 release。

gitpush origin v1.1-stability ghprcreate--basemain--headv1.1-stability\--title"v1.1 Stability Release"ghprmerge1--squash--delete-branch gh release create v1.1--title"v1.1 Stability Release"

仓库地址：https://github.com/nakajimamiyuki/ai_collector_project

7 个 commit，+1021/-174 行，10 个文件改动：

八、复盘：今天到底学到了什么

v1.0 教会我"怎么做"，v1.1 教会我"怎么做对"

v1.0：让代码能跑 v1.1：让代码该跑成什么样 v1.0：用 print 看进度 v1.1：用 logger 收集 metrics v1.0：失败抛 exception v1.1：失败分类 → 自愈策略 v1.0：手动跑 v1.1：自动跑 + 智能摘要 v1.0：写完就 commit v1.1：先诊断 → 再修 → 再验证

5 个关键工程教训

1. 修代码前先做"诊断"

我不是上来就改代码，是先跑了一次 v1.0、看清楚 4 个真实问题、按优先级排序，再下刀。这步省了至少 2 小时。

2. 错误码是金矿，不是噪音

B 站的 -799 / -412 / -111 不是"反爬"，是精确的状态信号。把它们识别出来，就能写出"智能降级"而不是"暴力重试"的代码。

3. 失败要分类

LLM 阶段失败 ≠ collector 阶段失败 ≠ 网络失败。每种失败有对应的恢复策略。统一 try/except 抹平了所有有用的信息。

4. 自愈机制比无 bug 更值钱

任何系统都会出错。真正的"生产级"不是"不出错"，是"出错了能自己恢复"。

5. 自动化的最后一公里是通知

调度跑通了不等于完成。怎么把结果优雅地呈现给用户——寂静策略、精简摘要、错误提醒——决定了系统的"用户体验"。

数字总结

代码层面： +1021 / -174 行 7 commits 10 文件改动 0 lint 错误 数据层面： 数据库从 1 → 27 条真实 AI 行业新闻 字段命中率：title 100% / tags 100% / summary 100% FAILED 任务从 1 → 0（全部自动复活） 工程层面： 从"手动跑"→"每天 09:30 自动跑" 从"挂掉等修"→"失败自动重试 3 次" 从"看天用"→"双路径 fallback"

距离 AI Agent 工程师的进步

如果说 v1.0 让我证明"我能写出 AI Agent 项目"，v1.1 让我证明"我能把项目工程化"。

这两件事在面试官眼里完全不是一个量级：

• 写出 demo：很多人能做
• 把 demo 升级成系统：少数人能做
• 能讲清楚"为什么这么升级"：稀缺

我现在能完整讲清楚：

"v1.0 之后我做了一份复盘，发现 4 个真实问题。我没有一上来就改代码，而是先把问题分类——配置层、解析层、可观测层、可恢复层。

然后按价值/难度排序，从最容易的 LLM 修起，逐步推进到 B 站反爬、失败重试、定时调度。

中间踩到了 async event loop 的反模式 bug、kimi-k2.6 的长尾延迟问题、main.py 直接调 SQLite 破坏抽象的设计缺陷。每一个都有 git commit message 记录根因和修法。

最终发布了 v1.1 release，附带完整 changelog 和 roadmap。整个仓库公开在 GitHub。"

这一段话，是这个项目对我求职最大的"产出"。比代码本身值钱 10 倍。

尾声

这个项目对我来说，是从"能写代码"到"能交付系统"的分水岭。

如果你跟我一样：

• 在转型路上
• 想做但不知道做到什么程度才算"能拿出手"
• 时间有限（我每天只有 2-3 小时）

那我建议你也试试这种节奏：

第 1 天：做 v1.0，让它能跑（哪怕踩 8 个坑） 第 2 天：写复盘，列出问题 第 3 天：做 v1.1，把问题分类逐个修 第 4 天：写博客，把"为什么这么做"讲清楚

代码会被遗忘，但工程思维不会。

项目地址：https://github.com/nakajimamiyuki/ai_collector_project

下一篇预计写：怎么把 final_results 表里的数据"用起来"——daily brief、语义搜索、RAG 应用。

一个还在路上的人。