每天固定花两小时刷新闻、查招标公告、看行业动态还是容易漏掉关键信息。如果让 AI 每天自己联网去“找项目、找融资线索、查风险”再把分析结果整理成日报推送过来这就能把重复劳动彻底自动化。本文从一个可落地的角度出发完整拆解这套 AI 自动情报助理系统的设计和实现包含搜索模块、大模型分析、风险识别、定时调度和消息推送等核心工程细节适合想入门 AI Agent 开发或者需要在团队内部做自动化信息采集的开发者参考。我尽量把每个环节都讲清楚先讲为什么需要这样一套系统再讲整体架构和模块划分然后给出可运行的 Python 代码最后补充高频问题排查和工程化建议。代码都以完整片段形式给出标注了文件路径你可以直接照着建项目。1. 需求与技术背景1.1 信息采集与整理的痛点在业务拓展、投资研究、行业分析这类场景里信息采集通常有很明显的重复性。每天要从新闻网站、招标平台、企业公告、招聘信息里筛选出“跟我有关”的内容再整理成简报发给团队成员。这个过程有三个痛点时间成本高。人工搜索、阅读、摘录一个调研任务常常要半天。覆盖面有限。搜索引擎返回结果多但真正相关的可能只有几条很容易漏掉。结果不可复用。每次调研都是临时手动操作没有形成持续的数据积累。其实这些问题可以用 AI Agent 来解决。所谓 AI Agent是一个具备“感知、决策、行动”闭环的智能程序它通过工具获取外部信息把信息交给大模型分析再根据分析结果执行下一步动作。放到“情报采集”场景里就是让 AI 每天定时上网搜索把结果做结构化分析最后把高价值的信息推送到你需要的地方。1.2 “找项目、找钱、查风险”分别是什么这套系统围绕三个核心任务展开每个任务有不同的数据源和分析逻辑。找项目重点是发现可参与的商机或可研究的技术方向。可以关注政府招标信息、开源社区高关注度项目、企业服务领域的招标动态。找钱重点不是字面上的“找资金”而是发现融资事件、政策补贴、行业扶持等商业机会。查风险重点是监控行业监管政策变化、企业负面舆情、法律诉讼、经营异动等风险信号。这三个任务本质上都是“搜索 分析 告警”的流程只是关键词和分析维度不同非常适合抽成一套通用框架。1.3 AI 上网的常见实现方式让 AI 获得实时信息有几种常见方案直接调用大模型自带的联网搜索能力比如部分商业模型的 Web Search 功能。用搜索 API 获取搜索结果再交给大模型分析。自己写爬虫抓取目标网站做页面解析。第一种最简单但受限于模型服务商的能力和额度第三种灵活但维护成本高还涉及合规问题。第二种是工程上最常用的方式通过 Tavily、Bing Web Search、DuckDuckGo 等搜索接口获取结果然后拼接成上下文交给大模型做二次分析。本文采用第二种方案并且把搜索模块抽象为统一接口方便替换服务商。2. 系统整体设计2.1 架构设计整个系统按“采集 → 分析 → 存储 → 通知”四个层次划分模块之间通过标准数据结构解耦。定时调度APScheduler ↓ 搜索引擎SearchEngine ↓ 搜索结果SearchResult ↓ LLM 分析LLMAnalyzer ↓ SQLite 存储Storage ↓ 消息推送Notifier ↓ 手机 / 邮件 / 群机器人定时调度触发后先根据任务类型生成搜索关键词调用搜索引擎获取原始结果再由大模型把原始结果提炼成结构化结论写入数据库最后决定是否需要推送。2.2 技术选型模块选型说明编程语言Python 3.10生态成熟适合快速开发搜索服务Tavily / DuckDuckGoTavily 专为 AI 场景设计DuckDuckGo 免密钥大模型OpenAI 兼容接口可接 OpenAI、DeepSeek、通义千问、Ollama 等定时调度APScheduler轻量适合进程内定时任务数据存储SQLite单机场景足够零部署成本消息推送Server酱 / Webhook / 邮件按团队习惯选择这里强调一下大模型部分我用 OpenAI 兼容接口因为目前绝大多数国内模型服务商都提供 OpenAI 兼容的 HTTP 接口只需要修改base_url和model就能切换代码不用大改。2.3 模块划分项目结构如下ai-autointel/ ├── config.py # 配置加载 ├── main.py # 入口支持单次执行和定时执行 ├── core/ │ ├── __init__.py │ ├── models.py # 数据结构定义 │ ├── search_engine.py # 搜索模块 │ ├── analyzer.py # LLM 分析模块 │ ├── risk_monitor.py # 风险监控模块 │ ├── storage.py # SQLite 存储 │ ├── notifier.py # 消息推送 │ └── runner.py # 任务编排 ├── requirements.txt └── .env.example下面先讲环境准备再逐个模块实现。3. 环境准备与版本说明3.1 Python 环境示例代码以 Python 3.10 为基准建议使用虚拟环境隔离依赖。python3 -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate3.2 安装第三方依赖需要安装的核心库如下pip install openai python-dotenv apscheduler requests如果搜索服务选择 DuckDuckGo需要额外安装pip install duckduckgo-search版本建议使用当前最新稳定版不同小版本之间接口差异一般不大但如果你遇到方法不存在或字段变化请以官方文档为准。3.3 密钥与环境变量新建.env文件内容参考.env.example# 搜索服务 SEARCH_PROVIDERddgs TAVILY_API_KEY # 大模型 LLM_API_KEYsk-your-key LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini # 通知服务 NOTIFY_TYPEconsole SERVERCHAN_KEY WEBHOOK_URL # 数据路径 DB_PATH./data/intelligence.db如果你的大模型服务商支持 OpenAI 兼容格式把LLM_BASE_URL改成服务商提供的地址即可。没有密钥也可以先用 DuckDuckGo 搜索并把NOTIFY_TYPE设为console这样系统会先把结果打印到控制台方便验证。4. 核心模块代码实现4.1 标准数据结构 core/models.py为了让搜索、分析、存储各层解耦先定义统一的SearchResult和IntelligenceItem。# file: core/models.py from dataclasses import dataclass, field from typing import Dict, Any dataclass class SearchResult: 标准化的搜索结果结构。 title: str url: str snippet: str source: str raw: Dict[str, Any] field(default_factorydict) dataclass class IntelligenceItem: 经过 LLM 分析后的情报条目。 task_type: str # project / money / risk title: str url: str snippet: str analysis: str # 模型生成的结论摘要 value_type: str info # opportunity / warning / info score: float 0.0 raw_json: Dict[str, Any] field(default_factorydict)4.2 搜索模块 core/search_engine.py搜索模块采用抽象基类 工厂模式便于在不同搜索服务之间切换。# file: core/search_engine.py from abc import ABC, abstractmethod from typing import List, Optional import requests from core.models import SearchResult class BaseSearchEngine(ABC): 所有搜索引擎的抽象基类。 abstractmethod def search(self, query: str, max_results: int 10, **kwargs) - List[SearchResult]: raise NotImplementedError class TavilySearchEngine(BaseSearchEngine): Tavily 搜索专为 AI 场景设计。 def __init__(self, api_key: str): self.api_key api_key self.endpoint https://api.tavily.com/search def search(self, query: str, max_results: int 10, **kwargs) - List[SearchResult]: headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { query: query, search_depth: kwargs.get(search_depth, advanced), max_results: max_results, include_answer: kwargs.get(include_answer, False), } resp requests.post(self.endpoint, jsonpayload, headersheaders, timeout15) resp.raise_for_status() data resp.json() results [] for item in data.get(results, []): results.append( SearchResult( titleitem.get(title, ), urlitem.get(url, ), snippetitem.get(content, ), sourcetavily, rawitem, ) ) return results class DuckDuckGoSearchEngine(BaseSearchEngine): DuckDuckGo 搜索免密钥适合本地测试。 def search(self, query: str, max_results: int 10, **kwargs) - List[SearchResult]: from duckduckgo_search import DDGS results [] with DDGS() as ddgs: for item in ddgs.text(query, max_resultsmax_results): results.append( SearchResult( titleitem.get(title, ), urlitem.get(href, ), snippetitem.get(body, ), sourceduckduckgo, rawitem, ) ) return results class SearchEngineFactory: 根据配置创建搜索引擎实例。 staticmethod def create(provider: str, api_key: Optional[str] None) - BaseSearchEngine: provider provider.lower() if provider tavily: if not api_key: raise ValueError(使用 Tavily 搜索时必须配置 TAVILY_API_KEY) return TavilySearchEngine(api_key) if provider in (ddgs, duckduckgo): return DuckDuckGoSearchEngine() raise ValueError(f不支持的搜索服务: {provider})这段代码里需要注意几点搜索接口返回字段可能随服务商变化统一转换为SearchResult后再向上传递业务层完全不感知搜索来源。Tavily 的鉴权方式支持 Header 和 body 两种不同时期文档有差异建议以官方文档为准示例中采用标准的Authorization: Bearer写法。DuckDuckGo 免费但存在频率限制搜索频率过高可能返回空结果。生产环境优先推荐 Tavily 或企业级搜索 API。4.3 关键词生成与结果去重搜索模块只能回答问题还需要一个组件把“找项目”“找钱”“查风险”这类任务转成具体的关键词列表并对多次搜索结果去重。# file: core/keywords.py from typing import List, Dict def build_keywords(task_type: str) - List[str]: 根据任务类型生成搜索关键词。你可以按行业自定义。 keyword_map: Dict[str, List[str]] { project: [ 开源项目 最新 高星, 信息化项目 招标公告, 企业服务 项目机会, ], money: [ 融资快讯 最新, 产业政策 补贴 申报, 投资机构 赛道 布局, ], risk: [ 数据安全 处罚 企业, 行业监管 政策 变动, 企业 诉讼 裁员 负面, ], } return keyword_map.get(task_type, []) def deduplicate(results: List[SearchResult]) - List[SearchResult]: 按 URL 去重保留第一个出现的条目。 seen set() unique [] for item in results: key item.url.strip().lower() if not key: continue if key in seen: continue seen.add(key) unique.append(item) return unique关键词是最需要运营调优的部分。同一个任务不同时间、不同行业关键词应该定期调整。建议把关键词放到配置文件里避免每次改逻辑都重新部署。4.4 LLM 分析模块 core/analyzer.py搜索返回的是零散网页片段必须经过大模型分析才能变成可用的结论。分析模块接收关键词和搜索结果列表返回结构化 JSON。# file: core/analyzer.py import json import re from typing import List, Dict, Any from openai import OpenAI from core.models import SearchResult, IntelligenceItem class LLMAnalyzer: 基于大模型对搜索结果进行结构化分析。 def __init__(self, api_key: str, base_url: str , model: str gpt-4o-mini): if base_url: self.client OpenAI(api_keyapi_key, base_urlbase_url) else: self.client OpenAI(api_keyapi_key) self.model model def analyze( self, task_type: str, query: str, results: List[SearchResult], max_items: int 5, ) - List[IntelligenceItem]: 把搜索结果交给大模型提取高价值情报。 if not results: return [] # 只保留标题、链接、摘要控制上下文长度 compact_results [] for r in results[:10]: compact_results.append( {title: r.title, url: r.url, snippet: r.snippet[:500]} ) prompt f你是一名商业情报分析师。请基于以下搜索结果围绕任务「{query}」提取关键信息。 搜索结果列表 {json.dumps(compact_results, ensure_asciiFalse, indent2)} 要求 1. 输出 JSON 数组不要输出 Markdown 代码块。 2. 每条结果必须包含 title、url、summary、value_type、score 五个字段。 3. value_type 只能是 opportunity机会、warning风险、info普通信息。 4. score 是 0-10 的整数表示与任务的相关度和价值10 分最高。 5. 过滤掉广告、无关内容、重复信息。 6. 最多返回 {max_items} 条。 只输出合法 JSON。 try: resp self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是一个严格输出 JSON 的情报分析助手。}, {role: user, content: prompt}, ], temperature0.2, max_tokens1000, ) content resp.choices[0].message.content data self._parse_json(content) items [] for item in data: items.append( IntelligenceItem( task_typetask_type, titleitem.get(title, ), urlitem.get(url, ), snippetitem.get(summary, ), value_typeitem.get(value_type, info), scorefloat(item.get(score, 0)), analysisitem.get(summary, ), raw_jsonitem, ) ) return items except Exception as exc: # 模型输出异常时降级保留原文标题作为普通信息 print(f[analyzer] 大模型分析失败降级处理: {exc}) return [ IntelligenceItem( task_typetask_type, titler.title, urlr.url, snippetr.snippet, value_typeinfo, score0.0, ) for r in results[:max_items] ] staticmethod def _parse_json(content: str) - List[Dict[str, Any]]: 容错解析模型输出的 JSON。 text content.strip() if text.startswith(): text re.sub(r^[a-zA-Z]*, , text) text text.rstrip() try: return json.loads(text) except json.JSONDecodeError: # 尝试从文本中提取 JSON 数组 match re.search(r\[.*\], text, re.S) if match: return json.loads(match.group(0)) return []这段代码有几个值得说明的点temperature设置为 0.2降低输出随机性保证信息抽取稳定性。如果大模型输出不是合法 JSON会触发降级策略直接把原始结果标题保存为普通信息避免任务中断。base_url允许为空为空时使用 OpenAI 默认地址如果你接的是国内模型填服务商提供的兼容地址即可。4.5 风险监控模块 core/risk_monitor.py风险监控和普通任务类似但更关注“最新负面变化”。它接收一个监控目标列表对每个目标做多关键词搜索再用大模型判断是否存在风险。# file: core/risk_monitor.py from typing import List from core.models import SearchResult, IntelligenceItem class RiskMonitor: 针对特定实体或行业做持续风险扫描。 def __init__(self, analyzer, search_engine): self.analyzer analyzer self.search_engine search_engine def scan( self, targets: List[str], risk_keywords: List[str], max_results: int 5, ) - List[IntelligenceItem]: all_results: List[SearchResult] [] for target in targets: for keyword in risk_keywords: query f{target} {keyword} try: results self.search_engine.search(query, max_resultsmax_results) all_results.extend(results) except Exception as exc: print(f[risk_monitor] 搜索失败 target{target} query{query} error{exc}) if not all_results: return [] return self.analyzer.analyze( task_typerisk, query重点关注负面风险、合规风险、经营异常, resultsall_results, max_items10, )实际使用中risk_keywords 可以设计成“处罚、诉讼、裁员、违规、立案、关停、数据泄露”等组合词越具体效果越好。4.6 存储模块 core/storage.py存储模块使用 SQLite简单直接方便后续查询历史记录。# file: core/storage.py import sqlite3 from datetime import datetime from typing import List from core.models import IntelligenceItem class Storage: def __init__(self, db_path: str ./data/intelligence.db): self.db_path db_path self._init_db() def _connect(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self): conn self._connect() conn.execute( CREATE TABLE IF NOT EXISTS intelligence ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_type TEXT NOT NULL, title TEXT NOT NULL, url TEXT, snippet TEXT, value_type TEXT DEFAULT info, score REAL DEFAULT 0, created_at TEXT NOT NULL ) ) conn.commit() conn.close() def save_items(self, items: List[IntelligenceItem]): conn self._connect() now datetime.now().isoformat(timespecseconds) for item in items: conn.execute( INSERT INTO intelligence (task_type, title, url, snippet, value_type, score, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) , ( item.task_type, item.title[:500], item.url[:1000], item.snippet[:2000], item.value_type, item.score, now, ), ) conn.commit() conn.close() def query_latest(self, task_type: str , limit: int 20) - List[sqlite3.Row]: conn self._connect() sql SELECT * FROM intelligence params [] if task_type: sql WHERE task_type ? params.append(task_type) sql ORDER BY id DESC LIMIT ? params.append(limit) rows conn.execute(sql, params).fetchall() conn.close() return rows存储层目前只做了新增和查询没有做按 URL 去重。如果你想避免重复推送可以在建表时为 URL 增加唯一索引但要注意部分搜索结果的 URL 为空。4.7 消息推送模块 core/notifier.py推送模块支持三种方式控制台输出、Server酱、通用 Webhook。# file: core/notifier.py import requests class Notifier: def __init__(self, notify_type: str console, serverchan_key: str , webhook_url: str ): self.notify_type notify_type self.serverchan_key serverchan_key self.webhook_url webhook_url def send(self, title: str, content: str): if self.notify_type console: print(\n 推送内容 ) print(title) print(content) print(\n) elif self.notify_type serverchan: self._send_serverchan(title, content) elif self.notify_type webhook: self._send_webhook(title, content) else: print(f[notifier] 未识别的推送类型: {self.notify_type}) def _send_serverchan(self, title: str, content: str): # 接口地址以 Server酱官方文档为准 if not self.serverchan_key: print([notifier] SERVERCHAN_KEY 未配置跳过推送) return url fhttps://sctapi.ftqq.com/{self.serverchan_key}.send resp requests.post(url, data{title: title, desp: content}, timeout10) resp.raise_for_status() def _send_webhook(self, title: str, content: str): if not self.webhook_url: print([notifier] WEBHOOK_URL 未配置跳过推送) return payload {title: title, content: content} resp requests.post(self.webhook_url, jsonpayload, timeout10) resp.raise_for_status()推送模块不直接依赖某个具体渠道后续要接企业微信机器人、钉钉机器人或邮件只需要增加一个私有方法。4.8 任务编排 core/runner.py任务编排是整个系统的核心入口。它把搜索、关键词生成、分析、存储、推送串起来并对每个任务分别配置最大返回条数。# file: core/runner.py from typing import List from core.keywords import build_keywords, deduplicate from core.models import IntelligenceItem from core.search_engine import BaseSearchEngine from core.analyzer import LLMAnalyzer from core.storage import Storage from core.notifier import Notifier class TaskRunner: def __init__( self, search_engine: BaseSearchEngine, analyzer: LLMAnalyzer, storage: Storage, notifier: Notifier, ): self.search_engine search_engine self.analyzer analyzer self.storage storage self.notifier notifier def run_task(self, task_type: str, max_results_per_query: int 8) - List[IntelligenceItem]: keywords build_keywords(task_type) all_results [] for keyword in keywords: print(f[runner] 任务{task_type} 关键词{keyword}) try: results self.search_engine.search(keyword, max_resultsmax_results_per_query) all_results.extend(results) except Exception as exc: print(f[runner] 搜索失败 keyword{keyword} error{exc}) unique_results deduplicate(all_results) print(f[runner] 任务{task_type} 共获取 {len(all_results)} 条去重后 {len(unique_results)} 条) items self.analyzer.analyze( task_typetask_type, query、.join(keywords), resultsunique_results, max_items5, ) if items: self.storage.save_items(items) # 推送高价值信息 high_value [item for item in items if item.score 7] if high_value: lines [] for item in high_value: lines.append( f- [{item.value_type}] {item.title}\n {item.url}\n {item.snippet} ) self.notifier.send( titlef[AI情报] {task_type} 发现 {len(high_value)} 条高价值信息, content\n.join(lines), ) return items def run_risk_scan(self, targets: List[str], risk_keywords: List[str]) - List[IntelligenceItem]: # 风险扫描单独走专用逻辑 from core.risk_monitor import RiskMonitor monitor RiskMonitor(analyzerself.analyzer, search_engineself.search_engine) items monitor.scan(targetstargets, risk_keywordsrisk_keywords) if items: self.storage.save_items(items) warnings [item for item in items if item.value_type warning] if warnings: lines [ f- {item.title}\n {item.url}\n {item.snippet} for item in warnings ] self.notifier.send( titlef[AI风险预警] 发现 {len(warnings)} 条风险信号, content\n.join(lines), ) return items这里的评分阈值 7 分是经验值你可以根据实际输出调整。如果推送太频繁就提高阈值如果觉得漏报就降低阈值。5. 完整实战跑一次全流程5.1 初始化项目按前面 2.3 节的目录结构创建项目然后在项目根目录添加.env.example环境变量文件。# 创建数据目录 mkdir -p data touch data/.gitkeep5.2 编写配置文件 core/config.py# file: config.py import os from dataclasses import dataclass, field from dotenv import load_dotenv load_dotenv() dataclass class Config: search_provider: str field(default_factorylambda: os.getenv(SEARCH_PROVIDER, ddgs)) tavily_api_key: str field(default_factorylambda: os.getenv(TAVILY_API_KEY, )) llm_api_key: str field(default_factorylambda: os.getenv(LLM_API_KEY, )) llm_base_url: str field(default_factorylambda: os.getenv(LLM_BASE_URL, )) llm_model: str field(default_factorylambda: os.getenv(LLM_MODEL, gpt-4o-mini)) notify_type: str field(default_factorylambda: os.getenv(NOTIFY_TYPE, console)) serverchan_key: str field(default_factorylambda: os.getenv(SERVERCHAN_KEY, )) webhook_url: str field(default_factorylambda: os.getenv(WEBHOOK_URL, )) db_path: str field(default_factorylambda: os.getenv(DB_PATH, ./data/intelligence.db))5.3 编写主程序 main.py# file: main.py import argparse from config import Config from core.search_engine import SearchEngineFactory from core.analyzer import LLMAnalyzer from core.storage import Storage from core.notifier import Notifier from core.runner import TaskRunner from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def build_runner(cfg: Config) - TaskRunner: search_engine SearchEngineFactory.create(cfg.search_provider, cfg.tavily_api_key) analyzer LLMAnalyzer( api_keycfg.llm_api_key, base_urlcfg.llm_base_url, modelcfg.llm_model, ) storage Storage(db_pathcfg.db_path) notifier Notifier( notify_typecfg.notify_type, serverchan_keycfg.serverchan_key, webhook_urlcfg.webhook_url, ) return TaskRunner( search_enginesearch_engine, analyzeranalyzer, storagestorage, notifiernotifier, ) def run_once(cfg: Config): runner build_runner(cfg) runner.run_task(project) runner.run_task(money) runner.run_task(risk) # 自定义风险扫描目标 runner.run_risk_scan( targets[某行业头部企业], risk_keywords[处罚, 诉讼, 裁员, 数据泄露], ) print(本轮任务执行完成) def start_schedule(cfg: Config): runner build_runner(cfg) scheduler BlockingScheduler(timezoneAsia/Shanghai) # 每天早上 9 点执行全部任务 scheduler.add_job( run_once, CronTrigger(hour9, minute0), args[cfg], iddaily_all_tasks, replace_existingTrue, ) # 风险任务每 3 小时执行一次 scheduler.add_job( run_risk_scan_job, CronTrigger(hour*/3, minute15), args[runner], idrisk_scan, replace_existingTrue, ) print(定时任务已启动按 CtrlC 停止) try: scheduler.start() except (KeyboardInterrupt, SystemExit): print(定时任务已停止) def run_risk_scan_job(runner: TaskRunner): runner.run_risk_scan( targets[某行业头部企业], risk_keywords[处罚, 诉讼, 裁员, 数据泄露], ) if __name__ __main__: parser argparse.ArgumentParser(descriptionAI 自动情报助理) group parser.add_mutually_exclusive_group() group.add_argument(--once, actionstore_true, help只执行一次) group.add_argument(--schedule, actionstore_true, help启动定时调度) args parser.parse_args() cfg Config() if args.schedule: start_schedule(cfg) else: run_once(cfg)5.4 运行与预期输出如果只配置了 DuckDuckGo可以这样运行python main.py --once预期输出类似[runner] 任务project 关键词开源项目 最新 高星 [runner] 任务project 关键词信息化项目 招标公告 [runner] 任务project 关键词企业服务 项目机会 [runner] 任务project 共获取 24 条去重后 18 条 [analyzer] 大模型分析完成提取 5 条情报 [runner] 任务money 关键词融资快讯 最新 ... 推送内容 [AI情报] project 发现 2 条高价值信息 - [opportunity] 某开源项目发布 3.0 版本 https://example.com/news/xxx ......控制台会逐行打印每个任务的执行状态最终把高价值信息推送到控制台、Server酱或 Webhook。6. 常见问题与排查思路问题现象常见原因解决思路Tavily 返回 401 UnauthorizedAPI Key 未配置或已失效检查.env中的TAVILY_API_KEY确认 Key 正确且账户有余额DuckDuckGo 搜索返回空列表请求频率过高触发限流降低调度频率在关键词之间增加 sleep或切换到 Tavily 等付费服务大模型输出不是合法 JSON模型指令遵循能力不足或上下文过长压缩搜索结果片段把 snippet 截断到 500 字以内增加 JSON 输出约束使用更强的模型推送没有收到消息NOTIFY_TYPE配置错误或 key 为空先设置NOTIFY_TYPEconsole验证链路再切换到 Server酱或 Webhook定时任务不执行时区或调度器配置问题检查BlockingScheduler(timezone...)是否设置正确单次执行--once可先验证任务逻辑数据重复入库未对 URL 做唯一约束在 SQLite 表结构中增加 URL 去重或保存前用 deduplicate 过滤需要手动清理历史数据时先备份除了上面这些还有一个容易忽略的问题大模型 API 调用失败时任务可能会中断。我在analyzer.analyze中做了降级处理但如果你修改了调用逻辑记得保留 try-except避免某一个任务的异常影响整个调度。7. 工程化建议与合规边界7.1 合理设计提示词提示词是整个系统效果的上限。建议把提示词独立成一个prompts.py文件方便版本管理。升级提示词前准备好一批固定测试样本对比新旧输出质量避免上线后才发现分析结果漂移。7.2 引入失败重试与降级策略网络请求很容易受服务商限流、超时等因素影响。建议对搜索调用和大模型调用增加有限次重试比如指数退避重试 3 次。连续失败时把异常信息记录到日志并跳过本轮任务而不是让整个进程崩溃。7.3 控制成本搜索 API 和大模型 API 都是按量计费。建议限制每轮搜索的条数和 LLM 生成的 token 数。在代码中我已经把搜索结果截断为最多 10 条、每条 snippet 截断为 500 字LLM 返回max_tokens1000这能显著降低调用成本。生产环境还可以按任务设置每日额度。7.4 数据合规与授权边界这是一条非常重要的底线。AI 自动采集信息必须遵守以下原则只能采集公开信息或你已获得合法授权的数据。使用搜索 API、官方开放接口不要绕过登录鉴权、验证码或反爬策略。不采集个人敏感信息不将采集到的数据用于与授权目的无关的用途。涉及企业数据监控时建议先确认数据来源和使用范围符合协议要求。对推送内容做人工抽检避免大模型“幻觉”导致错误信息扩散。7.5 最小权限与数据保留数据库中的情报数据也可能包含一定业务敏感性。建议对数据库文件设置文件系统访问权限避免明文暴露。设置数据保留周期定期清理旧数据。API Key 放在.env中且.env加入.gitignore绝不提交到代码仓库。如果系统部署在服务器上使用单独的低权限用户运行遵循最小权限原则。7.6 日志与可观测性建议用标准logging替代print。生产环境把日志输出到文件并记录每次任务的运行时间、结果数量、搜索服务商、模型名称后续排查问题时会有非常大的帮助。8. 总结与下一步这篇文章从业务痛点出发设计并实现了一个“AI 自动情报助理”系统核心能力是让 AI 定时联网完成找项目、找融资商机、查风险三个任务。代码覆盖了搜索接口抽象、LLM 结构化分析、SQLite 存储、任务调度和消息推送已经是一套可运行的工程雏形。下一步你可以从这几个方向继续深入把关键词配置化做成 Web 管理后台运营同学可以随时调整。增加网页内容爬取模块对高价值结果做正文全文分析。接入企业微信或钉钉机器人把推送链路接入团队协作工具。把分析结果沉淀到向量数据库支持语义检索与自动摘要。动手修改关键词、跑几次真实任务、观察模型输出是否合理是掌握这套系统的关键。如果本文对你有帮助欢迎收藏备用后续我也会继续分享 AI Agent 工程落地的更多细节。