Pine
e44fea96ca
- relay 鉴权重写：JWT→引擎 PAT 逐用户归属，拒绝匿名调用，余额<=0 预检 403 - compute_client 增 list_engine_users/get_engine_user/sync_user_mirror 等 - 运营端新增 /admin/compute/platform-users 平台用户视图(引擎真实余额/折扣/用量)与充值镜像回写 Co-Authored-By: Claude <noreply@anthropic.com>
136 lines
5.4 KiB
Python
136 lines
5.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""OpenAI 兼容模型中转端点。
|
||
|
||
PineAgents(8088) 的 ``pineagents`` provider 把 base_url 指向本服务的 ``/v1``，
|
||
请求进来后丢弃客户端任何 ``Authorization``，统一替换为服务端凭据。
|
||
|
||
模型调用统一经 **compute-engine**（loopback :3000）中转计量（引擎持有渠道 key，
|
||
前端不直连）；走 ``PINEAGENTS_COMPUTE_ADMIN_TOKEN`` 管理令牌。流式 SSE 原样透传。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
from starlette.background import BackgroundTask
|
||
|
||
from ... import config
|
||
from ...services import compute_catalog, compute_client
|
||
|
||
router = APIRouter(prefix="/v1", tags=["relay"])
|
||
|
||
|
||
def _is_jwt(token: str) -> bool:
|
||
"""平台 JWT 形如 header.payload.signature（两处 '.'）；引擎 PAT 不含 '.'。"""
|
||
return token.count(".") == 2
|
||
|
||
|
||
async def _user_engine_pat(token: str) -> str:
|
||
"""把平台 JWT 解析为该用户的引擎消费令牌（PAT）。
|
||
|
||
取该用户引擎名下第一枚有效令牌；没有则现场签发一枚，保证「登录用户=引擎用户」。
|
||
"""
|
||
from ...jwt import decode_access_token
|
||
|
||
claims = decode_access_token(token)
|
||
username = (claims or {}).get("username") or ""
|
||
if not username:
|
||
raise HTTPException(status_code=401, detail="登录态无效，请重新登录")
|
||
await compute_client.ensure_user(username)
|
||
items = await compute_client.list_user_tokens(username)
|
||
for it in items or []:
|
||
key = it.get("key") or it.get("token")
|
||
if key:
|
||
return key
|
||
issued = await compute_client.issue_user_token(username, name="平台中继令牌")
|
||
key = issued.get("key") or issued.get("token") or ""
|
||
if not key:
|
||
raise HTTPException(status_code=502, detail="算力令牌签发失败，请联系运营方")
|
||
return key
|
||
|
||
|
||
async def _auth_headers(request: Request) -> dict[str, str]:
|
||
"""模型调用鉴权：把请求归到「真实用户令牌」，由引擎按该用户余额计量。
|
||
|
||
- 客户端携带引擎 PAT：直接透传（compute 校验令牌与余额，无效即 401）。
|
||
- 客户端携带平台 JWT（桌面端/网页端登录态）：解析用户 → 归户到其引擎 PAT。
|
||
- 二者皆无：**拒绝匿名**——不回落 COMPUTE_RELAY_TOKEN/COMPUTE_ADMIN_TOKEN，
|
||
否则「用户无余额仍能免费使用」。
|
||
"""
|
||
client_auth = (request.headers.get("Authorization", "") or "").strip()
|
||
if not client_auth:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail="缺少算力令牌：请登录后调用（或携带用户算力令牌 Bearer <PAT>）",
|
||
)
|
||
token = client_auth[7:].strip() if client_auth.lower().startswith("bearer ") else client_auth
|
||
if not token:
|
||
raise HTTPException(status_code=401, detail="算力令牌为空")
|
||
if _is_jwt(token):
|
||
try:
|
||
pat = await _user_engine_pat(token)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc: # noqa: BLE001
|
||
raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
|
||
return {"Authorization": f"Bearer {pat}"}
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
@router.post("/chat/completions")
|
||
async def relay_chat_completions(request: Request):
|
||
"""把 chat/completions 转发到 compute-engine，流式 SSE 原样透传。"""
|
||
try:
|
||
body: dict[str, Any] = await request.json()
|
||
except Exception as exc: # noqa: BLE001
|
||
raise HTTPException(status_code=400, detail="Invalid JSON body") from exc
|
||
|
||
chat_url = f"{config.COMPUTE_BASE_URL.rstrip('/')}/v1/chat/completions"
|
||
stream = bool(body.get("stream", False))
|
||
headers = {**(await _auth_headers(request)), "Content-Type": "application/json"}
|
||
|
||
client = httpx.AsyncClient(timeout=None)
|
||
upstream_request = client.build_request("POST", chat_url, json=body, headers=headers)
|
||
|
||
if not stream:
|
||
try:
|
||
upstream = await client.send(upstream_request)
|
||
payload = upstream.json() if upstream.content else None
|
||
except Exception: # noqa: BLE001
|
||
await client.aclose()
|
||
raise HTTPException(status_code=502, detail="Upstream relay failed") from None
|
||
await client.aclose()
|
||
return JSONResponse(status_code=upstream.status_code, content=payload)
|
||
|
||
try:
|
||
upstream = await client.send(upstream_request, stream=True)
|
||
except Exception: # noqa: BLE001
|
||
await client.aclose()
|
||
raise HTTPException(status_code=502, detail="Upstream relay failed") from None
|
||
|
||
resp_headers: dict[str, str] = {}
|
||
if "content-type" in upstream.headers:
|
||
resp_headers["content-type"] = upstream.headers["content-type"]
|
||
if "x-request-id" in upstream.headers:
|
||
resp_headers["x-request-id"] = upstream.headers["x-request-id"]
|
||
|
||
return StreamingResponse(
|
||
upstream.aiter_raw(),
|
||
status_code=upstream.status_code,
|
||
media_type="text/event-stream",
|
||
headers=resp_headers,
|
||
background=BackgroundTask(client.aclose),
|
||
)
|
||
|
||
|
||
@router.get("/models")
|
||
async def relay_models():
|
||
"""返回模型清单（= 算力中心 admin 模型，供 OpenAIProvider.fetch_models 使用）。"""
|
||
ids = await compute_catalog.model_ids()
|
||
return {
|
||
"object": "list",
|
||
"data": [{"id": m, "object": "model", "owned_by": "pineagents"} for m in ids],
|
||
}
|