问题：每次加载模型都要 30 秒后才能开始查询
解决：改为 Web 服务常驻内存，无需再加载模型
方法：一个关键改造，让你的 RAG 从"一次性工具"变成"企业级服务"

---

如果你正在部署 RAG（检索增强生成）系统，很可能遇到过这个尴尬的场景：

第一次查询：

加载模型...（30 秒）

第二次查询：

加载模型...（30 秒）← 又要重新加载！

我就查个问题，为什么要等这么久？！

模型明明之前加载过了，为什么每次都要重新加载？

我在部署一个股票知识库 RAG 系统时，就遇到了这个问题。经过一天的改造，把 RAG 从"命令行脚本"改造成"Web 服务"，模型加载从30秒降为0秒。

今天把完整的改造过程和经验分享出来，帮你避开我踩过的坑。

---

问题诊断：为什么每次都重新加载模型？

初始架构（命令行脚本）

用户查询 → 运行 rag_search.py → 加载模型 → 检索 → 返回答案
                        ↓
                    每次都要重新加载！

问题根源：

• 脚本执行完就退出，模型从内存中卸载
• 下次查询重新运行脚本，重新加载模型

---

为什么模型加载这么慢？

原因 1：模型文件大

Embedding 模型（moka-ai/m3e-base）：

• 文件大小：~500MB
• 加载时间：~15 秒
• 用途：文本向量化

Rerank 模型（BAAI/bge-reranker-base）：

• 文件大小：~300MB
• 加载时间：~15 秒
• 用途：重排序检索结果

总计：800MB，30 秒

---

原因 2：每次从磁盘读取

命令行脚本的执行流程：

1. 运行 rag_search.py
2. import rag_core.py
3. 初始化 RAGSystem
4. 从磁盘加载模型文件（30 秒）← 这里最慢
5. 执行检索
6. 脚本退出，模型从内存卸载
7. 下次查询，重复步骤 1-6

问题：每次都从磁盘读取 800MB 模型文件！

---

原因 3：没有进程常驻

命令行脚本：

• 简单，适合测试
• 执行完就退出
• 无法保持模型在内存中

Web 服务：

• 进程常驻，模型只加载一次
• 后续查询直接使用内存中的模型
• 需要改造架构

---

关键改造：从命令行脚本到 Web 服务

改造思路

之前（命令行脚本）：

用户查询 → 运行脚本 → 加载模型 → 检索 → 返回 → 退出
                    ↓
                每次都加载

改造后（Web 服务）：

启动服务 → 加载模型（一次） → 常驻内存
              ↓
用户查询 → HTTP 请求 → 直接使用模型 → 返回
                    ↓
                无需重新加载

---

改造步骤

步骤 1：创建 HTTP 服务器

新建 rag_server.py：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
RAG Web 服务 - HTTP 服务器
模型只加载一次，后续查询秒级响应
"""
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
from rag_core import RAGSystem

# ========== 全局变量：模型常驻内存 ==========
print("启动 RAG 服务...")
start_time = time.time()

# 加载 RAG 系统（只加载一次！）
rag = RAGSystem('config.json')

load_time = time.time() - start_time
print(f"\n 模型加载完成！耗时：{load_time:.1f}秒")
print(f" 知识库文档数：{rag.collection.count()}")
print("\n" + "="*60)
print(" 服务地址：http://127.0.0.1:8000")
print(" 提示：按 Ctrl+C 停止服务")
print("="*60 + "\n")

# ========== HTTP 请求处理器 ==========
class RAGHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        """处理 POST 请求（检索）"""
        start = time.time()
        
        # 解析请求
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length).decode('utf-8')
        data = json.loads(body)
        
        query = data.get('query', '')
        top_k = data.get('top_k', 3)
        use_llm = data.get('use_llm', False)
        
        # 检索（模型已在内存中，直接使用）
        result = rag.search(query, top_k=top_k)
        
        # 构建响应
        response = {
            'query': query,
            'top_docs': result['top_docs'],
            'top_sources': result['top_sources'],
            'elapsed': f"{time.time() - start:.2f}秒"
        }
        
        # 返回 JSON
        self.send_response(200)
        self.send_header('Content-type', 'application/json; charset=utf-8')
        self.end_headers()
        self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8'))
    
    def log_message(self, format, *args):
        """禁用默认日志"""
        pass

# ========== 启动服务 ==========
if __name__ == '__main__':
    server = HTTPServer(('127.0.0.1', 8000), RAGHandler)
    print(" 服务启动成功！等待请求...\n")
    server.serve_forever()

关键点：

• rag = RAGSystem('config.json') 在全局作用域
• 只执行一次，服务启动时加载
• 后续请求直接使用 rag 对象

---

步骤 2：创建 HTTP 客户端

新建 rag_client.py：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
RAG 客户端 - 命令行工具
通过 HTTP 请求后台服务，秒级响应
"""
import sys
import json
import requests

SERVER_URL = 'http://127.0.0.1:8000'

def search(query, top_k=3, use_llm=False):
    """发送检索请求"""
    payload = {
        'query': query,
        'top_k': top_k,
        'use_llm': use_llm
    }
    
    response = requests.post(SERVER_URL, json=payload, timeout=60)
    response.raise_for_status()
    return response.json()

def main():
    # 解析参数
    args = sys.argv[1:]
    query = None
    use_llm = False
    
    for arg in args:
        if arg == '--llm':
            use_llm = True
        elif not arg.startswith('-'):
            query = arg
    
    if not query:
        print("="*60)
        print("RAG 客户端 - 后台服务模式")
        print("="*60)
        print("\n用法：python rag_client.py \"你的问题\"")
        print("\n示例：")
        print('  python rag_client.py "交保险时间"')
        print('  python rag_client.py "散户特点" --llm')
        return
    
    # 检查服务
    try:
        response = requests.get(SERVER_URL, timeout=2)
        if response.status_code == 200:
            data = response.json()
            print(f" RAG 服务运行中（知识库：{data.get('docs_count', 0)} 个文档）\n")
        else:
            print(" 无法连接到 RAG 服务")
            print("\n请先启动后台服务：")
            print("  python rag_server.py")
            return
    except requests.exceptions.ConnectionError:
        print(" 无法连接到 RAG 服务")
        print("\n请先启动后台服务：")
        print("  python rag_server.py")
        return
    
    # 发送请求
    result = search(query, use_llm=use_llm)
    
    # 显示结果
    print("\n" + "="*60)
    print(f"问题：{result['query']}")
    print("="*60)
    print(f"\n  检索耗时：{result['elapsed']}")
    print("\n检索到的相关资料：\n")
    for i, (doc, source) in enumerate(zip(result['top_docs'], result['top_sources']), 1):
        print(f"【资料{i}】来源：{source}")
        print("-"*60)
        content = doc[:500] + "..." if len(doc) > 500 else doc
        print(content)
        print()
    print("="*60)

if __name__ == '__main__':
    main()

---

步骤 3：创建启动脚本

新建 start_rag_service.bat（Windows）：

@echo off
chcp 65001 >nul
cd /d "%~dp0"

echo ============================================================
echo RAG Backend Service Startup
echo ============================================================
echo.
echo Starting RAG backend service...
echo.

REM 启动服务（新窗口）
start "RAG Backend Service" python rag_server.py

echo Service started in new window!
echo.
echo Usage:
echo   python rag_client.py "your question"
echo.
echo Stop service: Close "RAG Backend Service" window
echo.
pause

新建 start_rag_service.sh（Linux/Mac）：

#!/bin/bash
cd "$(dirname "$0")"

echo "============================================================"
echo "RAG Backend Service Startup"
echo "============================================================"
echo ""
echo "Starting RAG backend service..."
echo ""

# 后台启动
nohup python rag_server.py > rag_server.log 2>&1 &
PID=$!

echo "Service started! PID: $PID"
echo ""
echo "Usage:"
echo "  python rag_client.py \"your question\""
echo ""
echo "Stop service:"
echo "  kill $PID"
echo "  或：pkill -f rag_server.py"
echo ""

---

架构对比

改造前（命令行脚本）

┌─────────────────────────────────────┐
│  用户                                │
│  ↓ python rag_search.py "问题"       │
│  ↓                                   │
│  1. 启动 Python 进程                  │
│  2. 加载 Embedding 模型（15 秒）       │
│  3. 加载 Rerank 模型（15 秒）          │
│  4. 检索                             │
│  5. 大模型生成                       │
│  6. 显示结果                         │
│  7. 进程退出，模型卸载                │
└─────────────────────────────────────┘
     ↓
下次查询，重复以上步骤

---

改造后（Web 服务）

┌─────────────────────────────────────┐
│  启动服务                            │
│  ↓ python rag_server.py              │
│  ↓                                   │
│  1. 启动 Python 进程                  │
│  2. 加载 Embedding 模型（15 秒）       │
│  3. 加载 Rerank 模型（15 秒）          │
│  4. 启动 HTTP 服务器                  │
│  5. 等待请求 ← 服务常驻               │
└─────────────────────────────────────┘
     ↓
┌─────────────────────────────────────┐
│  用户查询                            │
│  ↓ python rag_client.py "问题"       │
│  ↓                                   │
│  1. HTTP POST 请求                   │
│  2. 直接使用内存中的模型              │
│  3. 检索                             │
│  4. 大模型生成                        │
│  5. 返回 JSON 结果                    │
│  6. 显示结果                         │
└─────────────────────────────────────┘
     ↓
下次查询，直接使用内存中的模型

首次查询：加载模型30 秒
后续查询：无需加载

---

进阶优化

优化 1：使用 NSSM 部署为 Windows 服务

好处：

• 开机自启
• 后台运行，不占用窗口
• 崩溃自动重启

步骤：

1. 下载 NSSM：https://nssm.cc/download

2. 安装服务：

nssm install RAGServer "C:\Python314\python.exe"
nssm set RAGServer AppDirectory "C:\path\tools\rag"
nssm set RAGServer AppParameters "rag_server.py"
nssm set RAGServer DisplayName "RAG Backend Service"
nssm set RAGServer Start SERVICE_AUTO_START

3. 启动服务：

nssm start RAGServer

---

优化 2：使用 systemd 部署为 Linux 服务

创建 /etc/systemd/system/rag-server.service：

[Unit]
Description=RAG Backend Service
After=network.target

[Service]
Type=simple
User=your_user
WorkingDirectory=/path/to/rag
ExecStart=/usr/bin/python3 rag_server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

启动服务：

sudo systemctl daemon-reload
sudo systemctl start rag-server
sudo systemctl enable rag-server  # 开机自启

---

优化 3：添加健康检查接口

修改 rag_server.py：

class RAGHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        """处理 GET 请求（健康检查）"""
        self.send_response(200)
        self.send_header('Content-type', 'application/json; charset=utf-8')
        self.end_headers()
        response = {
            'status': 'ok',
            'message': 'RAG 服务运行中',
            'docs_count': rag.collection.count()
        }
        self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8'))

检查服务状态：

curl http://127.0.0.1:8000

---

优化 4：添加日志记录

修改 rag_server.py：

import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('rag_server.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class RAGHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        start = time.time()
        logger.info(f"Received query: {query}")
        
        # ... 处理请求 ...
        
        elapsed = time.time() - start
        logger.info(f"Completed in {elapsed:.2f}s")

---

关键经验

1. 模型常驻内存是必须的

• 命令行脚本：适合测试，不适合生产
• Web 服务：模型只加载一次，后续查询秒级响应
• 结论：生产环境必须用 Web 服务

2. 进程管理很重要

• 手动启动：适合开发
• 系统服务：适合生产（NSSM/systemd）
• 容器部署：适合大规模（Docker/K8s）

3. 监控和日志不可少

• 日志：记录每次查询的耗时、错误
• 监控：服务状态、内存使用、请求量
• 告警：服务崩溃、响应超时

4. 并发处理要考虑

• 单线程：适合测试（HTTPServer）
• 多线程：适合生产（ThreadingHTTPServer）
• 多进程：适合高并发（Gunicorn/uWSGI）

5. 安全配置不能忘

• 防火墙：只允许内网访问
• 认证：API Key 验证
• 限流：防止滥用

---

下一步优化方向

1. 负载均衡

场景：单服务无法支撑高并发

方案：

• Nginx 反向代理
• 多实例部署
• 健康检查 + 自动故障转移

2. 缓存层

场景：相同查询频繁出现

方案：

• Redis 缓存查询结果
• TTL 过期策略
• 缓存命中率监控

3. 异步处理

场景：大模型生成耗时较长

方案：

• 消息队列（RabbitMQ/Kafka）
• 异步任务（Celery）
• WebSocket 推送结果

---

---

参考资料

• Python HTTPServer 官方文档
• NSSM - Non-Sucking Service Manager
• systemd 官方文档
• LangChain Serve 部署指南

---

如果你的 RAG 系统也慢，试试改造成 Web 服务，让模型常驻内存！

---

作者：阿财
发布时间：2026-03-18