DeepSeek-R1-Distill-Qwen-1.5B实战手册：API接口封装（FastAPI）与前端对接-程序员充电站

DeepSeek-R1-Distill-Qwen-1.5B实战手册：API接口封装（FastAPI）与前端对接

1. 为什么需要把Streamlit聊天界面“拆开”？

你已经用上那个清爽的Streamlit本地对话助手了——输入即响应、思考过程自动展开、显存一键清理，确实省心。但很快你会发现：它像一个功能完整的“黑盒子”，漂亮好用，却不太方便集成进你自己的系统。

比如：

你想在公司内部知识库网页里嵌入一个AI问答小窗口，而不是跳转到整个Streamlit页面；
你想用Python脚本批量测试模型对不同提示词的响应质量；
你想让手机App通过HTTP请求调用这个本地模型，而不是依赖浏览器；
你想把多个AI能力（文本、代码、推理）统一接入同一个后端网关，做权限、日志、限流管理。

这时候，Streamlit就不再是终点，而是起点。它验证了模型能跑、效果不错、交互友好；而真正的工程落地，需要把它变成一个可编程、可调度、可组合的服务接口。

本文不讲怎么再部署一遍模型，也不重复Streamlit界面操作。我们直接从你已有的/root/ds_1.5b本地环境出发，用最轻量、最稳妥的方式，把DeepSeek-R1-Distill-Qwen-1.5B的能力“抽出来”，封装成标准HTTP API，并完成与真实前端页面的对接验证。全程不改模型、不重训、不换框架，只加一层薄薄的胶水代码。

2. FastAPI封装：三步构建稳定可靠的推理服务

2.1 为什么选FastAPI而不是Flask或其它？

不是因为“新潮”，而是因为它刚好踩中了这个场景的三个关键点：

异步支持原生：模型推理是I/O密集型（加载权重、读写显存），FastAPI的async def能自然挂起等待GPU计算，避免阻塞其他请求；
自动文档生成：/docs页面自动生成OpenAPI规范，你不用写一行文档，前端同事就能看清每个参数怎么填、返回什么结构；
类型安全即契约：用Python类型注解（str,int,List[Message]）定义输入输出，Pydantic自动校验+转换，前端传错格式会立刻报422错误，而不是让模型崩溃或返回乱码。

更重要的是：它和你已有的Streamlit项目完全兼容——共享同一套模型加载逻辑、同一份tokenizer、同一组推理参数。你不是在重建轮子，而是在已有轮子上加个轴承。

2.2 核心服务代码（精简可运行版）

新建文件api_server.py，内容如下（已适配你本地的/root/ds_1.5b路径）：

# api_server.py import torch from fastapi import FastAPI, HTTPException from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForCausalLM from typing import List, Dict, Any # === 模型加载（复用Streamlit逻辑，仅执行一次）=== MODEL_PATH = "/root/ds_1.5b" # 使用torch.no_grad() + auto device_map，与Streamlit一致 model = AutoModelForCausalLM.from_pretrained( MODEL_PATH, device_map="auto", torch_dtype="auto", trust_remote_code=True, ) tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True) # 确保模型处于评估模式 model.eval() # === 数据模型定义 === class Message(BaseModel): role: str # "user" or "assistant" content: str class ChatRequest(BaseModel): messages: List[Message] max_new_tokens: int = 2048 temperature: float = 0.6 top_p: float = 0.95 class ChatResponse(BaseModel): response: str usage: Dict[str, Any] # === FastAPI应用 === app = FastAPI( title="DeepSeek-R1-Distill-Qwen-1.5B API", description="本地化、低显存、高可读的轻量推理服务", version="1.0" ) @app.post("/v1/chat/completions", response_model=ChatResponse) async def chat_completions(request: ChatRequest): try: # 1. 构建输入：复用官方chat template input_text = tokenizer.apply_chat_template( request.messages, tokenize=False, add_generation_prompt=True ) # 2. 编码 & 推理 inputs = tokenizer(input_text, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=request.max_new_tokens, temperature=request.temperature, top_p=request.top_p, do_sample=True, pad_token_id=tokenizer.eos_token_id, eos_token_id=tokenizer.eos_token_id, ) # 3. 解码并截断输入部分 full_response = tokenizer.decode(outputs[0], skip_special_tokens=True) # 去掉输入前缀，只保留模型生成内容（含思考链） if "assistant" in full_response: response_only = full_response.split("assistant")[-1].strip() else: response_only = full_response.strip() # 4. 自动格式化：将 <think>...</think> 转为「思考过程」+「回答」 # （此处简化处理，实际可复用Streamlit中成熟逻辑） if "<think>" in response_only and "</think>" in response_only: parts = response_only.split("<think>", 1) if len(parts) == 2: after_think = parts[1].split("</think>", 1) if len(after_think) == 2: thought = after_think[0].strip() answer = after_think[1].strip() formatted = f"「思考过程」\n{thought}\n\n「回答」\n{answer}" response_only = formatted return ChatResponse( response=response_only, usage={ "prompt_tokens": len(inputs["input_ids"][0]), "completion_tokens": len(outputs[0]) - len(inputs["input_ids"][0]), "total_tokens": len(outputs[0]) } ) except Exception as e: raise HTTPException(status_code=500, detail=f"推理失败: {str(e)}") @app.get("/health") def health_check(): return {"status": "ok", "model": "DeepSeek-R1-Distill-Qwen-1.5B", "device": str(model.device)}

2.3 启动与验证

安装依赖（确保已激活你的Python环境）：

pip install fastapi uvicorn transformers torch

启动服务：

uvicorn api_server:app --host 0.0.0.0 --port 8000 --reload

验证是否成功：

访问http://localhost:8000/health→ 应返回{"status":"ok",...}
访问http://localhost:8000/docs→ 自动生成交互式API文档，可直接试用

关键设计说明：

所有device_map="auto"、torch_dtype="auto"、torch.no_grad()均与Streamlit项目严格一致，避免因配置差异导致效果不一致；
apply_chat_template调用方式完全复用官方模板，保证多轮对话上下文拼接逻辑100%对齐；
返回结构遵循OpenAI-style（/v1/chat/completions），便于未来无缝切换至其他兼容服务；
错误处理明确：422用于参数错误，500用于推理异常，前端可精准捕获。

3. 前端对接：用纯HTML+JavaScript实现最小可行界面

3.1 不依赖任何框架，5分钟搭出可用UI

新建index.html，放入与api_server.py同级目录：

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>DeepSeek-R1 本地API对接</title> <style> body { font-family: "Segoe UI", system-ui; max-width: 800px; margin: 0 auto; padding: 20px; background: #f8f9fa; } #chat-container { height: 500px; overflow-y: auto; border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; background: white; } .message { margin-bottom: 16px; line-height: 1.5; } .user { text-align: right; } .assistant { background: #f0f8ff; padding: 12px 16px; border-radius: 12px; display: inline-block; max-width: 80%; } .user .assistant { background: #e8f4f8; } #input-area { width: 100%; padding: 12px; border: 1px solid #ccc; border-radius: 6px; margin-top: 16px; } button { background: #007bff; color: white; border: none; padding: 10px 20px; border-radius: 6px; cursor: pointer; } button:disabled { background: #ccc; cursor: not-allowed; } .thinking { color: #6c757d; font-size: 0.9em; } </style> </head> <body> <h1> DeepSeek-R1-Distill-Qwen-1.5B API对接演示</h1> <p><strong>说明：</strong>本页直连本地FastAPI服务（<code>http://localhost:8000</code>），所有数据不出设备。</p> <div id="chat-container"></div> <textarea id="input-area" placeholder="输入问题，例如：'推导勾股定理的三种方法'..." rows="3"></textarea> <br> <button id="send-btn">发送</button> <button id="clear-btn">清空对话</button> <script> const chatContainer = document.getElementById('chat-container'); const inputArea = document.getElementById('input-area'); const sendBtn = document.getElementById('send-btn'); const clearBtn = document.getElementById('clear-btn'); let messages = []; function addMessage(role, content) { const div = document.createElement('div'); div.className = `message ${role}`; if (role === 'assistant') { // 简单解析「思考过程」和「回答」 if (content.includes('「思考过程」')) { const parts = content.split('「思考过程」'); if (parts.length > 1) { const [thought, answer] = parts[1].split('「回答」').map(s => s.trim()); div.innerHTML = ` <div class="assistant"> <div class="thinking"><strong> 思考过程：</strong>${thought.replace(/\n/g, '<br>')}</div> <div><strong> 回答：</strong>${answer.replace(/\n/g, '<br>')}</div> </div> `; } else { div.innerHTML = `<div class="assistant">${content.replace(/\n/g, '<br>')}</div>`; } } else { div.innerHTML = `<div class="assistant">${content.replace(/\n/g, '<br>')}</div>`; } } else { div.innerHTML = `<div class="assistant">${content.replace(/\n/g, '<br>')}</div>`; } chatContainer.appendChild(div); chatContainer.scrollTop = chatContainer.scrollHeight; } async function sendMessage() { const userMsg = inputArea.value.trim(); if (!userMsg) return; // 添加用户消息 addMessage('user', userMsg); messages.push({ role: 'user', content: userMsg }); inputArea.value = ''; // 禁用按钮防重复提交 sendBtn.disabled = true; sendBtn.textContent = '思考中...'; try { const response = await fetch('http://localhost:8000/v1/chat/completions', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages: messages, max_new_tokens: 2048, temperature: 0.6, top_p: 0.95 }) }); if (!response.ok) { throw new Error(`HTTP ${response.status}: ${response.statusText}`); } const data = await response.json(); const assistantMsg = data.response || '抱歉，模型未返回有效内容。'; addMessage('assistant', assistantMsg); messages.push({ role: 'assistant', content: assistantMsg }); } catch (err) { addMessage('assistant', ` 请求失败：${err.message}`); console.error('API Error:', err); } finally { sendBtn.disabled = false; sendBtn.textContent = '发送'; } } sendBtn.addEventListener('click', sendMessage); inputArea.addEventListener('keydown', (e) => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); sendMessage(); } }); clearBtn.addEventListener('click', () => { messages = []; chatContainer.innerHTML = ''; }); // 初始化欢迎语 addMessage('assistant', '你好！我是本地运行的 DeepSeek-R1-Distill-Qwen-1.5B，支持逻辑推理、数学解题、代码生成等。请开始提问吧 👇'); </script> </body> </html>

3.2 对接要点解析：为什么这样写？

零构建工具链：纯HTML+JS，双击即可打开，无需Node.js、Webpack、Vite，适合快速验证、给非前端同事演示；
消息结构严格对齐：前端维护messages数组，格式与FastAPI期望的List[Message]完全一致，避免JSON序列化歧义；
思考过程可视化还原：前端主动识别「思考过程」标签并分段渲染，复现Streamlit中用户熟悉的阅读体验；
错误反馈友好：网络失败、API报错、模型无响应均有明确提示，不静默失败；
隐私明确声明：页面顶部清晰标注“所有数据不出设备”，增强用户信任。

效果验证：

启动FastAPI（uvicorn api_server:app --host 0.0.0.0 --port 8000）
用浏览器打开index.html
输入“用Python写一个快速排序函数”，点击发送 → 看到带思考过程的结构化回复

4. 工程级增强：让API真正“可交付”

4.1 加入请求队列与并发控制（防OOM）

1.5B模型虽轻，但在多用户并发时仍可能因显存争抢导致OOM。FastAPI本身不提供队列，我们用asyncio.Semaphore轻量实现：

# 在 api_server.py 顶部添加 import asyncio # ... # 全局信号量：最多允许2个并发推理请求 semaphore = asyncio.Semaphore(2) # 修改 chat_completions 函数开头： @app.post("/v1/chat/completions", response_model=ChatResponse) async def chat_completions(request: ChatRequest): async with semaphore: # ⬅ 关键：请求进入前先获取许可 try: # ...原有推理逻辑保持不变...

这样，第3个请求会自动等待前2个完成，避免显存超载，比粗暴的503 Service Unavailable更友好。

4.2 日志与性能追踪（定位慢请求）

在推理前后加入毫秒级计时与结构化日志：

import time import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 在 generate 前后加： start_time = time.time() outputs = model.generate(...) end_time = time.time() logger.info(f"推理完成 | 输入长度: {len(inputs['input_ids'][0])} | 输出长度: {len(outputs[0])} | 耗时: {end_time - start_time:.2f}s")

日志示例：

INFO:__main__:推理完成 | 输入长度: 42 | 输出长度: 187 | 耗时: 3.21s

便于你判断：是模型本身慢？还是网络传输慢？或是前端渲染慢？

4.3 支持CORS（跨域调试必备）

开发时前端常跑在http://localhost:5173，而API在8000，需显式放行：

# 在 api_server.py 中 app 创建后添加 from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://localhost:8080"], # 允许的前端地址 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )