在实际项目中我们经常需要从多个来源实时追踪和分析热点信息比如技术趋势、行业动态或舆情监控。传统做法是编写独立的爬虫脚本然后手动或通过定时任务进行数据清洗、分析和存储流程割裂且难以维护。随着大语言模型LLM和智能体Agent技术的发展我们可以构建一个更智能、更自动化的系统让多个具备不同能力的 Agent 协同工作形成一个完整的数据处理与洞察生成工作流。本文将带你从零开始构建一个基于 LangGraph、LangChain、FastAPI 和 Nuxt 的 AI 热点追踪分析平台。这个平台的核心思想是将数据采集、内容解析、信息聚合、趋势分析和报告生成等任务分配给不同的 AI Agent并通过 LangGraph 的工作流引擎来编排它们的执行顺序与状态流转。前端使用 Nuxt 提供友好的交互界面后端使用 FastAPI 提供高效、类型安全的 API 服务。通过本文你将掌握如何将这些前沿技术栈组合起来实现一个可学习、可复现的智能分析系统。无论你是想深入理解多 Agent 系统设计还是希望构建自己的自动化信息处理管道本文都将提供清晰的路径和具体的代码实现。1. 理解多 Agent 系统与 LangGraph 的工作流编排在开始编码之前我们需要厘清几个核心概念理解它们如何在这个项目中协同工作。1.1 什么是 Agent 与多 Agent 系统在 LangChain 的语境中一个Agent是一个具备使用工具Tools能力的大语言模型LLM。它可以根据用户的目标自主决定调用哪个工具、以什么参数调用并解析工具的返回结果。例如一个“爬虫 Agent”可能拥有“获取网页内容”和“解析 HTML”两个工具。多 Agent 系统则由多个这样的 Agent 组成每个 Agent 专精于某一特定领域如爬取、分析、总结。它们之间可以传递信息、协作完成任务。相比于单个“全能”Agent多 Agent 系统的好处在于职责分离、易于调试和扩展。在我们的热点追踪平台中可以设计以下 Agent采集 Agent负责调度爬虫从指定源如新闻网站、社交媒体 API获取原始数据。清洗与解析 Agent负责从原始 HTML 或 JSON 中提取结构化信息标题、正文、发布时间、来源。聚合与去重 Agent负责合并来自不同源的相似内容并过滤掉低质量或重复信息。分析 Agent负责对聚合后的信息进行情感分析、关键词提取、趋势判断。报告生成 Agent负责将分析结果组织成结构化的报告如每日摘要。1.2 LangGraph 如何编排工作流LangChain 提供了构建单个 Agent 的基础但当多个 Agent 需要以复杂、有状态的方式协作时就需要更强大的编排工具。这就是LangGraph的用武之地。LangGraph 允许你将整个系统建模为一个有向图Graph。图中的节点Node代表一个执行单元可以是一个 Agent 的调用一个工具的执行或一个条件判断边Edge代表执行流的方向。它核心管理的是状态State一个在所有节点间共享和传递的字典。在我们的场景中工作流可以这样设计开始节点接收用户查询如“追踪今日 AI 热点”。状态传递给“采集 Agent”节点它调用爬虫工具获取数据。状态现在包含了原始数据传递给“清洗 Agent”节点进行处理。清洗后的数据传递给“聚合 Agent”节点。根据聚合结果的数量通过条件边决定下一步如果数据量足够进入“分析 Agent”否则结束流程或触发重新采集。“分析 Agent”节点处理数据然后将结果传递给“报告生成 Agent”节点。最终状态包含生成的分析报告流程结束。LangGraph 保证了工作流的有序、可靠执行并方便地处理循环、分支和并行通过异步。1.3 技术栈选型理由LangChain LangGraph当前构建 LLM 应用最成熟的框架之一提供了丰富的 Agent、Tool、Memory 等抽象LangGraph 是其官方的工作流/状态机管理库两者结合紧密。FastAPI现代、高性能的 Python Web 框架自动生成 OpenAPI 文档异步支持好非常适合作为 AI 服务的后端方便前端调用和调试。Nuxt基于 Vue.js 的元框架支持服务端渲染SSR、静态站点生成SSG和单页面应用SPA。它提供了良好的开发体验和项目结构适合构建需要 SEO 友好和复杂交互的管理界面。爬虫技术如requests,BeautifulSoup,Playwright作为 Agent 的工具Tools被集成负责实际的数据获取任务。2. 环境准备与项目初始化我们将创建一个前后端分离的项目。后端Backend使用 FastAPI 提供工作流执行和管理的 API前端Frontend使用 Nuxt 构建管理界面。2.1 后端环境准备FastAPI LangGraph首先创建项目目录并初始化 Python 环境。# 创建项目根目录 mkdir ai-hotspot-platform cd ai-hotspot-platform # 创建后端目录 mkdir backend cd backend # 创建 Python 虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 创建 requirements.txt 并安装核心依赖backend/requirements.txt文件内容如下fastapi0.104.1 uvicorn[standard]0.24.0 langchain0.0.353 langgraph0.0.13 langchain-openai0.0.2 # 使用 OpenAI 模型也可替换为其他适配器 openai1.3.0 requests2.31.0 beautifulsoup44.12.2 playwright1.40.0 # 用于处理动态网页 pydantic2.5.0 pydantic-settings2.1.0 python-dotenv1.0.0 sqlalchemy2.0.23 alembic1.12.1 # 数据库迁移 psycopg2-binary2.9.9 # PostgreSQL 驱动可按需更换 redis5.0.1 # 可选用于缓存或状态存储 celery5.3.4 # 可选用于异步任务队列安装依赖pip install -r requirements.txt # 安装 Playwright 浏览器 playwright install chromium2.2 前端环境准备Nuxt在项目根目录下初始化 Nuxt 项目。cd ai-hotspot-platform # 使用 npx 创建 Nuxt 项目选择适合的选项 npx nuxilatest init frontend # 进入项目并安装依赖 cd frontend npm install # 安装 UI 库这里以 Element Plus 为例 npm install element-plus element-plus/nuxt # 安装 HTTP 客户端 npm install axios修改frontend/nuxt.config.ts引入 Element Plusexport default defineNuxtConfig({ modules: [element-plus/nuxt], css: [element-plus/dist/index.css], // 其他配置... })2.3 项目结构规划一个清晰的项目结构有助于维护。建议如下ai-hotspot-platform/ ├── backend/ # FastAPI 后端 │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI 应用入口 │ │ ├── core/ # 核心配置、依赖项 │ │ ├── agents/ # 各个 Agent 的定义 │ │ │ ├── __init__.py │ │ │ ├── crawler_agent.py │ │ │ ├── parser_agent.py │ │ │ └── ... │ │ ├── graphs/ # LangGraph 工作流定义 │ │ │ ├── __init__.py │ │ │ └── hotspot_graph.py # 核心热点分析工作流 │ │ ├── tools/ # Agent 可用的工具 │ │ │ ├── __init__.py │ │ │ ├── web_tools.py │ │ │ └── ... │ │ ├── models/ # Pydantic 模型和 SQLAlchemy ORM 模型 │ │ ├── schemas/ # Pydantic 响应/请求模型 │ │ ├── api/ # API 路由 │ │ │ ├── v1/ │ │ │ │ ├── __init__.py │ │ │ │ ├── endpoints/ │ │ │ │ │ ├── workflow.py │ │ │ │ │ └── ... │ │ │ │ └── deps.py # 依赖项 │ │ ├── db/ # 数据库会话、迁移 │ │ └── config.py # 配置文件 │ ├── alembic/ # 数据库迁移目录 │ ├── tests/ │ ├── requirements.txt │ └── .env.example └── frontend/ # Nuxt 前端 ├── pages/ ├── components/ ├── composables/ # 状态管理、API 调用 ├── layouts/ ├── assets/ ├── nuxt.config.ts └── package.json3. 构建核心LangGraph 热点分析工作流这是项目的核心。我们将在backend/app/graphs/hotspot_graph.py中定义工作流。3.1 定义共享状态State状态是所有节点共享和修改的数据容器。我们使用TypedDict来定义其结构。# backend/app/graphs/hotspot_graph.py from typing import TypedDict, List, Optional, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.messages import BaseMessage, HumanMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from app.agents.crawler_agent import get_crawler_agent from app.agents.parser_agent import get_parser_agent from app.agents.analyzer_agent import get_analyzer_agent from pydantic import BaseModel class GraphState(TypedDict): 工作流的状态定义 # 用户输入 query: str # 中间数据 raw_data: Optional[List[str]] # 爬取的原始内容HTML/文本 parsed_articles: Optional[List[dict]] # 解析后的结构化文章列表 aggregated_articles: Optional[List[dict]] # 聚合去重后的文章列表 analysis_result: Optional[dict] # 分析结果 final_report: Optional[str] # 最终报告 # 控制流 should_continue: bool # 决定是否继续流程 error: Optional[str] # 错误信息3.2 实现各个功能节点Nodes每个节点是一个函数接收当前State返回更新后的State。节点1采集节点Crawler Node这个节点调用采集 Agent。def crawl_node(state: GraphState) - GraphState: 调用采集 Agent 获取原始数据 print(f“--- 执行采集节点查询: {state[‘query’]} ---“) try: # 获取配置好的采集 Agent 执行器 crawler_agent_executor get_crawler_agent() # 这里简化处理实际应根据 query 选择数据源 result crawler_agent_executor.invoke({ “input”: f“请根据主题‘{state[‘query’]}’从预设的技术新闻源获取最新的文章链接和简要内容。” }) # 假设 Agent 返回一个包含 ‘raw_data’ 键的字典 raw_data result.get(“output”, []).get(“raw_data”, []) return {“raw_data”: raw_data} except Exception as e: return {“error”: f“采集失败: {str(e)}”, “should_continue”: False}节点2解析节点Parser Node这个节点调用清洗与解析 Agent。def parse_node(state: GraphState) - GraphState: 调用解析 Agent 处理原始数据 if not state.get(“raw_data”): return {“error”: “无原始数据可供解析”, “should_continue”: False} print(“--- 执行解析节点 ---“) try: parser_agent_executor get_parser_agent() # 将原始数据传递给解析 Agent result parser_agent_executor.invoke({ “input”: f“请解析以下原始内容提取每篇文章的标题、正文、发布时间、来源和URL。内容{state[‘raw_data’]}” }) parsed_articles result.get(“output”, {}).get(“articles”, []) return {“parsed_articles”: parsed_articles} except Exception as e: return {“error”: f“解析失败: {str(e)}”, “should_continue”: False}节点3路由节点Router Node这是一个条件判断节点决定流程走向。def route_after_parse(state: GraphState) - str: 根据解析结果决定下一步聚合还是直接结束 articles state.get(“parsed_articles”, []) if not articles or len(articles) 0: print(“--- 未解析到有效文章流程结束 ---“) return “end” elif len(articles) 3: # 假设数量太少不值得深入分析 print(f“--- 解析到 {len(articles)} 篇文章数量较少跳过深度分析 ---“) return “generate_simple_report” else: print(f“--- 解析到 {len(articles)} 篇文章进入聚合分析流程 ---“) return “aggregate”节点4聚合与分析节点Aggregate Analyze Node这里可以串联或并行执行聚合与分析。def aggregate_and_analyze_node(state: GraphState) - GraphState: 聚合文章并进行趋势分析 articles state.get(“parsed_articles”, []) if not articles: return {“error”: “无文章可聚合分析”, “should_continue”: False} print(“--- 执行聚合与分析节点 ---“) # 这里简化聚合逻辑去重基于标题或URL seen set() unique_articles [] for article in articles: identifier article.get(“title”, “”) article.get(“url”, “”) if identifier not in seen: seen.add(identifier) unique_articles.append(article) print(f“聚合后剩余 {len(unique_articles)} 篇唯一文章。”) # 调用分析 Agent try: analyzer_agent_executor get_analyzer_agent() analysis_input f“请分析以下文章列表总结出主要话题、关键词、情感倾向和热度趋势。文章列表{unique_articles}” result analyzer_agent_executor.invoke({“input”: analysis_input}) analysis_result result.get(“output”, {}) return { “aggregated_articles”: unique_articles, “analysis_result”: analysis_result } except Exception as e: return {“error”: f“分析失败: {str(e)}”, “should_continue”: False}节点5报告生成节点Report Nodedef report_node(state: GraphState) - GraphState: 生成最终分析报告 print(“--- 执行报告生成节点 ---“) analysis state.get(“analysis_result”, {}) articles state.get(“aggregated_articles”, []) query state.get(“query”, “”) # 这里可以调用一个专门的报告生成 Agent或直接格式化 report f“”” # 热点分析报告{query} **生成时间** {datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’)} **分析文章数** {len(articles)} ## 核心发现 {analysis.get(‘summary’, ‘暂无总结’)} ## 关键词 {‘, ‘.join(analysis.get(‘keywords’, []))} ## 趋势判断 {analysis.get(‘trend’, ‘暂无趋势判断’)} ## 文章列表 “”” for i, art in enumerate(articles, 1): report f“{i}. [{art.get(‘title’, ‘No Title’)}]({art.get(‘url’, ‘#’)})\n” return {“final_report”: report}3.3 组装工作流图将节点和边组合起来形成完整的工作流。def create_hotspot_workflow() - StateGraph: 创建并返回热点分析工作流图 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(“crawl”, crawl_node) workflow.add_node(“parse”, parse_node) workflow.add_node(“aggregate_analyze”, aggregate_and_analyze_node) workflow.add_node(“generate_report”, report_node) workflow.add_node(“generate_simple_report”, lambda s: {“final_report”: “数据量不足无法生成深度报告。”}) # 设置入口点 workflow.set_entry_point(“crawl”) # 添加边 workflow.add_edge(“crawl”, “parse”) # 从 parse 出来后根据条件路由 workflow.add_conditional_edges( “parse”, route_after_parse, { “end”: END, # 直接结束 “generate_simple_report”: “generate_simple_report”, “aggregate”: “aggregate_analyze” } ) workflow.add_edge(“aggregate_analyze”, “generate_report”) workflow.add_edge(“generate_report”, END) workflow.add_edge(“generate_simple_report”, END) # 编译图 return workflow.compile()4. 实现 Agent 与工具工作流中的节点依赖于具体的 Agent。我们以实现Crawler Agent为例。4.1 定义爬虫工具Tools工具是 Agent 可以调用的函数。我们先创建几个基础的网页抓取工具。# backend/app/tools/web_tools.py import requests from bs4 import BeautifulSoup from langchain.tools import tool from typing import Optional import logging logger logging.getLogger(__name__) tool def fetch_webpage(url: str) - str: “””获取指定URL的网页内容静态。“”” try: headers {‘User-Agent’: ‘Mozilla/5.0’} response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 可以在这里添加简单的编码处理 return response.text except Exception as e: logger.error(f“抓取网页失败 {url}: {e}”) return f“抓取失败: {str(e)}” tool def parse_html_with_bs(html_content: str, title_selector: Optional[str] “h1”, content_selector: Optional[str] “article”) - dict: “””使用 BeautifulSoup 解析 HTML提取标题和主要内容。“”” try: soup BeautifulSoup(html_content, ‘html.parser’) title_elem soup.select_one(title_selector) content_elem soup.select_one(content_selector) title title_elem.get_text(stripTrue) if title_elem else “” # 获取段落文本 if content_elem: paragraphs content_elem.find_all([‘p’, ‘h2’, ‘h3’]) content ‘\n’.join([p.get_text(stripTrue) for p in paragraphs]) else: content “” return {“title”: title, “content”: content[:1000]} # 限制长度 except Exception as e: logger.error(f“解析HTML失败: {e}”) return {“error”: str(e)}4.2 构建采集 AgentAgent 由 LLM、工具和提示词Prompt构成。# backend/app/agents/crawler_agent.py from langchain_openai import ChatOpenAI from langchain.agents import create_openai_tools_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from app.tools.web_tools import fetch_webpage, parse_html_with_bs import os from dotenv import load_dotenv load_dotenv() def get_crawler_agent() - AgentExecutor: “””创建并返回一个配置好的采集 Agent 执行器。“”” # 初始化 LLM这里使用 OpenAI GPT-4你需要设置 OPENAI_API_KEY llm ChatOpenAI(model“gpt-4-1106-preview”, temperature0, api_keyos.getenv(“OPENAI_API_KEY”)) # 定义 Agent 可用的工具 tools [fetch_webpage, parse_html_with_bs] # 构建提示词指导 Agent 如何工作 prompt ChatPromptTemplate.from_messages([ (“system”, “””你是一个专业的网络信息采集助手。你的任务是理解用户的需求并智能地调用工具来获取网页内容或解析数据。 用户可能会给你一个主题你需要从你已知的可靠来源或根据上下文获取相关信息。 如果你需要抓取网页请使用 fetch_webpage 工具。 如果你需要从 HTML 中提取特定信息请使用 parse_html_with_bs 工具。 请一步步思考并清晰地向用户汇报你的行动和发现。“””), MessagesPlaceholder(variable_name“chat_history”, optionalTrue), (“human”, “{input}”), MessagesPlaceholder(variable_name“agent_scratchpad”), ]) # 创建 Agent agent create_openai_tools_agent(llm, tools, prompt) # 创建执行器它负责运行 Agent 并处理工具调用循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor类似地你需要实现parser_agent.py,analyzer_agent.py等。解析 Agent 的工具可能更专注于文本清洗和结构化提取分析 Agent 则可能不需要外部工具直接利用 LLM 的强大分析能力或者调用情感分析、关键词提取的专用 API。5. 使用 FastAPI 暴露工作流 API现在我们需要一个方式来触发和执行这个工作流。FastAPI 将提供 RESTful API。5.1 创建 API 路由# backend/app/api/v1/endpoints/workflow.py from fastapi import APIRouter, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional from app.graphs.hotspot_graph import create_hotspot_workflow import asyncio from app.core.celery_app import celery_app # 假设使用 Celery 处理异步任务 router APIRouter() class WorkflowRequest(BaseModel): query: str async_exec: bool False # 是否异步执行 class WorkflowResponse(BaseModel): task_id: Optional[str] None status: str result: Optional[dict] None message: Optional[str] None router.post(“/run”, response_modelWorkflowResponse) async def run_hotspot_workflow(request: WorkflowRequest, background_tasks: BackgroundTasks): “””执行热点分析工作流“”” if not request.query or len(request.query.strip()) 0: raise HTTPException(status_code400, detail“查询内容不能为空”) if request.async_exec: # 异步执行提交给 Celery 任务队列 from .tasks import execute_workflow_task task execute_workflow_task.delay(request.query) return WorkflowResponse(task_idtask.id, status“PENDING”, message“任务已提交到队列”) else: # 同步执行注意长时间任务会阻塞仅用于演示或快速测试 try: graph create_hotspot_workflow() # LangGraph 的 invoke 是同步的对于复杂图考虑在线程池中运行 final_state graph.invoke({“query”: request.query}) if final_state.get(“error”): return WorkflowResponse(status“ERROR”, messagefinal_state[“error”]) return WorkflowResponse( status“SUCCESS”, result{ “report”: final_state.get(“final_report”, “”), “analysis”: final_state.get(“analysis_result”), “article_count”: len(final_state.get(“aggregated_articles”, [])) } ) except Exception as e: raise HTTPException(status_code500, detailf“工作流执行失败: {str(e)}”) router.get(“/task/{task_id}”) async def get_task_status(task_id: str): “””查询异步任务状态“”” from celery.result import AsyncResult task_result AsyncResult(task_id, appcelery_app) response { “task_id”: task_id, “status”: task_result.status, “result”: task_result.result if task_result.ready() else None } return response5.2 集成到主应用并配置 CORS# backend/app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.v1.api import api_router from app.core.config import settings app FastAPI(title“AI Hotspot Tracking Platform API”, version“1.0.0”) # 设置 CORS允许前端访问 app.add_middleware( CORSMiddleware, allow_origins[“http://localhost:3000”], # Nuxt 开发服务器默认端口 allow_credentialsTrue, allow_methods[“*”], allow_headers[“*”], ) # 包含 API 路由 app.include_router(api_router, prefix“/api/v1”) app.get(“/”) def read_root(): return {“message”: “AI Hotspot Tracking Platform API is running.”}使用 Uvicorn 启动后端cd backend uvicorn app.main:app --reload --host 0.0.0.0 --port 8000访问http://localhost:8000/docs查看自动生成的 API 文档。6. 构建 Nuxt 前端管理界面前端负责提供输入界面、触发工作流、展示结果和任务状态。6.1 创建可复用的 API 调用 Composables// frontend/composables/useWorkflowApi.ts import { ref } from ‘vue’; import axios from ‘axios’; const apiClient axios.create({ baseURL: ‘http://localhost:8000/api/v1’, // 后端地址 timeout: 60000, // 长任务需要更长时间 }); export interface WorkflowRequest { query: string; async_exec: boolean; } export interface WorkflowResponse { task_id?: string; status: string; result?: any; message?: string; } export function useWorkflowApi() { const loading ref(false); const error refstring | null(null); const runWorkflow async (request: WorkflowRequest): PromiseWorkflowResponse { loading.value true; error.value null; try { const response await apiClient.postWorkflowResponse(‘/workflow/run’, request); return response.data; } catch (err: any) { error.value err.response?.data?.detail || err.message; throw err; } finally { loading.value false; } }; const getTaskStatus async (taskId: string): Promiseany { try { const response await apiClient.get(/workflow/task/${taskId}); return response.data; } catch (err: any) { console.error(‘获取任务状态失败:’, err); throw err; } }; return { loading, error, runWorkflow, getTaskStatus, }; }6.2 创建热点分析页面!-- frontend/pages/index.vue -- template div class“container” h1AI 热点追踪分析平台/h1 el-card class“input-card” template #header span分析任务配置/span /template el-form :model“form” label-width“100px” el-form-item label“分析主题” el-input v-model“form.query” type“textarea” :rows“3” placeholder“请输入您想追踪的热点主题例如‘今日人工智能领域重大突破’” /el-input /el-form-item el-form-item label“执行方式” el-radio-group v-model“form.async_exec” el-radio :label“false”同步执行快速测试/el-radio el-radio :label“true”异步执行推荐长任务/el-radio /el-radio-group /el-form-item el-form-item el-button type“primary” click“handleSubmit” :loading“api.loading”开始分析/el-button el-button click“resetForm”重置/el-button /el-form-item /el-form /el-card el-card v-if“result || api.error” class“result-card” template #header span分析结果/span el-button v-if“taskId” size“small” click“pollTaskStatus” :loading“polling” 刷新状态 /el-button /template div v-if“api.error” class“error-message” el-alert :title“api.error” type“error” show-icon / /div div v-else-if“result” el-tabs v-model“activeTab” el-tab-pane label“报告” name“report” !-- 使用 v-html 需注意 XSS确保后端返回的是可信内容 -- div class“report-content” v-html“renderedReport”/div /el-tab-pane el-tab-pane label“原始数据” name“raw” pre{{ JSON.stringify(result, null, 2) }}/pre /el-tab-pane /el-tabs /div div v-else-if“taskId” el-alert :title“任务已提交任务ID: ${taskId}” type“info” show-icon / p状态: {{ taskStatus }}/p el-progress v-if“taskStatus ‘PENDING’ || taskStatus ‘STARTED’” :percentage“50” status“success” :indeterminate“true” / /div /el-card /div /template script setup lang“ts” import { ref, reactive, computed } from ‘vue’; import { useWorkflowApi } from ‘~/composables/useWorkflowApi’; import { marked } from ‘marked’; // 用于渲染 Markdown 报告 const api useWorkflowApi(); const form reactive({ query: ‘’, async_exec: true, }); const result refany(null); const taskId refstring | null(null); const taskStatus refstring(‘’); const polling ref(false); const activeTab ref(‘report’); const renderedReport computed(() { if (result.value?.result?.report) { return marked(result.value.result.report); } return ‘’; }); const handleSubmit async () { if (!form.query.trim()) { ElMessage.warning(‘请输入分析主题’); return; } try { const response await api.runWorkflow(form); if (response.task_id) { // 异步任务 taskId.value response.task_id; taskStatus.value response.status; result.value null; startPolling(); } else { // 同步任务直接返回结果 taskId.value null; result.value response; } } catch (error) { console.error(‘提交任务失败:’, error); } }; const startPolling () { const intervalId setInterval(async () { if (!taskId.value) { clearInterval(intervalId); return; } polling.value true; try { const statusResp await api.getTaskStatus(taskId.value); taskStatus.value statusResp.status; if (statusResp.status ‘SUCCESS’) { result.value { result: statusResp.result }; taskId.value null; clearInterval(intervalId); ElMessage.success(‘分析完成’); } else if (statusResp.status ‘FAILURE’) { ElMessage.error(‘任务执行失败’); taskId.value null; clearInterval(intervalId); } } catch (error) { console.error(‘轮询状态失败:’, error); } finally { polling.value false; } }, 2000); // 每2秒轮询一次 }; const pollTaskStatus async () { if (!taskId.value) return; try { const statusResp await api.getTaskStatus(taskId.value); taskStatus.value statusResp.status; } catch (error) { console.error(‘获取任务状态失败:’, error); } }; const resetForm () { form.query ‘’; form.async_exec true; result.value null; taskId.value null; }; /script style scoped .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .input-card, .result-card { margin-top: 20px; } .report-content { line-height: 1.6; } .error-message { margin-bottom: 15px; } /style启动前端开发服务器cd frontend npm run dev访问http://localhost:3000即可使用平台。7. 常见问题排查与优化实践将系统跑起来只是第一步在实际运行中你会遇到各种问题。以下是典型问题的排查路径和优化建议。7.1 工作流执行失败或卡住问题现象可能原因检查方式处理建议调用/runAPI 超时1. 同步执行模式工作流本身耗时过长。2. LLM API 调用慢或失败。3. 爬虫工具遇到反爬或网络问题。1. 查看后端日志uvicorn输出。2. 检查OPENAI_API_KEY是否正确网络是否通畅。3. 在爬虫工具中添加超时和重试逻辑并打印日志。1.务必使用异步模式async_exectrue并搭配 Celery/RQ 等任务队列。2. 为 LLM 调用设置合理的超时时间。3. 在爬虫工具中使用try-catch返回明确错误信息。Agent 陷入循环或无法决策提示词Prompt不够清晰导致 LLM 无法正确选择工具。查看 LangChain Agent 执行时的详细日志verboseTrue观察其“思考”过程。优化系统提示词明确每个工具的用途和调用条件。可以增加max_iterations参数限制循环次数。状态State传递错误GraphState 中字段定义与节点返回值不匹配。在每个节点函数的开始和结束打印state内容。确保每个节点返回的字典键名与GraphState定义一致。使用 Pydantic 模型进行状态验证是更严谨的做法。7.2 爬虫工具被网站屏蔽这是爬虫类项目最常见的问题。现象fetch_webpage返回 403、429 状态码或返回验证页面如“安全验证”。排查检查返回的 HTML 内容是否包含captcha、verification、access denied等关键词。使用工具如curl -I手动请求目标 URL查看响应头中的X-RateLimit-*等信息。解决策略遵守robots.txt在工具中集成robotparser尊重网站规则。设置请求头模拟真实浏览器如User-Agent,Referer,Accept-Language。使用会话和 Cookies对于需要登录的站点使用requests.Session()。添加延迟在连续请求间使用time.sleep(random.uniform(1, 3))避免高频访问。使用代理 IP 池对于反爬严格的站点这是必要手段。升级到无头浏览器对于 JavaScript 渲染的页面将工具替换为使用Playwright或Selenium。例如from langchain.tools import tool from playwright.sync_api import sync_playwright tool def fetch_dynamic_page(url: str) - str: “””使用无头浏览器获取动态渲染的页面内容。“”” with sync_playwright() as p: browser p.chromium.launch(headlessTrue) # 生产环境建议 headlessTrue page browser.new_page() try: page.goto(url, wait_until“networkidle”) content page.content() return content except Exception as e: return f“动态抓取失败: {str(e)}” finally: browser.close()7.3 LLM 调用成本与性能优化频繁调用 GPT-4 等模型成本很高且速度慢。策略一缓存对相同的查询或中间结果进行缓存。可以使用langchain.cache如SQLiteCache,RedisCache或自定义缓存逻辑。策略二使用更经济的模型对于清洗、解析等结构化提取任务可以尝试使用gpt-3.5-turbo或本地部署的小模型通过Ollama,Llama.cpp集成。策略三优化提示词清晰、具体的提示词能减少模型的“思考”时间Token 消耗和提高输出质量。使用少样本Few-shot提示。策略四流式输出与异步对于报告生成等长文本任务考虑使用流式响应提升前端用户体验。后端使用异步调用 LLM API。7.4 生产环境部署建议配置管理使用pydantic-settings管理所有敏感配置API Keys、数据库连接通过环境变量加载。数据库将工作流执行记录、分析结果、文章缓存存入 PostgreSQL 或 MySQL便于查询和历史回溯。任务队列必须使用Celery 或 RQ 处理工作流执行避免 HTTP 请求超时。并配置 Redis 作为 Broker 和结果后端。日志与监控为每个 Agent、工具和工作流节点添加结构化日志。集成 Sentry 或 Logstash 进行错误追踪和性能监控。前端优化对于长任务使用 WebSocket 或 Server-Sent Events (SSE) 向前端推送实时进度比轮询更高效。对分析结果进行分页和搜索。使用 ECharts 等库将趋势分析结果可视化。安全性API 接口添加认证如 JWT。对用户输入的query进行清洗防止注入攻击。限制单个用户/IP 的请求频率。8. 扩展方向与最佳实践构建出基础平台后可以从以下几个方向深化和优化增加更多数据源和 Agent社交媒体 Agent集成 Twitter/X、Reddit、微博等平台的 API 或爬虫。学术论文 Agent从 arXiv、Google Scholar 抓取最新论文。舆情情感 Agent专门分析评论和情感倾向。实现长期记忆与知识库使用LangGraph的Checkpointer或向量数据库如 Chroma, Weaviate存储历史分析结果。新的分析可以关联历史数据判断趋势是“持续升温”还是“新爆发”。工作流可视化与调试利用LangGraph的get_graph().draw_mermaid()功能生成工作流图。开发一个管理后台可以可视化查看每次工作流执行的状态流转和中间结果极大方便调试。评估与迭代设计评估指标信息抓取率、分析准确度、报告可读性。定期用一批标准问题测试系统根据结果调整提示词、工具或工作流逻辑。代码组织最佳实践依赖注入将 LLM 实例、工具集等通过依赖注入方式提供给 Agent 创建函数便于测试和配置切换。配置化将数据源列表、关键词、爬虫间隔等配置外置到 YAML 或数据库。单元测试为每个工具函数和独立的节点编写单元测试。使用pytest和pytest-asyncio。错误处理与重试在工作流层面增加错误处理节点对可重试的错误如网络超时进行有限次重试。这个基于多 Agent 和 LangGraph 的热点追踪分析平台提供了一个将 AI 能力工程化、管道化的范本。其价值不在于某个 Agent 有多智能而在于通过清晰的工作流将不确定的 LLM 调用、易碎的爬虫脚本、复杂的业务逻辑串联成一个稳定、可观测、可扩展的系统。在开发过程中最耗时的部分往往不是编写单个 Agent而是设计清晰的状态流、健壮的错误处理以及适配真实网络环境的工具。从这个小项目出发你可以逐步替换更强的模型、接入更丰富的数据源、设计更复杂的分析策略最终构建出真正实用的智能信息处理系统。