前言

RAG（Retrieval-Augmented Generation）是一种结合了信息检索和生成式AI的技术架构，能够让大语言模型基于外部知识库进行问答，有效解决模型知识截止日期和幻觉问题。

本文将详细介绍如何用LangChain+Milvus+Ollama+Streamlit搭建一个完全本地化的RAG系统。

技术架构

  • LangChain: 负责RAG管道的编排和文档处理
  • Milvus: 高性能向量数据库，负责文档向量的存储和检索
  • Ollama: 本地大语言模型运行框架
  • Streamlit: 提供友好的Web前端交互界面

环境准备

1. 安装依赖

# 创建虚拟环境
python -m venv rag_env
source rag_env/bin/activate  # Linux/Mac
# rag_env\Scripts\activate  # Windows

# 安装核心依赖
pip install langchain langchain-ollama pymilvus sentence-transformers streamlit streamlit-chat
pip install pypdf2 tiktoken chromadb plotly pandas  # 前端界面和数据分析

2. 启动Milvus(Standalone版)

# 启动Milvus（推荐使用Docker）
docker run -d \
  --name milvus-standalone \
  --security-opt seccomp:unconfined \
  -p 19530:19530 \
  -p 9091:9091 \
  -v milvus-data:/milvus/data \
  milvusdb/milvus:latest

# 下载并启动Ollama模型
ollama serve
ollama pull llama2  # 或其他模型

核心代码实现

1. 文档处理模块

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import Milvus

class DocumentProcessor:
    def __init__(self):
        # 初始化嵌入模型（Ollama提供）
        self.embeddings = OllamaEmbeddings(model="llama2")
        # 文档分割器
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len,
        )
        
    def load_and_split_documents(self, file_paths):
        """加载并分割文档"""
        documents = []
        for file_path in file_paths:
            if file_path.endswith('.pdf'):
                loader = PyPDFLoader(file_path)
                docs = loader.load()
                documents.extend(docs)
            else:
                # 处理其他格式文档
                pass
        
        # 分割文档
        splits = self.text_splitter.split_documents(documents)
        return splits
    
    def create_vectorstore(self, documents, collection_name="rag_docs"):
        """创建向量数据库"""
        # 清空现有集合
        vectorstore = Milvus.from_documents(
            documents=documents,
            embedding=self.embeddings,
            collection_name=collection_name,
            drop_old=True  # 删除旧集合
        )
        return vectorstore

2. RAG查询引擎

from langchain_ollama import ChatOllama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class RAGQueryEngine:
    def __init__(self, vectorstore):
        # 初始化本地LLM
        self.llm = ChatOllama(
            model="llama2",
            temperature=0.1,
            max_tokens=2000
        )
        
        # 创建检索器
        self.retriever = vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 4}  # 返回4个最相关的文档片段
        )
        
        # 创建自定义提示词
        self.prompt = PromptTemplate(
            template="""基于以下上下文信息回答问题：
            
            上下文：
            {context}
            
            问题：{question}
            
            回答要求：
            1. 基于上下文信息回答
            2. 如果上下文中没有相关信息，说明无法回答
            3. 保持回答简洁明确
            
            回答：""",
            input_variables=["context", "question"]
        )
        
        # 创建QA链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            chain_type_kwargs={"prompt": self.prompt},
            return_source_documents=True  # 返回源文档
        )
    
    def query(self, question):
        """执行查询"""
        result = self.qa_chain({"query": question})
        return {
            "answer": result["result"],
            "source_documents": result["source_documents"]
        }

3. 系统集成模块

class LocalRAGSystem:
    def __init__(self):
        self.doc_processor = DocumentProcessor()
        self.vectorstore = None
        self.query_engine = None
        
    def build_system(self, document_paths):
        """构建RAG系统"""
        print("📚 正在处理文档...")
        documents = self.doc_processor.load_and_split_documents(document_paths)
        print(f"✅ 文档处理完成，共{len(documents)}个片段")
        
        print("🗄️  正在构建向量数据库...")
        self.vectorstore = self.doc_processor.create_vectorstore(documents)
        print("✅ 向量数据库构建完成")
        
        print("🤖 正在初始化查询引擎...")
        self.query_engine = RAGQueryEngine(self.vectorstore)
        print("✅ 查询引擎初始化完成")
        
    def chat(self, question):
        """执行对话"""
        if not self.query_engine:
            return "系统未初始化，请先调用build_system()"
        
        print(f"❓ 问题：{question}")
        result = self.query_engine.query(question)
        print(f"💬 回答：{result['answer']}")
        
        # 显示引用源
        print("\n📖 参考文档片段：")
        for i, doc in enumerate(result['source_documents'][:2], 1):
            print(f"{i}. {doc.page_content[:100]}...")
        
        return result

完整使用示例

def main():
    # 初始化RAG系统
    rag_system = LocalRAGSystem()
    
    # 构建系统（提供文档路径）
    document_paths = ["./documents/技术文档.pdf", "./documents/FAQ.md"]
    rag_system.build_system(document_paths)
    
    # 执行查询
    questions = [
        "什么是LangChain框架？",
        "如何使用Milvus进行向量检索？",
        "RAG系统的优势是什么？"
    ]
    
    for question in questions:
        print("\n" + "="*50)
        rag_system.chat(question)
        print("="*50)

if __name__ == "__main__":
    main()

高级功能扩展

1. 多文档类型支持

from langchain_community.document_loaders import (
    TextLoader, CSVLoader, ExcelLoader, UnstructuredFileLoader
)

class AdvancedDocumentProcessor(DocumentProcessor):
    def load_and_split_documents(self, file_paths):
        """支持多种文档格式"""
        documents = []
        
        for file_path in file_paths:
            if file_path.endswith('.pdf'):
                loader = PyPDFLoader(file_path)
            elif file_path.endswith('.txt'):
                loader = TextLoader(file_path, encoding='utf-8')
            elif file_path.endswith('.csv'):
                loader = CSVLoader(file_path)
            elif file_path.endswith(('.xlsx', '.xls')):
                loader = ExcelLoader(file_path)
            else:
                # 使用通用文档处理器
                loader = UnstructuredFileLoader(file_path)
            
            docs = loader.load()
            documents.extend(docs)
        
        return self.text_splitter.split_documents(documents)

2. 查询结果过滤

class EnhancedRAGQueryEngine(RAGQueryEngine):
    def __init__(self, vectorstore):
        super().__init__(vectorstore)
        self.confidence_threshold = 0.7
    
    def query_with_confidence(self, question):
        """带置信度评估的查询"""
        result = self.query_engine.query(question)
        
        # 简单的置信度评估（基于相似度分数）
        similarities = [doc.metadata.get('score', 0) 
                       for doc in result['source_documents']]
        avg_similarity = sum(similarities) / len(similarities) if similarities else 0
        
        if avg_similarity < self.confidence_threshold:
            result['answer'] += f"\n\n⚠️  注意：检索置信度较低({avg_similarity:.2f})，回答仅供参考。"
        
        result['confidence'] = avg_similarity
        return result

Streamlit前端界面设计

1. 主界面应用

import streamlit as st
import tempfile
import os
from streamlit_chat import message
import json

class StreamlitRAGInterface:
    def __init__(self):
        self.initialize_session_state()
    
    def initialize_session_state(self):
        """初始化Streamlit会话状态"""
        if 'rag_system' not in st.session_state:
            st.session_state.rag_system = LocalRAGSystem()
        if 'chat_history' not in st.session_state:
            st.session_state.chat_history = []
        if 'system_built' not in st.session_state:
            st.session_state.system_built = False
        if 'uploaded_files' not in st.session_state:
            st.session_state.uploaded_files = []
    
    def render_sidebar(self):
        """渲染侧边栏"""
        with st.sidebar:
            st.title("🔧 RAG系统配置")
            
            # 模型配置
            st.subheader("🤖 模型配置")
            model_options = ["llama2", "llama3", "codellama", "mistral"]
            selected_model = st.selectbox("选择模型", model_options, index=0)
            
            temperature = st.slider("生成温度", 0.0, 1.0, 0.1)
            max_tokens = st.slider("最大令牌数", 500, 4000, 2000)
            
            # 文档上传
            st.subheader("📚 文档管理")
            uploaded_files = st.file_uploader(
                "上传文档", 
                type=['pdf', 'txt', 'md', 'csv', 'xlsx'],
                accept_multiple_files=True
            )
            
            if uploaded_files:
                st.session_state.uploaded_files = uploaded_files
            
            # 系统构建
            if st.button("🚀 构建RAG系统", type="primary"):
                self.build_system_from_uploads()
            
            # 系统状态
            st.subheader("📊 系统状态")
            if st.session_state.system_built:
                st.success("✅ 系统已就绪")
                if st.button("🔄 重置系统"):
                    self.reset_system()
            else:
                st.warning("⏳ 等待系统构建")
                
            # 参数配置
            st.subheader("⚙️ 检索参数")
            k_value = st.slider("检索文档数量", 1, 10, 4)
            
            return {
                'model': selected_model,
                'temperature': temperature,
                'max_tokens': max_tokens,
                'k_value': k_value
            }
    
    def build_system_from_uploads(self):
        """从上传的文件构建系统"""
        if not st.session_state.uploaded_files:
            st.error("请先上传文档")
            return
        
        try:
            with st.spinner("正在处理文档..."):
                # 保存临时文件
                temp_files = []
                for uploaded_file in st.session_state.uploaded_files:
                    temp_dir = tempfile.mkdtemp()
                    temp_path = os.path.join(temp_dir, uploaded_file.name)
                    
                    with open(temp_path, "wb") as f:
                        f.write(uploaded_file.getvalue())
                    temp_files.append(temp_path)
                
                # 构建系统
                st.session_state.rag_system.build_system(temp_files)
                st.session_state.system_built = True
                st.session_state.chat_history = []
                
                st.success(f"✅ 成功处理 {len(temp_files)} 个文档")
                
        except Exception as e:
            st.error(f"构建失败: {str(e)}")
    
    def reset_system(self):
        """重置系统"""
        st.session_state.rag_system = LocalRAGSystem()
        st.session_state.chat_history = []
        st.session_state.system_built = False
        st.session_state.uploaded_files = []
        st.rerun()
    
    def render_chat_interface(self, config):
        """渲染聊天界面"""
        st.title("💬 本地RAG智能问答系统")
        
        if not st.session_state.system_built:
            st.warning("请先在侧边栏构建RAG系统")
            return
        
        # 聊天历史显示
        chat_container = st.container()
        
        with chat_container:
            for i, (user_msg, bot_msg) in enumerate(st.session_state.chat_history):
                message(user_msg, is_user=True, key=f"user_{i}")
                message(bot_msg, is_user=False, key=f"bot_{i}")
        
        # 用户输入
        if query := st.chat_input("请输入您的问题..."):
            # 用户消息
            with st.spinner("正在思考..."):
                try:
                    # 配置模型参数
                    st.session_state.rag_system.query_engine.llm.temperature = config['temperature']
                    st.session_state.rag_system.query_engine.llm.max_tokens = config['max_tokens']
                    
                    # 执行查询
                    result = st.session_state.rag_system.chat(query)
                    
                    # 更新聊天历史
                    st.session_state.chat_history.append((query, result['answer']))
                    
                    # 刷新页面显示新消息
                    st.rerun()
                    
                except Exception as e:
                    st.error(f"查询失败: {str(e)}")

def main():
    """主函数"""
    st.set_page_config(
        page_title="RAG智能问答系统",
        page_icon="🤖",
        layout="wide"
    )
    
    # 添加自定义CSS
    st.markdown("""
    .main-header {
        text-align: center;
        padding: 1rem;
        background: linear-gradient(90deg, #667eea 0%, #764ba2 100%);
        border-radius: 10px;
        margin-bottom: 2rem;
    }

    .chat-message {
        padding: 1rem;
        border-radius: 10px;
        margin: 0.5rem 0;
    }

    .user-message {
        background-color: #e3f2fd;
        border-left: 5px solid #2196f3;
    }

    .bot-message {
        background-color: #f3e5f5;
        border-left: 5px solid #9c27b0;
    }

    """, unsafe_allow_html=True)

    # 初始化界面
    app = StreamlitRAGInterface()
    config = app.render_sidebar()
    app.render_chat_interface(config)

if __name__ == "__main__":
    main()

2. 配置文件

# config.py - 配置文件
import os
from typing import Dict, Any

class Config:
    # 模型配置
    DEFAULT_MODEL = "llama2"
    AVAILABLE_MODELS = ["llama2", "llama3", "codellama", "mistral"]
    
    # 向量数据库配置
    MILVUS_HOST = os.getenv("MILVUS_HOST", "localhost")
    MILVUS_PORT = os.getenv("MILVUS_PORT", "19530")
    COLLECTION_NAME = "rag_docs"
    
    # 文档处理配置
    CHUNK_SIZE = 1000
    CHUNK_OVERLAP = 200
    SUPPORTED_FILE_TYPES = ['pdf', 'txt', 'md', 'csv', 'xlsx']
    
    # Streamlit配置
    PAGE_TITLE = "RAG智能问答系统"
    PAGE_ICON = "🤖"
    LAYOUT = "wide"
    
    # UI配置
    MAX_UPLOAD_SIZE = 50 * 1024 * 1024  # 50MB
    CHAT_HISTORY_LIMIT = 100
    
    # 检索配置
    DEFAULT_TOP_K = 4
    CONFIDENCE_THRESHOLD = 0.7

部署建议

1. Docker化部署

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8501

# 启动命令
CMD ["streamlit", "run", "streamlit_app.py", "--server.port=8501", "--server.address=0.0.0.0"]

2. Docker Compose配置

# docker-compose.yml
version: '3.8'

services:
  milvus:
    image: milvusdb/milvus:latest
    container_name: milvus-standalone
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    volumes:
      - milvus_data:/milvus/data
    ports:
      - "19530:19530"
    networks:
      - rag_network

  streamlit-app:
    build: .
    container_name: streamlit-rag
    ports:
      - "8501:8501"
    environment:
      - MILVUS_HOST=milvus
      - MILVUS_PORT=19530
    depends_on:
      - milvus
    networks:
      - rag_network

volumes:
  milvus_data:

networks:
  rag_network:
    driver: bridge

3. 独立部署（开发环境）

# 1. 启动Milvus
docker run -d --name milvus-standalone \
  --security-opt seccomp:unconfined \
  -p 19530:19530 \
  -p 9091:9091 \
  -v milvus-data:/milvus/data \
  milvusdb/milvus:latest

# 2. 启动Ollama
ollama serve
ollama pull llama2

# 3. 启动Streamlit应用
streamlit run streamlit_app.py --server.port 8501 --server.address 0.0.0.0

# 4. 访问 http://localhost:8501

4. 一键启动脚本

#!/bin/bash
# setup.sh - 一键安装脚本

echo "🚀 开始搭建RAG系统..."

# 1. 创建虚拟环境
echo "📦 创建Python虚拟环境..."
python -m venv rag_env
source rag_env/bin/activate

# 2. 安装依赖
echo "📚 安装Python依赖..."
pip install --upgrade pip
pip install langchain langchain-ollama pymilvus sentence-transformers streamlit streamlit-chat
pip install pypdf2 tiktoken chromadb plotly pandas

# 3. 启动Milvus
echo "🗄️ 启动Milvus向量数据库..."
docker run -d \
  --name milvus-standalone \
  --security-opt seccomp:unconfined \
  -p 19530:19530 \
  -p 9091:9091 \
  -v milvus-data:/milvus/data \
  milvusdb/milvus:latest

# 4. 等待服务就绪
echo "⏳ 等待服务启动..."
sleep 30

# 5. 检查Ollama
echo "🤖 检查Ollama服务..."
if ! command -v ollama &> /dev/null; then
    echo "❌ 请先安装Ollama: https://ollama.ai"
    exit 1
fi

# 6. 启动模型
echo "📥 下载模型..."
ollama pull llama2

# 7. 启动应用
echo "🌐 启动Streamlit应用..."
echo "🎉 RAG系统已启动！请访问: http://localhost:8501"
streamlit run streamlit_app.py --server.port 8501 --server.address 0.0.0.0

快速测试

创建测试文档进行验证：

# test_document.txt

什么是RAG系统？
RAG（Retrieval-Augmented Generation）是一种结合了信息检索和生成式AI的技术架构。

LangChain有哪些主要功能？
LangChain是一个用于开发基于LLM的应用程序的框架，提供了丰富的组件和工具。

Milvus是什么数据库？
Milvus是一个开源的向量数据库，专门用于存储和检索大规模向量数据。

总结

通过LangChain+Milvus+Ollama+Streamlit的完整组合，我们成功搭建了一个功能完善的本地化RAG系统，具有以下特点：

🔧 核心技术栈

  • ✅ LangChain: RAG管道编排和文档处理
  • ✅ Milvus: 高性能向量数据库存储
  • ✅ Ollama: 完全本地化的大语言模型
  • ✅ Streamlit: 友好的Web前端交互界面

🌟 系统特色

  • ✅ 数据隐私保护: 完全本地化，数据不上传云端
  • ✅ 灵活文档处理: 支持PDF、TXT、CSV等多种格式
  • ✅ 可视化交互: 美观的Web界面，支持聊天式问答
  • ✅ 实时系统监控: 聊天历史、统计分析等功能
  • ✅ 一键部署: Docker支持，快速部署上线
  • ✅ 高可扩展性: 模块化设计，便于二次开发

🎯 应用场景

  • 企业内部知识库: 安全访问公司文档和FAQ
  • 客户服务系统: 基于企业资料的智能客服
  • 教育培训平台: 个性化学习助手
  • 科研文献助手: 快速检索和分析学术论文

📈 性能优化建议

1.模型选择: 根据硬件资源选择合适的本地模型

2.索引优化: 调整Milvus索引参数提升检索速度

3.缓存策略: 对热门查询结果进行缓存

4.批量处理: 文档向量化时使用批处理提升效率

5.负载均衡: 生产环境考虑多实例部署

    这种架构特别适合需要数据安全控制、高度定制化和良好用户体验的企业环境。

    # milvus # RAG
    Logo
    AI Agent技术社区

    Agent 垂直技术社区，欢迎活跃、内容共建。

    更多推荐

    cover

    从一张商品图到正式上架：实测店匠 Shoplazza Athena，AI Agent 不只是聊天框

    avatar AI Agent技术社区
    cover

    我测试了一个0.9B模型，它竟然能自动区分多人讲话）

    avatar AI Agent技术社区
    cover

    【AI Agent】《2026年9月 AI Agent全栈开发技术选型专项面试宝典》

    avatar AI Agent技术社区