Vue + Java + Python 打造企业级 AI 知识库与任务分发系统（RAG架构全解析）

前言

一个支持 AI 对话的“任务分发与管理系统”。我的需求是：系统不仅能让用户手动创建和管理任务，还要包含一个AI 知识库（RAG，检索增强生成）模块。用户可以通过对话的方式，让 AI 基于项目内的专属文档、规范和历史记录进行问答，甚至自动下发任务。

面对复杂的业务逻辑和前沿的 AI 能力，我们最终敲定了Vue (前端) + Java (主业务后端) + Python (AI 专属微服务)的架构。这篇文章将完整复盘我们的架构设计、数据流转逻辑，并附上 Python 侧的核心源码，希望能为正在做大模型企业级落地的同学提供一些参考。

一、 为什么选择 Java + Python 双后端微服务架构？

在企业级 AI 应用落地中，语言的选择至关重要。纯 Java 搞 AI 生态稍弱，纯 Python 搞复杂业务和高并发又略显吃力。因此，“各司其职”是最佳方案：

• Vue (前端)：负责知识库的管理页面（上传文档、音视频）和类似 ChatGPT 的打字机对话界面。

• Java (主服务 / Spring Boot)：守住业务底线。负责网关鉴权、数据 CRUD（用户、项目、任务表）、文件 OSS 存储、以及代理前端与 Python 之间的流式通信（SSE）。

• Python (AI 微服务 / FastAPI)：拓展 AI 上限。提供纯粹的 AI API，负责文档解析、文本切分（Chunking）、向量化（Embedding）、操作向量数据库（Vector DB），以及与大语言模型（LLM）交互。

• 向量数据库：用于存储文档片段的向量数据（如 Milvus, Qdrant, 或 Chroma）

二、 核心工作流拆解

知识库的运作分为两个完全独立的阶段：“建库（让 AI 学习）”和“检索（让 AI 回答）”。

阶段一：知识录入与建库（异步处理是关键）

处理几百页的 PDF 或长视频非常耗时，绝不能使用同步的 HTTP 请求，否则前端必定超时。

1. 文件上传：前端上传文档至 Java 后端。

2. 落库存根：Java 校验权限，将文件存入 OSS，并在 MySQL 中生成一条“解析中”的记录，拿到document_id。

3. 消息队列 (MQ)：Java 将文件 URL 和项目 ID 丢入 RabbitMQ / Redis 队列后直接返回前端。

4. 后台解析：Python 消费队列，下载文件，解析切片，调用 Embedding 模型转化为向量，并打上project_id等元数据标签存入向量库。

5. 回调通知：Python 处理完毕后，通过 HTTP 回调 Java 接口，更新 MySQL 状态为“已完成”。

阶段二：流式问答与任务下发（SSE 协议代理）

大模型的回答是一个字一个字生成的，需要用到 Server-Sent Events (SSE) 协议。

1. 发起提问：用户提问请求打到 Java，Java 校验项目权限。

2. 请求透传：Java 携带project_id和用户问题，向 Python 发起流式 HTTP 请求。

3. 数据检索：Python 接收请求，严格根据project_id过滤向量数据库，召回最相关的文档片段。

4. 组装 Prompt 提示词：Python 将召回的知识与用户问题组装成 Prompt，发给大模型。

5. 流式返回：Python 通过 SSE 将大模型结果流式返回给 Java，Java 作为“透明代理”将数据流推送给 Vue 前端，实现打字机效果。

三、 Python AI 微服务核心代码实战

这里分享我们 Python 侧基于FastAPI+LangChain的极简核心代码。它包含了“文档向量化录入”和“流式对话检索”两个关键接口。

依赖安装：

Bash

pip install fastapi uvicorn langchain langchain-openai chromadb sse-starlette tiktoken

核心逻辑 (main.py)：

Python

from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough import uvicorn app = FastAPI(title="AI Knowledge Base Microservice") # 1. 初始化模型与向量库 (这里演示用 OpenAI，可无缝切换至 DeepSeek 或私有化 Ollama 模型) api_key = "your_api_key_here" embeddings = OpenAIEmbeddings(openai_api_key=api_key, model="text-embedding-3-small") llm = ChatOpenAI(openai_api_key=api_key, model="gpt-3.5-turbo", streaming=True) # 初始化本地向量数据库 Chroma vector_store = Chroma( collection_name="project_knowledge", embedding_function=embeddings, persist_directory="./chroma_db" ) # 2. 定义请求体参数 class BuildRequest(BaseModel): project_id: str document_id: str text_content: str class ChatRequest(BaseModel): project_id: str query: str # 3. 接口：知识库录入 (建库) @app.post("/api/knowledge/build") async def build_knowledge(req: BuildRequest): # 文本切分 text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) chunks = text_splitter.split_text(req.text_content) # 【核心踩坑点】注入元数据，用于后续的数据隔离 metadatas = [{"project_id": req.project_id, "document_id": req.document_id} for _ in chunks] # 存入向量库 vector_store.add_texts(texts=chunks, metadatas=metadatas) return {"status": "success", "message": f"成功录入 {len(chunks)} 个文本块"} # 4. 接口：带数据隔离的流式对话检索 (RAG) @app.post("/api/chat/stream") async def chat_stream(req: ChatRequest): # 【核心踩坑点】配置检索器：必须加上 filter 条件！ retriever = vector_store.as_retriever( search_kwargs={ "k": 3, "filter": {"project_id": req.project_id} } ) template = """你是一个专业的项目助手。请严格根据以下【背景知识】回答问题。 如果背景知识中没有相关内容，请回答“知识库中未找到相关信息”，不要编造。 【背景知识】\n{context}\n\n【用户问题】\n{question}""" prompt = ChatPromptTemplate.from_template(template) def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) # LangChain 表达式 (LCEL) 构建流水线 rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) async def generate_stream(): async for chunk in rag_chain.astream(req.query): # 必须严格遵守 SSE 协议格式 yield f"data: {chunk}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate_stream(), media_type="text/event-stream") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

快速测试：启动服务后，可以通过curl进行极速测试。

• 建库测试：curl -X POST "http://127.0.0.1:8000/api/knowledge/build" -H "Content-Type: application/json" -d '{"project_id": "P-1001", "document_id": "D-01", "text_content": "项目后端用Java，前端用Vue"}'

• 流式对话测试：curl -N -X POST "http://127.0.0.1:8000/api/chat/stream" -H "Content-Type: application/json" -d '{"project_id": "P-1001", "query": "项目用的什么技术栈？"}'

四、 深度避坑指南（干货）

在实际落地过程中，我们踩过几个大坑，分享给大家：

1. 数据串流灾难（Metadata 隔离）：千万不要把所有项目的文档混在一个向量库里直接查。录入时必须注入project_id作为 Metadata，检索时使用filter过滤。否则 A 项目的成员可能会问出 B 项目的机密设计图。

2. Java 端代理 SSE 流断连问题：Java 代理 Python 的流式响应时，Spring Boot 的默认超时时间可能很短。大模型思考慢时容易抛出TimeoutException，建议使用WebFlux的WebClient来进行代理，并显式调优超时配置。

3. 音视频文件的降维打击：传统的 RAG 只懂文本。如果前端允许上传视频会议记录，必须在 Python 端多加一层处理：先用FFmpeg提取音频，再用Whisper大模型将语音转成文本字幕，带上时间轴后再进行 Chunking。

结语

将 RAG 知识库与现有的任务系统结合，能极大地提升团队的协作效率。这套Vue + Java + Python的架构不仅能保证现有业务的绝对稳定，还能快速接入各种最新的大模型能力。

后续如果有机会，我会继续分享Java (Spring Boot) 端如何优雅地代理 SSE 流，以及Python 端如何处理复杂的 PDF 和 Excel 解析。欢迎大家在评论区交流指正！