基于AI Agent工作流构建自动化行业趋势报告生成器

📅 2026/7/4 23:10:32
基于AI Agent工作流构建自动化行业趋势报告生成器
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度在实际 AI 项目开发和技术趋势跟踪中开发者经常面临一个矛盾一方面需要快速获取并理解最新的行业动态、开源项目和技术突破另一方面信息源分散、访问不稳定、内容重复等问题又严重消耗了研发效率。手动整理日报、周报或专题报告从信息采集、去重、验证到格式化输出是一个重复且繁琐的过程。这正是 AI Agent 和工作流技术可以大显身手的场景——将固定的信息处理流程自动化、智能化。本文将以一个具体的开源项目ai-trend-radar-report为例深入探讨如何构建一个面向中文 AI 行业监测的 Agent Skill。这不是一个简单的爬虫脚本而是一个完整的工作流它能够被集成到支持 Agent Skills 的工具如 Claude Code中自动生成结构清晰、来源可溯的行业报告。我们将从概念理解开始逐步拆解其工作机制然后提供一个可运行的本地部署与集成示例最后深入分析其配置、常见问题及在生产环境中的最佳实践。无论你是希望为自己的团队搭建一个自动化的信息雷达还是想深入学习 Agent 工作流的设计与实现这篇文章都将提供一条清晰的路径。1. 理解 Agent Skill 与自动化工作流的核心价值在深入代码之前我们需要厘清几个关键概念Agent、Agent Skill 和工作流。它们共同构成了现代自动化信息处理系统的基石。1.1 什么是 AI AgentAI Agent 通常指一个能够感知环境、自主决策并执行行动以实现特定目标的智能体。在软件开发语境下一个 Agent 可以理解为一个封装了特定能力如搜索、分析、写作的程序模块它能够接收指令通常通过自然语言调用工具Tools或技能Skills并返回结构化的结果。与传统的脚本不同Agent 更强调自主性和上下文理解能力。1.2 Agent Skill 的本质Agent Skill 是 Agent 可调用的、完成特定任务的标准化能力单元。你可以把它类比为手机上的一个“小程序”或一个“插件”。一个设计良好的 Skill 应该具备明确的输入输出接口定义好接收什么参数返回什么格式的数据。可复用的逻辑其核心处理流程不依赖于特定的 Agent 实例。自包含性尽可能减少外部依赖或明确声明依赖。错误处理能够妥善处理异常情况并给出有意义的反馈。ai-trend-radar-report项目就是一个典型的 Agent Skill它的核心任务是“生成 AI 行业趋势报告”。1.3 工作流如何将 Skill 串联起来单个 Skill 可能只完成一个步骤比如“获取新闻列表”。而一个复杂任务如生成日报通常需要多个步骤协同获取信息、清洗去重、分析筛选、格式化输出、来源校验等。工作流Workflow就是定义这些步骤执行顺序、数据传递和条件分支的蓝图。在ai-trend-radar-report中其工作流可以抽象为以下阶段采集Crawl从预设的、优先国内可访问的信息源如 AIbase、量子位拉取原始数据。清洗与去重Deduplicate去除重复转载的新闻合并相似内容提取核心信息。上下文筛选Filter Enrich根据时效性、重要性等规则筛选条目并可能回溯官方页面、论文等补充详细信息。报告生成与校验Generate Validate将筛选后的内容按照固定模板日报、周报、专题组织成 Markdown 报告并校验报告是否包含无关的中间过程信息如搜索日志确保最终输出干净。这个工作流的设计正是为了解决项目简介中提到的痛点信息源分散、访问不稳定、内容重复、AI 幻觉混入报告等。2. 环境准备与项目结构解析要运行或集成这个 Agent Skill首先需要理解它的运行环境和代码结构。该项目强调“不依赖第三方 Python 包只使用标准库”这极大地降低了部署的复杂性。2.1 基础环境要求Python 3.8项目基于 Python 标准库确保你安装了合适版本的 Python。网络连接需要能够访问项目预设的信息源网站如aibase.cn,qbitai.com等。支持 Agent Skills 的工具可选如 Claude Code、Cursor 等用于直接集成和调用。本文也会演示如何在纯本地 Python 环境中运行核心逻辑。首先将项目克隆到本地git clone https://github.com/lgy1027/ai-trend-radar-report.git cd ai-trend-radar-report2.2 项目目录结构分析查看项目根目录理解各个文件的作用ai-trend-radar-report/ ├── README.md # 项目说明文档 ├── skill.json # Agent Skill 的元数据描述文件关键 ├── main.py # Skill 的入口主程序 ├── config.yaml # 配置文件信息源、报告模板等 ├── sources/ # 信息源采集模块目录 │ ├── __init__.py │ ├── aibase.py # AIbase 信息源实现 │ ├── qbitai.py # 量子位信息源实现 │ └── ... # 其他信息源 ├── processors/ # 数据处理器目录 │ ├── __init__.py │ ├── deduplicator.py # 去重处理器 │ ├── filter.py # 过滤器 │ └── validator.py # 输出校验器 ├── templates/ # 报告模板目录 │ ├── daily.md.j2 # 日报模板 │ ├── weekly.md.j2 # 周报模板 │ └── ... └── outputs/ # 报告输出目录通常由程序生成关键文件解析skill.json这是 Agent Skill 的“身份证”。它定义了 Skill 的名称、描述、输入参数、输出格式等信息使得支持 Agent Skills 的平台能够自动识别和加载它。理解这个文件是集成到 Claude Code 等工具的关键。config.yaml集中管理所有配置如启用哪些信息源、报告生成的模式fast/standard/deep、输出风格等。修改配置比修改代码更安全、更灵活。sources/目录每个.py文件对应一个信息源的采集逻辑。这种模块化设计使得新增一个信息源如“机器之心”只需在目录下添加一个新文件并在配置中启用即可。processors/目录实现了工作流中的各个处理环节。这种设计符合“单一职责原则”每个处理器只做一件事便于测试和维护。templates/目录使用 Jinja2 模板引擎Python 标准库jinja2需单独安装但项目声明“不依赖第三方包”此处可能内嵌了简易模板逻辑或依赖环境已有定义报告格式。分离格式和逻辑便于定制报告样式。2.3 核心依赖检查虽然项目宣称不依赖第三方包但为了处理 HTML 和生成报告通常会用到一些库。让我们检查main.py或相关文件的导入部分# 示例可能用到的标准库或内置功能 import json import yaml # 可能需要 pyyaml但有时可用标准库替代或自定义加载 import datetime import re import urllib.request from pathlib import Path import hashlib注意如果运行时报错缺少yaml或jinja2模块你需要通过pip install pyyaml jinja2安装。这与项目“标准库”的声明可能略有出入在实际工程中明确声明依赖并管理requirements.txt是更推荐的做法。为了快速验证可以先安装这些常用库。3. 核心工作流实现与配置详解现在我们深入代码看这个 Agent Skill 是如何将工作流一步步实现的。3.1 工作流引擎main.py解析main.py是整个 Skill 的调度中心。它通常遵循以下流程# 伪代码展示 main.py 的核心逻辑 def generate_report(report_typedaily, modestandard, stylestandard): 生成报告的主函数。 Args: report_type: 报告类型如 daily, weekly, brief, topic mode: 采集模式如 fast, standard, deep style: 输出风格如 concise, standard, detailed Returns: 生成的报告文本Markdown格式 # 1. 加载配置 config load_config(config.yaml) # 2. 根据模式和类型确定要使用的信息源和处理器 enabled_sources config[sources][mode] enabled_processors config[processors][report_type] # 3. 数据采集阶段 all_raw_items [] for source_name in enabled_sources: source_module importlib.import_module(fsources.{source_name}) items source_module.fetch() # 调用每个信息源的 fetch 函数 all_raw_items.extend(items) # 4. 数据处理阶段工作流核心 processed_items all_raw_items for processor_name in enabled_processors: processor_module importlib.import_module(fprocessors.{processor_name}) # 将上一步处理的结果传递给下一步形成管道 processed_items processor_module.process(processed_items) # 5. 报告生成阶段 template load_template(ftemplates/{report_type}.md.j2) report_content template.render( itemsprocessed_items, generated_timedatetime.now(), stylestyle ) # 6. 输出校验阶段 if is_valid_report(report_content): return report_content else: raise ValueError(生成的报告未通过校验可能混入了调试信息。)这个流程清晰体现了工作流的思想数据像在流水线上一样依次经过各个“工位”处理器每个工位完成特定加工最终产出成品。3.2 关键配置config.yaml解读配置文件是控制工作流行为的枢纽。一个典型的config.yaml可能如下所示# config.yaml 示例 sources: fast: - aibase - qbitai standard: - aibase - qbitai - infoq deep: - aibase - qbitai - infoq - official_blog # 回溯官方博客 - arxiv # 回溯论文 processors: daily: - deduplicator - time_filter # 过滤24小时内的 - validator weekly: - deduplicator - time_filter # 过滤7天内的 - topic_cluster # 按主题聚类 - validator output: default_format: markdown styles: concise: include_sections: [highlights, trends] standard: include_sections: [highlights, trends, details, sources] detailed: include_sections: [highlights, trends, details, sources, analysis, risks] report_types: daily: description: AI Daily Report max_items: 15 weekly: description: AI Weekly Report max_items: 30配置项说明sources: 定义了不同采集模式下的信息源组合。fast模式只查最快的信息源deep模式会进行深度回溯验证。processors: 定义了不同类型报告对应的处理管道。日报可能只需要去重和时效过滤周报则可能增加主题聚类。output: 控制报告的最终呈现形式。report_types: 定义每种报告的特性如最大条目数。通过修改这个文件你可以轻松地增加或移除信息源。调整处理流程的顺序和组合。改变报告的风格和包含的模块。3.3 信息源实现示例sources/aibase.py看看一个具体的信息源是如何工作的# sources/aibase.py import urllib.request from bs4 import BeautifulSoup # 可能需要安装 bs4但项目声明用标准库这里可能用 html.parser import html.parser import json from datetime import datetime, timedelta def fetch(): 从 AIbase 获取最新的 AI 资讯列表。 url https://www.aibase.com/ headers {User-Agent: Mozilla/5.0} # 模拟浏览器访问 req urllib.request.Request(url, headersheaders) try: with urllib.request.urlopen(req, timeout10) as response: html_content response.read().decode(utf-8) except Exception as e: print(fFailed to fetch from AIbase: {e}) return [] # 优雅降级一个源失败不影响整体 # 使用标准库 html.parser 解析假设项目坚持不用第三方库 # 这里简化处理实际解析逻辑会更复杂 soup BeautifulSoup(html_content, html.parser) # 如果不用bs4则需用 html.parser 手动解析 news_items [] # 假设 AIbase 的新闻条目有特定的 CSS 类 for article in soup.find_all(div, class_news-item): title_elem article.find(h3) link_elem article.find(a, hrefTrue) time_elem article.find(span, class_time) if title_elem and link_elem: title title_elem.text.strip() link link_elem[href] # 处理相对链接 if link.startswith(/): link https://www.aibase.com link pub_time time_elem.text.strip() if time_elem else datetime.now().strftime(%Y-%m-%d) news_items.append({ title: title, url: link, source: AIbase, published_at: pub_time, summary: , # 初始摘要为空可由后续处理器补充 category: general }) # 返回结构化的数据列表 return news_items[:10] # 限制每次获取的数量关键点错误处理网络请求可能失败必须有try...except包裹返回空列表而非崩溃。结构化数据每个新闻条目被转换为字典包含标题、链接、来源、时间等统一字段。这是工作流中数据流转的基础。优雅降级一个信息源失败不应导致整个工作流中断。3.4 处理器示例processors/deduplicator.py去重是信息聚合的关键步骤。一个简单的基于标题相似度的去重器# processors/deduplicator.py import hashlib from difflib import SequenceMatcher def calculate_similarity(str1, str2): 计算两个字符串的相似度0-1。 return SequenceMatcher(None, str1, str2).ratio() def process(items, similarity_threshold0.8): 对新闻条目列表进行去重。 基于标题相似度和来源进行判断。 unique_items [] seen_hashes set() for item in items: # 生成一个基于标题和来源的“指纹”用于精确匹配 item_hash hashlib.md5(f{item[title]}_{item[source]}.encode()).hexdigest() if item_hash in seen_hashes: continue # 完全重复跳过 # 模糊匹配检查与已保留条目的相似度 is_duplicate False for unique_item in unique_items: if calculate_similarity(item[title], unique_item[title]) similarity_threshold: # 如果相似度很高认为是同一事件合并来源或保留更早/更权威的 # 这里简单策略跳过新的保留旧的 is_duplicate True # 可选合并来源链接到 unique_item # unique_item[alternative_sources].append(item[url]) break if not is_duplicate: unique_items.append(item) seen_hashes.add(item_hash) return unique_items这个处理器展示了工作流中“数据转换”的典型模式接收一个列表处理后返回一个新的列表。4. 本地运行与集成到 Agent 工具理解了核心代码后我们来看看如何实际使用它。4.1 在本地命令行运行最简单的方式是直接运行main.py脚本。项目可能提供了命令行接口。我们可以查看main.py的if __name__ __main__:部分或者自己封装一个简单的运行脚本# run_local.py import sys sys.path.insert(0, .) # 确保可以导入项目模块 from main import generate_report if __name__ __main__: # 生成今日AI日报 report_type daily mode standard style standard try: report generate_report(report_typereport_type, modemode, stylestyle) # 打印到控制台 print(report) # 同时保存到文件 filename fai_report_{report_type}_{datetime.now().strftime(%Y%m%d_%H%M%S)}.md with open(filename, w, encodingutf-8) as f: f.write(report) print(f\n报告已保存至: {filename}) except Exception as e: print(f生成报告时出错: {e}) sys.exit(1)运行命令python run_local.py如果一切正常你将在当前目录下得到一个以时间戳命名的 Markdown 文件内容就是生成的 AI 日报。4.2 解析skill.json并集成到 Claude Code要让这个 Skill 被 Claude Code 等工具识别关键在于skill.json文件。这个文件遵循一定的规范如 MCPModel Context Protocol 或工具自定义的格式。{ name: ai-trend-radar-report, description: An Agent Skill for generating Chinese AI industry trend reports (daily, weekly, brief, topic)., version: 0.1.0, author: lgy1027, entrypoint: main.py, protocol: mcp, // 或特定工具要求的协议 capabilities: { generate_report: { description: Generate an AI trend report., parameters: { report_type: { type: string, enum: [daily, weekly, brief, topic], description: Type of report to generate., default: daily }, mode: { type: string, enum: [fast, standard, deep], description: Depth of information gathering., default: standard }, style: { type: string, enum: [concise, standard, detailed], description: Output style of the report., default: standard } }, returns: { type: string, description: The generated report in Markdown format. } } } }集成步骤以 Claude Code 为例放置 Skill将整个ai-trend-radar-report项目文件夹放到 Claude Code 指定的 Skills 目录下具体路径需查看 Claude Code 文档。刷新或重启工具让 Claude Code 重新扫描并加载新的 Skill。调用 Skill在 Claude Code 的对话或命令窗口中你可以通过自然语言或特定命令调用该 Skill例如“请使用ai-trend-radar-reportSkill 生成一份本周的 AI 周报使用深度模式。”Agent 自动执行Claude Code 中的 Agent 会解析你的指令匹配到skill.json中定义的generate_report能力并传入相应的参数report_type: weekly,mode: deep然后执行main.py中的逻辑最后将生成的 Markdown 报告返回给你。这个过程实现了“一句话指令自动完成复杂工作流”。5. 常见问题排查与优化实践在实际部署和运行过程中你可能会遇到一些问题。以下是典型的排查路径和解决方案。5.1 问题一运行时报网络错误或获取不到数据现象运行脚本后控制台输出Failed to fetch from [SourceName]或长时间无响应后超时。可能原因与排查步骤网络连接问题检查尝试在浏览器中直接访问config.yaml中配置的信息源网址如https://www.aibase.com。解决如果无法访问可能是网络环境导致。考虑为urllib.request配置代理需谨慎处理代理设置仅用于开发测试环境。或者在sources配置中暂时注释掉不可用的源。网站反爬机制检查脚本使用的User-Agent过于简单被网站拒绝。查看返回的 HTTP 状态码如 403。解决在sources/下的各个.py文件中优化请求头。可以模拟更真实的浏览器或添加Referer等头部信息。headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, }网页结构变更检查这是最常见的问题。信息源网站的 HTML 结构更新了导致解析代码找不到对应的标签如div.news-item。解决需要更新对应的sources/[source_name].py文件中的解析逻辑。使用浏览器的开发者工具F12重新分析目标网站的元素结构调整find或find_all中使用的标签和类名。5.2 问题二生成的报告内容空洞或重复条目多现象报告生成了但条目很少或者不同信息源的同一新闻被重复列出多次。可能原因与排查步骤去重阈值设置不当检查查看processors/deduplicator.py中的similarity_threshold参数。默认值 0.8 可能过高或过低。解决调整阈值。可以尝试打印出被判定为重复的标题对及其相似度分数根据实际情况调整。# 在 deduplicator.py 的 process 函数中添加调试信息 sim calculate_similarity(item[title], unique_item[title]) if sim similarity_threshold: print(f判定为重复: {item[title]} 与 {unique_item[title]}, 相似度: {sim:.2f}) is_duplicate True信息源失效或解析逻辑过时检查单独测试每个信息源的fetch函数看其是否能返回预期数量的数据。解决同问题一的第3点更新解析逻辑。采集模式mode配置问题检查config.yaml中sources下对应模式如standard配置的信息源列表是否完整且有效。解决确保配置的信息源模块名与sources/目录下的文件名一致并且该模块能正确导入。5.3 问题三集成到 Claude Code 后无法识别或调用失败现象将项目文件夹放入 Skills 目录后在 Claude Code 中无法看到或调用该 Skill。可能原因与排查步骤skill.json格式错误或路径不对检查确保skill.json文件位于项目根目录并且是有效的 JSON 格式。可以使用在线 JSON 校验工具检查。解决修正 JSON 格式错误。确认 Claude Code 支持哪种 Skill 协议如 MCP并按照其官方文档调整skill.json的结构。依赖缺失检查虽然项目声明使用标准库但如果实际代码用了yaml、jinja2或bs4而 Claude Code 的运行环境没有这些包就会导致导入失败。解决在 Claude Code 的运行环境中安装缺失的包。或者修改项目代码用纯标准库实现例如用json代替yaml用string.Template代替jinja2用html.parser代替bs4。权限或执行路径问题检查Claude Code 是否有权限执行项目目录下的 Python 脚本。解决检查文件权限并确保main.py有可执行权限在 Linux/macOS 上chmod x main.py。同时检查skill.json中entrypoint指定的路径是否正确。5.4 性能与稳定性优化实践当这个 Skill 用于生产环境如定时生成日报并推送到团队群时需要考虑更多增加缓存机制问题频繁运行会反复请求相同网站可能触发反爬也浪费资源。实践为每个信息源的fetch函数增加简单的文件缓存或内存缓存例如将结果按源和时间戳缓存1小时。import os import pickle import time CACHE_DIR ./cache os.makedirs(CACHE_DIR, exist_okTrue) def fetch_with_cache(source_name, url, ttl3600): cache_file os.path.join(CACHE_DIR, f{source_name}_{hash(url)}.pkl) if os.path.exists(cache_file) and (time.time() - os.path.getmtime(cache_file)) ttl: with open(cache_file, rb) as f: return pickle.load(f) # 否则执行网络请求 data real_fetch(url) with open(cache_file, wb) as f: pickle.dump(data, f) return data完善日志记录问题运行出错时难以定位是哪个信息源或处理器出了问题。实践使用 Python 的logging模块替代print为不同模块设置不同日志级别并将日志输出到文件。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(ai_radar.log), logging.StreamHandler()]) logger logging.getLogger(__name__) # 在代码中使用 logger.info(f开始从 {source_name} 采集数据) try: items fetch() logger.info(f从 {source_name} 成功采集到 {len(items)} 条数据) except Exception as e: logger.error(f从 {source_name} 采集数据失败: {e}, exc_infoTrue)配置外部化与敏感信息管理问题信息源 URL、请求头等信息硬编码在代码中不便于管理和更新。实践将所有可配置项包括 URL、请求头、CSS 选择器、阈值参数都移至config.yaml或一个专门的sources_config.yaml中。这样无需修改代码即可调整采集行为。设置超时与重试问题某个信息源响应慢会导致整个工作流卡住。实践为urllib.request.urlopen设置合理的超时时间如 15 秒并实现简单的重试逻辑如最多重试2次。import socket def fetch_with_retry(url, max_retries2, timeout15): for attempt in range(max_retries): try: req urllib.request.Request(url, headersheaders) with urllib.request.urlopen(req, timeouttimeout) as response: return response.read().decode(utf-8) except (urllib.error.URLError, socket.timeout) as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避6. 扩展方向与自定义开发ai-trend-radar-report项目提供了一个优秀的起点你可以基于它进行深度定制打造属于自己的信息雷达。6.1 添加新的信息源这是最常见的扩展需求。假设你想添加“机器之心”作为信息源在sources/目录下创建新文件如jqr.py。实现fetch()函数参照现有信息源的代码结构适配“机器之心”网站的页面结构。在config.yaml的sources部分将jqr添加到相应的模式列表中。测试单独运行这个新源的fetch函数确保能正确解析数据。6.2 开发新的处理器工作流的强大之处在于可插拔的处理器。例如你想增加一个“情感分析”处理器为每条新闻打上积极/消极/中性的标签在processors/目录下创建新文件如sentiment_analyzer.py。实现process(items)函数接收条目列表为每个item添加一个sentiment字段。# 简化示例实际可能需要调用NLP API或使用本地模型 def simple_sentiment(text): positive_words [突破, 领先, 开源, 免费, 强大] negative_words [争议, 漏洞, 泄露, 下跌, 制裁] # ... 简单的关键词匹配逻辑 return neutral def process(items): for item in items: item[sentiment] simple_sentiment(item[title] item.get(summary, )) return items在config.yaml的processors部分将sentiment_analyzer插入到合适的位置例如在deduplicator之后。修改报告模板在templates/daily.md.j2中可以利用新的sentiment字段对新闻进行分类展示。6.3 输出到更多渠道默认输出是 Markdown 文件。你可以轻松扩展将报告发送到更多地方邮件推送使用smtplib库将生成的 Markdown 内容作为邮件正文发送给团队成员。Webhook 通知将报告内容通过 HTTP POST 请求发送到团队协作工具如钉钉、飞书、Slack的 Webhook。生成图文并茂的公众号推文可以结合第三方库将 Markdown 转换为更美观的 HTML甚至生成图片。6.4 构建定时任务要让这个工作流完全自动化可以将其部署为定时任务Linux/macOS使用cron。# 每天上午9点运行生成日报 0 9 * * * cd /path/to/ai-trend-radar-report /usr/bin/python3 run_local.py /path/to/logs/cron.log 21Windows使用任务计划程序。云服务器/容器使用systemd timer或 Docker 容器的健康检查与定时任务结合。通过以上步骤你不仅能够运行一个现成的 AI 行业趋势报告生成器更能深入理解 Agent Skill 和工作流的设计思想并具备根据自身需求进行定制和扩展的能力。这种将固定、繁琐的信息处理流程自动化的模式可以广泛应用于竞品分析、舆情监控、技术动态追踪等多个领域是提升研发和信息处理效率的利器。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度