前言一个支持 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 请求否则前端必定超时。文件上传前端上传文档至 Java 后端。落库存根Java 校验权限将文件存入 OSS并在 MySQL 中生成一条“解析中”的记录拿到document_id。消息队列 (MQ)Java 将文件 URL 和项目 ID 丢入 RabbitMQ / Redis 队列后直接返回前端。后台解析Python 消费队列下载文件解析切片调用 Embedding 模型转化为向量并打上project_id等元数据标签存入向量库。回调通知Python 处理完毕后通过 HTTP 回调 Java 接口更新 MySQL 状态为“已完成”。阶段二流式问答与任务下发SSE 协议代理大模型的回答是一个字一个字生成的需要用到 Server-Sent Events (SSE) 协议。发起提问用户提问请求打到 JavaJava 校验项目权限。请求透传Java 携带project_id和用户问题向 Python 发起流式 HTTP 请求。数据检索Python 接收请求严格根据project_id过滤向量数据库召回最相关的文档片段。组装 Prompt 提示词Python 将召回的知识与用户问题组装成 Prompt发给大模型。流式返回Python 通过 SSE 将大模型结果流式返回给 JavaJava 作为“透明代理”将数据流推送给 Vue 前端实现打字机效果。三、 Python AI 微服务核心代码实战这里分享我们 Python 侧基于FastAPILangChain的极简核心代码。它包含了“文档向量化录入”和“流式对话检索”两个关键接口。依赖安装Bashpip install fastapi uvicorn langchain langchain-openai chromadb sse-starlette tiktoken核心逻辑 (main.py)Pythonfrom 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(titleAI Knowledge Base Microservice) # 1. 初始化模型与向量库 (这里演示用 OpenAI可无缝切换至 DeepSeek 或私有化 Ollama 模型) api_key your_api_key_here embeddings OpenAIEmbeddings(openai_api_keyapi_key, modeltext-embedding-3-small) llm ChatOpenAI(openai_api_keyapi_key, modelgpt-3.5-turbo, streamingTrue) # 初始化本地向量数据库 Chroma vector_store Chroma( collection_nameproject_knowledge, embedding_functionembeddings, 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_size500, chunk_overlap50) 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(textschunks, metadatasmetadatas) 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 fdata: {chunk}\n\n yield data: [DONE]\n\n return StreamingResponse(generate_stream(), media_typetext/event-stream) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)快速测试启动服务后可以通过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: 项目用的什么技术栈}四、 深度避坑指南干货在实际落地过程中我们踩过几个大坑分享给大家数据串流灾难Metadata 隔离千万不要把所有项目的文档混在一个向量库里直接查。录入时必须注入project_id作为 Metadata检索时使用filter过滤。否则 A 项目的成员可能会问出 B 项目的机密设计图。Java 端代理 SSE 流断连问题Java 代理 Python 的流式响应时Spring Boot 的默认超时时间可能很短。大模型思考慢时容易抛出TimeoutException建议使用WebFlux的WebClient来进行代理并显式调优超时配置。音视频文件的降维打击传统的 RAG 只懂文本。如果前端允许上传视频会议记录必须在 Python 端多加一层处理先用FFmpeg提取音频再用Whisper大模型将语音转成文本字幕带上时间轴后再进行 Chunking。结语将 RAG 知识库与现有的任务系统结合能极大地提升团队的协作效率。这套Vue Java Python的架构不仅能保证现有业务的绝对稳定还能快速接入各种最新的大模型能力。后续如果有机会我会继续分享Java (Spring Boot) 端如何优雅地代理 SSE 流以及Python 端如何处理复杂的 PDF 和 Excel 解析。欢迎大家在评论区交流指正
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 请求否则前端必定超时。文件上传前端上传文档至 Java 后端。落库存根Java 校验权限将文件存入 OSS并在 MySQL 中生成一条“解析中”的记录拿到document_id。消息队列 (MQ)Java 将文件 URL 和项目 ID 丢入 RabbitMQ / Redis 队列后直接返回前端。后台解析Python 消费队列下载文件解析切片调用 Embedding 模型转化为向量并打上project_id等元数据标签存入向量库。回调通知Python 处理完毕后通过 HTTP 回调 Java 接口更新 MySQL 状态为“已完成”。阶段二流式问答与任务下发SSE 协议代理大模型的回答是一个字一个字生成的需要用到 Server-Sent Events (SSE) 协议。发起提问用户提问请求打到 JavaJava 校验项目权限。请求透传Java 携带project_id和用户问题向 Python 发起流式 HTTP 请求。数据检索Python 接收请求严格根据project_id过滤向量数据库召回最相关的文档片段。组装 Prompt 提示词Python 将召回的知识与用户问题组装成 Prompt发给大模型。流式返回Python 通过 SSE 将大模型结果流式返回给 JavaJava 作为“透明代理”将数据流推送给 Vue 前端实现打字机效果。三、 Python AI 微服务核心代码实战这里分享我们 Python 侧基于FastAPILangChain的极简核心代码。它包含了“文档向量化录入”和“流式对话检索”两个关键接口。依赖安装Bashpip install fastapi uvicorn langchain langchain-openai chromadb sse-starlette tiktoken核心逻辑 (main.py)Pythonfrom 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(titleAI Knowledge Base Microservice) # 1. 初始化模型与向量库 (这里演示用 OpenAI可无缝切换至 DeepSeek 或私有化 Ollama 模型) api_key your_api_key_here embeddings OpenAIEmbeddings(openai_api_keyapi_key, modeltext-embedding-3-small) llm ChatOpenAI(openai_api_keyapi_key, modelgpt-3.5-turbo, streamingTrue) # 初始化本地向量数据库 Chroma vector_store Chroma( collection_nameproject_knowledge, embedding_functionembeddings, 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_size500, chunk_overlap50) 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(textschunks, metadatasmetadatas) 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 fdata: {chunk}\n\n yield data: [DONE]\n\n return StreamingResponse(generate_stream(), media_typetext/event-stream) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)快速测试启动服务后可以通过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: 项目用的什么技术栈}四、 深度避坑指南干货在实际落地过程中我们踩过几个大坑分享给大家数据串流灾难Metadata 隔离千万不要把所有项目的文档混在一个向量库里直接查。录入时必须注入project_id作为 Metadata检索时使用filter过滤。否则 A 项目的成员可能会问出 B 项目的机密设计图。Java 端代理 SSE 流断连问题Java 代理 Python 的流式响应时Spring Boot 的默认超时时间可能很短。大模型思考慢时容易抛出TimeoutException建议使用WebFlux的WebClient来进行代理并显式调优超时配置。音视频文件的降维打击传统的 RAG 只懂文本。如果前端允许上传视频会议记录必须在 Python 端多加一层处理先用FFmpeg提取音频再用Whisper大模型将语音转成文本字幕带上时间轴后再进行 Chunking。结语将 RAG 知识库与现有的任务系统结合能极大地提升团队的协作效率。这套Vue Java Python的架构不仅能保证现有业务的绝对稳定还能快速接入各种最新的大模型能力。后续如果有机会我会继续分享Java (Spring Boot) 端如何优雅地代理 SSE 流以及Python 端如何处理复杂的 PDF 和 Excel 解析。欢迎大家在评论区交流指正
