尧图网站建设 
Agent 架构师 - 第5课实战课资讯日报生成器 - 模型参与的复杂工作流这次我们来看一个非常典型的 Agent 实战项目资讯日报生成器。它的重点不是“再写一个爬虫”而是解决一个更麻烦的问题——当一个工作流里既有信息采集、又有内容清洗、又有分类归纳、又有最终成稿时Agent 架构应该怎么拆模型在哪些环节真正参与决策哪些环节用普通代码处理就够了。对于正在学习 Agent 开发、做架构师转型、或者准备把“工作流 模型调用”落到实际项目里的同学来说这个例子比单纯写个 LLM 接口调用更有价值。因为日报生成器看起来简单但它完整覆盖了 Agent 工作流里最常见的几个难点任务拆解、依赖编排、模型输入输出控制、失败重试和结果校验。这篇文章会从架构设计开始讲然后给一套可落地的代码结构再说明如何用模型在关键环节做决策最后补充批量任务、接口 API 化和常见问题排查。读完后你可以直接照着一套最小复现版本把它改造成自己的资讯聚合工具。1. 核心能力速览先给一个整体的能力画像方便你快速判断这个课程内容适不适合自己。能力项说明项目类型Agent 实战教程 / 复杂工作流案例核心玩法用模型参与资讯采集、清洗、分类、摘要、日报生成关键设计多步骤工作流编排 模型节点 非模型节点模型参与点标签分类、摘要生成、内容去重、日报标题生成工程重点任务依赖、状态管理、重试机制、输出格式校验适用人群Agent 开发者、系统架构师、AI 应用开发工程师运行环境Python、LLM API 或本地模型服务是否支持批量支持可将多来源资讯批量处理成统一结构是否支持 API支持工作流完成后可通过接口对外提供日报服务启动方式脚本启动 / Web 服务启动相关技术栈Agent、工作流编排、模型调用、结构化输出如果已经在用市面上的 Workflow 类产品比如 Dify、Coze、n8n、扣子工作流那么这个项目能帮助你理解这些平台背后“节点编排 模型参与”的本质。如果你是从零开始学 Agent 开发这个案例也比单纯调用一个模型更适合作为“架构设计”的第一课。2. 适用场景与使用边界资讯日报生成器适合谁使用能解决什么问题先说清楚。适合做资讯聚合和内容运营的人每天需要从多个渠道收集新闻、技术博客、行业动态然后整理成一份结构清晰、可读性强的日报。适合 Agent 开发初学者希望通过一个真实案例理解“工作流到底是怎么组织的”而不是只会在一个循环里反复调用 Prompt。适合准备做系统架构师或 AI 应用开发的人这个案例会涉及任务拆解、模块边界、数据流设计和模型调用策略属于非常典型的“AI 应用系统设计”训练。适合技术团队做内部服务化验证把日报生成器封装成 API接入企业微信、钉钉、飞书或者自己的内容平台。需要强调边界。资讯日报生成器涉及抓取第三方网站信息使用前必须确认目标来源的版权政策、robots 协议和平台服务条款。用于内部研究、个人学习、内容聚合时应做内容摘录和改编避免大段原文搬运。如果日报内容要对外发布尤其是商业用途更要做版权复核和来源标注。涉及企业内网数据、用户隐私数据的采集必须遵守相关法律法规和公司安全规范。任何绕过目标网站访问限制、抓取非公开内容、批量爬取个人数据的行为都不在这个项目的合理使用范围内。3. 资讯日报生成器的工作流拆解在设计 Agent 架构时第一步不是写代码而是把“生成一份资讯日报”这个过程拆成多个清晰步骤。日报的生成流程通常包括资讯采集从 RSS、新闻 API、技术社区、公众号热文等渠道获取原始内容。内容清洗去掉 HTML 标签、广告、无关导航、重复段落。条目识别从清洗后的文本里抽取有效资讯条目比如标题、链接、发布时间、来源。标签分类判断每一条资讯属于哪个领域比如 AI、前端、后端、云计算、开源、公司动态。内容摘要对每一条资讯生成简短摘要方便阅读者快速判断是否点开原文。去重合并把多个来源重复报道的资讯合并成一个条目。日报编排按分类、时间或重要程度排序生成最终 Markdown 或 HTML 日报。输出与分发保存到本地文件、发送到 Webhook 或通过 API 提供查询。这个流程的特点是有确定性的数据处理步骤采集、清洗、排序也有需要模型判断的步骤分类、摘要、去重决策。Agent 架构的核心就是把这些步骤组合成一个可控制、可观测、可恢复的工作流。InfoAgent 架构的重点不是让一个模型从头到尾“自由发挥”而是把每个环节的输入输出都定义清楚让模型在合理的边界内做决策。日报生成器就是非常好的教学示例。4. Agent 架构设计与节点划分在开始写代码之前先明确每个节点的职责和输入输出。这是 Agent 架构里面最容易出错的地方。4.1 节点职责表节点名类型输入输出是否用模型SourceCollector普通代码来源配置原始 HTML / RSS 数据否ContentCleaner普通代码原始 HTML / 文本清洗后的正文否EntryExtractor普通代码 / 正则清洗后文本候选资讯列表否可用规则TagClassifier模型节点资讯标题和正文分类标签是Summarizer模型节点资讯正文一句话摘要是DuplicateMerger普通代码 模型辅助候选条目列表去重后的条目部分使用模型ReportComposer普通代码结构化条目Markdown 日报否模板渲染Distributor普通代码日报文本文件 / API / Webhook否为什么有的节点必须用模型有的节点不建议用模型因为模型有延迟、有成本、有不确定性能用规则解决的问题不要交给模型。采集、清洗、排序、格式渲染这些步骤用普通代码更稳定、更可控、更容易调试。TagClassifier 和 Summarizer 是模型中真正有价值的应用场景。4.2 工作流编排模型日报生成器可以看作一个有向无环图执行过程。SourceCollector 执行完成后才能执行 ContentCleanerContentCleaner 完成后再执行 EntryExtractor后级节点可以并行执行。为了代码可维护建议用三种方式之一来组织工作流纯 Python 函数调用顺序执行适合原型验证。使用自研 Task 类 状态枚举适合理解 Agent 状态流转。使用现有工作流框架比如 Dify、Coze、n8n、LangGraph适合生产项目。本文给出一个自定义 Task 类的方案因为它能帮助你理解底层状态管理且不依赖特定工作流平台。5. 环境准备与本地部署前置条件资讯日报生成器的运行环境不复杂核心依赖是 Python、模型 API 或本地模型服务。5.1 基础环境需要准备Python 3.10 或更高版本。一个可调用的 LLM API比如 OpenAI 兼容接口或者本地部署的模型服务。如果本地部署模型需要 CUDA 环境、对应显卡驱动和一个支持 OpenAI 兼容接口的推理服务。用于测试的 RSS 地址或新闻 API本文示例会用模拟数据代替真实网络请求。5.2 安装依赖python -m venv agent_env source agent_env/bin/activate # Windows 下请使用 agent_env\Scripts\activate pip install requests openai beautifulsoup4 markdownify pydantic如果本地已有模型服务只需要保证 base_url 指向本地服务地址即可。比如使用本地 OpenAI 兼容服务时base_url 通常为http://127.0.0.1:8000/v1如果使用在线模型 API则保持默认的 OpenAI 服务地址并在环境变量中配置 API Key。export OPENAI_API_KEY你的APIKey export OPENAI_BASE_URLhttps://api.openai.com/v15.3 模型服务选择建议日报生成器对模型的要求不算高但也不建议用太弱的模型。TagClassifier 需要模型理解资讯内容Summarizer 需要模型进行信息压缩如果模型能力不足分类和摘要都会不稳定。推荐使用具备中等以上逻辑能力的模型。如果本地部署建议观察显存占用和推理延迟。也可以先用一个较小的量化模型做测试再切换到更大模型。6. 安装部署与启动方式下面给出一套最小可复现代码。为了演示 Agent 架构这里不依赖任何重量级框架只使用 Python 标准库和少量第三方库。6.1 定义任务与状态# tasks.py from enum import Enum from dataclasses import dataclass, field from typing import Any, Dict, Optional class TaskStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed dataclass class Task: name: str func: callable depends_on: list field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None error: Optional[str] None config: Dict[str, Any] field(default_factorydict)Task 类的作用是统一节点行为。每个节点都有名字、执行函数、依赖列表、状态、结果和错误信息。在 Agent 架构里这种统一结构非常关键因为工作流引擎只需要看状态和数据不用关心每个节点内部的实现。6.2 工作流执行器# engine.py import time from typing import Dict, List from tasks import Task, TaskStatus class WorkflowEngine: def __init__(self): self.tasks: Dict[str, Task] {} def add_task(self, task: Task): self.tasks[task.name] task def get_ready_tasks(self) - List[str]: ready [] for name, task in self.tasks.items(): if task.status TaskStatus.PENDING: deps_done all( self.tasks[d].status TaskStatus.SUCCESS for d in task.depends_on if d in self.tasks ) if deps_done: ready.append(name) return ready def run(self, max_retries: int 2): while True: ready self.get_ready_tasks() if not ready: break for name in ready: task self.tasks[name] task.status TaskStatus.RUNNING print(f[Engine] 执行节点: {name}) try: task.result task.func(**task.config) task.status TaskStatus.SUCCESS print(f[Engine] 节点成功: {name}) except Exception as e: task.error str(e) task.status TaskStatus.FAILED print(f[Engine] 节点失败: {name}, error{e}) time.sleep(0.1) failed [name for name, t in self.tasks.items() if t.status TaskStatus.FAILED] if failed: print(f[Engine] 存在失败节点: {failed}) else: print([Engine] 工作流全部执行完成)这个执行器非常简单但它已经包含了 Agent 工作流最基本的要素任务注册、依赖判断、状态流转、错误收集。后续要加入重试机制只需要在 except 块里增加循环。6.3 定义收集与清洗节点先定义两个非模型节点。它们使用普通代码处理数据。# nodes_collect.py from bs4 import BeautifulSoup def collect_rss_source(rss_url: str, max_items: int 10): 模拟从 RSS 获取数据真实项目中可替换为 requests 请求。 # 为演示稳定性这里返回模拟数据不执行真实网络请求。 mock_items [ { title: AI Agent 开发实战从零构建日报系统, link: https://example.com/1, published: 2025-01-01T08:00:00Z, content: p本文介绍如何用 Agent 架构构建日报系统。Agent 可以自动采集资讯、分类、摘要并生成最终报告。/p }, { title: 开源工作流引擎的选型指南, link: https://example.com/2, published: 2025-01-01T09:00:00Z, content: p从 n8n 到 Dify再到自研引擎不同团队需要根据自己的业务规模选择合适的工作流方案。本文分析了主流引擎的优缺点。/p }, { title: 大模型在内容行业的生产实践, link: https://example.com/3, published: 2025-01-01T10:00:00Z, content: p内容团队开始使用大模型进行标题生成、摘要提取和辅助写作。实际使用中需要关注输出质量和版权合规。/p } ] return mock_items[:max_items] def clean_html_content(raw_items): 清洗 HTML把正文转换为纯文本。 cleaned [] for item in raw_items: soup BeautifulSoup(item[content], html.parser) plain_text soup.get_text(separator , stripTrue) cleaned.append({ title: item[title], link: item[link], published: item[published], content: plain_text }) return cleaned这里故意使用模拟数据是为了保证起始阶段不依赖网络。真实项目中把 collect_rss_source 替换成 requests 请求 RSS/API 再解析即可。6.4 定义模型节点模型节点的核心是“输入提示词输出结构化结果”。这里使用 OpenAI 兼容接口。# nodes_model.py import json from openai import OpenAI client OpenAI() def classify_news(item: str) - str: prompt f 你是一个资讯分类器。请判断下面这条资讯所属的技术领域。 只输出一个最合适的标签可选范围AI、前端、后端、云计算、开源、架构、编程语言、数据库。 资讯内容 {item} 输出格式只输出标签本身不要输出其他内容。 resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是专业的技术资讯分类助手。}, {role: user, content: prompt} ], temperature0, max_tokens10 ) tag resp.choices[0].message.content.strip() return tag def summarize_news(item: str) - str: prompt f 请用一句话概括下面这条资讯的核心内容不超过50字。 要求保留关键信息不添加原文没有的结论。 资讯内容 {item} 输出格式只输出摘要文本。 resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是专业的技术编辑。}, {role: user, content: prompt} ], temperature0.3, max_tokens200 ) summary resp.choices[0].message.content.strip() return summarynode_model.py 里两个函数的输出都要求是纯文本。这个设计可以避免在后续流程中处理复杂 JSON 结构。对于需要多个字段的模型输出建议采用 JSON 输出并设置 response_format 为 json_object。但这个案例中每个模型节点只需要一个字段所以纯文本输出更简单稳定。6.5 组装工作流在组装阶段把普通节点和模型节点连接起来。# main.py from tasks import Task from engine import WorkflowEngine from nodes_collect import collect_rss_source, clean_html_content from nodes_model import classify_news, summarize_news def extract_entries(cleaned_items): 从清洗结果中直接构造条目真实场景可做更细的抽取。 entries [] for idx, item in enumerate(cleaned_items): entries.append({ id: idx, title: item[title], link: item[link], published: item[published], content: item[content] }) return entries def build_workflow(): engine WorkflowEngine() collect Task( namecollect, funccollect_rss_source, config{rss_url: https://example.com/rss, max_items: 5} ) clean Task( nameclean, funcclean_html_content, depends_on[collect], config{raw_items: None} # 运行时需要从 collect 结果填充 ) # 注意到依赖节点之间的数据传递这里需要修改执行逻辑。 # 我们可以用共享上下文来传递数据。 engine.add_task(collect) engine.add_task(clean) return engine上面的 config 传递方式有个问题clean 节点需要 collect 节点的结果作为输入。简单的 Task 类没有共享上下文机制。这里需要一个上下文对象或者直接在 Task 执行时注入依赖结果。更合理的设计是在 execute 时给函数传入依赖结果。# engine_v2.py import time from typing import Dict, List from tasks import Task, TaskStatus class WorkflowEngineV2: def __init__(self): self.tasks: Dict[str, Task] {} def add_task(self, task: Task): self.tasks[task.name] task def _build_call_kwargs(self, task: Task) - Dict: kwargs dict(task.config) for dep in task.depends_on: if dep in self.tasks and self.tasks[dep].result is not None: kwargs[dep] self.tasks[dep].result return kwargs def get_ready_tasks(self) - List[str]: ready [] for name, task in self.tasks.items(): if task.status TaskStatus.PENDING: deps_done all( self.tasks[d].status TaskStatus.SUCCESS for d in task.depends_on if d in self.tasks ) if deps_done: ready.append(name) return ready def run(self): while True: ready self.get_ready_tasks() if not ready: break for name in ready: task self.tasks[name] task.status TaskStatus.RUNNING try: kwargs self._build_call_kwargs(task) task.result task.func(**kwargs) task.status TaskStatus.SUCCESS except Exception as e: task.error str(e) task.status TaskStatus.FAILED time.sleep(0.1)现在clean 函数可以接收名为 collect 的参数。但是为了代码可读性建议将函数签名中的参数名与上游节点名保持一致。def clean_html_content(collectNone): raw_items collect # 后续清洗逻辑这种“依赖注入”方式虽然简单但在多节点工作流里很实用。只要命名统一数据流就会非常清晰。这也是很多 Agent 框架内部所做的事情。6.6 完整工作流组装示例# main_full.py from tasks import Task from engine_v2 import WorkflowEngineV2 from nodes_collect import collect_rss_source, clean_html_content from nodes_model import classify_news, summarize_news def extract_entries(cleanNone): entries [] for idx, item in enumerate(clean): entries.append({ id: idx, title: item[title], link: item[link], published: item[published], content: item[content] }) return entries def classify_all(entriesNone): result [] for item in entries: tag classify_news(item[title] \n item[content]) item[tag] tag result.append(item) return result def summarize_all(entriesNone): result [] for item in entries: summary summarize_news(item[content]) item[summary] summary result.append(item) return result def compose_report(entriesNone): lines [# 资讯日报\n] for item in entries: lines.append(f## [{item[tag]}] {item[title]}) lines.append(f**来源**: {item[link]}) lines.append(f**时间**: {item[published]}) lines.append(f**摘要**: {item[summary]}) lines.append() return \n.join(lines) def main(): engine WorkflowEngineV2() engine.add_task(Task( namecollect, funccollect_rss_source, config{rss_url: https://example.com/rss, max_items: 5} )) engine.add_task(Task( nameclean, funcclean_html_content, depends_on[collect] )) engine.add_task(Task( nameextract, funcextract_entries, depends_on[clean] )) engine.add_task(Task( nameclassify_all, funcclassify_all, depends_on[extract] )) engine.add_task(Task( namesummarize_all, funcsummarize_all, depends_on[extract] )) engine.add_task(Task( namecompose_report, funccompose_report, depends_on[classify_all, summarize_all] )) engine.run() report_task engine.tasks[compose_report] if report_task.status TaskStatus.SUCCESS: print(日报生成成功内容如下) print(report_task.result) else: print(日报生成失败请查看日志) for name, task in engine.tasks.items(): if task.status TaskStatus.FAILED: print(f失败节点: {name}, 错误: {task.error}) if __name__ __main__: main()这个示例展示了 Agent 工作流的核心模式classify_all 和 summarize_all 都依赖 extract它们之间是并行关系compose_report 需要等待两者完成。这是典型的 DAG 依赖结构。7. 功能测试与效果验证按照下面的维度验证工作流是否正常工作。7.1 基础流程测试先运行一次完整工作流观察每个节点是否按预期顺序执行。python main_full.py预期输出collect 节点成功返回模拟资讯列表。clean 节点成功输出清洗后的纯文本。extract 节点成功输出结构化条目。classify_all 节点成功每条资讯获得一个标签。summarize_all 节点成功每条资讯获得摘要。compose_report 节点成功输出 Markdown 日报。判断标准最终打印的日报包含标题、来源、时间、标签和摘要。7.2 模型节点稳定性测试模型节点最大的风险是输出格式不稳定。可以单独测试 classify_news 和 summarize_news# test_model.py from nodes_model import classify_news, summarize_news sample OpenAI 发布新一代推理模型在数学和代码任务上表现优异。 print(分类结果:, classify_news(sample)) print(摘要结果:, summarize_news(sample))观察点分类结果是否在可选标签范围内。摘要是否超过 50 字。多次运行结果是否稳定。模型返回的内容是否有额外解释文字。如果模型输出不稳定可以在 Prompt 中增加“禁止输出其他内容”的说明并适当调低 temperature。7.3 依赖失败测试模拟一个节点抛异常观察工作流是否会卡住。可以临时把 clean_html_content 改成一个抛出 RuntimeError 的函数然后运行工作流。预期结果依赖 clean 的 extract 节点不会执行因为依赖不满足。引擎打印失败节点信息。工作流最终不会生成日报而会打印失败节点列表。这个测试能帮助你确认工作流的依赖判断和错误处理生效。7.4 超时与重试测试模型 API 调用可能超时。在真实项目中建议为每个模型节点增加超时设置和重试机制。from openai import OpenAI client OpenAI(timeout30.0, max_retries2)如果模型服务不稳定可以在工作流引擎层面增加重试逻辑。# 在节点执行失败时进行重试最大次数 3 def execute_with_retry(task, kwargs, retries3): for attempt in range(retries): try: task.status TaskStatus.RUNNING task.result task.func(**kwargs) task.status TaskStatus.SUCCESS return except Exception as e: task.error str(e) print(f节点 {task.name} 第 {attempt 1} 次执行失败: {e}) time.sleep(2 ** attempt) task.status TaskStatus.FAILED重试策略建议使用指数退避第一次等待 2 秒第二次等待 4 秒第三次等待 8 秒避免把模型服务压垮。8. 接口 API 与批量任务落地日报生成器如果只跑脚本价值有限。实际生产中通常需要暴露 API 或支持定时批量触发。8.1 封装为 Web API可以使用 FastAPI 把工作流包装成接口服务。# api.py from fastapi import FastAPI from pydantic import BaseModel from main_full import build_report app FastAPI() class ReportRequest(BaseModel): rss_url: str max_items: int 5 class ReportResponse(BaseModel): report: str success: bool app.post(/api/report/generate, response_modelReportResponse) def generate_report(req: ReportRequest): try: report build_report(req.rss_url, req.max_items) return ReportResponse(reportreport, successTrue) except Exception as e: return ReportResponse(reportf生成失败: {e}, successFalse) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)在这个设计中build_report 函数需要从 main_full.py 中抽象出来接收 rss_url 和 max_items 参数返回日报字符串。这个封装方式也说明了一个重要架构原则工作流逻辑和展示逻辑分离。Agent 工作流本身不关心是脚本调用还是 API 调用。8.2 批量任务设计当需要处理多个来源时可以增加一个批量任务入口。# batch.py import concurrent.futures def build_report_for_source(rss_url, max_items5): # 实际调用 build_report return rss_url, build_report(rss_url, max_items) def run_batch(rss_sources): with concurrent.futures.ThreadPoolExecutor(max_workers3) as executor: futures [ executor.submit(build_report_for_source, url) for url in rss_sources ] for future in concurrent.futures.as_completed(futures): url, report future.result() print(f完成来源: {url}, 日报长度: {len(report)})批量任务需要考虑以下问题模型 API 的服务限流并发太高会触发 429需要控制 max_workers。结果保存策略每个来源独立保存一个 Markdown 文件或合并成一个汇总日报。失败隔离某个来源生成失败不能影响其他来源需要捕获异常并记录日志。数据来源合法性批量拉取多个来源时更需要遵守目标站点的服务政策。8.3 curl 调用示例curl -X POST http://127.0.0.1:8000/api/report/generate \ -H Content-Type: application/json \ -d {rss_url: https://example.com/rss, max_items: 5}返回结果{ report: # 资讯日报\n\n## [AI] 标题...\n\n, success: true }通过接口调用后续可以接入企业微信机器人、飞书机器人、定时任务调度平台实现每天自动生成日报并推送到指定群聊。9. 资源占用与性能观察日报生成器的主要资源消耗集中在模型调用阶段。普通代码节点运行时间通常是毫秒级模型节点则可能是秒级。模型推理的延迟受以下因素影响模型大小。大模型推理速度慢显存占用高小模型速度快但输出质量可能下降。输入文本长度。资讯正文越长prompt 越长处理时间越长。输出 token 数。摘要节点需要控制 max_tokens避免一次生成过长文本。并发数量。多个节点同时调用模型 API 时受限于 API 限流或本地显存。网络延迟。使用远程 API 时网络状态会影响整体耗时。如果使用本地模型服务建议观察显存占用。在类 Dify、Coze 等平台里这个问题已经抽象成“节点执行时间”用户需要重点观察的则是工作流整体耗时。优化性能的方法先做内容清洗和截断再送入模型降低 prompt 长度。把多条分类请求合并成一次模型调用让模型返回 JSON 数组减少请求次数。对相同来源的资讯做初步规则去重减少模型无用计算。对模型节点设置合理的缓存相同文本不重复调用。在本地部署时使用 vLLM 或类似框架提升推理吞吐。本地模型服务的显存占用受具体模型影响较大无法给出统一数字。建议第一次以最小 batch size、短文本内容测试再逐步增大并发和文本长度观察服务延迟和显存变化。10. 常见问题与排查方法下面是资讯日报生成器在开发和运行阶段最常遇到的问题。问题现象可能原因排查方式解决方案工作流一直不执行下一步依赖节点失败或循环依赖检查节点状态和依赖列表修正依赖关系增加失败后中断模型返回的标签不在可选范围内prompt 约束不够或模型能力不足打印模型返回原始内容增强 prompt加入格式示例降低 temperature摘要过长或包含额外解释max_tokens 设置过大或 prompt 不明确检查模型输出 token 数限制 max_tokens增加“只输出摘要”约束模型 API 调用超时网络问题或模型服务负载高查看服务日志和响应时间增加超时时间和重试机制日报内容重复多个来源报道同一事件检查去重节点是否生效在 summarize 前增加标题相似度去重本地模型显存不足模型太大或并发过高使用 nvidia-smi 查看显存切换小模型降低并发使用量化模型清洗后内容为空HTML 解析失败或正文结构特殊打印清洗后的文本增加备份解析规则或人工标注模板批量任务部分失败API 限流或单个来源超时记录每个任务的异常日志增加失败重试和来源隔离生成日报格式变乱模型输出插入特殊字符检查 compose_report 输入对模型输出做转义或清洗本地 API 端口冲突端口被其他服务占用检查端口监听情况更换端口或杀掉占用进程11. 最佳实践与使用建议资讯日报生成器虽然代码量不大但要做得工程化需要关注以下几点。11.1 第一次先小参数测试不要一开始就抓取几十个来源、生成几千条日报。先用一个来源、三条资讯跑通全流程确认每个节点都能正常输出再逐步增加数据量。这样可以避免问题堆积也方便定位故障节点。11.2 保留一套最小可运行配置将 rss_url、max_items、模型名称等配置放到配置文件或环境变量中。# config.yaml 示例实际解析时需要安装 pyyaml sources: - url: https://example.com/rss max_items: 5 model: api_key_env: OPENAI_API_KEY base_url: https://api.openai.com/v1 name: gpt-4o-mini timeout: 30 max_retries: 2 workflow: output_dir: ./outputs save_markdown: true save_json: true11.3 模型文件、输入素材、输出结果分目录管理即使没有模型文件也需要把输出目录与代码目录分开。project/ ├── nodes_collect.py ├── nodes_model.py ├── engine_v2.py ├── tasks.py ├── main_full.py ├── api.py ├── config/ │ └── config.yaml └── outputs/ ├── reports/ └── logs/这样生产环境出问题时日志和产物不会污染代码目录。11.4 批量任务要加日志和失败重试每个任务都要记录开始时间、结束时间、状态、错误原因。建议把任务日志写入 JSONL 文件方便后续分析。失败重试建议使用指数退避避免频繁重试导致 API 被限流。11.5 接口服务要限制访问范围如果日报服务部署在公网建议增加 API 密钥或 IP 白名单。避免内部工作流接口被外部随意调用产生不必要的费用和安全风险。11.6 涉及人脸、声音、版权素材时必须确认授权虽然日报生成器主要是文本但如果在资讯中处理图片、视频内容或者后续扩展为“多模态日报”使用任何他人的肖像、声音、版权素材时必须确认授权。包括人脸图片的采集、公司品牌素材的使用、第三方文章的全文搬运都在合规风险范围内。11.7 发布或商用前要做效果复核模型生成的摘要和分类不代表一定正确。对外发布前建议安排人工抽检尤其是涉及金融、医疗、法律、政治等敏感领域的内容模型输出可能包含错误或偏见需要人工把关。12. 总结与下一步这个实战课的核心价值不是教你调一个 API而是让你看到模型参与工作流的正确姿势。资讯日报生成器把采集、清洗、分类、摘要、编排、分发这套流程完整地串了一遍它有清晰的节点边界、可控的依赖关系、明确的模型参与点和可扩展的 API 接口。如果你要动手跑一遍最先验证的应该是“工作流是否能按依赖顺序完整执行”。把模拟数据跑通后再去接入真实 RSS 源和模型 API观察分类和摘要的质量然后逐步增加内容去重、批量任务和定时调度。最容易踩的坑有三个一是没有在节点层做错误收集导致工作流卡死二是模型输出格式不稳定但代码没有做校验三是直接抓取大量来源忽略版权和访问政策。前两个是技术问题第三个是合规问题都需要认真对待。下一步可以扩展的方向包括接入更多资讯来源、增加自定义分类规则、把日报发送到企业微信群、把模型分类结果保存到数据库做趋势分析或者直接用自研 Agent 框架替代当前的简易引擎。这些方向都能从本课的架构基础上继续延伸建议收藏备用。最后说一句Agent 不是让模型自由发挥而是把复杂的任务拆成可控的节点让模型只做它擅长的事情。资讯日报生成器把这个原则体现得非常清楚。 Agent架构实战：资讯日报生成器中的模型决策工作流
Agent 架构师 - 第5课实战课资讯日报生成器 - 模型参与的复杂工作流这次我们来看一个非常典型的 Agent 实战项目资讯日报生成器。它的重点不是“再写一个爬虫”而是解决一个更麻烦的问题——当一个工作流里既有信息采集、又有内容清洗、又有分类归纳、又有最终成稿时Agent 架构应该怎么拆模型在哪些环节真正参与决策哪些环节用普通代码处理就够了。对于正在学习 Agent 开发、做架构师转型、或者准备把“工作流 模型调用”落到实际项目里的同学来说这个例子比单纯写个 LLM 接口调用更有价值。因为日报生成器看起来简单但它完整覆盖了 Agent 工作流里最常见的几个难点任务拆解、依赖编排、模型输入输出控制、失败重试和结果校验。这篇文章会从架构设计开始讲然后给一套可落地的代码结构再说明如何用模型在关键环节做决策最后补充批量任务、接口 API 化和常见问题排查。读完后你可以直接照着一套最小复现版本把它改造成自己的资讯聚合工具。1. 核心能力速览先给一个整体的能力画像方便你快速判断这个课程内容适不适合自己。能力项说明项目类型Agent 实战教程 / 复杂工作流案例核心玩法用模型参与资讯采集、清洗、分类、摘要、日报生成关键设计多步骤工作流编排 模型节点 非模型节点模型参与点标签分类、摘要生成、内容去重、日报标题生成工程重点任务依赖、状态管理、重试机制、输出格式校验适用人群Agent 开发者、系统架构师、AI 应用开发工程师运行环境Python、LLM API 或本地模型服务是否支持批量支持可将多来源资讯批量处理成统一结构是否支持 API支持工作流完成后可通过接口对外提供日报服务启动方式脚本启动 / Web 服务启动相关技术栈Agent、工作流编排、模型调用、结构化输出如果已经在用市面上的 Workflow 类产品比如 Dify、Coze、n8n、扣子工作流那么这个项目能帮助你理解这些平台背后“节点编排 模型参与”的本质。如果你是从零开始学 Agent 开发这个案例也比单纯调用一个模型更适合作为“架构设计”的第一课。2. 适用场景与使用边界资讯日报生成器适合谁使用能解决什么问题先说清楚。适合做资讯聚合和内容运营的人每天需要从多个渠道收集新闻、技术博客、行业动态然后整理成一份结构清晰、可读性强的日报。适合 Agent 开发初学者希望通过一个真实案例理解“工作流到底是怎么组织的”而不是只会在一个循环里反复调用 Prompt。适合准备做系统架构师或 AI 应用开发的人这个案例会涉及任务拆解、模块边界、数据流设计和模型调用策略属于非常典型的“AI 应用系统设计”训练。适合技术团队做内部服务化验证把日报生成器封装成 API接入企业微信、钉钉、飞书或者自己的内容平台。需要强调边界。资讯日报生成器涉及抓取第三方网站信息使用前必须确认目标来源的版权政策、robots 协议和平台服务条款。用于内部研究、个人学习、内容聚合时应做内容摘录和改编避免大段原文搬运。如果日报内容要对外发布尤其是商业用途更要做版权复核和来源标注。涉及企业内网数据、用户隐私数据的采集必须遵守相关法律法规和公司安全规范。任何绕过目标网站访问限制、抓取非公开内容、批量爬取个人数据的行为都不在这个项目的合理使用范围内。3. 资讯日报生成器的工作流拆解在设计 Agent 架构时第一步不是写代码而是把“生成一份资讯日报”这个过程拆成多个清晰步骤。日报的生成流程通常包括资讯采集从 RSS、新闻 API、技术社区、公众号热文等渠道获取原始内容。内容清洗去掉 HTML 标签、广告、无关导航、重复段落。条目识别从清洗后的文本里抽取有效资讯条目比如标题、链接、发布时间、来源。标签分类判断每一条资讯属于哪个领域比如 AI、前端、后端、云计算、开源、公司动态。内容摘要对每一条资讯生成简短摘要方便阅读者快速判断是否点开原文。去重合并把多个来源重复报道的资讯合并成一个条目。日报编排按分类、时间或重要程度排序生成最终 Markdown 或 HTML 日报。输出与分发保存到本地文件、发送到 Webhook 或通过 API 提供查询。这个流程的特点是有确定性的数据处理步骤采集、清洗、排序也有需要模型判断的步骤分类、摘要、去重决策。Agent 架构的核心就是把这些步骤组合成一个可控制、可观测、可恢复的工作流。InfoAgent 架构的重点不是让一个模型从头到尾“自由发挥”而是把每个环节的输入输出都定义清楚让模型在合理的边界内做决策。日报生成器就是非常好的教学示例。4. Agent 架构设计与节点划分在开始写代码之前先明确每个节点的职责和输入输出。这是 Agent 架构里面最容易出错的地方。4.1 节点职责表节点名类型输入输出是否用模型SourceCollector普通代码来源配置原始 HTML / RSS 数据否ContentCleaner普通代码原始 HTML / 文本清洗后的正文否EntryExtractor普通代码 / 正则清洗后文本候选资讯列表否可用规则TagClassifier模型节点资讯标题和正文分类标签是Summarizer模型节点资讯正文一句话摘要是DuplicateMerger普通代码 模型辅助候选条目列表去重后的条目部分使用模型ReportComposer普通代码结构化条目Markdown 日报否模板渲染Distributor普通代码日报文本文件 / API / Webhook否为什么有的节点必须用模型有的节点不建议用模型因为模型有延迟、有成本、有不确定性能用规则解决的问题不要交给模型。采集、清洗、排序、格式渲染这些步骤用普通代码更稳定、更可控、更容易调试。TagClassifier 和 Summarizer 是模型中真正有价值的应用场景。4.2 工作流编排模型日报生成器可以看作一个有向无环图执行过程。SourceCollector 执行完成后才能执行 ContentCleanerContentCleaner 完成后再执行 EntryExtractor后级节点可以并行执行。为了代码可维护建议用三种方式之一来组织工作流纯 Python 函数调用顺序执行适合原型验证。使用自研 Task 类 状态枚举适合理解 Agent 状态流转。使用现有工作流框架比如 Dify、Coze、n8n、LangGraph适合生产项目。本文给出一个自定义 Task 类的方案因为它能帮助你理解底层状态管理且不依赖特定工作流平台。5. 环境准备与本地部署前置条件资讯日报生成器的运行环境不复杂核心依赖是 Python、模型 API 或本地模型服务。5.1 基础环境需要准备Python 3.10 或更高版本。一个可调用的 LLM API比如 OpenAI 兼容接口或者本地部署的模型服务。如果本地部署模型需要 CUDA 环境、对应显卡驱动和一个支持 OpenAI 兼容接口的推理服务。用于测试的 RSS 地址或新闻 API本文示例会用模拟数据代替真实网络请求。5.2 安装依赖python -m venv agent_env source agent_env/bin/activate # Windows 下请使用 agent_env\Scripts\activate pip install requests openai beautifulsoup4 markdownify pydantic如果本地已有模型服务只需要保证 base_url 指向本地服务地址即可。比如使用本地 OpenAI 兼容服务时base_url 通常为http://127.0.0.1:8000/v1如果使用在线模型 API则保持默认的 OpenAI 服务地址并在环境变量中配置 API Key。export OPENAI_API_KEY你的APIKey export OPENAI_BASE_URLhttps://api.openai.com/v15.3 模型服务选择建议日报生成器对模型的要求不算高但也不建议用太弱的模型。TagClassifier 需要模型理解资讯内容Summarizer 需要模型进行信息压缩如果模型能力不足分类和摘要都会不稳定。推荐使用具备中等以上逻辑能力的模型。如果本地部署建议观察显存占用和推理延迟。也可以先用一个较小的量化模型做测试再切换到更大模型。6. 安装部署与启动方式下面给出一套最小可复现代码。为了演示 Agent 架构这里不依赖任何重量级框架只使用 Python 标准库和少量第三方库。6.1 定义任务与状态# tasks.py from enum import Enum from dataclasses import dataclass, field from typing import Any, Dict, Optional class TaskStatus(str, Enum): PENDING pending RUNNING running SUCCESS success FAILED failed dataclass class Task: name: str func: callable depends_on: list field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None error: Optional[str] None config: Dict[str, Any] field(default_factorydict)Task 类的作用是统一节点行为。每个节点都有名字、执行函数、依赖列表、状态、结果和错误信息。在 Agent 架构里这种统一结构非常关键因为工作流引擎只需要看状态和数据不用关心每个节点内部的实现。6.2 工作流执行器# engine.py import time from typing import Dict, List from tasks import Task, TaskStatus class WorkflowEngine: def __init__(self): self.tasks: Dict[str, Task] {} def add_task(self, task: Task): self.tasks[task.name] task def get_ready_tasks(self) - List[str]: ready [] for name, task in self.tasks.items(): if task.status TaskStatus.PENDING: deps_done all( self.tasks[d].status TaskStatus.SUCCESS for d in task.depends_on if d in self.tasks ) if deps_done: ready.append(name) return ready def run(self, max_retries: int 2): while True: ready self.get_ready_tasks() if not ready: break for name in ready: task self.tasks[name] task.status TaskStatus.RUNNING print(f[Engine] 执行节点: {name}) try: task.result task.func(**task.config) task.status TaskStatus.SUCCESS print(f[Engine] 节点成功: {name}) except Exception as e: task.error str(e) task.status TaskStatus.FAILED print(f[Engine] 节点失败: {name}, error{e}) time.sleep(0.1) failed [name for name, t in self.tasks.items() if t.status TaskStatus.FAILED] if failed: print(f[Engine] 存在失败节点: {failed}) else: print([Engine] 工作流全部执行完成)这个执行器非常简单但它已经包含了 Agent 工作流最基本的要素任务注册、依赖判断、状态流转、错误收集。后续要加入重试机制只需要在 except 块里增加循环。6.3 定义收集与清洗节点先定义两个非模型节点。它们使用普通代码处理数据。# nodes_collect.py from bs4 import BeautifulSoup def collect_rss_source(rss_url: str, max_items: int 10): 模拟从 RSS 获取数据真实项目中可替换为 requests 请求。 # 为演示稳定性这里返回模拟数据不执行真实网络请求。 mock_items [ { title: AI Agent 开发实战从零构建日报系统, link: https://example.com/1, published: 2025-01-01T08:00:00Z, content: p本文介绍如何用 Agent 架构构建日报系统。Agent 可以自动采集资讯、分类、摘要并生成最终报告。/p }, { title: 开源工作流引擎的选型指南, link: https://example.com/2, published: 2025-01-01T09:00:00Z, content: p从 n8n 到 Dify再到自研引擎不同团队需要根据自己的业务规模选择合适的工作流方案。本文分析了主流引擎的优缺点。/p }, { title: 大模型在内容行业的生产实践, link: https://example.com/3, published: 2025-01-01T10:00:00Z, content: p内容团队开始使用大模型进行标题生成、摘要提取和辅助写作。实际使用中需要关注输出质量和版权合规。/p } ] return mock_items[:max_items] def clean_html_content(raw_items): 清洗 HTML把正文转换为纯文本。 cleaned [] for item in raw_items: soup BeautifulSoup(item[content], html.parser) plain_text soup.get_text(separator , stripTrue) cleaned.append({ title: item[title], link: item[link], published: item[published], content: plain_text }) return cleaned这里故意使用模拟数据是为了保证起始阶段不依赖网络。真实项目中把 collect_rss_source 替换成 requests 请求 RSS/API 再解析即可。6.4 定义模型节点模型节点的核心是“输入提示词输出结构化结果”。这里使用 OpenAI 兼容接口。# nodes_model.py import json from openai import OpenAI client OpenAI() def classify_news(item: str) - str: prompt f 你是一个资讯分类器。请判断下面这条资讯所属的技术领域。 只输出一个最合适的标签可选范围AI、前端、后端、云计算、开源、架构、编程语言、数据库。 资讯内容 {item} 输出格式只输出标签本身不要输出其他内容。 resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是专业的技术资讯分类助手。}, {role: user, content: prompt} ], temperature0, max_tokens10 ) tag resp.choices[0].message.content.strip() return tag def summarize_news(item: str) - str: prompt f 请用一句话概括下面这条资讯的核心内容不超过50字。 要求保留关键信息不添加原文没有的结论。 资讯内容 {item} 输出格式只输出摘要文本。 resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是专业的技术编辑。}, {role: user, content: prompt} ], temperature0.3, max_tokens200 ) summary resp.choices[0].message.content.strip() return summarynode_model.py 里两个函数的输出都要求是纯文本。这个设计可以避免在后续流程中处理复杂 JSON 结构。对于需要多个字段的模型输出建议采用 JSON 输出并设置 response_format 为 json_object。但这个案例中每个模型节点只需要一个字段所以纯文本输出更简单稳定。6.5 组装工作流在组装阶段把普通节点和模型节点连接起来。# main.py from tasks import Task from engine import WorkflowEngine from nodes_collect import collect_rss_source, clean_html_content from nodes_model import classify_news, summarize_news def extract_entries(cleaned_items): 从清洗结果中直接构造条目真实场景可做更细的抽取。 entries [] for idx, item in enumerate(cleaned_items): entries.append({ id: idx, title: item[title], link: item[link], published: item[published], content: item[content] }) return entries def build_workflow(): engine WorkflowEngine() collect Task( namecollect, funccollect_rss_source, config{rss_url: https://example.com/rss, max_items: 5} ) clean Task( nameclean, funcclean_html_content, depends_on[collect], config{raw_items: None} # 运行时需要从 collect 结果填充 ) # 注意到依赖节点之间的数据传递这里需要修改执行逻辑。 # 我们可以用共享上下文来传递数据。 engine.add_task(collect) engine.add_task(clean) return engine上面的 config 传递方式有个问题clean 节点需要 collect 节点的结果作为输入。简单的 Task 类没有共享上下文机制。这里需要一个上下文对象或者直接在 Task 执行时注入依赖结果。更合理的设计是在 execute 时给函数传入依赖结果。# engine_v2.py import time from typing import Dict, List from tasks import Task, TaskStatus class WorkflowEngineV2: def __init__(self): self.tasks: Dict[str, Task] {} def add_task(self, task: Task): self.tasks[task.name] task def _build_call_kwargs(self, task: Task) - Dict: kwargs dict(task.config) for dep in task.depends_on: if dep in self.tasks and self.tasks[dep].result is not None: kwargs[dep] self.tasks[dep].result return kwargs def get_ready_tasks(self) - List[str]: ready [] for name, task in self.tasks.items(): if task.status TaskStatus.PENDING: deps_done all( self.tasks[d].status TaskStatus.SUCCESS for d in task.depends_on if d in self.tasks ) if deps_done: ready.append(name) return ready def run(self): while True: ready self.get_ready_tasks() if not ready: break for name in ready: task self.tasks[name] task.status TaskStatus.RUNNING try: kwargs self._build_call_kwargs(task) task.result task.func(**kwargs) task.status TaskStatus.SUCCESS except Exception as e: task.error str(e) task.status TaskStatus.FAILED time.sleep(0.1)现在clean 函数可以接收名为 collect 的参数。但是为了代码可读性建议将函数签名中的参数名与上游节点名保持一致。def clean_html_content(collectNone): raw_items collect # 后续清洗逻辑这种“依赖注入”方式虽然简单但在多节点工作流里很实用。只要命名统一数据流就会非常清晰。这也是很多 Agent 框架内部所做的事情。6.6 完整工作流组装示例# main_full.py from tasks import Task from engine_v2 import WorkflowEngineV2 from nodes_collect import collect_rss_source, clean_html_content from nodes_model import classify_news, summarize_news def extract_entries(cleanNone): entries [] for idx, item in enumerate(clean): entries.append({ id: idx, title: item[title], link: item[link], published: item[published], content: item[content] }) return entries def classify_all(entriesNone): result [] for item in entries: tag classify_news(item[title] \n item[content]) item[tag] tag result.append(item) return result def summarize_all(entriesNone): result [] for item in entries: summary summarize_news(item[content]) item[summary] summary result.append(item) return result def compose_report(entriesNone): lines [# 资讯日报\n] for item in entries: lines.append(f## [{item[tag]}] {item[title]}) lines.append(f**来源**: {item[link]}) lines.append(f**时间**: {item[published]}) lines.append(f**摘要**: {item[summary]}) lines.append() return \n.join(lines) def main(): engine WorkflowEngineV2() engine.add_task(Task( namecollect, funccollect_rss_source, config{rss_url: https://example.com/rss, max_items: 5} )) engine.add_task(Task( nameclean, funcclean_html_content, depends_on[collect] )) engine.add_task(Task( nameextract, funcextract_entries, depends_on[clean] )) engine.add_task(Task( nameclassify_all, funcclassify_all, depends_on[extract] )) engine.add_task(Task( namesummarize_all, funcsummarize_all, depends_on[extract] )) engine.add_task(Task( namecompose_report, funccompose_report, depends_on[classify_all, summarize_all] )) engine.run() report_task engine.tasks[compose_report] if report_task.status TaskStatus.SUCCESS: print(日报生成成功内容如下) print(report_task.result) else: print(日报生成失败请查看日志) for name, task in engine.tasks.items(): if task.status TaskStatus.FAILED: print(f失败节点: {name}, 错误: {task.error}) if __name__ __main__: main()这个示例展示了 Agent 工作流的核心模式classify_all 和 summarize_all 都依赖 extract它们之间是并行关系compose_report 需要等待两者完成。这是典型的 DAG 依赖结构。7. 功能测试与效果验证按照下面的维度验证工作流是否正常工作。7.1 基础流程测试先运行一次完整工作流观察每个节点是否按预期顺序执行。python main_full.py预期输出collect 节点成功返回模拟资讯列表。clean 节点成功输出清洗后的纯文本。extract 节点成功输出结构化条目。classify_all 节点成功每条资讯获得一个标签。summarize_all 节点成功每条资讯获得摘要。compose_report 节点成功输出 Markdown 日报。判断标准最终打印的日报包含标题、来源、时间、标签和摘要。7.2 模型节点稳定性测试模型节点最大的风险是输出格式不稳定。可以单独测试 classify_news 和 summarize_news# test_model.py from nodes_model import classify_news, summarize_news sample OpenAI 发布新一代推理模型在数学和代码任务上表现优异。 print(分类结果:, classify_news(sample)) print(摘要结果:, summarize_news(sample))观察点分类结果是否在可选标签范围内。摘要是否超过 50 字。多次运行结果是否稳定。模型返回的内容是否有额外解释文字。如果模型输出不稳定可以在 Prompt 中增加“禁止输出其他内容”的说明并适当调低 temperature。7.3 依赖失败测试模拟一个节点抛异常观察工作流是否会卡住。可以临时把 clean_html_content 改成一个抛出 RuntimeError 的函数然后运行工作流。预期结果依赖 clean 的 extract 节点不会执行因为依赖不满足。引擎打印失败节点信息。工作流最终不会生成日报而会打印失败节点列表。这个测试能帮助你确认工作流的依赖判断和错误处理生效。7.4 超时与重试测试模型 API 调用可能超时。在真实项目中建议为每个模型节点增加超时设置和重试机制。from openai import OpenAI client OpenAI(timeout30.0, max_retries2)如果模型服务不稳定可以在工作流引擎层面增加重试逻辑。# 在节点执行失败时进行重试最大次数 3 def execute_with_retry(task, kwargs, retries3): for attempt in range(retries): try: task.status TaskStatus.RUNNING task.result task.func(**kwargs) task.status TaskStatus.SUCCESS return except Exception as e: task.error str(e) print(f节点 {task.name} 第 {attempt 1} 次执行失败: {e}) time.sleep(2 ** attempt) task.status TaskStatus.FAILED重试策略建议使用指数退避第一次等待 2 秒第二次等待 4 秒第三次等待 8 秒避免把模型服务压垮。8. 接口 API 与批量任务落地日报生成器如果只跑脚本价值有限。实际生产中通常需要暴露 API 或支持定时批量触发。8.1 封装为 Web API可以使用 FastAPI 把工作流包装成接口服务。# api.py from fastapi import FastAPI from pydantic import BaseModel from main_full import build_report app FastAPI() class ReportRequest(BaseModel): rss_url: str max_items: int 5 class ReportResponse(BaseModel): report: str success: bool app.post(/api/report/generate, response_modelReportResponse) def generate_report(req: ReportRequest): try: report build_report(req.rss_url, req.max_items) return ReportResponse(reportreport, successTrue) except Exception as e: return ReportResponse(reportf生成失败: {e}, successFalse) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)在这个设计中build_report 函数需要从 main_full.py 中抽象出来接收 rss_url 和 max_items 参数返回日报字符串。这个封装方式也说明了一个重要架构原则工作流逻辑和展示逻辑分离。Agent 工作流本身不关心是脚本调用还是 API 调用。8.2 批量任务设计当需要处理多个来源时可以增加一个批量任务入口。# batch.py import concurrent.futures def build_report_for_source(rss_url, max_items5): # 实际调用 build_report return rss_url, build_report(rss_url, max_items) def run_batch(rss_sources): with concurrent.futures.ThreadPoolExecutor(max_workers3) as executor: futures [ executor.submit(build_report_for_source, url) for url in rss_sources ] for future in concurrent.futures.as_completed(futures): url, report future.result() print(f完成来源: {url}, 日报长度: {len(report)})批量任务需要考虑以下问题模型 API 的服务限流并发太高会触发 429需要控制 max_workers。结果保存策略每个来源独立保存一个 Markdown 文件或合并成一个汇总日报。失败隔离某个来源生成失败不能影响其他来源需要捕获异常并记录日志。数据来源合法性批量拉取多个来源时更需要遵守目标站点的服务政策。8.3 curl 调用示例curl -X POST http://127.0.0.1:8000/api/report/generate \ -H Content-Type: application/json \ -d {rss_url: https://example.com/rss, max_items: 5}返回结果{ report: # 资讯日报\n\n## [AI] 标题...\n\n, success: true }通过接口调用后续可以接入企业微信机器人、飞书机器人、定时任务调度平台实现每天自动生成日报并推送到指定群聊。9. 资源占用与性能观察日报生成器的主要资源消耗集中在模型调用阶段。普通代码节点运行时间通常是毫秒级模型节点则可能是秒级。模型推理的延迟受以下因素影响模型大小。大模型推理速度慢显存占用高小模型速度快但输出质量可能下降。输入文本长度。资讯正文越长prompt 越长处理时间越长。输出 token 数。摘要节点需要控制 max_tokens避免一次生成过长文本。并发数量。多个节点同时调用模型 API 时受限于 API 限流或本地显存。网络延迟。使用远程 API 时网络状态会影响整体耗时。如果使用本地模型服务建议观察显存占用。在类 Dify、Coze 等平台里这个问题已经抽象成“节点执行时间”用户需要重点观察的则是工作流整体耗时。优化性能的方法先做内容清洗和截断再送入模型降低 prompt 长度。把多条分类请求合并成一次模型调用让模型返回 JSON 数组减少请求次数。对相同来源的资讯做初步规则去重减少模型无用计算。对模型节点设置合理的缓存相同文本不重复调用。在本地部署时使用 vLLM 或类似框架提升推理吞吐。本地模型服务的显存占用受具体模型影响较大无法给出统一数字。建议第一次以最小 batch size、短文本内容测试再逐步增大并发和文本长度观察服务延迟和显存变化。10. 常见问题与排查方法下面是资讯日报生成器在开发和运行阶段最常遇到的问题。问题现象可能原因排查方式解决方案工作流一直不执行下一步依赖节点失败或循环依赖检查节点状态和依赖列表修正依赖关系增加失败后中断模型返回的标签不在可选范围内prompt 约束不够或模型能力不足打印模型返回原始内容增强 prompt加入格式示例降低 temperature摘要过长或包含额外解释max_tokens 设置过大或 prompt 不明确检查模型输出 token 数限制 max_tokens增加“只输出摘要”约束模型 API 调用超时网络问题或模型服务负载高查看服务日志和响应时间增加超时时间和重试机制日报内容重复多个来源报道同一事件检查去重节点是否生效在 summarize 前增加标题相似度去重本地模型显存不足模型太大或并发过高使用 nvidia-smi 查看显存切换小模型降低并发使用量化模型清洗后内容为空HTML 解析失败或正文结构特殊打印清洗后的文本增加备份解析规则或人工标注模板批量任务部分失败API 限流或单个来源超时记录每个任务的异常日志增加失败重试和来源隔离生成日报格式变乱模型输出插入特殊字符检查 compose_report 输入对模型输出做转义或清洗本地 API 端口冲突端口被其他服务占用检查端口监听情况更换端口或杀掉占用进程11. 最佳实践与使用建议资讯日报生成器虽然代码量不大但要做得工程化需要关注以下几点。11.1 第一次先小参数测试不要一开始就抓取几十个来源、生成几千条日报。先用一个来源、三条资讯跑通全流程确认每个节点都能正常输出再逐步增加数据量。这样可以避免问题堆积也方便定位故障节点。11.2 保留一套最小可运行配置将 rss_url、max_items、模型名称等配置放到配置文件或环境变量中。# config.yaml 示例实际解析时需要安装 pyyaml sources: - url: https://example.com/rss max_items: 5 model: api_key_env: OPENAI_API_KEY base_url: https://api.openai.com/v1 name: gpt-4o-mini timeout: 30 max_retries: 2 workflow: output_dir: ./outputs save_markdown: true save_json: true11.3 模型文件、输入素材、输出结果分目录管理即使没有模型文件也需要把输出目录与代码目录分开。project/ ├── nodes_collect.py ├── nodes_model.py ├── engine_v2.py ├── tasks.py ├── main_full.py ├── api.py ├── config/ │ └── config.yaml └── outputs/ ├── reports/ └── logs/这样生产环境出问题时日志和产物不会污染代码目录。11.4 批量任务要加日志和失败重试每个任务都要记录开始时间、结束时间、状态、错误原因。建议把任务日志写入 JSONL 文件方便后续分析。失败重试建议使用指数退避避免频繁重试导致 API 被限流。11.5 接口服务要限制访问范围如果日报服务部署在公网建议增加 API 密钥或 IP 白名单。避免内部工作流接口被外部随意调用产生不必要的费用和安全风险。11.6 涉及人脸、声音、版权素材时必须确认授权虽然日报生成器主要是文本但如果在资讯中处理图片、视频内容或者后续扩展为“多模态日报”使用任何他人的肖像、声音、版权素材时必须确认授权。包括人脸图片的采集、公司品牌素材的使用、第三方文章的全文搬运都在合规风险范围内。11.7 发布或商用前要做效果复核模型生成的摘要和分类不代表一定正确。对外发布前建议安排人工抽检尤其是涉及金融、医疗、法律、政治等敏感领域的内容模型输出可能包含错误或偏见需要人工把关。12. 总结与下一步这个实战课的核心价值不是教你调一个 API而是让你看到模型参与工作流的正确姿势。资讯日报生成器把采集、清洗、分类、摘要、编排、分发这套流程完整地串了一遍它有清晰的节点边界、可控的依赖关系、明确的模型参与点和可扩展的 API 接口。如果你要动手跑一遍最先验证的应该是“工作流是否能按依赖顺序完整执行”。把模拟数据跑通后再去接入真实 RSS 源和模型 API观察分类和摘要的质量然后逐步增加内容去重、批量任务和定时调度。最容易踩的坑有三个一是没有在节点层做错误收集导致工作流卡死二是模型输出格式不稳定但代码没有做校验三是直接抓取大量来源忽略版权和访问政策。前两个是技术问题第三个是合规问题都需要认真对待。下一步可以扩展的方向包括接入更多资讯来源、增加自定义分类规则、把日报发送到企业微信群、把模型分类结果保存到数据库做趋势分析或者直接用自研 Agent 框架替代当前的简易引擎。这些方向都能从本课的架构基础上继续延伸建议收藏备用。最后说一句Agent 不是让模型自由发挥而是把复杂的任务拆成可控的节点让模型只做它擅长的事情。资讯日报生成器把这个原则体现得非常清楚。