尧图网页设计 
从业务痛点出发用模型编排一条“资讯日报自动生产流水线”把 Agent 架构师视角下的任务拆分、模型参与决策和容错降级思路讲清楚。本文的代码可以直接运行重点不是贴一堆零散接口而是让你理解复杂工作流到底应该怎么设计以及每一步为什么这样做。1. 背景与核心概念先说一个很常见的场景。很多团队每天早上需要有人把竞品动态、行业新闻、技术博客、开源项目更新整理成一份日报发到群里或者邮件列表里。早期做法是人工搜索、复制、排版一个人光收集信息就要占用半小时以上还要花时间写摘要、加点评。后来有人用脚本抓取 RSS 或接口再拼出固定模板的日报虽然省了人工但内容只是“原样堆砌”没有分类、没有重点、没有摘要读起来像信息流而不是日报。于是资讯日报生成器这类应用开始走向“模型参与的复杂工作流”形态让大模型在流程的关键节点参与决策——判断文章重要程度、给内容打标签、生成摘要、写导读、挑出值得推荐的内容最后再排版输出。这就是 Agent 架构师课程第 5 课要实战的内容。从概念上说Agent 架构的核心不是“单次调用一个大模型”而是把任务拆成多个节点让模型在不同的节点承担不同的职责。资讯日报生成器恰好是一个典型的多节点协作场景。如果把整套系统拆开看至少包含以下节点节点职责模型参与方式采集节点从 RSS、API、网页抓取原始内容不参与或仅参与源筛选解析节点清洗正文、去重、提取元信息不参与分类节点判断内容属于哪个主题分类模型参与分类决策摘要节点对每篇文章生成一句话摘要模型参与内容理解推荐节点从候选内容中挑出值得收录的条目模型参与重要性评估编排节点按模板组装日报不参与或模型参与格式生成所以本课实战并不是简单写一个requests.get加openai.ChatCompletion.create就完事而是要设计一套可维护、可扩展、可容错的工作流让“模型参与”成为一个稳定可复用的能力。2. 系统目标与工作流总览2.1 日报生成器的功能目标在设计之前先明确资讯日报生成器的功能目标每天定时抓取多个资讯源。过滤掉重复内容和无效内容。对文章进行主题分类。为每篇文章生成摘要。挑选出最值得阅读的 Top 内容。按固定 Markdown 模板生成日报正文。最终输出到本地文件可继续扩展发送到钉钉、飞书、邮件或企业微信群。这个过程如果全部用规则实现代码会非常复杂而且无法适应内容变化。比如想判断“这篇文章是不是重要的行业动态”规则很难写但模型可以结合标题、正文和时间因素给出判断。2.2 模型参与的复杂工作流设计可以把整套流程分成两个层次理解。第一层是固定流程层。这个层次使用代码串联任务节点例如获取数据、清洗数据、调用模型、写文件这些步骤是确定的不依赖模型决策。第二层是模型决策层。这个层次的任务无法使用固定规则解决必须交给大模型。例如判断一篇技术博客属于“前端”还是“后端”或者生成一段适合日报阅读的摘要。整个工作流可以用下面的流水线表示读取配置文件获取资讯源列表和模型参数。并行或串行抓取每个资讯源的数据。清洗 HTML 正文提取标题、链接、发布时间。对文章做去重处理。调用大模型完成文章分类。调用大模型生成摘要并评估重要性。根据评分过滤出日报候选内容。调用大模型生成日报导读或按模板装配 Markdown。输出最终日报文件。这里的关键设计思路是不要把模型调用写死在一个大函数里而是把“模型能力”抽象成可复用的服务每个节点按需调用。2.3 技术选型本文的实战示例采用 Python 实现使用的第三方库尽量保持精简库用途requests抓取 RSS 或 API 数据feedparser解析 RSS Feedopenai调用大模型接口python-dotenv读取环境变量管理 API Keyjinja2生成日报模板如果出于合规或网络环境限制无法直接访问某些接口可以把大模型客户端替换为兼容 OpenAI 协议的本地模型服务代码逻辑不变。这一点只做思路说明具体接入方式需根据你所在环境选择合法的模型服务渠道。3. 环境准备与项目结构3.1 环境版本说明本文示例代码基于 Python 3.10 以上版本开发。大模型接口兼容 OpenAI Chat Completions 格式具体版本需要根据你的项目实际情况调整。建议使用虚拟环境安装依赖python -m venv venv source venv/bin/activate然后安装依赖pip install requests feedparser openai python-dotenv jinja2需要提醒的是不同版本的 openai 库 API 调用方式有差异。本文以较新的接口风格为例如果你的 openai 版本较低需要将客户端初始化方式按官方文档调整。3.2 项目目录结构实际项目采用下面的目录结构daily-news-agent/ ├── .env ├── config.yaml ├── requirements.txt ├── src/ │ ├── __init__.py │ ├── config.py │ ├── models.py │ ├── fetcher.py │ ├── parser.py │ ├── llm_client.py │ ├── summary.py │ ├── workflow.py │ └── daily_builder.py ├── output/ │ └── daily_report.md └── main.py其中config.yaml存放资讯源和分类规则src/llm_client.py封装模型调用src/workflow.py负责编排整个工作流main.py是入口。3.3 配置示例创建.env文件内容如下LLM_API_KEYyour_api_key_here LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini这里的LLM_BASE_URL可以根据你实际使用的模型服务调整。注意不要把自己的 API Key 提交到代码仓库应该通过.gitignore忽略.env文件。再创建config.yamlsources: - name: 开源中国 type: rss url: https://www.oschina.net/news/rss - name: 36氪 type: rss url: https://36kr.com/feed - name: InfoQ 中文 type: rss url: https://www.infoq.cn/feed categories: - 前端 - 后端 - 人工智能 - 云计算 - 开源工具 - 业界动态 max_items: 20 output_path: output/daily_report.md把资讯源放在配置文件里后续新增源只需要修改配置不需要改动代码。4. 核心模块实现4.1 配置加载模块先实现src/config.py负责读取 YAML 和环境变量。这里没有引入pyyaml如果需要解析 YAML可以在依赖中加上pyyaml然后使用yaml.safe_load。# src/config.py import os from pathlib import Path from typing import Any, Dict PROJECT_ROOT Path(__file__).resolve().parent.parent def load_config() - Dict[str, Any]: 加载 config.yaml 配置。 config_path PROJECT_ROOT / config.yaml if not config_path.exists(): raise FileNotFoundError(f配置文件不存在: {config_path}) # 这里使用 yaml 库读取也可以在依赖中安装 pyyaml import yaml with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) # 从环境变量读取模型配置 config[llm] { api_key: os.getenv(LLM_API_KEY, ), base_url: os.getenv(LLM_BASE_URL, ), model: os.getenv(LLM_MODEL, gpt-4o-mini), } if not config[llm][api_key]: raise ValueError(请先在 .env 中配置 LLM_API_KEY) return config这里把配置集中管理后续调整资讯源、模型参数都不需要改动主流程代码。4.2 数据模型定义为了让工作流中的数据传递更清晰可以定义几个简单的数据类。# src/models.py from dataclasses import dataclass, field from typing import Optional dataclass class NewsItem: 标准化后的资讯条目。 title: str url: str source: str published: str summary: str category: str importance_score: float 0.0 raw_content: str def to_dict(self): return { title: self.title, url: self.url, source: self.source, published: self.published, summary: self.summary, category: self.category, importance_score: self.importance_score, }定义NewsItem数据类以后不同模块之间传递的就是结构化的对象而不是字典或字符串代码可读性会好很多。4.3 资讯采集模块采集模块负责从 RSS 源获取数据。使用feedparser解析 RSS 是常见做法。# src/fetcher.py import time import feedparser from typing import List, Dict, Any from src.models import NewsItem def fetch_rss(url: str, source_name: str, max_items: int 30) - List[NewsItem]: 抓取单个 RSS 源返回标准化条目。 try: feed feedparser.parse(url) except Exception as e: print(f[采集失败] {source_name}: {e}) return [] items [] for entry in feed.entries[:max_items]: title entry.get(title, ).strip() link entry.get(link, ).strip() published entry.get(published, ) if not title or not link: continue item NewsItem( titletitle, urllink, sourcesource_name, publishedpublished, ) items.append(item) return items def fetch_all_sources(sources: List[Dict[str, Any]]) - List[NewsItem]: 串行抓取所有资讯源。 all_items [] for source in sources: source_type source.get(type, rss) url source.get(url, ) name source.get(name, 未知来源) if source_type rss: items fetch_rss(url, name) else: print(f[跳过] 不支持的源类型: {source_type}) items [] print(f[抓取完成] {name}: 获取 {len(items)} 条) all_items.extend(items) time.sleep(0.5) # 避免对目标站点造成压力 return all_items实际生产项目中抓取频率需要遵守目标网站的 robots 协议和服务条款同时做好限速和缓存。4.4 大模型客户端封装模型调用是整套工作流的核心。要考虑三个问题不同业务节点需要不同的 prompt。模型输出可能不稳定需要容错。多次调用模型需要注意 Token 成本。封装src/llm_client.py# src/llm_client.py import json from typing import List, Dict, Any from openai import OpenAI class LLMClient: 通用大模型客户端封装。 def __init__(self, api_key: str, base_url: str, model: str): self.model model self.client OpenAI(api_keyapi_key, base_urlbase_url) def chat(self, messages: List[Dict[str, str]], temperature: float 0.3) - str: 发送对话消息返回文本内容。 try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, ) return response.choices[0].message.content or except Exception as e: print(f[模型调用失败] {e}) return def chat_json(self, messages: List[Dict[str, str]], temperature: float 0.2) - Dict[str, Any]: 发送对话消息并解析 JSON 输出。 messages messages [ { role: system, content: 请只输出 JSON 格式内容不要输出额外解释。, } ] content self.chat(messages, temperaturetemperature) try: return json.loads(content) except json.JSONDecodeError: # 尝试提取第一个 { ... } 片段 start content.find({) end content.rfind(}) if start ! -1 and end ! -1 and end start: try: return json.loads(content[start : end 1]) except json.JSONDecodeError: return {} return {}chat_json方法非常重要。资讯分类、摘要、重要性评估都可以要求模型输出 JSON然后由代码解析。但模型偶尔会输出多余文字因此我们做了两段式提取先整体解析失败后提取花括号片段。4.5 模型参与的文章分类节点接下来实现一个“模型参与分类”的函数。它的输入是一篇NewsItem输出是分类结果。# src/classifier.py from typing import List from src.models import NewsItem from src.llm_client import LLMClient CATEGORY_SYSTEM_PROMPT 你是一名资讯分类专家。 请根据文章标题和内容将文章划分到以下分类之一{categories} 请只输出 JSON格式为{{category: 分类名, reason: 分类理由}} def classify_item(llm: LLMClient, item: NewsItem, categories: List[str]) - str: 调用模型对单篇文章分类。 category_text 、.join(categories) system_prompt CATEGORY_SYSTEM_PROMPT.format(categoriescategory_text) user_content f文章标题{item.title}\n文章来源{item.source} result llm.chat_json( [ {role: system, content: system_prompt}, {role: user, content: user_content}, ] ) category result.get(category, ) if category not in categories: return 未分类 return category这里没有把文章全文带入因为单次分类只需要标题和来源就能完成能有效降低 Token 消耗。如果是更复杂的分类任务可以再把正文摘要加入。4.6 模型参与的摘要与评分节点摘要和重要性评分可以合并成一次模型调用减少调用次数。# src/summary.py from src.models import NewsItem from src.llm_client import LLMClient SUMMARY_SYSTEM_PROMPT 你是一名资深编辑负责为日报撰写内容摘要。 请根据文章内容生成一句话摘要并给出重要性评分1-10分10分最重要。 请只输出 JSON格式为 {{summary: 一句话摘要, importance_score: 8}} def summarize_item(llm: LLMClient, item: NewsItem) - NewsItem: 调用模型生成摘要和评分并更新到 item 上。 user_content f文章标题{item.title}\n文章内容{item.raw_content[:800]} result llm.chat_json( [ {role: system, content: SUMMARY_SYSTEM_PROMPT}, {role: user, content: user_content}, ] ) summary result.get(summary, ) try: score float(result.get(importance_score, 0)) except (TypeError, ValueError): score 0 item.summary summary item.importance_score score return item抓取 RSS 时通常拿不到完整正文所以raw_content可能为空。在实际项目中可以增加一个正文抓取逻辑比如用requests抓取链接后提取p标签文本。4.7 日报组装模块日报输出使用jinja2模板。模板文件可以放在项目templates/目录下。# src/daily_builder.py from pathlib import Path from typing import List from jinja2 import Environment, FileSystemLoader from src.models import NewsItem REPORT_TEMPLATE # 每日资讯日报 生成时间{{ generated_time }} {% for category, items in grouped_items.items() %} ## {{ category }} {% for item in items %} - [{{ item.title }}]({{ item.url }}) - 来源{{ item.source }} - 摘要{{ item.summary }} - 重要度{{ item.importance_score }} {% endfor %} {% endfor %} def build_markdown_report(items: List[NewsItem], generated_time: str) - str: 按分类组装 Markdown 日报。 env Environment() template env.from_string(REPORT_TEMPLATE) grouped_items {} for item in items: category item.category or 未分类 grouped_items.setdefault(category, []).append(item) # 对每个分类内部的条目按重要度排序 for category in grouped_items: grouped_items[category].sort(keylambda x: x.importance_score, reverseTrue) return template.render( generated_timegenerated_time, grouped_itemsgrouped_items, )直接在代码里写模板字符串可以让示例更直观实际项目可以把模板抽成独立文件方便编辑。5. 工作流编排模块5.1 设计工作流状态在复杂工作流中多个节点需要共享数据。最简单可靠的方式是使用一个WorkflowContext对象作为统一的上下文。# src/workflow.py from pathlib import Path from datetime import datetime from typing import List, Dict, Any from src.config import load_config from src.fetcher import fetch_all_sources from src.llm_client import LLMClient from src.classifier import classify_item from src.summary import summarize_item from src.daily_builder import build_markdown_report from src.models import NewsItem class WorkflowContext: 工作流上下文保存中间状态。 def __init__(self): self.config None self.llm None self.all_items: List[NewsItem] [] self.filtered_items: List[NewsItem] [] self.report_content: str 5.2 编排流程工作流编排函数可以按节点拆开这样每一步都可以单独测试。def run_workflow() - Path: 执行资讯日报生成工作流。 ctx WorkflowContext() # 1. 加载配置 ctx.config load_config() # 2. 初始化模型客户端 llm_config ctx.config[llm] ctx.llm LLMClient( api_keyllm_config[api_key], base_urlllm_config[base_url], modelllm_config[model], ) # 3. 抓取资讯 print( 开始抓取资讯 ) ctx.all_items fetch_all_sources(ctx.config[sources]) print(f共抓取到 {len(ctx.all_items)} 条资讯) # 4. 分类和摘要 print( 模型分类与摘要 ) categories ctx.config[categories] for item in ctx.all_items: item.category classify_item(ctx.llm, item, categories) item summarize_item(ctx.llm, item) print(f [{item.category}] {item.title}) # 5. 过滤低质量内容 print( 过滤低质量内容 ) ctx.filtered_items [item for item in ctx.all_items if item.importance_score 5] ctx.filtered_items.sort(keylambda x: x.importance_score, reverseTrue) ctx.filtered_items ctx.filtered_items[: ctx.config.get(max_items, 20)] # 6. 生成 Markdown 日报 print( 生成日报 ) generated_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) ctx.report_content build_markdown_report(ctx.filtered_items, generated_time) # 7. 输出文件 output_path Path(ctx.config[output_path]) output_path.parent.mkdir(parentsTrue, exist_okTrue) output_path.write_text(ctx.report_content, encodingutf-8) print(f日报已生成{output_path}) return output_path这里使用了最简单的串行编排方式。真实生产环境中如果资讯源很多可以对抓取和摘要节点做并发优化例如使用ThreadPoolExecutor。但并发会引入速率限制和 API 超时的问题本文先以串行为主保证逻辑清晰。5.3 入口文件main.py只需要调用run_workflow()# main.py from src.workflow import run_workflow if __name__ __main__: run_workflow()这样设计的好处是后续如果你想增加“发送到钉钉”或“发送到邮件”的能力只需要在run_workflow末尾再追加一个节点不必改动既有节点。6. 运行与验证6.1 运行命令在项目根目录执行python main.py如果一切正常你会看到类似下面的日志 开始抓取资讯 [抓取完成] 开源中国: 获取 25 条 [抓取完成] 36氪: 获取 20 条 [抓取完成] InfoQ 中文: 获取 18 条 共抓取到 63 条资讯 模型分类与摘要 [人工智能] 某某大模型发布新版本 [后端] 微服务架构演进实践 [开源工具] 一款新的命令行工具 ... 过滤低质量内容 生成日报 日报已生成output/daily_report.md6.2 验证输出打开output/daily_report.md应该能看到分类清晰、带摘要和重要度评分的资讯列表。# 每日资讯日报 生成时间2025-01-02 09:30:00 ## 人工智能 - [某某大模型发布新版本](https://example.com/news/1) - 来源36氪 - 摘要该大模型在推理能力上实现大幅提升并开放了 API 接口。 - 重要度9.0到这里一个简单的模型参与资讯日报工作流已经能够运行。7. 常见问题与排查思路在实际使用过程中最常见的问题集中在模型调用、数据抓取和输出格式三方面。问题现象常见原因解决思路模型返回空字符串API Key 错误、网络超时或余额不足检查.env配置单独写脚本测试模型接口连通性模型返回内容不是 JSON模型输出不稳定增强chat_json容错逻辑或改用更低 temperatureRSS 抓取不到数据目标站点屏蔽爬虫或网络不通检查 URL 是否能直接访问尝试添加 User-Agent 请求头分类结果不准确Prompt 分类定义不够清晰增加分类描述和示例或者给模型提供正文摘要Token 消耗过快每篇文章都传入了全文限制raw_content长度或先提取正文关键段落日报内容太多没有设置过滤阈值调整max_items和importance_score阈值这里列一个排查 checklist先测试 API 密钥能否正常调用模型接口。再单独测试一个 RSS 源能否抓取到数据。手动构造一条NewsItem测试分类和摘要节点。确认输出文件路径是否有写入权限。查看日志中每个节点的输出数量是否合理。8. 模型参与复杂工作流的工程建议8.1 模型调用要做降级与缓存资讯日报生成器涉及多轮模型调用如果模型服务不稳定整个工作流可能中断。建议在工程层面做两件事。第一是降级。如果摘要节点调用失败可以直接用文章标题作为摘要而不是中断整个流程。如果分类节点失败可以回退到“未分类”。这样做不能保证结果完美但能保证日报不会被意外打断。第二是缓存。同一天对同一篇文章的摘要和分类结果是稳定不变的。可以在本地维护一个简单缓存键是文章 URL值是模型返回结果。下次再遇到同一文章时直接读缓存不再调用模型。8.2 控制上下文长度与 Token 成本RSS 抓取到的内容可能很长。如果每篇文章都传完整正文Token 消耗会很高。推荐做法是先提取正文前 300 到 500 字。去除 HTML 标签和无意义符号。再传给模型做摘要。对于评分任务甚至可以只传标题和第一段文字效果往往也不错。8.3 Prompt 设计要稳定模型参与的复杂工作流里Prompt 是核心资产。建议把不同的 Prompt 集中管理不要散落在业务代码里。可以单独创建prompts.py文件或者把 Prompt 写入配置目录。每次修改 Prompt 后记录版本变化方便回溯。8.4 工作流节点要可观测复杂工作流排错难度大建议为每个节点增加日志输出。至少包含节点名称。输入数据量。输出数据量。耗时。失败条数。比如fetch_all_sources返回后打印“来源 A 成功 25 条来源 B 失败 0 条”。这样日报内容变少时能快速定位是哪一步过滤掉了数据。8.5 安全与合规边界开发资讯采集类工具时需要注意目标网站的版权与使用条款不要无限频抓取也不要直接复制全文进行商用。如果模型服务需要付费务必设置预算上限避免异常循环导致费用飙升。在生产环境中API Key 应该放在密钥管理系统中而不是以明文形式写进配置仓库。如果使用 Git 管理项目建议提交前检查.env是否已被忽略。8.6 从串行编排走向可扩展架构本文的run_workflow是串行函数适合课程演示和中小规模场景。如果团队已经开始做 Agent 架构下一步可以考虑引入正式的流程引擎把每个节点抽象成独立组件。例如可以定义统一的接口class WorkflowNode: def execute(self, context: WorkflowContext) - WorkflowContext: pass每个节点只关心自己的输入输出由编排器负责组装。这样后续增加“查重节点”“人工审核节点”“多渠道推送节点”都只需要新增一个类不影响其他节点。在这个方向上可以进一步研究 Dify 和 Coze 这类可视化工作流平台它们同样围绕“节点编排 模型调用”的思路只不过把流程放到了图形界面上。理解本文的代码实现后再去看这些工作流平台会更容易理解节点的输入输出和模型连接方式。9. 后续扩展方向资讯日报生成器虽然是一个小项目但它几乎覆盖了 Agent 开发的基础套路任务拆分、模型调用、流程编排、容错降级、结果输出。你可以顺着下面几个方向继续扩展。第一把日报推送到不同渠道。例如封装一个Notifier接口分别实现钉钉机器人通知、飞书机器人通知和邮件通知。工作流只需要多执行一个推送节点。第二引入定时调度。使用cron或APScheduler每天早上 8 点自动运行一次。注意多线程运行时的并发问题避免上一个任务还没结束下一个任务又开始执行。第三个性化订阅。部分用户只关注人工智能分类可以为他们单独生成过滤后的日报。这就是把“用户偏好”引入工作流的一种方式。第四引入人工反馈机制。对模型生成的摘要做点赞/点踩积累数据后可用于 Prompt 调优。第五升级为更完整的 AI Agent。如果不只是做日报而是希望 Agent 能根据用户问题自己决定抓取哪些源、生成什么内容就需要引入 Agent 决策层让模型在“规划节点”选择执行路径。这就是后续课程中 Agent 架构师要解决的更深层问题。最后想说的是模型参与的复杂工作流并不神秘核心在于把“模型能力”和“工程能力”组合起来。模型负责理解、生成、决策工程负责采集、清洗、编排、输出、容错。两者结合才能做出真正稳定可用的产品。希望这篇实战笔记能帮你迈出第一步。如果你在运行过程中遇到问题欢迎留言交流。 用Agent架构打造资讯日报自动生成流水线
从业务痛点出发用模型编排一条“资讯日报自动生产流水线”把 Agent 架构师视角下的任务拆分、模型参与决策和容错降级思路讲清楚。本文的代码可以直接运行重点不是贴一堆零散接口而是让你理解复杂工作流到底应该怎么设计以及每一步为什么这样做。1. 背景与核心概念先说一个很常见的场景。很多团队每天早上需要有人把竞品动态、行业新闻、技术博客、开源项目更新整理成一份日报发到群里或者邮件列表里。早期做法是人工搜索、复制、排版一个人光收集信息就要占用半小时以上还要花时间写摘要、加点评。后来有人用脚本抓取 RSS 或接口再拼出固定模板的日报虽然省了人工但内容只是“原样堆砌”没有分类、没有重点、没有摘要读起来像信息流而不是日报。于是资讯日报生成器这类应用开始走向“模型参与的复杂工作流”形态让大模型在流程的关键节点参与决策——判断文章重要程度、给内容打标签、生成摘要、写导读、挑出值得推荐的内容最后再排版输出。这就是 Agent 架构师课程第 5 课要实战的内容。从概念上说Agent 架构的核心不是“单次调用一个大模型”而是把任务拆成多个节点让模型在不同的节点承担不同的职责。资讯日报生成器恰好是一个典型的多节点协作场景。如果把整套系统拆开看至少包含以下节点节点职责模型参与方式采集节点从 RSS、API、网页抓取原始内容不参与或仅参与源筛选解析节点清洗正文、去重、提取元信息不参与分类节点判断内容属于哪个主题分类模型参与分类决策摘要节点对每篇文章生成一句话摘要模型参与内容理解推荐节点从候选内容中挑出值得收录的条目模型参与重要性评估编排节点按模板组装日报不参与或模型参与格式生成所以本课实战并不是简单写一个requests.get加openai.ChatCompletion.create就完事而是要设计一套可维护、可扩展、可容错的工作流让“模型参与”成为一个稳定可复用的能力。2. 系统目标与工作流总览2.1 日报生成器的功能目标在设计之前先明确资讯日报生成器的功能目标每天定时抓取多个资讯源。过滤掉重复内容和无效内容。对文章进行主题分类。为每篇文章生成摘要。挑选出最值得阅读的 Top 内容。按固定 Markdown 模板生成日报正文。最终输出到本地文件可继续扩展发送到钉钉、飞书、邮件或企业微信群。这个过程如果全部用规则实现代码会非常复杂而且无法适应内容变化。比如想判断“这篇文章是不是重要的行业动态”规则很难写但模型可以结合标题、正文和时间因素给出判断。2.2 模型参与的复杂工作流设计可以把整套流程分成两个层次理解。第一层是固定流程层。这个层次使用代码串联任务节点例如获取数据、清洗数据、调用模型、写文件这些步骤是确定的不依赖模型决策。第二层是模型决策层。这个层次的任务无法使用固定规则解决必须交给大模型。例如判断一篇技术博客属于“前端”还是“后端”或者生成一段适合日报阅读的摘要。整个工作流可以用下面的流水线表示读取配置文件获取资讯源列表和模型参数。并行或串行抓取每个资讯源的数据。清洗 HTML 正文提取标题、链接、发布时间。对文章做去重处理。调用大模型完成文章分类。调用大模型生成摘要并评估重要性。根据评分过滤出日报候选内容。调用大模型生成日报导读或按模板装配 Markdown。输出最终日报文件。这里的关键设计思路是不要把模型调用写死在一个大函数里而是把“模型能力”抽象成可复用的服务每个节点按需调用。2.3 技术选型本文的实战示例采用 Python 实现使用的第三方库尽量保持精简库用途requests抓取 RSS 或 API 数据feedparser解析 RSS Feedopenai调用大模型接口python-dotenv读取环境变量管理 API Keyjinja2生成日报模板如果出于合规或网络环境限制无法直接访问某些接口可以把大模型客户端替换为兼容 OpenAI 协议的本地模型服务代码逻辑不变。这一点只做思路说明具体接入方式需根据你所在环境选择合法的模型服务渠道。3. 环境准备与项目结构3.1 环境版本说明本文示例代码基于 Python 3.10 以上版本开发。大模型接口兼容 OpenAI Chat Completions 格式具体版本需要根据你的项目实际情况调整。建议使用虚拟环境安装依赖python -m venv venv source venv/bin/activate然后安装依赖pip install requests feedparser openai python-dotenv jinja2需要提醒的是不同版本的 openai 库 API 调用方式有差异。本文以较新的接口风格为例如果你的 openai 版本较低需要将客户端初始化方式按官方文档调整。3.2 项目目录结构实际项目采用下面的目录结构daily-news-agent/ ├── .env ├── config.yaml ├── requirements.txt ├── src/ │ ├── __init__.py │ ├── config.py │ ├── models.py │ ├── fetcher.py │ ├── parser.py │ ├── llm_client.py │ ├── summary.py │ ├── workflow.py │ └── daily_builder.py ├── output/ │ └── daily_report.md └── main.py其中config.yaml存放资讯源和分类规则src/llm_client.py封装模型调用src/workflow.py负责编排整个工作流main.py是入口。3.3 配置示例创建.env文件内容如下LLM_API_KEYyour_api_key_here LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini这里的LLM_BASE_URL可以根据你实际使用的模型服务调整。注意不要把自己的 API Key 提交到代码仓库应该通过.gitignore忽略.env文件。再创建config.yamlsources: - name: 开源中国 type: rss url: https://www.oschina.net/news/rss - name: 36氪 type: rss url: https://36kr.com/feed - name: InfoQ 中文 type: rss url: https://www.infoq.cn/feed categories: - 前端 - 后端 - 人工智能 - 云计算 - 开源工具 - 业界动态 max_items: 20 output_path: output/daily_report.md把资讯源放在配置文件里后续新增源只需要修改配置不需要改动代码。4. 核心模块实现4.1 配置加载模块先实现src/config.py负责读取 YAML 和环境变量。这里没有引入pyyaml如果需要解析 YAML可以在依赖中加上pyyaml然后使用yaml.safe_load。# src/config.py import os from pathlib import Path from typing import Any, Dict PROJECT_ROOT Path(__file__).resolve().parent.parent def load_config() - Dict[str, Any]: 加载 config.yaml 配置。 config_path PROJECT_ROOT / config.yaml if not config_path.exists(): raise FileNotFoundError(f配置文件不存在: {config_path}) # 这里使用 yaml 库读取也可以在依赖中安装 pyyaml import yaml with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) # 从环境变量读取模型配置 config[llm] { api_key: os.getenv(LLM_API_KEY, ), base_url: os.getenv(LLM_BASE_URL, ), model: os.getenv(LLM_MODEL, gpt-4o-mini), } if not config[llm][api_key]: raise ValueError(请先在 .env 中配置 LLM_API_KEY) return config这里把配置集中管理后续调整资讯源、模型参数都不需要改动主流程代码。4.2 数据模型定义为了让工作流中的数据传递更清晰可以定义几个简单的数据类。# src/models.py from dataclasses import dataclass, field from typing import Optional dataclass class NewsItem: 标准化后的资讯条目。 title: str url: str source: str published: str summary: str category: str importance_score: float 0.0 raw_content: str def to_dict(self): return { title: self.title, url: self.url, source: self.source, published: self.published, summary: self.summary, category: self.category, importance_score: self.importance_score, }定义NewsItem数据类以后不同模块之间传递的就是结构化的对象而不是字典或字符串代码可读性会好很多。4.3 资讯采集模块采集模块负责从 RSS 源获取数据。使用feedparser解析 RSS 是常见做法。# src/fetcher.py import time import feedparser from typing import List, Dict, Any from src.models import NewsItem def fetch_rss(url: str, source_name: str, max_items: int 30) - List[NewsItem]: 抓取单个 RSS 源返回标准化条目。 try: feed feedparser.parse(url) except Exception as e: print(f[采集失败] {source_name}: {e}) return [] items [] for entry in feed.entries[:max_items]: title entry.get(title, ).strip() link entry.get(link, ).strip() published entry.get(published, ) if not title or not link: continue item NewsItem( titletitle, urllink, sourcesource_name, publishedpublished, ) items.append(item) return items def fetch_all_sources(sources: List[Dict[str, Any]]) - List[NewsItem]: 串行抓取所有资讯源。 all_items [] for source in sources: source_type source.get(type, rss) url source.get(url, ) name source.get(name, 未知来源) if source_type rss: items fetch_rss(url, name) else: print(f[跳过] 不支持的源类型: {source_type}) items [] print(f[抓取完成] {name}: 获取 {len(items)} 条) all_items.extend(items) time.sleep(0.5) # 避免对目标站点造成压力 return all_items实际生产项目中抓取频率需要遵守目标网站的 robots 协议和服务条款同时做好限速和缓存。4.4 大模型客户端封装模型调用是整套工作流的核心。要考虑三个问题不同业务节点需要不同的 prompt。模型输出可能不稳定需要容错。多次调用模型需要注意 Token 成本。封装src/llm_client.py# src/llm_client.py import json from typing import List, Dict, Any from openai import OpenAI class LLMClient: 通用大模型客户端封装。 def __init__(self, api_key: str, base_url: str, model: str): self.model model self.client OpenAI(api_keyapi_key, base_urlbase_url) def chat(self, messages: List[Dict[str, str]], temperature: float 0.3) - str: 发送对话消息返回文本内容。 try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, ) return response.choices[0].message.content or except Exception as e: print(f[模型调用失败] {e}) return def chat_json(self, messages: List[Dict[str, str]], temperature: float 0.2) - Dict[str, Any]: 发送对话消息并解析 JSON 输出。 messages messages [ { role: system, content: 请只输出 JSON 格式内容不要输出额外解释。, } ] content self.chat(messages, temperaturetemperature) try: return json.loads(content) except json.JSONDecodeError: # 尝试提取第一个 { ... } 片段 start content.find({) end content.rfind(}) if start ! -1 and end ! -1 and end start: try: return json.loads(content[start : end 1]) except json.JSONDecodeError: return {} return {}chat_json方法非常重要。资讯分类、摘要、重要性评估都可以要求模型输出 JSON然后由代码解析。但模型偶尔会输出多余文字因此我们做了两段式提取先整体解析失败后提取花括号片段。4.5 模型参与的文章分类节点接下来实现一个“模型参与分类”的函数。它的输入是一篇NewsItem输出是分类结果。# src/classifier.py from typing import List from src.models import NewsItem from src.llm_client import LLMClient CATEGORY_SYSTEM_PROMPT 你是一名资讯分类专家。 请根据文章标题和内容将文章划分到以下分类之一{categories} 请只输出 JSON格式为{{category: 分类名, reason: 分类理由}} def classify_item(llm: LLMClient, item: NewsItem, categories: List[str]) - str: 调用模型对单篇文章分类。 category_text 、.join(categories) system_prompt CATEGORY_SYSTEM_PROMPT.format(categoriescategory_text) user_content f文章标题{item.title}\n文章来源{item.source} result llm.chat_json( [ {role: system, content: system_prompt}, {role: user, content: user_content}, ] ) category result.get(category, ) if category not in categories: return 未分类 return category这里没有把文章全文带入因为单次分类只需要标题和来源就能完成能有效降低 Token 消耗。如果是更复杂的分类任务可以再把正文摘要加入。4.6 模型参与的摘要与评分节点摘要和重要性评分可以合并成一次模型调用减少调用次数。# src/summary.py from src.models import NewsItem from src.llm_client import LLMClient SUMMARY_SYSTEM_PROMPT 你是一名资深编辑负责为日报撰写内容摘要。 请根据文章内容生成一句话摘要并给出重要性评分1-10分10分最重要。 请只输出 JSON格式为 {{summary: 一句话摘要, importance_score: 8}} def summarize_item(llm: LLMClient, item: NewsItem) - NewsItem: 调用模型生成摘要和评分并更新到 item 上。 user_content f文章标题{item.title}\n文章内容{item.raw_content[:800]} result llm.chat_json( [ {role: system, content: SUMMARY_SYSTEM_PROMPT}, {role: user, content: user_content}, ] ) summary result.get(summary, ) try: score float(result.get(importance_score, 0)) except (TypeError, ValueError): score 0 item.summary summary item.importance_score score return item抓取 RSS 时通常拿不到完整正文所以raw_content可能为空。在实际项目中可以增加一个正文抓取逻辑比如用requests抓取链接后提取p标签文本。4.7 日报组装模块日报输出使用jinja2模板。模板文件可以放在项目templates/目录下。# src/daily_builder.py from pathlib import Path from typing import List from jinja2 import Environment, FileSystemLoader from src.models import NewsItem REPORT_TEMPLATE # 每日资讯日报 生成时间{{ generated_time }} {% for category, items in grouped_items.items() %} ## {{ category }} {% for item in items %} - [{{ item.title }}]({{ item.url }}) - 来源{{ item.source }} - 摘要{{ item.summary }} - 重要度{{ item.importance_score }} {% endfor %} {% endfor %} def build_markdown_report(items: List[NewsItem], generated_time: str) - str: 按分类组装 Markdown 日报。 env Environment() template env.from_string(REPORT_TEMPLATE) grouped_items {} for item in items: category item.category or 未分类 grouped_items.setdefault(category, []).append(item) # 对每个分类内部的条目按重要度排序 for category in grouped_items: grouped_items[category].sort(keylambda x: x.importance_score, reverseTrue) return template.render( generated_timegenerated_time, grouped_itemsgrouped_items, )直接在代码里写模板字符串可以让示例更直观实际项目可以把模板抽成独立文件方便编辑。5. 工作流编排模块5.1 设计工作流状态在复杂工作流中多个节点需要共享数据。最简单可靠的方式是使用一个WorkflowContext对象作为统一的上下文。# src/workflow.py from pathlib import Path from datetime import datetime from typing import List, Dict, Any from src.config import load_config from src.fetcher import fetch_all_sources from src.llm_client import LLMClient from src.classifier import classify_item from src.summary import summarize_item from src.daily_builder import build_markdown_report from src.models import NewsItem class WorkflowContext: 工作流上下文保存中间状态。 def __init__(self): self.config None self.llm None self.all_items: List[NewsItem] [] self.filtered_items: List[NewsItem] [] self.report_content: str 5.2 编排流程工作流编排函数可以按节点拆开这样每一步都可以单独测试。def run_workflow() - Path: 执行资讯日报生成工作流。 ctx WorkflowContext() # 1. 加载配置 ctx.config load_config() # 2. 初始化模型客户端 llm_config ctx.config[llm] ctx.llm LLMClient( api_keyllm_config[api_key], base_urlllm_config[base_url], modelllm_config[model], ) # 3. 抓取资讯 print( 开始抓取资讯 ) ctx.all_items fetch_all_sources(ctx.config[sources]) print(f共抓取到 {len(ctx.all_items)} 条资讯) # 4. 分类和摘要 print( 模型分类与摘要 ) categories ctx.config[categories] for item in ctx.all_items: item.category classify_item(ctx.llm, item, categories) item summarize_item(ctx.llm, item) print(f [{item.category}] {item.title}) # 5. 过滤低质量内容 print( 过滤低质量内容 ) ctx.filtered_items [item for item in ctx.all_items if item.importance_score 5] ctx.filtered_items.sort(keylambda x: x.importance_score, reverseTrue) ctx.filtered_items ctx.filtered_items[: ctx.config.get(max_items, 20)] # 6. 生成 Markdown 日报 print( 生成日报 ) generated_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) ctx.report_content build_markdown_report(ctx.filtered_items, generated_time) # 7. 输出文件 output_path Path(ctx.config[output_path]) output_path.parent.mkdir(parentsTrue, exist_okTrue) output_path.write_text(ctx.report_content, encodingutf-8) print(f日报已生成{output_path}) return output_path这里使用了最简单的串行编排方式。真实生产环境中如果资讯源很多可以对抓取和摘要节点做并发优化例如使用ThreadPoolExecutor。但并发会引入速率限制和 API 超时的问题本文先以串行为主保证逻辑清晰。5.3 入口文件main.py只需要调用run_workflow()# main.py from src.workflow import run_workflow if __name__ __main__: run_workflow()这样设计的好处是后续如果你想增加“发送到钉钉”或“发送到邮件”的能力只需要在run_workflow末尾再追加一个节点不必改动既有节点。6. 运行与验证6.1 运行命令在项目根目录执行python main.py如果一切正常你会看到类似下面的日志 开始抓取资讯 [抓取完成] 开源中国: 获取 25 条 [抓取完成] 36氪: 获取 20 条 [抓取完成] InfoQ 中文: 获取 18 条 共抓取到 63 条资讯 模型分类与摘要 [人工智能] 某某大模型发布新版本 [后端] 微服务架构演进实践 [开源工具] 一款新的命令行工具 ... 过滤低质量内容 生成日报 日报已生成output/daily_report.md6.2 验证输出打开output/daily_report.md应该能看到分类清晰、带摘要和重要度评分的资讯列表。# 每日资讯日报 生成时间2025-01-02 09:30:00 ## 人工智能 - [某某大模型发布新版本](https://example.com/news/1) - 来源36氪 - 摘要该大模型在推理能力上实现大幅提升并开放了 API 接口。 - 重要度9.0到这里一个简单的模型参与资讯日报工作流已经能够运行。7. 常见问题与排查思路在实际使用过程中最常见的问题集中在模型调用、数据抓取和输出格式三方面。问题现象常见原因解决思路模型返回空字符串API Key 错误、网络超时或余额不足检查.env配置单独写脚本测试模型接口连通性模型返回内容不是 JSON模型输出不稳定增强chat_json容错逻辑或改用更低 temperatureRSS 抓取不到数据目标站点屏蔽爬虫或网络不通检查 URL 是否能直接访问尝试添加 User-Agent 请求头分类结果不准确Prompt 分类定义不够清晰增加分类描述和示例或者给模型提供正文摘要Token 消耗过快每篇文章都传入了全文限制raw_content长度或先提取正文关键段落日报内容太多没有设置过滤阈值调整max_items和importance_score阈值这里列一个排查 checklist先测试 API 密钥能否正常调用模型接口。再单独测试一个 RSS 源能否抓取到数据。手动构造一条NewsItem测试分类和摘要节点。确认输出文件路径是否有写入权限。查看日志中每个节点的输出数量是否合理。8. 模型参与复杂工作流的工程建议8.1 模型调用要做降级与缓存资讯日报生成器涉及多轮模型调用如果模型服务不稳定整个工作流可能中断。建议在工程层面做两件事。第一是降级。如果摘要节点调用失败可以直接用文章标题作为摘要而不是中断整个流程。如果分类节点失败可以回退到“未分类”。这样做不能保证结果完美但能保证日报不会被意外打断。第二是缓存。同一天对同一篇文章的摘要和分类结果是稳定不变的。可以在本地维护一个简单缓存键是文章 URL值是模型返回结果。下次再遇到同一文章时直接读缓存不再调用模型。8.2 控制上下文长度与 Token 成本RSS 抓取到的内容可能很长。如果每篇文章都传完整正文Token 消耗会很高。推荐做法是先提取正文前 300 到 500 字。去除 HTML 标签和无意义符号。再传给模型做摘要。对于评分任务甚至可以只传标题和第一段文字效果往往也不错。8.3 Prompt 设计要稳定模型参与的复杂工作流里Prompt 是核心资产。建议把不同的 Prompt 集中管理不要散落在业务代码里。可以单独创建prompts.py文件或者把 Prompt 写入配置目录。每次修改 Prompt 后记录版本变化方便回溯。8.4 工作流节点要可观测复杂工作流排错难度大建议为每个节点增加日志输出。至少包含节点名称。输入数据量。输出数据量。耗时。失败条数。比如fetch_all_sources返回后打印“来源 A 成功 25 条来源 B 失败 0 条”。这样日报内容变少时能快速定位是哪一步过滤掉了数据。8.5 安全与合规边界开发资讯采集类工具时需要注意目标网站的版权与使用条款不要无限频抓取也不要直接复制全文进行商用。如果模型服务需要付费务必设置预算上限避免异常循环导致费用飙升。在生产环境中API Key 应该放在密钥管理系统中而不是以明文形式写进配置仓库。如果使用 Git 管理项目建议提交前检查.env是否已被忽略。8.6 从串行编排走向可扩展架构本文的run_workflow是串行函数适合课程演示和中小规模场景。如果团队已经开始做 Agent 架构下一步可以考虑引入正式的流程引擎把每个节点抽象成独立组件。例如可以定义统一的接口class WorkflowNode: def execute(self, context: WorkflowContext) - WorkflowContext: pass每个节点只关心自己的输入输出由编排器负责组装。这样后续增加“查重节点”“人工审核节点”“多渠道推送节点”都只需要新增一个类不影响其他节点。在这个方向上可以进一步研究 Dify 和 Coze 这类可视化工作流平台它们同样围绕“节点编排 模型调用”的思路只不过把流程放到了图形界面上。理解本文的代码实现后再去看这些工作流平台会更容易理解节点的输入输出和模型连接方式。9. 后续扩展方向资讯日报生成器虽然是一个小项目但它几乎覆盖了 Agent 开发的基础套路任务拆分、模型调用、流程编排、容错降级、结果输出。你可以顺着下面几个方向继续扩展。第一把日报推送到不同渠道。例如封装一个Notifier接口分别实现钉钉机器人通知、飞书机器人通知和邮件通知。工作流只需要多执行一个推送节点。第二引入定时调度。使用cron或APScheduler每天早上 8 点自动运行一次。注意多线程运行时的并发问题避免上一个任务还没结束下一个任务又开始执行。第三个性化订阅。部分用户只关注人工智能分类可以为他们单独生成过滤后的日报。这就是把“用户偏好”引入工作流的一种方式。第四引入人工反馈机制。对模型生成的摘要做点赞/点踩积累数据后可用于 Prompt 调优。第五升级为更完整的 AI Agent。如果不只是做日报而是希望 Agent 能根据用户问题自己决定抓取哪些源、生成什么内容就需要引入 Agent 决策层让模型在“规划节点”选择执行路径。这就是后续课程中 Agent 架构师要解决的更深层问题。最后想说的是模型参与的复杂工作流并不神秘核心在于把“模型能力”和“工程能力”组合起来。模型负责理解、生成、决策工程负责采集、清洗、编排、输出、容错。两者结合才能做出真正稳定可用的产品。希望这篇实战笔记能帮你迈出第一步。如果你在运行过程中遇到问题欢迎留言交流。