从资讯日报生成器看Agent工作流编排：模型与规则的混合实战

资讯日报生成器是“模型参与的复杂工作流”里一个非常典型的小型产品。它要抓取多个资讯源清洗去重用大模型生成摘要和分类最后按固定结构输出一份 Markdown 日报。这个任务看起来简单但一旦把范围从“跑通一次”扩大到“每天稳定跑一次”难点就不再是调用模型接口而是如何拆分流程、控制上下文、约束输出、处理失败和防止幻觉。写这类工作流时模型只是整条流水线里的一个组件。真正决定日报质量的是哪些步骤让模型做哪些步骤用规则做模型返回内容如何校验某一节点失败时整条链路的降级策略是什么。这篇文章用资讯日报生成器作为实战案例完整走一遍采集、清洗、摘要、分类、渲染、编排的实现过程。阅读前需要了解 Python 基础、REST API 调用和环境变量配置读完后能够自己搭一个最小可用的资讯日报工作流并在此基础上扩展成更复杂的 Agent 应用。1. 资讯日报生成器为什么能训练手中的 Agent 架构思维1.1 从“模型调用”到“工作流编排”差的不是模型能力而是任务拆解很多初学者会把 Agent 开发理解成“写好 prompt 然后反复问模型”。这个思路在单轮问答里没有大问题但放到资讯日报场景里会迅速失控。一份日报如果让模型一口气完成“抓取、去重、摘要、分类、排序、渲染”它会面临三个现实问题。第一模型拿不到实时网络数据就算拿到也容易把无关链接拼进正文。第二抓取到的几十条资讯可能远超单次上下文窗口强塞进去只会导致截断和摘要质量下降。第三输出既要符合 JSON 结构又要保证标题和来源正确一次生成的失败率会随内容长度显著上升。正确做法是先把任务拆成阶段。每个阶段有明确输入、输出、成功条件和失败策略。模型只处理它擅长的事情理解语义、压缩信息、给内容排序。网络请求、文本清洗、格式渲染这些确定性操作交给代码完成。所谓 Agent 架构本质上就是设计这条流水线的数据流、状态和护栏。1.2 资讯日报是理想的教学案例脏活、脑力活和展示活天然分开资讯日报里包含三类明显不同的工作。采集和去重是脏活摘要和分类是脑力活Markdown 渲染是展示活。三种工作的最优处理方式完全不同。采集需要处理网络超时、反爬限制、RSS 字段缺失、HTML 标签混杂等问题这些问题靠模型处理效率很低而且每次结果不稳定。摘要和分类恰好相反它需要理解时间、主体、关键数据和事件含义规则脚本很难覆盖。渲染则是典型的模板工作要求输出格式统一、日期一致、分类顺序稳定模板引擎比模型更合适。当一个任务能把“规则”和“模型”的边界划得这么清楚时它就非常适合作为工作流入门案例。你可以先理解模型应该放在哪个位置再理解每个模型节点前后的数据校验最后再考虑如何用状态机或图形框架把这些节点串起来。1.3 模型在复杂工作流里的三种参与方式在实际编排中模型不会只做“生成正文”这一件事。资讯日报工作流里至少会出现三种参与方式。第一种是生成式参与比如把一篇长资讯压缩成 80 字摘要或者给新闻写一个更适合日报展示的标题。这种场景对格式要求低但对事实准确性要求高需要约束模型不要加入原文没有的信息。第二种是判定式参与比如给资讯打分类标签、评估重要程度、判断两篇文章是否讲述同一事件。模型输出的是一个结构化的判定结果适合用 JSON Schema 校验。第三种是改写式参与比如把口语化的 RSS 摘要改写成技术博客风格的日报语句。它介于生成和改写之间风险在于改写后可能失真因此需要保留原文链接作为溯源字段。设计工作流时先想清楚每个模型节点属于哪种参与方式。生成式节点重提示词约束判定式节点重输出结构改写式节点重溯源校验。混为一谈会让后续排错变得非常困难。注意不要因为项目里接入了大模型就把所有步骤都交给模型处理。一个稳定工作流的标志是模型失败时系统还能用规则降级而不是直接崩溃。2. 工作流的执行方案和运行环境要提前定好2.1 技术选型用 OpenAI 兼容接口降低绑定成本资讯日报生成器涉及网络请求、RSS 解析、JSON 校验、模板渲染和模型调用。为了降低学习成本这里选择 Python 作为主语言模型调用通过 OpenAI 兼容接口完成。选用 OpenAI 兼容协议有几个原因。第一不管是商用模型服务还是本地部署的 Ollama、vLLM 等推理服务大多数都提供/v1/chat/completions兼容端点代码可以一次编写多处切换。第二Python SDK 提供了结构清晰的消息、参数和响应对象便于在编排器里统一处理。第三如果你后续要切换到其他模型服务只需要改base_url、api_key和model三个配置项不需要改业务逻辑。这里不会绑定具体模型名称。示例代码里的模型名只是占位落地前要确认你的模型服务真实支持的模型标识以及它是否支持response_format或 function calling。2.2 Python 环境准备与依赖安装建议使用 Python 3.10 或更高版本。低版本对类型标注和 dataclass 的支持虽然也够用但后续扩展 LangGraph 之类依赖时会比较麻烦。创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate pip install openai1.30,2 feedparser6,7 requests2.31,3 pip install jinja23,4 pydantic2,3 python-dotenv1,2这些依赖的用途如下依赖作用学习环境openai调用 OpenAI 兼容接口任意版本均可建议 1.xfeedparser解析 RSS/Atom 资讯源替代手写 XML 解析requests下载 RSS 内容设置超时和 User-Agentjinja2渲染 Markdown 日报模板也可以直接用 f-stringpydantic校验模型输出的 JSON 结构保证节点间数据契约python-dotenv加载.env配置避免密钥写死到代码如果你的网络环境无法访问公共模型服务可以把模型端点指向本地推理服务。保持 OpenAI 兼容协议代码不需要大改。需要注意的是本地模型对结构化输出的支持参差不齐后面会讲到如何做兜底解析。2.3 配置文件设计来源、模型、调度参数全部外置日报工作流最忌讳把 RSS 来源、模型名、重复运行天数、最大条数都写死在代码里。这样每次调整都要改代码而且容易在多人协作时产生冲突。推荐在项目根目录创建.env文件LLM_BASE_URLhttps://your-llm-service.example.com/v1 LLM_API_KEYsk-your-key LLM_MODELyour-chat-model RSS_SOURCEShttps://example.com/feed.xml,https://example.org/rss MAX_ITEMS_PER_SOURCE20 REPORT_TOP_N10 REPORT_DATE2026-01-01然后在config.py里统一读取import os from dataclasses import dataclass, field from dotenv import load_dotenv load_dotenv() dataclass class Config: base_url: str api_key: str model: str sources: list[str] field(default_factorylist) max_items_per_source: int 20 report_top_n: int 10 report_date: str def load_config() - Config: sources [ s.strip() for s in os.getenv(RSS_SOURCES, ).split(,) if s.strip() ] return Config( base_urlos.getenv(LLM_BASE_URL, ), api_keyos.getenv(LLM_API_KEY, ), modelos.getenv(LLM_MODEL, ), sourcessources, max_items_per_sourceint(os.getenv(MAX_ITEMS_PER_SOURCE, 20)), report_top_nint(os.getenv(REPORT_TOP_N, 10)), report_dateos.getenv(REPORT_DATE, ), )这样做的目的有两点。一是把容易变的部分集中到配置里二是为后续接入配置中心或环境变量注入留下位置。生产环境不要直接把.env提交到代码仓库密钥应该由部署平台注入。2.4 输入输出结构与状态对象设计多阶段工作流最容易出现的问题是每个节点都返回自己的数据结构节点之间没有统一约定。日报只有五六个阶段还能靠记忆支撑一旦阶段变多维护成本会迅速上升。这里定义一个贯穿全流程的状态对象每一阶段把结果写到对应字段from dataclasses import dataclass, field dataclass class RawItem: title: str url: str raw_summary: str published: str source: str dataclass class CleanedItem: title: str url: str clean_summary: str published: str source: str digest: str dataclass class SummarizedItem: title: str url: str clean_summary: str daily_summary: str source: str dataclass class CategorizedItem: title: str url: str daily_summary: str category: str score: int reason: str source: str dataclass class WorkflowContext: raw_items: list[RawItem] field(default_factorylist) cleaned_items: list[CleanedItem] field(default_factorylist) summarized_items: list[SummarizedItem] field(default_factorylist) categorized_items: list[CategorizedItem] field(default_factorylist) report: str 有了这个状态对象每个节点只需要关注“从哪个字段读往哪个字段写”。中间某一步失败时也可以把已经完成的数据序列化到本地文件下次从断点继续。3. 分阶段实现资讯日报生成器3.1 采集阶段先用规则把原始数据抓下来模型先不参与采集阶段的目标是把多个 RSS 源的原始条目合并起来。这个阶段尽量不让模型参与因为网络请求和 XML 解析是确定性工作模型参与只会增加延迟和失败率。用一个通用函数抓取单个 RSS 源import feedparser import requests def fetch_rss(url: str, timeout: int 10) - list[RawItem]: headers { User-Agent: daily-news-agent/0.1, } resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() feed feedparser.parse(resp.content) items [] for entry in feed.entries: title entry.get(title, ).strip() link entry.get(link, ).strip() summary entry.get(summary, ).strip() published entry.get(published, ) if title and link: items.append(RawItem( titletitle, urllink, raw_summarysummary, publishedpublished, sourceurl, )) return items这里要注意两点。第一必须设置超时时间避免某个源卡住整个工作流第二RSS 的summary往往是 HTML 片段不能直接交给模型需要在下一阶段清洗。多源抓取可以简单循环也可以使用线程池并发。日报场景对实时性要求不高循环加异常捕获就够用。如果某一条源失败把它记入日志并继续抓其他源比整体失败更合理。3.2 清洗与去重用哈希和规则减少模型处理量清洗阶段做三件事去掉 HTML 标签、压缩空白、过滤内容过短的条目。import hashlib import re def clean_html(text: str) - str: text re.sub(r[^], , text or ) text re.sub(r\s, , text) return text.strip() def clean_items(raw_items: list[RawItem]) - list[CleanedItem]: cleaned [] for item in raw_items: clean_summary clean_html(item.raw_summary) if len(clean_summary) 20: continue digest_text f{item.title}|{item.url} digest hashlib.sha256(digest_text.encode(utf-8)).hexdigest() cleaned.append(CleanedItem( titleitem.title, urlitem.url, clean_summaryclean_summary, publisheditem.published, sourceitem.source, digestdigest, )) return cleaned def dedup_items(cleaned_items: list[CleanedItem]) - list[CleanedItem]: seen set() result [] for item in cleaned_items: if item.digest in seen: continue seen.add(item.digest) result.append(item) return result为什么用标题 链接做哈希而不是只用标题因为不同网站的相同事件标题可能相近但链接通常不同只用链接又可能漏掉同一篇文章在不同站点的不同链接。日报场景里先用这个简单哈希足够生产环境可以再引入向量相似度去重。实际项目中还要注意一个坑RSS 源可能每天更新但同一条新闻会挂在源里好几天。如果不去重日报会出现大量“昨天看过”的内容。可以把历史标题存进 SQLite 或 Redis按 URL 维度做跨天去重。3.3 标题改写与摘要生成让模型只做一句话的精准压缩清洗完成后就可以进入模型节点。摘要生成是日报工作流里最重要的一步。这里的目标不是让模型重新创作而是把原文压缩成 80 字以内的日报摘要保留时间、主体、关键数据。先定义摘要调用函数from openai import OpenAI SUMMARY_SYSTEM_PROMPT 你是一个资讯日报编辑。你负责把用户提供的资讯压缩成适合日报阅读的摘要。 要求 1. 摘要不超过 80 字。 2. 只根据原文内容压缩不添加原文没有的信息。 3. 保留时间、主体、关键数据。 4. 返回 JSON格式为 {summary: 摘要内容}。 def summarize_one(item: CleanedItem, client: OpenAI, model: str) - SummarizedItem: user_content ( f标题{item.title}\n f正文{item.clean_summary[:500]}\n ) resp client.chat.completions.create( modelmodel, messages[ {role: system, content: SUMMARY_SYSTEM_PROMPT}, {role: user, content: user_content}, ], temperature0.3, response_format{type: json_object}, ) content resp.choices[0].message.content data parse_json(content) return SummarizedItem( titleitem.title, urlitem.url, clean_summaryitem.clean_summary, daily_summarydata[summary], sourceitem.source, )这里我刻意让函数只处理一条资讯而不是批量处理。原因在于批量会让 prompt 变长模型更容易把 A 文章的事实写到 B 文章的摘要里。日报场景每天几十条内容即使单条调用也花不了太久换来的是更稳定的输出。如果一定要批量调用以降低延迟可以在输入中为每篇文章加id并强制模型在返回结果里带着id回传。这样后续校验时才能确认“每一条摘要对应的确实是同一条原文”。parse_json是模型工作流里最容易被忽略的组件。它不能只是json.loads还要处理模型偶尔输出前后缀文本、Markdown 代码块、逗号错误等问题。import json import re def parse_json(content: str) - dict: text content.strip() text re.sub(r^(?:json)?, , text) text re.sub(r$, , text).strip() try: return json.loads(text) except json.JSONDecodeError: start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(模型返回值中未找到 JSON 对象) return json.loads(text[start:end 1])这里有一个值得记住的原则模型返回的内容永远是“不可信输入”不能因为看起来像 JSON 就直接当作 JSON 使用。注意不要把response_format{type: json_object}当成万能开关。部分本地模型或兼容服务并不支持该参数即使支持也仍然需要客户端做 JSON 解析和字段校验。3.4 分类与排序用结构化输出约束模型结果摘要生成之后日报需要把条目按照“技术、产品、会议、开源、商业、其他”等分类整理并给出重要性打分。分类和打分适合放在同一个模型请求里因为它们都属于“对一条已有摘要的判定”。这里引入 Pydantic 来定义模型输出结构让校验在数据到达下一节点之前完成。from pydantic import BaseModel, Field class CategoryResult(BaseModel): url: str Field(description资讯链接必须与输入一致) category: str Field(description分类名) score: int Field(ge0, le10, description重要性打分) reason: str Field(description打分理由)分类调用的输入是上一步生成的摘要而不是原始全文。因为摘要更短模型更容易做出稳定判断同时也能节省 Token。def classify_one(item: SummarizedItem, client: OpenAI, model: str) - CategorizedItem: payload { url: item.url, title: item.title, daily_summary: item.daily_summary, } resp client.chat.completions.create( modelmodel, messages[ {role: system, content: CATEGORY_SYSTEM_PROMPT}, {role: user, content: json.dumps(payload, ensure_asciiFalse)}, ], temperature0.1, response_format{type: json_object}, ) raw parse_json(resp.choices[0].message.content) result CategoryResult.model_validate(raw) return CategorizedItem( titleitem.title, urlitem.url, daily_summaryitem.daily_summary, categoryresult.category, scoreresult.score, reasonresult.reason, sourceitem.source, )这里使用CategoryResult.model_validate(raw)做校验。如果模型返回的score是字符串或者缺少url字段Pydantic 会直接抛异常。你要在编排器里捕获这个异常决定是重试一次还是降级为默认分类。排序为什么不用模型做因为排序可以交给 Python 的sort。模型只需要给出score最终顺序由规则决定这样更可控也更容易复现。def rank_items(items: list[CategorizedItem], top_n: int) - list[CategorizedItem]: sorted_items sorted(items, keylambda x: x.score, reverseTrue) return sorted_items[:top_n]这样设计的好处是如果你想让“商业”类优先展示只需要在规则里调整权重不需要重新调用模型。3.5 渲染阶段把结构化数据交给 Jinja2 输出 Markdown日报的 Markdown 结构应该稳定标题、日期、分类分组、条目列表。这个工作用 Jinja2 模板实现比让模型生成更可靠。from jinja2 import Template REPORT_TEMPLATE # {{ date }} 资讯日报 {% for group_name, group_items in grouped_items %} ## {{ group_name }} {% for item in group_items %} - [{{ item.title }}]({{ item.url }})重要度 {{ item.score }} - {{ item.daily_summary }} - 来源{{ item.source }} {% endfor %} {% endfor %} def render_report(items: list[CategorizedItem], report_date: str) - str: grouped: dict[str, list[CategorizedItem]] {} for item in items: grouped.setdefault(item.category, []).append(item) for key in grouped: grouped[key].sort(keylambda x: x.score, reverseTrue) template Template(REPORT_TEMPLATE) return template.render( datereport_date, grouped_itemsgrouped, )为了让渲染结果便于排查建议在这个节点把日报内容同时写进文件和控制台日志。不要把渲染逻辑和保存逻辑耦合在一起保存属于持久化节点后续可以替换成邮件发送、飞书机器人通知或者自动发布流程。3.6 编排器把各阶段串成可重试、可断点的状态流所有节点准备好之后需要一个编排器把它们按顺序组织起来。编排器是最能体现“Agent 架构师”工作量的地方它决定了一个节点失败时是整个流程回滚还是选择跳过并继续。最简单的编排器是一个顺序执行的函数def run_workflow(cfg: Config) - WorkflowContext: client OpenAI(base_urlcfg.base_url, api_keycfg.api_key) ctx WorkflowContext() ctx.raw_items [] for source in cfg.sources: try: ctx.raw_items.extend(fetch_rss(source)) except Exception as exc: print(f[warn] fetch failed: {source}, {exc}) ctx.cleaned_items dedup_items(clean_items(ctx.raw_items)) print(f[info] cleaned items: {len(ctx.cleaned_items)}) for item in ctx.cleaned_items[:cfg.max_items_per_source * len(cfg.sources)]: try: ctx.summarized_items.append( summarize_one(item, client, cfg.model) ) except Exception as exc: print(f[warn] summarize failed: {item.url}, {exc}) continue for item in ctx.summarized_items: try: ctx.categorized_items.append( classify_one(item, client, cfg.model) ) except Exception as exc: print(f[warn] classify failed: {item.url}, {exc}) ctx.categorized_items.append( CategorizedItem( titleitem.title, urlitem.url, daily_summaryitem.daily_summary, category其他, score0, reason规则降级, sourceitem.source, ) ) ranked rank_items(ctx.categorized_items, cfg.report_top_n) ctx.report render_report(ranked, cfg.report_date) return ctx这个编排器解决了几个关键问题。采集失败时只跳过单个来源不终止整个工作流。摘要失败时丢弃这条资讯因为日报宁缺毋滥。分类失败时使用规则降级为“其他”和分数 0保证日报结构完整。渲染阶段基于排好序的数据生成 Markdown此时已经不再依赖模型。如果项目继续变大比如要增加“人工审核节点”“历史相似文章过滤节点”建议换成 LangGraph 或自研状态机。LangGraph 的StateGraph允许把每个节点定义成独立函数并且通过显式边表达流转关系代码结构会更接近流程图# 示意代码需要安装 langgraph 和相关依赖 from langgraph.graph import StateGraph, END graph StateGraph(WorkflowContext) graph.add_node(fetch, fetch_node) graph.add_node(summary, summary_node) graph.add_node(classify, classify_node) graph.add_node(render, render_node) graph.set_entry_point(fetch) graph.add_edge(fetch, summary) graph.add_edge(summary, classify) graph.add_edge(classify, render) graph.add_edge(render, END)这里使用同一个WorkflowContext作为状态类型和前面的设计是一致的。换成 LangGraph 不是重写业务逻辑而是把编排方式从顺序函数升级为显式状态图排查问题时能更直观地看到当前停在哪个节点。4. 运行、验证和结果分析4.1 运行命令与预期日志确保.env配置正确后在项目根目录运行python main.py这里的main.py只需要几行代码from config import load_config from workflow import run_workflow if __name__ __main__: cfg load_config() ctx run_workflow(cfg) print(ctx.report)第一次运行建议只配一个 RSS 源、MAX_ITEMS_PER_SOURCE设小一点比如 5。这样做是为了先验证“模型调用是否通、JSON 解析是否成功、Markdown 渲染是否正常”再逐步扩大到全量来源。正常的日志大致如下[info] cleaned items: 5 [info] summarize done: 5 [info] classify done: 5 [info] report rendered, length2140如果某一条摘要失败会在日志里看到[warn] summarize failed这是预期中的降级行为不代表工作流崩溃。4.2 输出检查既看格式也看内容质量运行成功不等于结果正确。日报工作流至少要从三个层面检查输出。第一层是格式。Markdown 标题是否齐全分类是否分组链接是否可点击日期是否正确。如果发现None或空字符串说明清洗阶段有遗漏。第二层是结构。检查输出 JSON 中的url字段是否和输入一致分类名是否都在允许的集合内打分是否都在 0 到 10 之间。这个检查可以由校验脚本自动完成也可以在开发阶段人工抽样。第三层是内容质量。随机抽取 5 条摘要对照原文看是否有事实错误是否出现“张冠李戴”。有些模型喜欢在摘要里补充“业内专家表示”“这一技术突破”等原文没有的内容这类输出在日报里应该被拦截。建议在代码里加入一个简单的模型检查器专门负责验证模型输出是否满足字段约束def validate_category_output(url: str, raw: dict) - CategoryResult: result CategoryResult.model_validate(raw) if result.url ! url: raise ValueError(furl mismatch: {result.url} ! {url}) allowed {技术, 产品, 会议, 开源, 商业, 其他} if result.category not in allowed: raise ValueError(fcategory not allowed: {result.category}) return result模型检查器是模型工作流里容易被低估的一环。没有它任何一次模型输出异常都会直接污染最终日报而且往往是在读者看到错误内容之后才被发现。4.3 Token 成本与耗时估算资讯日报工作流的成本主要来自摘要和分类两个模型节点。假设每天 50 条资讯摘要输入约 500 字输出约 100 字分类输入约 200 字输出约 50 字那么单次调用的 Token 消耗大概是节点输入 Token 估算输出 Token 估算调用次数摘要50 条 x 70050 条 x 10050分类50 条 x 30050 条 x 10050合计5000010000100这只是一个数量级参考实际会因模型、摘要长度和是否启用 JSON 模式而不同。重点在于工作流里的 Token 消耗是可预测的而不是一次性把所有内容塞进一个大请求。如果成本超标优先做三件事。第一清洗阶段把正文截断到 300 到 500 字。第二把摘要和分类合并成一次调用前提是你已经验证过批量输出不会导致数据错位。第三对已经处理过的 URL 做缓存重复运行时不重复调用模型。5. 模型参与工作流时的常见故障与排查路径5.1 模型不返回合法 JSON现象程序在parse_json抛出ValueError或者CategoryResult.model_validate报字段缺失。原因通常是三类。模型服务不支持response_format即使传了也没效果提示词没有明确指出“必须返回 JSON”模型输出被长度截断导致 JSON 结束符丢失。检查顺序先打印模型返回的原始内容确认是不是 JSON再确认调用时是否真的传入了response_format最后检查模型服务日志看是否有截断或超时记录。解决方案去掉response_format改为在系统提示词里明确要求 JSON。调用成功后增加 schema 校验用 Pydantic 强制字段类型。做一次重试重试时把原始错误信息拼进用户消息提醒模型修正格式。预防建议不要假设模型输出永远是合法 JSON要把它当成外部输入对待。5.2 上下文超长导致请求失败现象模型调用返回类似context length exceeded的错误或者请求总是超时。原因很简单单个 RSS 条目的正文可能很长或者一条请求里带入了太多条目。检查方式是看请求体的字符数和模型服务的最大上下文长度做对比。解决方式清洗阶段截断正文clean_summary只保留前 500 字。摘要节点每次只处理一条。分类节点输入摘要而不是全文。如果依然超长考虑用向量检索召回相关片段再用模型加工而不是把全部内容塞给模型。日报工作流里很容易犯“为了让模型理解更多就把更多内容扔给它”的错。实际效果恰恰相反内容越长模型越容易忽略关键信息。5.3 摘要内容张冠李戴现象日报摘要读起来很流畅但和原文链接里的内容对不上或者把另外一篇文章的事实写进了当前摘要。根因通常是批量调用导致模型混淆上下文或者 prompt 约束不足。检查方式把模型返回的摘要和输入原文打印到一行人工对照。如果发现批量请求里返回的id和输入不一致基本可以确定是张冠李戴。解决方案是强制模型在返回 JSON 中携带原始url字段并用代码校验返回的url与输入url是否一致。校验不过就丢弃本条结果而不是继续往下游传递。推荐做法日报场景对延迟不敏感摘要单独调用分类可以小批量但也要携带url做关联校验。5.4 某个节点失败拖垮整个流程现象一个 RSS 源超时导致整个日报没有生成或者一个分类调用失败导致后面的渲染全部停止。根因是编排器没有做好节点隔离。你在run_workflow里看到的 try except 就是为了避免这种连锁失败。检查方式看日志里是哪一个阶段先出现异常。如果异常出现在采集阶段但摘要和分类没有执行说明异常没有被捕获如果异常出现在摘要阶段但整个程序终止说明缺少单条失败继续处理的逻辑。解决方式每个节点独立 try except。对模型调用设置超时和重试次数。对失败条目执行“跳过”或“降级”不让数据停留在半成品状态。将中间产物定期保存到本地文件比如artifacts/summarized.json方便断点排查。生产环境还要考虑更严格的幂等设计。即使重试导致某一条被处理两次也要保证最终的日报内容不会出现重复。5.5 重复采集导致日报内容雷同现象连续两天生成的日报里大量条目是重复的。根因是 RSS 源本身会把老文章保留在 feed 里而当前去重只使用了单次运行内部的seen集合没有跨天记忆。检查方式查看cleaned_items里相同标题的出现次数或者统计当天与前一天的 URL 交集。解决方式把历史 URL 存入 SQLite查询时跳过已出现过的链接。联合标题 链接生成哈希避免不同源链接导致重复。引入基于摘要的向量相似度去重但这是后期优化不必在第一步就做。日报工具的价值是“每天提供新信息”重复内容会直接削弱产品价值因此这条排查路径值得格外重视。6. 从日报生成器到生产级 Agent 工作流6.1 平台化路线与自研路线的取舍日报生成器跑通后你可能会考虑要不要直接用现成工作流平台比如 Dify、Coze、n8n或者继续自研。这里没有绝对答案取决于团队目标和维护能力。方案适合场景需要关注的点Dify快速搭建可视化 Agent 应用业务人员也能改流程需要维护平台版本和模型配置Coze快速验证插件、对话类 Agent调试方便平台绑定程度较高迁移成本需要评估n8n偏自动化任务编排适合对接 Webhook、邮件、数据库模型节点能力依赖你接入的 API 和插件LangGraph代码团队希望把状态机写进版本库方便测试和回滚需要自己编写节点和状态对象自研脚本流程固定、逻辑简单、依赖少后续扩展时编排代码会越来越难维护这篇实战里的run_workflow其实就是一个最简自研编排器。它的优点是清晰、可控、容易排查缺点是节点一多代码里会出现大量 if else 和异常捕获边界会越来越模糊。这时候可以平滑迁移到 LangGraph因为状态机制是一致的。6.2 引入向量检索和排序让信息量不只靠时间日报工作流还有一个常见瓶颈RSS 源里可能有上百条资讯但日报只需要 10 条。简单按时间取前面几条很容易漏掉重要信息让模型给每篇打分Token 成本又高。更合理的生产方案是分两层处理。第一层用规则和向量检索做粗召回。先把资讯变成向量存入向量库再用用户关心的主题向量检索最相关的候选集。第二层再把候选集交给模型做精细打分和分类。如果你还需要判断“两条资讯是否在讲同一件事”可以引入 reranker 模型做二次排序会有更准确的效果。引入这些能力的前提是你已经拥有或者能部署 embedding 和 reranker 推理服务。它们可以以 OpenAI 兼容接口或独立 HTTP 接口的方式接入现有工作流不需要推翻日报生成器的主体结构。这里要注意向量检索和 reranker 是锦上添花的扩展不该在第一版就引入。先把规则、模型、校验、降级这四层做扎实再考虑信息召回质量。6.3 上线前检查清单生产环境的日报生成器不只是“能出 Markdown”还需要具备可运营性。下面是一份可以直接使用的上线前检查清单。检查项具体内容模型服务可用性确认模型接口地址、超时时间、最大重试次数密钥管理API Key 不在代码仓库中出现由部署平台注入节点异常策略每个节点都定义了失败时是重试、跳过还是降级输出校验所有模型输出经过 JSON 解析和 schema 校验中间产物采集、摘要、分类结果可落盘便于排查重复控制跨天去重逻辑已实现历史 URL 有持久化存储Token 监控统计每天的输入输出 Token设置成本告警质量抽查每天有固定流程抽查摘要和分类正确性告警通知工作流整体失败时能通过邮件或消息机器人收到通知回滚方案旧模板、旧配置、旧依赖有备份能快速切换这份清单同样适用于其他模型参与的定时任务类工作流。核心判断标准是如果今天我收到告警说“日报没生成”我能不能在 10 分钟内定位到是采集、模型还是渲染出了问题。6.4 学习建议如何从单机脚本走向 Agent 架构师这篇实战完成的是一个顺序工作流但 Agent 架构师真正要掌握的是“模型能决策、系统能兜底”的设计能力。建议按照下面顺序继续练习。第一给日报生成器增加“人工审核节点”。模型把分类结果输出后需要人工批准才能发布研究一下如何在状态机里表达“等待人工”这个状态。第二把摘要和分类改成由模型决定“是否需要更多信息”。当模型判断某条资讯信息不足时工作流自动触发一次网页正文抓取再重新摘要。这就是从“单向工作流”走向“带反馈回路的 Agent 行为”。第三研究 ReAct 这类 Agent 循环。让模型在“推理、调用工具、观察结果、再次推理”之间循环。你会发现日报生成器里的模型检查器、JSON 校验、降级策略在 Agent 循环里依然适用。第四回到 Dify、Coze 或 LangGraph把你写好的节点翻译成可视化流程图。对比两种实现方式的差异能加深对工作流状态管理的理解。最后一件事不要追求把所有场景都做成 Agent。资讯日报生成器里采集是规则、摘要是模型、分类是模型、排序是规则、渲染是模板这种“规则与模型混合编排”的方式才是生产系统里最常见的形态。稳定的工作流不是模型强大到不会出错而是模型出错时系统仍有办法给出一个可接受的结果。