LingBot-Vision 通过 FastAPI 暴露本地推理服务的实现

把LingBot-Vision变成HTTP服务，关键不是写路由，而是保证模型只加载一次、请求响应结构稳定。FastAPI的lifespan机制能管理模型生命周期，Pydantic能校验输入输出，配合uvicorn就能在本地跑起来。以下骨架假设你已经有一个能接收图像并返回文本的LingBot-Vision调用对象，具体模型类名和预处理函数要按实际SDK替换。

适用场景：已有LingBot-Vision本地推理脚本，需要封装成HTTP接口供团队调用。操作动作：用FastAPI定义/infer接口，lifespan加载模型，启动uvicorn。验证方式：用curl发送base64图片，检查返回JSON是否包含结果。风险边界：模型加载时长和显存占用需结合硬件确认，并发请求需要自行加锁或调整worker数。

项目结构与依赖安装

创建项目目录，例如lingbot-vision-api/，内部放requirements.txt和main.py。目录结构很简单：

lingbot-vision-api/
├── main.py
└── requirements.txt

requirements.txt内容如下，核心是fastapi和uvicorn，其他按模型实际依赖追加：

fastapi
uvicorn[standard]
pillow
torch
torchvision
# 如果模型基于transformers，再添加：
transformers

安装依赖：pip install -r requirements.txt。这里没有列出具体版本，因为不同环境下的默认版本可能冲突，建议安装后先跑一次import fastapi, uvicorn确认可用。

模型加载与全局生命周期管理

不要在请求处理函数里加载模型，否则每次调用都会重新初始化，既慢又占内存。用FastAPI的lifespan上下文管理器，在服务启动时加载一次LingBot-Vision模型，并存入全局变量。关闭服务时释放模型，避免显存残留。

from contextlib import asynccontextmanager
from fastapi import FastAPI

model = None

def load_lingbot_vision_model():
    # 这里替换成实际的LingBot-Vision初始化代码
    # 例如 model = LingBotVision.from_pretrained("checkpoint")
    pass

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    model = load_lingbot_vision_model()
    yield
    model = None

app = FastAPI(lifespan=lifespan)

注意：load_lingbot_vision_model是占位函数，你要按实际的SDK写法替换。如果模型加载失败，应用启动时会报错，此时服务不会进入端口监听，这是正常表现。

设计请求与响应模型

用Pydantic定义输入输出，能减少参数错误。请求体里必须包含image_base64字符串，可选参数可以额外增加，比如控制生成长度的max_new_tokens。响应里同时返回模型输出和耗时，方便调用方记录性能。

from pydantic import BaseModel, Field

class InferRequest(BaseModel):
    image_base64: str = Field(..., description="图片的base64编码字符串")
    max_new_tokens: int = Field(64, description="生成的最大token数", ge=1)

class InferResponse(BaseModel):
    result: str = Field(..., description="模型输出文本")
    elapsed_ms: float = Field(..., description="推理耗时（毫秒）")

这里只加了max_new_tokens一个可选参数，实际使用时可以按模型支持的能力再扩展，比如temperature、top_p等。

推理路由的实现与异常兜底

路由处理流程是：接收JSON -> base64解码 -> 图像预处理 -> 调用模型 -> 返回结构化结果。下面是一个完整的POST /infer示例：

import base64
import time
from io import BytesIO
from fastapi import HTTPException
from PIL import Image

def preprocess_image(image_bytes):
    # 替换成LingBot-Vision实际需要的预处理步骤
    img = Image.open(BytesIO(image_bytes))
    # 例如：resize、归一化、转tensor
    return img

@app.post("/infer", response_model=InferResponse)
async def infer(req: InferRequest):
    try:
        image_bytes = base64.b64decode(req.image_base64.split(",")[-1])
        img = preprocess_image(image_bytes)
        start = time.perf_counter()
        # 替换成模型实际推理方法，这里假设是 model.generate
        output = model.generate(img, max_new_tokens=req.max_new_tokens)
        elapsed_ms = (time.perf_counter() - start) * 1000
        return InferResponse(result=output, elapsed_ms=elapsed_ms)
    except Exception as e:
        # 记录详细异常日志，再返回500
        raise HTTPException(status_code=500, detail=f"推理失败: {str(e)}")

异常兜底覆盖了base64解码失败、图片格式不对、模型调用异常等情况。返回500时，服务端控制台会打印完整异常堆栈，方便定位。

本地启动与用curl验证接口

在项目目录下启动uvicorn：

uvicorn main:app `--host` 0.0.0.0 `--port` 8000

看到Application startup complete后，说明模型加载完成，服务可以接受请求。另开一个终端，先用base64编码图片再发送：

BASE64=$(base64 -w0 test.jpg)
curl -X POST http://127.0.0.1:8000/infer \
  -H "Content-Type: application/json" \
  -d "{\"image_base64\": \"$BASE64\", \"max_new_tokens\": 128}"

如果curl返回的内容里出现"result"和"elapsed_ms"字段，说明接口成功。如果返回{\"detail\":\"推理失败: ...\"}，检查服务端日志。网络不通时，先确认uvicorn是否监听在0.0.0.0:8000，以及防火墙是否放行。