DeerFlow研究助手实战：Python爬虫+LLM实现智能数据采集与分析

DeerFlow研究助手实战：Python爬虫+LLM实现智能数据采集与分析

1. 引言

想象一下这个场景：你正在做一个市场分析项目，需要收集某个行业的最新动态、竞品信息和用户反馈。传统做法是什么？打开浏览器，手动搜索、复制、粘贴，然后整理到Excel里，最后再写分析报告。整个过程繁琐、耗时，而且容易遗漏关键信息。

现在，有了DeerFlow这样的深度研究框架，这一切都可以自动化完成。DeerFlow是一个社区驱动的多智能体系统，它把大语言模型（LLM）和专业工具（比如网络搜索、爬虫、Python代码执行）结合起来，帮你完成从数据采集到分析报告的全流程。

今天我要分享的，就是如何用DeerFlow的Researcher Agent和Coder Agent，配合Python爬虫技术，构建一个智能数据采集与分析系统。我会用一个完整的案例，带你走一遍从网页数据采集到智能分析的自动化流程。

2. DeerFlow核心架构快速了解

在深入实战之前，先简单了解一下DeerFlow是怎么工作的。它的架构设计得很清晰，主要由四个核心组件组成：

协调器（Coordinator）：相当于项目的总指挥，负责接收你的研究需求，然后决定下一步该怎么做。

规划器（Planner）：这是系统的“大脑”，它会分析你的研究目标，制定详细的执行计划。比如，需要搜索哪些信息、要爬取哪些网站、需要做什么数据分析等等。

研究团队（Research Team）：这里有两个专业角色：

• 研究员（Researcher）：负责信息搜集，可以用各种搜索引擎和爬虫工具
• 编码员（Coder）：负责技术实现，主要是用Python代码处理数据、执行分析

报告员（Reporter）：最后把所有的发现汇总起来，生成一份完整的研究报告。

整个系统基于LangGraph构建，各个组件之间通过状态传递信息，形成一个灵活的工作流。你可以把它想象成一个智能的研究团队，每个成员各司其职，协同完成复杂的任务。

3. 环境准备与快速部署

3.1 系统要求

要运行DeerFlow，你的系统需要满足以下基本要求：

• Python 3.12或更高版本
• Node.js 22或更高版本（如果你要用Web UI的话）

3.2 一键安装

DeerFlow推荐使用uv来管理Python环境，这个工具会自动帮你创建虚拟环境并安装所有依赖，非常方便。

# 克隆项目仓库 git clone https://github.com/bytedance/deer-flow.git cd deer-flow # 安装所有依赖（uv会自动处理Python环境和包管理） uv sync # 复制配置文件模板 cp .env.example .env cp conf.yaml.example conf.yaml

3.3 关键配置

安装完成后，需要配置几个关键的API密钥。打开刚才复制的.env文件，你会看到类似这样的内容：

# 搜索引擎配置（选一个配置即可） # SEARCH_API=tavily # TAVILY_API_KEY=your_tavily_api_key_here # SEARCH_API=infoquest # INFOQUEST_API_KEY=your_infoquest_api_key_here # 如果你要用语音合成功能 # VOLCENGINE_TTS_ACCESS_KEY=your_tts_access_key # VOLCENGINE_TTS_SECRET_KEY=your_tts_secret_key

对于爬虫功能，DeerFlow支持两种引擎：

• Jina Reader：免费的网页内容提取工具，基础功能不需要API密钥
• InfoQuest：字节推出的智能搜索与爬取工具集，功能更强大

在conf.yaml文件中配置爬虫引擎：

CRAWLER_ENGINE: # 引擎类型："jina"（默认）或 "infoquest" engine: jina

如果你刚开始接触，建议先用Jina，因为它不需要API密钥就能用基础功能。

3.4 启动项目

最简单的启动方式是使用控制台界面：

# 启动控制台UI uv run main.py

启动后，你会看到一个交互式界面，可以直接输入研究问题，系统就会开始工作了。

如果你想用更直观的Web界面，也可以启动Web UI：

# 先安装Web UI依赖（如果需要的话） cd web pnpm install cd .. # 启动Web UI（macOS/Linux） ./bootstrap.sh -d # Windows系统用这个 bootstrap.bat -d

启动成功后，在浏览器打开http://localhost:3000就能看到Web界面了。

4. 实战案例：智能电商数据监控系统

现在进入正题，我用一个实际的电商数据监控案例，来展示DeerFlow的爬虫和数据分析能力。

4.1 场景设定

假设你是一家电商公司的数据分析师，需要定期监控竞品的产品信息、价格变化和用户评价。传统做法是每天手动去各个电商平台查看，既费时又容易出错。

我们的目标是：用DeerFlow构建一个自动化系统，每天自动采集指定竞品的信息，分析价格趋势，生成监控报告。

4.2 研究计划制定

首先，我们让DeerFlow的规划器来制定研究计划。在控制台输入：

分析小米14手机在京东、天猫、拼多多三大平台的价格趋势和用户评价，需要采集近一个月的价格数据和用户评论，最后给出购买建议。

规划器会分析这个需求，然后生成一个结构化的执行计划。在Web UI里，你能看到类似这样的计划：

研究计划：小米14手机多平台价格监控分析 步骤1：搜索小米14手机的基本信息和市场定位 步骤2：采集京东平台的价格数据和用户评价 步骤3：采集天猫平台的价格数据和用户评价 步骤4：采集拼多多平台的价格数据和用户评价 步骤5：分析三个平台的价格趋势对比 步骤6：汇总用户评价的关键点 步骤7：生成购买建议报告

你可以审查这个计划，如果觉得哪里不合适，可以直接用自然语言修改。比如你可以说：“[EDIT PLAN] 增加对历史价格曲线的分析，看看有没有促销规律。”

4.3 Researcher Agent调用爬虫工具

计划确定后，研究团队开始工作。Researcher Agent会负责信息搜集。对于电商数据采集，我们主要用爬虫工具。

DeerFlow内置了Jina Reader工具，可以智能提取网页内容。Researcher Agent会根据计划，自动调用这个工具来采集数据。

让我写一个简单的示例，展示Researcher Agent如何工作：

# 这是一个简化的示例，展示Researcher Agent调用爬虫的逻辑 # 在实际的DeerFlow中，这些调用是自动完成的 from langchain_community.tools import JinaReader # 初始化Jina Reader工具 jina_reader = JinaReader() # Researcher Agent会分析需要采集的网站 target_urls = [ "https://item.jd.com/10012345678.html", # 京东商品页 "https://detail.tmall.com/item.htm?id=123456789", # 天猫商品页 "https://yangkeduo.com/goods.html?goods_id=987654321" # 拼多多商品页 ] # 对每个URL进行内容提取 for url in target_urls: print(f"正在采集: {url}") # 调用Jina Reader提取网页主要内容 # 在实际的DeerFlow中，Researcher Agent会自动处理这些调用 content = jina_reader.run(url) # 提取关键信息：价格、评价数、评分等 # Jina Reader会智能识别网页结构，提取正文内容 print(f"采集到的内容长度: {len(content)} 字符") # 这里可以添加进一步的信息提取逻辑 # 比如用LLM从提取的内容中解析出价格、评价等信息

在实际运行中，你不需要手动写这些代码。Researcher Agent会根据研究计划，自动决定要爬取哪些网站，调用相应的工具，然后把采集到的信息整理好，传递给下一个环节。

4.4 Coder Agent执行数据清洗与分析

数据采集回来后，往往是杂乱无章的。这时候Coder Agent就派上用场了。它负责用Python代码处理这些数据。

假设我们已经采集到了三个平台的数据，现在需要清洗和分析。Coder Agent会自动编写和执行相应的Python代码。

# Coder Agent生成的代码示例：数据清洗与价格分析 import pandas as pd import numpy as np from datetime import datetime, timedelta import matplotlib.pyplot as plt import seaborn as sns # 假设我们已经有了采集到的原始数据 # 这里模拟一些数据 raw_data = { 'platform': ['京东', '京东', '京东', '天猫', '天猫', '天猫', '拼多多', '拼多多', '拼多多'], 'date': ['2024-01-01', '2024-01-02', '2024-01-03', '2024-01-01', '2024-01-02', '2024-01-03', '2024-01-01', '2024-01-02', '2024-01-03'], 'price': [3999, 3899, 3799, 4050, 3950, 3850, 3650, 3550, 3450], 'rating': [4.8, 4.8, 4.7, 4.6, 4.6, 4.5, 4.3, 4.3, 4.2], 'review_count': [12500, 12650, 12800, 8900, 8950, 9000, 15600, 15800, 16000] } # 创建DataFrame df = pd.DataFrame(raw_data) df['date'] = pd.to_datetime(df['date']) print("原始数据概览:") print(df.head()) print(f"\n数据形状: {df.shape}") print(f"时间范围: {df['date'].min()} 到 {df['date'].max()}") # 数据清洗：检查缺失值 print("\n=== 数据清洗 ===") print("缺失值检查:") print(df.isnull().sum()) # 价格趋势分析 print("\n=== 价格趋势分析 ===") # 按平台计算平均价格 platform_avg_price = df.groupby('platform')['price'].mean() print("各平台平均价格:") for platform, avg_price in platform_avg_price.items(): print(f"{platform}: ¥{avg_price:.2f}") # 计算价格变化率 df['price_change'] = df.groupby('platform')['price'].pct_change() print("\n每日价格变化率:") print(df[['platform', 'date', 'price', 'price_change']].head(6)) # 可视化价格趋势 plt.figure(figsize=(12, 6)) for platform in df['platform'].unique(): platform_data = df[df['platform'] == platform] plt.plot(platform_data['date'], platform_data['price'], marker='o', label=platform, linewidth=2) plt.title('小米14手机各平台价格趋势（模拟数据）', fontsize=14) plt.xlabel('日期', fontsize=12) plt.ylabel('价格（元）', fontsize=12) plt.legend() plt.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() # 保存图表 plt.savefig('price_trend.png', dpi=300, bbox_inches='tight') print("\n价格趋势图已保存为 price_trend.png") # 评分与价格的相关性分析 print("\n=== 评分与价格相关性分析 ===") correlation = df.groupby('platform').apply( lambda x: x['price'].corr(x['rating']) ) print("各平台价格与评分的相关性:") print(correlation) # 生成分析报告 print("\n=== 分析报告摘要 ===") report_summary = f""" 小米14手机多平台监控分析报告（{datetime.now().strftime('%Y-%m-%d')}） 1. 价格对比： - 京东平均价格：¥{df[df['platform']=='京东']['price'].mean():.2f} - 天猫平均价格：¥{df[df['platform']=='天猫']['price'].mean():.2f} - 拼多多平均价格：¥{df[df['platform']=='拼多多']['price'].mean():.2f} 2. 价格趋势： - 所有平台均呈现下降趋势 - 拼多多价格最低，但下降幅度相对较小 - 京东价格最高，但下降速度最快 3. 用户评价： - 京东评分最高（{df[df['platform']=='京东']['rating'].mean():.1f}） - 拼多多评价数量最多（平均{df[df['platform']=='拼多多']['review_count'].mean():.0f}条） 4. 购买建议： - 追求性价比：考虑拼多多平台 - 注重服务保障：考虑京东平台 - 建议关注周末促销活动 """ print(report_summary) # 保存分析结果 df.to_csv('xiaomi14_analysis.csv', index=False, encoding='utf-8-sig') print("\n分析结果已保存为 xiaomi14_analysis.csv")

这段代码展示了Coder Agent可能生成的数据分析逻辑。在实际的DeerFlow运行中，Coder Agent会根据具体的数据情况，动态生成最适合的分析代码。

4.5 完整流程演示

现在让我们看一个完整的端到端示例。假设我们要分析“Python爬虫技术的最新发展”，我们可以这样操作：

# 在DeerFlow控制台中输入研究问题 请输入研究问题：分析Python爬虫技术的最新发展趋势，包括常用框架、反爬对策、以及AI在爬虫中的应用 # DeerFlow会自动执行以下流程： # 1. Coordinator接收问题，交给Planner # 2. Planner制定研究计划： # - 搜索Python爬虫的最新文章和技术博客 # - 采集GitHub上热门爬虫项目的README和issue # - 分析Stack Overflow上关于爬虫的最新问题 # - 总结AI辅助爬虫的最新进展 # 3. Research Team执行： # - Researcher Agent搜索相关文章，用Jina爬虫采集内容 # - Coder Agent分析GitHub数据，统计框架流行度 # 4. Reporter生成最终报告

在Web UI中，你可以实时看到每个步骤的执行情况。Researcher Agent会显示它搜索了哪些关键词，采集了哪些网页。Coder Agent会显示它执行了哪些Python代码，得到了什么结果。

最终生成的报告会包含：

• 关键发现摘要
• Python爬虫生态现状
• 主流框架对比（Scrapy、BeautifulSoup、Selenium等）
• 反爬技术发展趋势
• AI在爬虫中的应用案例
• 未来发展趋势预测

5. 高级技巧与实战建议

5.1 定制化爬虫策略

虽然DeerFlow内置的Jina Reader很好用，但有时候我们需要更定制化的爬取策略。这时候可以利用Coder Agent的Python执行能力。

比如，我们需要爬取一个需要登录的网站，或者需要处理JavaScript渲染的页面：

# Coder Agent可以生成更复杂的爬虫代码 # 示例：使用Selenium处理动态加载的电商页面 from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.options import Options import time import pandas as pd def crawl_dynamic_ecommerce(url): """爬取动态加载的电商商品页面""" # 设置Chrome选项（无头模式） chrome_options = Options() chrome_options.add_argument('--headless') # 无头模式，不显示浏览器窗口 chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-dev-shm-usage') driver = webdriver.Chrome(options=chrome_options) try: driver.get(url) # 等待页面加载完成 wait = WebDriverWait(driver, 10) # 等待商品标题加载 title_element = wait.until( EC.presence_of_element_located((By.CSS_SELECTOR, ".product-title")) ) title = title_element.text # 等待价格加载 price_element = wait.until( EC.presence_of_element_located((By.CSS_SELECTOR, ".product-price")) ) price = price_element.text # 滚动加载更多评价（如果需要） driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") time.sleep(2) # 等待加载 # 获取评价信息 reviews = [] review_elements = driver.find_elements(By.CSS_SELECTOR, ".review-item") for review in review_elements[:10]: # 取前10条评价 try: rating = review.find_element(By.CSS_SELECTOR, ".rating").get_attribute("data-score") text = review.find_element(By.CSS_SELECTOR, ".review-text").text reviews.append({"rating": rating, "text": text}) except: continue return { "title": title, "price": price, "reviews": reviews, "review_count": len(reviews) } finally: driver.quit() # 使用示例 if __name__ == "__main__": result = crawl_dynamic_ecommerce("https://example-ecommerce.com/product/123") print(f"商品标题: {result['title']}") print(f"价格: {result['price']}") print(f"采集到 {result['review_count']} 条评价")

5.2 处理反爬机制

在实际的爬虫项目中，经常会遇到各种反爬措施。Coder Agent可以帮我们智能应对：

# 应对常见反爬策略的代码示例 import requests import random import time from fake_useragent import UserAgent class SmartCrawler: """智能爬虫类，处理常见反爬措施""" def __init__(self): self.ua = UserAgent() self.session = requests.Session() self.setup_session() def setup_session(self): """设置会话，模拟真实浏览器""" headers = { 'User-Agent': self.ua.random, 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', } self.session.headers.update(headers) def get_with_retry(self, url, max_retries=3): """带重试机制的GET请求""" for attempt in range(max_retries): try: # 随机延迟，避免请求过于频繁 time.sleep(random.uniform(1, 3)) # 每次请求更换User-Agent self.session.headers['User-Agent'] = self.ua.random response = self.session.get(url, timeout=10) # 检查是否被屏蔽 if response.status_code == 403: print(f"第{attempt+1}次请求被屏蔽，尝试更换策略...") self.rotate_proxy() # 如果有代理，可以轮换代理 continue if response.status_code == 200: return response else: print(f"请求失败，状态码: {response.status_code}") except requests.exceptions.RequestException as e: print(f"请求异常: {e}") if attempt == max_retries - 1: raise return None def rotate_proxy(self): """轮换代理IP（如果有代理池的话）""" # 这里可以集成代理服务 # 比如从代理池获取新的代理IP pass def parse_with_fallback(self, html, selectors): """多选择器回退解析""" for selector in selectors: try: # 这里可以用BeautifulSoup或lxml解析 # 示例：提取价格信息 pass except: continue return None # 使用示例 crawler = SmartCrawler() response = crawler.get_with_retry("https://example.com/product") if response: print(f"成功获取页面，长度: {len(response.text)} 字符")

5.3 数据质量监控

在自动化爬虫系统中，数据质量监控很重要。我们可以让Coder Agent定期检查数据质量：

# 数据质量监控脚本 import pandas as pd import numpy as np from datetime import datetime class DataQualityMonitor: """数据质量监控类""" @staticmethod def check_completeness(df, required_columns): """检查数据完整性""" missing_cols = [col for col in required_columns if col not in df.columns] if missing_cols: return False, f"缺少必要列: {missing_cols}" return True, "所有必要列都存在" @staticmethod def check_consistency(df, column_rules): """检查数据一致性""" issues = [] for column, rules in column_rules.items(): if column in df.columns: # 检查数据类型 if 'type' in rules: expected_type = rules['type'] actual_type = df[column].dtype if not np.issubdtype(actual_type, expected_type): issues.append(f"{column}列类型不匹配: 期望{expected_type}, 实际{actual_type}") # 检查值范围 if 'range' in rules: min_val, max_val = rules['range'] out_of_range = df[(df[column] < min_val) | (df[column] > max_val)] if len(out_of_range) > 0: issues.append(f"{column}列有{len(out_of_range)}个值超出范围[{min_val}, {max_val}]") # 检查唯一性 if rules.get('unique', False): duplicates = df[df.duplicated(subset=[column], keep=False)] if len(duplicates) > 0: issues.append(f"{column}列有{len(duplicates)}个重复值") if issues: return False, issues return True, "数据一致性检查通过" @staticmethod def check_freshness(df, date_column, max_age_days=7): """检查数据新鲜度""" if date_column not in df.columns: return False, f"找不到日期列: {date_column}" # 转换为日期类型 df[date_column] = pd.to_datetime(df[date_column]) # 计算最新数据的日期 latest_date = df[date_column].max() days_old = (datetime.now() - latest_date).days if days_old > max_age_days: return False, f"数据已过期{days_old}天，最新数据日期: {latest_date.date()}" return True, f"数据新鲜度良好，最新数据日期: {latest_date.date()}" # 使用示例 def monitor_crawled_data(data_file): """监控爬取的数据质量""" # 读取数据 df = pd.read_csv(data_file) print("=== 数据质量检查报告 ===") print(f"数据文件: {data_file}") print(f"数据形状: {df.shape}") print(f"检查时间: {datetime.now()}") # 定义检查规则 required_columns = ['product_id', 'product_name', 'price', 'rating', 'review_count', 'crawl_date'] column_rules = { 'price': {'type': np.float64, 'range': (0, 100000)}, 'rating': {'type': np.float64, 'range': (0, 5)}, 'review_count': {'type': np.int64, 'range': (0, 1000000)}, 'product_id': {'unique': True} } # 执行检查 completeness_ok, completeness_msg = DataQualityMonitor.check_completeness(df, required_columns) print(f"\n1. 完整性检查: {'通过' if completeness_ok else '失败'}") print(f" 详情: {completeness_msg}") consistency_ok, consistency_msg = DataQualityMonitor.check_consistency(df, column_rules) print(f"\n2. 一致性检查: {'通过' if consistency_ok else '失败'}") if not consistency_ok: for issue in consistency_msg: print(f" - {issue}") freshness_ok, freshness_msg = DataQualityMonitor.check_freshness(df, 'crawl_date', max_age_days=1) print(f"\n3. 新鲜度检查: {'通过' if freshness_ok else '失败'}") print(f" 详情: {freshness_msg}") # 生成质量评分 checks_passed = sum([completeness_ok, consistency_ok, freshness_ok]) quality_score = checks_passed / 3 * 100 print(f"\n=== 质量评分: {quality_score:.1f}/100 ===") if quality_score < 80: print("警告: 数据质量需要关注，建议检查爬虫配置") return quality_score # 运行监控 if __name__ == "__main__": score = monitor_crawled_data("crawled_data.csv")

6. 实际应用场景扩展

DeerFlow的爬虫+LLM组合不仅限于电商监控，在很多场景下都能发挥价值：

6.1 竞品分析自动化

对于市场部门来说，定期分析竞品动态是日常工作。用DeerFlow可以自动化这个流程：

# 输入研究问题 监控苹果、华为、小米最新手机的产品特性、价格策略和用户反馈，每周生成竞品分析报告 # DeerFlow会自动： # 1. 爬取各品牌官网的产品页面 # 2. 采集电商平台的销售数据和评价 # 3. 分析社交媒体上的用户讨论 # 4. 生成包含数据可视化的竞品报告

6.2 舆情监控系统

对于公关或品牌部门，舆情监控很重要：

# 输入研究问题 监控“新能源汽车”相关的舆情，包括新闻、社交媒体、论坛讨论，识别正面和负面情绪，发现潜在危机 # DeerFlow会自动： # 1. 爬取新闻网站的相关报道 # 2. 采集微博、知乎等社交平台讨论 # 3. 用情感分析模型判断舆情倾向 # 4. 生成舆情日报和预警提示

6.3 学术研究辅助

对于研究人员，文献调研和数据收集可以自动化：

# 输入研究问题 收集关于“深度学习在医疗影像诊断中的应用”的最新研究论文，分析研究趋势、主要方法和未来方向 # DeerFlow会自动： # 1. 搜索arXiv、PubMed等学术数据库 # 2. 爬取相关论文的摘要和关键信息 # 3. 分析研究方法和技术路线 # 4. 生成文献综述报告

7. 总结

用下来感觉DeerFlow确实是个挺有意思的工具，特别是它把多智能体协作和实际的数据采集需求结合得不错。对于需要定期做数据监控、竞品分析或者舆情跟踪的团队来说，能省不少事。

最大的优点是整个流程可以自动化，从制定计划到执行采集，再到分析报告，一气呵成。而且因为用了多智能体架构，每个环节都有专门的“专家”负责，结果质量比单一工具要好。

当然，实际用的时候也会遇到一些挑战。比如有些网站的反爬措施比较严格，可能需要调整爬虫策略。还有数据清洗这块，虽然Coder Agent能写代码处理，但特别复杂的数据结构可能还是需要人工干预一下。

如果你刚开始接触，建议从小项目做起，先试试简单的数据采集任务，熟悉了各个Agent的工作方式，再逐步扩展到更复杂的场景。配置方面，记得把API密钥和环境变量设置好，这是最容易出问题的地方。

总的来说，DeerFlow为Python爬虫和数据分析提供了一个新的思路——不是单纯写代码，而是让AI来帮你思考整个流程，你只需要告诉它你想要什么。这种工作方式，可能会是未来的一个趋势。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。