DeepSeek-R1-Distill-Qwen-7B部署案例：Ollama + FastAPI 构建私有LLM服务

想不想在自己的服务器上搭建一个专属的AI助手？不用依赖任何外部API，完全私有化部署，数据安全自己掌控。今天我就带你用Ollama和FastAPI，把DeepSeek-R1-Distill-Qwen-7B这个强大的推理模型变成你自己的私有LLM服务。

你可能听说过DeepSeek-R1系列模型在推理任务上的出色表现，但直接使用大模型往往需要复杂的部署流程。别担心，通过Ollama这个工具，我们可以像安装普通软件一样轻松部署模型，再用FastAPI给它套上一个漂亮的Web接口。整个过程就像搭积木一样简单，跟着我做，30分钟内你就能拥有一个可以对外提供服务的AI接口。

1. 为什么选择这个技术方案？

1.1 模型优势：DeepSeek-R1-Distill-Qwen-7B到底强在哪？

DeepSeek-R1-Distill-Qwen-7B不是普通的语言模型，它是专门为推理任务优化的。想象一下，你问AI一个数学题，普通模型可能直接给你答案，但这个模型会像人一样一步步思考，把解题过程都展示出来。

这个模型有几个特别的地方：

• 推理能力强：在数学、代码、逻辑推理任务上表现突出
• 体积适中：7B参数规模，对硬件要求相对友好
• 开源免费：完全开源，不用担心使用限制或费用问题
• 蒸馏版本：从更大的R1模型蒸馏而来，保留了核心推理能力

我测试过，让它解一个初中数学题，它不仅给出正确答案，还会详细解释每一步的思路。这种能力在很多实际场景中特别有用，比如教育辅导、代码调试、数据分析等。

1.2 技术栈选择：Ollama + FastAPI的黄金组合

为什么选Ollama？因为它让模型部署变得异常简单。传统部署大模型需要处理各种依赖、配置环境变量、调整参数，而Ollama把这些都封装好了。你只需要一条命令就能拉取模型，再一条命令就能启动服务。

FastAPI则是Python里构建API最快、最现代的框架。它的自动文档生成、类型检查、异步支持这些特性，让我们能快速构建出高性能、易维护的API服务。

这个组合的好处很明显：

• 部署简单：Ollama负责模型管理，FastAPI负责接口封装
• 性能不错：异步处理支持，能同时服务多个请求
• 易于扩展：后续想加缓存、加监控、加认证都很方便
• 维护省心：两个都是成熟稳定的技术栈

2. 环境准备与快速部署

2.1 系统要求与安装准备

在开始之前，先确认你的环境是否符合要求。我建议使用Linux系统，Ubuntu 20.04或22.04都可以。Windows和macOS也能用，但Linux的兼容性最好。

硬件方面，DeepSeek-R1-Distill-Qwen-7B是7B参数的模型，建议至少：

• 16GB内存（模型本身需要约14GB）
• 支持CUDA的NVIDIA显卡（显存8GB以上）
• 50GB可用磁盘空间

如果你没有显卡，用CPU也能跑，就是速度会慢一些。我测试过，在RTX 3060（12GB显存）上，生成一段200字的回答大概需要3-5秒。

先更新系统并安装必要的工具：

# 更新系统包
sudo apt update && sudo apt upgrade -y

# 安装Python和相关工具
sudo apt install python3 python3-pip python3-venv curl -y

# 安装Docker（可选，用于容器化部署）
sudo apt install docker.io docker-compose -y

2.2 安装Ollama

Ollama的安装非常简单，一条命令搞定：

# 下载并安装Ollama
curl -fsSL https://ollama.com/install.sh | sh

安装完成后，启动Ollama服务：

# 启动Ollama服务
ollama serve &

你可以检查服务是否正常运行：

# 检查Ollama服务状态
curl http://localhost:11434/api/tags

如果看到返回JSON数据，说明Ollama服务已经启动成功了。

2.3 拉取DeepSeek-R1-Distill-Qwen-7B模型

现在来拉取我们需要的模型。Ollama支持很多开源模型，DeepSeek-R1-Distill-Qwen-7B也在其中：

# 拉取模型（这步需要一些时间，模型大小约4GB）
ollama pull deepseek-r1:7b

拉取过程中，你会看到下载进度。完成后，可以验证模型是否可用：

# 测试模型是否能正常工作
ollama run deepseek-r1:7b "你好，请介绍一下你自己"

如果模型开始生成回答，说明一切正常。按Ctrl+C退出交互模式。

3. 构建FastAPI服务

3.1 创建项目目录和虚拟环境

我们先创建一个专门的项目目录，这样管理起来更清晰：

# 创建项目目录
mkdir deepseek-api && cd deepseek-api

# 创建虚拟环境
python3 -m venv venv

# 激活虚拟环境
source venv/bin/activate

3.2 安装必要的Python包

在虚拟环境中安装FastAPI和相关依赖：

# 安装FastAPI和相关包
pip install fastapi uvicorn httpx python-dotenv

# 安装异步HTTP客户端
pip install aiohttp

3.3 编写FastAPI应用代码

创建一个main.py文件，这是我们的主应用文件：

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List
import httpx
import asyncio
import json
import os
from datetime import datetime

# 创建FastAPI应用实例
app = FastAPI(
    title="DeepSeek-R1 API服务",
    description="基于Ollama部署的DeepSeek-R1-Distill-Qwen-7B模型API接口",
    version="1.0.0"
)

# 添加CORS中间件，允许跨域请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应该限制具体的域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 定义请求和响应的数据模型
class ChatRequest(BaseModel):
    """聊天请求模型"""
    prompt: str
    max_tokens: Optional[int] = 512
    temperature: Optional[float] = 0.7
    top_p: Optional[float] = 0.9
    stream: Optional[bool] = False

class ChatResponse(BaseModel):
    """聊天响应模型"""
    response: str
    model: str
    created_at: str
    usage: dict

class HealthResponse(BaseModel):
    """健康检查响应模型"""
    status: str
    model: str
    ollama_status: str
    timestamp: str

# Ollama API配置
OLLAMA_BASE_URL = "http://localhost:11434"
MODEL_NAME = "deepseek-r1:7b"

async def check_ollama_health():
    """检查Ollama服务是否健康"""
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
            return response.status_code == 200
    except Exception:
        return False

@app.get("/")
async def root():
    """根路径，返回服务信息"""
    return {
        "service": "DeepSeek-R1 API",
        "version": "1.0.0",
        "model": MODEL_NAME,
        "endpoints": {
            "chat": "/api/v1/chat",
            "health": "/api/v1/health",
            "docs": "/docs"
        }
    }

@app.get("/api/v1/health", response_model=HealthResponse)
async def health_check():
    """健康检查接口"""
    ollama_healthy = await check_ollama_health()
    
    return HealthResponse(
        status="healthy" if ollama_healthy else "unhealthy",
        model=MODEL_NAME,
        ollama_status="running" if ollama_healthy else "stopped",
        timestamp=datetime.now().isoformat()
    )

@app.post("/api/v1/chat", response_model=ChatResponse)
async def chat_completion(request: ChatRequest):
    """
    聊天补全接口
    
    参数:
    - prompt: 用户输入的提示词
    - max_tokens: 最大生成token数（默认512）
    - temperature: 温度参数，控制随机性（默认0.7）
    - top_p: 核采样参数（默认0.9）
    - stream: 是否流式输出（默认False）
    """
    
    # 检查Ollama服务是否可用
    if not await check_ollama_health():
        raise HTTPException(
            status_code=503,
            detail="Ollama服务不可用，请检查Ollama是否正在运行"
        )
    
    # 构建Ollama API请求
    ollama_request = {
        "model": MODEL_NAME,
        "prompt": request.prompt,
        "stream": request.stream,
        "options": {
            "num_predict": request.max_tokens,
            "temperature": request.temperature,
            "top_p": request.top_p
        }
    }
    
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            if request.stream:
                # 流式响应处理
                response = await client.post(
                    f"{OLLAMA_BASE_URL}/api/generate",
                    json=ollama_request,
                    headers={"Content-Type": "application/json"}
                )
                
                if response.status_code != 200:
                    raise HTTPException(
                        status_code=response.status_code,
                        detail=f"Ollama API错误: {response.text}"
                    )
                
                # 处理流式响应
                full_response = ""
                async for line in response.aiter_lines():
                    if line.strip():
                        try:
                            data = json.loads(line)
                            if "response" in data:
                                full_response += data["response"]
                        except json.JSONDecodeError:
                            continue
                
                return ChatResponse(
                    response=full_response,
                    model=MODEL_NAME,
                    created_at=datetime.now().isoformat(),
                    usage={"tokens": len(full_response.split())}
                )
            else:
                # 非流式响应
                response = await client.post(
                    f"{OLLAMA_BASE_URL}/api/generate",
                    json=ollama_request,
                    headers={"Content-Type": "application/json"}
                )
                
                if response.status_code != 200:
                    raise HTTPException(
                        status_code=response.status_code,
                        detail=f"Ollama API错误: {response.text}"
                    )
                
                data = response.json()
                
                return ChatResponse(
                    response=data.get("response", ""),
                    model=MODEL_NAME,
                    created_at=datetime.now().isoformat(),
                    usage={
                        "total_tokens": data.get("total_duration", 0),
                        "prompt_tokens": len(request.prompt.split()),
                        "completion_tokens": len(data.get("response", "").split())
                    }
                )
    
    except httpx.TimeoutException:
        raise HTTPException(
            status_code=504,
            detail="请求超时，请稍后重试"
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"内部服务器错误: {str(e)}"
        )

@app.get("/api/v1/models")
async def list_models():
    """获取可用模型列表"""
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
            if response.status_code == 200:
                return response.json()
            else:
                return {"models": []}
    except Exception:
        return {"models": []}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3.4 创建配置文件和环境变量

创建一个.env文件来管理配置：

# 创建.env文件
cat > .env << EOF
# FastAPI配置
API_HOST=0.0.0.0
API_PORT=8000
API_WORKERS=4

# Ollama配置
OLLAMA_HOST=http://localhost:11434
OLLAMA_MODEL=deepseek-r1:7b

# 模型参数默认值
DEFAULT_MAX_TOKENS=512
DEFAULT_TEMPERATURE=0.7
DEFAULT_TOP_P=0.9

# 日志配置
LOG_LEVEL=INFO
LOG_FILE=logs/api.log
EOF

再创建一个requirements.txt文件：

fastapi==0.104.1
uvicorn[standard]==0.24.0
httpx==0.25.1
python-dotenv==1.0.0
aiohttp==3.9.1
pydantic==2.5.0

4. 启动与测试服务

4.1 启动FastAPI服务

现在我们可以启动服务了。有两种启动方式，选一种就行：

方式一：直接运行

# 确保在虚拟环境中
source venv/bin/activate

# 启动FastAPI服务
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

方式二：使用脚本启动

创建一个启动脚本start.sh：

#!/bin/bash

# 激活虚拟环境
source venv/bin/activate

# 启动服务
uvicorn main:app --host 0.0.0.0 --port 8000 \
  --workers 4 \
  --log-level info \
  --access-log

给脚本执行权限并运行：

chmod +x start.sh
./start.sh

服务启动后，你会看到类似这样的输出：

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

4.2 测试API接口

打开浏览器，访问 http://localhost:8000/docs，你会看到自动生成的API文档页面。这是FastAPI的一个很棒的功能，不需要额外写文档。

让我们测试几个接口：

测试健康检查接口：

curl http://localhost:8000/api/v1/health

应该返回类似这样的响应：

{
  "status": "healthy",
  "model": "deepseek-r1:7b",
  "ollama_status": "running",
  "timestamp": "2024-01-15T10:30:00.123456"
}

测试聊天接口：

curl -X POST http://localhost:8000/api/v1/chat \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "请用Python写一个快速排序算法",
    "max_tokens": 300,
    "temperature": 0.7
  }'

你会看到模型生成的代码和解释。DeepSeek-R1模型的一个特点是它会展示推理过程，所以回答会比较详细。

测试流式输出：

curl -X POST http://localhost:8000/api/v1/chat \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "解释一下什么是机器学习",
    "stream": true
  }'

流式输出会逐块返回结果，适合需要实时显示的场景。

5. 实际应用示例

5.1 代码助手应用

让我们创建一个简单的代码助手应用，展示如何在实际项目中使用这个API：

# code_assistant.py
import requests
import json
from typing import List, Dict

class CodeAssistant:
    def __init__(self, api_url: str = "http://localhost:8000"):
        self.api_url = api_url
        self.chat_endpoint = f"{api_url}/api/v1/chat"
    
    def generate_code(self, requirement: str, language: str = "python") -> str:
        """生成代码"""
        prompt = f"""请用{language}语言实现以下功能：
{requirement}

要求：
1. 代码要有完整的注释
2. 包含必要的错误处理
3. 提供使用示例
4. 代码要简洁高效"""
        
        response = requests.post(
            self.chat_endpoint,
            json={
                "prompt": prompt,
                "max_tokens": 800,
                "temperature": 0.3  # 低温度让代码更稳定
            }
        )
        
        if response.status_code == 200:
            return response.json()["response"]
        else:
            raise Exception(f"API请求失败: {response.text}")
    
    def explain_code(self, code: str) -> str:
        """解释代码"""
        prompt = f"""请详细解释以下代码的功能和工作原理：
```python
{code}

请从以下几个方面解释：

1. 代码的整体功能

2. 关键函数的作用

3. 算法的时间复杂度

4. 可能的优化点"""

    response = requests.post(
        self.chat_endpoint,
        json={
            "prompt": prompt,
            "max_tokens": 600
        }
    )
    
    if response.status_code == 200:
        return response.json()["response"]
    else:
        raise Exception(f"API请求失败: {response.text}")
   

   def debug_code(self, code: str, error: str) -> str: """调试代码""" prompt = f"""以下代码有错误：

{code}

错误信息： {error}

请：

1. 分析错误原因

2. 提供修复后的代码

3. 解释修复的原理"""

    response = requests.post(
        self.chat_endpoint,
        json={
            "prompt": prompt,
            "max_tokens": 700
        }
    )
    
    if response.status_code == 200:
        return response.json()["response"]
    else:
        raise Exception(f"API请求失败: {response.text}")
   

使用示例

if name == "main": assistant = CodeAssistant()

# 生成代码
requirement = "一个函数，接收整数列表，返回列表中的最大值和最小值"
code = assistant.generate_code(requirement)
print("生成的代码：")
print(code)
print("\n" + "="*50 + "\n")

# 解释代码
explanation = assistant.explain_code(code)
print("代码解释：")
print(explanation)

### 5.2 数学解题助手

DeepSeek-R1在数学推理方面特别强，我们可以创建一个数学助手：

```python
# math_assistant.py
import requests
import re

class MathAssistant:
    def __init__(self, api_url: str = "http://localhost:8000"):
        self.api_url = api_url
        self.chat_endpoint = f"{api_url}/api/v1/chat"
    
    def solve_math_problem(self, problem: str, show_steps: bool = True) -> str:
        """解决数学问题"""
        if show_steps:
            prompt = f"""请解决以下数学问题，并展示详细的解题步骤：

问题：{problem}

要求：
1. 分步骤解答
2. 解释每一步的原理
3. 给出最终答案
4. 如果有多种解法，请都列出来"""
        else:
            prompt = f"""请直接给出以下数学问题的答案：

问题：{problem}"""
        
        response = requests.post(
            self.chat_endpoint,
            json={
                "prompt": prompt,
                "max_tokens": 1000,
                "temperature": 0.1  # 低温度确保答案准确
            }
        )
        
        if response.status_code == 200:
            return response.json()["response"]
        else:
            raise Exception(f"API请求失败: {response.text}")
    
    def check_answer(self, problem: str, user_answer: str) -> dict:
        """检查答案是否正确"""
        prompt = f"""问题：{problem}

用户给出的答案：{user_answer}

请判断这个答案是否正确。
如果正确，请说"正确"并简要解释。
如果不正确，请：
1. 指出错误在哪里
2. 给出正确解法
3. 提供正确答案"""
        
        response = requests.post(
            self.chat_endpoint,
            json={
                "prompt": prompt,
                "max_tokens": 600
            }
        )
        
        if response.status_code == 200:
            result = response.json()["response"]
            is_correct = "正确" in result[:50]  # 简单判断
            
            return {
                "is_correct": is_correct,
                "feedback": result,
                "problem": problem,
                "user_answer": user_answer
            }
        else:
            raise Exception(f"API请求失败: {response.text}")

# 使用示例
if __name__ == "__main__":
    assistant = MathAssistant()
    
    # 解决数学问题
    problem = "一个长方形的长是8厘米，宽是5厘米，求它的面积和周长。"
    solution = assistant.solve_math_problem(problem)
    print("数学问题解答：")
    print(solution)
    print("\n" + "="*50 + "\n")
    
    # 检查答案
    user_answer = "面积是40平方厘米，周长是26厘米"
    check_result = assistant.check_answer(problem, user_answer)
    print("答案检查结果：")
    print(f"是否正确：{check_result['is_correct']}")
    print(f"反馈：{check_result['feedback']}")

6. 部署优化与监控

6.1 使用Docker容器化部署

为了更方便地部署和迁移，我们可以把整个服务打包成Docker镜像。

创建Dockerfile：

# Dockerfile
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

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

# 安装Ollama
RUN curl -fsSL https://ollama.com/install.sh | sh

# 复制Python依赖文件
COPY requirements.txt .

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

# 复制应用代码
COPY . .

# 拉取模型（可以在构建时或运行时进行）
# RUN ollama pull deepseek-r1:7b

# 暴露端口
EXPOSE 8000 11434

# 启动脚本
COPY start.sh .
RUN chmod +x start.sh

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/api/v1/health || exit 1

# 启动命令
CMD ["./start.sh"]

创建docker-compose.yml：

# docker-compose.yml
version: '3.8'

services:
  ollama:
    image: ollama/ollama:latest
    container_name: deepseek-ollama
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    command: serve
    restart: unless-stopped
  
  api:
    build: .
    container_name: deepseek-api
    ports:
      - "8000:8000"
    depends_on:
      - ollama
    environment:
      - OLLAMA_HOST=http://ollama:11434
    restart: unless-stopped
    volumes:
      - ./logs:/app/logs

volumes:
  ollama_data:

6.2 添加监控和日志

修改main.py，添加更完善的监控和日志：

# 在main.py中添加
import logging
from logging.handlers import RotatingFileHandler
import time

# 配置日志
def setup_logging():
    """配置日志系统"""
    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    
    # 文件日志
    file_handler = RotatingFileHandler(
        'logs/api.log',
        maxBytes=10*1024*1024,  # 10MB
        backupCount=5
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(logging.Formatter(log_format))
    
    # 控制台日志
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(logging.Formatter(log_format))
    
    # 获取根日志器
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    return logger

logger = setup_logging()

# 添加请求日志中间件
@app.middleware("http")
async def log_requests(request, call_next):
    """记录请求日志"""
    start_time = time.time()
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    logger.info(
        f"{request.method} {request.url.path} - "
        f"Status: {response.status_code} - "
        f"Time: {process_time:.3f}s"
    )
    
    return response

# 在聊天接口中添加详细日志
@app.post("/api/v1/chat", response_model=ChatResponse)
async def chat_completion(request: ChatRequest):
    # ... 原有代码 ...
    
    logger.info(f"收到聊天请求: {request.prompt[:100]}...")
    logger.info(f"请求参数: max_tokens={request.max_tokens}, "
                f"temperature={request.temperature}")
    
    try:
        # ... 处理逻辑 ...
        
        logger.info(f"请求处理完成，生成{len(full_response)}字符")
        return response
        
    except Exception as e:
        logger.error(f"处理请求时出错: {str(e)}")
        raise

6.3 性能优化建议

根据我的使用经验，这里有几个优化建议：

1. 启用模型缓存：

   # 在启动Ollama时启用缓存
   ollama serve --cache-size 2048
   

2. 调整FastAPI工作进程数：

   # 根据CPU核心数调整
   uvicorn main:app --workers $(nproc) --host 0.0.0.0 --port 8000
   

3. 添加请求限流：

   # 安装限流库
   # pip install slowapi
   
   from slowapi import Limiter, _rate_limit_exceeded_handler
   from slowapi.util import get_remote_address
   from slowapi.errors import RateLimitExceeded
   
   limiter = Limiter(key_func=get_remote_address)
   app.state.limiter = limiter
   app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
   
   @app.post("/api/v1/chat")
   @limiter.limit("10/minute")  # 每分钟10次
   async def chat_completion(request: ChatRequest):
       # ...
   

4. 使用连接池：

   # 创建全局HTTP客户端连接池
   import httpx
   
   class OllamaClient:
       def __init__(self):
           self.client = httpx.AsyncClient(
               base_url="http://localhost:11434",
               timeout=30.0,
               limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
           )
       
       async def generate(self, prompt: str, **kwargs):
           response = await self.client.post("/api/generate", json={
               "model": "deepseek-r1:7b",
               "prompt": prompt,
               **kwargs
           })
           return response.json()
   

7. 总结

通过这个教程，我们完成了一个完整的私有LLM服务搭建。从Ollama部署DeepSeek-R1-Distill-Qwen-7B模型，到用FastAPI构建RESTful API，再到实际应用示例和优化建议，每一步我都尽量用最直白的方式讲解。

这个方案有几个明显的优势：

• 完全私有化：数据不出本地，安全可控
• 成本可控：一次部署，长期使用，没有API调用费用
• 灵活定制：可以根据需要调整模型参数、添加业务逻辑
• 易于集成：标准的HTTP接口，任何编程语言都能调用

在实际使用中，DeepSeek-R1-Distill-Qwen-7B展现出了不错的推理能力，特别是在需要逐步思考的任务上。虽然7B参数规模不算大，但对于很多实际应用场景已经足够用了。

如果你需要更强的能力，也可以尝试DeepSeek-R1的更大版本，或者探索其他开源模型。Ollama支持很多主流模型，切换起来很方便。

部署过程中如果遇到问题，记得检查Ollama服务是否正常运行，端口是否被占用，以及模型是否成功加载。大多数问题都能通过查看日志找到原因。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。