| |
| |
| """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 |
| from toy_dual_head import load_torch |
|
|
| import torch |
| from torch import nn |
| import torch.nn.functional as F |
|
|
| 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 |
| 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() |
|
|
