本地部署大模型完全指南④：API服务化与远程访问

第①~③篇完成了本地模型和知识库的搭建。现在我们要做的是——把它变成真正的服务，供团队和外部系统使用。

前言：为什么要API服务化？

本地模型跑在终端里自娱自乐有什么用？真正的价值在于让别人也能用：

• 团队协作：前端、后端、产品经理都能调用同一个模型
• 系统集成：集成到公司现有的IM、文档系统、CI/CD流水线
• 移动端访问：手机、平板通过API调用本地模型
• 弹性扩展：从单机到多机集群，API接口不变

---

一、Ollama原生API详解

Ollama本身就内置了HTTP API，部署后默认监听 11434 端口。

1.1 API速查表

端点	方法	功能
/api/generate	POST	文本生成（单轮）
/api/chat	POST	对话（多轮+工具调用）
/api/embeddings	POST	文本向量化
/api/tags	GET	列出所有模型
/api/show	POST	查看模型详情
/api/pull	POST	下载模型
/api/push	POST	上传模型
/api/delete	DELETE	删除模型
/api/copy	POST	复制模型
/api/version	GET	查看版本

1.2 流式与非流式调用

非流式（等待完整响应）：

import requests
import json

def generate(prompt: str):
    response = requests.post("http://localhost:11434/api/generate", json={
        "model": "deepseek-r1:7b",
        "prompt": prompt,
        "stream": False,
        "options": {
            "temperature": 0.7,
            "num_predict": 1024
        }
    })
    return response.json()["response"]

流式（逐token返回，体验更好）：

def generate_stream(prompt: str):
    response = requests.post("http://localhost:11434/api/generate", json={
        "model": "deepseek-r1:7b",
        "prompt": prompt,
        "stream": True  # 开启流式
    }, stream=True)
    
    for line in response.iter_lines():
        if line:
            chunk = json.loads(line)
            if "response" in chunk:
                yield chunk["response"]

# 使用
for token in generate_stream("用Python写一个冒泡排序"):
    print(token, end="", flush=True)

1.3 多轮对话API

def chat(messages: list):
    """Ollama对话API"""
    response = requests.post("http://localhost:11434/api/chat", json={
        "model": "deepseek-r1:7b",
        "messages": messages,
        "stream": False
    })
    return response.json()["message"]["content"]

# 多轮对话示例
history = [
    {"role": "system", "content": "你是一个专业的Python开发助手"},
    {"role": "user", "content": "Python的装饰器是什么？"},
]

# 第一轮
reply1 = chat(history)
print(f"AI: {reply1}")

# 第二轮（带上历史）
history.append({"role": "assistant", "content": reply1})
history.append({"role": "user", "content": "给我举个例子"})

reply2 = chat(history)
print(f"AI: {reply2}")

---

二、构建API网关

直接用Ollama的API虽然能用，但缺乏权限控制、限流、日志等企业级功能。我们需要一个API网关。

2.1 用Nginx做反向代理+鉴权

# /etc/nginx/conf.d/ollama-gateway.conf

upstream ollama_backend {
    server localhost:11434;
    keepalive 64;
}

server {
    listen 8080;
    server_name ai-api.company.com;
    
    # 启用HTTPS（生产环境必配）
    # listen 443 ssl;
    # ssl_certificate /etc/nginx/certs/ai-api.crt;
    # ssl_certificate_key /etc/nginx/certs/ai-api.key;
    
    # 请求体大小限制
    client_max_body_size 100m;
    
    # 速率限制（每个IP每分钟30次请求）
    limit_req_zone $binary_remote_addr zone=ollama:10m rate=30r/m;
    
    location / {
        limit_req zone=ollama burst=5 nodelay;
        
        # API Key鉴权
        auth_request /auth;
        auth_request_set $auth_status $upstream_status;
        
        proxy_pass http://ollama_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        
        # 流式响应支持
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding on;
        proxy_buffering off;
    }
    
    # 鉴权端点
    location = /auth {
        internal;
        proxy_pass http://127.0.0.1:8081/auth;
        proxy_pass_request_body off;
        proxy_set_header Content-Length "";
        proxy_set_header X-Original-URI $request_uri;
    }
}

2.2 Python鉴权服务

# auth_server.py — API Key鉴权服务
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import hmac

# 预设的API Keys（实际使用应存储在数据库）
VALID_KEYS = {
    "sk-prod-abc123": {"user": "admin", "role": "admin"},
    "sk-prod-def456": {"user": "developer", "role": "user"},
    "sk-dev-xyz789": {"user": "tester", "role": "user", "rate_limit": 10},
}

class AuthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 从Header中提取API Key
        api_key = self.headers.get("X-API-Key", "")
        
        if api_key in VALID_KEYS:
            user_info = VALID_KEYS[api_key]
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(user_info).encode())
        else:
            self.send_response(401)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": "Invalid API Key"}).encode())
    
    def log_message(self, format, *args):
        """静默日志"""
        pass

if __name__ == "__main__":
    server = HTTPServer(("127.0.0.1", 8081), AuthHandler)
    print("Auth server running on port 8081...")
    server.serve_forever()

2.3 完整的API调用示例

# 通过网关调用（需要API Key）
curl -X POST http://localhost:8080/api/generate \
  -H "X-API-Key: sk-prod-abc123" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "deepseek-r1:7b",
    "prompt": "用Python写一个HTTP服务器",
    "stream": false
  }'

---

三、FastAPI封装的模型服务

Nginx方案适合简单的反向代理。如果你需要更多的控制逻辑（如多模型路由、负载均衡、使用统计），建议用FastAPI封装一层。

3.1 完整的模型服务

# model_service.py — 完整的模型API服务
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
from typing import Optional, List
import requests
import time
import json
from datetime import datetime

app = FastAPI(title="Local LLM API Service", version="1.0.0")

# 配置
OLLAMA_BASE = "http://localhost:11434"
API_KEYS = {"sk-123": "admin", "sk-456": "user"}

# 使用统计
usage_stats = {"total_requests": 0, "total_tokens": 0}

# 请求模型
class GenerateRequest(BaseModel):
    model: str = "deepseek-r1:7b"
    prompt: str
    system: Optional[str] = None
    temperature: float = 0.7
    max_tokens: int = 1024
    stream: bool = False

class ChatRequest(BaseModel):
    model: str = "deepseek-r1:7b"
    messages: list
    temperature: float = 0.7
    stream: bool = False

class EmbeddingRequest(BaseModel):
    model: str = "bge-m3"
    input: str

async def verify_api_key(x_api_key: str = Header(None)):
    """验证API Key"""
    if x_api_key not in API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return API_KEYS[x_api_key]

@app.post("/v1/generate")
async def generate(req: GenerateRequest, user: str = None):
    """文本生成接口"""
    start = time.time()
    
    # 构造Ollama请求
    ollama_request = {
        "model": req.model,
        "prompt": req.prompt,
        "stream": False,
        "options": {
            "temperature": req.temperature,
            "num_predict": req.max_tokens
        }
    }
    
    if req.system:
        ollama_request["system"] = req.system
    
    # 调用Ollama
    response = requests.post(
        f"{OLLAMA_BASE}/api/generate",
        json=ollama_request
    )
    
    if response.status_code != 200:
        raise HTTPException(status_code=500, detail="Model inference failed")
    
    data = response.json()
    
    # 更新统计
    elapsed = time.time() - start
    usage_stats["total_requests"] += 1
    usage_stats["total_tokens"] += data.get("eval_count", 0)
    
    return {
        "response": data["response"],
        "model": req.model,
        "usage": {
            "total_tokens": data.get("eval_count", 0),
            "inference_time_ms": int(elapsed * 1000)
        }
    }

@app.post("/v1/chat")
async def chat(req: ChatRequest):
    """多轮对话接口"""
    ollama_request = {
        "model": req.model,
        "messages": req.messages,
        "stream": False,
        "options": {
            "temperature": req.temperature
        }
    }
    
    response = requests.post(
        f"{OLLAMA_BASE}/api/chat",
        json=ollama_request
    )
    
    if response.status_code != 200:
        raise HTTPException(status_code=500, detail="Chat failed")
    
    data = response.json()
    return {
        "response": data["message"]["content"],
        "model": req.model
    }

@app.post("/v1/embeddings")
async def embeddings(req: EmbeddingRequest):
    """文本向量化接口"""
    response = requests.post(
        f"{OLLAMA_BASE}/api/embeddings",
        json={
            "model": req.model,
            "prompt": req.input
        }
    )
    
    if response.status_code != 200:
        raise HTTPException(status_code=500, detail="Embedding failed")
    
    data = response.json()
    return {
        "embedding": data["embedding"],
        "dimensions": len(data["embedding"])
    }

@app.get("/v1/models")
async def list_models():
    """列出可用模型"""
    response = requests.get(f"{OLLAMA_BASE}/api/tags")
    models = response.json().get("models", [])
    
    return {
        "models": [
            {
                "name": m["name"],
                "size_gb": round(m["size"] / (1024**3), 2),
                "modified_at": m.get("modified_at", "")
            }
            for m in models
        ]
    }

@app.get("/v1/stats")
async def get_stats():
    """使用统计"""
    return usage_stats

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

3.2 启动服务

# 安装依赖
pip install fastapi uvicorn requests

# 启动（开发环境）
python model_service.py

# 启动（生产环境，多进程）
uvicorn model_service:app --host 0.0.0.0 --port 8000 --workers 4

# 生产环境建议用systemd管理

3.3 客户端库

# llm_client.py — 封装好的客户端
import requests
from typing import Optional, Generator

class LocalLLMClient:
    """本地大模型API客户端"""
    
    def __init__(self, base_url: str = "http://localhost:8000", 
                 api_key: str = "sk-123"):
        self.base_url = base_url
        self.headers = {"X-API-Key": api_key}
    
    def generate(self, prompt: str, model: str = "deepseek-r1:7b",
                 temperature: float = 0.7, max_tokens: int = 1024) -> str:
        """文本生成"""
        response = requests.post(
            f"{self.base_url}/v1/generate",
            headers=self.headers,
            json={
                "model": model,
                "prompt": prompt,
                "temperature": temperature,
                "max_tokens": max_tokens,
                "stream": False
            }
        )
        return response.json()["response"]
    
    def chat(self, messages: list, model: str = "deepseek-r1:7b") -> str:
        """多轮对话"""
        response = requests.post(
            f"{self.base_url}/v1/chat",
            headers=self.headers,
            json={"model": model, "messages": messages}
        )
        return response.json()["response"]
    
    def list_models(self) -> list:
        """列出可用模型"""
        response = requests.get(
            f"{self.base_url}/v1/models",
            headers=self.headers
        )
        return response.json()["models"]

# 使用
client = LocalLLMClient()
print(client.generate("用Python写一个文件监控工具"))

---

四、内网穿透：让外网访问你的模型

本地部署的最大限制是——只能在局域网用。如何让外网也能访问？

4.1 方案一：frp内网穿透（推荐）

# frpc.ini (内网机)
[common]
server_addr = your-server.com
server_port = 7000

[ollama-api]
type = tcp
local_ip = 127.0.0.1
local_port = 8000
remote_port = 8000

[open-webui]
type = tcp
local_ip = 127.0.0.1
local_port = 3000
remote_port = 3000

# frps.ini (公网服务器)
[common]
bind_port = 7000
vhost_http_port = 8080

4.2 方案二：Tailscale（零配置VPN）

# 安装Tailscale
# https://tailscale.com/download

# 所有设备登录同一账号后，自动组网
tailscale up

# 获取虚拟IP
tailscale ip

# 其他设备通过虚拟IP访问：http://100.x.x.x:8000

4.3 方案三：Cloudflare Tunnel（免费HTTPS）

# 安装cloudflared
# https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/

cloudflared tunnel create my-llm-tunnel
cloudflared tunnel route dns my-llm-tunnel ai-api.your-domain.com

# 创建配置文件
cat > ~/.cloudflared/config.yml << EOF
tunnel: my-llm-tunnel
credentials-file: /root/.cloudflared/my-llm-tunnel.json

ingress:
  - hostname: ai-api.your-domain.com
    service: http://localhost:8000
  - service: http_status:404
EOF

# 启动
cloudflared tunnel run my-llm-tunnel

---

五、安全加固建议

5.1 必做的安全措施

措施	说明	紧急程度
API Key认证	所有请求必须携带有效Key	⭐⭐⭐
HTTPS加密	防止中间人攻击	⭐⭐⭐
速率限制	防止滥用	⭐⭐⭐
请求审计日志	记录谁在什么时候调用了什么	⭐⭐
内容过滤	防止生成违规内容	⭐⭐
白名单IP	限制来源IP	⭐⭐

5.2 日志审计

# 简单审计日志中间件
import logging
from datetime import datetime

logging.basicConfig(
    filename="api_audit.log",
    level=logging.INFO,
    format="%(asctime)s | %(message)s"
)

async def audit_middleware(request, call_next):
    """审计日志"""
    start = time.time()
    
    # 记录请求
    api_key = request.headers.get("x-api-key", "unknown")
    path = request.url.path
    client_ip = request.client.host
    
    # 处理请求
    response = await call_next(request)
    
    # 记录结果
    elapsed = time.time() - start
    logging.info(
        f"{client_ip} | {api_key[:8]}... | {path} | "
        f"{response.status_code} | {elapsed:.2f}s"
    )
    
    return response

---

总结

至此，你的本地大模型已经从"个人玩具"升级为"企业级服务"——有API鉴权、有网关、有内网穿透、有日志审计。团队所有人都能用，也能集成到现有系统中。

下一篇预告：第⑤篇《搭建美观的对话界面（Open WebUI）》—— 给你的AI服务加上专业级的前端界面。

---

需要完整脚本和配置文件的同学，可以看我主页的付费资源专栏。

有问题欢迎评论区留言，大家一起讨论！