企业官网建设流程全解析

每天花一两个小时整理资讯，从 RSS、网站列表、文档源里抓几十篇内容，再人工完成摘要、分类、排版，这个动作长期做下去非常消耗精力。这次我们来看一个能把这套流程压缩成“一条可重跑工作流”的实战项目：资讯日报生成器。它本质上是一个多模型参与的复杂工作流——采集、清洗、摘要、分类、排版、导出，每一站由不同角色完成，模型只做自己最擅长的那一部分。这一课作为 Agent 架构实战的第 5 课，重点不是讲单点模型能力，而是把多个模型任务串成一条可控、可观测、可恢复的工作流。

这篇文章会直接从 Agent 架构角度拆解：日报生成器为什么适合做复杂工作流样例、节点与状态怎么设计、模型怎么分工、怎么把流程封装成 HTTP 接口、怎么支持批量生成和定时任务，以及最常见的排错思路。文中实现不锁定具体平台，使用 OpenAI 兼容接口加 Python 完成，迁移到 Dify、Coze、n8n 或自研 Agent 框架都是同一套思路。如果你正在学 Agent 架构、想做一个自动信息聚合产品，或者每天都需要整理大量资料，这篇文章建议收藏备用。

1. 核心能力速览

从表格可以看出，这套工作流的运行成本不在 GPU 而在 Token。建模时要提前想清楚：哪些节点用便宜模型批量跑，哪些节点用强推理模型精修，这是整个架构设计中最核心的资源决策。

2. 适用场景与使用边界

资讯日报生成器适合三类人：第一类是内容运营，需要每天早上快速浏览多个平台，找出行业关键动态；第二类是产品研发，想在业务系统里内置“每日摘要”能力；第三类是 Agent 学习者，希望用一个小而完整的例子理解多模型协作的工程结构。

它能解决的问题很直接：把“找内容、读内容、总结内容、分类整理”从人工操作变成自动化流程。你只需要维护一个源列表，定时触发，工作流就会把几十篇原始资讯变成一份带摘要、带分类标签、带原文链接的 Markdown 日报。

但也要说清不适合什么场景。第一，它不适合低延迟强交互场景，比如聊天机器人、实时问答，因为工作流有固定步骤，单次完整生成可能需要几十秒甚至更久；第二，它不适合做“一次生成即交付”的内容产品，模型摘要可能出错、可能有幻觉，重要资讯必须保留原文出处和人工复核入口；第三，它不适合抓取需要登录、有反爬限制或明确禁止采集的站点。

合规边界必须反复强调。使用 RSS、公开 API、或站点明确允许转载的内容是安全的；批量抓取网页文本时，要尊重目标站点的 robots 协议和服务条款。模型生成的摘要不要直接当作原创内容商用，引用新闻时保留原文链接。如果后续加入人脸识别、声音处理、私域数据解析等能力，需要额外评估肖像权、隐私和内部数据保密要求。

3. 为什么需要“模型参与的复杂工作流”

很多人第一反应是：把几十篇资讯一次性丢给大模型，让它直接输出日报，不是更简单吗？从演示角度看确实简单，但进入工程化后问题立刻暴露。

第一个问题是上下文长度。十几篇资讯轻松超过几万字，普通模型的上下文窗口不一定装得下，强行塞入会导致中间内容被截断或注意力分散。即便上下文足够，单次调用也容易超过网关超时时间，重试成本非常高。

第二个问题是成本不可控。所有资讯都让最贵的强模型处理，Token 费用会成倍上涨。合理的做法是拆节点：信息抓取用规则和库完成，分类打标用轻量模型，只有最后的摘要和排版才动用强推理模型。

第三个问题是可观测性差。单次大 Prompt 生成的日报，一旦输出质量不对，你无法判断是采集阶段出了问题、摘要阶段出了问题，还是排版阶段出了问题。拆成工作流节点后，每一步都可以单独测试、单独替换模型、单独重试。

这就引出了 Agent 架构里“执行编排层（harness）”的概念。工作流引擎负责维护状态、传递数据、重试失败节点，而每个模型节点只负责“理解输入并产出结果”。资讯日报生成器就是一个很好的编排样例：它不需要复杂的工具调用，也不需要多轮自我反思，但要求节点之间状态清晰、数据格式稳定、失败可恢复。这些能力恰好是 Agent 工程的通用基础。

4. 系统架构与工作流节点设计

整个工作流按数据流方向拆成六个节点。每个节点都是独立函数，通过一个结构化数据对象传递结果。

这个设计的核心是数据协议。每个节点的输出必须是一个明确的 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

各依赖用途说明：

模型侧建议准备一个 OpenAI 兼容接口，可以是云端服务，也可以是本地部署的 vLLM、Ollama 等网关。将接口地址和密钥写入.env文件：

LLM_API_KEY=your_api_key_here LLM_BASE_URL=https://your-llm-endpoint.example.com/v1 LLM_MODEL=gpt-4o-mini LLM_MODEL_CHEAP=gpt-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_key=os.getenv("LLM_API_KEY"), base_url=os.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( model=os.getenv("LLM_MODEL"), response_format={"type": "json_object"}, messages=[ {"role": "system", "content": "你是严谨的信息摘要引擎，只输出 JSON。"}, {"role": "user", "content": prompt}, ], temperature=0.3, timeout=60, ) 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( model=os.getenv("LLM_MODEL_CHEAP"), messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=10, ) 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_ok=True) out_path = os.path.join(output_dir, f"daily_{date.today().isoformat()}.md") with open(out_path, "w", encoding="utf-8") as f: f.write(article) return out_path

这一段是工作流主循环：抓取尽量不中断，清洗失败跳过，摘要失败重试逻辑可根据项目需要补充。这样设计的好处是单源异常不会导致整个日报失败。

7. 服务化接入与批量任务队列

脚本能跑通之后，下一步是把它封装成 HTTP 接口，让其他系统可以调用。这里用 FastAPI 写一个最小服务：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI(title="Daily 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_code=500, detail=str(e))

启动服务：

uvicorn app:app --host 0.0.0.0 --port 8000

接口调用示例，先看 curl：

curl -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, json=payload, timeout=300) 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"), trigger="cron", hour=8, minute=0, ) 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, retries=2): for i in range(retries): try: return summarize_item(item) except Exception as e: print(f"retry {i+1}: {e}") time.sleep(1) return None

8.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. 常见问题与排查方法

这里特别提醒一点：如果用到 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 方式做周报或月度趋势分析，整体架构无需推倒重来。