#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Phase 10.1（RESEARCH_ROADMAP）：/v1/systemone 本地兼容服务（Jev 客户端可无缝切换）。 契约（对齐官方 typesafe_demo.py + jev_live_eval.py 实测行级输出）： POST /v1/systemone {"state": str, "model": str, "questions": {name: {"type": "noul"|"choice"|"score", "instructions": str, "criteria": ...}}} - choice: criteria = {id: description}（无序） - score: criteria = [description, ...]（由低到高有序等级） - noul: 无 criteria，instructions 即判据 响应 {"answers": {name: answer}, "model", "usage": {"input_tokens", "output_tokens"}, "latency_ms"} - noul: {"noul": p_yes, "confidence": max(p_yes, 1-p_yes), "probabilities": {"yes": p, "no": 1-p}} - choice: {"choice": id, "confidence": maxp, "probabilities": {id: p}} - score: {"score": E[rank]/(k-1) ∈ [0,1], "confidence": maxp, "probabilities": {description: p}} 实现：frozen Qwen3.5-4B + 三种打分模式 --mode direct SemIf direct-logit（末位 logits 限 A-P slot → softmax，enable_thinking=False 硬前提） --mode lora LoRA checkpoint（scale_lora_train/rlcd_train 产物）merge 进基座后走 direct-logit；最终模型 rlcd_score43k_v5.pt（Phase 11 选型，authored 0.958/perturb 0.972） --mode head AttentionHead checkpoint（scale_head_train 产物）+ 冻结编码器（KV-Cache 广播是 10.2，本版逐问题前向） confidence 一律为分布汇总值（无第二头，对齐 Jev 黑盒形态）。 CLI（远程 4090）：CUDA_VISIBLE_DEVICES=1 python systemone_server.py --model /home/chenzhimin/modelscope/models/Qwen--Qwen3.5-4B/snapshots/master --port 8100 --mode lora --lora-checkpoint /data20T/semif-results/rlcd_score43k_v5.pt 自测：python systemone_server.py --selftest ...（跑官方 demo 5 场景，不调外部 API） """ from __future__ import annotations import argparse import json import math import os import sys import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parent sys.path.insert(0, str(ROOT / "src")) sys.path.insert(0, str(Path(__file__).parent)) from semif_phase1.core import LETTERS, DIRECT_SYSTEM # noqa: E402 from toy_dual_head import load_torch # noqa: E402 import torch # noqa: E402 from torch import nn # noqa: E402 import torch.nn.functional as F # noqa: E402 LETTER_SLOTS_CACHE: dict[str, int] = {} def softmax(values: list[float]) -> list[float]: peak = max(values) weights = [math.exp(v - peak) for v in values] total = sum(weights) return [w / total for w in weights] class DirectScorer: """SemIf direct-logit：末位 logits 限 A-P slot → softmax。score_batch 为 10.2 优化路径（共享前缀 KV-Cache 广播）。""" def __init__(self, model, tokenizer, device) -> None: self.model = model self.tokenizer = tokenizer self.device = device def build_prompt(self, state: str, instructions: str, option_descriptions: list[str]) -> str: payload = {"evidence": state, "criterion": instructions, "options": [{"letter": LETTERS[i], "description": description} for i, description in enumerate(option_descriptions)]} return self.tokenizer.apply_chat_template([{"role": "system", "content": DIRECT_SYSTEM}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}], tokenize=False, add_generation_prompt=True, enable_thinking=False) def score(self, state: str, instructions: str, option_descriptions: list[str]) -> list[float]: prompt = self.build_prompt(state, instructions, option_descriptions) ids = self.tokenizer.encode(prompt, add_special_tokens=False) with torch.inference_mode(): inp = torch.tensor([ids], dtype=torch.long, device=self.device) logits = self.model(input_ids=inp).logits[0, -1].float().cpu() return self._slot_softmax(logits, option_descriptions) def _slot_softmax(self, logits: torch.Tensor, option_descriptions: list[str]) -> list[float]: slots = [] for i in range(len(option_descriptions)): letter = LETTERS[i] if letter not in LETTER_SLOTS_CACHE: token_ids = self.tokenizer.encode(letter, add_special_tokens=False) if len(token_ids) != 1: raise ValueError(f"slot {letter} not single token: {token_ids}") LETTER_SLOTS_CACHE[letter] = token_ids[0] slots.append(LETTER_SLOTS_CACHE[letter]) selected = [float(logits[slot]) for slot in slots] return softmax(selected) def score_batch(self, state: str, questions: list[tuple[str, list[str]]]) -> list[list[float]]: """10.2：同一 state 的多问题共享前缀——prefix 编码一次（线性态缓存），每问题恢复快照再前向后缀。失败自动回退逐条全量前向。""" if len(questions) <= 1 or not self._cache_supported(): return [self.score(state, instructions, option_descriptions) for instructions, option_descriptions in questions] prompts = [self.build_prompt(state, instructions, option_descriptions) for instructions, option_descriptions in questions] encoded = [self.tokenizer.encode(prompt, add_special_tokens=False) for prompt in prompts] common = 0 limit = min(len(ids) for ids in encoded) while common < limit and all(ids[common] == encoded[0][common] for ids in encoded): common += 1 if common < 32: return [self.score(state, instructions, option_descriptions) for instructions, option_descriptions in questions] try: results = [] prefix_ids = encoded[0][:common] with torch.inference_mode(): prefix_tensor = torch.tensor([prefix_ids], dtype=torch.long, device=self.device) cache = self._make_cache() self.model(input_ids=prefix_tensor, past_key_values=cache, use_cache=True) snapshot = self._snapshot(cache) for index, ids in enumerate(encoded): self._restore(cache, snapshot) suffix = torch.tensor([ids[common:]], dtype=torch.long, device=self.device) logits = self.model(input_ids=suffix, past_key_values=cache, use_cache=True).logits[0, -1].float().cpu() option_descriptions = questions[index][1] results.append(self._slot_softmax(logits, option_descriptions)) return results except Exception as error: print(f"[score_batch] cache path failed ({type(error).__name__}: {error}), falling back to full forward", flush=True) self._cache_ok = False return [self.score(state, instructions, option_descriptions) for instructions, option_descriptions in questions] def _cache_supported(self) -> bool: return bool(getattr(self, "_cache_ok", True)) def _make_cache(self): from transformers import DynamicCache config = getattr(self.model, "config", None) if config is not None: try: return DynamicCache(config=config) except Exception: pass return DynamicCache() def _snapshot(self, cache) -> list: snapshot = [] names = ("conv_states", "recurrent_states", "key_states", "value_states") for layer in cache.layers: snapshot.append(tuple((name, self._clone_state(getattr(layer, name, None))) for name in names)) return snapshot def _restore(self, cache, snapshot: list) -> None: for layer, entries in zip(cache.layers, snapshot): for name, state in entries: if state is not None: setattr(layer, name, self._clone_state(state)) def _clone_state(self, state): if state is None: return None if isinstance(state, torch.Tensor): return state.detach().clone() if isinstance(state, dict): return {key: self._clone_state(value) for key, value in state.items()} if isinstance(state, (list, tuple)): cloned = [self._clone_state(item) for item in state] return tuple(cloned) if isinstance(state, tuple) else cloned return state class HeadScorer: """AttentionHead（scale_head_train checkpoint）+ 冻结编码器。context = chat prompt 全 token hidden，option = 描述 mean-pool。""" def __init__(self, model, tokenizer, device, checkpoint: Path, rank: int = 64, context_tokens: int = 640, option_tokens: int = 64) -> None: from scale_head_train import AttentionHead # noqa: F401 — 同步实现复用 self.model = model self.tokenizer = tokenizer self.device = device self.context_tokens = context_tokens self.option_tokens = option_tokens payload = torch.load(checkpoint, map_location="cpu", weights_only=False) config = payload.get("config", {}) self.head = AttentionHead(self.resolve_hidden(), int(config.get("head_rank", rank))).to(device) self.head.load_state_dict(payload["state_dict"]) self.head.eval() def resolve_hidden(self) -> int: hidden = getattr(self.model.config, "hidden_size", None) if hidden is None and hasattr(self.model.config, "get_text_config"): hidden = self.model.config.get_text_config().hidden_size if hidden is None: raise ValueError(f"cannot resolve hidden_size from config {type(self.model.config)}") return int(hidden) def encode_context(self, prompt: str) -> tuple[torch.Tensor, torch.Tensor]: ids = self.tokenizer.encode(prompt, add_special_tokens=False)[-self.context_tokens:] mask = torch.ones((1, len(ids)), dtype=torch.bool, device=self.device) with torch.inference_mode(): inp = torch.tensor([ids], dtype=torch.long, device=self.device) hidden = self.model.model(input_ids=inp).last_hidden_state.to(torch.float16) return hidden, mask def encode_options(self, descriptions: list[str]) -> torch.Tensor: vectors = [] with torch.inference_mode(): for description in descriptions: ids = self.tokenizer.encode(description, add_special_tokens=True)[: self.option_tokens] if not ids: raise ValueError("empty option tokens") inp = torch.tensor([ids], dtype=torch.long, device=self.device) hidden = self.model.model(input_ids=inp).last_hidden_state.float() vectors.append(hidden[0].mean(dim=0)) return torch.stack(vectors).to(torch.float16).unsqueeze(0) def score(self, state: str, instructions: str, option_descriptions: list[str]) -> list[float]: payload = {"evidence": state, "criterion": instructions, "options": [{"letter": LETTERS[i], "description": description} for i, description in enumerate(option_descriptions)]} prompt = self.tokenizer.apply_chat_template([{"role": "system", "content": DIRECT_SYSTEM}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}], tokenize=False, add_generation_prompt=True, enable_thinking=False) context, context_mask = self.encode_context(prompt) options = self.encode_options(option_descriptions) option_mask = torch.zeros((1, options.shape[1]), dtype=torch.bool, device=self.device) option_mask[0, :] = True with torch.inference_mode(): logits = self.head(context, context_mask, options, option_mask)[0] probs = F.softmax(logits, dim=-1).cpu().numpy().tolist() return probs NOUL_YES = "The criterion holds" NOUL_NO = "The criterion does not hold" def question_spec(question: dict) -> tuple[str, list[str]]: """把 Jev 三类型问题归一为 (instructions, option_descriptions)。""" question_type = question.get("type") instructions = str(question.get("instructions", "")).strip() if not instructions: raise ValueError("missing instructions") if question_type == "noul": return instructions, [NOUL_YES, NOUL_NO] if question_type == "choice": criteria = question.get("criteria") if not isinstance(criteria, dict) or not criteria: raise ValueError("choice requires non-empty criteria dict") return instructions, [str(value) for value in criteria.values()] if question_type == "score": criteria = question.get("criteria") if not isinstance(criteria, list) or len(criteria) < 2: raise ValueError("score requires ordered criteria list (>=2 levels)") return instructions, [str(item) for item in criteria] raise ValueError(f"unknown type {question_type}") def answer_request(scorer, state: str, questions: dict) -> dict: """请求级入口：同 state 多问题走 score_batch（10.2 共享前缀 KV-Cache 广播）。""" names = list(questions.keys()) specs = [question_spec(questions[name]) for name in names] if hasattr(scorer, "score_batch"): prob_lists = scorer.score_batch(state, specs) else: prob_lists = [scorer.score(state, instructions, option_descriptions) for instructions, option_descriptions in specs] answers = {} for name, question, (instructions, _), probs in zip(names, [questions[n] for n in names], specs, prob_lists): answers[name] = format_answer(question, probs) return answers def format_answer(question: dict, probs: list[float]) -> dict: question_type = question.get("type") if question_type == "noul": p_yes = float(probs[0]) return {"noul": p_yes, "confidence": max(p_yes, 1.0 - p_yes), "probabilities": {"yes": p_yes, "no": 1.0 - p_yes}} if question_type == "choice": criteria = question["criteria"] ids = list(criteria.keys()) best = int(np.argmax(probs)) return {"choice": ids[best], "confidence": float(probs[best]), "probabilities": {key: float(p) for key, p in zip(ids, probs)}} if question_type == "score": criteria = question["criteria"] k = len(criteria) expectation = sum(p * (i / (k - 1)) for i, p in enumerate(probs)) return {"score": float(expectation), "confidence": float(max(probs)), "probabilities": {str(item): float(p) for item, p in zip(criteria, probs)}} raise ValueError(f"unknown type {question_type}") def answer_question(scorer, state: str, name: str, question: dict) -> dict: question_type = question.get("type") instructions = str(question.get("instructions", "")).strip() if not instructions: raise ValueError(f"question {name} missing instructions") if question_type == "noul": probs = scorer.score(state, instructions, [NOUL_YES, NOUL_NO]) p_yes = float(probs[0]) return {"noul": p_yes, "confidence": max(p_yes, 1.0 - p_yes), "probabilities": {"yes": p_yes, "no": 1.0 - p_yes}} if question_type == "choice": criteria = question.get("criteria") if not isinstance(criteria, dict) or not criteria: raise ValueError(f"question {name} choice requires non-empty criteria dict") ids = list(criteria.keys()) descriptions = [str(criteria[key]) for key in ids] probs = scorer.score(state, instructions, descriptions) best = int(np.argmax(probs)) return {"choice": ids[best], "confidence": float(probs[best]), "probabilities": {key: float(p) for key, p in zip(ids, probs)}} if question_type == "score": criteria = question.get("criteria") if not isinstance(criteria, list) or len(criteria) < 2: raise ValueError(f"question {name} score requires ordered criteria list (>=2 levels)") descriptions = [str(item) for item in criteria] probs = scorer.score(state, instructions, descriptions) k = len(descriptions) expectation = sum(p * (i / (k - 1)) for i, p in enumerate(probs)) return {"score": float(expectation), "confidence": float(max(probs)), "probabilities": {description: float(p) for description, p in zip(descriptions, probs)}} raise ValueError(f"question {name} unknown type {question_type}") SELFTEST_SCENARIOS = [ {"name": "support ticket routing", "state": "Hi, I've been trying to connect my Stripe account for 3 days and the integration keeps failing. I'm losing sales. Please help ASAP.", "questions": {"is_urgent": {"type": "noul", "instructions": "The message conveys urgency or time-sensitivity"}, "department": {"type": "choice", "instructions": "Which team should handle this", "criteria": {"billing": "Payment or subscription issues", "technical": "Bugs or integration problems", "sales": "Pricing or account questions"}}, "frustration": {"type": "score", "instructions": "How frustrated the customer appears", "criteria": ["Calm, just stating facts", "Frustrated but civil", "Very angry, strong language"]}}}, {"name": "financial news risk", "state": "The Federal Reserve signaled it will hold interest rates steady this quarter, while inflation remains above target. Tech stocks fell 2% in pre-market trading.", "questions": {"is_bearish": {"type": "noul", "instructions": "The headline is overall negative for equity markets"}, "asset_class": {"type": "choice", "instructions": "Which asset class is most directly affected", "criteria": {"equities": "Stocks", "rates": "Bonds / interest rates", "crypto": "Cryptocurrencies", "fx": "Currencies"}}, "impact": {"type": "score", "instructions": "How large is the expected market impact", "criteria": ["Negligible", "Moderate", "Major"]}}}, {"name": "phishing email", "state": "Dear Customer, your account will be suspended in 24 hours unless you verify your password immediately by clicking this link: http://secure-login.xyz/verify", "questions": {"is_spam": {"type": "noul", "instructions": "This is unsolicited bulk or promotional email"}, "is_phishing": {"type": "noul", "instructions": "This email tries to trick the user into revealing credentials or clicking a malicious link"}, "risk": {"type": "choice", "instructions": "The primary risk category", "criteria": {"credential_theft": "Asks for login credentials or passwords", "malware": "Encourages downloading or running something", "scam": "Asks for money or personal data", "benign": "No clear malicious intent"}}}}, {"name": "code review", "state": "def divide(a, b):\n return a / b\n\n# used as: divide(10, 0)", "questions": {"has_bug": {"type": "noul", "instructions": "The code has a correctness bug that can crash or produce wrong results"}, "bug_type": {"type": "choice", "instructions": "The main defect, if any", "criteria": {"division_by_zero": "Does not guard against divide-by-zero", "type_error": "Wrong types or misuse", "logic_error": "Wrong logic / off-by-one", "none": "No clear defect"}}, "readability": {"type": "score", "instructions": "How readable and well-structured the code is", "criteria": ["Poor", "Okay", "Clean"]}}}, {"name": "review scoring", "state": "The laptop is fast and the screen is gorgeous, but the battery only lasts 3 hours and the price is way over what I expected. Would buy it again for the performance.", "questions": {"performance": {"type": "score", "instructions": "How well performance is reviewed", "criteria": ["Poor", "Average", "Excellent"]}, "battery": {"type": "score", "instructions": "How well battery life is reviewed", "criteria": ["Poor", "Average", "Excellent"]}, "value_for_money": {"type": "score", "instructions": "How well price/value is reviewed", "criteria": ["Poor", "Average", "Excellent"]}, "would_recommend": {"type": "noul", "instructions": "The reviewer would recommend this product"}}}, ] def run_selftest(scorer, tokenizer) -> None: for scenario in SELFTEST_SCENARIOS: started = time.time() answers = answer_request(scorer, scenario["state"], scenario["questions"]) latency = round((time.time() - started) * 1000, 1) print("=" * 70) print(f"【{scenario['name']}】latency={latency}ms") print(json.dumps(answers, ensure_ascii=False, indent=2)) class Handler(BaseHTTPRequestHandler): scorer = None tokenizer = None model_name = "semif-systemone" def do_POST(self): if self.path != "/v1/systemone": self.send_json(404, {"error": f"unknown path {self.path}"}) return try: length = int(self.headers.get("Content-Length", "0")) body = json.loads(self.rfile.read(length).decode("utf-8")) state = str(body.get("state", "")) questions = body.get("questions") if not state or not isinstance(questions, dict) or not questions: raise ValueError("body requires state and questions") started = time.time() answers = answer_request(self.scorer, state, questions) latency = round((time.time() - started) * 1000, 1) input_tokens = sum(len(self.tokenizer.encode(json.dumps({"state": state, "question": question}, ensure_ascii=False), add_special_tokens=False)) for question in questions.values()) self.send_json(200, {"answers": answers, "model": self.model_name, "usage": {"input_tokens": input_tokens, "output_tokens": 0}, "latency_ms": latency}) except Exception as error: self.send_json(400, {"error": f"{type(error).__name__}: {error}"}) def send_json(self, status: int, payload: dict) -> None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, format, *args): print(f"[{time.strftime('%H:%M:%S')}] {args[0]}", flush=True) def load_lora_checkpoint(model, checkpoint: Path): """把 scale_lora_train 保存的 lora_state merge 进基座（merge_and_unload 后即普通模型）。""" from peft import LoraConfig, get_peft_model payload = torch.load(checkpoint, map_location="cpu", weights_only=False) lora_state = payload["lora_state"] config = payload.get("config", {}) rank = int(config.get("rank", 32)) lora_config = LoraConfig(r=rank, lora_alpha=rank * 2, target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], lora_dropout=0.0, bias="none", task_type="CAUSAL_LM") wrapped = get_peft_model(model, lora_config) wrapped.load_state_dict(lora_state, strict=False) current = wrapped.state_dict() matched = sum(1 for name, tensor in lora_state.items() if name in current and current[name].shape == tensor.shape) print(f"lora checkpoint {checkpoint}: {len(lora_state)} tensors, matched={matched}", flush=True) if matched < len(lora_state) * 0.9: raise ValueError(f"lora state mismatch: matched {matched}/{len(lora_state)}") wrapped.eval() return wrapped.merge_and_unload() def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model", type=str, default=os.path.expanduser("~/modelscope/models/Qwen--Qwen3.5-4B/snapshots/master")) parser.add_argument("--port", type=int, default=8100) parser.add_argument("--mode", choices=("direct", "lora", "head"), default="direct") parser.add_argument("--lora-checkpoint", type=Path, default=None) parser.add_argument("--head-checkpoint", type=Path, default=None) parser.add_argument("--selftest", action="store_true") args = parser.parse_args() device = torch.device("cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu")) print(f"loading {args.model} on {device} (mode={args.mode})", flush=True) model, tokenizer, _ = load_torch(args.model) if device.type == "cuda": model = model.to("cuda:0") if args.mode == "lora": if args.lora_checkpoint is None: parser.error("--mode lora requires --lora-checkpoint (scale_lora_train 产物)") model = load_lora_checkpoint(model, args.lora_checkpoint) scorer = DirectScorer(model, tokenizer, device) Handler.model_name = "semif-systemone-lora" elif args.mode == "head": if args.head_checkpoint is None: parser.error("--mode head requires --head-checkpoint (scale_head_train 产物)") scorer = HeadScorer(model, tokenizer, device, args.head_checkpoint) Handler.model_name = "semif-systemone-head" else: scorer = DirectScorer(model, tokenizer, device) Handler.scorer = scorer Handler.tokenizer = tokenizer if args.selftest: run_selftest(scorer, tokenizer) return server = ThreadingHTTPServer(("0.0.0.0", args.port), Handler) print(f"listening on 0.0.0.0:{args.port} (POST /v1/systemone)", flush=True) server.serve_forever() if __name__ == "__main__": main()