# -*- coding: utf-8 -*- """내 파인튜닝 모델을 격투장에 붙이는 제일 짧은 길. 파이썬으로 파인튜닝했다면 이미 이런 코드를 쓰고 있을 겁니다: tok = AutoTokenizer.from_pretrained("./my-model") model = AutoModelForCausalLM.from_pretrained("./my-model") out = model.generate(**tok(prompt, return_tensors="pt")) 이 파일은 거기에 껍데기만 씌운 것입니다. 새로 배울 게 없습니다 - 같은 코드가 HTTP 뒤에 앉아 있을 뿐이고, 그러면 격투장 클라이언트가 알아서 찾아냅니다. pip install transformers torch # 이미 있다면 건너뛰기 python serve_my_model.py ./my-model # 켜두기 ColosseumPet.exe # 딴 창에서. 알아서 붙습니다. vLLM을 이미 쓰고 있다면 이 파일은 필요 없습니다. `vllm serve ./my-model` 쪽이 훨씬 빠릅니다. 이건 "그건 또 뭘 깔라는 거야" 소리가 나올 때 쓰는 길입니다. 가중치는 이 컴퓨터를 안 떠납니다. 격투장에 가는 건 점수와 수순뿐입니다. """ import argparse import json import sys import threading from urllib.parse import urlsplit from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # 한국어 윈도우 콘솔은 cp949 라서 이 파일의 안내문에 쓰인 문자를 못 찍고 # UnicodeEncodeError 를 낸다. 그러면 모델을 다 불러온 뒤에 인사말 한 줄 # 때문에 서버가 죽는다 - 실제로 그렇게 죽었다. 못 찍는 글자는 물음표로 # 흘려보내고 서버는 계속 살아 있게 한다. for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(errors="replace") except Exception: pass # 답은 숫자 하나지만 추론 모델은 그 앞에 생각을 적는다. 좁게 잡으면 # 생각하다 잘려서 답이 아예 안 나온다. 그래서 요청이 정한 max_tokens 를 따른다 # (격투장 exe·킷은 16384 를 보낸다). 요청에 없으면 예전 값 1024, 많아도 # 16384 에서 자른다 - 고장 난 요청 하나가 GPU 를 오래 붙잡지 않게. MAX_NEW_TOKENS = 1024 # 요청에 max_tokens 가 없을 때 MAX_NEW_TOKENS_CAP = 16384 # 요청이 더 달라고 해도 여기까지 def _max_new_tokens(req) -> int: for key in ("max_tokens", "max_completion_tokens"): try: n = int(req.get(key) or 0) except (TypeError, ValueError): n = 0 if n > 0: return min(n, MAX_NEW_TOKENS_CAP) return MAX_NEW_TOKENS _lock = threading.Lock() # 모델은 한 번에 하나씩. 동시 호출은 줄을 섭니다. _state = {"tok": None, "model": None, "name": "", "spec": {}} def load(path: str, device: str, dtype: str): """평소 쓰던 그 두 줄. 여기서만 부릅니다.""" from transformers import AutoModelForCausalLM, AutoTokenizer import torch print(f"[모델] 불러오는 중: {path}") tok = AutoTokenizer.from_pretrained(path) kw = {} if dtype != "auto": kw["torch_dtype"] = getattr(torch, dtype) else: kw["torch_dtype"] = "auto" # device_map="auto" 는 accelerate 를 요구한다. 파인튜닝은 했어도 # accelerate 는 안 깔린 경우가 흔해서, 있으면 쓰고 없으면 평범하게 올린다. # 없다고 여기서 막히면 "그럼 또 뭘 깔라는 거야" 가 된다. have_accelerate = False if device == "auto": try: import accelerate # noqa: F401 have_accelerate = True except ImportError: print("[모델] accelerate 가 없어 단순 방식으로 올립니다") if device == "auto" and have_accelerate: kw["device_map"] = "auto" model = AutoModelForCausalLM.from_pretrained(path, **kw) if device != "auto": model = model.to(device) elif not have_accelerate: model = model.to("cuda" if torch.cuda.is_available() else "cpu") model.eval() # 체급 판정에 쓰이는 제원. 이름이 아니라 실제 파라미터 수를 셉니다 - # 격투장은 체급을 이름으로 나누지 않습니다. cfg = model.config _state["spec"] = { "params": int(sum(p.numel() for p in model.parameters())), "architecture": (getattr(cfg, "model_type", "") or "").lower(), "blocks": int(getattr(cfg, "num_hidden_layers", 0) or 0), "embedding": int(getattr(cfg, "hidden_size", 0) or 0), "feed_forward": int(getattr(cfg, "intermediate_size", 0) or 0), "head_count": int(getattr(cfg, "num_attention_heads", 0) or 0), "head_count_kv": int(getattr(cfg, "num_key_value_heads", 0) or getattr(cfg, "num_attention_heads", 0) or 0), } hs = _state["spec"]["embedding"] hc = _state["spec"]["head_count"] if hs and hc: _state["spec"]["key_length"] = hs // hc _state["spec"]["value_length"] = hs // hc _state["tok"], _state["model"] = tok, model print("[모델] 준비 완료 - %.2fB (%s, %d층)" % ( _state["spec"]["params"] / 1e9, _state["spec"]["architecture"] or "?", _state["spec"]["blocks"])) def generate(messages, max_new_tokens=MAX_NEW_TOKENS): """메시지 한 벌을 받아 (답, 끝난 이유). 끝난 이유는 stop 또는 length.""" import torch tok, model = _state["tok"], _state["model"] # 채팅 템플릿이 있으면 그걸 씁니다. 파인튜닝한 형식을 존중하는 게 맞습니다. if getattr(tok, "chat_template", None): text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) else: text = "\n".join(m.get("content", "") for m in messages) inputs = tok(text, return_tensors="pt").to(model.device) with _lock, torch.no_grad(): out = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False, pad_token_id=tok.eos_token_id) new = out[0][inputs["input_ids"].shape[1]:] # 한도까지 다 썼으면 생각하다 잘린 것이다. OpenAI 처럼 length 로 알려준다. reason = "length" if len(new) >= max_new_tokens else "stop" return tok.decode(new, skip_special_tokens=True).strip(), reason class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *a): pass def _send(self, obj, status=200): body = json.dumps(obj).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): # ?model= 같은 질의 문자열은 떼고 본다. 새 exe·킷은 제원을 # /v1/model_spec?model=<이름> 으로 먼저 묻는다. path = urlsplit(self.path).path.rstrip("/") # 클라이언트가 서버를 찾을 때 두드리는 곳입니다. if path in ("/v1/models", "/models"): return self._send({"object": "list", "data": [ {"id": _state["name"], "object": "model", "owned_by": "local"}]}) # 격투장 클라이언트가 체급 판정에 쓰는 제원. OpenAI 규격에는 없는 # 우리 확장이고, 없으면 그냥 "체급 미확인"이 됩니다. if path in ("/v1/model_spec", "/model_spec"): return self._send(dict(_state["spec"], name=_state["name"])) return self._send({"error": "not found"}, 404) def do_POST(self): if urlsplit(self.path).path.rstrip("/") not in ("/v1/chat/completions", "/chat/completions"): return self._send({"error": "not found"}, 404) try: n = int(self.headers.get("Content-Length") or 0) req = json.loads(self.rfile.read(n) or b"{}") except Exception: return self._send({"error": "bad json"}, 400) try: reply, reason = generate(req.get("messages") or [], _max_new_tokens(req)) except Exception as e: # 모델이 터져도 서버는 산다 return self._send({"error": str(e)[:200]}, 500) return self._send({ "id": "chatcmpl-local", "object": "chat.completion", "model": _state["name"], "choices": [{"index": 0, "finish_reason": reason, "message": {"role": "assistant", "content": reply}}], }) def main() -> int: ap = argparse.ArgumentParser( description="파인튜닝한 모델을 OpenAI 호환 서버로 띄웁니다.") ap.add_argument("model", help="모델 폴더 또는 허브 이름 (예: ./my-tuned-8b)") ap.add_argument("--port", type=int, default=8000, help="기본 8000. 격투장 클라이언트가 여기부터 찾습니다.") ap.add_argument("--device", default="auto", help="auto | cuda | cpu") ap.add_argument("--dtype", default="auto", help="auto | float16 | bfloat16 | float32") ap.add_argument("--name", default="", help="점수판에 뜰 이름 (기본: 폴더 이름)") args = ap.parse_args() _state["name"] = args.name or args.model.rstrip("/\\").split("/")[-1].split("\\")[-1] try: load(args.model, args.device, args.dtype) except ImportError: print("\n[오류] transformers 와 torch 가 필요합니다:\n") print(" pip install transformers torch\n") return 2 except Exception as e: print(f"\n[오류] 모델을 못 불러왔습니다: {e}\n") return 2 srv = ThreadingHTTPServer(("127.0.0.1", args.port), Handler) srv.daemon_threads = True print() print("=" * 52) print(f" 준비됐습니다 - http://localhost:{args.port}/v1") print(f" 모델 이름: {_state['name']}") print() print(" 이 창은 켜둔 채로, 다른 창에서 ColosseumPet 를 실행하세요.") print(" 클라이언트가 이 주소를 알아서 찾습니다.") print(" 끄기: Ctrl+C") print("=" * 52) try: srv.serve_forever() except KeyboardInterrupt: print("\n종료.") return 0 if __name__ == "__main__": sys.exit(main())