games-development-ai/wg1/gen-worker/worker/_client.py

"""_client.py —— W-G1 L1 裸路模型客户端（复用 spike 代理旁路 + 裸 openai 客户端 + usage 读取）。

L1 纪律（spike VERDICT §3/§6）：裸 llm_client，不引 AgentScope（框架 token 膨胀+本地开销吃便宜档单价）。
唯一模型出口 = new-api OpenAI 兼容网关（BASE_URL/v1），绝不直连厂商 SDK；密钥只走环境变量/本地 .env，绝不入库。
"""

import os
import time
from pathlib import Path
from urllib.parse import urlparse

# ── 本地 .env 加载（wg1/gen-worker/.env，gitignored）：密钥外置，不落代码 ──
_ENV_PATH = Path(__file__).resolve().parent.parent / ".env"


def _load_env():
    """把 wg1/gen-worker/.env 里的 KEY=VAL 注入环境（已存在的不覆盖）。"""
    if not _ENV_PATH.exists():
        return
    for line in _ENV_PATH.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        os.environ.setdefault(k.strip(), v.strip())


_load_env()

# 网关地址（可被环境变量覆盖；默认 new-api Tailscale 网关 OpenAI 兼容路径）。
BASE_URL = os.environ.get("NEWAPI_BASE_URL", "http://100.64.0.8:3000/v1")
# L1 目标模型（可被 WG1_MODEL 覆盖；冒烟默认最便宜档 deepseek-v4-flash）。
DEFAULT_MODEL = os.environ.get("WG1_MODEL", "deepseek-v4-flash")

# ── 代理旁路（spike VERDICT §4-坑1，必须在 import openai 前）──
# 本机 clash 代理(HTTP(S)_PROXY=127.0.0.1:7897) 会把发往 100.64.0.8 的请求经代理转发→502；
# httpx/openai 默认 trust_env=True，须把网关 host 并入大小写两个 NO_PROXY 才直连。
_GW_HOST = urlparse(BASE_URL).hostname or "100.64.0.8"
for _k in ("NO_PROXY", "no_proxy"):
    _cur = os.environ.get(_k, "")
    if _GW_HOST not in _cur:
        os.environ[_k] = (_cur + "," + _GW_HOST) if _cur else _GW_HOST

import openai  # noqa: E402  —— 必须在代理旁路之后导入

# WU2 §3.7:回落全局 env key 时打 WARN;但 get_api_key 每次 chat 都会被调,故一进程只 WARN 一次防刷屏。
_ENV_FALLBACK_WARNED = False


class NewapiQuotaExhaustedError(RuntimeError):
    """new-api 网关按 per-user token 判额度耗尽(WU2 §3.9)。

    据 one-api 惯例:HTTP 402/403 或错误体含额度耗尽关键词。属**不可重试**干净失败——worker 据此回调
    failure_reason=quota_exhausted(而非含糊 llm_error);凭据失效(401)不塌成本错误、维持可重试 llm_error。
    【阶段2 待验】402/403 与确切错误体 message S0 只坐实管理面、未测 /v1/chat/completions 耗尽错误体,
    确切匹配待 S4 真机耗尽/凭据失效测试;现按 one-api 惯例初值分辨(状态码优先、关键词兜底)。
    """


def _mask_token(tok) -> str:
    """token 日志脱敏(WU2 §3.8):前后各留 4 位、中间省略;过短/空则整体隐藏。绝不整条打凭据。"""
    if not tok:
        return "<none>"
    s = str(tok)
    return "****" if len(s) <= 8 else f"{s[:4]}…{s[-4:]}"


def _newapi_error_status_text(exc):
    """从 new-api 调用异常尽力抽 (http_status, 错误体文本);抽不到 status 返 None,文本恒兜底 str(exc)。

    兼容 openai SDK 的 APIStatusError(.status_code / .response.status_code / .body / .message)与被上层
    框架包装后的普通异常(只剩 str)。供 classify_newapi_failure_reason 分辨。
    """
    status = getattr(exc, "status_code", None)
    if status is None:
        status = getattr(exc, "code", None)
    if status is None:
        resp = getattr(exc, "response", None)
        status = getattr(resp, "status_code", None) if resp is not None else None
    parts = []
    for attr in ("message", "body"):
        v = getattr(exc, attr, None)
        if v:
            parts.append(str(v))
    parts.append(str(exc))
    return status, " ".join(parts)


def classify_newapi_failure_reason(status=None, text: str = "") -> str:
    """据 one-api 惯例把网关非 2xx 分辨成失败因(WU2 §3.9)。

    返回 'quota_exhausted'(额度耗尽,不可重试)或 'llm_error'(凭据失效 401/其他,可重试、不塌成额度耗尽)。
    判据:HTTP 402/403 → 额度耗尽;401 → 凭据失效;状态码取不到时按错误体关键词兜底。
    【阶段2 待验】确切 message/code 需 S4 真机耗尽/凭据失效测试坐实;此处状态码优先、关键词兜底为初值。
    """
    try:
        s = int(status) if status is not None else None
    except (TypeError, ValueError):
        s = None
    if s in (402, 403):
        return "quota_exhausted"
    if s == 401:
        return "llm_error"  # 凭据失效:维持可重试,不塌成 quota_exhausted(§3.9 ②)
    t = (text or "").lower()
    quota_markers = ("insufficient user quota", "insufficient quota", "余额不足", "额度不足", "quota not enough")
    if any(m in t for m in quota_markers):
        return "quota_exhausted"
    return "llm_error"


def get_api_key(user_token=None):
    """解析本次调用的 new-api 凭据(WU2 §3.7 F)。

    优先用传入的 per-user token(user_token,随 §6.1 job 派来);缺失回落全局 env NEWAPI_KEY(缺失即抛,
    密钥不入库、复现前先设 .env 或 export)。回落是系统/编排/bake-off/spike 触发的正常预期(无 game_player
    成员身份、后端 dispatch 已按身份旁路不塞 token);额度门在后端 dispatch 层按身份堵,worker 层只做「有则用、
    无则回落」。回落时打一条 WARN(一进程一次,防每次 chat 刷屏)便于发现本该带 token 的 member create job 异常。
    """
    if user_token:
        return user_token
    key = os.environ.get("NEWAPI_KEY")
    if not key:
        raise RuntimeError("NEWAPI_KEY 未设：请在 wg1/gen-worker/.env 写入或 export NEWAPI_KEY=...")
    global _ENV_FALLBACK_WARNED
    if not _ENV_FALLBACK_WARNED:
        _ENV_FALLBACK_WARNED = True
        print(f"[wg1-worker] 无 per-user userToken → 回落全局 env NEWAPI_KEY({_mask_token(key)})。"
              f"系统/编排/bake-off/spike 为正常预期;若为真实 member create job 则异常走共享计量(§3.7)。",
              flush=True)
    return key


def get_client(user_token=None):
    """构建指向 new-api 的裸 OpenAI 兼容客户端(WU2 §3.7:凭据优先 per-user token、缺失回落 env)。"""
    return openai.OpenAI(api_key=get_api_key(user_token), base_url=BASE_URL, max_retries=2, timeout=180.0)


def _extra_body(model, enable_thinking):
    """按模型构造 extra_body：MiniMax 系走官方 thinking 控制（默认 disabled，关思维链防截断）；其他模型不带厂商专有参。"""
    if "minimax" in (model or "").lower():
        return {"thinking": {"type": "adaptive" if enable_thinking else "disabled"}}
    return {}


def _cache_hit_from_usage(usage):
    """从 openai 原始 usage 抽命中缓存的 prompt token（兼容两套字段取 max，U4/B9）。

    DeepSeek `usage.prompt_cache_hit_tokens` / MiniMax `usage.prompt_tokens_details.cached_tokens`，取 max；
    字段全缺/异常 → 0（按 miss 计，不报错）。逻辑与 cost.cache_hit_tokens 同口径，但本处只依赖 openai 原始对象、
    不反向 import cost（避免 _client→cost 循环依赖；cost 已 import _client）。
    """
    if usage is None:
        return None
    try:
        cands = []
        ds = getattr(usage, "prompt_cache_hit_tokens", None)
        if ds is not None:
            cands.append(int(ds))
        details = getattr(usage, "prompt_tokens_details", None)
        cached = getattr(details, "cached_tokens", None) if details is not None else None
        if cached is not None:
            cands.append(int(cached))
        return max(cands) if cands else 0
    except Exception:
        return 0


def chat(model, system, user, max_tokens=16000, temperature=0.0, tries=3, retry_delay=2.0,
         enable_thinking=False, user_token=None):
    """单次裸调用 chat.completions。确定性产出 temperature=0；瞬时网关错误(502 burst)重试。

    返回 dict：{content, prompt_tokens, completion_tokens, prompt_cache_hit_tokens, total_tokens, model, wall_s, id, finish_reason}。
    usage 字段用 openai 2.41.1 口径：prompt_tokens / completion_tokens（供成本折算与 token 取证）；
    prompt_cache_hit_tokens（U4/B9 新增，additive 键）= 命中缓存的 prompt token（两套字段取 max；缺则 0），供 cost.py 折缓存价。
    WU2 §3.7：user_token 给定则用它调网关(per-user 额度)、缺失回落全局 env key。
    WU2 §3.9：网关非 2xx 按 one-api 惯例分辨——额度耗尽(402/403)立即抛 NewapiQuotaExhaustedError 不再重试;
    凭据失效(401)/瞬时错(502)维持原重试语义(不塌成额度耗尽)。
    """
    client = get_client(user_token)
    last_err = None
    for attempt in range(tries):
        try:
            t0 = time.perf_counter()
            resp = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": system},
                    {"role": "user", "content": user},
                ],
                temperature=temperature,
                max_tokens=max_tokens,
                stream=False,
                # 关推理（默认）：MiniMax-M3 在 OpenAI 端点默认 thinking=adaptive(开)，思维链内联进 content，复杂局
                # 飙 token → max_tokens 处截断、永远到不了 JSON（finish_reason=length，表现为"parse 失败"，实测
                # match3/simon）。官方关法(MiniMax docs)= 顶层 thinking={type:disabled}；本 worker 只要结构化 JSON、
                # 不需 CoT，故对 MiniMax 系默认关（enable_thinking=True 可覆盖）。非 MiniMax 模型不带此厂商专有参。
                extra_body=_extra_body(model, enable_thinking),
            )
            wall = time.perf_counter() - t0
            choice = resp.choices[0]
            usage = resp.usage
            return {
                "content": choice.message.content or "",
                "finish_reason": getattr(choice, "finish_reason", None),
                "prompt_tokens": getattr(usage, "prompt_tokens", None),
                "completion_tokens": getattr(usage, "completion_tokens", None),
                # U4/B9：命中缓存的 prompt token（additive 键；旧 dict 消费方不受影响，新口径供 cost 折缓存价）。
                "prompt_cache_hit_tokens": _cache_hit_from_usage(usage),
                "total_tokens": getattr(usage, "total_tokens", None),
                "model": model,
                "wall_s": round(wall, 3),
                "id": getattr(resp, "id", None),
            }
        except Exception as e:  # 瞬时网关错误重试（spike 实测 new-api 偶发 502 burst）
            # WU2 §3.9:额度耗尽(402/403)重试无意义,立即抛典型错误让 worker 回调 quota_exhausted;
            #   凭据失效(401)/瞬时错(502)不塌成额度耗尽,维持下方重试语义(可重试 llm_error)。
            if classify_newapi_failure_reason(*_newapi_error_status_text(e)) == "quota_exhausted":
                raise NewapiQuotaExhaustedError(
                    f"new-api 额度耗尽(按 §3.9 分辨,阶段2 待精确 message)：{type(e).__name__}: {e}") from e
            last_err = e
            if attempt < tries - 1:
                time.sleep(retry_delay)
    raise RuntimeError(f"模型调用失败（{tries} 次重试后）：{last_err}")


if __name__ == "__main__":
    # 自测：列模型（零 token 成本）确认 客户端+代理旁路+密钥 三件齐活。
    c = get_client()
    ids = sorted(m.id for m in c.models.list().data)
    print(f"[_client] BASE_URL={BASE_URL}  NO_PROXY 含网关={_GW_HOST in os.environ.get('NO_PROXY','')}")
    print(f"[_client] 可用模型 {len(ids)} 个：")
    for i in ids:
        print("   ", i)