FastAPI + Ollama 打造生产级本地 LLM API 服务

在本地运行语言模型并将其暴露为标准的 REST API 有真实的实用价值：没有速率限制、没有按 token 计费、数据不会离开你的基础设施。问题在于 Ollama 原生的 HTTP 接口是有意简化的——能用，但在任何实际工作负载中使用之前，你肯定需要认证、结构化的请求验证、流式响应支持和完善的错误处理。

这份指南构建的是一个你真正会部署的、围绕 Ollama 的薄层 FastAPI 封装。

你需要安装并运行 Ollama，且至少 pull 一个模型：

# Install Ollama (Linux)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull a model
ollama pull llama3.2:3b

# Verify
ollama list

对于 FastAPI 层：

pip install fastapi uvicorn httpx python-dotenv

Ollama 在 localhost:11434 上运行自己的 HTTP 服务器。这里的策略很简单：

FastAPI 处理来自客户端的请求

它用 Pydantic 验证载荷

将请求转发给 Ollama

返回响应——要么是完整的 JSON，要么是流式 SSE 响应

这样 Ollama 进程就保持为哑后端，让你在 FastAPI 层叠加认证、日志和请求转换，而无需触碰 Ollama 的配置。

# server.py
import os
import httpx
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
API_KEY = os.getenv("API_KEY", "change-me-in-production")

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

class ChatMessage(BaseModel):
    role: str = Field(..., pattern="^(system|user|assistant)$")
    content: str

class ChatRequest(BaseModel):
    model: str = Field(default="llama3.2:3b")
    messages: list[ChatMessage]
    stream: bool = False
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=1024, ge=1, le=8192)

def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

async def stream_ollama_response(payload: dict) -> AsyncGenerator[str, None]:
    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream(
            "POST",
            f"{OLLAMA_BASE_URL}/api/chat",
            json=payload,
        ) as response:
            async for line in response.aiter_lines():
                if line:
                    yield f"data: {line}\n\n"

@app.post("/v1/chat")
async def chat(request: ChatRequest, _: str = Depends(verify_api_key)):
    payload = {
        "model": request.model,
        "messages": [m.model_dump() for m in request.messages],
        "stream": request.stream,
        "options": {
            "temperature": request.temperature,
            "num_predict": request.max_tokens,
        },
    }

    if request.stream:
        return StreamingResponse(
            stream_ollama_response(payload),
            media_type="text/event-stream",
        )

    async with httpx.AsyncClient(timeout=120) as client:
        try:
            response = await client.post(
                f"{OLLAMA_BASE_URL}/api/chat",
                json=payload,
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise HTTPException(status_code=502, detail=f"Ollama error: {e.response.text}")
        except httpx.ConnectError:
            raise HTTPException(status_code=503, detail="Ollama is not running")

    return response.json()

@app.get("/v1/models")
async def list_models(_: str = Depends(verify_api_key)):
    async with httpx.AsyncClient(timeout=10) as client:
        try:
            resp = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
            resp.raise_for_status()
            return resp.json()
        except httpx.ConnectError:
            raise HTTPException(status_code=503, detail="Ollama is not running")

@app.get("/health")
async def health():
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            resp = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
        return {"status": "ok", "ollama": resp.status_code == 200}
    except httpx.ConnectError:
        return {"status": "degraded", "ollama": False}

API_KEY=my-secret-key uvicorn server:app --host 0.0.0.0 --port 8000

curl -X POST http://localhost:8000/v1/chat \
  -H "x-api-key: my-secret-key" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "llama3.2:3b",
    "messages": [{"role": "user", "content": "What is a JWT?"}]
  }'

处理流式响应

Ollama 在流式响应时返回换行符分隔的 JSON。stream_ollama_response 生成器将每个 chunk 包装为 Server-Sent Event。以下是一个消费它的客户端：

# client_stream.py
import json
import httpx

def stream_chat(prompt: str, api_key: str, base_url: str = "http://localhost:8000"):
    payload = {
        "model": "llama3.2:3b",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    with httpx.Client(timeout=120) as client:
        with client.stream(
            "POST",
            f"{base_url}/v1/chat",
            json=payload,
            headers={"x-api-key": api_key},
        ) as response:
            for line in response.iter_lines():
                if line.startswith("data: "):
                    chunk = json.loads(line[6:])
                    if not chunk.get("done"):
                        print(chunk["message"]["content"], end="", flush=True)
    print()

if __name__ == "__main__":
    stream_chat("Explain TLS handshake in 3 sentences.", "my-secret-key")

输出逐 token 出现——这与任何托管 API 的行为一致。

生产环境注意事项

在 localhost 之外部署之前，有几件事需要注意。

模型验证。当前代码接受任意模型字符串，让 Ollama 以 500 错误失败。更好的做法是在启动时获取可用模型列表，然后在每个请求时验证模型是否在那个集合中。

并发控制。Ollama 默认一次只处理一个请求（除非你的 GPU 支持真正的并行）。并发请求会在内部排队，但在高负载下会迅速堆积。用 asyncio.Semaphore 限制并发推理调用数：

_sem = asyncio.Semaphore(2)  # max 2 concurrent Ollama requests

@app.post("/v1/chat")
async def chat(request: ChatRequest, _: str = Depends(verify_api_key)):
    async with _sem:
        # ... forward to Ollama
        pass

超时。timeout=120 是一个上限，不是保证。在消费级硬件上，一个 13B 模型生成长响应可能超过 2 分钟。按模型调优超时时间，或者将超时暴露为可选的请求字段。

日志。在请求级别添加结构化日志：模型名称、输入 token 数（Ollama 会返回这个）、延迟和状态码。没有这些，调试慢响应会很痛苦。

安全响应头。如果这个 API 放在 nginx 后面，添加 X-Content-Type-Options、X-Frame-Options 和一个严格的 Content-Security-Policy。关于结构化 API 加固参考，这份安全检查清单涵盖了部署时容易被忽视的缺口。

TLS。不要在没有 TLS 终止的情况下直接暴露在公网接口上。在前面用 Caddy 或 nginx，即使对内部使用也是如此。

在语言模型前面加一层 FastAPI 封装大约需要 100 行代码，换来的是一个具有完善验证、认证和流式响应能力的自托管推理端点。与托管 API 相比，主要的权衡是：你拥有硬件和延迟特征，模型大小受限于显存，而且你需要负责更新和可靠性。

对于内部工具、空气隔离环境或成本敏感的工作负载——每天调用 API 数千次的情况下——这个技术栈具有真正的竞争力。它的局限在于高并发吞吐量——到了那个程度，改用 vLLM 而不是 Ollama，vLLM 是专为高吞吐量而非单请求延迟构建的。

---