摘要
本文详细介绍如何利用 Python 生态中的 LangChain/LlamaIndex 框架，结合 FastAPI 构建后端服务，使用 Streamlit 开发前端界面，并集成本地部署的 Ollama 与 Qwen 大语言模型，打造一套完全离线、可运行于企业内网环境的 RAG（检索增强生成）知识问答系统。文章将从系统架构设计、环境搭建、核心代码实现、前后端集成、模型本地化部署到最终系统展示，提供完整的实战指南。
1. 系统架构与设计目标
在企业内网环境中，数据安全与隐私保护是首要考虑。本系统设计目标如下：
- 完全离线：所有组件（大模型、向量数据库、前后端服务）均部署于内网，无需连接外网。
- 开源技术栈：采用成熟的 Python 开源框架，降低技术门槛与授权成本。
- 模块化设计：前后端分离，便于独立开发、测试与部署。
- 易于扩展：支持更换不同的本地大模型、向量数据库及前端框架。
整体架构如下图所示：
架构分为四层：
- 数据层：本地文档（PDF、Word、TXT等）经过文本提取与分割后，存入本地向量数据库（如 Chroma、FAISS）。
- 模型层：本地部署的 Ollama 服务，加载 Qwen 等开源大模型。
- 后端服务层：基于 FastAPI 构建，集成 LangChain/LlamaIndex 实现 RAG 检索与生成流程。
- 前端交互层：使用 Streamlit 构建简洁的 Web 界面，供用户提问并展示答案。
2. 环境准备与依赖安装
首先，确保内网服务器或开发机已安装 Python 3.8+。创建虚拟环境并安装核心依赖。
# 创建项目目录 mkdir enterprise-rag-system && cd enterprise-rag-system 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows 安装核心 Python 包 pip install fastapi uvicorn pip install langchain langchain-community langchain-chroma 或者使用 LlamaIndex pip install llama-index llama-index-vector-stores-chroma pip install streamlit pip install pypdf python-docx # 文档解析 pip install sentence-transformers # 本地 embedding 模型安装并启动 Ollama：
从 Ollama 官网下载适用于内网系统的离线安装包，或在一台可联网的机器上提前拉取模型再迁移至内网。
# 在可联网机器上拉取 Qwen 模型 (例如 7B 版本) ollama pull qwen2.5:7b 将模型文件（通常位于 ~/.ollama/models）拷贝至内网服务器对应目录 在内网服务器启动 Ollama 服务 ollama serve 默认服务地址为 http://localhost:114343. 后端服务开发 (FastAPI + LangChain)
后端主要负责文档处理、向量检索和调用本地大模型生成答案。
3.1 项目结构
backend/ ├── main.py # FastAPI 主应用 ├── rag_core.py # RAG 核心逻辑 ├── models.py # 数据模型 ├── chroma_db/ # 向量数据库存储目录 └── documents/ # 待处理的原始文档目录3.2 核心 RAG 逻辑 (rag_core.py)
import os from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader, TextLoader, Docx2txtLoader from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma from langchain_community.llms import Ollama from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate class RAGSystem: def init(self, model_name="qwen2.5:7b", persist_directory="./chroma_db"): # 初始化本地 embedding 模型 self.embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2" ) # 初始化本地 LLM 通过 Ollama self.llm = Ollama(base_url="http://localhost:11434", model=model_name) self.persist_directory = persist_directory self.vectorstore = None self.qa_chain = None def load_and_split_documents(self, document_dir): """加载并分割文档""" documents = [] for filename in os.listdir(document_dir): filepath = os.path.join(document_dir, filename) if filename.endswith(".pdf"): loader = PyPDFLoader(filepath) elif filename.endswith(".txt"): loader = TextLoader(filepath) elif filename.endswith(".docx"): loader = Docx2txtLoader(filepath) else: continue documents.extend(loader.load()) # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) splits = text_splitter.split_documents(documents) return splits def create_vectorstore(self, splits): """创建向量数据库""" self.vectorstore = Chroma.from_documents( documents=splits, embedding=self.embeddings, persist_directory=self.persist_directory ) self.vectorstore.persist() return self.vectorstore def load_existing_vectorstore(self): """加载已存在的向量数据库""" self.vectorstore = Chroma( persist_directory=self.persist_directory, embedding_function=self.embeddings ) return self.vectorstore def build_qa_chain(self): """构建 QA 链""" # 自定义提示词 prompt_template = """基于以下上下文信息，请用中文回答用户的问题。如果上下文信息不足以回答问题，请直接说“根据现有资料无法回答该问题”，不要编造答案。 上下文： {context} 问题：{question} 答案：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 创建检索器 retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3}) # 构建 QA 链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=retriever, chain_type_kwargs={"prompt": PROMPT} ) return self.qa_chain def query(self, question): """执行查询""" if not self.qa_chain: raise ValueError("请先调用 build_qa_chain() 初始化 QA 链") result = self.qa_chain.run(question) return result</code></pre> 3.3 FastAPI 主应用 (main.py) from fastapi import FastAPI, HTTPException from pydantic import BaseModel from rag_core import RAGSystem import uvicorn app = FastAPI(title="企业内网 RAG 问答系统后端") 初始化 RAG 系统 rag_system = RAGSystem() 数据模型 class QueryRequest(BaseModel): question: str class IngestRequest(BaseModel): document_dir: str @app.on_event("startup") async def startup_event(): """启动时尝试加载已有的向量数据库""" try: rag_system.load_existing_vectorstore() rag_system.build_qa_chain() print("向量数据库加载成功，QA 链已就绪。") except: print("未找到已有向量数据库，请先调用 /ingest 接口导入文档。") @app.post("/ingest") async def ingest_documents(request: IngestRequest): """导入文档并构建向量数据库""" try: splits = rag_system.load_and_split_documents(request.document_dir) rag_system.create_vectorstore(splits) rag_system.build_qa_chain() return {"message": f"文档导入成功，共处理 {len(splits)} 个文本块。"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/query") async def query_documents(request: QueryRequest): """用户提问""" try: answer = rag_system.query(request.question) return {"question": request.question, "answer": answer} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): return {"status": "healthy"} if name == "main": uvicorn.run(app, host="0.0.0.0", port=8000) 后端服务运行截图如下（启动后访问 http://localhost:8000/docs 查看 API 文档）： 4. 前端界面开发 (Streamlit) Streamlit 可以快速构建交互式 Web 应用，非常适合作为本系统的前端。 4.1 前端代码 (frontend/app.py) import streamlit as st import requests import json 后端 API 地址 (根据实际部署修改) BACKEND_URL = "http://localhost:8000" st.set_page_config(page_title="企业内网知识问答系统", layout="wide") st.title("🧠 企业内网 RAG 知识问答系统") st.markdown("基于 LangChain + FastAPI + Ollama(Qwen) 构建的完全离线问答系统。") 侧边栏 - 文档管理 with st.sidebar: st.header("📁 文档管理") doc_dir = st.text_input("文档目录路径", value="./documents") if st.button("导入文档并构建知识库"): with st.spinner("正在处理文档并构建向量数据库..."): try: resp = requests.post( f"{BACKEND_URL}/ingest", json={"document_dir": doc_dir} ) if resp.status_code == 200: st.success(resp.json()["message"]) else: st.error(f"导入失败: {resp.text}") except Exception as e: st.error(f"连接后端失败: {e}") st.divider() st.header("⚙️ 系统状态") try: health_resp = requests.get(f"{BACKEND_URL}/health") if health_resp.status_code == 200: st.success("后端服务运行正常") else: st.warning("后端服务异常") except: st.error("无法连接到后端服务") 主界面 - 问答 st.header("💬 知识问答") question = st.text_area("请输入您的问题：", height=100, placeholder="例如：公司今年的战略重点是什么？") if st.button("提交问题", type="primary"): if not question.strip(): st.warning("请输入问题") else: with st.spinner("正在检索并生成答案..."): try: resp = requests.post( f"{BACKEND_URL}/query", json={"question": question} ) if resp.status_code == 200: result = resp.json() st.subheader("❓ 问题") st.info(result["question"]) st.subheader("🤖 答案") st.success(result["answer"]) else: st.error(f"请求失败: {resp.text}") except Exception as e: st.error(f"连接后端失败: {e}") 历史记录 (简化示例) st.divider() st.header("📜 最近问答") if "history" not in st.session_state: st.session_state.history = [] if st.session_state.history: for q, a in st.session_state.history[-5:]: with st.expander(f"Q: {q}"): st.write(a) else: st.caption("暂无历史记录") 4.2 启动前端 在 frontend 目录下运行 streamlit run app.py 前端运行截图如下： 5. 系统部署与运行 启动 Ollama 服务：确保 Ollama 服务在后台运行，并已加载 Qwen 模型。 启动后端服务：在 backend 目录下运行 python main.py。 启动前端服务：在 frontend 目录下运行 streamlit run app.py。 访问系统：在浏览器中打开 Streamlit 提供的本地地址（通常是 http://localhost:8501）。 导入文档：在侧边栏输入文档目录路径，点击“导入文档并构建知识库”。 开始问答：在主界面输入问题，点击“提交问题”获取答案。 6. 优化与扩展建议 性能优化：对于大量文档，可考虑使用更高效的向量数据库（如 Milvus）或对文档进行预处理筛选。 多模型支持：Ollama 支持众多开源模型，可轻松切换为 Llama、Mistral 等。 前端美化：Streamlit 支持自定义主题和组件，可进一步优化 UI 体验。 权限管理：可根据企业需求，在后端增加 API 密钥认证或用户登录功能。 日志与监控：添加详细的日志记录和系统健康监控。 7. 总结 本文完整演示了如何利用 Python 开源技术栈，构建一个完全离线、安全可控的企业内网 RAG 知识问答系统。该系统将企业内部文档转化为可查询的知识库，通过本地大模型生成准确、可靠的答案，有效保护了企业数据隐私。读者可根据自身业务需求，对代码进行修改和扩展，快速落地适用于特定场景的知识管理解决方案。最后再附上实际使用的效果截图：
特别说明，本项目可完全在内网中使用，解决了用户数据安全以及离线AI的功能。
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
