Ollama+FastAPI：150行代码搭建本地LLM API服务

在本地运行语言模型意味着你掌握数据控制权、没有按 token 计费的烦恼、还可以根据使用场景调优延迟。问题是 Ollama 内置的 HTTP API 过于精简——无认证、无 schema 校验、与现有 Python 技术栈集成也不方便。用 FastAPI 包装一下，不到 150 行代码就能解决这些问题，并给你一个可以交给团队成员或接入流水线的端点，而无需暴露底层推理基础设施。

Ollama 提供了一种极其简单的方式来拉取并本地或在私有服务器上运行开源模型（Llama 3、Mistral、Qwen、Phi 等）。它的 REST API 功能齐全但比较基础：没有中间件、没有请求校验、没有用量追踪。FastAPI 正好补齐这些缺失的部分：

• Pydantic 模型——输入校验和清晰的错误信息
• 兼容 Ollama SSE 流式输出的异步支持
• 认证中间件、限流、结构化日志
• 开箱即用的 OpenAPI 文档，位于 /docs

最终得到的是一个可直接替换云端 LLM API 的本地方案，完全由你掌控。数据不离开服务器、没有按 token 计费、高并发时也没有厂商限流。

在 Linux 或 macOS 上安装 Ollama：

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2
ollama serve  # 默认启动在 http://localhost:11434

验证裸 API 是否正常响应：

curl http://localhost:11434/api/generate \
  -d '{"model": "llama3.2", "prompt": "Hello", "stream": false}'

Ollama 暴露两个主要端点：/api/generate 用于单轮补全，/api/chat 用于带消息历史的多轮对话。我们将对这两个端点做代理和增强。

安装依赖：

pip install fastapi uvicorn httpx pydantic

以下是完整的服务器代码：

import httpx
import json
import os
import time
import logging
from typing import Optional, List

from fastapi import FastAPI, HTTPException, Depends, Header, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="Local LLM API", version="1.0.0")
logger = logging.getLogger("llm_api")
logging.basicConfig(level=logging.INFO)

OLLAMA_BASE_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
API_KEY = os.getenv("API_KEY", "")

# --- Auth ---

async def verify_api_key(x_api_key: Optional[str] = Header(None)):
    if API_KEY and x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")

# --- Request/Response models ---

class Message(BaseModel):
    role: str   # "user" or "assistant"
    content: str

class ChatRequest(BaseModel):
    model: str = "llama3.2"
    messages: List[Message]
    stream: bool = False
    temperature: Optional[float] = 0.7
    max_tokens: Optional[int] = None

# --- Middleware for latency logging ---

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.monotonic()
    response = await call_next(request)
    duration = time.monotonic() - start
    logger.info(
        "method=%s path=%s status=%d duration=%.3fs",
        request.method, request.url.path,
        response.status_code, duration,
    )
    return response

# --- Routes ---

@app.get("/health")
async def health():
    async with httpx.AsyncClient() as client:
        try:
            r = await client.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=3.0)
            r.raise_for_status()
            models = [m["name"] for m in r.json().get("models", [])]
            return {"status": "ok", "models": models}
        except Exception:
            raise HTTPException(status_code=503, detail="Ollama unreachable")

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

    if req.stream:
        return StreamingResponse(
            _stream_ollama("/api/chat", payload),
            media_type="text/event-stream",
        )

    async with httpx.AsyncClient(timeout=120.0) as client:
        r = await client.post(f"{OLLAMA_BASE_URL}/api/chat", json=payload)
        if r.status_code != 200:
            raise HTTPException(status_code=r.status_code, detail=r.text)
        data = r.json()
        return {
            "model": req.model,
            "content": data["message"]["content"],
            "done": data.get("done", True),
        }

async def _stream_ollama(path: str, payload: dict):
    async with httpx.AsyncClient(timeout=120.0) as client:
        async with client.stream(
            "POST", f"{OLLAMA_BASE_URL}{path}", json=payload
        ) as r:
            async for line in r.aiter_lines():
                if line:
                    yield f"data: {line}\n\n"
    yield "data: [DONE]\n\n"

启动服务：

API_KEY=mysecret uvicorn server:app --host 0.0.0.0 --port 8080

/health 端点告诉你当前加载了哪些模型。访问 /docs 可以看到自动生成的 OpenAPI UI——方便团队成员探索服务器支持哪些功能。

消费流式端点

对于聊天界面和长输出场景，流式传输很关键。首 token 延迟决定了感知响应速度。以下是一个极简的 Python 客户端，在 token 到达时直接打印：

import httpx, json

def stream_chat(prompt: str, api_key: str = "mysecret"):
    messages = [{"role": "user", "content": prompt}]
    with httpx.Client() as client:
        with client.stream(
            "POST",
            "http://localhost:8080/v1/chat",
            json={"messages": messages, "stream": True},
            headers={"x-api-key": api_key},
            timeout=120.0,
        ) as r:
            for line in r.iter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    chunk = json.loads(line[6:])
                    token = chunk.get("message", {}).get("content", "")
                    if token:
                        print(token, end="", flush=True)
    print()

stream_chat("Explain mTLS in two sentences")

[DONE] 哨兵让客户端循环保持干净，不依赖连接关闭检测，如果你在做前端也容易改写成 JavaScript EventSource 客户端。

部署为 Systemd 服务

要让本地 API 持久运行，需要把 Ollama 和 FastAPI 都部署为 systemd 服务。创建 /etc/systemd/system/llm-api.service：

[Unit]
Description=Local LLM API (FastAPI)
After=ollama.service

[Service]
User=llm
WorkingDirectory=/opt/llm-api
ExecStart=/opt/llm-api/venv/bin/uvicorn server:app --host 127.0.0.1 --port 8080
Restart=on-failure
Environment=API_KEY=your_secret_here
Environment=OLLAMA_URL=http://127.0.0.1:11434

[Install]
WantedBy=multi-user.target

启用服务：

systemctl daemon-reload
systemctl enable --now llm-api

在前面加一层 Nginx 终止 TLS，你就拥有了一个生产级的本地 LLM 端点，而且重启后依然存活。关于 API 层的安全加固——认证模式、限流配置、允许模型白名单——我们发布的安全加固清单里有这些模式的详细配置和具体示例。

基础跑起来后，有几个自然的扩展方向：

请求缓存：对 (model, messages, temperature) 元组做哈希，把响应存到 Redis 或 SQLite。重复的相同 prompt 零推理成本瞬间返回——对 FAQ 类机器人很有用。

模型路由：根据 prompt 长度和复杂度判断，短的事实类查询走小而快的模型（Phi-3 mini），需要更多推理的长任务走更大的模型。Ollama 的 /api/tags 告诉你有哪些可用模型。

Token 用量追踪：每次请求往 SQLite 写一行——模型名、Ollama 的 eval_count 和 prompt_eval_count 字段、延迟。即使是本地推理也能拿到等效成本数据，从而公平地做模型对比。

按 API Key 限流：当前实现只有一个 key。在 dict 或 SQLite 表里加一个 key→层级映射，对每个 key 用令牌桶装饰器做限流。

Ollama 负责模型管理和推理。FastAPI 负责它周围的一切：校验、认证、流式输出、可观测性。这个组合大约 150 行 Python，给你一个端到端可控的私有、免费、API 兼容的 LLM 端点。

裸 Ollama API 对本地小打小闹够用了。一旦你需要把端点分享给团队成员、加认证、或者接入生产流水线，用 FastAPI 包装一下只需要一个下午，随即在清晰度和可维护性上立刻回本。