你是不是经常感觉信息过载但又怕错过关键信息每天花大量时间刷新闻、看行业报告、跟踪竞品动态结果发现信息要么滞后要么零散真正有价值的情报总是姗姗来迟。这种“信息差”不仅让你在决策上慢人一步更可能让你在技术选型、市场判断上陷入被动。今天我们不再依赖人工搜索和碎片化阅读。我将带你用 AI 技术亲手搭建一个自动化、智能化、可定制的个人信息情报系统。这个系统的核心不是简单地聚合 RSS而是利用 AI Agent 的能力理解你的需求主动从海量信息中筛选、分析、提炼最终将结构化的“情报”推送到你面前。我们将使用当下热门的DeepSeek大模型作为“大脑”结合浏览器自动化等工具作为“手脚”实现一个低成本、高效率的解决方案。读完本文你将能独立部署一个 7x24 小时运行的 AI 情报官。它不仅能帮你监控特定技术话题如“Vibe Coding 的最新实践”、跟踪开源项目动态、分析行业趋势还能生成每日简报让你用一杯咖啡的时间掌握别人需要一整天才能搜集到的信息。1. 这篇文章真正要解决的问题从“信息过载”到“情报主动送达”在深入技术细节之前我们必须先厘清一个核心问题我们到底要解决什么很多人误以为搭建情报系统就是做一个“高级爬虫”或“RSS 阅读器”这完全低估了 AI 的潜力也误解了“情报”的本质。情报Intelligence与信息Information有本质区别。信息是原始的、未经处理的信号比如一篇新的技术博客、一个 GitHub 的 commit、一条行业新闻。情报则是经过筛选、关联、分析后能直接支撑决策的结论例如“竞争对手 A 在项目 B 中开始大量使用向量数据库这可能意味着他们正在加强其产品的语义搜索能力。”传统的信息获取方式存在三大痛点被动接收你需要主动去各个平台“找”信息效率低下。缺乏分析即使找到了信息也需要你人工阅读、理解、关联耗时耗力。难以定制通用信息流无法精准匹配你的个人兴趣和专业领域。本文要构建的 AI 情报系统旨在通过技术手段彻底扭转这一局面变被动为主动系统按你设定的规则自动、持续地抓取目标信息源。变信息为情报利用大模型的理解和总结能力自动提炼核心观点、识别潜在影响、进行多源信息交叉验证。变通用为专属你可以用自然语言定义情报任务比如“监控所有关于‘浏览器自动化测试’与‘Playwright’结合的新文章并对比其与 Selenium 的方案优劣”。接下来的内容我们将聚焦于如何用可落地的技术栈实现这一构想核心会围绕DeepSeek API、浏览器自动化Playwright以及任务编排展开。你会发现搭建这样一个系统其技术门槛远没有想象中高。2. 核心概念与系统架构设计在动手写代码之前我们需要建立几个关键概念并设计出系统的整体架构。这能帮助你在后续配置和开发时清楚每一部分在扮演什么角色。2.1 关键概念解析AI Agent智能体这是我们系统的“大脑”。它不是一个简单的聊天机器人而是一个能根据目标如“获取今日AI编程领域动态”自主规划步骤思考、调用工具行动、并评估结果观察的智能程序。在本系统中DeepSeek 模型将承担 Agent 的“思考”和“分析”核心。工具ToolsAgent 的“手脚”。Agent 本身无法直接浏览网页或读取文件它需要通过调用预设的工具来与外界交互。我们将为 Agent 配备的关键工具包括网页抓取工具用于访问指定 URL 并获取页面内容。搜索引擎工具可选用于扩大信息发现范围。数据存储工具用于保存历史情报和原始数据。任务编排Orchestration决定系统何时运行、以什么顺序执行哪些任务的中枢。例如每天上午9点自动执行“监控任务”然后将结果交给“分析任务”最后触发“推送任务”。我们将使用简单的Cron 定时任务或更强大的工作流引擎如 Apache Airflow 的轻量替代来实现。信息源Sources系统监控的目标。这不仅仅是 RSS 订阅源还可以是特定技术博客、新闻网站。GitHub 仓库的 Releases、Issues、Commits。社交媒体如 Twitter/X的关键词或用户。行业报告网站、论坛的热门板块。2.2 系统架构设计一个健壮的情报系统应该模块清晰、易于扩展。以下是推荐的核心架构[信息源列表] | v [定时调度器] (Cron Job / Workflow Engine) | v [采集器模块] (使用 Playwright 进行浏览器自动化抓取) | \ | \ (原始 HTML/数据) | \ v v [内容解析器] [原始存储] (如 SQLite/JSON文件) | v [AI 分析引擎] (调用 DeepSeek API) | \ | \ (分析结果摘要、分类、情感、关联) | \ v v [情报生成器] [分析结果存储] | v [推送模块] (邮件、Slack、钉钉、Telegram) | v [用户终端]各模块职责调度器系统的发条按计划触发整个流水线。采集器系统的眼睛和手负责从信息源获取原始内容。我们选择Playwright而非简单requests是因为现代网站大量使用 JavaScript 渲染Playwright 能模拟真实浏览器行为确保抓到完整内容。解析器从杂乱无章的 HTML 中提取出干净的标题、正文、发布时间等结构化数据。AI 分析引擎系统的核心大脑。它将解析后的文本发送给 DeepSeek并提出精准的指令Prompt要求其进行总结、提取关键词、判断相关性、甚至进行情感分析。情报生成器将 AI 的分析结果格式化为易于阅读的报告如 Markdown。推送模块将最终报告送达给你。这个架构的优点是松耦合。你可以轻易替换其中任何一个模块例如将 DeepSeek 换成其他大模型 API或将邮件推送改为微信机器人。3. 环境准备与工具选型工欲善其事必先利其器。我们将选择一个简单、高效且免费资源丰富的技术栈确保每个人都能快速上手。3.1 基础运行环境操作系统推荐 Linux (Ubuntu 20.04) 或 macOS。Windows 也可行但部分命令行操作略有差异。Python 版本Python 3.9。这是大多数现代AI库的基础要求。包管理使用pip或更推荐的poetry/pipenv来管理项目依赖避免环境冲突。3.2 核心工具与库我们将基于 Python 生态来构建。请在你的项目目录下创建一个requirements.txt文件或直接安装以下库# 创建并进入项目目录 mkdir ai-intelligence-system cd ai-intelligence-system # 初始化Python虚拟环境强烈推荐 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install playwright # 浏览器自动化用于抓取动态网页 pip install beautifulsoup4 # HTML 解析库用于从Playwright获取的页面中提取内容 pip install requests # 用于调用DeepSeek API pip install schedule # 轻量级定时任务库用于简单调度 # 或者使用更强大的APScheduler # pip install apscheduler pip install python-dotenv # 用于管理环境变量如API密钥 pip install sqlite3 # Python内置用于轻量级数据存储安装 Playwright 浏览器 Playwright 需要安装它自带的浏览器内核。playwright install chromium3.3 获取 DeepSeek API 密钥本系统的“智能”来源于 DeepSeek 大模型。你需要前往 DeepSeek 开放平台 注册账号并获取 API Key。登录后在控制台找到“API Keys”部分。创建一个新的 Key并妥善保存。它看起来像一串长字符sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx。重要安全提示永远不要将 API Key 直接硬编码在代码中或上传到 GitHub 等公开仓库。我们将使用环境变量来管理它。在项目根目录创建一个.env文件# .env 文件内容 DEEPSEEK_API_KEYsk-你的实际API密钥 DEEPSEEK_API_BASEhttps://api.deepseek.com # API基础地址请以官方文档为准并在代码中通过python-dotenv加载# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量 DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) API_BASE os.getenv(DEEPSEEK_API_BASE, https://api.deepseek.com)4. 核心模块实现分步构建你的 AI 情报官现在我们开始动手实现架构图中的每一个核心模块。我们将采用自底向上的方式先实现基础工具再组合成完整流程。4.1 模块一智能网页采集器Playwright为什么不用简单的requests.get()因为很多网站如技术博客、新闻站采用前端框架React, Vue.js动态加载内容直接请求只能得到空壳 HTML。Playwright 能模拟真人浏览器操作等待页面完全加载后再获取内容。# crawler/playwright_crawler.py import asyncio from playwright.async_api import async_playwright from bs4 import BeautifulSoup import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class PlaywrightCrawler: def __init__(self, headlessTrue): # headlessTrue 表示无头模式不显示浏览器界面 self.headless headless async def fetch_url(self, url: str, wait_for_selector: str None, timeout: int 30000): 使用 Playwright 抓取指定 URL 的完整内容。 :param url: 目标网址 :param wait_for_selector: 可选等待某个CSS选择器出现后再抓取用于应对动态加载 :param timeout: 超时时间毫秒 :return: 包含标题和正文文本的字典 async with async_playwright() as p: # 启动浏览器推荐使用 Chromium兼容性好 browser await p.chromium.launch(headlessself.headless) context await browser.new_context() page await context.new_page() try: logger.info(f正在访问: {url}) await page.goto(url, wait_untilnetworkidle, timeouttimeout) # 等待网络空闲 # 如果提供了选择器则等待该元素出现 if wait_for_selector: await page.wait_for_selector(wait_for_selector, timeout10000) # 获取页面完整的 HTML 内容 html_content await page.content() # 使用 BeautifulSoup 解析 HTML提取标题和正文 soup BeautifulSoup(html_content, html.parser) # 提取标题 - 尝试多种可能的选择器 title soup.title.string if soup.title else # 如果 og:title 存在有时它更准确 og_title soup.find(meta, propertyog:title) if og_title and og_title.get(content): title og_title[content] # 提取正文 - 这是一个简化的示例实际中可能需要针对不同网站定制 # 常见策略寻找 article, main 标签或者包含大量文本的 div article_body article_tag soup.find(article) or soup.find(main) or soup.find(div, class_lambda c: c and (content in c or post in c or article in c)) if article_tag: article_body article_tag.get_text(separator\n, stripTrue) else: # 备选方案获取整个 body 的文本噪声会多 article_body soup.body.get_text(separator\n, stripTrue) if soup.body else logger.info(f成功抓取: {title[:50]}...) return { url: url, title: title.strip(), body: article_body.strip(), raw_html: html_content[:5000] # 保存部分HTML供调试避免过大 } except Exception as e: logger.error(f抓取 {url} 时出错: {e}) return {url: url, title: , body: , error: str(e)} finally: await browser.close() # 同步函数包装便于在非异步环境中调用 def fetch_url_sync(url: str, **kwargs): 同步接口内部运行异步函数 crawler PlaywrightCrawler(headlessTrue) return asyncio.run(crawler.fetch_url(url, **kwargs)) if __name__ __main__: # 测试代码 test_url https://example.com # 请替换为一个真实的动态网站进行测试 result fetch_url_sync(test_url, wait_for_selectorarticle) print(f标题: {result[title]}) print(f正文预览: {result[body][:200]}...)关键点解析wait_untilnetworkidle这是 Playwright 的核心优势之一它会等待页面网络请求基本停止后再返回确保动态内容已加载。wait_for_selector对于某些特定网站你可以指定一个元素如article作为页面加载完成的标志更加精准。正文提取代码中的正文提取逻辑是通用的但效果可能因网站结构而异。在真实项目中你可能需要为重要的信息源编写特定的提取规则可配置化这是提升情报质量的关键一步。4.2 模块二AI 分析引擎DeepSeek API 调用这是系统的“大脑”。我们将定义一个类专门负责与 DeepSeek API 交互并封装几个常用的分析任务。# analyzer/deepseek_analyzer.py import requests import json import logging from typing import List, Dict, Any from config import DEEPSEEK_API_KEY, API_BASE # 导入配置文件 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DeepSeekAnalyzer: def __init__(self, api_key: str None, model: str deepseek-chat): self.api_key api_key or DEEPSEEK_API_KEY if not self.api_key: raise ValueError(DeepSeek API Key 未设置。请在 .env 文件中配置 DEEPSEEK_API_KEY) self.model model self.api_url f{API_BASE}/chat/completions self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } def _call_api(self, messages: List[Dict[str, str]]) - str: 调用 DeepSeek API 的基础方法 payload { model: self.model, messages: messages, temperature: 0.2, # 较低的温度使输出更稳定、更事实性 max_tokens: 2000 } try: response requests.post(self.api_url, headersself.headers, jsonpayload, timeout60) response.raise_for_status() # 如果状态码不是200抛出异常 result response.json() return result[choices][0][message][content] except requests.exceptions.RequestException as e: logger.error(fAPI 请求失败: {e}) if hasattr(e.response, text): logger.error(f错误响应: {e.response.text}) return except (KeyError, IndexError, json.JSONDecodeError) as e: logger.error(f解析 API 响应失败: {e}) return def summarize_article(self, title: str, content: str) - Dict[str, Any]: 总结一篇文章的核心内容。 返回结构化的摘要信息。 prompt f请对以下技术文章进行专业摘要和分析。 文章标题{title} 文章内容 {content[:6000]} # 限制内容长度避免超出Token限制 请以JSON格式返回分析结果包含以下字段 1. summary: 文章的核心内容摘要不超过300字。 2. key_points: 一个数组列出3-5个最关键的技术点或观点。 3. keywords: 一个数组提取5-8个核心关键词。 4. relevance_score: 一个1-10的整数评估这篇文章与“软件开发、AI编程、前沿技术”领域的相关性。 5. sentiment: 文章的整体情感倾向可选值: 积极, 消极, 中性。 messages [ {role: system, content: 你是一个专注于技术领域的信息分析专家。请准确、简洁地分析文章。}, {role: user, content: prompt} ] analysis_text self._call_api(messages) # 尝试从返回文本中解析JSON。DeepSeek有时会在JSON外加 Markdown 代码块标记。 try: # 清理可能的代码块标记 if analysis_text.startswith(json): analysis_text analysis_text[7:-3] # 移除 json 和结尾的 elif analysis_text.startswith(): analysis_text analysis_text[3:-3] analysis_data json.loads(analysis_text.strip()) return analysis_data except json.JSONDecodeError: logger.warning(f无法解析AI返回的JSON返回原始文本。原文: {analysis_text[:200]}) # 返回一个包含原始文本的默认结构 return { summary: analysis_text, key_points: [], keywords: [], relevance_score: 5, sentiment: 中性, raw_output: analysis_text } def generate_daily_brief(self, articles_analysis: List[Dict]) - str: 根据多篇文章的分析结果生成一份每日简报。 # 将分析结果格式化为文本作为新的Prompt输入 input_text 以下是今日监控到的多篇文章分析结果\n\n for i, analysis in enumerate(articles_analysis, 1): input_text f【文章{i}】{analysis.get(title, 无标题)}\n input_text f摘要{analysis.get(summary, 无摘要)}\n input_text f关键词{, .join(analysis.get(keywords, []))}\n input_text f相关性评分{analysis.get(relevance_score, 5)}/10\n input_text - * 40 \n prompt f{input_text} 请基于以上信息生成一份面向技术负责人或开发者的《每日技术情报简报》。 简报要求 1. 用一段话概括今日整体技术风向。 2. 分点列出最重要的2-3个技术动态或趋势并简要说明其潜在影响。 3. 给出一个“今日值得关注”的总结指出最值得深入阅读或跟进的一两篇文章或话题。 请使用清晰、专业的Markdown格式输出。 messages [ {role: system, content: 你是一名资深技术布道师擅长从零散信息中提炼出有洞察力的趋势简报。}, {role: user, content: prompt} ] return self._call_api(messages) # 测试分析功能 if __name__ __main__: analyzer DeepSeekAnalyzer() # 假设我们已经抓取了一篇文章 test_data { title: Vibe Coding一种新兴的AI辅助编程范式, body: ...这里是一大段关于Vibe Coding的文章内容... } result analyzer.summarize_article(test_data[title], test_data[body]) print(json.dumps(result, indent2, ensure_asciiFalse))关键点解析结构化 Prompt 工程我们通过精心设计的 Prompt引导 AI 返回结构化的 JSON 数据这极大方便了后续的数据处理。Prompt 中明确了角色、任务和输出格式。Token 长度管理content[:6000]用于限制发送给 API 的文本长度防止超出模型上下文限制并控制 API 成本。对于超长文章更复杂的策略是分段总结再聚合。错误处理与降级API 调用可能失败AI 也可能不按格式返回。代码中包含了异常捕获和降级处理返回原始文本保证系统的鲁棒性。温度参数temperature0.2设置较低使模型输出更确定、更少“创造性”更适合事实性分析任务。4.3 模块三数据存储与任务调度我们需要一个地方来存储抓取到的原始数据、分析结果以及记录任务运行状态。同时需要一个调度器来定时执行任务。轻量级数据存储SQLite# storage/db_client.py import sqlite3 import json from datetime import datetime from typing import List, Dict, Any class IntelligenceDB: def __init__(self, db_path: str intelligence.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._init_tables() def _init_tables(self): 初始化数据库表 cursor self.conn.cursor() # 原始文章表 cursor.execute( CREATE TABLE IF NOT EXISTS raw_articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE, title TEXT, body TEXT, fetch_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ) # 分析结果表 cursor.execute( CREATE TABLE IF NOT EXISTS analysis_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER, summary TEXT, key_points TEXT, -- 存储为JSON字符串 keywords TEXT, -- 存储为JSON字符串 relevance_score INTEGER, sentiment TEXT, analysis_time DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (article_id) REFERENCES raw_articles (id) ) ) # 任务运行日志表 cursor.execute( CREATE TABLE IF NOT EXISTS task_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_name TEXT, status TEXT, -- SUCCESS, FAILED message TEXT, run_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_article(self, url: str, title: str, body: str) - int: 保存抓取到的文章返回文章ID cursor self.conn.cursor() try: cursor.execute( INSERT OR REPLACE INTO raw_articles (url, title, body) VALUES (?, ?, ?), (url, title, body) ) self.conn.commit() return cursor.lastrowid except sqlite3.Error as e: print(f保存文章失败: {e}) return -1 def save_analysis(self, article_id: int, analysis_data: Dict[str, Any]): 保存AI分析结果 cursor self.conn.cursor() cursor.execute( INSERT INTO analysis_results (article_id, summary, key_points, keywords, relevance_score, sentiment) VALUES (?, ?, ?, ?, ?, ?) , ( article_id, analysis_data.get(summary), json.dumps(analysis_data.get(key_points, []), ensure_asciiFalse), json.dumps(analysis_data.get(keywords, []), ensure_asciiFalse), analysis_data.get(relevance_score, 5), analysis_data.get(sentiment, 中性) )) self.conn.commit() def log_task(self, task_name: str, status: str, message: str ): 记录任务运行日志 cursor self.conn.cursor() cursor.execute( INSERT INTO task_logs (task_name, status, message) VALUES (?, ?, ?), (task_name, status, message) ) self.conn.commit() def close(self): self.conn.close()简单任务调度器 对于个人使用schedule库足够轻量。对于更复杂的依赖关系可以考虑APScheduler。# scheduler/daily_scheduler.py import schedule import time import threading from datetime import datetime from crawler.playwright_crawler import fetch_url_sync from analyzer.deepseek_analyzer import DeepSeekAnalyzer from storage.db_client import IntelligenceDB import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DailyIntelligenceJob: def __init__(self): self.db IntelligenceDB() self.analyzer DeepSeekAnalyzer() # 定义你的监控源列表 self.target_sources [ {name: TechCrunch AI, url: https://techcrunch.com/category/artificial-intelligence/, selector: article}, {name: Hacker News Top, url: https://news.ycombinator.com/, selector: .athing}, # 添加更多源例如特定博客的RSS页面或GitHub趋势页 # {name: GitHub Trending Python, url: https://github.com/trending/python, selector: article}, ] def run(self): 核心任务执行函数 task_name fDailyIntelligence-{datetime.now().strftime(%Y%m%d)} logger.info(f开始执行情报收集任务: {task_name}) all_analysis [] for source in self.target_sources: try: logger.info(f抓取源: {source[name]}) # 1. 抓取 article_data fetch_url_sync(source[url], wait_for_selectorsource.get(selector)) if article_data.get(error) or not article_data.get(body): logger.warning(f抓取失败或内容为空: {source[name]}) continue # 2. 存储原始数据 article_id self.db.save_article(article_data[url], article_data[title], article_data[body]) # 3. AI分析 logger.info(f分析文章: {article_data[title][:50]}...) analysis_result self.analyzer.summarize_article(article_data[title], article_data[body]) analysis_result[title] article_data[title] # 把标题也加入结果方便后续生成简报 # 4. 存储分析结果 if article_id 0: self.db.save_analysis(article_id, analysis_result) all_analysis.append(analysis_result) time.sleep(2) # 礼貌性延迟避免对目标网站造成压力 except Exception as e: logger.error(f处理源 {source[name]} 时发生错误: {e}) self.db.log_task(task_name, FAILED, f{source[name]}: {str(e)}) # 5. 生成并保存每日简报 if all_analysis: try: logger.info(生成每日简报...) daily_brief self.analyzer.generate_daily_brief(all_analysis) # 这里可以将简报保存到文件或数据库也可以直接调用推送模块 with open(fbriefs/daily_brief_{datetime.now().strftime(%Y%m%d)}.md, w, encodingutf-8) as f: f.write(daily_brief) logger.info(每日简报已生成。) except Exception as e: logger.error(f生成简报失败: {e}) self.db.log_task(task_name, SUCCESS, f处理了 {len(all_analysis)} 篇文章) logger.info(f情报收集任务完成: {task_name}) def run_scheduler(): 启动调度器 job DailyIntelligenceJob() # 定义调度规则每天上午9点运行 schedule.every().day.at(09:00).do(job.run) # 也可以每小时运行一次用于测试 # schedule.every().hour.do(job.run) logger.info(情报系统调度器已启动等待执行计划...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次 if __name__ __main__: # 创建简报输出目录 import os os.makedirs(briefs, exist_okTrue) # 可以直接运行一次任务进行测试 # job DailyIntelligenceJob() # job.run() # 或者启动调度器 run_scheduler()5. 系统整合与运行验证现在我们已经有了所有核心模块。让我们把它们整合起来并验证整个流程是否能跑通。5.1 项目目录结构一个清晰的项目结构有助于维护。你的项目目录应大致如下ai-intelligence-system/ ├── .env # 环境变量API密钥等 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── main.py # 主程序入口 ├── crawler/ │ └── playwright_crawler.py ├── analyzer/ │ └── deepseek_analyzer.py ├── storage/ │ └── db_client.py ├── scheduler/ │ └── daily_scheduler.py ├── briefs/ # 简报输出目录 └── intelligence.db # SQLite数据库文件运行后生成5.2 编写主程序入口创建一个main.py文件提供两种运行模式立即执行一次任务或启动后台调度服务。# main.py import argparse import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from scheduler.daily_scheduler import DailyIntelligenceJob, run_scheduler def main(): parser argparse.ArgumentParser(descriptionAI 信息情报系统) parser.add_argument(--mode, choices[run-once, schedule], defaultrun-once, help运行模式: run-once (立即执行一次), schedule (启动定时调度器)) args parser.parse_args() if args.mode run-once: print(开始执行单次情报收集任务...) job DailyIntelligenceJob() job.run() print(单次任务执行完毕。) elif args.mode schedule: print(启动定时调度服务每天上午9点执行...) print(程序将持续运行按 CtrlC 终止。) try: run_scheduler() except KeyboardInterrupt: print(\n调度服务已停止。) if __name__ __main__: main()5.3 运行与验证安装所有依赖pip install -r requirements.txt playwright install chromium配置 API 密钥确保.env文件已正确创建并填写了你的 DeepSeek API Key。测试单次运行python main.py --mode run-once预期输出控制台会依次打印抓取、分析、存储的日志信息。验证结果检查是否生成了intelligence.db文件。可以使用 SQLite 命令行工具或 DB Browser for SQLite 查看raw_articles和analysis_results表是否有数据。检查briefs/目录下是否生成了当日的 Markdown 简报文件。启动后台调度服务适用于服务器长期运行# 在Linux/macOS上可以使用 nohup 或 systemd 让它在后台运行 python main.py --mode schedule 这将启动一个永不退出的进程每天上午9点自动执行任务。6. 常见问题与排查思路在搭建和运行过程中你可能会遇到以下问题。这里提供一份快速排查指南。问题现象可能原因排查方式解决方案Playwright 浏览器启动失败1. 未安装浏览器内核。2. 系统缺少依赖库多见于Linux。3. 权限问题。1. 运行playwright install --help查看安装状态。2. 查看错误日志通常会有明确提示。1. 执行playwright install chromium。2. 根据Playwright官方文档安装系统依赖。3. 确保有足够的文件执行权限。DeepSeek API 调用返回 401 或 403 错误1. API Key 错误或过期。2. API Key 未正确加载。3. 请求的模型名称不对。1. 检查.env文件中的DEEPSEEK_API_KEY是否正确。2. 在代码中打印DEEPSEEK_API_KEY的前几位确认已加载。3. 查看 DeepSeek 官方文档确认模型名和API地址。1. 在 DeepSeek 平台重新生成 Key。2. 确保.env文件在项目根目录且load_dotenv()被调用。3. 更新config.py中的API_BASE和model参数。抓取到的正文内容为空或杂乱1. 网页结构复杂通用选择器无效。2. 页面加载未完成。3. 网站有反爬机制。1. 使用浏览器开发者工具分析目标网页的DOM结构。2. 增加wait_for_selector参数或延长timeout。3. 检查raw_html字段看是否抓到了完整HTML。1. 为特定网站编写自定义的解析函数覆盖playwright_crawler.py中的提取逻辑。2. 尝试不同的等待策略如wait_untildomcontentloaded或等待特定元素。3. 考虑添加随机延迟、使用代理池针对反爬。AI 分析结果不符合 JSON 格式1. Prompt 指令不够清晰。2. 模型“不听话”输出了额外文本。3. 返回内容被截断。1. 打印出原始的analysis_text查看模型实际返回了什么。2. 检查发送给 API 的content是否过长。1. 优化 Prompt在 system message 中更严格地要求 JSON 格式。2. 在代码中添加更健壮的文本清理逻辑如移除 Markdown 代码块标记。3. 减少发送的文本长度或使用支持更长上下文的模型。定时任务不执行1. 服务器时间时区问题。2.schedule库在后台线程中运行异常。3. 主进程被杀死。1. 检查服务器系统时间。2. 查看任务日志表task_logs是否有错误记录。3. 检查进程是否存活。1. 在服务器上使用date命令确认时间或在代码中打印当前时间调试。2. 将调度任务包装在 try-catch 中确保单次失败不影响后续调度。3. 在生产环境使用更稳定的方案如Systemd Timer、Cron或Celery Beat。数据库文件被锁或操作失败1. 多线程/多进程同时写入。2. 数据库连接未正确关闭。1. 检查是否在多个地方实例化了IntelligenceDB。2. 使用with语句管理连接或确保close()被调用。1. 改为使用连接池或确保数据库操作是串行的。2. 在IntelligenceDB类中使用连接池或为每个线程创建独立连接。对于简单场景SQLite 的串行访问可以接受。7. 进阶优化与最佳实践基础系统跑通后你可以从以下几个方向进行深化和优化让它更强大、更智能、更稳定。7.1 信息源管理的工程化配置化将监控源列表URL、名称、选择器移出代码放入一个配置文件如sources.yaml或sources.json中便于动态增删改。# sources.yaml - name: AI 科技评论 url: https://www.aitechtalk.com/ type: blog extractor: css_selector # 指定提取器类型 selector: .post-content schedule: daily - name: GitHub Trending Go url: https://github.com/trending/go type: github_trending extractor: github_trending # 使用专用的GitHub解析器 schedule: daily专用解析器为不同类型的网站GitHub、Twitter、新闻站、论坛编写专用的解析类继承自一个基础解析器提高内容提取的准确率。7.2 提升 AI 分析质量与深度链式思考Chain-of-Thought对于复杂分析任务可以设计多轮对话的 Prompt让 AI 先拆解问题再逐步推理最后给出结论。多模型对比除了 DeepSeek可以集成其他 API如 OpenAI GPT、Claude、国内其他大模型对同一份内容进行交叉分析或根据任务类型选择最合适的模型。长期记忆与关联分析将历史分析结果向量化存入向量数据库如 Chroma、Milvus Lite。当分析新文章时可以先进行语义检索找出历史上的相关文章让 AI 在简报中建立关联发现趋势脉络。7.3 系统可靠性保障异常重试与降级在网络请求、API 调用失败时加入指数退避的重试机制。对于非核心源可以设置失败跳过保证整体任务完成。监控与告警除了记录日志到数据库可以集成告警功能。当连续多次任务失败或每日抓取文章数异常下降时通过邮件、钉钉机器人等方式通知你。数据备份与清理定期如每周对 SQLite 数据库进行备份。同时设置数据保留策略自动清理过早的原始数据只保留结构化分析结果和简报。7.4 输出与推送的多样化简报模板美化设计更精美的 Markdown/HTML 简报模板加入图表如通过分析关键词生成词云图。多通道推送集成多种推送方式让情报主动找到你。邮件使用smtplib和email库。钉钉/飞书群机器人调用 Webhook。Telegram Bot通过 Bot API 发送消息。生成静态页面将每日简报自动发布到 GitHub Pages 或你的个人博客形成一个可回溯的知识库。7.5 安全与成本控制API 成本DeepSeek 等 API 按 Token 收费。在代码中估算 Token 消耗对超长文本进行智能截断或分块处理。为每月使用量设置预算告警。速率限制遵守目标网站的robots.txt规则在抓取时添加合理的延迟time.sleep避免被封 IP。敏感信息永远不要将.env文件提交到版本控制系统。使用.gitignore将其忽略。8. 总结从项目到习惯通过本文你不仅获得了一套可运行的代码更重要的是掌握了一种用技术对抗信息不对称的思维模式。这个 AI 情报系统的价值会随着你的“调教”而不断增长。下一步你可以做什么定义你的专属情报维度除了技术新闻尝试监控你所在领域的竞品公司的官网、博客、招聘信息。你关注的技术领袖或投资人的社交媒体动态。与你技术栈相关的GitHub 仓库的 issue 和 release。特定专利数据库或论文预印本网站的更新。实现反馈闭环在生成的简报中加入简单的反馈机制如“这篇相关度打分是否准确”。收集你的反馈数据微调 AI 分析的 Prompt让系统越来越懂你。从监控到预测当积累了足够多的历史数据后你可以尝试让 AI 做一些简单的预测分析比如“根据近三个月‘向量数据库’相关文章的 sentiment 变化推测其市场热度趋势”。技术的最终目的是为人服务。这个由你亲手搭建、为你量身定制的 AI 情报官将成为你在信息洪流中的“导航仪”和“过滤器”。它不能替代你的深度思考但能为你节省出大量用于信息搜集和初步筛选的时间让你更专注于决策、创造和学习本身。开始动手吧从配置第一个信息源开始让你的信息差优势从今天开始建立。 基于DeepSeek与Playwright构建AI智能情报系统：从信息过载到精准推送
你是不是经常感觉信息过载但又怕错过关键信息每天花大量时间刷新闻、看行业报告、跟踪竞品动态结果发现信息要么滞后要么零散真正有价值的情报总是姗姗来迟。这种“信息差”不仅让你在决策上慢人一步更可能让你在技术选型、市场判断上陷入被动。今天我们不再依赖人工搜索和碎片化阅读。我将带你用 AI 技术亲手搭建一个自动化、智能化、可定制的个人信息情报系统。这个系统的核心不是简单地聚合 RSS而是利用 AI Agent 的能力理解你的需求主动从海量信息中筛选、分析、提炼最终将结构化的“情报”推送到你面前。我们将使用当下热门的DeepSeek大模型作为“大脑”结合浏览器自动化等工具作为“手脚”实现一个低成本、高效率的解决方案。读完本文你将能独立部署一个 7x24 小时运行的 AI 情报官。它不仅能帮你监控特定技术话题如“Vibe Coding 的最新实践”、跟踪开源项目动态、分析行业趋势还能生成每日简报让你用一杯咖啡的时间掌握别人需要一整天才能搜集到的信息。1. 这篇文章真正要解决的问题从“信息过载”到“情报主动送达”在深入技术细节之前我们必须先厘清一个核心问题我们到底要解决什么很多人误以为搭建情报系统就是做一个“高级爬虫”或“RSS 阅读器”这完全低估了 AI 的潜力也误解了“情报”的本质。情报Intelligence与信息Information有本质区别。信息是原始的、未经处理的信号比如一篇新的技术博客、一个 GitHub 的 commit、一条行业新闻。情报则是经过筛选、关联、分析后能直接支撑决策的结论例如“竞争对手 A 在项目 B 中开始大量使用向量数据库这可能意味着他们正在加强其产品的语义搜索能力。”传统的信息获取方式存在三大痛点被动接收你需要主动去各个平台“找”信息效率低下。缺乏分析即使找到了信息也需要你人工阅读、理解、关联耗时耗力。难以定制通用信息流无法精准匹配你的个人兴趣和专业领域。本文要构建的 AI 情报系统旨在通过技术手段彻底扭转这一局面变被动为主动系统按你设定的规则自动、持续地抓取目标信息源。变信息为情报利用大模型的理解和总结能力自动提炼核心观点、识别潜在影响、进行多源信息交叉验证。变通用为专属你可以用自然语言定义情报任务比如“监控所有关于‘浏览器自动化测试’与‘Playwright’结合的新文章并对比其与 Selenium 的方案优劣”。接下来的内容我们将聚焦于如何用可落地的技术栈实现这一构想核心会围绕DeepSeek API、浏览器自动化Playwright以及任务编排展开。你会发现搭建这样一个系统其技术门槛远没有想象中高。2. 核心概念与系统架构设计在动手写代码之前我们需要建立几个关键概念并设计出系统的整体架构。这能帮助你在后续配置和开发时清楚每一部分在扮演什么角色。2.1 关键概念解析AI Agent智能体这是我们系统的“大脑”。它不是一个简单的聊天机器人而是一个能根据目标如“获取今日AI编程领域动态”自主规划步骤思考、调用工具行动、并评估结果观察的智能程序。在本系统中DeepSeek 模型将承担 Agent 的“思考”和“分析”核心。工具ToolsAgent 的“手脚”。Agent 本身无法直接浏览网页或读取文件它需要通过调用预设的工具来与外界交互。我们将为 Agent 配备的关键工具包括网页抓取工具用于访问指定 URL 并获取页面内容。搜索引擎工具可选用于扩大信息发现范围。数据存储工具用于保存历史情报和原始数据。任务编排Orchestration决定系统何时运行、以什么顺序执行哪些任务的中枢。例如每天上午9点自动执行“监控任务”然后将结果交给“分析任务”最后触发“推送任务”。我们将使用简单的Cron 定时任务或更强大的工作流引擎如 Apache Airflow 的轻量替代来实现。信息源Sources系统监控的目标。这不仅仅是 RSS 订阅源还可以是特定技术博客、新闻网站。GitHub 仓库的 Releases、Issues、Commits。社交媒体如 Twitter/X的关键词或用户。行业报告网站、论坛的热门板块。2.2 系统架构设计一个健壮的情报系统应该模块清晰、易于扩展。以下是推荐的核心架构[信息源列表] | v [定时调度器] (Cron Job / Workflow Engine) | v [采集器模块] (使用 Playwright 进行浏览器自动化抓取) | \ | \ (原始 HTML/数据) | \ v v [内容解析器] [原始存储] (如 SQLite/JSON文件) | v [AI 分析引擎] (调用 DeepSeek API) | \ | \ (分析结果摘要、分类、情感、关联) | \ v v [情报生成器] [分析结果存储] | v [推送模块] (邮件、Slack、钉钉、Telegram) | v [用户终端]各模块职责调度器系统的发条按计划触发整个流水线。采集器系统的眼睛和手负责从信息源获取原始内容。我们选择Playwright而非简单requests是因为现代网站大量使用 JavaScript 渲染Playwright 能模拟真实浏览器行为确保抓到完整内容。解析器从杂乱无章的 HTML 中提取出干净的标题、正文、发布时间等结构化数据。AI 分析引擎系统的核心大脑。它将解析后的文本发送给 DeepSeek并提出精准的指令Prompt要求其进行总结、提取关键词、判断相关性、甚至进行情感分析。情报生成器将 AI 的分析结果格式化为易于阅读的报告如 Markdown。推送模块将最终报告送达给你。这个架构的优点是松耦合。你可以轻易替换其中任何一个模块例如将 DeepSeek 换成其他大模型 API或将邮件推送改为微信机器人。3. 环境准备与工具选型工欲善其事必先利其器。我们将选择一个简单、高效且免费资源丰富的技术栈确保每个人都能快速上手。3.1 基础运行环境操作系统推荐 Linux (Ubuntu 20.04) 或 macOS。Windows 也可行但部分命令行操作略有差异。Python 版本Python 3.9。这是大多数现代AI库的基础要求。包管理使用pip或更推荐的poetry/pipenv来管理项目依赖避免环境冲突。3.2 核心工具与库我们将基于 Python 生态来构建。请在你的项目目录下创建一个requirements.txt文件或直接安装以下库# 创建并进入项目目录 mkdir ai-intelligence-system cd ai-intelligence-system # 初始化Python虚拟环境强烈推荐 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install playwright # 浏览器自动化用于抓取动态网页 pip install beautifulsoup4 # HTML 解析库用于从Playwright获取的页面中提取内容 pip install requests # 用于调用DeepSeek API pip install schedule # 轻量级定时任务库用于简单调度 # 或者使用更强大的APScheduler # pip install apscheduler pip install python-dotenv # 用于管理环境变量如API密钥 pip install sqlite3 # Python内置用于轻量级数据存储安装 Playwright 浏览器 Playwright 需要安装它自带的浏览器内核。playwright install chromium3.3 获取 DeepSeek API 密钥本系统的“智能”来源于 DeepSeek 大模型。你需要前往 DeepSeek 开放平台 注册账号并获取 API Key。登录后在控制台找到“API Keys”部分。创建一个新的 Key并妥善保存。它看起来像一串长字符sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx。重要安全提示永远不要将 API Key 直接硬编码在代码中或上传到 GitHub 等公开仓库。我们将使用环境变量来管理它。在项目根目录创建一个.env文件# .env 文件内容 DEEPSEEK_API_KEYsk-你的实际API密钥 DEEPSEEK_API_BASEhttps://api.deepseek.com # API基础地址请以官方文档为准并在代码中通过python-dotenv加载# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量 DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) API_BASE os.getenv(DEEPSEEK_API_BASE, https://api.deepseek.com)4. 核心模块实现分步构建你的 AI 情报官现在我们开始动手实现架构图中的每一个核心模块。我们将采用自底向上的方式先实现基础工具再组合成完整流程。4.1 模块一智能网页采集器Playwright为什么不用简单的requests.get()因为很多网站如技术博客、新闻站采用前端框架React, Vue.js动态加载内容直接请求只能得到空壳 HTML。Playwright 能模拟真人浏览器操作等待页面完全加载后再获取内容。# crawler/playwright_crawler.py import asyncio from playwright.async_api import async_playwright from bs4 import BeautifulSoup import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class PlaywrightCrawler: def __init__(self, headlessTrue): # headlessTrue 表示无头模式不显示浏览器界面 self.headless headless async def fetch_url(self, url: str, wait_for_selector: str None, timeout: int 30000): 使用 Playwright 抓取指定 URL 的完整内容。 :param url: 目标网址 :param wait_for_selector: 可选等待某个CSS选择器出现后再抓取用于应对动态加载 :param timeout: 超时时间毫秒 :return: 包含标题和正文文本的字典 async with async_playwright() as p: # 启动浏览器推荐使用 Chromium兼容性好 browser await p.chromium.launch(headlessself.headless) context await browser.new_context() page await context.new_page() try: logger.info(f正在访问: {url}) await page.goto(url, wait_untilnetworkidle, timeouttimeout) # 等待网络空闲 # 如果提供了选择器则等待该元素出现 if wait_for_selector: await page.wait_for_selector(wait_for_selector, timeout10000) # 获取页面完整的 HTML 内容 html_content await page.content() # 使用 BeautifulSoup 解析 HTML提取标题和正文 soup BeautifulSoup(html_content, html.parser) # 提取标题 - 尝试多种可能的选择器 title soup.title.string if soup.title else # 如果 og:title 存在有时它更准确 og_title soup.find(meta, propertyog:title) if og_title and og_title.get(content): title og_title[content] # 提取正文 - 这是一个简化的示例实际中可能需要针对不同网站定制 # 常见策略寻找 article, main 标签或者包含大量文本的 div article_body article_tag soup.find(article) or soup.find(main) or soup.find(div, class_lambda c: c and (content in c or post in c or article in c)) if article_tag: article_body article_tag.get_text(separator\n, stripTrue) else: # 备选方案获取整个 body 的文本噪声会多 article_body soup.body.get_text(separator\n, stripTrue) if soup.body else logger.info(f成功抓取: {title[:50]}...) return { url: url, title: title.strip(), body: article_body.strip(), raw_html: html_content[:5000] # 保存部分HTML供调试避免过大 } except Exception as e: logger.error(f抓取 {url} 时出错: {e}) return {url: url, title: , body: , error: str(e)} finally: await browser.close() # 同步函数包装便于在非异步环境中调用 def fetch_url_sync(url: str, **kwargs): 同步接口内部运行异步函数 crawler PlaywrightCrawler(headlessTrue) return asyncio.run(crawler.fetch_url(url, **kwargs)) if __name__ __main__: # 测试代码 test_url https://example.com # 请替换为一个真实的动态网站进行测试 result fetch_url_sync(test_url, wait_for_selectorarticle) print(f标题: {result[title]}) print(f正文预览: {result[body][:200]}...)关键点解析wait_untilnetworkidle这是 Playwright 的核心优势之一它会等待页面网络请求基本停止后再返回确保动态内容已加载。wait_for_selector对于某些特定网站你可以指定一个元素如article作为页面加载完成的标志更加精准。正文提取代码中的正文提取逻辑是通用的但效果可能因网站结构而异。在真实项目中你可能需要为重要的信息源编写特定的提取规则可配置化这是提升情报质量的关键一步。4.2 模块二AI 分析引擎DeepSeek API 调用这是系统的“大脑”。我们将定义一个类专门负责与 DeepSeek API 交互并封装几个常用的分析任务。# analyzer/deepseek_analyzer.py import requests import json import logging from typing import List, Dict, Any from config import DEEPSEEK_API_KEY, API_BASE # 导入配置文件 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DeepSeekAnalyzer: def __init__(self, api_key: str None, model: str deepseek-chat): self.api_key api_key or DEEPSEEK_API_KEY if not self.api_key: raise ValueError(DeepSeek API Key 未设置。请在 .env 文件中配置 DEEPSEEK_API_KEY) self.model model self.api_url f{API_BASE}/chat/completions self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } def _call_api(self, messages: List[Dict[str, str]]) - str: 调用 DeepSeek API 的基础方法 payload { model: self.model, messages: messages, temperature: 0.2, # 较低的温度使输出更稳定、更事实性 max_tokens: 2000 } try: response requests.post(self.api_url, headersself.headers, jsonpayload, timeout60) response.raise_for_status() # 如果状态码不是200抛出异常 result response.json() return result[choices][0][message][content] except requests.exceptions.RequestException as e: logger.error(fAPI 请求失败: {e}) if hasattr(e.response, text): logger.error(f错误响应: {e.response.text}) return except (KeyError, IndexError, json.JSONDecodeError) as e: logger.error(f解析 API 响应失败: {e}) return def summarize_article(self, title: str, content: str) - Dict[str, Any]: 总结一篇文章的核心内容。 返回结构化的摘要信息。 prompt f请对以下技术文章进行专业摘要和分析。 文章标题{title} 文章内容 {content[:6000]} # 限制内容长度避免超出Token限制 请以JSON格式返回分析结果包含以下字段 1. summary: 文章的核心内容摘要不超过300字。 2. key_points: 一个数组列出3-5个最关键的技术点或观点。 3. keywords: 一个数组提取5-8个核心关键词。 4. relevance_score: 一个1-10的整数评估这篇文章与“软件开发、AI编程、前沿技术”领域的相关性。 5. sentiment: 文章的整体情感倾向可选值: 积极, 消极, 中性。 messages [ {role: system, content: 你是一个专注于技术领域的信息分析专家。请准确、简洁地分析文章。}, {role: user, content: prompt} ] analysis_text self._call_api(messages) # 尝试从返回文本中解析JSON。DeepSeek有时会在JSON外加 Markdown 代码块标记。 try: # 清理可能的代码块标记 if analysis_text.startswith(json): analysis_text analysis_text[7:-3] # 移除 json 和结尾的 elif analysis_text.startswith(): analysis_text analysis_text[3:-3] analysis_data json.loads(analysis_text.strip()) return analysis_data except json.JSONDecodeError: logger.warning(f无法解析AI返回的JSON返回原始文本。原文: {analysis_text[:200]}) # 返回一个包含原始文本的默认结构 return { summary: analysis_text, key_points: [], keywords: [], relevance_score: 5, sentiment: 中性, raw_output: analysis_text } def generate_daily_brief(self, articles_analysis: List[Dict]) - str: 根据多篇文章的分析结果生成一份每日简报。 # 将分析结果格式化为文本作为新的Prompt输入 input_text 以下是今日监控到的多篇文章分析结果\n\n for i, analysis in enumerate(articles_analysis, 1): input_text f【文章{i}】{analysis.get(title, 无标题)}\n input_text f摘要{analysis.get(summary, 无摘要)}\n input_text f关键词{, .join(analysis.get(keywords, []))}\n input_text f相关性评分{analysis.get(relevance_score, 5)}/10\n input_text - * 40 \n prompt f{input_text} 请基于以上信息生成一份面向技术负责人或开发者的《每日技术情报简报》。 简报要求 1. 用一段话概括今日整体技术风向。 2. 分点列出最重要的2-3个技术动态或趋势并简要说明其潜在影响。 3. 给出一个“今日值得关注”的总结指出最值得深入阅读或跟进的一两篇文章或话题。 请使用清晰、专业的Markdown格式输出。 messages [ {role: system, content: 你是一名资深技术布道师擅长从零散信息中提炼出有洞察力的趋势简报。}, {role: user, content: prompt} ] return self._call_api(messages) # 测试分析功能 if __name__ __main__: analyzer DeepSeekAnalyzer() # 假设我们已经抓取了一篇文章 test_data { title: Vibe Coding一种新兴的AI辅助编程范式, body: ...这里是一大段关于Vibe Coding的文章内容... } result analyzer.summarize_article(test_data[title], test_data[body]) print(json.dumps(result, indent2, ensure_asciiFalse))关键点解析结构化 Prompt 工程我们通过精心设计的 Prompt引导 AI 返回结构化的 JSON 数据这极大方便了后续的数据处理。Prompt 中明确了角色、任务和输出格式。Token 长度管理content[:6000]用于限制发送给 API 的文本长度防止超出模型上下文限制并控制 API 成本。对于超长文章更复杂的策略是分段总结再聚合。错误处理与降级API 调用可能失败AI 也可能不按格式返回。代码中包含了异常捕获和降级处理返回原始文本保证系统的鲁棒性。温度参数temperature0.2设置较低使模型输出更确定、更少“创造性”更适合事实性分析任务。4.3 模块三数据存储与任务调度我们需要一个地方来存储抓取到的原始数据、分析结果以及记录任务运行状态。同时需要一个调度器来定时执行任务。轻量级数据存储SQLite# storage/db_client.py import sqlite3 import json from datetime import datetime from typing import List, Dict, Any class IntelligenceDB: def __init__(self, db_path: str intelligence.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._init_tables() def _init_tables(self): 初始化数据库表 cursor self.conn.cursor() # 原始文章表 cursor.execute( CREATE TABLE IF NOT EXISTS raw_articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE, title TEXT, body TEXT, fetch_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ) # 分析结果表 cursor.execute( CREATE TABLE IF NOT EXISTS analysis_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER, summary TEXT, key_points TEXT, -- 存储为JSON字符串 keywords TEXT, -- 存储为JSON字符串 relevance_score INTEGER, sentiment TEXT, analysis_time DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (article_id) REFERENCES raw_articles (id) ) ) # 任务运行日志表 cursor.execute( CREATE TABLE IF NOT EXISTS task_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_name TEXT, status TEXT, -- SUCCESS, FAILED message TEXT, run_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_article(self, url: str, title: str, body: str) - int: 保存抓取到的文章返回文章ID cursor self.conn.cursor() try: cursor.execute( INSERT OR REPLACE INTO raw_articles (url, title, body) VALUES (?, ?, ?), (url, title, body) ) self.conn.commit() return cursor.lastrowid except sqlite3.Error as e: print(f保存文章失败: {e}) return -1 def save_analysis(self, article_id: int, analysis_data: Dict[str, Any]): 保存AI分析结果 cursor self.conn.cursor() cursor.execute( INSERT INTO analysis_results (article_id, summary, key_points, keywords, relevance_score, sentiment) VALUES (?, ?, ?, ?, ?, ?) , ( article_id, analysis_data.get(summary), json.dumps(analysis_data.get(key_points, []), ensure_asciiFalse), json.dumps(analysis_data.get(keywords, []), ensure_asciiFalse), analysis_data.get(relevance_score, 5), analysis_data.get(sentiment, 中性) )) self.conn.commit() def log_task(self, task_name: str, status: str, message: str ): 记录任务运行日志 cursor self.conn.cursor() cursor.execute( INSERT INTO task_logs (task_name, status, message) VALUES (?, ?, ?), (task_name, status, message) ) self.conn.commit() def close(self): self.conn.close()简单任务调度器 对于个人使用schedule库足够轻量。对于更复杂的依赖关系可以考虑APScheduler。# scheduler/daily_scheduler.py import schedule import time import threading from datetime import datetime from crawler.playwright_crawler import fetch_url_sync from analyzer.deepseek_analyzer import DeepSeekAnalyzer from storage.db_client import IntelligenceDB import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DailyIntelligenceJob: def __init__(self): self.db IntelligenceDB() self.analyzer DeepSeekAnalyzer() # 定义你的监控源列表 self.target_sources [ {name: TechCrunch AI, url: https://techcrunch.com/category/artificial-intelligence/, selector: article}, {name: Hacker News Top, url: https://news.ycombinator.com/, selector: .athing}, # 添加更多源例如特定博客的RSS页面或GitHub趋势页 # {name: GitHub Trending Python, url: https://github.com/trending/python, selector: article}, ] def run(self): 核心任务执行函数 task_name fDailyIntelligence-{datetime.now().strftime(%Y%m%d)} logger.info(f开始执行情报收集任务: {task_name}) all_analysis [] for source in self.target_sources: try: logger.info(f抓取源: {source[name]}) # 1. 抓取 article_data fetch_url_sync(source[url], wait_for_selectorsource.get(selector)) if article_data.get(error) or not article_data.get(body): logger.warning(f抓取失败或内容为空: {source[name]}) continue # 2. 存储原始数据 article_id self.db.save_article(article_data[url], article_data[title], article_data[body]) # 3. AI分析 logger.info(f分析文章: {article_data[title][:50]}...) analysis_result self.analyzer.summarize_article(article_data[title], article_data[body]) analysis_result[title] article_data[title] # 把标题也加入结果方便后续生成简报 # 4. 存储分析结果 if article_id 0: self.db.save_analysis(article_id, analysis_result) all_analysis.append(analysis_result) time.sleep(2) # 礼貌性延迟避免对目标网站造成压力 except Exception as e: logger.error(f处理源 {source[name]} 时发生错误: {e}) self.db.log_task(task_name, FAILED, f{source[name]}: {str(e)}) # 5. 生成并保存每日简报 if all_analysis: try: logger.info(生成每日简报...) daily_brief self.analyzer.generate_daily_brief(all_analysis) # 这里可以将简报保存到文件或数据库也可以直接调用推送模块 with open(fbriefs/daily_brief_{datetime.now().strftime(%Y%m%d)}.md, w, encodingutf-8) as f: f.write(daily_brief) logger.info(每日简报已生成。) except Exception as e: logger.error(f生成简报失败: {e}) self.db.log_task(task_name, SUCCESS, f处理了 {len(all_analysis)} 篇文章) logger.info(f情报收集任务完成: {task_name}) def run_scheduler(): 启动调度器 job DailyIntelligenceJob() # 定义调度规则每天上午9点运行 schedule.every().day.at(09:00).do(job.run) # 也可以每小时运行一次用于测试 # schedule.every().hour.do(job.run) logger.info(情报系统调度器已启动等待执行计划...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次 if __name__ __main__: # 创建简报输出目录 import os os.makedirs(briefs, exist_okTrue) # 可以直接运行一次任务进行测试 # job DailyIntelligenceJob() # job.run() # 或者启动调度器 run_scheduler()5. 系统整合与运行验证现在我们已经有了所有核心模块。让我们把它们整合起来并验证整个流程是否能跑通。5.1 项目目录结构一个清晰的项目结构有助于维护。你的项目目录应大致如下ai-intelligence-system/ ├── .env # 环境变量API密钥等 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── main.py # 主程序入口 ├── crawler/ │ └── playwright_crawler.py ├── analyzer/ │ └── deepseek_analyzer.py ├── storage/ │ └── db_client.py ├── scheduler/ │ └── daily_scheduler.py ├── briefs/ # 简报输出目录 └── intelligence.db # SQLite数据库文件运行后生成5.2 编写主程序入口创建一个main.py文件提供两种运行模式立即执行一次任务或启动后台调度服务。# main.py import argparse import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from scheduler.daily_scheduler import DailyIntelligenceJob, run_scheduler def main(): parser argparse.ArgumentParser(descriptionAI 信息情报系统) parser.add_argument(--mode, choices[run-once, schedule], defaultrun-once, help运行模式: run-once (立即执行一次), schedule (启动定时调度器)) args parser.parse_args() if args.mode run-once: print(开始执行单次情报收集任务...) job DailyIntelligenceJob() job.run() print(单次任务执行完毕。) elif args.mode schedule: print(启动定时调度服务每天上午9点执行...) print(程序将持续运行按 CtrlC 终止。) try: run_scheduler() except KeyboardInterrupt: print(\n调度服务已停止。) if __name__ __main__: main()5.3 运行与验证安装所有依赖pip install -r requirements.txt playwright install chromium配置 API 密钥确保.env文件已正确创建并填写了你的 DeepSeek API Key。测试单次运行python main.py --mode run-once预期输出控制台会依次打印抓取、分析、存储的日志信息。验证结果检查是否生成了intelligence.db文件。可以使用 SQLite 命令行工具或 DB Browser for SQLite 查看raw_articles和analysis_results表是否有数据。检查briefs/目录下是否生成了当日的 Markdown 简报文件。启动后台调度服务适用于服务器长期运行# 在Linux/macOS上可以使用 nohup 或 systemd 让它在后台运行 python main.py --mode schedule 这将启动一个永不退出的进程每天上午9点自动执行任务。6. 常见问题与排查思路在搭建和运行过程中你可能会遇到以下问题。这里提供一份快速排查指南。问题现象可能原因排查方式解决方案Playwright 浏览器启动失败1. 未安装浏览器内核。2. 系统缺少依赖库多见于Linux。3. 权限问题。1. 运行playwright install --help查看安装状态。2. 查看错误日志通常会有明确提示。1. 执行playwright install chromium。2. 根据Playwright官方文档安装系统依赖。3. 确保有足够的文件执行权限。DeepSeek API 调用返回 401 或 403 错误1. API Key 错误或过期。2. API Key 未正确加载。3. 请求的模型名称不对。1. 检查.env文件中的DEEPSEEK_API_KEY是否正确。2. 在代码中打印DEEPSEEK_API_KEY的前几位确认已加载。3. 查看 DeepSeek 官方文档确认模型名和API地址。1. 在 DeepSeek 平台重新生成 Key。2. 确保.env文件在项目根目录且load_dotenv()被调用。3. 更新config.py中的API_BASE和model参数。抓取到的正文内容为空或杂乱1. 网页结构复杂通用选择器无效。2. 页面加载未完成。3. 网站有反爬机制。1. 使用浏览器开发者工具分析目标网页的DOM结构。2. 增加wait_for_selector参数或延长timeout。3. 检查raw_html字段看是否抓到了完整HTML。1. 为特定网站编写自定义的解析函数覆盖playwright_crawler.py中的提取逻辑。2. 尝试不同的等待策略如wait_untildomcontentloaded或等待特定元素。3. 考虑添加随机延迟、使用代理池针对反爬。AI 分析结果不符合 JSON 格式1. Prompt 指令不够清晰。2. 模型“不听话”输出了额外文本。3. 返回内容被截断。1. 打印出原始的analysis_text查看模型实际返回了什么。2. 检查发送给 API 的content是否过长。1. 优化 Prompt在 system message 中更严格地要求 JSON 格式。2. 在代码中添加更健壮的文本清理逻辑如移除 Markdown 代码块标记。3. 减少发送的文本长度或使用支持更长上下文的模型。定时任务不执行1. 服务器时间时区问题。2.schedule库在后台线程中运行异常。3. 主进程被杀死。1. 检查服务器系统时间。2. 查看任务日志表task_logs是否有错误记录。3. 检查进程是否存活。1. 在服务器上使用date命令确认时间或在代码中打印当前时间调试。2. 将调度任务包装在 try-catch 中确保单次失败不影响后续调度。3. 在生产环境使用更稳定的方案如Systemd Timer、Cron或Celery Beat。数据库文件被锁或操作失败1. 多线程/多进程同时写入。2. 数据库连接未正确关闭。1. 检查是否在多个地方实例化了IntelligenceDB。2. 使用with语句管理连接或确保close()被调用。1. 改为使用连接池或确保数据库操作是串行的。2. 在IntelligenceDB类中使用连接池或为每个线程创建独立连接。对于简单场景SQLite 的串行访问可以接受。7. 进阶优化与最佳实践基础系统跑通后你可以从以下几个方向进行深化和优化让它更强大、更智能、更稳定。7.1 信息源管理的工程化配置化将监控源列表URL、名称、选择器移出代码放入一个配置文件如sources.yaml或sources.json中便于动态增删改。# sources.yaml - name: AI 科技评论 url: https://www.aitechtalk.com/ type: blog extractor: css_selector # 指定提取器类型 selector: .post-content schedule: daily - name: GitHub Trending Go url: https://github.com/trending/go type: github_trending extractor: github_trending # 使用专用的GitHub解析器 schedule: daily专用解析器为不同类型的网站GitHub、Twitter、新闻站、论坛编写专用的解析类继承自一个基础解析器提高内容提取的准确率。7.2 提升 AI 分析质量与深度链式思考Chain-of-Thought对于复杂分析任务可以设计多轮对话的 Prompt让 AI 先拆解问题再逐步推理最后给出结论。多模型对比除了 DeepSeek可以集成其他 API如 OpenAI GPT、Claude、国内其他大模型对同一份内容进行交叉分析或根据任务类型选择最合适的模型。长期记忆与关联分析将历史分析结果向量化存入向量数据库如 Chroma、Milvus Lite。当分析新文章时可以先进行语义检索找出历史上的相关文章让 AI 在简报中建立关联发现趋势脉络。7.3 系统可靠性保障异常重试与降级在网络请求、API 调用失败时加入指数退避的重试机制。对于非核心源可以设置失败跳过保证整体任务完成。监控与告警除了记录日志到数据库可以集成告警功能。当连续多次任务失败或每日抓取文章数异常下降时通过邮件、钉钉机器人等方式通知你。数据备份与清理定期如每周对 SQLite 数据库进行备份。同时设置数据保留策略自动清理过早的原始数据只保留结构化分析结果和简报。7.4 输出与推送的多样化简报模板美化设计更精美的 Markdown/HTML 简报模板加入图表如通过分析关键词生成词云图。多通道推送集成多种推送方式让情报主动找到你。邮件使用smtplib和email库。钉钉/飞书群机器人调用 Webhook。Telegram Bot通过 Bot API 发送消息。生成静态页面将每日简报自动发布到 GitHub Pages 或你的个人博客形成一个可回溯的知识库。7.5 安全与成本控制API 成本DeepSeek 等 API 按 Token 收费。在代码中估算 Token 消耗对超长文本进行智能截断或分块处理。为每月使用量设置预算告警。速率限制遵守目标网站的robots.txt规则在抓取时添加合理的延迟time.sleep避免被封 IP。敏感信息永远不要将.env文件提交到版本控制系统。使用.gitignore将其忽略。8. 总结从项目到习惯通过本文你不仅获得了一套可运行的代码更重要的是掌握了一种用技术对抗信息不对称的思维模式。这个 AI 情报系统的价值会随着你的“调教”而不断增长。下一步你可以做什么定义你的专属情报维度除了技术新闻尝试监控你所在领域的竞品公司的官网、博客、招聘信息。你关注的技术领袖或投资人的社交媒体动态。与你技术栈相关的GitHub 仓库的 issue 和 release。特定专利数据库或论文预印本网站的更新。实现反馈闭环在生成的简报中加入简单的反馈机制如“这篇相关度打分是否准确”。收集你的反馈数据微调 AI 分析的 Prompt让系统越来越懂你。从监控到预测当积累了足够多的历史数据后你可以尝试让 AI 做一些简单的预测分析比如“根据近三个月‘向量数据库’相关文章的 sentiment 变化推测其市场热度趋势”。技术的最终目的是为人服务。这个由你亲手搭建、为你量身定制的 AI 情报官将成为你在信息洪流中的“导航仪”和“过滤器”。它不能替代你的深度思考但能为你节省出大量用于信息搜集和初步筛选的时间让你更专注于决策、创造和学习本身。开始动手吧从配置第一个信息源开始让你的信息差优势从今天开始建立。