每天花一两个小时整理资讯从 RSS、网站列表、文档源里抓几十篇内容再人工完成摘要、分类、排版这个动作长期做下去非常消耗精力。这次我们来看一个能把这套流程压缩成“一条可重跑工作流”的实战项目资讯日报生成器。它本质上是一个多模型参与的复杂工作流——采集、清洗、摘要、分类、排版、导出每一站由不同角色完成模型只做自己最擅长的那一部分。这一课作为 Agent 架构实战的第 5 课重点不是讲单点模型能力而是把多个模型任务串成一条可控、可观测、可恢复的工作流。这篇文章会直接从 Agent 架构角度拆解日报生成器为什么适合做复杂工作流样例、节点与状态怎么设计、模型怎么分工、怎么把流程封装成 HTTP 接口、怎么支持批量生成和定时任务以及最常见的排错思路。文中实现不锁定具体平台使用 OpenAI 兼容接口加 Python 完成迁移到 Dify、Coze、n8n 或自研 Agent 框架都是同一套思路。如果你正在学 Agent 架构、想做一个自动信息聚合产品或者每天都需要整理大量资料这篇文章建议收藏备用。1. 核心能力速览能力项说明项目类型Agent 架构实战 / 信息聚合工作流核心思路多模型节点 多工具协作按管道方式生成资讯日报输入RSS 链接、URL 列表、本地文本或 Markdown 文件输出Markdown 日报、JSON 结构化结果主要功能信息采集、正文提取、模型摘要、自动分类、智能排版、定时生成模型接入OpenAI 兼容接口可替换为本地部署模型或国内大模型 API是否支持批量任务支持可一次处理多源、多日期、多主题日报是否支持 API支持用 FastAPI 封装后可通过 HTTP 调用资源占用云端 LLM 场景主要消耗 Token 与网络延迟本地模型场景以对应模型显存需求为准启动方式Python 脚本直接运行或启动本地 HTTP 服务适合场景个人早报、团队资讯周报、内容聚合类产品学习价值Agent 工作流设计、节点拆分、状态流转、失败重试、接口化封装从表格可以看出这套工作流的运行成本不在 GPU 而在 Token。建模时要提前想清楚哪些节点用便宜模型批量跑哪些节点用强推理模型精修这是整个架构设计中最核心的资源决策。2. 适用场景与使用边界资讯日报生成器适合三类人第一类是内容运营需要每天早上快速浏览多个平台找出行业关键动态第二类是产品研发想在业务系统里内置“每日摘要”能力第三类是 Agent 学习者希望用一个小而完整的例子理解多模型协作的工程结构。它能解决的问题很直接把“找内容、读内容、总结内容、分类整理”从人工操作变成自动化流程。你只需要维护一个源列表定时触发工作流就会把几十篇原始资讯变成一份带摘要、带分类标签、带原文链接的 Markdown 日报。但也要说清不适合什么场景。第一它不适合低延迟强交互场景比如聊天机器人、实时问答因为工作流有固定步骤单次完整生成可能需要几十秒甚至更久第二它不适合做“一次生成即交付”的内容产品模型摘要可能出错、可能有幻觉重要资讯必须保留原文出处和人工复核入口第三它不适合抓取需要登录、有反爬限制或明确禁止采集的站点。合规边界必须反复强调。使用 RSS、公开 API、或站点明确允许转载的内容是安全的批量抓取网页文本时要尊重目标站点的 robots 协议和服务条款。模型生成的摘要不要直接当作原创内容商用引用新闻时保留原文链接。如果后续加入人脸识别、声音处理、私域数据解析等能力需要额外评估肖像权、隐私和内部数据保密要求。3. 为什么需要“模型参与的复杂工作流”很多人第一反应是把几十篇资讯一次性丢给大模型让它直接输出日报不是更简单吗从演示角度看确实简单但进入工程化后问题立刻暴露。第一个问题是上下文长度。十几篇资讯轻松超过几万字普通模型的上下文窗口不一定装得下强行塞入会导致中间内容被截断或注意力分散。即便上下文足够单次调用也容易超过网关超时时间重试成本非常高。第二个问题是成本不可控。所有资讯都让最贵的强模型处理Token 费用会成倍上涨。合理的做法是拆节点信息抓取用规则和库完成分类打标用轻量模型只有最后的摘要和排版才动用强推理模型。第三个问题是可观测性差。单次大 Prompt 生成的日报一旦输出质量不对你无法判断是采集阶段出了问题、摘要阶段出了问题还是排版阶段出了问题。拆成工作流节点后每一步都可以单独测试、单独替换模型、单独重试。这就引出了 Agent 架构里“执行编排层harness”的概念。工作流引擎负责维护状态、传递数据、重试失败节点而每个模型节点只负责“理解输入并产出结果”。资讯日报生成器就是一个很好的编排样例它不需要复杂的工具调用也不需要多轮自我反思但要求节点之间状态清晰、数据格式稳定、失败可恢复。这些能力恰好是 Agent 工程的通用基础。4. 系统架构与工作流节点设计整个工作流按数据流方向拆成六个节点。每个节点都是独立函数通过一个结构化数据对象传递结果。节点输入模型/工具动作输出失败策略节点 1 源加载sources.json 源列表读取 RSS / URL / 本地文件原始文本列表单源失败跳过记录日志节点 2 内容清洗原始文本正则去噪、HTML 标签清理、去重干净正文列表空内容直接丢弃节点 3 模型摘要干净正文LLM 按摘要模板生成 1-3 句摘要摘要对象列表单篇失败重试 2 次节点 4 标签分类摘要 正文开头LLM 按分类枚举打标带标签的条目分类失败默认“未分类”节点 5 聚合排版带标签条目LLM 或模板引擎排序、整合Markdown 日报排版异常时降级为模板拼接节点 6 输出校验Markdown 日报规则校验标题、链接、编码最终日报文件失败则保留原始数据备查这个设计的核心是数据协议。每个节点的输出必须是一个明确的 dict 或 Pydantic 对象不能是自由文本。这样后续节点可以稳定读取字段模型返回异常时也能快速定位到具体节点。建议协议字段统一为{ id: unique_id, title: 标题, source: 来源名称, url: 原文链接, content: 清洗后的正文, summary: 模型生成的摘要, category: 分类标签 }工作流引擎只需要循环执行节点列表并把上一个节点的输出作为下一个节点的输入。这一步看着简单却是 Agent 架构里最值得花时间设计的部分数据协议稳定后面接什么模型都可以。5. 环境准备与前置条件本工作流的核心实现是 Python 3.10需要安装以下依赖pip install feedparser requests openai pydantic fastapi uvicorn apscheduler python-dotenv各依赖用途说明依赖用途feedparser解析 RSS/Atom 订阅源requests抓取网页和调用通用 HTTP 接口openai调用 OpenAI 兼容的 Chat Completions 接口pydantic定义结构化输出协议fastapi uvicorn封装 HTTP API 服务apscheduler定时触发日报生成python-dotenv读取 .env 配置模型侧建议准备一个 OpenAI 兼容接口可以是云端服务也可以是本地部署的 vLLM、Ollama 等网关。将接口地址和密钥写入.env文件LLM_API_KEYyour_api_key_here LLM_BASE_URLhttps://your-llm-endpoint.example.com/v1 LLM_MODELgpt-4o-mini LLM_MODEL_CHEAPgpt-4o-mini REPORT_OUTPUT_DIR./outputs需要说明的是这里没有写死模型名称。实际使用时把LLM_BASE_URL换成你本地或云端的地址即可只要接口兼容 Chat Completions。源列表用 JSON 维护格式如下{ sources: [ { name: 示例科技媒体, type: rss, url: https://example.com/rss }, { name: 示例站点专栏, type: url, url: https://example.com/column } ] }6. 工作流核心代码实现简化示例6.1 RSS 抓取与清洗这部分用 feedparser 负责解析避免重复造轮子import feedparser import re def fetch_rss(url: str, limit: int 10): feed feedparser.parse(url) items [] for entry in feed.entries[:limit]: content entry.get(summary) or entry.get(description) or text re.sub(r[^], , content) text re.sub(r\s, , text).strip() items.append({ title: entry.get(title, ).strip(), url: entry.get(link, ).strip(), source: feed.feed.get(title, url), content: text }) return items这段代码会自动跳过没有内容的条目并统一去掉 HTML 标签。如果源是普通网页而非 RSS可以用 requests BeautifulSoup 做正文提取也可以直接让模型从网页文本中提取正文——后者更灵活但 Token 成本更高。6.2 模型摘要节点摘要节点是工作流里最标准的 LLM 调用。为了让模型输出可控采用 JSON Mode 并要求返回固定字段import os from openai import OpenAI from pydantic import BaseModel client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), ) class SummaryResult(BaseModel): title: str summary: str keywords: list[str] def summarize_item(item: dict) - SummaryResult: prompt f 你是一名资讯编辑。请根据下面的文章信息生成一句不超过50字的摘要并提取3个关键词。 文章标题{item[title]} 文章正文{item[content][:1500]} 要求 1. 摘要要保留时间、主体、核心事件。 2. 输出 JSON字段为 title, summary, keywords。 resp client.chat.completions.create( modelos.getenv(LLM_MODEL), response_format{type: json_object}, messages[ {role: system, content: 你是严谨的信息摘要引擎只输出 JSON。}, {role: user, content: prompt}, ], temperature0.3, timeout60, ) text resp.choices[0].message.content cleaned text.strip().strip(json).strip() return SummaryResult.model_validate_json(cleaned)这里用 Pydantic 做了解析模型一旦返回非法 JSON 就会抛异常方便上层做重试。如果你用的模型不支持response_format就移除此参数改用正则或 JSON 解析器兜底。6.3 分类与排版分类节点继续复用 LLM但这里可以用便宜模型CATEGORIES [AI, 云计算, 开源项目, 行业动态, 产品发布, 其他] def classify_item(summarized: dict) - str: prompt f 文章标题{summarized[title]} 摘要{summarized[summary]} 请从以下分类中选择一个最匹配的{, .join(CATEGORIES)} 只输出分类名称不要解释。 resp client.chat.completions.create( modelos.getenv(LLM_MODEL_CHEAP), messages[{role: user, content: prompt}], temperature0, max_tokens10, ) result resp.choices[0].message.content.strip() return result if result in CATEGORIES else 其他排版节点不需要调用模型。直接用 Python 按分类分组再用 Markdown 语法拼接模板def render_report(items: list[dict], report_date: str) - str: lines [f# 每日资讯日报 {report_date}, ] for cat in CATEGORIES: cat_items [i for i in items if i[category] cat] if not cat_items: continue lines.append(f## {cat}) lines.append() for item in cat_items: lines.append(f- [{item[title]}]({item[url]})) lines.append(f - {item[summary]}) lines.append() return \n.join(lines)模板拼接的好处是稳定、不需要额外 Token。如果你希望日报语言更口语化、更像人工编辑撰写可以再接入一个“润色排版”节点但初期建议用模板降低成本且方便排查。6.4 主流程串联把上述节点串成完整工作流def run_daily_report(source_path: str, output_dir: str): sources load_sources(source_path) all_items [] for src in sources: try: items fetch_rss(src[url]) all_items.extend(items) except Exception as e: print(f[fetch failed] {src[name]}: {e}) continue processed [] for item in all_items: try: summary summarize_item(item) category classify_item({title: item[title], summary: summary.summary}) processed.append({ **item, summary: summary.summary, category: category }) except Exception as e: print(f[process failed] {item[title]}: {e}) continue article render_report(processed, date.today().isoformat()) os.makedirs(output_dir, exist_okTrue) out_path os.path.join(output_dir, fdaily_{date.today().isoformat()}.md) with open(out_path, w, encodingutf-8) as f: f.write(article) return out_path这一段是工作流主循环抓取尽量不中断清洗失败跳过摘要失败重试逻辑可根据项目需要补充。这样设计的好处是单源异常不会导致整个日报失败。7. 服务化接入与批量任务队列脚本能跑通之后下一步是把它封装成 HTTP 接口让其他系统可以调用。这里用 FastAPI 写一个最小服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleDaily Report Generator) class ReportRequest(BaseModel): source_path: str ./sources.json output_dir: str ./outputs report_date: str app.post(/api/daily-report) def create_report(req: ReportRequest): try: out_path run_daily_report(req.source_path, req.output_dir) return {status: ok, output_path: out_path} except Exception as e: raise HTTPException(status_code500, detailstr(e))启动服务uvicorn app:app --host 0.0.0.0 --port 8000接口调用示例先看 curlcurl -X POST http://127.0.0.1:8000/api/daily-report \ -H Content-Type: application/json \ -d {source_path: ./sources.json, output_dir: ./outputs}Python 调用方式import requests url http://127.0.0.1:8000/api/daily-report payload { source_path: ./sources.json, output_dir: ./outputs } resp requests.post(url, jsonpayload, timeout300) print(resp.status_code) print(resp.json())批量任务场景下建议在外部维护一个批量目录每次调用传不同参数inputs/ sources_tech.json sources_finance.json outputs/ 2025-06-19_tech.md 2025-06-19_finance.md批量脚本可以循环调用/api/daily-report也可以直接在进程内循环调用run_daily_report。对日报类任务来说并发没必要开太高串行加超时控制反而更稳。如果同一时间触发过多任务最简单的方式是加一个队列用任务状态文件记录进度失败任务允许重试。定时任务用 APScheduler 实现每天 8 点执行一次from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.add_job( lambda: run_daily_report(./sources.json, ./outputs), triggercron, hour8, minute0, ) scheduler.start()8. 分节点功能测试与效果验证工作流部署后不要直接跑全量按节点逐层验证。8.1 单源采集测试先用一个 RSS 源测试fetch_rss。判断成功的标准是能打印出标题、URL、清洗后的正文且正文没有 HTML 标签和多余空白。如果返回空列表优先检查网络能否访问该 RSS 地址以及 feedparser 是否解析成功。8.2 摘要一致性验证对同一篇文章连续调用三次摘要接口比较输出内容。正常现象是表达用词略有差异但核心事实一致。如果三次输出差异过大说明 temperature 设置偏高建议降到 0.2 到 0.3如果摘要丢失时间、主体、事件需要调整 Prompt 结构把要求拆得更细。8.3 分类准确率抽查准备一批已知分类的资讯运行classify_item统计准确率。初版达到 80% 就算合格因为后续可以用规则修正。常见问题是“其他”分类占比过高说明分类枚举不够覆盖业务场景可以增加分类项或让模型优先输出最接近的类别。8.4 超长文本处理单篇文章正文超过 1500 字时摘要 Prompt 做了截断处理。这里要验证截断后是否丢失关键信息。稳妥做法是提取正文前 1500 字同时把标题放在最前面让模型优先理解主题。如果站点正文结构复杂建议先做正文抽取再截断不要直接把整个 HTML 丢给模型。8.5 批量与失败重试模拟一个失效 RSS 源观察全流程是否继续执行。工作流应该跳过失败源输出剩余正常源的日报并在日志中记录失败原因。摘要环节连续失败两次时可以加入重试机制def summarize_with_retry(item, retries2): for i in range(retries): try: return summarize_item(item) except Exception as e: print(fretry {i1}: {e}) time.sleep(1) return None8.6 整体效果验证最终日报文件应当包含标题、生成日期、分类小标题、资讯链接、摘要。发布前人工抽查 5 到 10 条确认链接可访问、摘要准确、分类合理。这一步是日报类产品不可省略的质检环节。9. 资源占用与性能观察资讯日报生成器的资源占用和传统推理服务不一样重点看四个指标Token 消耗、单次调用延迟、总任务耗时、失败率。Token 消耗可以直接从模型服务后台统计也可以在工作流里打印每次请求的 usage 字段。估算公式是每条摘要大约消耗输入 500 到 2000 Token输出 100 到 200 Token。十条资讯大概消耗 1 万到 3 万 Token按接口单价能算出单份日报的成本。要降本可以做内容去重、只摘要新增内容、分类用便宜模型、排版用模板。延迟方面云端 LLM 单次摘要调用约 3 到 10 秒分类调用约 1 到 3 秒。十条资讯耗时可能在 1 到 3 分钟之间。如果引入超时和并发单份日报可以压到 30 秒以内但要注意接口限流。如果整个工作流改用本地模型资源占用就转移到显存和 CPU。一个 7B 到 14B 的模型大约需要 6G 到 16G 显存具体以模型版本和量化精度为准本地部署时建议用 vLLM 或 Ollama 启动一个兼容接口然后工作流代码完全不用改。实际占用要以本机测试为准不要用网上某一篇博客的数字直接套用。10. 常见问题与排查方法问题现象可能原因排查方式解决方案RSS 抓取结果为空网络不通、RSS 地址失效、feedparser 解析失败用浏览器打开 RSS 地址确认返回 XML更换源地址或改用 URL 抓取模型返回内容不是 JSON模型能力不足或 Prompt 指令不够明确打印原始返回文本启用 JSON Mode增加示例降低 temperature摘要结果截断输出 max_tokens 设置过小查看返回内容结尾是否完整提高 max_tokens或压缩输入长度分类结果总是“其他”分类枚举不匹配、Prompt 描述不清抽查 10 条分类结果扩大分类定义增加少量示例API 调用超时网络慢、模型负载高、单次请求 Prompt 过长查看服务端日志和调用耗时缩短输入文本、缩短 timeout、增加重试服务启动后端口冲突8000 端口被占用检查端口占用情况更换端口启动 uvicorn批量任务卡住单篇摘要无响应或线程阻塞查看任务日志定位卡住的源增加单任务超时失败自动跳过日报 Markdown 乱码编码未指定 UTF-8检查输出文件编码写入文件时强制 encodingutf-8同一篇文章反复出现源列表里有重复链接检查源配置和抓取日志按 URL 去重Agent 执行超时工作流单步等待模型响应过久开启 trace 日志观察每一步耗时为每个模型调用设置独立超时这里特别提醒一点如果用到 Agent 框架常见报错类似“execution provider did not respond in time”本质都是单步调用超时。排除思路是先把模型接口单独 curl 一遍确认模型本身响应正常再检查工作流超时配置。11. 最佳实践与使用建议第一次搭建时不要追求全功能先跑通最小闭环。用两到三个 RSS 源配一个便宜模型输出一份不分类的 Markdown 日报。确认这条链路稳定后再加分类节点、加 HTML 抓取、加定时任务、加 API 服务。每一步都做单独验证避免一次引入太多变量导致问题难定位。模型选择上建议分层规划。抓取和清洗用规则代码完成不消耗 Token分类打标用便宜小模型追求速度摘要生成用中等模型保证信息密度和准确性如果后续要写日报开头的“今日观察”再调用强推理模型生成 200 到 500 字的趋势分析。这样分配成本最合理。工作流数据结构要稳定。所有节点只通过 Pydantic 模型传参禁止节点之间互相透传 JSON 字符串。后期要加节点或换模型只需要保证数据协议不变代码改动会小很多。日志和缓存是日报系统稳定运行的基础。每次抓取后把原始文本缓存到本地这样模型摘要失败时不需要重新抓取每次模型调用记录 Token、耗时、模型名方便月底核算成本。批量任务建议给每个任务分配一个任务 ID日志文件名带上任务 ID排查问题会轻松很多。涉及版权和隐私时必须谨慎。采集公开 RSS 和允许访问的页面没有问题但不能为了抓取绕过站点限制。日报里展示摘要时应附上原文链接避免把模型摘要直接作为原创内容发布。如果企业内部使用要确认数据是否涉密模型接口是否允许传输这些数据。内容生成后一定要保留人工复核环节。模型摘要偶尔会出现张冠李戴、数字写错、来源混淆等问题。可以引入“置信度”概念模型对摘要结果不确定时输出时打标“需人工复核”运营人员看到这类标记再做二次确认。12. 总结与下一步资讯日报生成器最适合作为 Agent 复杂工作流的入门实战因为它把多模型协作、节点状态、失败重试、服务化封装都包含在内但又不引入过于复杂的推理规划。最值得先验证的功能是“RSS 抓取 模型摘要 模板排版”这个最小闭环先确认模型输出稳定再逐步扩展分类和接口。最容易踩的坑有三个一是把所有资讯塞进一个大 Prompt导致上下文超长和超时二是模型返回的 JSON 没做解析保护运行中频繁中断三是没有设置单源失败跳过一个源挂了整个日报都跑不出来。这三个坑在设计阶段就可以避开。下一步扩展方向很多接入飞书或钉钉机器人每天早上自动推送日报支持更多来源类型比如微信公众号文章、PDF 文档、数据库记录按用户兴趣做个性化日报先用向量化手段召回相关内容再用模型重排生成摘要还可以把多份历史日报存进知识库用 RAG 方式做周报或月度趋势分析整体架构无需推倒重来。