1. 从一条消息说起为什么我要给 WorkBuddy 设个“十点半闹钟”每天早上到工位第一件事是打开电脑、登录各种后台、翻一遍昨天的数据、看看有没有异常告警、再顺手刷一下行业动态。这套动作我做了快两年熟练到闭着眼都能点完但熟练不代表不烦——它每天雷打不动吃掉我将近四十分钟而且全是低价值的重复劳动。真正让我下决心动手的是某天早上我一边啃包子一边手动整理日报突然意识到这些信息本来就是结构化的抓取、汇总、排版、推送每一步都是确定性操作凭什么要人来做于是就有了这个项目——给 WorkBuddy 设一个“闹钟”每天上午十点半让它自动把一份 AI 日报生成好直接送进微信。这里说的 WorkBuddy是我自己搭的一套轻量自动化助手核心能力就三块定时触发、调用大模型做内容加工、把结果推到指定渠道。它不是什么现成的商业产品而是我用 Python 脚本 定时任务 大模型 API 拼出来的一个“私人助理”。之所以叫它 WorkBuddy是因为它干的活就是替我在工作场景里跑腿。而“闹钟”这个词指的是定时调度机制——到点自动醒、自动干活、干完自动汇报。这份日报里有什么大致是四块内容昨天我关注的几个信息源更新了什么、AI 圈当天值得留意的动态摘要、我自己项目里的一些关键指标快照、以及一段由大模型生成的“今日建议”。生成完之后通过微信的推送通道发到我手机上我睁眼或者刚到工位就能看到不用再手动翻。适合谁看这篇三类人。第一类是被日常重复信息整理折磨的职场人想用自动化把自己解放出来第二类是对 AI Agent、大模型 API 调用感兴趣想找个真实场景练手的开发者第三类是做自动化测试、运维出身手里有脚本能力但没想好往哪用的朋友。这篇不讲虚的从思路到代码到踩坑我尽量把能复现的都写清楚。2. 整体设计思路为什么是“定时 大模型 微信推送”这套组合2.1 拆解需求日报这件事到底难在哪很多人一听“自动生成日报”第一反应是“写个爬虫抓一抓不就行了”。真动手就会发现难点根本不在抓取而在三件事上。第一是信息源的异构性。我的信息源有 RSS、有网页、有 API 返回的 JSON、还有我自己数据库里的指标。格式五花八门如果每个源都写一套解析逻辑维护成本会爆炸。第二是内容的可读性。原始数据抓下来是一堆标题和数字直接推给我等于没推我需要的是“人话版”的摘要和解读。第三是触达的及时性。日报生成得再好如果我要主动去某个后台看那它就没意义必须主动推到我每天必看的地方——微信。这三件事对应到技术选型上就是统一的数据采集层、大模型做内容加工、微信做推送终端。而把它们串起来的是定时调度。2.2 为什么选大模型做“加工”而不是写死模板一开始我试过模板方案抓到的标题按固定格式拼一拼加个时间戳就推。跑了一周我就放弃了。原因是模板出来的东西没有“判断力”——十条动态里哪条重要、哪条是噪音模板分不出来全给你堆上去读起来跟没读一样。换成大模型之后情况完全不一样。我把抓到的原始素材丢给模型让它做三件事去重、排序、摘要。去重是把同一件事的不同报道合并排序是按“对我这个领域的相关性”排摘要是用一两句话把一件事说清楚。这三步做完日报的信息密度立刻上来了。这里我用的是 DeepSeek 的 API。选它的理由很实际中文理解好、价格便宜、接口稳定、文档清晰。对于日报这种每天跑一次、每次消耗几千 token 的场景成本几乎可以忽略。当然你用别的大模型也完全可以逻辑是通用的差别只在调用参数和返回格式上。2.3 为什么推送选微信而不是邮件或钉钉这是个很现实的选择。邮件我一天可能只看两次钉钉消息一多就被淹没只有微信是我真正高频打开的东西。日报要想起作用就得出现在我必然会看的地方。微信推送有几种路子企业微信机器人、公众号模板消息、服务号推送。我最后用的是企业微信的群机器人 Webhook原因很简单——配置成本最低一个 Webhook 地址就能推不需要认证、不需要审核、不需要用户授权而且消息能同步到我的个人微信里看。对于个人自用场景这是性价比最高的方案。提示企业微信群机器人 Webhook 有频率限制每分钟最多 20 条。日报一天一条完全够用但如果你要做高频推送得注意这个限制。2.4 整体架构一张图在脑子里整个系统的数据流是这样的定时器到点触发主流程 → 采集层并行拉取各信息源 → 清洗层做去重和格式化 → 大模型层做摘要和排序 → 渲染层拼成日报文本 → 推送层发到微信 → 记录日志。每一层都是独立的函数层与层之间用统一的数据结构我用的就是一个 dict 列表传递。这样设计的好处是任何一层出问题都不影响其他层而且换信息源、换模型、换推送渠道都只需要改对应那一层不用动整体逻辑。3. 核心细节解析每个环节的关键点和坑3.1 定时调度为什么我最终没用操作系统的 crontab最朴素的定时方案是 Linux 的 crontab一行配置搞定。我一开始也是这么干的但很快遇到两个问题。第一个问题是环境变量。crontab 执行时的环境跟你登录 shell 的环境不一样我脚本里依赖的 API Key 是通过环境变量传的结果 crontab 跑起来找不到直接报错。第二个问题是调试困难。crontab 跑失败了日志默认发到系统邮件我根本看不到排查一次要折腾半天。后来我换成了 Python 的APScheduler把定时逻辑写进脚本本身。好处是环境跟手动执行完全一致、日志我自己控制、失败重试逻辑可以自己写。配置大概是这样from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job( generate_and_push_daily_report, CronTrigger(hour10, minute30), iddaily_report, misfire_grace_time600, # 错过触发时间后 10 分钟内仍执行 coalesceTrue, # 多次错过只执行一次 max_instances1 # 防止任务重叠 ) scheduler.start()misfire_grace_time这个参数很关键。如果机器在十点半那一刻正好在重启或者负载高任务可能被错过。设了这个参数只要在十分钟内恢复任务还会补跑。coalesceTrue是防止补跑时重复执行多次。max_instances1是防止上一次还没跑完下一次又来了。注意如果你把脚本部署在会休眠的机器上比如某些云函数的冷启动场景定时可能不准。日报这种场景对时间精度要求不高差个几分钟无所谓但如果你要精确到秒得用常驻进程的方案。3.2 信息采集统一接口是维护成本的关键采集层我踩过最大的坑就是一开始给每个源写了独立的抓取函数返回格式各不相同。结果加一个新源就要改主流程改着改着代码就乱了。后来我定了一个规矩所有采集函数必须返回统一格式的列表每个元素是一个 dict包含title、content、source、timestamp四个字段。这样主流程完全不用关心数据从哪来只管往下传。def fetch_rss(url, source_name): RSS 源采集返回统一格式 import feedparser feed feedparser.parse(url) items [] for entry in feed.entries[:20]: items.append({ title: entry.get(title, ), content: entry.get(summary, ), source: source_name, timestamp: entry.get(published, ) }) return items def fetch_api(url, source_name, headersNone): API 源采集 import requests resp requests.get(url, headersheaders, timeout15) resp.raise_for_status() data resp.json() items [] for row in data.get(items, []): items.append({ title: row.get(name, ), content: row.get(desc, ), source: source_name, timestamp: row.get(time, ) }) return items采集层还要处理异常隔离。某个源挂了不能拖垮整个流程所以每个采集函数都包一层 try-except失败了就返回空列表并记日志主流程继续跑其他源。def safe_fetch(fetch_func, *args, **kwargs): try: return fetch_func(*args, **kwargs) except Exception as e: logger.error(f采集失败 {fetch_func.__name__}: {e}) return []这个safe_fetch包装器是我后来加的加完之后日报的稳定性肉眼可见地提升了。以前一个源超时整个日报就没了现在最多是少一块内容。3.3 大模型加工Prompt 设计决定了日报的质量上限这是整个项目里最需要打磨的部分。同样一批素材Prompt 写得好和写得差出来的日报质量天差地别。我最终的 Prompt 结构是这样的先给模型设定角色和任务再给输出格式要求最后贴素材。角色设定很重要我写的是“你是一个关注 AI 领域的技术分析师擅长从大量信息中筛选出真正重要的内容”。这个设定会显著影响模型的筛选倾向。输出格式我用的是结构化要求让模型返回 JSON方便我后续渲染PROMPT_TEMPLATE 你是一个关注 AI 领域的技术分析师。 请从下面的原始素材中筛选出最重要的 5-8 条按重要性排序并生成日报。 要求 1. 合并重复或高度相似的内容 2. 每条用一句话概括核心信息不超过 50 字 3. 对每条给出一个为什么值得关注的简短点评 4. 最后用一段话总结今天的整体趋势 严格按以下 JSON 格式返回不要有多余文字 { items: [ {title: ..., summary: ..., comment: ...} ], overview: ... } 原始素材 {raw_content} 这里有几个细节值得说。第一我明确要求“合并重复内容”因为不同源经常报道同一件事不合并的话日报会很啰嗦。第二我限制了每条摘要的字数防止模型写太长。第三我要求返回纯 JSON这样解析不会出错。调用的时候要注意超时和重试。大模型 API 偶尔会慢或者返回异常我设了 60 秒超时失败重试两次def call_llm(prompt, retries2): for i in range(retries 1): try: resp requests.post( LLM_API_URL, headers{Authorization: fBearer {API_KEY}}, json{ model: deepseek-chat, messages: [{role: user, content: prompt}], temperature: 0.3 }, timeout60 ) resp.raise_for_status() return resp.json()[choices][0][message][content] except Exception as e: if i retries: raise logger.warning(fLLM 调用失败重试 {i1}: {e}) time.sleep(3)temperature我设的 0.3偏低。日报这种场景要的是稳定和准确不需要模型发挥创造力温度低一点输出更可控。3.4 微信推送Webhook 的细节和消息格式企业微信机器人的推送非常简单一个 POST 请求就搞定。但消息格式有讲究用 Markdown 格式能让日报看起来清爽很多。def push_to_wechat(content): webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的KEY payload { msgtype: markdown, markdown: {content: content} } resp requests.post(webhook, jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: raise Exception(f推送失败: {result})Markdown 格式支持有限标题、加粗、链接、引用都能用但表格和复杂嵌套不支持。我日报的排版就是标题用##每条内容用加粗标题 普通描述 引用块点评整体读起来层次分明。注意企业微信 Markdown 消息有长度限制大概 4096 字节。日报内容多的时候要截断或者分条发送。我的做法是控制在 8 条以内一般不会超。4. 完整实操流程从零到跑通的每一步4.1 环境准备和依赖安装我用的 Python 3.10依赖不多一个 requirements.txt 搞定pip install requests apscheduler feedparserrequests负责所有网络请求apscheduler负责定时feedparser负责解析 RSS。如果你不用 RSS 源feedparser 可以省掉。API Key 和 Webhook 地址我放在环境变量里不写进代码export LLM_API_KEY你的大模型APIKey export WECHAT_WEBHOOK你的企业微信Webhook地址这样做的好处是代码可以随便分享不怕泄露密钥。读取的时候用os.environ.get()取不到就报错退出避免用空 Key 去调接口。4.2 主流程串联把各层拼起来主流程就是一个顺序执行的函数逻辑很直白def generate_and_push_daily_report(): logger.info(日报任务开始) # 1. 采集 all_items [] all_items safe_fetch(fetch_rss, https://example.com/feed1, 源A) all_items safe_fetch(fetch_rss, https://example.com/feed2, 源B) all_items safe_fetch(fetch_api, https://api.example.com/news, 源C) if not all_items: logger.error(所有源采集失败任务终止) return # 2. 预处理去重 seen set() unique_items [] for item in all_items: key item[title][:30] if key not in seen: seen.add(key) unique_items.append(item) # 3. 大模型加工 raw_text \n.join( f[{it[source]}] {it[title]}: {it[content][:200]} for it in unique_items[:50] ) prompt PROMPT_TEMPLATE.format(raw_contentraw_text) llm_output call_llm(prompt) # 4. 解析结果 result json.loads(llm_output) # 5. 渲染 markdown render_report(result) # 6. 推送 push_to_wechat(markdown) logger.info(日报任务完成)每一步都有日志出问题能快速定位。unique_items[:50]是限制送给模型的素材数量太多会超 token 限制太少又可能漏掉重要信息50 条是我实测下来比较合适的值。4.3 渲染函数把 JSON 变成好看的日报渲染这块没什么技术含量但细节决定观感def render_report(result): lines [## AI 日报, f {datetime.now().strftime(%Y-%m-%d %H:%M)}, ] for i, item in enumerate(result[items], 1): lines.append(f**{i}. {item[title]}**) lines.append(item[summary]) lines.append(f 点评{item[comment]}) lines.append() lines.append(---) lines.append(f**今日趋势**{result[overview]}) return \n.join(lines)标题、时间、编号、点评、趋势层次清楚。我特意把点评做成引用块视觉上跟正文区分开读起来不累。4.4 部署和守护让它稳定跑起来脚本写好了怎么让它一直跑我用的是最土但最稳的办法nohup后台运行 日志重定向。nohup python daily_report.py report.log 21 这样脚本就在后台常驻日志写到 report.log。想看运行情况就tail -f report.log。但这样有个问题机器重启后脚本就没了。所以我又加了一个 systemd 服务让它开机自启、崩溃自动重启[Unit] DescriptionDaily Report Bot Afternetwork.target [Service] Typesimple Useryouruser WorkingDirectory/home/youruser/report EnvironmentLLM_API_KEYxxx EnvironmentWECHAT_WEBHOOKxxx ExecStart/usr/bin/python3 /home/youruser/report/daily_report.py Restartalways RestartSec10 [Install] WantedBymulti-user.targetRestartalways是关键脚本万一崩了systemd 会在 10 秒后自动拉起来。环境变量也在这里配比 export 更可靠。5. 常见问题与排查技巧实录5.1 日报内容质量不稳定的排查症状有时候日报很精炼有时候一堆废话。排查思路先看送给模型的原始素材。如果素材本身重复度高、噪音多模型再强也救不回来。我后来在预处理阶段加了一个简单的关键词过滤把明显不相关的内容先剔掉效果立竿见影。另一个原因是 Prompt 里的素材顺序。模型对开头和结尾的内容更敏感我把最重要的源放在素材列表最前面重要内容被选中的概率明显提高。5.2 推送失败的几种情况和处理错误码含义处理方式93000Webhook 无效检查地址是否完整key 是否正确45009频率超限降低推送频率加延时40058内容格式错误检查 Markdown 语法避免不支持的元素超时网络问题加重试逻辑设合理超时我遇到最多的是内容格式错误原因是日报里偶尔会出现企业微信不支持的 Markdown 语法。后来我在渲染函数里加了一个清洗步骤把不支持的字符替换掉就再没出过这个问题。5.3 大模型返回格式不对怎么办模型偶尔会不按 JSON 格式返回比如多写了一段解释文字。我的处理是先用正则把 JSON 部分抠出来再解析import re def parse_llm_json(text): match re.search(r\{.*\}, text, re.DOTALL) if not match: raise ValueError(未找到 JSON 内容) return json.loads(match.group())这个兜底逻辑加上之后解析失败率从偶尔变成几乎为零。另外我在 Prompt 里也强调了“不要有多余文字”双管齐下。5.4 几个我踩过的坑坑一时区问题。服务器默认 UTC 时间我设的十点半结果变成了北京时间下午六点半。后来在 APScheduler 里显式指定了timezoneAsia/Shanghai才解决。坑二日志没轮转。脚本跑了一个月日志文件涨到几百兆。后来加了logging.handlers.RotatingFileHandler按大小切割最多保留 5 个文件。坑三API Key 硬编码。早期图省事把 Key 写死在代码里后来要分享代码给别人看差点泄露。现在一律走环境变量养成习惯就好。坑四素材太多超 token。有一次某个源突然返回了几百条直接超了模型上下文限制。现在我在采集层就限制了每个源最多取 20 条从源头控制。6. 还能怎么扩展几个我打算继续折腾的方向日报跑稳之后我开始想让它更“聪明”一点。第一个想法是加交互——日报推过去之后我回复某个编号它就把那条的详细内容发给我。这需要在推送层加一个接收消息的入口技术上不难就是多一个轮询或者回调。第二个想法是多端同步。现在只推微信我还想同步一份到我的笔记软件里方便以后检索。这个只需要在推送层加一个函数把同样的内容推到另一个渠道。第三个想法是个性化权重。现在模型筛选靠的是通用判断我想让它学习我的偏好——比如我经常点开某类内容就给它更高权重。这需要记录我的点击行为喂给模型做参考是个长期优化的活。第四个想法是异常告警。日报里如果出现某些关键词比如“故障”“下线”“涨价”自动标红并单独提醒。这个用简单的关键词匹配就能做成本很低但很实用。这些扩展我还没全部实现但思路是清晰的。核心原则不变每一层独立、接口统一、出问题不影响全局。这套架构搭好之后往上加功能就是搭积木不用推倒重来。最后分享一个我自己的体会自动化这件事最大的价值不是省时间而是把人的注意力从重复劳动里解放出来。日报这件事我做了两年每天四十分钟一年就是一百多个小时。现在这时间省下来了我可以用来做真正需要思考的事。而且自动化跑出来的东西比人手整理的更稳定、更不容易漏。如果你手里也有类似的重复劳动真的值得花一个周末把它自动化掉回报率远超你的想象。
基于大模型API与定时调度的微信自动日报生成实践
1. 从一条消息说起为什么我要给 WorkBuddy 设个“十点半闹钟”每天早上到工位第一件事是打开电脑、登录各种后台、翻一遍昨天的数据、看看有没有异常告警、再顺手刷一下行业动态。这套动作我做了快两年熟练到闭着眼都能点完但熟练不代表不烦——它每天雷打不动吃掉我将近四十分钟而且全是低价值的重复劳动。真正让我下决心动手的是某天早上我一边啃包子一边手动整理日报突然意识到这些信息本来就是结构化的抓取、汇总、排版、推送每一步都是确定性操作凭什么要人来做于是就有了这个项目——给 WorkBuddy 设一个“闹钟”每天上午十点半让它自动把一份 AI 日报生成好直接送进微信。这里说的 WorkBuddy是我自己搭的一套轻量自动化助手核心能力就三块定时触发、调用大模型做内容加工、把结果推到指定渠道。它不是什么现成的商业产品而是我用 Python 脚本 定时任务 大模型 API 拼出来的一个“私人助理”。之所以叫它 WorkBuddy是因为它干的活就是替我在工作场景里跑腿。而“闹钟”这个词指的是定时调度机制——到点自动醒、自动干活、干完自动汇报。这份日报里有什么大致是四块内容昨天我关注的几个信息源更新了什么、AI 圈当天值得留意的动态摘要、我自己项目里的一些关键指标快照、以及一段由大模型生成的“今日建议”。生成完之后通过微信的推送通道发到我手机上我睁眼或者刚到工位就能看到不用再手动翻。适合谁看这篇三类人。第一类是被日常重复信息整理折磨的职场人想用自动化把自己解放出来第二类是对 AI Agent、大模型 API 调用感兴趣想找个真实场景练手的开发者第三类是做自动化测试、运维出身手里有脚本能力但没想好往哪用的朋友。这篇不讲虚的从思路到代码到踩坑我尽量把能复现的都写清楚。2. 整体设计思路为什么是“定时 大模型 微信推送”这套组合2.1 拆解需求日报这件事到底难在哪很多人一听“自动生成日报”第一反应是“写个爬虫抓一抓不就行了”。真动手就会发现难点根本不在抓取而在三件事上。第一是信息源的异构性。我的信息源有 RSS、有网页、有 API 返回的 JSON、还有我自己数据库里的指标。格式五花八门如果每个源都写一套解析逻辑维护成本会爆炸。第二是内容的可读性。原始数据抓下来是一堆标题和数字直接推给我等于没推我需要的是“人话版”的摘要和解读。第三是触达的及时性。日报生成得再好如果我要主动去某个后台看那它就没意义必须主动推到我每天必看的地方——微信。这三件事对应到技术选型上就是统一的数据采集层、大模型做内容加工、微信做推送终端。而把它们串起来的是定时调度。2.2 为什么选大模型做“加工”而不是写死模板一开始我试过模板方案抓到的标题按固定格式拼一拼加个时间戳就推。跑了一周我就放弃了。原因是模板出来的东西没有“判断力”——十条动态里哪条重要、哪条是噪音模板分不出来全给你堆上去读起来跟没读一样。换成大模型之后情况完全不一样。我把抓到的原始素材丢给模型让它做三件事去重、排序、摘要。去重是把同一件事的不同报道合并排序是按“对我这个领域的相关性”排摘要是用一两句话把一件事说清楚。这三步做完日报的信息密度立刻上来了。这里我用的是 DeepSeek 的 API。选它的理由很实际中文理解好、价格便宜、接口稳定、文档清晰。对于日报这种每天跑一次、每次消耗几千 token 的场景成本几乎可以忽略。当然你用别的大模型也完全可以逻辑是通用的差别只在调用参数和返回格式上。2.3 为什么推送选微信而不是邮件或钉钉这是个很现实的选择。邮件我一天可能只看两次钉钉消息一多就被淹没只有微信是我真正高频打开的东西。日报要想起作用就得出现在我必然会看的地方。微信推送有几种路子企业微信机器人、公众号模板消息、服务号推送。我最后用的是企业微信的群机器人 Webhook原因很简单——配置成本最低一个 Webhook 地址就能推不需要认证、不需要审核、不需要用户授权而且消息能同步到我的个人微信里看。对于个人自用场景这是性价比最高的方案。提示企业微信群机器人 Webhook 有频率限制每分钟最多 20 条。日报一天一条完全够用但如果你要做高频推送得注意这个限制。2.4 整体架构一张图在脑子里整个系统的数据流是这样的定时器到点触发主流程 → 采集层并行拉取各信息源 → 清洗层做去重和格式化 → 大模型层做摘要和排序 → 渲染层拼成日报文本 → 推送层发到微信 → 记录日志。每一层都是独立的函数层与层之间用统一的数据结构我用的就是一个 dict 列表传递。这样设计的好处是任何一层出问题都不影响其他层而且换信息源、换模型、换推送渠道都只需要改对应那一层不用动整体逻辑。3. 核心细节解析每个环节的关键点和坑3.1 定时调度为什么我最终没用操作系统的 crontab最朴素的定时方案是 Linux 的 crontab一行配置搞定。我一开始也是这么干的但很快遇到两个问题。第一个问题是环境变量。crontab 执行时的环境跟你登录 shell 的环境不一样我脚本里依赖的 API Key 是通过环境变量传的结果 crontab 跑起来找不到直接报错。第二个问题是调试困难。crontab 跑失败了日志默认发到系统邮件我根本看不到排查一次要折腾半天。后来我换成了 Python 的APScheduler把定时逻辑写进脚本本身。好处是环境跟手动执行完全一致、日志我自己控制、失败重试逻辑可以自己写。配置大概是这样from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job( generate_and_push_daily_report, CronTrigger(hour10, minute30), iddaily_report, misfire_grace_time600, # 错过触发时间后 10 分钟内仍执行 coalesceTrue, # 多次错过只执行一次 max_instances1 # 防止任务重叠 ) scheduler.start()misfire_grace_time这个参数很关键。如果机器在十点半那一刻正好在重启或者负载高任务可能被错过。设了这个参数只要在十分钟内恢复任务还会补跑。coalesceTrue是防止补跑时重复执行多次。max_instances1是防止上一次还没跑完下一次又来了。注意如果你把脚本部署在会休眠的机器上比如某些云函数的冷启动场景定时可能不准。日报这种场景对时间精度要求不高差个几分钟无所谓但如果你要精确到秒得用常驻进程的方案。3.2 信息采集统一接口是维护成本的关键采集层我踩过最大的坑就是一开始给每个源写了独立的抓取函数返回格式各不相同。结果加一个新源就要改主流程改着改着代码就乱了。后来我定了一个规矩所有采集函数必须返回统一格式的列表每个元素是一个 dict包含title、content、source、timestamp四个字段。这样主流程完全不用关心数据从哪来只管往下传。def fetch_rss(url, source_name): RSS 源采集返回统一格式 import feedparser feed feedparser.parse(url) items [] for entry in feed.entries[:20]: items.append({ title: entry.get(title, ), content: entry.get(summary, ), source: source_name, timestamp: entry.get(published, ) }) return items def fetch_api(url, source_name, headersNone): API 源采集 import requests resp requests.get(url, headersheaders, timeout15) resp.raise_for_status() data resp.json() items [] for row in data.get(items, []): items.append({ title: row.get(name, ), content: row.get(desc, ), source: source_name, timestamp: row.get(time, ) }) return items采集层还要处理异常隔离。某个源挂了不能拖垮整个流程所以每个采集函数都包一层 try-except失败了就返回空列表并记日志主流程继续跑其他源。def safe_fetch(fetch_func, *args, **kwargs): try: return fetch_func(*args, **kwargs) except Exception as e: logger.error(f采集失败 {fetch_func.__name__}: {e}) return []这个safe_fetch包装器是我后来加的加完之后日报的稳定性肉眼可见地提升了。以前一个源超时整个日报就没了现在最多是少一块内容。3.3 大模型加工Prompt 设计决定了日报的质量上限这是整个项目里最需要打磨的部分。同样一批素材Prompt 写得好和写得差出来的日报质量天差地别。我最终的 Prompt 结构是这样的先给模型设定角色和任务再给输出格式要求最后贴素材。角色设定很重要我写的是“你是一个关注 AI 领域的技术分析师擅长从大量信息中筛选出真正重要的内容”。这个设定会显著影响模型的筛选倾向。输出格式我用的是结构化要求让模型返回 JSON方便我后续渲染PROMPT_TEMPLATE 你是一个关注 AI 领域的技术分析师。 请从下面的原始素材中筛选出最重要的 5-8 条按重要性排序并生成日报。 要求 1. 合并重复或高度相似的内容 2. 每条用一句话概括核心信息不超过 50 字 3. 对每条给出一个为什么值得关注的简短点评 4. 最后用一段话总结今天的整体趋势 严格按以下 JSON 格式返回不要有多余文字 { items: [ {title: ..., summary: ..., comment: ...} ], overview: ... } 原始素材 {raw_content} 这里有几个细节值得说。第一我明确要求“合并重复内容”因为不同源经常报道同一件事不合并的话日报会很啰嗦。第二我限制了每条摘要的字数防止模型写太长。第三我要求返回纯 JSON这样解析不会出错。调用的时候要注意超时和重试。大模型 API 偶尔会慢或者返回异常我设了 60 秒超时失败重试两次def call_llm(prompt, retries2): for i in range(retries 1): try: resp requests.post( LLM_API_URL, headers{Authorization: fBearer {API_KEY}}, json{ model: deepseek-chat, messages: [{role: user, content: prompt}], temperature: 0.3 }, timeout60 ) resp.raise_for_status() return resp.json()[choices][0][message][content] except Exception as e: if i retries: raise logger.warning(fLLM 调用失败重试 {i1}: {e}) time.sleep(3)temperature我设的 0.3偏低。日报这种场景要的是稳定和准确不需要模型发挥创造力温度低一点输出更可控。3.4 微信推送Webhook 的细节和消息格式企业微信机器人的推送非常简单一个 POST 请求就搞定。但消息格式有讲究用 Markdown 格式能让日报看起来清爽很多。def push_to_wechat(content): webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的KEY payload { msgtype: markdown, markdown: {content: content} } resp requests.post(webhook, jsonpayload, timeout10) result resp.json() if result.get(errcode) ! 0: raise Exception(f推送失败: {result})Markdown 格式支持有限标题、加粗、链接、引用都能用但表格和复杂嵌套不支持。我日报的排版就是标题用##每条内容用加粗标题 普通描述 引用块点评整体读起来层次分明。注意企业微信 Markdown 消息有长度限制大概 4096 字节。日报内容多的时候要截断或者分条发送。我的做法是控制在 8 条以内一般不会超。4. 完整实操流程从零到跑通的每一步4.1 环境准备和依赖安装我用的 Python 3.10依赖不多一个 requirements.txt 搞定pip install requests apscheduler feedparserrequests负责所有网络请求apscheduler负责定时feedparser负责解析 RSS。如果你不用 RSS 源feedparser 可以省掉。API Key 和 Webhook 地址我放在环境变量里不写进代码export LLM_API_KEY你的大模型APIKey export WECHAT_WEBHOOK你的企业微信Webhook地址这样做的好处是代码可以随便分享不怕泄露密钥。读取的时候用os.environ.get()取不到就报错退出避免用空 Key 去调接口。4.2 主流程串联把各层拼起来主流程就是一个顺序执行的函数逻辑很直白def generate_and_push_daily_report(): logger.info(日报任务开始) # 1. 采集 all_items [] all_items safe_fetch(fetch_rss, https://example.com/feed1, 源A) all_items safe_fetch(fetch_rss, https://example.com/feed2, 源B) all_items safe_fetch(fetch_api, https://api.example.com/news, 源C) if not all_items: logger.error(所有源采集失败任务终止) return # 2. 预处理去重 seen set() unique_items [] for item in all_items: key item[title][:30] if key not in seen: seen.add(key) unique_items.append(item) # 3. 大模型加工 raw_text \n.join( f[{it[source]}] {it[title]}: {it[content][:200]} for it in unique_items[:50] ) prompt PROMPT_TEMPLATE.format(raw_contentraw_text) llm_output call_llm(prompt) # 4. 解析结果 result json.loads(llm_output) # 5. 渲染 markdown render_report(result) # 6. 推送 push_to_wechat(markdown) logger.info(日报任务完成)每一步都有日志出问题能快速定位。unique_items[:50]是限制送给模型的素材数量太多会超 token 限制太少又可能漏掉重要信息50 条是我实测下来比较合适的值。4.3 渲染函数把 JSON 变成好看的日报渲染这块没什么技术含量但细节决定观感def render_report(result): lines [## AI 日报, f {datetime.now().strftime(%Y-%m-%d %H:%M)}, ] for i, item in enumerate(result[items], 1): lines.append(f**{i}. {item[title]}**) lines.append(item[summary]) lines.append(f 点评{item[comment]}) lines.append() lines.append(---) lines.append(f**今日趋势**{result[overview]}) return \n.join(lines)标题、时间、编号、点评、趋势层次清楚。我特意把点评做成引用块视觉上跟正文区分开读起来不累。4.4 部署和守护让它稳定跑起来脚本写好了怎么让它一直跑我用的是最土但最稳的办法nohup后台运行 日志重定向。nohup python daily_report.py report.log 21 这样脚本就在后台常驻日志写到 report.log。想看运行情况就tail -f report.log。但这样有个问题机器重启后脚本就没了。所以我又加了一个 systemd 服务让它开机自启、崩溃自动重启[Unit] DescriptionDaily Report Bot Afternetwork.target [Service] Typesimple Useryouruser WorkingDirectory/home/youruser/report EnvironmentLLM_API_KEYxxx EnvironmentWECHAT_WEBHOOKxxx ExecStart/usr/bin/python3 /home/youruser/report/daily_report.py Restartalways RestartSec10 [Install] WantedBymulti-user.targetRestartalways是关键脚本万一崩了systemd 会在 10 秒后自动拉起来。环境变量也在这里配比 export 更可靠。5. 常见问题与排查技巧实录5.1 日报内容质量不稳定的排查症状有时候日报很精炼有时候一堆废话。排查思路先看送给模型的原始素材。如果素材本身重复度高、噪音多模型再强也救不回来。我后来在预处理阶段加了一个简单的关键词过滤把明显不相关的内容先剔掉效果立竿见影。另一个原因是 Prompt 里的素材顺序。模型对开头和结尾的内容更敏感我把最重要的源放在素材列表最前面重要内容被选中的概率明显提高。5.2 推送失败的几种情况和处理错误码含义处理方式93000Webhook 无效检查地址是否完整key 是否正确45009频率超限降低推送频率加延时40058内容格式错误检查 Markdown 语法避免不支持的元素超时网络问题加重试逻辑设合理超时我遇到最多的是内容格式错误原因是日报里偶尔会出现企业微信不支持的 Markdown 语法。后来我在渲染函数里加了一个清洗步骤把不支持的字符替换掉就再没出过这个问题。5.3 大模型返回格式不对怎么办模型偶尔会不按 JSON 格式返回比如多写了一段解释文字。我的处理是先用正则把 JSON 部分抠出来再解析import re def parse_llm_json(text): match re.search(r\{.*\}, text, re.DOTALL) if not match: raise ValueError(未找到 JSON 内容) return json.loads(match.group())这个兜底逻辑加上之后解析失败率从偶尔变成几乎为零。另外我在 Prompt 里也强调了“不要有多余文字”双管齐下。5.4 几个我踩过的坑坑一时区问题。服务器默认 UTC 时间我设的十点半结果变成了北京时间下午六点半。后来在 APScheduler 里显式指定了timezoneAsia/Shanghai才解决。坑二日志没轮转。脚本跑了一个月日志文件涨到几百兆。后来加了logging.handlers.RotatingFileHandler按大小切割最多保留 5 个文件。坑三API Key 硬编码。早期图省事把 Key 写死在代码里后来要分享代码给别人看差点泄露。现在一律走环境变量养成习惯就好。坑四素材太多超 token。有一次某个源突然返回了几百条直接超了模型上下文限制。现在我在采集层就限制了每个源最多取 20 条从源头控制。6. 还能怎么扩展几个我打算继续折腾的方向日报跑稳之后我开始想让它更“聪明”一点。第一个想法是加交互——日报推过去之后我回复某个编号它就把那条的详细内容发给我。这需要在推送层加一个接收消息的入口技术上不难就是多一个轮询或者回调。第二个想法是多端同步。现在只推微信我还想同步一份到我的笔记软件里方便以后检索。这个只需要在推送层加一个函数把同样的内容推到另一个渠道。第三个想法是个性化权重。现在模型筛选靠的是通用判断我想让它学习我的偏好——比如我经常点开某类内容就给它更高权重。这需要记录我的点击行为喂给模型做参考是个长期优化的活。第四个想法是异常告警。日报里如果出现某些关键词比如“故障”“下线”“涨价”自动标红并单独提醒。这个用简单的关键词匹配就能做成本很低但很实用。这些扩展我还没全部实现但思路是清晰的。核心原则不变每一层独立、接口统一、出问题不影响全局。这套架构搭好之后往上加功能就是搭积木不用推倒重来。最后分享一个我自己的体会自动化这件事最大的价值不是省时间而是把人的注意力从重复劳动里解放出来。日报这件事我做了两年每天四十分钟一年就是一百多个小时。现在这时间省下来了我可以用来做真正需要思考的事。而且自动化跑出来的东西比人手整理的更稳定、更不容易漏。如果你手里也有类似的重复劳动真的值得花一个周末把它自动化掉回报率远超你的想象。
