在实际 AI 应用开发中单靠一个语言模型往往难以完成复杂任务。真正的挑战在于如何让多个 AI 智能体协同工作每个智能体专注于特定子任务并通过有效的通信机制组合成完整的工作流。FastAPI 提供了高性能的异步后端支持Vue3 构建了现代化的前端交互界面而 LangGraph 则负责管理多智能体之间的状态流转和协作逻辑。本文将基于这三个技术栈实现一个完整的 AI 论文写作系统展示从需求分析到代码落地的全过程。这个系统不是简单的提示词拼接而是构建了一个包含文献检索、大纲生成、内容撰写、格式校对等多个专业智能体的协作网络。每个智能体都有自己的专业领域和决策逻辑通过 LangGraph 的有向图控制流实现任务分解和结果整合。最终用户只需要输入论文主题和基本要求系统就能自动完成从开题到成稿的完整流程。适合阅读本文的读者包括有一定 Python 和 JavaScript 基础的全栈开发者、对 AI 应用架构感兴趣的技术决策者、需要构建复杂 AI 工作流的工程团队。通过本文你将掌握多智能体系统的设计思路、LangGraph 的状态管理机制、FastAPI 的异步并发处理以及 Vue3 与前端的实时交互技巧。1. 理解多智能体系统的核心价值与架构设计1.1 为什么单智能体无法满足复杂 AI 任务需求传统的单智能体系统在处理论文写作这类复杂任务时面临几个关键瓶颈。首先单一模型很难同时具备文献检索、学术写作、格式规范等多领域专业知识。虽然大型语言模型在通用知识上表现优秀但涉及到特定领域的深度任务时需要更专业的子模块分工协作。其次长文本生成存在连贯性问题。当需要生成数千字的学术论文时单一提示词容易导致内容重复、结构松散或主题偏离。通过多智能体分工每个智能体只负责特定章节既能保证内容专业性又能通过整体架构维持逻辑连贯。最后是错误处理和质量控制的复杂性。在单智能体系统中任何一个环节出错都需要重新生成整个文档成本极高。而多智能体架构允许对单个环节进行局部优化和重试大大提升了系统的鲁棒性和可维护性。1.2 LangGraph 如何管理多智能体协作流程LangGraph 的核心创新在于将有向图的概念引入到智能体工作流管理中。与传统的线性链式调用不同LangGraph 允许定义复杂的条件分支和循环逻辑智能体之间的消息传递通过通道Channels机制实现。在一个典型的论文写作流程中LangGraph 会维护一个共享状态对象包含当前进度、已生成内容、用户需求等关键信息。每个智能体作为图中的一个节点只关注状态中与自己相关的部分完成任务后更新状态并决定下一步要激活哪个节点。这种架构的优势在于清晰的责任分离和灵活的工作流设计。例如当大纲生成智能体完成任务后可以根据内容复杂度决定是否激活文献检索智能体或者直接进入章节写作阶段。这种动态决策能力是传统链式调用无法实现的。1.3 系统整体架构与技术选型理由整个系统采用前后端分离架构后端使用 FastAPI 提供 RESTful API 和 WebSocket 支持前端使用 Vue3 构建响应式单页应用智能体引擎基于 LangGraph 构建。FastAPI 的选择主要基于其出色的异步性能和自动 API 文档生成能力。在多智能体场景下长时间任务处理是常态FastAPI 的异步特性可以高效管理并发请求避免阻塞主线程。同时其基于 Pydantic 的类型提示系统能够确保接口数据的规范性。Vue3 的 Composition API 特别适合管理复杂的组件状态。在论文写作过程中前端需要实时显示多个智能体的工作状态、生成进度和中间结果Vue3 的响应式系统能够优雅地处理这种多维状态变化。LangGraph 作为智能体编排框架相比 LangChain 提供了更精细的工作流控制能力。特别是其状态持久化和断点续传功能对于可能长时间运行的论文写作任务至关重要。2. 环境准备与项目初始化2.1 后端环境配置与依赖管理首先创建项目目录结构后端代码放在server目录下mkdir ai-paper-writer cd ai-paper-writer mkdir server client cd server创建 Python 虚拟环境并激活python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows安装核心依赖包创建requirements.txtfastapi0.104.1 uvicorn0.24.0 langgraph0.0.40 langchain0.0.350 openai1.3.0 python-dotenv1.0.0 pydantic2.5.0 sqlalchemy2.0.23 alembic1.12.1 websockets12.0安装依赖pip install -r requirements.txt环境变量配置创建.env文件OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLsqlite:///./paper_writer.db LOG_LEVELINFO2.2 前端项目初始化与依赖配置进入客户端目录并初始化 Vue3 项目cd ../client npm create vuelatest .选择需要的特性Router, Pinia, TypeScript等然后安装依赖npm install安装额外的 UI 库和工具包npm install element-plus element-plus/icons-vue npm install axios vue-axios npm install vueuse/core配置 Vue3 的主要入口文件main.tsimport { createApp } from vue import { createPinia } from pinia import ElementPlus from element-plus import * as ElementPlusIconsVue from element-plus/icons-vue import App from ./App.vue import router from ./router import element-plus/dist/index.css const app createApp(App) const pinia createPinia() for (const [key, component] of Object.entries(ElementPlusIconsVue)) { app.component(key, component) } app.use(pinia) app.use(router) app.use(ElementPlus) app.mount(#app)2.3 数据库设计与模型定义使用 SQLAlchemy 定义数据模型创建server/models.pyfrom sqlalchemy import Column, Integer, String, Text, DateTime, JSON from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class PaperProject(Base): __tablename__ paper_projects id Column(Integer, primary_keyTrue, indexTrue) title Column(String(200), nullableFalse) description Column(Text) status Column(String(50), defaultdraft) # draft, writing, reviewing, completed created_at Column(DateTime, defaultdatetime.utcnow) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) config Column(JSON) # 存储写作配置参数 class PaperSection(Base): __tablename__ paper_sections id Column(Integer, primary_keyTrue, indexTrue) project_id Column(Integer, indexTrue) section_type Column(String(50)) # abstract, introduction, method, result, discussion content Column(Text) status Column(String(50), defaultpending) # pending, writing, completed agent_logs Column(JSON) # 存储智能体执行日志 created_at Column(DateTime, defaultdatetime.utcnow)数据库初始化脚本server/database.pyfrom sqlalchemy import create_engine from models import Base import os DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./paper_writer.db) engine create_engine(DATABASE_URL) def init_db(): Base.metadata.create_all(bindengine) if __name__ __main__: init_db()3. 构建多智能体论文写作引擎3.1 定义智能体状态与工作流图创建智能体状态模型server/agents/state.pyfrom typing import TypedDict, List, Optional, Annotated from langgraph.graph import add_messages class PaperWritingState(TypedDict): messages: Annotated[List[str], add_messages] project_id: int current_section: str completed_sections: List[str] outline: Optional[dict] references: List[str] writing_style: str max_iterations: int current_iteration: int定义智能体工作流图server/agents/graph.pyfrom langgraph.graph import StateGraph, END from .state import PaperWritingState from .nodes import ( outline_agent_node, research_agent_node, writing_agent_node, review_agent_node, coordinator_agent_node ) def create_paper_writing_graph(): workflow StateGraph(PaperWritingState) # 添加节点 workflow.add_node(outline_agent, outline_agent_node) workflow.add_node(research_agent, research_agent_node) workflow.add_node(writing_agent, writing_agent_node) workflow.add_node(review_agent, review_agent_node) workflow.add_node(coordinator, coordinator_agent_node) # 设置入口点 workflow.set_entry_point(coordinator) # 定义边工作流逻辑 workflow.add_conditional_edges( coordinator, coordinator_agent_node.should_continue, { outline: outline_agent, research: research_agent, write: writing_agent, review: review_agent, end: END } ) workflow.add_edge(outline_agent, coordinator) workflow.add_edge(research_agent, coordinator) workflow.add_edge(writing_agent, coordinator) workflow.add_edge(review_agent, coordinator) return workflow.compile()3.2 实现各个专业智能体节点大纲生成智能体server/agents/nodes/outline_agent.pyfrom langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import json class OutlineAgent: def __init__(self): self.llm ChatOpenAI(modelgpt-4-1106-preview, temperature0.7) self.system_prompt 你是一位学术论文写作专家负责根据用户需求生成详细的论文大纲。 请按照学术论文的标准结构摘要、引言、相关工作、方法、实验、结果、讨论、结论来组织内容。 对于每个章节提供3-5个关键要点。 def generate_outline(self, topic: str, requirements: dict) - dict: user_prompt f 论文主题{topic} 写作要求{json.dumps(requirements, ensure_asciiFalse)} 请生成详细的论文大纲包含每个章节的核心要点。 messages [ SystemMessage(contentself.system_prompt), HumanMessage(contentuser_prompt) ] response self.llm.invoke(messages) return self._parse_outline(response.content) def _parse_outline(self, outline_text: str) - dict: # 解析LLM返回的大纲文本转换为结构化数据 sections {} current_section None for line in outline_text.split(\n): line line.strip() if line.startswith(##): current_section line.replace(##, ).strip() sections[current_section] [] elif line.startswith(-) and current_section: point line.replace(-, ).strip() sections[current_section].append(point) return sections def outline_agent_node(state: dict) - dict: agent OutlineAgent() topic state.get(topic, ) requirements state.get(requirements, {}) outline agent.generate_outline(topic, requirements) return { outline: outline, messages: [f大纲生成完成共{len(outline)}个主要章节] }文献检索智能体server/agents/nodes/research_agent.pyimport asyncio from scholarly import scholarly from langchain.tools import Tool from langchain.agents import AgentType, initialize_agent class ResearchAgent: def __init__(self): self.search_tool Tool( nameliterature_search, funcself.search_literature, description搜索相关学术文献 ) def search_literature(self, query: str) - str: 使用scholarly库搜索学术文献 try: search_query scholarly.search_pubs(query) results [] for i, result in enumerate(search_query): if i 5: # 限制结果数量 break results.append({ title: result.get(bib, {}).get(title, ), author: result.get(bib, {}).get(author, ), year: result.get(bib, {}).get(year, ), abstract: result.get(bib, {}).get(abstract, )[:200] ... }) return str(results) except Exception as e: return f文献搜索失败: {str(e)} def research_agent_node(state: dict) - dict: agent ResearchAgent() topic state.get(topic, ) outline state.get(outline, {}) # 根据大纲中的关键章节生成搜索查询 search_queries [ f{topic} {section} for section in outline.keys() ] references [] for query in search_queries: results agent.search_literature(query) references.append({query: query, results: results}) return { references: references, messages: [f文献检索完成找到{len(references)}组相关文献] }3.3 配置智能体间的通信与状态管理创建智能体协调器server/agents/nodes/coordinator_agent.pyfrom langchain_core.messages import SystemMessage, HumanMessage from langchain_openai import ChatOpenAI from typing import Literal class CoordinatorAgent: def __init__(self): self.llm ChatOpenAI(modelgpt-4, temperature0.3) self.system_prompt 你是论文写作流程的协调器负责决定下一步应该执行哪个智能体任务。 可用的任务类型包括 - outline: 需要生成论文大纲 - research: 需要检索相关文献 - write: 需要开始章节写作 - review: 需要审核已写内容 - end: 所有任务已完成 根据当前状态决定下一步行动。 def decide_next_step(self, state: dict) - Literal[outline, research, write, review, end]: current_section state.get(current_section) completed_sections state.get(completed_sections, []) outline state.get(outline) references state.get(references, []) # 决策逻辑 if not outline: return outline elif not references and outline: return research elif not current_section and outline: # 选择第一个未完成的章节 all_sections list(outline.keys()) remaining [s for s in all_sections if s not in completed_sections] if remaining: return write elif current_section and current_section not in completed_sections: return write elif len(completed_sections) len(outline): return review else: return end return end def coordinator_agent_node(state: dict) - dict: agent CoordinatorAgent() next_step agent.decide_next_step(state) # 更新状态 update {next_step: next_step} if next_step write and not state.get(current_section): # 设置当前写作章节 outline state.get(outline, {}) completed state.get(completed_sections, []) remaining [s for s in outline.keys() if s not in completed] if remaining: update[current_section] remaining[0] return update def should_continue(state: dict) - str: 条件判断函数用于图的条件边 return state.get(next_step, end)4. FastAPI 后端接口设计与实现4.1 项目管理与状态查询接口创建主要的 API 路由server/api/routes/projects.pyfrom fastapi import APIRouter, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import List, Optional from models import PaperProject, PaperSection from database import SessionLocal from agents.graph import create_paper_writing_graph router APIRouter(prefix/api/projects, tags[projects]) class CreateProjectRequest(BaseModel): title: str description: Optional[str] None config: Optional[dict] None class ProjectResponse(BaseModel): id: int title: str description: Optional[str] status: str created_at: str router.post(/, response_modelProjectResponse) async def create_project(request: CreateProjectRequest, background_tasks: BackgroundTasks): db SessionLocal() try: project PaperProject( titlerequest.title, descriptionrequest.description, configrequest.config or {} ) db.add(project) db.commit() db.refresh(project) # 在后台启动写作流程 background_tasks.add_task(start_writing_process, project.id) return ProjectResponse( idproject.id, titleproject.title, descriptionproject.description, statusproject.status, created_atproject.created_at.isoformat() ) finally: db.close() router.get(/{project_id}, response_modelProjectResponse) async def get_project(project_id: int): db SessionLocal() try: project db.query(PaperProject).filter(PaperProject.id project_id).first() if not project: raise HTTPException(status_code404, detail项目不存在) return project finally: db.close() router.get(/{project_id}/sections) async def get_project_sections(project_id: int): db SessionLocal() try: sections db.query(PaperSection).filter(PaperSection.project_id project_id).all() return {sections: sections} finally: db.close() def start_writing_process(project_id: int): 启动论文写作流程 db SessionLocal() try: project db.query(PaperProject).filter(PaperProject.id project_id).first() if not project: return # 初始化写作图 graph create_paper_writing_graph() # 初始状态 initial_state { project_id: project_id, topic: project.title, requirements: project.config, messages: [], completed_sections: [], current_iteration: 0, max_iterations: 10 } # 执行图 for step in graph.stream(initial_state): # 更新项目状态和保存进度 update_project_progress(project_id, step) finally: db.close() def update_project_progress(project_id: int, step_state: dict): 更新项目进度到数据库 db SessionLocal() try: # 更新项目状态 project db.query(PaperProject).filter(PaperProject.id project_id).first() if project: project.status writing db.commit() # 保存章节内容 if current_section in step_state and content in step_state: section PaperSection( project_idproject_id, section_typestep_state[current_section], contentstep_state[content], statuscompleted, agent_logsstep_state.get(messages, []) ) db.add(section) db.commit() finally: db.close()4.2 WebSocket 实时进度推送创建 WebSocket 端点用于实时进度更新server/api/websockets/progress.pyimport json from fastapi import WebSocket, WebSocketDisconnect from typing import Dict class ConnectionManager: def __init__(self): self.active_connections: Dict[int, WebSocket] {} async def connect(self, websocket: WebSocket, project_id: int): await websocket.accept() self.active_connections[project_id] websocket def disconnect(self, project_id: int): if project_id in self.active_connections: del self.active_connections[project_id] async def send_progress_update(self, project_id: int, message: dict): if project_id in self.active_connections: try: await self.active_connections[project_id].send_json(message) except Exception: self.disconnect(project_id) manager ConnectionManager() router.websocket(/ws/progress/{project_id}) async def websocket_endpoint(websocket: WebSocket, project_id: int): await manager.connect(websocket, project_id) try: while True: # 保持连接等待客户端消息 data await websocket.receive_text() # 可以处理客户端发送的指令 if data get_status: # 返回当前状态 pass except WebSocketDisconnect: manager.disconnect(project_id)4.3 异步任务处理与性能优化配置异步任务处理器server/core/tasks.pyimport asyncio from concurrent.futures import ThreadPoolExecutor from langgraph.graph import StateGraph # 使用线程池处理CPU密集型任务 executor ThreadPoolExecutor(max_workers4) async def process_writing_task(graph: StateGraph, initial_state: dict): 异步处理写作任务 loop asyncio.get_event_loop() def sync_graph_execution(): results [] for step in graph.stream(initial_state): results.append(step) return results # 在线程池中执行同步的图计算 results await loop.run_in_executor(executor, sync_graph_execution) return results router.post(/{project_id}/start) async def start_writing(project_id: int, background_tasks: BackgroundTasks): 启动写作任务异步版本 db SessionLocal() try: project db.query(PaperProject).filter(PaperProject.id project_id).first() if not project: raise HTTPException(status_code404, detail项目不存在) graph create_paper_writing_graph() initial_state { project_id: project_id, topic: project.title, requirements: project.config or {} } # 使用后台任务异步处理 background_tasks.add_task( process_writing_task, graph, initial_state ) return {message: 写作任务已开始, project_id: project_id} finally: db.close()5. Vue3 前端界面开发5.1 项目创建与状态管理创建项目创建页面client/src/views/CreateProject.vuetemplate div classcreate-project el-card header新建论文项目 el-form :modelform :rulesrules refformRef label-width120px el-form-item label论文主题 proptitle el-input v-modelform.title placeholder请输入论文主题 / /el-form-item el-form-item label项目描述 propdescription el-input v-modelform.description typetextarea :rows3 placeholder简要描述论文要求和目标 / /el-form-item el-form-item label写作配置 el-collapse el-collapse-item title高级配置 el-form-item label写作风格 el-select v-modelform.config.style placeholder选择写作风格 el-option label学术严谨 valueacademic / el-option label技术报告 valuetechnical / el-option label综述文章 valuereview / /el-select /el-form-item el-form-item label目标字数 el-input-number v-modelform.config.targetWords :min1000 :max10000 :step500 / /el-form-item /el-collapse-item /el-collapse /el-form-item el-form-item el-button typeprimary clickhandleSubmit :loadingloading 创建项目并开始写作 /el-button /el-form-item /el-form /el-card /div /template script setup langts import { ref, reactive } from vue import { useRouter } from vue-router import { ElMessage } from element-plus import { useProjectStore } from /stores/project const router useRouter() const projectStore useProjectStore() const formRef ref() const loading ref(false) const form reactive({ title: , description: , config: { style: academic, targetWords: 3000 } }) const rules { title: [ { required: true, message: 请输入论文主题, trigger: blur }, { min: 5, max: 200, message: 长度在 5 到 200 个字符, trigger: blur } ] } const handleSubmit async () { if (!formRef.value) return try { await formRef.value.validate() loading.value true const project await projectStore.createProject(form) ElMessage.success(项目创建成功) router.push(/project/${project.id}) } catch (error) { ElMessage.error(创建失败 error.message) } finally { loading.value false } } /script创建状态管理 Storeclient/src/stores/project.tsimport { defineStore } from pinia import { ref } from vue import { projectApi } from /api/project interface Project { id: number title: string description: string status: string created_at: string } export const useProjectStore defineStore(project, () { const currentProject refProject | null(null) const projects refProject[]([]) const createProject async (data: any) { const response await projectApi.createProject(data) projects.value.push(response.data) return response.data } const fetchProject async (id: number) { const response await projectApi.getProject(id) currentProject.value response.data return response.data } const fetchProjects async () { const response await projectApi.getProjects() projects.value response.data return response.data } return { currentProject, projects, createProject, fetchProject, fetchProjects } })5.2 实时进度监控界面创建项目详情页面client/src/views/ProjectDetail.vuetemplate div classproject-detail el-page-header backrouter.back() template #content h1{{ project?.title }}/h1 el-tag :typestatusType{{ statusText }}/el-tag /template /el-page-header el-row :gutter20 classmt-4 el-col :span16 el-card header写作进度 div classprogress-container el-steps :activeactiveStep align-center el-step title大纲生成 :descriptionstepStatus.outline / el-step title文献检索 :descriptionstepStatus.research / el-step title内容写作 :descriptionstepStatus.writing / el-step title审核校对 :descriptionstepStatus.review / /el-steps div classagent-status mt-4 h3智能体工作状态/h3 el-timeline el-timeline-item v-for(log, index) in agentLogs :keyindex :timestamplog.timestamp :typelog.type {{ log.message }} /el-timeline-item /el-timeline /div /div /el-card el-card header生成内容 classmt-4 el-tabs v-modelactiveSection el-tab-pane v-forsection in sections :keysection.id :labelsection.section_type :namesection.section_type div classsection-content pre{{ section.content }}/pre /div /el-tab-pane /el-tabs /el-card /el-col el-col :span8 el-card header实时状态 div classwebsocket-status el-alert :titlewsStatus :typewsConnected ? success : warning :closablefalse / el-button clickconnectWebSocket :disabledwsConnected classmt-2 {{ wsConnected ? 已连接 : 连接实时更新 }} /el-button /div /el-card el-card header智能体详情 classmt-4 el-descriptions :column1 border el-descriptions-item label当前活跃智能体 {{ activeAgent || 无 }} /el-descriptions-item el-descriptions-item label已处理章节 {{ completedSections.length }} / {{ totalSections }} /el-descriptions-item el-descriptions-item label预计剩余时间 {{ estimatedTime }} /el-descriptions-item /el-descriptions /el-card /el-col /el-row /div /template script setup langts import { ref, computed, onMounted, onUnmounted } from vue import { useRoute, useRouter } from vue-router import { useProjectStore } from /stores/project import { useWebSocket } from /composables/useWebSocket const route useRoute() const router useRouter() const projectStore useProjectStore() const project ref(null) const sections ref([]) const activeSection ref() const agentLogs ref([]) const activeAgent ref() const { connect, disconnect, status: wsStatus, connected: wsConnected } useWebSocket() const statusType computed(() { const statusMap { draft: info, writing: warning, reviewing: primary, completed: success } return statusMap[project.value?.status] || info }) const statusText computed(() { const textMap { draft: 草稿, writing: 写作中, reviewing: 审核中, completed: 已完成 } return textMap[project.value?.status] || 未知 }) const activeStep computed(() { const stepMap { draft: 0, writing: 2, reviewing: 3, completed: 4 } return stepMap[project.value?.status] || 0 }) const completedSections computed(() { return sections.value.filter(s s.status completed) }) const totalSections computed(() sections.value.length) const estimatedTime computed(() { const remaining totalSections.value - completedSections.value.length return ${remaining * 5} 分钟 }) const stepStatus computed(() { return { outline: completedSections.value.length 0 ? 已完成 : 进行中, research: 待开始, writing: ${completedSections.value.length}/${totalSections.value}, review: 待开始 } }) onMounted(async () { const projectId parseInt(route.params.id as string) await loadProject(projectId) connectWebSocket() }) onUnmounted(() { disconnect() }) const loadProject async (id: number) { project.value await projectStore.fetchProject(id) sections.value await projectStore.fetchProjectSections(id) if (sections.value.length 0) { activeSection.value sections.value[0].section_type } } const connectWebSocket () { const projectId parseInt(route.params.id as string) connect(projectId, { onMessage: (data) { // 处理实时更新 if (data.type progress) { agentLogs.value.push({ timestamp: new Date().toLocaleTimeString(), message: data.message, type: success }) } else if (data.type agent_update) { activeAgent.value data.agent } } }) } /script5.3 WebSocket 连接管理与状态同步创建 WebSocket 组合式函数client/src/composables/useWebSocket.tsimport { ref, onUnmounted } from vue interface WebSocketOptions { onMessage: (data: any) void onError?: (error: Event) void onClose?: (event: CloseEvent) void } export function useWebSocket() { const ws refWebSocket | null(null) const connected ref(false) const status ref(未连接) const connect (projectId: number, options: WebSocketOptions) { if (ws.value) { disconnect() } const protocol window.location.protocol https: ? wss: : ws: const wsUrl ${protocol}//${window.location.host}/api/ws/progress/${projectId} try { ws.value new WebSocket(wsUrl) status.value 连接中... ws.value.onopen () { connected.value true status.value 已连接 } ws.value.onmessage (event) { try { const data JSON.parse(event.data) options.onMessage(data) } catch (error) { console.error(WebSocket消息解析错误:, error) } } ws.value.onerror (error) { status.value 连接错误 options.onError?.(error
