如果你正在构建一个AI驱动的复杂应用比如一个能自动完成文献检索、大纲生成、内容撰写、格式排版的AI论文写作系统你可能会面临这样的困境单个AI模型能力有限而将多个AI能力串联成一个稳定、可维护的工作流技术复杂度又急剧上升。这正是多智能体Multi-Agent系统要解决的核心问题。传统的单线程AI调用无法处理需要多步骤协作的复杂任务。而LangGraph的出现让构建基于状态机的多Agent工作流变得像写流程图一样直观。本文将通过一个完整的AI论文写作系统项目带你实战FastAPIVue3LangGraph的全栈技术栈。你将不仅学会如何搭建一个多Agent系统更重要的是理解在真实项目中前后端如何协同处理AI任务的状态管理、异步通信和用户体验优化。1. 这篇文章真正要解决的问题在AI应用开发中最常见的痛点是如何将多个AI能力组合成可靠的工作流。以论文写作为例一个完整的流程需要文献检索Agent根据主题查找相关论文大纲生成Agent基于文献生成论文结构内容撰写Agent按章节填充具体内容格式校对Agent检查格式和引用规范如果使用传统的LangChain你可能需要写大量的回调函数和条件判断。而LangGraph通过有向图的概念让整个工作流可视化、可调试。更关键的是工程化问题当AI任务执行时间较长时前端如何实时展示进度后端如何管理任务状态如何避免请求超时这些才是真实项目中会遇到的挑战。本文的项目实战将重点解决多Agent工作流的可视化设计与实现FastAPI处理长时间运行AI任务的异步方案Vue3前端的状态管理和实时进度展示生产环境下的错误处理和重试机制2. 基础概念与核心原理2.1 什么是LangGraphLangGraph是LangChain团队推出的用于构建多Agent系统的框架。与LangChain相比它的核心创新在于引入了**状态机StateGraph**的概念。传统LangChain的问题# 传统的链式调用难以处理复杂分支 chain prompt | model | parser result chain.invoke({topic: AI论文主题})LangGraph的解决方案# 基于状态机的可视化工作流 graph_builder StateGraph(ResearchState) graph_builder.add_node(literature_search, search_node) graph_builder.add_node(outline_generation, outline_node) graph_builder.add_edge(literature_search, outline_generation)关键区别在于LangGraph将整个流程抽象为节点Node和边Edge每个节点可以访问和修改共享的状态State。2.2 多Agent系统的核心组件一个完整的Multi-Agent系统包含以下要素组件作用在论文写作系统中的对应角色Agent执行特定任务的智能体文献检索Agent、内容生成Agent等State工作流共享的状态数据论文主题、检索结果、生成内容等Node工作流中的处理节点每个Agent对应一个NodeEdge节点间的流转条件根据上一步结果决定下一步走向2.3 FastAPI Vue3的技术分工在这个全栈架构中每个技术栈承担明确职责FastAPI后端提供RESTful API、管理AI任务队列、处理WebSocket实时通信LangGraph编排多Agent工作流、管理任务状态和流转逻辑Vue3前端用户交互界面、实时进度展示、结果渲染和编辑这种分离架构的优势在于前端专注于用户体验后端专注于AI任务调度LangGraph专注于业务流程编排。3. 环境准备与前置条件3.1 系统环境要求# 检查Python版本 python --version # 需要Python 3.8 node --version # 需要Node.js 163.2 后端依赖环境创建requirements.txt文件fastapi0.104.1 uvicorn0.24.0 langgraph0.0.40 langchain0.1.0 openai1.3.0 python-multipart0.0.6 websockets12.0 redis5.0.1 # 用于任务状态缓存3.3 前端依赖环境创建Vue3项目的package.json核心依赖{ dependencies: { vue: ^3.3.0, vue-router: ^4.2.0, axios: ^1.6.0, element-plus: ^2.4.0, socket.io-client: ^4.7.0 } }3.4 开发工具准备后端开发推荐使用PyCharm或VS Code with Python插件前端开发VS Code Volar Vue扩展API测试Postman或Thunder Client4. 核心流程拆解4.1 多Agent论文写作工作流设计整个系统的核心工作流如下图所示文字描述用户输入论文主题 ↓ 文献检索Agent → 检索相关文献 ↓ 大纲生成Agent → 生成论文结构 ↓ 内容撰写Agent → 分章节撰写内容 ↓ 格式校对Agent → 检查格式规范 ↓ 输出完整论文每个节点都可能失败或需要人工干预因此需要完善的状态管理。4.2 后端API设计要点FastAPI需要提供以下关键接口POST /api/tasks- 创建新的论文写作任务GET /api/tasks/{task_id}- 获取任务状态和结果GET /api/tasks/{task_id}/progress- WebSocket进度推送PUT /api/tasks/{task_id}/pause- 暂停任务DELETE /api/tasks/{task_id}- 取消任务4.3 前端状态管理设计Vue3使用Pinia管理应用状态// stores/taskStore.js export const useTaskStore defineStore(task, { state: () ({ currentTask: null, progress: 0, steps: [文献检索, 大纲生成, 内容撰写, 格式校对], results: {} }) })5. 完整示例与代码实现5.1 后端LangGraph工作流实现首先定义共享状态类# models/research_state.py from typing import List, Dict, Any, Optional from pydantic import BaseModel class ResearchState(BaseModel): topic: str literature_results: Optional[List[Dict]] None outline: Optional[str] None content: Optional[Dict[str, str]] None formatted_paper: Optional[str] None current_step: str initialized error: Optional[str] None然后构建论文写作图# services/paper_writing_graph.py from langgraph.graph import StateGraph, END from models.research_state import ResearchState class PaperWritingGraph: def __init__(self): self.graph_builder StateGraph(ResearchState) self._build_graph() def _build_graph(self): # 添加节点 self.graph_builder.add_node(literature_search, self._literature_search) self.graph_builder.add_node(outline_generation, self._outline_generation) self.graph_builder.add_node(content_writing, self._content_writing) self.graph_builder.add_node(format_checking, self._format_checking) # 设置入口点 self.graph_builder.set_entry_point(literature_search) # 添加边 self.graph_builder.add_edge(literature_search, outline_generation) self.graph_builder.add_edge(outline_generation, content_writing) self.graph_builder.add_edge(content_writing, format_checking) self.graph_builder.add_edge(format_checking, END) self.graph self.graph_builder.compile() async def _literature_search(self, state: ResearchState): # 文献检索逻辑 state.current_step literature_search # 模拟文献检索过程 state.literature_results [ {title: 相关论文1, authors: 作者1, abstract: 摘要1}, {title: 相关论文2, authors: 作者2, abstract: 摘要2} ] return state async def _outline_generation(self, state: ResearchState): # 大纲生成逻辑 state.current_step outline_generation state.outline 1. 引言 2. 相关工作 3. 方法论 4. 实验结果 5. 结论 return state async def _content_writing(self, state: ResearchState): # 内容撰写逻辑 state.current_step content_writing state.content { introduction: 引言内容..., methodology: 方法论内容..., conclusion: 结论内容... } return state async def _format_checking(self, state: ResearchState): # 格式检查逻辑 state.current_step format_checking state.formatted_paper f # {state.topic} ## 摘要 本文探讨了{state.topic}的相关研究... ## 正文 {state.content} return state async def execute(self, topic: str) - ResearchState: initial_state ResearchState(topictopic) result await self.graph.ainvoke(initial_state) return result5.2 FastAPI后端服务实现# main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import asyncio import uuid from typing import Dict from services.paper_writing_graph import PaperWritingGraph from models.research_state import ResearchState # 任务状态存储 task_status: Dict[str, Dict] {} asynccontextmanager async def lifespan(app: FastAPI): # 启动时初始化 app.state.paper_writing_graph PaperWritingGraph() yield # 关闭时清理 task_status.clear() app FastAPI(lifespanlifespan) # CORS配置 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.post(/api/tasks) async def create_task(background_tasks: BackgroundTasks, topic: str): task_id str(uuid.uuid4()) # 初始化任务状态 task_status[task_id] { status: pending, progress: 0, result: None, error: None } # 后台执行任务 background_tasks.add_task(execute_paper_writing, task_id, topic) return {task_id: task_id, status: created} async def execute_paper_writing(task_id: str, topic: str): try: task_status[task_id][status] running # 执行LangGraph工作流 graph app.state.paper_writing_graph result await graph.execute(topic) task_status[task_id].update({ status: completed, progress: 100, result: result.dict() }) except Exception as e: task_status[task_id].update({ status: failed, error: str(e) }) app.get(/api/tasks/{task_id}) async def get_task_status(task_id: str): if task_id not in task_status: return {error: Task not found} return task_status[task_id] app.websocket(/api/tasks/{task_id}/progress) async def websocket_progress(websocket: WebSocket, task_id: str): await websocket.accept() try: while True: if task_id not in task_status: await websocket.send_json({error: Task not found}) break status task_status[task_id] await websocket.send_json({ status: status[status], progress: status[progress], current_step: status.get(current_step, ) }) # 如果任务完成或失败关闭连接 if status[status] in [completed, failed]: break await asyncio.sleep(1) # 每秒推送一次进度 except WebSocketDisconnect: pass5.3 Vue3前端实现创建任务管理组件!-- components/PaperWriting.vue -- template div classpaper-writing-container el-card headerAI论文写作系统 el-form :modelform submit.preventcreateTask el-form-item label论文主题 el-input v-modelform.topic placeholder请输入论文主题/el-input /el-form-item el-button typeprimary :loadingloading clickcreateTask 开始写作 /el-button /el-form /el-card el-card v-ifcurrentTask header写作进度 classprogress-card el-steps :activecurrentStepIndex align-center el-step title文献检索 :descriptionstepStatus[0]/el-step el-step title大纲生成 :descriptionstepStatus[1]/el-step el-step title内容撰写 :descriptionstepStatus[2]/el-step el-step title格式校对 :descriptionstepStatus[3]/el-step /el-steps el-progress :percentageprogress :statusprogressStatus classprogress-bar /el-progress div v-ifresult classresult-section h3生成结果/h3 el-input typetextarea :rows20 v-modelresult.formatted_paper readonly /el-input /div /el-card /div /template script setup import { ref, reactive, onUnmounted } from vue import { ElMessage } from element-plus import { useTaskStore } from /stores/taskStore const taskStore useTaskStore() const form reactive({ topic: }) const loading ref(false) const currentTask ref(null) const progress ref(0) const result ref(null) const socket ref(null) const steps [文献检索, 大纲生成, 内容撰写, 格式校对] const stepStatus ref([等待中, 等待中, 等待中, 等待中]) const currentStepIndex ref(0) const progressStatus computed(() { if (progress.value 100) return success if (progress.value 0) return exception return undefined }) const createTask async () { if (!form.topic.trim()) { ElMessage.error(请输入论文主题) return } loading.value true try { const response await fetch(/api/tasks, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ topic: form.topic }) }) const data await response.json() currentTask.value data.task_id connectWebSocket(data.task_id) } catch (error) { ElMessage.error(创建任务失败) } finally { loading.value false } } const connectWebSocket (taskId) { const protocol window.location.protocol https: ? wss: : ws: const wsUrl ${protocol}//${window.location.host}/api/tasks/${taskId}/progress socket.value new WebSocket(wsUrl) socket.value.onmessage (event) { const data JSON.parse(event.data) progress.value data.progress || 0 // 更新步骤状态 const stepIndex steps.findIndex(step data.current_step data.current_step.includes(step.toLowerCase()) ) if (stepIndex ! -1) { currentStepIndex.value stepIndex stepStatus.value[stepIndex] 进行中 } if (data.status completed) { fetchResult(taskId) stepStatus.value.fill(已完成) } else if (data.status failed) { ElMessage.error(任务失败: ${data.error}) stepStatus.value.fill(失败) } } } const fetchResult async (taskId) { try { const response await fetch(/api/tasks/${taskId}) const data await response.json() result.value data.result ElMessage.success(论文生成完成) } catch (error) { ElMessage.error(获取结果失败) } } onUnmounted(() { if (socket.value) { socket.value.close() } }) /script style scoped .paper-writing-container { max-width: 1200px; margin: 0 auto; padding: 20px; } .progress-card { margin-top: 20px; } .progress-bar { margin: 20px 0; } .result-section { margin-top: 30px; } /style6. 运行结果与效果验证6.1 启动后端服务# 启动FastAPI服务 uvicorn main:app --reload --host 0.0.0.0 --port 8000访问 http://localhost:8000/docs 查看API文档确认所有接口正常。6.2 启动前端服务# 启动Vue开发服务器 npm run dev访问 http://localhost:3000 打开论文写作系统界面。6.3 功能测试流程输入论文主题在输入框中输入人工智能在医疗诊断中的应用点击开始写作观察进度条和步骤状态变化监控实时进度通过WebSocket连接查看每个Agent的执行状态查看生成结果完成后在文本框中查看格式化的论文内容6.4 预期输出效果系统应该能够实时显示每个Agent的执行进度在2-3分钟内完成整个论文写作流程生成结构完整、格式规范的论文草稿正确处理中途失败和重试情况7. 常见问题与排查思路7.1 后端常见问题问题现象可能原因排查方式解决方案LangGraph节点执行失败AI API密钥配置错误检查OpenAI等API密钥配置验证API密钥有效性WebSocket连接断开网络问题或服务重启查看浏览器控制台错误实现自动重连机制任务状态丢失Redis连接失败或内存泄漏检查Redis连接状态添加连接重试机制并发任务冲突共享状态管理不当检查任务ID唯一性使用数据库锁机制7.2 前端常见问题问题现象可能原因排查方式解决方案进度条不更新WebSocket连接失败检查浏览器Network标签添加连接状态提示步骤状态显示错误状态映射逻辑错误检查current_step匹配逻辑完善状态映射表结果渲染格式混乱文本格式化问题检查生成的论文内容添加Markdown渲染7.3 性能优化问题# 添加任务超时控制 async def execute_paper_writing(task_id: str, topic: str): try: async with asyncio.timeout(300): # 5分钟超时 # 执行任务逻辑 pass except asyncio.TimeoutError: task_status[task_id].update({ status: timeout, error: 任务执行超时 })8. 最佳实践与工程建议8.1 生产环境部署配置Docker化部署# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]环境变量配置# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) REDIS_URL os.getenv(REDIS_URL, redis://localhost:6379) MAX_WORKERS int(os.getenv(MAX_WORKERS, 10))8.2 错误处理与重试机制# utils/retry.py import asyncio from typing import Callable, Any async def retry_async( func: Callable, max_retries: int 3, delay: float 1.0 ) - Any: for attempt in range(max_retries): try: return await func() except Exception as e: if attempt max_retries - 1: raise e await asyncio.sleep(delay * (2 ** attempt)) # 指数退避8.3 监控与日志记录# middleware/logging.py import time import logging from fastapi import Request logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) async def log_requests(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time logger.info( f{request.method} {request.url} fCompleted in {process_time:.2f}s fStatus: {response.status_code} ) return response8.4 安全最佳实践API密钥管理使用环境变量或密钥管理服务严禁硬编码输入验证对用户输入进行严格的验证和清理速率限制防止API滥用添加请求频率限制数据加密敏感数据在传输和存储时进行加密9. 扩展功能与进阶优化9.1 添加更多写作Agent# 添加参考文献管理Agent async def reference_management(self, state: ResearchState): state.current_step reference_management # 自动管理引用格式和文献列表 return state # 添加 plagiarism_check Agent async def plagiarism_check(self, state: ResearchState): state.current_step plagiarism_check # 查重检测逻辑 return state9.2 支持自定义写作模板class WritingTemplate: def __init__(self, structure: Dict, style: str): self.structure structure self.style style # 支持学术论文、技术报告、博客文章等不同模板9.3 实现协作编辑功能!-- 实时协作编辑组件 -- template div classcollaborative-editor h4实时协作编辑/h4 textarea v-modelsharedContent inputonContentChange placeholder多人协同编辑论文内容... /textarea div classonline-users 在线用户: {{ onlineUsers.join(, ) }} /div /div /template这个完整的FastAPIVue3LangGraph多Agent论文写作系统展示了如何将现代AI技术与全栈开发相结合。关键在于理解多Agent工作流的状态管理以及前后端在长时间任务处理中的协同机制。实际项目中你可能会根据具体需求调整Agent的数量和功能但核心架构模式是通用的。建议从本文的示例出发逐步添加更多个性化功能如多语言支持、专业领域优化等。代码已提供完整可运行版本建议在理解的基础上进行修改和扩展。遇到具体技术问题可以重点研究LangGraph的状态机设计和FastAPI的异步任务管理机制。