大模型应用：大模型的本地 API 服务：FastAPI 封装与接口鉴权.44

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJsb2NhbF9sbG1fdXNlciIsImV4cCI6MTcxNzc0MjQwMH0.8Z7k1X8s7a9L0b8K7j6H5g4F3d2S1a0Z9Y8X7W6V5U4T3S2R1Q0P

{
  "alg": "HS256",  // 加密算法（我们用的是对称加密HS256）
  "typ": "JWT"     // 令牌类型，固定为JWT
}

token_data = {
    "sub": "local_llm_user",  # 自定义的用户标识
    "exp": datetime.utcnow() + timedelta(minutes=30)  # 过期时间
}

签名 = HMACSHA256(
    Base64编码(Header) + "." + Base64编码(Payload),
    服务器密钥（JWT_SECRET_KEY）
)

def create_jwt_token():
    token_data = {
        "sub": "local_llm_user",  # 身份标识
        "exp": datetime.utcnow() + timedelta(minutes=30)  # 过期时间
    }
    # 生成令牌：自动处理Header+Payload+Signature
    token = jwt.encode(token_data, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
    return token

Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJsb2NhbF9sbG1fdXNlciIsImV4cCI6MTcxNzc0MjQwMH0.8Z7k1X8s7a9L0b8K7j6H5g4F3d2S1a0Z9Y8X7W6V5U4T3S2R1Q0P

def verify_jwt_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):
    try:
        token = credentials.credentials
        # 解码+验证：自动校验签名+过期时间
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
        # 兜底检查过期时间（jwt库会自动校验，此处仅做提示）
        expire = payload.get("exp")
        if expire and datetime.utcfromtimestamp(expire) < datetime.utcnow():
            raise HTTPException(status_code=401, detail="令牌已过期")
        return payload.get("sub")  # 返回用户标识
    except JWTError:  # 签名错误/篡改都会触发这个异常
        raise HTTPException(status_code=401, detail="令牌无效")

# Windows
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple python-jose

# macOS/Linux
pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple python-jose

{
    "code": 200,
    "message": "令牌生成成功（有效期30分钟）",
    "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJsb2NhbF9sbG1fdXNlciIsImV4cCI6MTc2NzQ0OTkyOX0.LlpxEytdkOwzNcNLi-SZwYwwEMg093sJ-oSNpFMNpfM",
    "token_type": "bearer"
}

# model_path = "D:\\modelscope\\hub\\qwen\\Qwen1___5-1___8B-Chat"
# 导入需要的库
from fastapi import FastAPI, Header, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
from dotenv import load_dotenv
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os
import uvicorn

# ====================== 第一步：加载配置（API Key+JWT配置） ======================
try:
    load_dotenv()
    # API Key配置
    VALID_API_KEY = os.getenv("VALID_API_KEY", "default_key_123")  # 无配置时用默认值
    # JWT极简配置（新手不用改，理解即可）
    JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "my_jwt_secret_123")  # 加密密钥
    JWT_ALGORITHM = "HS256"  # 加密算法（固定）
    JWT_EXPIRE_MINUTES = 30  # 令牌过期时间（30分钟）
    print("✅ 配置加载成功")
except Exception as e:
    print(f"❌ 配置加载失败：{str(e)}")
    exit(1)

# ====================== 第二步：单例加载模型（修复路径+内存优化） ======================
class LocalLLM:
    _instance = None
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not LocalLLM._initialized:
            # 修复1：用超轻量化模型（DistilGPT2），避免内存不足
            # 先运行download_small_model.py下载模型到./model文件夹
            model_path = "D:\\modelscope\\hub\\qwen\\Qwen1___5-1___8B-Chat"  # 原始字符串避免反斜杠问题
            try:
                print("正在加载本地大模型...")
                if not os.path.exists(model_path):
                    raise FileNotFoundError(
                        f"模型文件夹不存在：{model_path}\n解决：1. 运行download_small_model.py下载小模型；2. 确认路径正确")
                
                # 修复2：添加low_cpu_mem_usage，减少内存占用
                self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
                self.model = AutoModelForCausalLM.from_pretrained(
                    model_path, 
                    torch_dtype=torch.float32, 
                    device_map="cpu", 
                    trust_remote_code=True,
                    low_cpu_mem_usage=True  # 减少CPU内存占用
                )
                self.model.eval()
                LocalLLM._initialized = True
                print("✅ 模型加载完成！")

            except FileNotFoundError as e:
                print(f"❌ 模型加载失败：{str(e)}")
                exit(1)
            except RuntimeError as e:
                if "out of memory" in str(e).lower():
                    print(f"❌ 内存不足，请关闭其他程序（如浏览器/微信），或换更小的模型")
                else:
                    print(f"❌ 模型运行错误：{str(e)}")
                exit(1)
            except Exception as e:
                print(f"❌ 模型加载未知错误：{str(e)}")
                exit(1)

    def generate_text(self, prompt, max_length):
        try:
            if not prompt.strip():
                raise ValueError("提示词不能为空")
            inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
            if inputs.input_ids.shape[1] > 512:
                raise ValueError(f"提示词过长（{inputs.input_ids.shape[1]}token），最大支持512token")
            
            with torch.no_grad():
                outputs = self.model.generate(**inputs, max_length=max_length, temperature=0.7, pad_token_id=self.tokenizer.eos_token_id)
            
            result = self.tokenizer.decode(outputs[0], skip_special_tokens=True).replace(prompt, "").strip()
            if not result:
                raise ValueError("模型未生成内容，请换提示词")
            return result

        except ValueError as e:
            raise HTTPException(status_code=400, detail=f"参数错误：{str(e)}")
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"推理错误：{str(e)}")

# 创建模型实例
try:
    llm = LocalLLM()
except Exception as e:
    print(f"❌ 模型初始化失败：{str(e)}")
    exit(1)

# ====================== 第三步：极简JWT鉴权核心函数 ======================
bearer_scheme = HTTPBearer()

def create_jwt_token():
    """生成JWT令牌（简化版：无用户名密码，直接生成）"""
    try:
        token_data = {
            "sub": "local_llm_user",
            "exp": datetime.utcnow() + timedelta(minutes=JWT_EXPIRE_MINUTES)
        }
        token = jwt.encode(token_data, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
        return token
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"生成令牌失败：{str(e)}")

def verify_jwt_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):
    """验证JWT令牌（简化版：只检查有效性和过期时间）"""
    try:
        token = credentials.credentials
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
        expire = payload.get("exp")
        if expire and datetime.utcfromtimestamp(expire) < datetime.utcnow():
            raise HTTPException(status_code=401, detail="❌ JWT令牌已过期\n解决：重新调用/get-token获取新令牌")
        return payload.get("sub")
    except JWTError as e:
        raise HTTPException(status_code=401, detail=f"❌ JWT令牌无效/签名错误\n解决：检查令牌是否正确，或重新获取")
    except Exception as e:
        raise HTTPException(status_code=401, detail=f"❌ 令牌验证失败：{str(e)}")

# ====================== 第四步：FastAPI接口（API Key+JWT双鉴权） ======================
app = FastAPI(title="本地大模型API（API Key+极简JWT鉴权）")

# 请求体模型（保留）
class LLMRequest(BaseModel):
    prompt: str = Field(..., description="提示词", max_length=500)
    max_length: int = Field(100, description="生成长度", ge=10, le=1000)

# 接口1：获取JWT令牌
@app.post("/get-token", summary="获取JWT令牌（无需密码，直接领）")
def get_jwt_token():
    token = create_jwt_token()
    return {
        "code": 200,
        "message": "令牌生成成功（有效期30分钟）",
        "access_token": token,
        "token_type": "bearer"
    }

# 接口2：API Key鉴权的生成接口
@app.post("/generate-text-apikey", summary="API Key鉴权-文本生成")
def generate_by_apikey(request: LLMRequest, x_api_key: str = Header(None)):
    try:
        if not x_api_key:
            raise HTTPException(status_code=401, detail="❌ 请传入X-API-Key")
        if x_api_key.strip() != VALID_API_KEY:
            raise HTTPException(status_code=401, detail="❌ API Key错误")
    except HTTPException as e:
        raise e

    try:
        result = llm.generate_text(request.prompt, request.max_length)
        return {"code": 200, "message": "生成成功", "data": {"prompt": request.prompt, "result": result}}
    except HTTPException as e:
        raise e

# 接口3：JWT鉴权的生成接口
@app.post("/generate-text-jwt", summary="JWT鉴权-文本生成")
def generate_by_jwt(
    request: LLMRequest,
    username: str = Depends(verify_jwt_token)
):
    try:
        result = llm.generate_text(request.prompt, request.max_length)
        return {
            "code": 200,
            "message": f"用户{username}生成成功",
            "data": {"prompt": request.prompt, "result": result}
        }
    except HTTPException as e:
        raise e

# ====================== 第五步：启动服务（修复端口+日志） ======================
if __name__ == "__main__":
    try:
        # 修复3：改端口为8080（避免8000被占用），添加更多日志
        uvicorn.run(
            "main:app",
            host="0.0.0.0",
            port=8080,  # 改端口，避免占用
            reload=False,
            log_level="debug",
            access_log=True  # 开启访问日志，便于排查
        )
    except OSError as e:
        if "address already in use" in str(e).lower():
            print(f"❌ 服务启动失败：端口8080被占用\n解决：1. 改端口为8090；2. 结束占用8080的程序")
        else:
            print(f"❌ 服务启动失败：{str(e)}")
        exit(1)
    except Exception as e:
        print(f"❌ 服务启动失败：{str(e)}\n解决：检查Python依赖是否安装完整")
        exit(1)