games-development-ai/cheap-worker/cheap_classify.py

"""cheap_classify.py — A11 切片三 M2 判意图(NL → 结构化建议改动 + 风险)。

两段式回路第一段:用户试玩后用自然语言提改 → 本模块据 base 源工程 + 原话,判出建议的
{category, mode, target, payload, riskLevel, needsConfirm, clarify},返前端供确认;确认后前端再调
已结构化的 /modify 执行(第二段)。本模块不执行、不建版本、不计费——只判意图。

落点(对齐生成侧工程规范性,创始人 2026-06-28):
  · 换资产 → assets.js 资产清单的一处(键→地址);
  · 调数值 → core.js 集中常量的一处;
  · 改玩法 → game-logic.js 模块重生成(intent 喂 LLM);
  · 大改/换品类/危险 → 高风险、回问用户确认;意图不明 → 回问澄清。

纯函数(build_classify_prompt / parse_classify_response)无网络无 LLM、可单测;classify 为集成路
(调便宜档 M3),complete_fn 可注入(单测桩/换模型)。
"""

import json
import re

# mode×category 锁定映射(可信边界:LLM 给的 mode 不可信,一律按 category 落 —— 对齐契约 StudioModifyReqVO)。
_DETERMINISTIC_CATEGORIES = ("asset", "config", "level")
_VALID_CATEGORIES = ("asset", "config", "level", "behavior", "big-change", "unclear")
_VALID_RISK = ("low", "medium", "high")

# 判意图给 LLM 看的可改面文件(资产清单 / 集中数值 / 玩法);其余 plumbing 文件不喂(判意图不需要)。
_SURFACE_FILES = ("src/assets.js", "src/core.js", "src/game-logic.js")


def _as_dict(source_project) -> dict:
    """source_project 容错成 dict(后端经 HTTP 传的是 JSON 串;也容忍已是 dict)。坏输入返 {}。"""
    if isinstance(source_project, dict):
        return source_project
    if isinstance(source_project, str) and source_project.strip():
        try:
            obj = json.loads(source_project)
            return obj if isinstance(obj, dict) else {}
        except Exception:  # noqa: BLE001 — 坏 JSON 容错返空,build_prompt 仍可只带原话
            return {}
    return {}


def build_classify_prompt(raw_text: str, source_project) -> str:
    """把 base 源的可改面(资产清单 / 集中数值 / 玩法 + 文件列表)与用户原话组成判意图 prompt。

    可改面对齐工程规范性:assets.js = 资产统一管理的一处、core.js = 数值集中的一处,喂给 LLM 以便它把
    改动定位到"那一处"。其余 L1 plumbing 文件不喂。
    """
    sp = _as_dict(source_project)
    files = sp.get("files") if isinstance(sp.get("files"), dict) else {}
    file_list = sorted(files.keys())

    surface_blocks = []
    for path in _SURFACE_FILES:
        content = files.get(path)
        if content:
            # 截断防超长(判意图只需看可改面结构,不需全文)。
            snippet = content if len(content) <= 1600 else content[:1600] + "\n…(截断)"
            surface_blocks.append(f"--- {path} ---\n{snippet}")
    surface_text = "\n\n".join(surface_blocks) if surface_blocks else "(无可改面文件)"

    return f"""你是游戏调整意图判断器。用户试玩了一款小游戏后,用自然语言提出修改。你的任务:判断这次修改属于哪一类、应落到哪一处,并评估风险。**只判断,不执行。**

【用户原话】
{raw_text}

【游戏文件列表】
{", ".join(file_list) if file_list else "(空)"}

【可改面(改动应落到这里的一处)】
{surface_text}

【分类规则】
- 换美术/贴图/配色/图标/角色外观 → category="asset"(改 assets.js 资产清单的一处键→地址,或配色常量);
- 调数值/难度/速度/时长/分数/节奏/数量 → category="config"(改 core.js 集中常量的一处);
- 改关卡/布局/场景排布 → category="level";
- 改玩法/规则/机制/胜负条件/逻辑 → category="behavior"(模块重生成 game-logic.js,intent 写改写意图);
- 换品类/整体重做/推倒重写/危险不可逆(清档/重置) → category="big-change"(超出局部调整);
- 意图不明/无法定位到具体一处 → category="unclear"(回问澄清)。

【输出】严格输出一个 JSON 对象(不要多余文字),字段:
{{"category": "asset|config|level|behavior|big-change|unclear",
  "target": {{"kind": "asset|config|level|behavior", "path": "落点(如 assets.js 的键名 / core.js 的常量名 / 文件路径)"}} 或 null,
  "payload": {{"value": {{"键": "新值"}}}}(确定性类) 或 {{"intent": "改写意图"}}(behavior) 或 null,
  "riskLevel": "low|medium|high",
  "rationale": "一句话判断依据(中文)",
  "clarify": "若 unclear/big-change,回问用户的话;否则 null"}}
"""


def _safe_fallback(reason: str) -> dict:
    """坏 JSON / 解析失败的安全兜底:判为 unclear + 需确认 + 不落任何可执行 mode(绝不静默判成可执行改动)。"""
    return {
        "category": "unclear", "mode": None, "target": None, "payload": None,
        "riskLevel": "medium", "needsConfirm": True,
        "rationale": f"未能解析判意图结果({reason})", "clarify": "没太理解你想改什么,能换种说法、或说明要改哪一处吗?",
    }


def _extract_json(text: str):
    """从 LLM 文本剥出 JSON 对象(剥 ```json 围栏 / 取首个 {…} 块);失败返 None。"""
    if not isinstance(text, str) or not text.strip():
        return None
    s = text.strip()
    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", s, re.DOTALL)
    if fence:
        s = fence.group(1)
    else:
        brace = re.search(r"\{.*\}", s, re.DOTALL)
        if brace:
            s = brace.group(0)
    try:
        obj = json.loads(s)
        return obj if isinstance(obj, dict) else None
    except Exception:  # noqa: BLE001
        return None


def parse_classify_response(text: str) -> dict:
    """解析 LLM 判意图输出 → 规范化建议改动 + 落 mode×category 锁定 + 风险/确认规则 + 坏 JSON 兜底。

    可信边界:LLM 给的 mode 不采信,一律按 category 落 mode(asset/config/level→deterministic、
    behavior→regenerate-module、big-change/unclear→None);needsConfirm 由本函数定(只有低风险确定性改
    才免确认),不交给 LLM——危险/大改/改玩法/意图不明一律回问(创始人 2026-06-28)。
    """
    obj = _extract_json(text)
    if obj is None:
        return _safe_fallback("非 JSON")

    category = obj.get("category")
    if category not in _VALID_CATEGORIES:
        return _safe_fallback(f"未知 category={category}")

    # mode 按 category 锁定(无视 LLM 给的 mode)。
    if category in _DETERMINISTIC_CATEGORIES:
        mode = "deterministic"
    elif category == "behavior":
        mode = "regenerate-module"
    else:  # big-change / unclear
        mode = None

    risk = obj.get("riskLevel")
    if risk not in _VALID_RISK:
        risk = "medium"
    if category == "big-change":
        risk = "high"  # 大改强制高风险

    # needsConfirm:只有"低风险 + 确定性类"才免确认;其余(改玩法/大改/意图不明/中高风险)一律回问。
    needs_confirm = not (mode == "deterministic" and risk == "low")

    clarify = obj.get("clarify")
    if not isinstance(clarify, str) or not clarify.strip():
        clarify = None
    if category in ("unclear", "big-change") and not clarify:
        clarify = "这处改动较大或意图不够明确,想确认一下你具体想怎么改?"

    return {
        "category": category,
        "mode": mode,
        "target": obj.get("target") if isinstance(obj.get("target"), dict) else None,
        "payload": obj.get("payload") if isinstance(obj.get("payload"), dict) else None,
        "riskLevel": risk,
        "needsConfirm": needs_confirm,
        "rationale": obj.get("rationale") if isinstance(obj.get("rationale"), str) else "",
        "clarify": clarify,
    }


def _default_complete(prompt: str, *, max_tokens: int = 1500) -> str:
    """默认 LLM 调用:openai 协议直调 new-api 网关(便宜档 M3,温度 0 求稳定分类)。集成路、非单测。

    复用框架代理旁路:load_framework() 触发 config import 把 new-api host 并入 NO_PROXY,故 openai
    客户端(httpx 读 NO_PROXY)对内网网关直连、不被系统代理 fake-ip 拦 502。base_url 取 host 根补 /v1
    (内网凭据文档 baseUrl 坑:Python openai 侧须带 /v1)。
    """
    import os

    import openai

    import _bootstrap

    fw = _bootstrap.load_framework()  # 装代理旁路 + ensure key(在 import openai 前已由 config 链装好)
    base = fw["client"].resolve_base_url().rstrip("/")
    if not base.endswith("/v1"):
        base = base + "/v1"
    oc = openai.OpenAI(base_url=base, api_key=os.environ["NEWAPI_KEY"])
    resp = oc.chat.completions.create(
        model=_bootstrap.SPIKE_MODEL, temperature=0, max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content or ""


def classify(raw_text: str, source_project, *, complete_fn=None) -> dict:
    """判意图集成入口:组 prompt → 调 LLM(complete_fn 可注入)→ 解析规范化。

    Args:
        raw_text: 用户自然语言改话。
        source_project: base 源工程(dict 或 2.0 JSON 串)。
        complete_fn: (prompt:str)->str 的 LLM 调用桩(默认 _default_complete);单测注入免真网络。

    Returns:
        规范化建议改动 dict(见 parse_classify_response);LLM 调用异常 → 安全兜底 unclear+需确认。
    """
    fn = complete_fn or _default_complete
    prompt = build_classify_prompt(raw_text, source_project)
    try:
        text = fn(prompt)
    except Exception as e:  # noqa: BLE001 — LLM 异常不外抛,兜底为需确认的 unclear(绝不静默判成可执行)
        return _safe_fallback(f"LLM 调用异常:{type(e).__name__}")
    return parse_classify_response(text)