Initial Gradio app for HF Space

app.py

@@ -0,0 +1,45 @@
+import gradio as gr
+from inference import run_inference, reload_model  # reload_model은 모델 재로딩 함수
+from utils_prompt import build_webtest_prompt
+
+def gradio_infer(npc_id, npc_location, player_utt):
+    prompt = build_webtest_prompt(npc_id, npc_location, player_utt)
+    result = run_inference(prompt)
+    return result["npc_output_text"], result["deltas"], result["flags_prob"]
+
+
+# API 호출용
+def api_infer(session_id, npc_id, prompt, max_tokens=200):
+    result = run_inference(prompt)
+    return {
+        "session_id": session_id,
+        "npc_id": npc_id,
+        "npc_response": result["npc_output_text"],
+        "deltas": result["deltas"],
+        "flags": result["flags_prob"],
+        "thresholds": result["flags_thr"]
+    }
+
+# Colab에서 호출할 ping endpoint
+def ping_reload():
+    reload_model(branch="latest")  # latest 브랜치에서 재다운로드 & 로드
+    return {"status": "reloaded"}
+
+with gr.Blocks() as demo:
+    gr.Markdown("## NPC Main Model Inference")
+
+    with gr.Tab("Web Test UI"):
+        npc_id = gr.Textbox(label="NPC ID")
+        npc_loc = gr.Textbox(label="NPC Location")
+        player_utt = gr.Textbox(label="Player Utterance")
+        npc_resp = gr.Textbox(label="NPC Response")
+        deltas = gr.JSON(label="Deltas")
+        flags = gr.JSON(label="Flags Probabilities")
+        btn = gr.Button("Run Inference")
+        btn.click(fn=gradio_infer, inputs=[npc_id, npc_loc, player_utt], outputs=[npc_resp, deltas, flags])
+
+    demo.add_api_route("/predict_main", api_infer, methods=["POST"], api_name="predict_main")
+    demo.add_api_route("/ping_reload", lambda: ping_reload(), methods=["POST"], api_name="ping_reload")
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

config.py

@@ -0,0 +1,17 @@
+import torch
+
+# 모델 경로
+BASE_MODEL = "Qwen/Qwen2.5-3B-Instruct"
+ADAPTER_MODEL = "m97j/npc-LoRA-fps"
+
+# 장치 설정
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+# 토크나이저/모델 공통
+MAX_LENGTH = 1024
+NUM_FLAGS = 7  # flags.json 길이와 일치
+
+# 생성 파라미터
+GEN_MAX_NEW_TOKENS = 200
+GEN_TEMPERATURE = 0.7
+GEN_TOP_P = 0.9

flags.json

@@ -0,0 +1,11 @@
+{
+  "ALL_FLAGS": [
+    "give_item",
+    "end_npc_main_story",
+    "quest_stage_change",
+    "change_game_state",
+    "change_player_state",
+    "npc_action",
+    "unlock_hidden_path"
+  ]
+}

inference.py

@@ -0,0 +1,59 @@
+import torch
+from config import DEVICE, MAX_LENGTH, GEN_MAX_NEW_TOKENS, GEN_TEMPERATURE, GEN_TOP_P
+from model_loader import ModelWrapper
+
+# 전역 로드 (서버 시작 시 1회)
+wrapper = ModelWrapper()
+tokenizer, model, flags_order = wrapper.get()
+
+GEN_PARAMS = {
+    "max_new_tokens": GEN_MAX_NEW_TOKENS,
+    "temperature": GEN_TEMPERATURE,
+    "top_p": GEN_TOP_P,
+    "do_sample": True,
+    "repetition_penalty": 1.05,
+}
+
+def run_inference(prompt: str):
+    inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=MAX_LENGTH).to(DEVICE)
+
+    with torch.no_grad():
+        gen_ids = model.generate(**inputs, **GEN_PARAMS)
+        generated_text = tokenizer.decode(
+            gen_ids[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True
+        )
+
+        outputs = model(**inputs, output_hidden_states=True)
+        h = outputs.hidden_states[-1]
+
+        STATE_ID = tokenizer.convert_tokens_to_ids("<STATE>")
+        ids = inputs["input_ids"]
+        mask = (ids == STATE_ID).unsqueeze(-1)
+        if mask.any():
+            counts = mask.sum(dim=1).clamp_min(1)
+            pooled = (h * mask).sum(dim=1) / counts
+        else:
+            pooled = h[:, -1, :]
+
+        delta_pred = torch.tanh(model.delta_head(pooled))[0].cpu().tolist()
+        flag_prob = torch.sigmoid(model.flag_head(pooled))[0].cpu().tolist()
+        flag_thr = torch.sigmoid(model.flag_threshold_head(pooled))[0].cpu().tolist()
+
+    flags_prob_dict = {name: round(prob, 6) for name, prob in zip(flags_order, flag_prob)}
+    flags_thr_dict = {name: round(thr, 6) for name, thr in zip(flags_order, flag_thr)}
+
+    return {
+        "npc_output_text": generated_text.strip(),
+        "deltas": {
+            "trust": float(delta_pred[0]),
+            "relationship": float(delta_pred[1]),
+        },
+        "flags_prob": flags_prob_dict,
+        "flags_thr": flags_thr_dict,
+    }
+
+def reload_model(branch="latest"):
+    global wrapper, tokenizer, model, flags_order
+    wrapper = ModelWrapper(branch=branch)  # branch 인자로 latest 전달
+    tokenizer, model, flags_order = wrapper.get()
+    print(f"Model reloaded from branch: {branch}")

model_loader.py

@@ -0,0 +1,43 @@
+import os, json, torch
+import torch.nn as nn
+from transformers import AutoTokenizer, AutoModelForCausalLM
+from peft import PeftModel
+from config import BASE_MODEL, ADAPTER_MODEL, DEVICE
+
+def get_current_branch():
+    if os.path.exists("current_branch.txt"):
+        with open("current_branch.txt", "r") as f:
+            return f.read().strip()
+    return "latest"  # fallback
+
+class ModelWrapper:
+    def __init__(self):
+        flags_path = os.path.join(os.path.dirname(__file__), "flags.json")
+        self.flags_order = json.load(open(flags_path, encoding="utf-8"))["ALL_FLAGS"]
+        self.num_flags = len(self.flags_order)
+
+        self.tokenizer = AutoTokenizer.from_pretrained(ADAPTER_MODEL, use_fast=True)
+        if self.tokenizer.pad_token is None:
+            self.tokenizer.pad_token = self.tokenizer.eos_token
+        self.tokenizer.padding_side = "right"
+
+        branch = get_current_branch()
+        base = AutoModelForCausalLM.from_pretrained(BASE_MODEL, device_map="auto", trust_remote_code=True)
+        self.model = PeftModel.from_pretrained(base, ADAPTER_MODEL, revision=branch, device_map="auto")
+
+        hidden_size = self.model.config.hidden_size
+        self.model.delta_head = nn.Linear(hidden_size, 2).to(DEVICE)
+        self.model.flag_head = nn.Linear(hidden_size, self.num_flags).to(DEVICE)
+        self.model.flag_threshold_head = nn.Linear(hidden_size, self.num_flags).to(DEVICE)
+
+        if os.path.exists("delta_head.pt"):
+            self.model.delta_head.load_state_dict(torch.load("delta_head.pt", map_location=DEVICE))
+        if os.path.exists("flag_head.pt"):
+            self.model.flag_head.load_state_dict(torch.load("flag_head.pt", map_location=DEVICE))
+        if os.path.exists("flag_threshold_head.pt"):
+            self.model.flag_threshold_head.load_state_dict(torch.load("flag_threshold_head.pt", map_location=DEVICE))
+
+        self.model.eval()
+
+    def get(self):
+        return self.tokenizer, self.model, self.flags_order

readme.md

@@ -0,0 +1,257 @@
+---
+title: NPC Main Model Inference Server
+emoji: 🤖
+colorFrom: blue
+colorTo: pink
+sdk: gradio
+sdk_version: 4.0.0
+python_version: 3.10
+app_file: app.py
+---
+
+# NPC 메인 모델 추론 서버 (hf-serve)
+
+이 Space는 **NPC 대화 메인 모델**의 추론 API와 간단한 Gradio UI를 제공합니다.  
+Hugging Face Hub에 업로드된  
+[Base model](https://huggingface.co/Qwen/Qwen2.5-3B-Instruct)과
+[LoRA adapter model](https://huggingface.co/m97j/npc_LoRA-fps)을 로드하여,  
+플레이어 발화와 게임 상태를 기반으로 NPC의 응답, 감정 변화량(delta),
+플래그 확률/임계값을 예측합니다.
+
+---
+
+## 🚀 주요 기능
+- **API 엔드포인트** `/predict_main`  
+  - JSON payload로 prompt를 받아 모델 추론 결과 반환
+- **웹 UI** `/ui`  
+  - NPC ID, 위치, 플레이어 발화를 입력해 실시간 응답 확인
+- **커스텀 헤드 예측**  
+  - `delta_head`: trust / relationship 변화량
+  - `flag_head`: 각 flag의 확률
+  - `flag_threshold_head`: 각 flag의 임계값
+- **모델 실시간 업데이트**  
+  - Colab 학습 후 `latest` 브랜치 업로드 → `/ping_reload` 호출 시 즉시 재로드
+
+---
+
+## 📂 디렉토리 구조
+```
+hf-serve/
+ ├─ app.py             # Gradio UI + API 라우팅
+ ├─ inference.py       # 모델 추론 로직
+ ├─ model_loader.py    # 모델/토크나이저 로드
+ ├─ utils_prompt.py    # prompt 생성 함수
+ ├─ flags.json         # flag index → name 매핑
+ ├─ requirements.txt   # 의존성 패키지
+ └─ README.md          # (현재 문서)
+```
+
+---
+
+## ⚙️ 추론 로직 개요
+
+이 서버의 핵심은 `run_inference()` 함수로,  
+NPC 메인 모델에 프롬프트를 입력하고 응답·상태 변화를 예측하는 전 과정을 담당합니다.
+
+### 처리 흐름
+1. **프롬프트 토크나이즈**
+   - 입력된 prompt를 토크나이저로 변환하여 텐서 형태로 준비
+   - 길이 제한(`MAX_LENGTH`)과 디바이스(`DEVICE`) 설정 적용
+
+2. **언어모델 응답 생성**
+   - 사전 정의된 추론 파라미터(`GEN_PARAMS`)로 `model.generate()` 실행  
+     → NPC의 대사 텍스트 생성
+   - 생성된 토큰을 디코딩하여 최종 문자열로 변환
+
+3. **히든 상태 추출**
+   - `output_hidden_states=True`로 모델 실행
+   - 마지막 레이어의 hidden state를 가져옴
+
+4. **<STATE> 토큰 위치 풀링**
+   - `<STATE>` 토큰이 있는 위치의 hidden state를 평균(pooling)  
+     → NPC 상태를 대표하는 벡터로 사용
+   - 없을 경우 마지막 토큰의 hidden state 사용
+
+5. **커스텀 헤드 예측**
+   - `delta_head`: trust / relationship 변화량 예측
+   - `flag_head`: 각 flag의 발생 확률 예측
+   - `flag_threshold_head`: 각 flag의 임계값 예측
+
+6. **index → name 매핑**
+   - `flags.json`의 순서(`flags_order`)를 기반으로  
+     예측 벡터를 `{flag_name: 값}` 형태의 딕셔너리로 변환
+
+### 반환 형식
+```json
+{
+  "npc_output_text": "<NPC 응답>",
+  "deltas": { "trust": 0.xx, "relationship": 0.xx },
+  "flags_prob": { "flag_name": 확률, ... },
+  "flags_thr": { "flag_name": 임계값, ... }
+}
+```
+
+---
+
+## 📜 Prompt 포맷
+모델은 학습 시 아래와 같은 구조의 prompt를 사용합니다.
+
+```
+<SYS>
+NPC_ID={npc_id}
+NPC_LOCATION={npc_location}
+TAGS:
+ quest_stage={quest_stage}
+ relationship={relationship}
+ trust={trust}
+ npc_mood={npc_mood}
+ player_reputation={player_reputation}
+ style={style}
+</SYS>
+<RAG>
+LORE: ...
+DESCRIPTION: ...
+</RAG>
+<PLAYER_STATE>
+...
+</PLAYER_STATE>
+<CTX>
+...
+</CTX>
+<PLAYER>...
+<STATE>
+<NPC>
+```
+---
+
+## 💡 **일반적인 LLM 추론과의 차이점**  
+이 서버는 단순히 텍스트를 생성하는 것에 그치지 않고,  
+`<STATE>` 토큰 기반 상태 벡터를 추출하여 커스텀 헤드에서 **감정 변화량(delta)**과  
+**플래그 확률/임계값**을 동시에 예측합니다.  
+이를 통해 대사 생성과 게임 상태 업데이트를 **한 번의 추론으로 처리**할 수 있습니다.
+
+---
+
+## 🎯 추론 파라미터
+
+| 파라미터 | 의미 | 영향 |
+|----------|------|------|
+| `temperature` | 샘플링 온도 (0.0~1.0+) | 낮을수록 결정적(Deterministic), 높을수록 다양성 증가 |
+| `do_sample` | 샘플링 여부 | `False`면 Greedy/Beam Search, `True`면 확률 기반 샘플링 |
+| `max_new_tokens` | 새로 생성할 토큰 수 제한 | 응답 길이 제한 |
+| `top_p` | nucleus sampling 확률 누적 컷오프 | 다양성 제어 (0.9면 상위 90% 확률만 사용) |
+| `top_k` | 확률 상위 k개 토큰만 샘플링 | 다양성 제어 (50이면 상위 50개 후보만) |
+| `repetition_penalty` | 반복 억제 계수 | 1.0보다 크면 반복 줄임 |
+| `stop` / `eos_token_id` | 생성 중단 토큰 | 특정 문자열/토큰에서 멈춤 |
+| `presence_penalty` / `frequency_penalty` | 특정 토큰 등장 빈도 제어 | OpenAI 계열에서 주로 사용 |
+| `seed` | 난�� 시드 | 재현성 확보 |
+
+위 파라미터들은 **학습 시에는 사용되지 않고**,  
+모델이 응답을 생성하는 **추론 시점**에만 적용됩니다.
+
+
+
+## 💡 사용 예시
+
+- **결정적 분류/판정용**  
+  (예: `_llm_trigger_check` YES/NO)
+  ```python
+  temperature = 0.0
+  do_sample = False
+  max_new_tokens = 2
+  ```
+  → 항상 같은 입력에 같은 출력, 짧고 확정적인 답변 [ai_server/의 local fallback model에 특정 조건을 지시할 때 사용]
+
+- **자연스러운 대화/창작용**  
+  (예: main/fallback 대사 생성)
+  ```python
+  temperature = 0.7
+  top_p = 0.9
+  do_sample = True
+  repetition_penalty = 1.05
+  max_new_tokens = 200
+  ```
+  → 다양성과 자연스러움 확보 [main model 추론시에 사용]
+
+hf-serve에서는 자연스러운 대화/창작용의 파라미터 예를 그대로 사용했습니다.
+
+---
+
+## 🌐 API & UI 차이
+
+| 경로 | 입력 형식 | 내부 처리 |
+|------|-----------|-----------|
+| `/predict_main` | 완성된 prompt 문자열 | 그대로 추론 |
+| `/ui` | NPC ID, Location, Utterance | `build_webtest_prompt()`로 prompt 생성 후 추론 |
+
+---
+
+## 📌 API 사용 예시
+
+### 요청
+```json
+POST /api/predict_main
+{
+  "session_id": "abc123",
+  "npc_id": "mother_abandoned_factory",
+  "prompt": "<SYS>...<NPC>",
+  "max_tokens": 200
+}
+```
+
+### 응답
+```json
+{
+  "session_id": "abc123",
+  "npc_id": "mother_abandoned_factory",
+  "npc_response": "그건 정말 놀라운 이야기군요.",
+  "deltas": { "trust": 0.42, "relationship": -0.13 },
+  "flags": { "give_item": 0.87, "end_npc_main_story": 0.02 },
+  "thresholds": { "give_item": 0.65, "end_npc_main_story": 0.5 }
+}
+```
+
+---
+
+## 🔄 모델 업데이트 흐름
+1. Colab에서 학습 완료
+2. Hugging Face Hub `latest` 브랜치에 업로드
+3. Colab에서 `/api/ping_reload` 호출
+4. Space가 최신 모델 재다운로드 & 로드
+
+---
+
+## 🛠 실행 방법
+
+### 로컬 실행
+```bash
+git clone https://huggingface.co/spaces/m97j/PersonaChatEngine
+cd PersonaChatEngine
+pip install -r requirements.txt
+python app.py
+```
+
+### Hugging Face Space에서 실행
+- 웹 UI: `https://m97j-PersonaChatEngine.hf.space/ui`
+- API: `POST https://m97j-PersonaChatEngine.hf.space/api/predict_main`
+
+---
+
+## 🛠 실행 환경
+- Python 3.10
+- FastAPI, Gradio, Transformers, PEFT, Torch
+- GPU 지원 시 추론 속도 향상
+---
+
+## 💡 비용 최적화 팁
+- Space Settings → Hardware에서 Free CPU로 전환 시 과금 없음
+- GPU 사용 시 테스트 후 Stop 버튼으로 Space 중지
+- 48시간 요청 없으면 자동 sleep
+
+---
+
+## 🔗 관련 리포지토리
+- **전체 프로젝트 개요 & AI 서버 코드**: [GitHub - persona-chat-engine](https://github.com/m97j/persona-chat-engine)
+- **모델 어댑터 파일(HF Hub)**: [Hugging Face Model Repo](https://huggingface.co/m97j/npc_LoRA-fps)
+
+---

requirements.txt

@@ -0,0 +1,7 @@
+gradio==4.44.0
+transformers==4.43.3
+torch==2.3.1
+accelerate==0.33.0
+peft==0.11.1
+sentence-transformers==3.0.1
+python-dotenv==1.0.1

utils_prompt.py

@@ -0,0 +1,81 @@
+from typing import Dict, Any
+
+def build_webtest_prompt(npc_id: str, npc_location: str, player_utt: str) -> str:
+    # 웹 테스트에서는 최소 필드만 채운 pre dict 생성
+    pre = {
+        "tags": {
+            "npc_id": npc_id,
+            "location": npc_location,
+            "quest_stage": "",
+            "relationship": "",
+            "trust": "",
+            "npc_mood": "",
+            "player_reputation": "",
+            "style": ""
+        },
+        "player_state": {},
+        "rag_main_docs": [],  # 웹 테스트에서는 RAG 문서 없음
+        "context": [],        # 대화 히스토리 없음
+        "player_utterance": player_utt
+    }
+    # session_id는 웹 테스트에서는 의미 없으니 빈 값
+    return build_main_prompt(pre, session_id="", npc_id=npc_id)
+
+
+def build_main_prompt(pre: Dict[str, Any], session_id: str, npc_id: str) -> str:
+    tags = pre.get("tags", {})
+    ps = pre.get("player_state", {})
+    rag_docs = pre.get("rag_main_docs", [])
+
+    # RAG 문서 분리
+    lore_text = ""
+    desc_text = ""
+    for doc in rag_docs:
+        if "LORE:" in doc:
+            lore_text += doc + "\n"
+        elif "DESCRIPTION:" in doc:
+            desc_text += doc + "\n"
+        else:
+            # fallback: type 기반 분리 가능
+            if "lore" in doc.lower():
+                lore_text += doc + "\n"
+            elif "description" in doc.lower():
+                desc_text += doc + "\n"
+
+    prompt = [
+        "<SYS>",
+        f"NPC_ID={tags.get('npc_id','')}",
+        f"NPC_LOCATION={tags.get('location','')}",
+        "TAGS:",
+        f" quest_stage={tags.get('quest_stage','')}",
+        f" relationship={tags.get('relationship','')}",
+        f" trust={tags.get('trust','')}",
+        f" npc_mood={tags.get('npc_mood','')}",
+        f" player_reputation={tags.get('player_reputation','')}",
+        f" style={tags.get('style','')}",
+        "</SYS>",
+        "<RAG>",
+        f"LORE: {lore_text.strip() or '(없음)'}",
+        f"DESCRIPTION: {desc_text.strip() or '(없음)'}",
+        "</RAG>",
+        "<PLAYER_STATE>"
+    ]
+
+    if ps.get("items"):
+        prompt.append(f"items={','.join(ps['items'])}")
+    if ps.get("actions"):
+        prompt.append(f"actions={','.join(ps['actions'])}")
+    if ps.get("position"):
+        prompt.append(f"position={ps['position']}")
+    prompt.append("</PLAYER_STATE>")
+
+    prompt.append("<CTX>")
+    for h in pre.get("context", []):
+        prompt.append(f"{h['role']}: {h['text']}")
+    prompt.append("</CTX>")
+
+    prompt.append(f"<PLAYER>{pre.get('player_utterance','').rstrip()}")
+    prompt.append("<STATE>")
+    prompt.append("<NPC>")
+
+    return "\n".join(prompt)