摘要

本文详细介绍如何利用 Python 生态中的 LangChain/LlamaIndex 框架，结合 FastAPI 构建后端服务，使用 Streamlit 开发前端界面，并集成本地部署的 Ollama 与 Qwen 大语言模型，打造一套完全离线、可运行于企业内网环境的 RAG（检索增强生成）知识问答系统。文章将从系统架构设计、环境搭建、核心代码实现、前后端集成、模型本地化部署到最终系统展示，提供完整的实战指南。

1. 系统架构与设计目标

在企业内网环境中，数据安全与隐私保护是首要考虑。本系统设计目标如下：

• 完全离线：所有组件（大模型、向量数据库、前后端服务）均部署于内网，无需连接外网。
• 开源技术栈：采用成熟的 Python 开源框架，降低技术门槛与授权成本。
• 模块化设计：前后端分离，便于独立开发、测试与部署。
• 易于扩展：支持更换不同的本地大模型、向量数据库及前端框架。

整体架构如下图所示：

架构分为四层：

1. 数据层：本地文档（PDF、Word、TXT等）经过文本提取与分割后，存入本地向量数据库（如 Chroma、FAISS）。
2. 模型层：本地部署的 Ollama 服务，加载 Qwen 等开源大模型。
3. 后端服务层：基于 FastAPI 构建，集成 LangChain/LlamaIndex 实现 RAG 检索与生成流程。
4. 前端交互层：使用 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:11434

3. 后端服务开发 (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的功能。