每天早上九点,你打开浏览器,先刷一遍招标平台看有没有新的采购项目,再去融资公告页面查一圈钱流向了哪家公司,最后还要搜一遍合作方的负面舆情。这个流程,我猜不少做企业服务、做FA、做投后管理的人都不陌生。它极其重要,又极其消耗精力。更难受的是,真正的机会和风险不会等你刷完网页才出现,它们可能在凌晨两点更新,或者在某个子页面里安静地挂了三天。
大模型出现后,很多人第一时间做的是“聊天机器人”,让 AI 陪你聊天、帮你写文案。但在我看来,这个方向对普通开发者来说并不算最优解。真正值得先做的,是一个看起来没那么性感、却能在每个工作日持续产生价值的 AI Agent:让 AI 每天自己上网,按照你预先设定的任务清单,去公开网站找项目线索、找资金动态、查合作风险,然后把结构化结果推到你面前。
这篇文章会讲清楚这个 Agent 的设计思路,并给出一套可以直接运行的 Python 原型代码。读完你会得到一个“定时任务 + 网页采集 + LLM 解析 + 去重存储 + 消息推送”的完整系统。这个系统不依赖复杂的 Agent 框架,也没有高不可攀的部署要求,一台普通电脑或者一个小型云服务器就能跑起来。更重要的是,我会把其中的坑一起讲透,包括页面抓不到、LLM 幻觉、重复推送、接口超时这些真实工程问题。
1. 为什么“每天自动上网”是 AI Agent 最有价值的落地场景
如果你仔细观察过当前大模型应用,会发现大多数产品的形态是“你问它答”。这种形态有一个天然限制:AI 不会主动开始工作。你不动,它就不动。但真实世界里的信息是主动更新的,招标公告不会等你提问才发布,一家公司的经营异常也不会因为你不知道而消失。
“定时自动上网”恰恰把这个限制打破了。它的本质是 Trigger-Action 模式:时间触发,Agent 执行采集、解析、判断、通知一系列动作。在这个模式里,LLM 不再是对话对象,而是整个流水线里的“语义理解引擎”,专门负责把网页里杂乱的文本转换成结构化数据。
我总结了三个最适合这类 Agent 先落地的场景:
找项目。很多做外包、做解决方案、做 SaaS 的公司,每天都在盯招标网站、政府采购平台、大企业采购页面。这些页面数据量大、更新快,而且不同平台页面结构差异很大。用固定的爬虫脚本去抓,任何一家网站改版就会让代码失效;用人力去盯,又养不起。这个场景天然适合“采集全量 + LLM 初筛 + 人工终审”。
找钱。这里的“钱”是指公开的资金线索。比如一家公司刚完成了新一轮融资,说明这条赛道正在吸纳资本;某个园区发布了补贴申报指南,说明有一批中小企业符合条件;某个平台挂出了预算充足的采购公告,说明这可能是你的下一个客户。这些信息散落在几十个网站里,靠人肉监控很难不漏。AI Agent 可以做的是每天定时抓取这类新闻,抽取公司名、金额、行业、时间,汇总成一张资金动态表。
查风险。这是三种场景里价值最高、也最需要谨慎的一种。合作方的经营异常、法律诉讼、负面舆情,直接决定要不要推进一笔生意。AI Agent 可以定时去公开的企业信用信息平台、法院公告网、新闻聚合站点采集指定公司的新增风险信号。但这里要特别强调:AI 只能做“初筛”,不能代替专业人士的最终判断,因为 LLM 天生存在幻觉风险,它可能会把不存在的风险事件编出来,这一点在后面的章节里我会专门展开。
这三个场景有一个共同特点:结果不是开放式的,而是明确的结构化信号。招标公告最终就是一个项目名称加预算金额,融资动态最终就是一家公司加一个轮次,风险信号最终就是一家企业加一种异常状态。这种“输入明确、任务明确、输出明确”的场景,正是 Agent 最容易做稳、做准、做出价值的地方。
2. AI Agent 的核心概念与整体架构设计
开始写代码之前,先把架构理清楚。很多 AI 初学者一上来就引入 LangChain、AutoGPT 这类重磅框架,结果是为了写一个每天抓网页的定时脚本,徒增了一堆抽象概念。我更推荐从最朴素的分层结构开始,等到真正需要多工具调度、多轮推理时再升级框架。
这个项目的整体架构分成五层:
| 层级 | 职责 | 对应模块 |
|---|---|---|
| 数据源层 | 配置要监控的网站和类型 | sources.yaml |
| 采集层 | 抓取网页可见文本 | collector.py |
| 解析层 | 用 LLM 抽取结构化信息 | parser.py、validator.py |
| 存储层 | 去重、落库、留档 | storage.py |
| 通知层 | 推送到企业微信/钉钉/邮件 | notifier.py |
整个数据流是单向的:定时器唤醒采集器,采集器把网页正文喂给 LLM 解析器,解析器返回 JSON 列表,存储层通过指纹去重,最后把新增条目推送到 IM 群。这个流程里没有复杂的 Agent 推理,也没有动态调用工具。用 Agent 的定义来说,它是一个相对“薄”的系统——任务编排靠调度器,理解能力靠大模型,稳定性靠工程兜底。
技术选型上,我对比过三种路线:
Python + Playwright + OpenAI SDK。本文采用的方式。代码量小、依赖少、调试直观,适合独立开发者和小团队。抓取动态网页用 Playwright,LLM 调用直接用 OpenAI 兼容接口,没有多余的封装层。
LangChain。如果你已经有 LangChain 生态的使用经验,用它做提示词模板和输出解析确实方便。但对于这个项目来说,LangChain 的抽象反而会掩盖底层的调用逻辑。一旦接口报错或者返回格式不稳,排查问题的链条会变长。
Spring AI。如果你的团队是纯 Java 技术栈,Spring AI 是目前比较值得关注的方案。它把模型调用封装成了类似 RestClient 的风格,可以比较平滑地嵌进现有 Spring Boot 业务系统里。但如果你只是要跑一个独立的定时 Agent,用 Java 写的成本明显高于 Python。
架构设计上最重要的一件事是:让 LLM 只做它擅长的事,也就是理解语义和抽取字段,不要让它做需要精确操作的事情。去重、定时、推送这些步骤全部用确定性代码完成。很多 Agent 项目翻车,就是因为把调度和去重的逻辑也交给了模型,结果模型在要不要重复采集这件事上“思考”出了千奇百怪的结果。
3. 环境准备与项目初始化
在实际动手之前,先把运行环境准备好。这个项目对操作系统没有太强要求,Windows、macOS、Linux 都可以。以下命令以 macOS / Linux 为例,Windows 下的区别只在激活虚拟环境的命令上。
mkdir ai-daily-agent cd ai-daily-agent python3 -m venv .venv source .venv/bin/activatePython 版本建议使用 3.10 及以上。然后安装依赖:
pip install playwright openai apscheduler pyyaml requests playwright install chromium这里需要解释一下为什么要装 Playwright。很多人写爬虫首选 requests 加 BeautifulSoup,这在抓静态页面时是够用的。但现实中的招标平台、融资公告页面大多数是 React、Vue 这类前端框架渲染出来的,直接用 requests 拿到的 HTML 可能是空的,核心数据要通过异步请求才能加载出来。Playwright 会启动一个无头浏览器,像真实用户一样执行页面脚本,我们就能通过page.inner_text("body")拿到渲染后的可见文本。
安装完成后,设置环境变量。这里用的是 OpenAI 兼容接口,方便接入 DeepSeek、通义、Moonshot 等国内可直接访问的大模型 API:
export LLM_API_KEY="你的 API Key" export LLM_BASE_URL="https://api.deepseek.com/v1" export LLM_MODEL="deepseek-chat" export NOTIFY_WEBHOOK="你的企业微信机器人 Webhook"如果你用的是 OpenAI 官方接口,把LLM_BASE_URL和LLM_MODEL换成你自己的服务即可。版本方面的细节请以你选择的模型文档为准,本文的重点是通用实现思路。
环境准备好之后,项目结构如下:
ai-daily-agent/ ├── .venv/ ├── sources.yaml ├── config.py ├── collector.py ├── parser.py ├── validator.py ├── storage.py ├── notifier.py └── main.py4. 数据源配置与信息采集模块
我先创建一个数据源配置文件。这个文件的格式特意用 YAML,方便非技术人员以后自己加网址,不用看代码。
# sources.yaml sources: projects: - name: "示例招标平台" url: "https://example.com/procurement" type: "bidding" enabled: true funding: - name: "示例融资公告" url: "https://example.com/funding" type: "funding" enabled: true risk: - name: "示例企业舆情监控" url: "https://example.com/news" type: "risk" enabled: true注意:实际项目里请把example.com替换成你真正要监控的公开页面,并且确保该网站允许爬虫访问。下面这个函数负责读取配置并打成 list:
# config.py import os import yaml LLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1") LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-chat") NOTIFY_WEBHOOK = os.getenv("NOTIFY_WEBHOOK", "") if not LLM_API_KEY: raise ValueError("请先设置 LLM_API_KEY 环境变量") def load_sources(path: str = "sources.yaml") -> list[dict]: with open(path, "r", encoding="utf-8") as f: raw = yaml.safe_load(f) or {} result = [] for group, items in raw.get("sources", {}).items(): for item in items: if item.get("enabled", True): result.append({ "name": item.get("name", group), "url": item["url"], "type": item.get("type", group), }) return result接下来是采集模块。核心逻辑很简单:用 Playwright 打开页面,等待核心内容加载,提取 body 的可见文本,最后交给解析层。
# collector.py import asyncio from playwright.async_api import ( async_playwright, TimeoutError as PlaywrightTimeoutError, ) USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" ) async def fetch_text(url: str, wait_ms: int = 2000) -> str: """ 抓取页面的可见文本。 参数 wait_ms 控制等待时间,给 JavaScript 渲染留出缓冲。 """ async with async_playwright() as p: browser = await p.chromium.launch(headless=True) ctx = await browser.new_context(user_agent=USER_AGENT, locale="zh-CN") page = await ctx.new_page() try: await page.goto(url, timeout=30000, wait_until="domcontentloaded") await page.wait_for_timeout(wait_ms) return await page.inner_text("body") except PlaywrightTimeoutError: print(f"[timeout] {url}") return "" finally: await browser.close()这里有三个工程细节值得展开。
第一,wait_until="domcontentloaded"表示最多等页面 DOM 解析完成就继续,不是等所有图片和广告加载完,速度会快很多。但这个时机不够保险,所以后面加了一个wait_for_timeout(2000),让前端框架有足够时间渲染内容。如果你的目标网站比较慢,可以把这个值调到 3000 或 5000。
第二,locale="zh-CN"会影响页面返回内容的语言。某些网站会根据浏览器的语言偏好展示不同的文案,如果你发现抓回来的文字是英文或者繁体,可以优先检查这个参数。
第三,这里抓的是body的可见文本,不是原始 HTML。原因是可见文本对 LLM 更友好,没有标签噪音,也大幅减少 token 消耗。如果你想抓结构化字段,比如表格里的金额、日期,可以在同一页面上用 Playwright 的locator精准提取,再用 LLM 做二次校验。对大多数 MVP 场景来说,body 文本已经足够。
在写采集层时,请务必把合规放在心上。只采集允许公开访问的页面,尊重网站的 robots.txt,控制抓取频率,不要绕过登录验证码,也不要对同一站点做高并发请求。这些不是教条,而是实际运行中避免 IP 被封、避免法律风险的基本操作。
5. 基于 LLM 的信息抽取与幻觉控制
采集到网页文本后,核心任务是把非结构化文本变成结构化 JSON。这一步如果不用 LLM,传统方案是给每个网站写一套 XPath 规则,采集规则的维护成本非常惊人。而用 LLM 以后,大多数情况下你只需要在提示词里描述“这是哪类来源”,模型就能完成任务。
先看解析模块:
# parser.py import json from openai import OpenAI from config import LLM_API_KEY, LLM_BASE_URL, LLM_MODEL client = OpenAI(api_key=LLM_API_KEY, base_url=LLM_BASE_URL) EXTRACT_PROMPT = """你是商业情报助理。请从网页文本中提取有价值的信息条目。 来源说明: - bidding:招标、中标、采购公告 - funding:融资、投资、补贴政策 - risk:企业负面、法律诉讼、经营异常 输出要求: 1. 只输出 JSON,格式为 {"items": [{...}]} 2. 每条 item 包含:title, company, summary, source_type, risk_level, date, url 3. risk_level 只允许取 high / medium / low 4. 如果没有值得关注的条目,输出 {"items": []} 5. 不要编造原文中不存在的信息 网页文本: {text} """ def parse_items(text: str, source_type: str) -> list[dict]: if not text.strip(): return [] prompt = EXTRACT_PROMPT.format(text=text[:8000]) resp = client.chat.completions.create( model=LLM_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.1, ) content = resp.choices[0].message.content or "" items = _safe_parse_json(content) for item in items: item["source_type"] = source_type return items def _safe_parse_json(content: str) -> list[dict]: try: data = json.loads(content) return data.get("items", []) if isinstance(data, dict) else [] except json.JSONDecodeError: # 模型偶尔会输出 Markdown 代码块,剥离后再解析一次 try: start = content.find("{") end = content.rfind("}") data = json.loads(content[start:end + 1]) return data.get("items", []) except Exception: return []提示词里最重要的两句话是“只输出 JSON”和“不要编造原文中不存在的信息”。前者保证下游解析的稳定性,后者是抑制幻觉的第一道防线。
关于幻觉,这是整个系统里最需要认真对待的问题。LLM 在抽取信息时,完全有可能把一家没出现过的公司名“联想”出来,或者把文章里一条中性的表述总结成负面风险。在“查风险”场景里,这种幻觉可能会让你误判一个合作方,导致真实的业务损失。
所以我在解析层之后加了一个独立的校验模块。它的思路不是用另一个模型来检查,而是用成本极低的确定性规则做粗筛:
# validator.py def validate_item(item: dict, raw_text: str) -> bool: """ 粗略校验 LLM 抽取结果是否在原文中有依据。 只要抽取到的公司名出现在原文中,就认为基本可信。 """ company = (item.get("company") or "").strip() title = (item.get("title") or "").strip() if not company and not title: return False if company and company in raw_text: return True if title and title in raw_text: return True return False这个校验非常宽松,我在实际使用中甚至考虑过更严格的方案:把title和company拆成关键词,逐个确认是否出现在原文中。但关键词匹配在中文语境下容易误伤,因为 LLM 抽取出来的title往往是归纳句,不会是原文原句。所以最终采用了“公司名必须在原文出现”这一个核心规则,再配合“高风险条目人工复核”的流程。
另外,我在提示词里把temperature设置成 0.1。对抽取任务来说,创造性越低越好。如果后续发现模型抽取结果总是不稳定,可以先检查这个参数是不是被改高了。
6. 定时调度、去重存储与通知推送
Agent 的核心价值是“每天自动”,所以定时调度是必不可少的环节。这里我选用 APScheduler,相比 Windows 计划任务或者系统 crontab,它的优势是配置简单、跨平台、还能在进程内管理任务。
先看存储模块。存储的目的是两件事:一是去重,防止同一则公告被反复推送;二是留档,方便后来检索和分析。
# storage.py import sqlite3 import hashlib from datetime import datetime DB_PATH = "agent_data.db" def init_db() -> None: conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS items ( fingerprint TEXT PRIMARY KEY, title TEXT, company TEXT, summary TEXT, source_type TEXT, risk_level TEXT, source_url TEXT, created_at TEXT ) """) conn.commit() conn.close() def save_unique_item(item: dict) -> bool: """ 返回 True 表示是新增条目,返回 False 表示已经存在。 指纹由 title 和 company 拼接后取 md5 得到。 """ fp = hashlib.md5( f"{item.get('title', '')}|{item.get('company', '')}".encode() ).hexdigest() conn = sqlite3.connect(DB_PATH) try: exists = conn.execute( "SELECT 1 FROM items WHERE fingerprint = ?", (fp,) ).fetchone() if exists: return False conn.execute( """ INSERT INTO items (fingerprint, title, company, summary, source_type, risk_level, source_url, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( fp, item.get("title", ""), item.get("company", ""), item.get("summary", ""), item.get("source_type", ""), item.get("risk_level", "low"), item.get("url", ""), datetime.now().isoformat(), ), ) conn.commit() return True finally: conn.close()去重的关键点是设计好指纹。用title + company的 md5 值作为主键,已经能覆盖绝大多数重复场景。同一家公司的同一条公告,LLM 两次抽取出来的title基本一致;即使有细微差别,company也足够兜底。这里不推荐用全文相似度去重,因为在数据量不大时,精确匹配已经足够,而且简单可靠。
再来看通知模块。我选择企业微信群机器人,因为它的 webhook 配置简单,不需要申请额外的应用权限,而且群消息天然适合团队协作场景。
# notifier.py import requests def send_wecom_webhook(webhook_url: str, message: str) -> None: if not webhook_url: return payload = { "msgtype": "text", "text": {"content": message}, } resp = requests.post(webhook_url, json=payload, timeout=10) resp.raise_for_status()企业微信机器人文本消息的长度上限约为 2048 字节,所以推送时要把条目截断控制住,不要一股脑把 50 条新结果全部塞进去。
7. 完整示例:把任务串起来跑通
现在把所有模块串进主程序。下面是完整的main.py,你可以直接复制到项目里运行。
# main.py import asyncio from apscheduler.schedulers.blocking import BlockingScheduler from collector import fetch_text from config import NOTIFY_WEBHOOK, load_sources from notifier import send_wecom_webhook from parser import parse_items from storage import init_db, save_unique_item from validator import validate_item def build_notify_text(source_name: str, items: list[dict]) -> str: lines = [f"[{source_name}] 今日新增 {len(items)} 条"] for item in items[:10]: title = item.get("title", "") company = item.get("company", "") level = item.get("risk_level", "low") lines.append(f"- {title} / {company} / {level}") return "\n".join(lines) def run_task(source: dict) -> None: url = source["url"] source_type = source["type"] print(f"[start] {source['name']} -> {url}") text = asyncio.run(fetch_text(url)) items = parse_items(text, source_type) new_items = [] for item in items: # 幻觉粗筛:公司名或标题必须在原文中出现过 if not validate_item(item, text): print(f"[filter] 未通过校验: {item.get('title')}") continue if save_unique_item(item): new_items.append(item) print(f"[done] {source['name']} 新增 {len(new_items)} 条") if new_items: msg = build_notify_text(source["name"], new_items) send_wecom_webhook(NOTIFY_WEBHOOK, msg) if __name__ == "__main__": init_db() sources = load_sources() scheduler = BlockingScheduler(timezone="Asia/Shanghai") for source in sources: scheduler.add_job( run_task, "cron", hour=9, minute=0, args=[source], ) print("AI Agent 已启动,每天 09:00 自动执行") scheduler.start()主程序里几个值得注意的点:
- 用
asyncio.run()包住 Playwright 的异步函数。这是因为 APScheduler 的 job 回调是同步的,而 Playwright 是异步 API,需要手动桥接一次。 - 先校验,再去重。校验通过后落库,既保证质量又保证幂等。
- 每个数据源独立注册任务,即使某个源报错,也不影响其他源的执行。这里的实际项目建议再包一层 try/except,避免单个源异常导致整个进程退出。
启动它非常简单:
python main.py运行日志应该类似这样:
AI Agent 已启动,每天 09:00 自动执行 [start] 示例招标平台 -> https://example.com/procurement [done] 示例招标平台 新增 3 条 [start] 示例融资公告 -> https://example.com/funding [done] 示例融资公告 新增 0 条 [start] 示例企业舆情监控 -> https://example.com/news [done] 示例企业舆情监控 新增 1 条8. 运行效果验证与常见问题排查
验证这个系统是否正常,我习惯按三个层次来检查。
第一层,手动触发一次任务。不要直接等定时器。最简单的方式是在main.py里临时加一行run_task(sources[0]),或者直接用 Python REPL 调用各个函数,确认每个环节都能跑通。
第二层,检查数据库。用 SQLite 命令行工具或者可视化工具查询items表:
SELECT title, company, source_type, risk_level, created_at FROM items ORDER BY created_at DESC LIMIT 20;如果表里有数据,说明采集和解析链路是通的。如果没有,优先检查fetch_text返回的文本是不是空的。
第三层,检查通知是否到达。企业微信 webhook 推送成功后,群里应该能收到消息。如果收不到,先查看notifier.py里的resp.raise_for_status()是否抛了异常,再看 webhook 是否被企业微信的 IP 白名单限制。
下面这个表格总结了实际运行中最容易遇到的几类问题:
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 任务不执行 | APScheduler 时区配置错误或进程退出 | 查看启动日志,确认 scheduler.start() 后的进程是否存活 | 设置 timezone="Asia/Shanghai",用 nohup 或 systemd 守护进程 |
| 抓到的文本为空 | 页面加载速度慢,或者 requests 无法执行 JS | 打印 fetch_text 返回的文本首尾各 100 字符 | 调大 wait_for_timeout,确认使用 Playwright 而不是 requests |
| LLM 返回内容无法解析 | 输出带 Markdown 代码块,或返回非 JSON | 打印 resp 的原始 content | 使用 _safe_parse_json 剥离代码块,检查提示词是否被截断 |
| 重复推送同一条信息 | 指纹生成规则覆盖不到内容变体 | 查询 items 表确认指纹是否一致 | 在指纹中加入 source_url,或对 title 做归一化 |
| 抓到大量无关内容 | 页面包含导航栏、推荐位、广告文本 | 观察 body 文本,确认噪音占比 | 用 Playwright locator 只抓核心区域,或在前端页面指定 main 节点 |
| 模型编造公司名 | LLM 幻觉 | 开启 validator 的日志,查看过滤条数 | 加入原文关键词校验,高风险条目人工复核 |
| 接口超时或限流 | API 服务不稳定,或并发请求过多 | 开启 requests 日志,观察状态码 | 每次请求间隔 1-2 秒,失败后指数退避重试 |
在这些问题里,最容易踩坑的是“LLM 返回内容无法解析”。即使提示词里明确要求“只输出 JSON”,有些模型偶尔还是会带上 Markdown 的 ```json 包裹。这也是我在_safe_parse_json里做二次剥离的原因。实际生产环境里,建议把每次调用的原始返回内容落一次日志,防止出问题时没有任何排查依据。
还有一个容易被忽视的点是时间戳问题。APScheduler 默认使用系统的本地时区,但不少云服务器的默认时区是 UTC。如果发现任务总是在北京时间下午五点左右执行,而不是早上九点,第一件事就是检查时区配置。
9. 工程实践与合规建议
从“能跑的脚本”到“能长期稳定运行的 Agent”,中间还有一段工程化距离。以下是我在做这类系统时的一些实践经验。
第一,把 Agent 当成一个长期运行的服务来设计,而不是一次性脚本。定时任务进程崩了怎么办?最直接的办法是在前面加nohup python main.py >> agent.log 2>&1 &,让它后台跑并输出日志。更稳妥的方案是用 systemd 维护一个 service,设置 Restart=always,进程异常退出后自动拉起。对大多数项目来说,维护一个 systemd 服务比引入容器编排简单得多。
第二,对 LLM 调用做异常兜底。大模型 API 的稳定性并不总让人放心。网络抖动、限流、服务端过载都可能发生。实际项目里,每次调用要加超时时间、重试机制和熔断逻辑。OpenAI 兼容 SDK 一般自带重试,但超时时间建议自己显式设置。比如:
client = OpenAI( api_key=LLM_API_KEY, base_url=LLM_BASE_URL, timeout=60.0, max_retries=2, )第三,日志越完整,事故恢复越快。至少记录每个数据源的开始时间、结束时间、抓取字符数、抽取条目数、新增条目数、校验过滤数。有了这些指标,你能在页面改版导致抓取内容异常时,马上定位是哪一类数据源出了问题,而不是面对一个干巴巴的报错信息发愁。
第四,敏感信息管理。API Key 一定不要硬编码在代码里,用环境变量或者配置文件,并确保这些文件不被提交到 git 仓库。.gitignore里至少要有.env、agent_data.db、*.log。
第五,合规是底线,不是附加项。这个 Agent 的整体设计目标是处理公开信息,但“公开”并不是随意的通行证。实际操作中要遵守以下原则:
- 只采集公开、无需登录即可访问的页面,不尝试绕过验证码和访问控制。
- 尊重网站的 robots.txt 和访问频率限制,单源每天只抓一次是合理的。
- 不采集实名个人信息,如果页面自动包含无关的个人手机号、邮箱,存储时尽量脱敏。
- 涉及风险判断的结果,明确标注“AI 初筛结果,仅供参考,不构成最终决策依据”。
- 对外提供信息摘要时,保留原始链接和抓取时间,方便溯源。
第六,AI 幻觉的最终防线是人。在“查风险”这类高影响场景里,无论模型提示词写得多严格、校验规则做得多严密,都应该保留人工复核的环节。Agent 的价值是把原本需要 3 小时的筛选压缩到 10 分钟,而不是替你做最后的决定。
10. 总结与下一步扩展方向
这个项目验证了一件事:AI Agent 不一定非要做得复杂才有价值,把确定的工程问题和不确定的语义理解问题拆开,各用最合适的手段解决,才是它能够长期运行的关键。定时、抓取、去重、通知全部用确定性代码,语义抽取交给大模型,二者之间用 JSON 和指纹去重做接口。这套思路可以复用到非常多场景里。
如果你想继续往深做,值得尝试的方向有几个。第一,把 SQLite 换成向量数据库,把历史条目做向量化,让 Agent 能够回答“过去一个月有哪些和 XX 行业相关的项目线索”这类检索问题。第二,在通知层加入每日摘要,让模型把所有新增条目压缩成一段 500 字以内的日报推送到群里,而不是把原始条目列表直接丢出来。第三,把规则校验器做得更细致,比如针对不同类型的来源定义不同的必填字段,bidding 类要求金额和截止时间,risk 类要求事件类型和发生日期。
还有一个很自然的扩展是给这个 Agent 加一个“人工反馈回路”。当用户标记一条推送为“有用”或“没用”时,把这个结果作为下一轮抽样的样本。初期可以用人工阅读日志的方式,样本量上来后再考虑用模型自动评价。没有反馈回路的信息系统,进步速度终究是有限的。
这个 Agent 的原型代码并不复杂,但每个模块都是可以独立替换的。换数据源只需要改 YAML,换模型只需要改环境变量,换通知渠道只需要重写 notifier 里的一个函数。建议你先拿自己的真实需求做第一次部署,跑通后再慢慢迭代。毕竟,让 AI 替你把每天盯网站的时间省下来,才是这套系统最直接的收益。