基于多Agent与LangGraph构建智能爬虫工作流：从数据采集到自动化分析

1. 先搞清楚这个平台到底能做什么以及它和普通爬虫的区别如果你正在找一个能自动追踪热点、分析趋势并且能自己思考、决策和协作的爬虫系统那这个基于多Agent和AI工作流构建的平台可能就是你需要的那个“智能副驾”。它不是一个简单的脚本而是一个由多个“智能体”分工协作的自动化大脑。核心价值在于它把“爬取数据”这个动作升级成了“感知-理解-决策-行动”的完整闭环。很多人一听到“多Agent 爬虫”第一反应是“用AI写爬虫代码”。这其实是个误解。这里的Agent不是帮你写代码的Copilot而是一个个具备特定能力的虚拟角色。比如一个Agent负责规划今天要爬哪些网站另一个Agent负责执行具体的网页请求和解析第三个Agent则对抓取到的内容进行摘要、分类或情感分析最后可能还有一个Agent负责把分析结果整理成报告。它们通过LangGraph定义的工作流来协同工作就像一支训练有素的数字特工小队。所以这个平台解决的核心问题是如何让数据采集和分析过程从“手动执行命令”变成“自动感知和响应”。它适合两类人一是需要长期、稳定监控特定领域信息如竞品动态、舆情、技术趋势的运营或分析师二是开发者希望构建一个可复用、可扩展的智能数据管道而不是写一堆零散且脆弱的爬虫脚本。最值得关注的点是“工作流”和“协作”。LangGraph让你能用图Graph的方式定义Agent之间的状态流转和调用逻辑这比写线性的if-else脚本要清晰和强大得多。而FastAPI和Nuxt分别提供了稳定可靠的后端API和灵活的前端展示让整个系统从一个实验性脚本变成了一个可以交付使用的平台。2. 环境与核心组件准备别急着写代码先理清依赖关系在动手之前我建议先花十分钟把整个技术栈和它们之间的关系画个草图。这能帮你避免后期陷入“模块A需要BB又依赖C”的混乱。整个平台可以粗略分为四层智能体与工作流层核心大脑LangChainLangGraph。LangChain是构建Agent的“乐高积木”提供了连接大语言模型、工具、记忆的基础组件。LangGraph则是“设计图纸”用来编排多个Agent定义它们何时、以何种顺序执行。数据获取层手脚Python爬虫。通常基于requests、BeautifulSoup、Playwright或Selenium。这部分被封装成一个个“工具”Tool供Agent调用。服务与接口层躯干FastAPI。它将AI工作流暴露为标准的HTTP API处理请求、调度任务、返回结果并管理并发、错误重试等。展示与交互层面孔Nuxt基于Vue.js。用于构建管理后台可视化工作流状态、查看分析结果、配置监控任务等。对于本地开发环境你的机器需要满足以下最低条件我一般会先检查这几项Python 3.9这是LangChain等库的主流支持版本。Node.js 16用于运行Nuxt前端。至少8GB内存运行大语言模型即使是本地小模型和多个服务时内存是关键。如果只是调用云端API如OpenAI4GB勉强够用。稳定的网络爬虫和调用AI API都依赖网络。接下来是Python虚拟环境和包安装。不要全局安装用venv或conda创建一个独立环境。# 创建并激活虚拟环境 python -m venv ai-tracker-env source ai-tracker-env/bin/activate # Linux/macOS # ai-tracker-env\Scripts\activate # Windows # 安装核心后端依赖 pip install langchain langgraph fastapi uvicorn requests beautifulsoup4 playwright # Playwright需要安装浏览器内核 playwright install前端部分如果你打算开发界面需要初始化Nuxt项目npx nuxilatest init ai-tracker-frontend cd ai-tracker-frontend npm install这里最容易忽略的是版本兼容性。LangChain和LangGraph更新较快直接用pip install langchain langgraph可能会装上最新版但某些教程或示例代码可能基于旧版。一个稳妥的做法是在项目初期先锁定一个已知稳定的版本组合例如pip install langchain0.1.0 langgraph0.0.40等核心工作流跑通后再考虑升级。另外如果计划使用本地大模型如通过Ollama还需要额外安装langchain-community和对应模型的集成包。3. 设计你的第一个AI工作流从单Agent到多Agent协作不要一上来就想设计一个监控全网热点的复杂系统。先从最小的、可验证的单元开始一个能爬取单一网页并总结内容的Agent。3.1 定义工具让Agent有“手”可用Agent本身不会爬虫它需要调用“工具”。我们首先创建一个最基础的爬虫工具函数。# tools.py import requests from bs4 import BeautifulSoup from langchain.tools import tool from typing import Optional tool def scrape_webpage(url: str, selector: Optional[str] None) - str: 爬取指定URL的网页内容并提取文本。 如果提供了CSS选择器则只提取该选择器匹配部分的内容。 try: headers {User-Agent: Mozilla/5.0} # 简单伪装浏览器 response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 检查HTTP错误 soup BeautifulSoup(response.content, html.parser) if selector: elements soup.select(selector) text .join([elem.get_text(stripTrue) for elem in elements]) else: # 移除脚本、样式等 for script in soup([script, style]): script.decompose() text soup.get_text(separator , stripTrue) # 返回前5000字符作为示例避免上下文过长 return text[:5000] if text else 未能获取到有效文本内容。 except Exception as e: return f爬取失败: {str(e)}这个工具函数用tool装饰器包装这样LangChain就能识别它并自动生成描述供Agent理解。selector参数很重要它让你能精准抓取页面特定区域如文章正文div.content避免抓到导航栏、广告等噪音。3.2 创建单Agent具备思考和行动能力接下来我们创建一个Agent它拥有上述工具并能根据我们的指令如“去分析一下XX网站的最新文章”来决定是否以及如何调用工具。# agent_basic.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI # 假设使用OpenAI API from langchain.prompts import PromptTemplate from tools import scrape_webpage # 1. 初始化大语言模型LLM # 请替换为你的API Key或使用本地模型如Ollama llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyyour-key) # 2. 准备工具列表 tools [scrape_webpage] # 3. 定义提示词模板告诉Agent它的角色和能力 prompt PromptTemplate.from_template( 你是一个专业的网络信息分析助手。你可以通过工具来获取网页内容。 请根据用户的问题决定是否需要爬取网页并给出分析。 当前问题{input} 请开始你的思考和工作。 ) # 4. 使用ReAct框架创建Agent agent create_react_agent(llmllm, toolstools, promptprompt) # 5. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 运行测试 if __name__ __main__: result agent_executor.invoke({ input: 请爬取https://news.example.com/tech这个页面并总结其主要科技新闻标题。 }) print(Agent输出:, result[output])运行这个脚本你会看到Agent的思考过程verboseTrue它先“思考”是否需要调用工具然后调用scrape_webpage拿到网页文本后再“思考”如何总结最后输出结果。这就是一个最基本的“感知-思考-行动”循环。3.3 引入LangGraph构建多Agent工作流单Agent能力有限。现在我们引入LangGraph创建两个协作的Agent一个爬虫调度员和一个内容分析师。# workflow_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from tools import scrape_webpage # 1. 定义工作流状态共享内存 class WorkflowState(TypedDict): 整个工作流的状态所有Agent都能读写其中部分字段。 original_query: str # 用户原始问题 target_urls: List[str] # 需要爬取的URL列表 scraped_contents: List[str] # 爬取到的原始内容 analysis_result: str # 最终分析报告 # 可以添加更多字段如爬取状态、错误信息等 # 2. 初始化模型和工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [scrape_webpage] # 3. 定义第一个节点URL规划员Agent def url_planner_node(state: WorkflowState): 根据用户查询分析并规划出需要爬取的URL列表。 from langchain.agents import AgentExecutor, create_react_agent from langchain.prompts import PromptTemplate prompt PromptTemplate.from_template( 你是信息搜集专家。用户想了解{query} 你的任务是分析这个需求并输出一个你认为最相关的、具体的网页URL列表最多3个。 只输出URL每行一个。如果你认为无法从公开网页获取请输出“无”。 ) planner_agent create_react_agent(llm, tools[], promptprompt) # 此Agent不需要爬虫工具 executor AgentExecutor(agentplanner_agent, tools[], verboseFalse) result executor.invoke({query: state[original_query]}) urls [line.strip() for line in result[output].split(\n) if line.strip() and line.startswith(http)] # 更新状态 return {target_urls: urls} # 4. 定义第二个节点爬虫执行员Agent def scraper_node(state: WorkflowState): 执行爬虫获取目标URL的内容。 contents [] for url in state.get(target_urls, []): content scrape_webpage.invoke({url: url}) contents.append(content) return {scraped_contents: contents} # 5. 定义第三个节点内容分析员Agent def analyst_node(state: WorkflowState): 分析爬取到的内容生成报告。 from langchain.schema import HumanMessage combined_content \n---\n.join(state[scraped_contents]) prompt f 用户最初的问题是{state[original_query]} 以下是爬取到的相关网页内容摘要 {combined_content[:3000]} # 限制长度避免token超限 请基于以上内容为用户生成一份简洁、有条理的分析报告。 message [HumanMessage(contentprompt)] response llm.invoke(message) return {analysis_result: response.content} # 6. 构建工作流图 graph_builder StateGraph(WorkflowState) # 添加节点 graph_builder.add_node(plan_urls, url_planner_node) graph_builder.add_node(scrape_data, scraper_node) graph_builder.add_node(analyze_content, analyst_node) # 设置边执行顺序 graph_builder.set_entry_point(plan_urls) graph_builder.add_edge(plan_urls, scrape_data) graph_builder.add_edge(scrape_data, analyze_content) graph_builder.add_edge(analyze_content, END) # 编译成可执行的工作流 workflow graph_builder.compile() # 7. 运行测试 if __name__ __main__: initial_state {original_query: 今天人工智能领域有什么最新突破或重磅论文} final_state workflow.invoke(initial_state) print(最终分析报告) print(final_state[analysis_result]) print(\n爬取的URL, final_state[target_urls])这个工作流清晰定义了三个节点的协作关系规划URL - 爬取 - 分析。LangGraph会自动管理状态在它们之间的传递。你可以通过workflow.get_graph().draw_mermaid()来生成可视化图直观看到流程。注意在实际项目中url_planner_node的实现会更复杂可能需要接入搜索引擎API或维护一个种子网站库。这里用LLM直接生成URL仅作演示其准确性和可靠性需要进一步设计。4. 用FastAPI封装为服务并处理并发与稳定性本地脚本跑通只是第一步。要让这个AI工作流成为一个“平台”必须用API包装起来使其能够被远程调用、排队、管理。FastAPI是绝佳选择因为它异步性能好、自动生成API文档。4.1 构建核心API端点创建一个main.py文件# main.py from fastapi import FastAPI, BackgroundTasks, HTTPException from pydantic import BaseModel, HttpUrl from typing import List, Optional import uuid import asyncio from workflow_graph import workflow # 导入前面定义的工作流 app FastAPI(titleAI热点追踪分析平台API) # 用于存储任务状态的内存字典生产环境应使用Redis或数据库 tasks {} class AnalysisRequest(BaseModel): query: str # 用户的分析请求如“追踪本周AI编程工具动态” callback_url: Optional[HttpUrl] None # 可选任务完成后回调通知的URL class TaskStatus(BaseModel): task_id: str status: str # pending, running, completed, failed result: Optional[str] None error: Optional[str] None app.post(/analyze, response_modelTaskStatus) async def create_analysis_task(request: AnalysisRequest, background_tasks: BackgroundTasks): 提交一个新的热点分析任务异步执行。 task_id str(uuid.uuid4()) tasks[task_id] {status: pending, result: None, error: None} # 将任务加入后台执行 background_tasks.add_task(run_analysis_workflow, task_id, request.query) return TaskStatus(task_idtask_id, statuspending) app.get(/task/{task_id}, response_modelTaskStatus) async def get_task_status(task_id: str): 查询指定任务的状态和结果。 task tasks.get(task_id) if not task: raise HTTPException(status_code404, detailTask not found) return TaskStatus(task_idtask_id, **task) async def run_analysis_workflow(task_id: str, query: str): 后台执行工作流的函数。 try: tasks[task_id][status] running # 注意LangGraph的invoke是同步的在异步环境中要用run_in_executor防止阻塞事件循环 loop asyncio.get_event_loop() final_state await loop.run_in_executor(None, workflow.invoke, {original_query: query}) tasks[task_id][status] completed tasks[task_id][result] final_state.get(analysis_result, No result generated.) except Exception as e: tasks[task_id][status] failed tasks[task_id][error] str(e) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)现在你可以通过POST /analyze提交任务并立即得到一个task_id。然后通过GET /task/{task_id}轮询结果。这种异步设计避免了HTTP请求长时间挂起。4.2 关键的生产化考量上面的示例为了简洁用了内存字典存任务状态。在实际生产环境中你必须考虑以下几点任务队列与持久化使用CeleryRedis或RabbitMQ来处理高并发任务并将任务状态和结果存入数据库如PostgreSQL。错误处理与重试爬虫极易因网络、反爬等原因失败。需要在工作流节点中增加重试逻辑并记录详细日志。速率限制与代理针对目标网站设置合理的爬取间隔time.sleep并准备IP代理池避免被封。结果存储不应只返回文本。应将原始内容、分析结果、来源URL、执行时间等结构化存储便于后续查询和可视化。配置化管理将目标网站、CSS选择器、爬取频率等配置外置到配置文件或数据库实现动态管理。5. 前端展示与管理用Nuxt构建控制台对于这样一个平台一个能查看任务历史、可视化工作流状态、配置监控规则的前端是必不可少的。Nuxt作为全栈框架能很好地与FastAPI后端配合。5.1 核心页面与功能任务提交页一个表单输入分析主题或关键词提交后跳转到任务详情页。任务列表页展示所有历史任务包括状态成功/失败/进行中、创建时间、查询内容。可以过滤和搜索。任务详情页展示某个任务的完整信息原始查询、爬取到的URL列表、最终分析报告。可以在这里重试失败的任务。工作流配置页进阶可视化编辑LangGraph工作流需要将图结构存储为JSON调整Agent的提示词或工具。5.2 与FastAPI交互示例在Nuxt页面组件中调用后端的API!-- pages/tasks/index.vue -- template div h1分析任务列表/h1 button clicksubmitNewTask新建分析任务/button ul li v-fortask in tasks :keytask.task_id {{ task.query }} - 状态: span :classtask.status{{ task.status }}/span button clickviewDetails(task.task_id)查看/button /li /ul /div /template script setup const { data: tasks, refresh } await useFetch(http://localhost:8000/tasks); // 需要后端新增一个获取所有任务的端点 const submitNewTask async () { const query prompt(请输入您想追踪分析的主题); if (query) { const { data } await useFetch(http://localhost:8000/analyze, { method: POST, body: { query } }); if (data.value) { alert(任务已提交ID: ${data.value.task_id}); refresh(); // 刷新列表 } } }; const viewDetails (taskId) { navigateTo(/tasks/${taskId}); }; /script style scoped .completed { color: green; } .running { color: orange; } .failed { color: red; } /style前端的关键是提供清晰的反馈和便捷的操作入口将后端AI工作流的复杂性对用户隐藏起来。6. 排查、优化与边界从能跑到好用平台搭建起来后真正的挑战在于稳定运行和持续优化。以下是几个最常见的排查点和优化方向。6.1 当工作流卡住或失败时先看哪里检查Agent的“思考”过程在开发阶段将各个Agent执行器的verbose参数设为True。这会打印出LLM的思考链Chain of Thought帮你判断是规划逻辑出错还是工具调用失败。检查网络与反爬爬虫节点失败最常见。查看返回的错误信息。如果是403、429说明触发了反爬。需要增加请求头、使用代理、降低频率。用try...except包裹爬虫调用并记录详细的错误日志。检查Token长度限制LLM有上下文窗口限制。如果爬取的内容过长在传递给分析Agent前必须进行截断或摘要。可以在scraper_node后增加一个“内容摘要”节点先用LLM对每篇文章进行压缩再将摘要传递给最终的分析节点。检查状态流转确保工作流每个节点的输出格式都符合下一个节点对输入状态的期望。LangGraph是强类型通过TypedDict类型不匹配会导致运行时错误。6.2 性能与成本优化并行爬取scraper_node中是顺序爬取URL。可以改用asyncio或并发库如concurrent.futures来并行爬取大幅缩短IO等待时间。缓存结果对于相同的查询或URL可以使用langchain.cache如SQLiteCache缓存LLM的响应和爬虫结果节省成本和时间。使用更经济的模型规划和分析可以使用不同的模型。URL规划可以用小模型如gpt-3.5-turbo最终分析报告生成再用大模型如gpt-4。本地部署模型如OllamaLlama 3可以彻底消除API成本但需要更强的本地算力。流式输出对于长报告可以让分析Agent流式输出前端逐步显示提升用户体验。FastAPI和LangChain都支持流式响应。6.3 系统的边界与局限认识到边界比盲目扩展更重要法律与伦理边界严格遵守网站的robots.txt协议尊重版权和个人隐私。本平台设计应用于公开信息的聚合分析切勿用于爬取个人敏感数据、进行商业侵权或攻击性行为。技术边界动态加载大量JavaScript的网站需要Playwright或Selenium资源消耗更大。验证码、登录墙等复杂反爬机制需要更高级的方案如OCR服务、Cookie池这会使系统复杂度急剧上升。LLM的可靠性Agent的规划和分析能力完全依赖LLM。LLM可能会“幻觉”出错误的URL或做出不合理的内容总结。需要在关键环节如URL验证加入人工规则或校验逻辑。并非全自动目前这仍是一个需要触发用户提交查询的系统。要实现真正的“热点追踪”还需要加上定时调度如使用APScheduler定期自动执行预设的查询任务。7. 从项目到平台下一步可以做什么当你把上述所有部分跑通一个最小可用的AI热点追踪分析平台就成型了。但要想把它变成一个真正有价值的工具还可以从以下几个方向深化记忆与持续学习利用LangGraph的“持久化检查点”功能让工作流记住之前的分析结论。下次遇到类似查询时可以直接给出历史结论或进行对比分析实现“越用越聪明”。工具扩展除了爬虫为Agent装备更多工具。例如search_web工具调用搜索引擎API获取更广泛的初步信息。send_email工具当分析出重大负面舆情或机会时自动发送预警邮件。save_to_database工具将结构化结果保存到Notion、Airtable或你的业务数据库。可视化与可解释性不仅展示最终报告还将工作流的执行过程哪些Agent被调用、输入输出是什么可视化出来增加系统的透明度和可信度。领域垂直化将通用平台改造成针对特定领域如科技资讯、电商价格、招聘市场的专用分析工具。这需要定制爬虫规则、分析提示词和输出模板。我个人更建议在初期不要追求大而全。先聚焦一个非常具体的垂直场景比如“追踪AI领域arXiv上每日高引论文”用这个最小闭环验证整个技术栈的可行性然后再逐步添加功能和扩大范围。这样能最快看到效果也最容易排查问题。这个项目的核心乐趣不在于爬虫本身而在于设计并见证多个AI智能体如何像一支团队一样自动完成一个复杂的认知任务。