之前一直在琢磨怎么把自己的日常工作“外包”出去。每天打开浏览器先刷一遍行业资讯再看几条项目动态还要抽空搜一下合作伙伴有没有负面新闻一套流程下来至少一个小时而且经常遗漏重要信息。后来我干脆写了一个 AI Agent 小系统让它每天早上自动上网按我设定的关键词完成三件事找项目、找钱、查风险最后生成一份 Markdown 格式的日报我只需要花五分钟看一眼结论就行。这篇文章就围绕这个系统展开从架构设计到代码实现把每一步拆开讲清楚。如果你也想让 AI 帮你处理重复性的信息搜集工作或者准备入门 AI Agent 开发本文将带你从零搭建一个可扩展的自动化信息监控 Agent。1. AI 每天上网的本质一个自动化 Agent先说一个容易混淆的概念这不是普通爬虫而是一个 AI Agent。普通爬虫做的事情是“抓取”把网页内容下载下来解析出标题、正文、链接然后原样保存。但面对“找项目、找钱、查风险”这类需求时爬虫远远不够因为海量信息里真正对你有用的可能只有几条而且每条信息还需要结合业务背景做判断。AI Agent 则多了一层“理解 决策 行动”的能力。它会先接收任务目标自己规划采集范围调用不同数据源获取信息再交给大模型分析最后把结果整理成结构化报告。整个过程不依赖人工干预每天按时执行。1.1 Agent 的三个核心能力拆解来看这套系统需要三个能力感知能力能够从互联网获取最新信息。这里我使用公开 API 和 RSS 作为数据源而不是爬取网页正文既稳定又合规。分析能力从原文中理解信息含义判断它跟我的目标是否相关。这里调用大模型接口完成本质上是一次文本分类和摘要生成。行动能力根据分析结果生成报告并在检测到高风险信息时发出提醒。这三个能力分别对应系统里的采集模块、分析模块和报告模块。1.2 Agent 与一次性脚本的区别你可能会说“这不就是一个定时脚本吗”从实现上看确实有相似之处但关键区别在于任务编排方式。传统脚本的执行路径是写死的抓哪个网站、提取哪些字段、输出什么格式全部在代码里硬编码。Agent 则更灵活你只需要告诉它“找风险信息”它可以自己决定去哪些源、用什么关键词组合、如何判断风险等级。这就是为什么这两年大家都在聊 Agent 工作流。当然完全自主决策在实际工程中不可控所以我在设计时采用了“半自动化”策略信息采集的范围由配置定义大模型负责分析判断最终高风险结论必须经过人工确认。这样既保留了 Agent 的智能性又控制了风险。2. 系统整体设计在写代码之前先把系统结构理清楚。整个项目我命名为ai-daily-scout通过组件解耦的方式设计方便后续扩展数据源和算法。2.1 模块划分系统分四个核心模块模块职责关键实现collector信息采集调用 GitHub API、RSS 解析器获取原始信息analyzer信息分析调用大模型对原始信息进行过滤、摘要、风险标注reporter报告生成将分析结果输出为 Markdown 日报scheduler定时调度每天定时触发上述流程这种分层的设计很容易扩展。比如后期想增加新闻源只需要在 collector 里新增一个函数想调整分析策略只需要修改 analyzer 里的 prompt。2.2 数据流设计整个流程可以概括为一条单向数据管道关键词配置 → 采集器 → 原始信息列表 → 大模型分析 → 结构化结果 → 报告生成 → 日报文件每个环节的输入都是上一个环节的输出结构清晰便于排查问题。2.3 项目目录结构这是一个单机 Python 项目目录结构如下ai-daily-scout/ ├── config.py # 全局配置关键词、数据源、模型参数 ├── collector.py # 信息采集模块 ├── analyzer.py # 大模型分析模块 ├── reporter.py # 报告生成模块 ├── scheduler.py # 定时调度模块 ├── main.py # 主流程入口 ├── requirements.txt # 依赖清单 ├── .env # 密钥配置不要提交到代码仓库 └── reports/ # 生成的日报存放目录3. 环境准备与项目初始化3.1 运行环境本文示例以 Python 3.10 为例操作系统不限Windows / macOS / Linux 均可。建议使用虚拟环境隔离依赖。主要依赖库如下requests发送 HTTP 请求beautifulsoup4解析 HTML 内容feedparser解析 RSS 订阅源openai调用 OpenAI 兼容的大模型接口schedule实现定时任务python-dotenv读取.env文件中的密钥3.2 安装依赖创建项目目录并进入mkdir ai-daily-scout cd ai-daily-scout python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate创建requirements.txtrequests2.31.0 beautifulsoup44.12.3 feedparser6.0.11 openai1.30.0 schedule1.2.1 python-dotenv1.0.1版本可以根据你的实际环境调整重点是掌握实现思路。安装pip install -r requirements.txt3.3 初始化配置在项目根目录创建.env文件存放大模型的密钥和接口地址。这里以 OpenAI 兼容接口为例如果你使用的是国内大模型服务只需要替换BASE_URL和MODEL_NAMELLM_API_KEY你的密钥 LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini同时创建config.py用于存放监控关键词、RSS 数据源等业务配置# config.py import os from dotenv import load_dotenv load_dotenv() # 大模型配置 LLM_API_KEY os.getenv(LLM_API_KEY) LLM_BASE_URL os.getenv(LLM_BASE_URL) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 监控目标 USER_GOAL 关注人工智能与开源领域的新项目机会、政策资金动态、以及相关企业风险信息 # 监控关键词用于 GitHub 项目搜索 SEARCH_KEYWORDS [AI agent, LLM application, open source] # RSS 数据源按需替换为真实数据源 RSS_SOURCES [ https://example.com/feed.xml, ] # 报告输出目录 OUTPUT_DIR reports这里的RSS_SOURCES是示例占位符实际使用时要替换成你感兴趣的合规 RSS 地址比如技术博客、新闻网站的官方订阅源。4. 核心代码实现4.1 信息采集 Agentcollector.py采集模块负责把不同来源的信息汇总成统一格式。这里我实现了两个采集函数fetch_github_repos()通过 GitHub Search API 搜索相关项目对应“找项目”场景。fetch_rss()解析 RSS 订阅源获取行业资讯对应“找信息”场景。# collector.py import requests import feedparser GITHUB_SEARCH_API https://api.github.com/search/repositories HEADERS { User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 } def fetch_github_repos(keyword: str, per_page: int 5) - list[dict]: 通过 GitHub Search API 搜索相关项目。 返回统一的条目格式title / url / description / source / type params { q: keyword, sort: updated, order: desc, per_page: per_page } resp requests.get(GITHUB_SEARCH_API, paramsparams, headersHEADERS, timeout15) resp.raise_for_status() data resp.json() items [] for repo in data.get(items, []): items.append({ title: repo.get(full_name), url: repo.get(html_url), description: repo.get(description), source: github, type: project }) return items def fetch_rss(rss_url: str, limit: int 10) - list[dict]: 解析 RSS 订阅源获取最新资讯条目。 feed feedparser.parse(rss_url) items [] for entry in feed.entries[:limit]: items.append({ title: entry.get(title), url: entry.get(link), summary: entry.get(summary, )[:300], source: rss_url, type: news }) return items关键点说明使用公开 API 而不是网页爬虫。GitHub Search API 是官方接口只要请求频率不超标基本不会被封。数据合规性也更好。统一条目格式。不管是项目还是新闻最终都转成包含title、url、source、type的字典这样下游模块不需要关心数据来源差异。设置请求超时。网络请求必须设置timeout避免程序卡死。实际场景中你还可以增加更多采集函数比如监控招标网站公告、政府政策文件页面、行业论坛热帖等只要把数据转成同样的格式即可。4.2 大模型分析 Agentanalyzer.py采集到的原始信息是碎片化的需要大模型来完成“过滤 摘要 风险判断”三个任务。我在这里采用了结构化输出方式要求模型返回 JSON 数组方便后续程序处理。# analyzer.py import json import os from openai import OpenAI # 初始化客户端兼容 OpenAI 与国内大模型服务 client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL) ) def analyze_with_llm(items: list[dict], user_goal: str) - list[dict]: 将原始信息交给大模型分析。 返回结构化列表每条包含 title / url / summary / risk_level / risk_reason # 控制传给模型的文本长度避免超限 text \n.join( [f- 标题{item[title]}\n 链接{item[url]}\n 描述{item.get(description) or item.get(summary, )} for item in items] )[:4000] prompt f你是我的信息分析助手我关注的目标是「{user_goal}」 下面是今天采集到的原始信息请帮我完成三件事 1. 筛选出与我的目标最相关的 5 条 2. 对每条信息写一句简洁摘要说清楚它和我目标的关系 3. 判断这条信息是否存在风险风险等级只能是 low / medium / high并给出简短风险原因。 按 JSON 数组格式返回不要输出其他解释 [ {{ title: 标题, url: 链接, summary: 一句话摘要, risk_level: low/medium/high, risk_reason: 风险原因或说明 }} ] 原始信息列表 {text} resp client.chat.completions.create( modelos.getenv(LLM_MODEL, gpt-4o-mini), messages[{role: user, content: prompt}], temperature0.3 ) content resp.choices[0].message.content.strip() # 清理大模型可能输出的代码块标记 if content.startswith(json): content content[7:] if content.endswith(): content content[:-3] content content.strip() return json.loads(content)这里有几个设计细节值得注意Prompt 里明确输出格式。大模型输出不稳定最常见的问题就是格式漂移所以我在 prompt 末尾指定了 JSON 结构并在代码里做了 json 标记清理。限制输入长度。把原始信息截断到 4000 字符以内一方面控制 token 成本另一方面防止上下文过长导致模型注意力分散。temperature 设置为 0.3。分析类任务需要稳定输出温度越低结果越确定不会出现“这次说风险高、下次说风险低”的随机情况。大模型返回的内容可以直接转成 Python 对象后续报告模块就能直接使用。4.3 风险识别与报告生成reporter.py分析模块返回的是带风险等级的结构化数据但用户直接看 JSON 太不友好所以需要一个报告生成模块。这里我选择生成 Markdown 格式日报既可以本地阅读也可以上传到语雀、Notion 等笔记工具。# reporter.py from datetime import datetime def generate_markdown_report(analyzed_items: list[dict], keywords: list[str]) - str: 将分析结果生成 Markdown 日报 now datetime.now().strftime(%Y-%m-%d %H:%M) lines [] lines.append(f# AI 每日信息日报 - {now}) lines.append() lines.append(f 监控关键词{, .join(keywords)}) lines.append() # 风险预警区 lines.append(## 风险预警) high_risk [item for item in analyzed_items if item.get(risk_level) high] if high_risk: for item in high_risk: lines.append(f- **{item[title]}**{item.get(risk_reason, )}) else: lines.append(- 今日无高风险预警) lines.append() # 项目动态区 lines.append(## 重点信息) for item in analyzed_items: risk_tag { high: , medium: , low: }.get(item.get(risk_level, low), ) lines.append(f- {risk_tag} [{item[title]}]({item[url]}){item.get(summary, )}) if item.get(risk_reason): lines.append(f - 风险说明{item[risk_reason]}) return \n.join(lines)为什么单独做一个报告模块因为 Agent 的“产出”不应该是大模型的原始回答而应该是经过整理的、可直接阅读的成果物。抽离成独立函数后后续可以轻松把日报改成 CSV、PDF 或者直接推送到钉钉、飞书群。4.4 主流程串联main.py主流程负责把所有模块串起来。这一步很关键Agent 不是各个函数的散装组合而是由主流程编排的完整流水线。# main.py import os from datetime import datetime from collector import fetch_github_repos, fetch_rss from analyzer import analyze_with_llm from reporter import generate_markdown_report from config import SEARCH_KEYWORDS, RSS_SOURCES, USER_GOAL, OUTPUT_DIR def run_daily_scan(): 执行一次完整的信息采集与分析流程 all_items [] # 1. 采集按关键词搜索 GitHub 项目 for kw in SEARCH_KEYWORDS: try: repos fetch_github_repos(kw, per_page5) all_items.extend(repos) print(f[collector] 关键词「{kw}」采集到 {len(repos)} 个项目) except Exception as e: print(f[collector] 关键词「{kw}」采集失败{e}) # 2. 采集解析 RSS 源 for rss_url in RSS_SOURCES: try: news fetch_rss(rss_url, limit10) all_items.extend(news) print(f[collector] RSS {rss_url} 采集到 {len(news)} 条资讯) except Exception as e: print(f[collector] RSS {rss_url} 采集失败{e}) # 3. 去重接根据 URL 去重 seen_urls set() unique_items [] for item in all_items: url item.get(url) if url and url not in seen_urls: seen_urls.add(url) unique_items.append(item) print(f[collector] 去重后共 {len(unique_items)} 条有效信息) if not unique_items: print([analyzer] 没有可分析的信息流程结束) return # 4. 分析交给大模型 analyzed_items analyze_with_llm(unique_items, USER_GOAL) print(f[analyzer] 大模型筛选出 {len(analyzed_items)} 条重点信息) # 5. 生成报告 report generate_markdown_report(analyzed_items, SEARCH_KEYWORDS) # 6. 保存报告 os.makedirs(OUTPUT_DIR, exist_okTrue) output_path os.path.join(OUTPUT_DIR, freport_{datetime.now().strftime(%Y%m%d)}.md) with open(output_path, w, encodingutf-8) as f: f.write(report) print(f[reporter] 报告已生成{output_path}) if __name__ __main__: run_daily_scan()这个主流程看起来简单但已经包含了 Agent 工作流的基本骨架采集 → 清洗 → 分析 → 产出。每一步都打印日志方便观察任务执行到哪个环节、是否出错。5. 定时调度让 Agent 自动运行主流程写完之后还需要一个“每天自动启动”的机制。这里我用schedule库实现定时调度。5.1 使用 schedule 实现每日任务创建一个scheduler.py# scheduler.py import schedule import time from main import run_daily_scan def start_scheduler(): # 每天早上 9 点执行一次 schedule.every().day.at(09:00).do(run_daily_scan) print(定时任务已启动每天 09:00 自动执行信息采集与分析) while True: schedule.run_pending() time.sleep(30) if __name__ __main__: start_scheduler()这段代码很简单解决的是“定时启动”的问题。每次循环休眠 30 秒检查是否有到期任务需要执行。这个方案适合个人电脑上长期运行。如果是部署在服务器上更推荐使用 crontab 或 systemd timer避免 Python 进程异常退出后没有兜底重启。5.2 手动运行验证在定时任务跑起来之前先手动执行一遍主流程python main.py如果一切正常会输出类似下面的日志[collector] 关键词「AI agent」采集到 5 个项目 [collector] 关键词「LLM application」采集到 5 个项目 [collector] 关键词「open source」采集到 5 个项目 [collector] RSS https://example.com/feed.xml 采集失败... [collector] 去重后共 15 条有效信息 [analyzer] 大模型筛选出 5 条重点信息 [reporter] 报告已生成reports/report_20241218.md这里 RSS 采集失败是正常的因为示例地址不可用。实际配置里替换成真实 RSS 源即可。5.3 一次运行结果示例生成的日报长这样# AI 每日信息日报 - 2024-12-18 09:02 监控关键词AI agent, LLM application, open source ## 风险预警 - [some/malicious-repo](https://github.com/some/malicious-repo)该项目声称提供免费 API但要求关闭本地防火墙存在安全风险。 ## 重点信息 - [mewamew/my_ai_town](https://github.com/mewamew/my_ai_town)开源 AI 小镇模拟项目提供 Mac 版游戏下载适合作为 Agent 行为实验环境。 - [open-source-llm-tracker](https://github.com/...)持续跟踪最新开源大模型榜单。 - [company-x-ai-product](https://example.com/news)某公司发布 AI 新品但公告未披露模型训练数据来源。这里的my_ai_town是真实的开源项目链接它是一个 AI 小镇模拟环境可以作为 Agent 行为的沙盒来学习和调试。6. 从示例到落地找项目、找钱、查风险的真实场景前面演示的是一个通用框架下面具体拆解如何把它应用到“找项目、找钱、查风险”三个实际场景。6.1 找项目开源与招标信息监控“找项目”在工程领域可以理解成“找可参与的项目机会”包括开源协作项目、招标公告、合作需求。对于开源项目GitHub Search API 是首选数据源。你可以在SEARCH_KEYWORDS里配置你关心的技术栈关键词比如langchain、rag、ai agent系统会每日搜索最新更新的仓库。对于招标公告很多政府采购网站没有开放 API这时可以用 RSS 或者页面探测方式。但需要注意必须先阅读目标网站的robots.txt确认抓取行为是否被允许。请求频率要控制建议单次任务最多抓取几个页面并添加延迟。优先使用官方网站提供的公告接口。6.2 找钱政策资金与行业动态监控“找钱”指的是挖掘商业机会信息包括政策补贴、融资新闻、人才需求、行业报告等。你的数据源设计可以参考这些政府政策文件 RSS部分地区政府网站提供政策文件 RSS订阅后可以第一时间获取资金申报、产业扶持文件。投资机构官网新闻页很多机构会发布投资动态、行业观察这些都是高质量的商业机会信号。招聘网站技术岗位列表大厂高薪岗位数量变化某种程度上反映了技术方向的投入力度这个信号也能帮助判断机会。把这些 RSS 地址加入RSS_SOURCES大模型的 prompt 里加入“是否包含商业机会、政策利好、融资动态”它就会自动帮你筛选出有价值的条目。6.3 查风险舆情与合规预警风险监控是这套系统最有价值的地方。你可以监控以下几类信息合作方动态对方是否出现负面新闻、法律诉讼、高管变动。行业风险监管政策变化、技术安全事故。项目风险开源项目是否停止维护、是否出现安全事故通报。风险判断不应只靠大模型“感觉”建议在 prompt 里明确风险标准例如风险判断标准 - high涉及法律诉讼、数据泄露、资金链断裂、产品停运等明确负面事件 - medium出现可能间接影响业务的不确定信息 - low正常动态无风险信号同时设置一条铁律AI 标记为高风险的信息必须人工复核后才能对外反馈或执行下一步操作。不要让 Agent 在没有监督的情况下把一个“疑似风险”直接发到工作群或者触发自动处置流程。6.4 与 AI 小镇等模拟环境的结合现在很多开源项目开始用游戏化的沙盒环境来训练 Agent比如 GitHub 上的my_ai_town项目。它提供了类似《模拟小镇》的环境你可以让 AI 扮演不同角色在虚拟环境里做任务观察 Agent 的行为是否合理。我建议的做法是先在模拟环境里验证 Agent 的任务规划能力等逻辑成熟后再把它接到真实互联网数据源上。因为真实互联网信息噪音大、格式乱、还有不合规内容如果 Agent 的基础行为策略没有调好一上来接真实数据会被各种异常打乱。7. 常见问题与排查思路开发过程中难免遇到各种问题下面整理几个高频场景。7.1 大模型返回 JSON 解析失败现象analyze_with_llm报JSONDecodeError。原因大模型有时会在 JSON 前后添加解释性文字或者输出里的引号、逗号不合法。排查步骤把resp.choices[0].message.content打印出来肉眼确认格式。确认removeprefix(json)清理逻辑是否覆盖所有情况。查看报错时的 JSON 片段锁定是哪个字段格式问题。解决方案在 prompt 里强调“只输出 JSON不要解释”并且把temperature调低到 0.3 以下。如果还不行可以引入json_repair库做容错。7.2 请求被拒绝或返回 403现象GitHub API 返回403或者 RSS 请求超时。原因请求频率过高、未设置 User-Agent、IP 被限流。排查步骤检查headers是否携带了合理的User-Agent。GitHub API 匿名请求有速率限制关注响应头里的X-RateLimit-Remaining字段。确认目标平台是否允许程序访问。解决方案降低抓取频率增加请求间隔优先使用官方 API为 API 请求配置密钥。7.3 定时任务不触发现象python scheduler.py启动后到了指定时间没有执行。原因进程可能在后台被杀死或者系统休眠schedule 库无法在休眠期间补执行任务。排查步骤先用python main.py手动执行确认主流程正常。查看 scheduler 是否打印了“定时任务已启动”。确认电脑在 9 点没有处于休眠状态。解决方案服务器环境使用 cron 替代在脚本开头加print日志方便定位。7.4 Agent 幻觉导致误判现象大模型在风险分析时把完全正常的信息判断为高风险或者凭空编造风险原因。原因大模型存在幻觉尤其是 prompt 语义模糊时模型会“脑补”不存在的细节。解决方案给风险等级增加明确的判断标准不要用模糊词汇。请求模型输出时必须引用原始信息中的证据禁止无依据判断。在报告里标记“AI 生成内容仅供参考需人工复核”。7.5 合规风险现象系统采集了不该采集的内容或者使用了未授权的数据源。原因对数据源规则不熟悉。解决方案所有数据采集必须遵守目标网站的用户协议和 robots 协议优先使用官方 API不在本地保存敏感个人信息涉及对外发布的信息必须人工审核。8. 工程化与部署建议代码能跑起来之后离“能长期稳定运行”还有一段距离。下面这些工程化建议值得收藏。8.1 日志与可观测性现在的 print 日志只能应付开发阶段长期运行建议使用logging模块按级别输出import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[ logging.FileHandler(agent.log, encodingutf-8), logging.StreamHandler() ] )有了日志文件即使某天任务异常也能快速定位是哪一步出了问题。8.2 密钥与配置管理这篇文章里密钥放在.env文件中这是一个很好的起点。但请注意.env文件不要提交到 Git 仓库建议在.gitignore中忽略。项目部署到服务器时可以使用环境变量或密钥管理服务注入不要把明文密钥写死在代码里。如果密钥泄露立即在云端控制台吊销并重新生成。8.3 成本控制大模型 API 调用不是免费的每日运行会积累成本。控制成本可以从几个方面先过滤再分析。优先用关键词或规则过滤掉明显不相关的信息再交给大模型减少传入 token。降低调用频率。如果信息变化不快可以改成每周两次而不是每天。选择更小的模型。不是所有任务都需要最强模型摘要和分类用轻量模型完全够用。设置调用上限。在代码里记录每日 token 消耗接近预算时自动停止。8.4 安全边界一定要明确 Agent 的权限边界。本文的系统只是“采集信息 生成报告”没有任何对外操作权限所以即使分析结果有误影响也可控。如果后续要扩展成自动发邮件、自动发布文章、甚至自动交易务必加上以下机制人工审批节点高风险动作必须人工确认。操作白名单Agent 只能调用白名单内的接口。全量操作日志记录 Agent 每次执行的输入、输出和动作便于回溯。9. 最后说几句回过头来看这个 AI Agent 系统的核心价值不是“替代你上网”而是把每天重复的信息搜集和初筛工作标准化、自动化。你依然需要用自己的判断力处理最终结果但节省下来的时间非常可观。建议你从最小闭环开始先配置两三个关键词接一个 RSS 源手动跑通主流程再逐步增加数据源和定时任务。等基本功能稳定后可以尝试接入企业微信或飞书机器人把日报推送到群里效果会更好。如果你在自己的实践过程中遇到了有意思的问题欢迎在评论区交流。觉得这篇文章有启发的话可以收藏备用也可以转发给身边在做 AI 应用开发的朋友。 AI Agent实战：构建自动化信息监控与风险预警系统
之前一直在琢磨怎么把自己的日常工作“外包”出去。每天打开浏览器先刷一遍行业资讯再看几条项目动态还要抽空搜一下合作伙伴有没有负面新闻一套流程下来至少一个小时而且经常遗漏重要信息。后来我干脆写了一个 AI Agent 小系统让它每天早上自动上网按我设定的关键词完成三件事找项目、找钱、查风险最后生成一份 Markdown 格式的日报我只需要花五分钟看一眼结论就行。这篇文章就围绕这个系统展开从架构设计到代码实现把每一步拆开讲清楚。如果你也想让 AI 帮你处理重复性的信息搜集工作或者准备入门 AI Agent 开发本文将带你从零搭建一个可扩展的自动化信息监控 Agent。1. AI 每天上网的本质一个自动化 Agent先说一个容易混淆的概念这不是普通爬虫而是一个 AI Agent。普通爬虫做的事情是“抓取”把网页内容下载下来解析出标题、正文、链接然后原样保存。但面对“找项目、找钱、查风险”这类需求时爬虫远远不够因为海量信息里真正对你有用的可能只有几条而且每条信息还需要结合业务背景做判断。AI Agent 则多了一层“理解 决策 行动”的能力。它会先接收任务目标自己规划采集范围调用不同数据源获取信息再交给大模型分析最后把结果整理成结构化报告。整个过程不依赖人工干预每天按时执行。1.1 Agent 的三个核心能力拆解来看这套系统需要三个能力感知能力能够从互联网获取最新信息。这里我使用公开 API 和 RSS 作为数据源而不是爬取网页正文既稳定又合规。分析能力从原文中理解信息含义判断它跟我的目标是否相关。这里调用大模型接口完成本质上是一次文本分类和摘要生成。行动能力根据分析结果生成报告并在检测到高风险信息时发出提醒。这三个能力分别对应系统里的采集模块、分析模块和报告模块。1.2 Agent 与一次性脚本的区别你可能会说“这不就是一个定时脚本吗”从实现上看确实有相似之处但关键区别在于任务编排方式。传统脚本的执行路径是写死的抓哪个网站、提取哪些字段、输出什么格式全部在代码里硬编码。Agent 则更灵活你只需要告诉它“找风险信息”它可以自己决定去哪些源、用什么关键词组合、如何判断风险等级。这就是为什么这两年大家都在聊 Agent 工作流。当然完全自主决策在实际工程中不可控所以我在设计时采用了“半自动化”策略信息采集的范围由配置定义大模型负责分析判断最终高风险结论必须经过人工确认。这样既保留了 Agent 的智能性又控制了风险。2. 系统整体设计在写代码之前先把系统结构理清楚。整个项目我命名为ai-daily-scout通过组件解耦的方式设计方便后续扩展数据源和算法。2.1 模块划分系统分四个核心模块模块职责关键实现collector信息采集调用 GitHub API、RSS 解析器获取原始信息analyzer信息分析调用大模型对原始信息进行过滤、摘要、风险标注reporter报告生成将分析结果输出为 Markdown 日报scheduler定时调度每天定时触发上述流程这种分层的设计很容易扩展。比如后期想增加新闻源只需要在 collector 里新增一个函数想调整分析策略只需要修改 analyzer 里的 prompt。2.2 数据流设计整个流程可以概括为一条单向数据管道关键词配置 → 采集器 → 原始信息列表 → 大模型分析 → 结构化结果 → 报告生成 → 日报文件每个环节的输入都是上一个环节的输出结构清晰便于排查问题。2.3 项目目录结构这是一个单机 Python 项目目录结构如下ai-daily-scout/ ├── config.py # 全局配置关键词、数据源、模型参数 ├── collector.py # 信息采集模块 ├── analyzer.py # 大模型分析模块 ├── reporter.py # 报告生成模块 ├── scheduler.py # 定时调度模块 ├── main.py # 主流程入口 ├── requirements.txt # 依赖清单 ├── .env # 密钥配置不要提交到代码仓库 └── reports/ # 生成的日报存放目录3. 环境准备与项目初始化3.1 运行环境本文示例以 Python 3.10 为例操作系统不限Windows / macOS / Linux 均可。建议使用虚拟环境隔离依赖。主要依赖库如下requests发送 HTTP 请求beautifulsoup4解析 HTML 内容feedparser解析 RSS 订阅源openai调用 OpenAI 兼容的大模型接口schedule实现定时任务python-dotenv读取.env文件中的密钥3.2 安装依赖创建项目目录并进入mkdir ai-daily-scout cd ai-daily-scout python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate创建requirements.txtrequests2.31.0 beautifulsoup44.12.3 feedparser6.0.11 openai1.30.0 schedule1.2.1 python-dotenv1.0.1版本可以根据你的实际环境调整重点是掌握实现思路。安装pip install -r requirements.txt3.3 初始化配置在项目根目录创建.env文件存放大模型的密钥和接口地址。这里以 OpenAI 兼容接口为例如果你使用的是国内大模型服务只需要替换BASE_URL和MODEL_NAMELLM_API_KEY你的密钥 LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini同时创建config.py用于存放监控关键词、RSS 数据源等业务配置# config.py import os from dotenv import load_dotenv load_dotenv() # 大模型配置 LLM_API_KEY os.getenv(LLM_API_KEY) LLM_BASE_URL os.getenv(LLM_BASE_URL) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 监控目标 USER_GOAL 关注人工智能与开源领域的新项目机会、政策资金动态、以及相关企业风险信息 # 监控关键词用于 GitHub 项目搜索 SEARCH_KEYWORDS [AI agent, LLM application, open source] # RSS 数据源按需替换为真实数据源 RSS_SOURCES [ https://example.com/feed.xml, ] # 报告输出目录 OUTPUT_DIR reports这里的RSS_SOURCES是示例占位符实际使用时要替换成你感兴趣的合规 RSS 地址比如技术博客、新闻网站的官方订阅源。4. 核心代码实现4.1 信息采集 Agentcollector.py采集模块负责把不同来源的信息汇总成统一格式。这里我实现了两个采集函数fetch_github_repos()通过 GitHub Search API 搜索相关项目对应“找项目”场景。fetch_rss()解析 RSS 订阅源获取行业资讯对应“找信息”场景。# collector.py import requests import feedparser GITHUB_SEARCH_API https://api.github.com/search/repositories HEADERS { User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 } def fetch_github_repos(keyword: str, per_page: int 5) - list[dict]: 通过 GitHub Search API 搜索相关项目。 返回统一的条目格式title / url / description / source / type params { q: keyword, sort: updated, order: desc, per_page: per_page } resp requests.get(GITHUB_SEARCH_API, paramsparams, headersHEADERS, timeout15) resp.raise_for_status() data resp.json() items [] for repo in data.get(items, []): items.append({ title: repo.get(full_name), url: repo.get(html_url), description: repo.get(description), source: github, type: project }) return items def fetch_rss(rss_url: str, limit: int 10) - list[dict]: 解析 RSS 订阅源获取最新资讯条目。 feed feedparser.parse(rss_url) items [] for entry in feed.entries[:limit]: items.append({ title: entry.get(title), url: entry.get(link), summary: entry.get(summary, )[:300], source: rss_url, type: news }) return items关键点说明使用公开 API 而不是网页爬虫。GitHub Search API 是官方接口只要请求频率不超标基本不会被封。数据合规性也更好。统一条目格式。不管是项目还是新闻最终都转成包含title、url、source、type的字典这样下游模块不需要关心数据来源差异。设置请求超时。网络请求必须设置timeout避免程序卡死。实际场景中你还可以增加更多采集函数比如监控招标网站公告、政府政策文件页面、行业论坛热帖等只要把数据转成同样的格式即可。4.2 大模型分析 Agentanalyzer.py采集到的原始信息是碎片化的需要大模型来完成“过滤 摘要 风险判断”三个任务。我在这里采用了结构化输出方式要求模型返回 JSON 数组方便后续程序处理。# analyzer.py import json import os from openai import OpenAI # 初始化客户端兼容 OpenAI 与国内大模型服务 client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL) ) def analyze_with_llm(items: list[dict], user_goal: str) - list[dict]: 将原始信息交给大模型分析。 返回结构化列表每条包含 title / url / summary / risk_level / risk_reason # 控制传给模型的文本长度避免超限 text \n.join( [f- 标题{item[title]}\n 链接{item[url]}\n 描述{item.get(description) or item.get(summary, )} for item in items] )[:4000] prompt f你是我的信息分析助手我关注的目标是「{user_goal}」 下面是今天采集到的原始信息请帮我完成三件事 1. 筛选出与我的目标最相关的 5 条 2. 对每条信息写一句简洁摘要说清楚它和我目标的关系 3. 判断这条信息是否存在风险风险等级只能是 low / medium / high并给出简短风险原因。 按 JSON 数组格式返回不要输出其他解释 [ {{ title: 标题, url: 链接, summary: 一句话摘要, risk_level: low/medium/high, risk_reason: 风险原因或说明 }} ] 原始信息列表 {text} resp client.chat.completions.create( modelos.getenv(LLM_MODEL, gpt-4o-mini), messages[{role: user, content: prompt}], temperature0.3 ) content resp.choices[0].message.content.strip() # 清理大模型可能输出的代码块标记 if content.startswith(json): content content[7:] if content.endswith(): content content[:-3] content content.strip() return json.loads(content)这里有几个设计细节值得注意Prompt 里明确输出格式。大模型输出不稳定最常见的问题就是格式漂移所以我在 prompt 末尾指定了 JSON 结构并在代码里做了 json 标记清理。限制输入长度。把原始信息截断到 4000 字符以内一方面控制 token 成本另一方面防止上下文过长导致模型注意力分散。temperature 设置为 0.3。分析类任务需要稳定输出温度越低结果越确定不会出现“这次说风险高、下次说风险低”的随机情况。大模型返回的内容可以直接转成 Python 对象后续报告模块就能直接使用。4.3 风险识别与报告生成reporter.py分析模块返回的是带风险等级的结构化数据但用户直接看 JSON 太不友好所以需要一个报告生成模块。这里我选择生成 Markdown 格式日报既可以本地阅读也可以上传到语雀、Notion 等笔记工具。# reporter.py from datetime import datetime def generate_markdown_report(analyzed_items: list[dict], keywords: list[str]) - str: 将分析结果生成 Markdown 日报 now datetime.now().strftime(%Y-%m-%d %H:%M) lines [] lines.append(f# AI 每日信息日报 - {now}) lines.append() lines.append(f 监控关键词{, .join(keywords)}) lines.append() # 风险预警区 lines.append(## 风险预警) high_risk [item for item in analyzed_items if item.get(risk_level) high] if high_risk: for item in high_risk: lines.append(f- **{item[title]}**{item.get(risk_reason, )}) else: lines.append(- 今日无高风险预警) lines.append() # 项目动态区 lines.append(## 重点信息) for item in analyzed_items: risk_tag { high: , medium: , low: }.get(item.get(risk_level, low), ) lines.append(f- {risk_tag} [{item[title]}]({item[url]}){item.get(summary, )}) if item.get(risk_reason): lines.append(f - 风险说明{item[risk_reason]}) return \n.join(lines)为什么单独做一个报告模块因为 Agent 的“产出”不应该是大模型的原始回答而应该是经过整理的、可直接阅读的成果物。抽离成独立函数后后续可以轻松把日报改成 CSV、PDF 或者直接推送到钉钉、飞书群。4.4 主流程串联main.py主流程负责把所有模块串起来。这一步很关键Agent 不是各个函数的散装组合而是由主流程编排的完整流水线。# main.py import os from datetime import datetime from collector import fetch_github_repos, fetch_rss from analyzer import analyze_with_llm from reporter import generate_markdown_report from config import SEARCH_KEYWORDS, RSS_SOURCES, USER_GOAL, OUTPUT_DIR def run_daily_scan(): 执行一次完整的信息采集与分析流程 all_items [] # 1. 采集按关键词搜索 GitHub 项目 for kw in SEARCH_KEYWORDS: try: repos fetch_github_repos(kw, per_page5) all_items.extend(repos) print(f[collector] 关键词「{kw}」采集到 {len(repos)} 个项目) except Exception as e: print(f[collector] 关键词「{kw}」采集失败{e}) # 2. 采集解析 RSS 源 for rss_url in RSS_SOURCES: try: news fetch_rss(rss_url, limit10) all_items.extend(news) print(f[collector] RSS {rss_url} 采集到 {len(news)} 条资讯) except Exception as e: print(f[collector] RSS {rss_url} 采集失败{e}) # 3. 去重接根据 URL 去重 seen_urls set() unique_items [] for item in all_items: url item.get(url) if url and url not in seen_urls: seen_urls.add(url) unique_items.append(item) print(f[collector] 去重后共 {len(unique_items)} 条有效信息) if not unique_items: print([analyzer] 没有可分析的信息流程结束) return # 4. 分析交给大模型 analyzed_items analyze_with_llm(unique_items, USER_GOAL) print(f[analyzer] 大模型筛选出 {len(analyzed_items)} 条重点信息) # 5. 生成报告 report generate_markdown_report(analyzed_items, SEARCH_KEYWORDS) # 6. 保存报告 os.makedirs(OUTPUT_DIR, exist_okTrue) output_path os.path.join(OUTPUT_DIR, freport_{datetime.now().strftime(%Y%m%d)}.md) with open(output_path, w, encodingutf-8) as f: f.write(report) print(f[reporter] 报告已生成{output_path}) if __name__ __main__: run_daily_scan()这个主流程看起来简单但已经包含了 Agent 工作流的基本骨架采集 → 清洗 → 分析 → 产出。每一步都打印日志方便观察任务执行到哪个环节、是否出错。5. 定时调度让 Agent 自动运行主流程写完之后还需要一个“每天自动启动”的机制。这里我用schedule库实现定时调度。5.1 使用 schedule 实现每日任务创建一个scheduler.py# scheduler.py import schedule import time from main import run_daily_scan def start_scheduler(): # 每天早上 9 点执行一次 schedule.every().day.at(09:00).do(run_daily_scan) print(定时任务已启动每天 09:00 自动执行信息采集与分析) while True: schedule.run_pending() time.sleep(30) if __name__ __main__: start_scheduler()这段代码很简单解决的是“定时启动”的问题。每次循环休眠 30 秒检查是否有到期任务需要执行。这个方案适合个人电脑上长期运行。如果是部署在服务器上更推荐使用 crontab 或 systemd timer避免 Python 进程异常退出后没有兜底重启。5.2 手动运行验证在定时任务跑起来之前先手动执行一遍主流程python main.py如果一切正常会输出类似下面的日志[collector] 关键词「AI agent」采集到 5 个项目 [collector] 关键词「LLM application」采集到 5 个项目 [collector] 关键词「open source」采集到 5 个项目 [collector] RSS https://example.com/feed.xml 采集失败... [collector] 去重后共 15 条有效信息 [analyzer] 大模型筛选出 5 条重点信息 [reporter] 报告已生成reports/report_20241218.md这里 RSS 采集失败是正常的因为示例地址不可用。实际配置里替换成真实 RSS 源即可。5.3 一次运行结果示例生成的日报长这样# AI 每日信息日报 - 2024-12-18 09:02 监控关键词AI agent, LLM application, open source ## 风险预警 - [some/malicious-repo](https://github.com/some/malicious-repo)该项目声称提供免费 API但要求关闭本地防火墙存在安全风险。 ## 重点信息 - [mewamew/my_ai_town](https://github.com/mewamew/my_ai_town)开源 AI 小镇模拟项目提供 Mac 版游戏下载适合作为 Agent 行为实验环境。 - [open-source-llm-tracker](https://github.com/...)持续跟踪最新开源大模型榜单。 - [company-x-ai-product](https://example.com/news)某公司发布 AI 新品但公告未披露模型训练数据来源。这里的my_ai_town是真实的开源项目链接它是一个 AI 小镇模拟环境可以作为 Agent 行为的沙盒来学习和调试。6. 从示例到落地找项目、找钱、查风险的真实场景前面演示的是一个通用框架下面具体拆解如何把它应用到“找项目、找钱、查风险”三个实际场景。6.1 找项目开源与招标信息监控“找项目”在工程领域可以理解成“找可参与的项目机会”包括开源协作项目、招标公告、合作需求。对于开源项目GitHub Search API 是首选数据源。你可以在SEARCH_KEYWORDS里配置你关心的技术栈关键词比如langchain、rag、ai agent系统会每日搜索最新更新的仓库。对于招标公告很多政府采购网站没有开放 API这时可以用 RSS 或者页面探测方式。但需要注意必须先阅读目标网站的robots.txt确认抓取行为是否被允许。请求频率要控制建议单次任务最多抓取几个页面并添加延迟。优先使用官方网站提供的公告接口。6.2 找钱政策资金与行业动态监控“找钱”指的是挖掘商业机会信息包括政策补贴、融资新闻、人才需求、行业报告等。你的数据源设计可以参考这些政府政策文件 RSS部分地区政府网站提供政策文件 RSS订阅后可以第一时间获取资金申报、产业扶持文件。投资机构官网新闻页很多机构会发布投资动态、行业观察这些都是高质量的商业机会信号。招聘网站技术岗位列表大厂高薪岗位数量变化某种程度上反映了技术方向的投入力度这个信号也能帮助判断机会。把这些 RSS 地址加入RSS_SOURCES大模型的 prompt 里加入“是否包含商业机会、政策利好、融资动态”它就会自动帮你筛选出有价值的条目。6.3 查风险舆情与合规预警风险监控是这套系统最有价值的地方。你可以监控以下几类信息合作方动态对方是否出现负面新闻、法律诉讼、高管变动。行业风险监管政策变化、技术安全事故。项目风险开源项目是否停止维护、是否出现安全事故通报。风险判断不应只靠大模型“感觉”建议在 prompt 里明确风险标准例如风险判断标准 - high涉及法律诉讼、数据泄露、资金链断裂、产品停运等明确负面事件 - medium出现可能间接影响业务的不确定信息 - low正常动态无风险信号同时设置一条铁律AI 标记为高风险的信息必须人工复核后才能对外反馈或执行下一步操作。不要让 Agent 在没有监督的情况下把一个“疑似风险”直接发到工作群或者触发自动处置流程。6.4 与 AI 小镇等模拟环境的结合现在很多开源项目开始用游戏化的沙盒环境来训练 Agent比如 GitHub 上的my_ai_town项目。它提供了类似《模拟小镇》的环境你可以让 AI 扮演不同角色在虚拟环境里做任务观察 Agent 的行为是否合理。我建议的做法是先在模拟环境里验证 Agent 的任务规划能力等逻辑成熟后再把它接到真实互联网数据源上。因为真实互联网信息噪音大、格式乱、还有不合规内容如果 Agent 的基础行为策略没有调好一上来接真实数据会被各种异常打乱。7. 常见问题与排查思路开发过程中难免遇到各种问题下面整理几个高频场景。7.1 大模型返回 JSON 解析失败现象analyze_with_llm报JSONDecodeError。原因大模型有时会在 JSON 前后添加解释性文字或者输出里的引号、逗号不合法。排查步骤把resp.choices[0].message.content打印出来肉眼确认格式。确认removeprefix(json)清理逻辑是否覆盖所有情况。查看报错时的 JSON 片段锁定是哪个字段格式问题。解决方案在 prompt 里强调“只输出 JSON不要解释”并且把temperature调低到 0.3 以下。如果还不行可以引入json_repair库做容错。7.2 请求被拒绝或返回 403现象GitHub API 返回403或者 RSS 请求超时。原因请求频率过高、未设置 User-Agent、IP 被限流。排查步骤检查headers是否携带了合理的User-Agent。GitHub API 匿名请求有速率限制关注响应头里的X-RateLimit-Remaining字段。确认目标平台是否允许程序访问。解决方案降低抓取频率增加请求间隔优先使用官方 API为 API 请求配置密钥。7.3 定时任务不触发现象python scheduler.py启动后到了指定时间没有执行。原因进程可能在后台被杀死或者系统休眠schedule 库无法在休眠期间补执行任务。排查步骤先用python main.py手动执行确认主流程正常。查看 scheduler 是否打印了“定时任务已启动”。确认电脑在 9 点没有处于休眠状态。解决方案服务器环境使用 cron 替代在脚本开头加print日志方便定位。7.4 Agent 幻觉导致误判现象大模型在风险分析时把完全正常的信息判断为高风险或者凭空编造风险原因。原因大模型存在幻觉尤其是 prompt 语义模糊时模型会“脑补”不存在的细节。解决方案给风险等级增加明确的判断标准不要用模糊词汇。请求模型输出时必须引用原始信息中的证据禁止无依据判断。在报告里标记“AI 生成内容仅供参考需人工复核”。7.5 合规风险现象系统采集了不该采集的内容或者使用了未授权的数据源。原因对数据源规则不熟悉。解决方案所有数据采集必须遵守目标网站的用户协议和 robots 协议优先使用官方 API不在本地保存敏感个人信息涉及对外发布的信息必须人工审核。8. 工程化与部署建议代码能跑起来之后离“能长期稳定运行”还有一段距离。下面这些工程化建议值得收藏。8.1 日志与可观测性现在的 print 日志只能应付开发阶段长期运行建议使用logging模块按级别输出import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[ logging.FileHandler(agent.log, encodingutf-8), logging.StreamHandler() ] )有了日志文件即使某天任务异常也能快速定位是哪一步出了问题。8.2 密钥与配置管理这篇文章里密钥放在.env文件中这是一个很好的起点。但请注意.env文件不要提交到 Git 仓库建议在.gitignore中忽略。项目部署到服务器时可以使用环境变量或密钥管理服务注入不要把明文密钥写死在代码里。如果密钥泄露立即在云端控制台吊销并重新生成。8.3 成本控制大模型 API 调用不是免费的每日运行会积累成本。控制成本可以从几个方面先过滤再分析。优先用关键词或规则过滤掉明显不相关的信息再交给大模型减少传入 token。降低调用频率。如果信息变化不快可以改成每周两次而不是每天。选择更小的模型。不是所有任务都需要最强模型摘要和分类用轻量模型完全够用。设置调用上限。在代码里记录每日 token 消耗接近预算时自动停止。8.4 安全边界一定要明确 Agent 的权限边界。本文的系统只是“采集信息 生成报告”没有任何对外操作权限所以即使分析结果有误影响也可控。如果后续要扩展成自动发邮件、自动发布文章、甚至自动交易务必加上以下机制人工审批节点高风险动作必须人工确认。操作白名单Agent 只能调用白名单内的接口。全量操作日志记录 Agent 每次执行的输入、输出和动作便于回溯。9. 最后说几句回过头来看这个 AI Agent 系统的核心价值不是“替代你上网”而是把每天重复的信息搜集和初筛工作标准化、自动化。你依然需要用自己的判断力处理最终结果但节省下来的时间非常可观。建议你从最小闭环开始先配置两三个关键词接一个 RSS 源手动跑通主流程再逐步增加数据源和定时任务。等基本功能稳定后可以尝试接入企业微信或飞书机器人把日报推送到群里效果会更好。如果你在自己的实践过程中遇到了有意思的问题欢迎在评论区交流。觉得这篇文章有启发的话可以收藏备用也可以转发给身边在做 AI 应用开发的朋友。