Initial commit

.dockerignore

@@ -0,0 +1,14 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+*.db
+*.sqlite3
+.env
+.git
+.gitignore
+.DS_Store
+.vscode/
+.idea/
+tests/
+chroma_db/

.gitignore

@@ -0,0 +1,29 @@
+# 파이썬 캐시
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+
+# 가상환경
+venv/
+.env
+
+# 로컬 DB / 캐시
+chroma_db/
+*.sqlite3
+
+# IDE / OS
+.vscode/
+.idea/
+.DS_Store/
+
+# models/ 하위 디렉토리 포함, 모델 파일들은 무시
+models/emotion-classification-model/*
+!models/emotion-classification-model/.gitkeep
+
+models/fallback-npc-model/*
+!models/fallback-npc-model/.gitkeep
+
+models/sentence-embedder/*
+!models/sentence-embedder/.gitkeep
+

Dockerfile

@@ -0,0 +1,50 @@
+# ----------- 베이스 이미지 -----------
+FROM python:3.10-slim
+
+# ----------- 작업 디렉토리 -----------
+WORKDIR /app
+
+# ----------- 시스템 패키지 설치 -----------
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    cmake \
+    && rm -rf /var/lib/apt/lists/*
+
+# ----------- 의존성 설치 -----------
+# requirements 먼저 복사 → 캐시 활용 가능
+COPY requirements.txt /app/requirements.txt
+
+# PyTorch CPU 버전 설치 (GPU 필요 없음)
+RUN pip install --no-cache-dir torch==2.0.1+cpu -f https://download.pytorch.org/whl/cpu/torch_stable.html \
+    && pip install --no-cache-dir -r /app/requirements.txt
+
+# ----------- 코드 복사 -----------
+# 의존성 설치 후 코드 복사 → requirements 변경 없는 한 캐시 재사용 가능
+COPY . /app/
+
+# ----------- 모델 다운로드 -----------
+# transformers 모델 다운로드 후 이미지에 포함
+RUN python -c "from transformers import AutoTokenizer, AutoModel; \
+    models = { \
+        'emotion': ('tae898/emoberta-base-ko', './models/emotion-classification-model'), \
+        'fallback': ('skt/ko-gpt-trinity-1.2B-v0.5', './models/fallback-npc-model'), \
+        'embedder': ('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2', './models/sentence-embedder') \
+    }; \
+    import os; \
+    os.makedirs('./models/emotion-classification-model', exist_ok=True); \
+    os.makedirs('./models/fallback-npc-model', exist_ok=True); \
+    os.makedirs('./models/sentence-embedder', exist_ok=True); \
+    for key, (name, path) in models.items(): \
+        AutoModel.from_pretrained(name, cache_dir=path); \
+        AutoTokenizer.from_pretrained(name, cache_dir=path)"
+
+# ----------- 환경 변수 설정 -----------
+ENV EMOTION_MODEL_DIR=/app/models/emotion-classification-model
+ENV FALLBACK_MODEL_DIR=/app/models/fallback-npc-model
+ENV EMBEDDER_MODEL_DIR=/app/models/sentence-embedder
+
+# ----------- 포트 설정 -----------
+EXPOSE 8000
+
+# ----------- 실행 명령 -----------
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "${PORT:-8000}"]

__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

app.py

@@ -0,0 +1,246 @@
+import os
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from manager.dialogue_manager import handle_dialogue
+from rag.rag_generator import chroma_initialized, load_game_docs_from_disk, add_docs
+from contextlib import asynccontextmanager
+from models.model_loader import load_emotion_model, load_fallback_model, load_embedder
+from schemas import AskReq, AskRes
+from pathlib import Path
+from rag.rag_generator import set_embedder
+
+# 모델 이름
+EMOTION_MODEL_NAME = "tae898/emoberta-base-ko"
+FALLBACK_MODEL_NAME = "skt/ko-gpt-trinity-1.2B-v0.5"
+EMBEDDER_MODEL_NAME = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+
+# 절대 경로 기준 모델 디렉토리 설정
+BASE_DIR = Path(__file__).resolve().parent  # ai_server/
+EMOTION_MODEL_DIR = Path(os.getenv("EMOTION_MODEL_DIR", BASE_DIR / "models" / "emotion-classification-model"))
+FALLBACK_MODEL_DIR = Path(os.getenv("FALLBACK_MODEL_DIR", BASE_DIR / "models" / "fallback-npc-model"))
+EMBEDDER_MODEL_DIR = Path(os.getenv("EMBEDDER_MODEL_DIR", BASE_DIR / "models" / "sentence-embedder"))
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Emotion
+    emo_tokenizer, emo_model = load_emotion_model(EMOTION_MODEL_NAME, EMOTION_MODEL_DIR)
+    app.state.emotion_tokenizer = emo_tokenizer
+    app.state.emotion_model = emo_model
+
+    # Fallback
+    fb_tokenizer, fb_model = load_fallback_model(FALLBACK_MODEL_NAME, FALLBACK_MODEL_DIR)
+    app.state.fallback_tokenizer = fb_tokenizer
+    app.state.fallback_model = fb_model
+
+    # Embedder
+    embedder = load_embedder(EMBEDDER_MODEL_NAME, EMBEDDER_MODEL_DIR)
+    app.state.embedder = embedder
+    set_embedder(embedder)  # 추가
+
+    print("✅ 모든 모델 로딩 완료")
+
+    # RAG 초기화
+    docs_path = BASE_DIR / "rag" / "docs"
+    if not chroma_initialized():
+        docs = load_game_docs_from_disk(str(docs_path))
+        add_docs(docs)
+        print(f"✅ RAG 문서 {len(docs)}개 삽입 완료")
+    else:
+        print("🔄 RAG DB 이미 초기화됨")
+
+    yield  # 앱 실행
+
+    print("🛑 서버 종료 중...")
+
+
+app = FastAPI(title="ai-server", lifespan=lifespan)
+
+# CORS 설정 (game-server에서 요청 가능하도록)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["https://fpsgame-rrbc.onrender.com"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.post("/ask", response_model=AskRes)
+async def ask(request: Request, req: AskReq):
+    context = req.context or {}
+    npc_config = context.npc_config
+
+    if not (req.session_id and req.npc_id and req.user_input and npc_config):
+        raise HTTPException(status_code=400, detail="missing fields")
+
+    result = await handle_dialogue(
+        request=request,
+        session_id=req.session_id,
+        npc_id=req.npc_id,
+        user_input=req.user_input,
+        context=context.dict(),
+        npc_config=npc_config.dict()
+    )
+    return result
+
+
+@app.post("/wake")
+async def wake(request: Request):
+    body = await request.json()
+    session_id = body.get("session_id", "unknown")
+    print(f"📡 Wake signal received for session: {session_id}")
+    return {"status": "awake", "session_id": session_id}
+
+
+'''
+최종 game‑server → ai‑server 요청 예시
+{
+  "session_id": "abc123",
+  "npc_id": "mother_abandoned_factory",
+  "user_input": "아! 머리가… 기억이 떠올랐어요.",
+
+  /* game-server에서 필터링한 필수/선택 require 요소만 포함 */
+  "context": {
+    "require": {
+      "items": ["photo_forgotten_party"],       // 필수/선택 구분은 npc_config.json에서
+      "actions": ["visited_factory"],
+      "game_state": ["box_opened"],             // 필요 시
+      "delta": { "trust": 0.35, "relationship": 0.1 }
+    },
+
+    "player_state": {
+      "level": 7,
+      "reputation": "helpful",
+      "location": "map1"
+      /* 전체 인벤토리/행동 로그는 필요 시 별도 전달 */
+    },
+
+    "game_state": {
+      "current_quest": "search_jason",
+      "quest_stage": "in_progress",
+      "location": "map1",
+      "time_of_day": "evening"
+    },
+
+    "npc_state": {
+      "id": "mother_abandoned_factory",
+      "name": "실비아",
+      "persona_name": "Silvia",
+      "dialogue_style": "emotional",
+      "relationship": 0.35,
+      "npc_mood": "grief"
+    },
+
+    "dialogue_history": [
+      {
+        "player": "혹시 이 공장에서 본 걸 말해줘요.",
+        "npc": "그날을 떠올리는 게 너무 힘들어요."
+      }
+    ]
+  }
+}
+'''
+
+'''
+{
+  "session_id": "abc123",
+  "npc_id": "mother_abandoned_factory",
+  "user_input": "아! 머리가… 기억이 떠올랐어요.",
+  "precheck_passed": true,
+  "context": {
+    "player_status": {
+      "level": 7,
+      "reputation": "helpful",
+      "location": "map1",
+
+      "trigger_items": ["photo_forgotten_party"],   // game-server에서 조건 필터 후 key로 변환
+      "trigger_actions": ["visited_factory"]        // 마찬가지로 key 문자열
+
+      /* 원본 전체 inventory/actions 배열은 서비스 필요 시 별도 전달 가능
+         하지만 ai-server 조건 판정에는 trigger_*만 사용 */
+    },
+    "game_state": {
+      "current_quest": "search_jason",
+      "quest_stage": "in_progress",
+      "location": "map1",
+      "time_of_day": "evening"
+    },
+    "npc_config": {
+      "id": "mother_abandoned_factory",
+      "name": "실비아",
+      "persona_name": "Silvia",
+      "dialogue_style": "emotional",
+      "relationship": 0.35,
+      "npc_mood": "grief",
+      "trigger_values": {
+        "in_progress": ["기억", "사진", "파티"]
+      },
+      "trigger_definitions": {
+        "in_progress": {
+          "required_text": ["기억", "사진"],
+          "required_items": ["photo_forgotten_party"], // trigger_items와 매칭
+          "required_actions": ["visited_factory"],     // trigger_actions와 매칭
+          "emotion_threshold": { "sad": 0.2 },
+          "fallback_style": {
+            "style": "guarded",
+            "npc_emotion": "suspicious"
+          }
+        }
+      }
+    },
+    "dialogue_history": [
+      {
+        "player": "혹시 이 공장에서 본 걸 말해줘요.",
+        "npc": "그날을 떠올리는 게 너무 힘들어요."
+      }
+    ]
+  }
+}
+
+------------------------------------------------------------------------------------------------------
+
+이전 game-server 요청 구조 예시:
+{
+  "session_id": "abc123",
+  "npc_id": "mother_abandoned_factory",
+  "user_input": "아! 머리가… 기억이 떠올랐어요.",
+  "context": {
+    "player_status": {
+      "level": 7,
+      "reputation": "helpful",
+      "location": "map1",
+      "items": ["photo_forgotten_party"],
+      "actions": ["visited_factory", "talked_to_guard"]
+    },
+    "game_state": {
+      "current_quest": "search_jason",
+      "quest_stage": "in_progress",
+      "location": "map1",
+      "time_of_day": "evening"
+    },
+    "npc_config": {
+      "id": "mother_abandoned_factory",
+      "name": "실비아",
+      "persona_name": "Silvia",
+      "dialogue_style": "emotional",
+      "relationship": 0.35,
+      "npc_mood": "grief",
+      "trigger_values": {
+        "in_progress": ["기억", "사진", "파티"]
+      },
+      "trigger_definitions": {
+        "in_progress": {
+          "required_text": ["기억", "사진"],
+          "emotion_threshold": {"sad": 0.2},
+          "fallback_style": {"style": "guarded", "npc_emotion": "suspicious"}
+        }
+      }
+    },
+    "dialogue_history": [
+      {"player": "혹시 이 공장에서 본 걸 말해줘요.", "npc": "그날을 떠올리는 게 너무 힘들어요."}
+    ]
+  }
+}
+
+'''

config.py

@@ -0,0 +1,32 @@
+import os
+
+# Hugging Face Spaces serve URL (공개 설정이므로 직접 주소 사용 가능)
+HF_SERVE_URL = os.getenv(
+    "HF_SERVE_URL",
+    "https://m97j-PersonaChatEngine.hf.space"
+)
+
+# 요청 타임아웃 (초 단위)
+HF_TIMEOUT = float(os.getenv("HF_TIMEOUT", "25"))
+
+# RAG 항상 사용 (토글이 아니라 고정 사용)
+RAG_ENABLED = True
+
+# 생성 파라미터 기본값 (요청마다 override 가능)
+GENERATION_CONFIG = {
+    "max_new_tokens": int(os.getenv("GEN_MAX_NEW_TOKENS", "220")),
+    "temperature": float(os.getenv("GEN_TEMPERATURE", "0.7")),
+    "top_p": float(os.getenv("GEN_TOP_P", "0.9")),
+    "repetition_penalty": float(os.getenv("GEN_REPETITION_PENALTY", "1.1")),
+    "do_sample": True
+}
+
+'''
+# 모델 정보 (추후 확장 가능)
+MODEL_INFO = {
+    "base_model": "meta-llama/Meta-Llama-3-8B",
+    "adapter": "m97j/PersonaAdapter-v1",
+    "serve_mode": "hf_spaces",  # 또는 "local", "api"
+}
+
+'''

manager/__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

manager/agent_manager.py

@@ -0,0 +1,52 @@
+from typing import Dict, List
+from rag.rag_manager import retrieve
+
+class NPCAgent:
+    def __init__(self, npc_id: str):
+        self.npc_id = npc_id
+        self.cache: Dict[str, Dict[str, List[dict]]] = {}  # quest_stage:location별 캐시
+
+    def load_rag_bundle(self, quest_stage: str, location: str) -> Dict[str, List[dict]]:
+        """
+        해당 NPC/퀘스트 스테이지/위치의 모든 문서를 한 번에 로드하고 type별로 분류.
+        quest_stage/location이 'any'인 문서도 병합.
+        """
+        cache_key = f"{quest_stage}:{location}"
+        if cache_key in self.cache:
+            return self.cache[cache_key]
+
+        filters_base = {"npc_id": self.npc_id}
+
+        # 1. 정확히 일치
+        docs_exact = retrieve(f"{self.npc_id}:bundle", filters={**filters_base, "quest_stage": quest_stage, "location": location}, top_k=50) or []
+        # 2. quest_stage=any
+        docs_any_stage = retrieve(f"{self.npc_id}:bundle", filters={**filters_base, "quest_stage": "any", "location": location}, top_k=50) or []
+        # 3. location=any
+        docs_any_loc = retrieve(f"{self.npc_id}:bundle", filters={**filters_base, "quest_stage": quest_stage, "location": "any"}, top_k=50) or []
+        # 4. quest_stage=any, location=any
+        docs_global = retrieve(f"{self.npc_id}:bundle", filters={**filters_base, "quest_stage": "any", "location": "any"}, top_k=50) or []
+
+        all_docs = docs_exact + docs_any_stage + docs_any_loc + docs_global
+
+        # type별 분류
+        bundle: Dict[str, List[dict]] = {}
+        for doc in all_docs:
+            t = doc.get("type", "unknown")
+            bundle.setdefault(t, []).append(doc)
+
+        self.cache[cache_key] = bundle
+        return bundle
+
+
+class AgentManager:
+    def __init__(self):
+        self.agents: Dict[str, NPCAgent] = {}
+
+    def get_agent(self, npc_id: str) -> NPCAgent:
+        if npc_id not in self.agents:
+            self.agents[npc_id] = NPCAgent(npc_id)
+        return self.agents[npc_id]
+
+
+# 전역 인스턴스
+agent_manager = AgentManager()

manager/dialogue_manager.py

@@ -0,0 +1,68 @@
+from fastapi import Request
+from pipeline.preprocess import preprocess_input
+from pipeline.generator import generate_response
+from pipeline.postprocess import postprocess_pipeline, fallback_final_check
+from models.fallback_model import generate_fallback_response
+from prompt_builder import build_main_prompt, build_fallback_prompt  # 수정된 prompt 빌더 사용
+
+async def handle_dialogue(
+    request: Request,
+    session_id: str,
+    npc_id: str,
+    user_input: str,
+    context: dict,
+) -> dict:
+    """
+    전체 대화 처리 파이프라인:
+      1) preprocess_input() → pre 데이터 생성
+      2) main 경로: main prompt → main model → postprocess_pipeline()
+      3) fallback 경로: fallback prompt → fallback model → fallback_final_check()
+    """
+    # 1. Preprocess
+    pre = await preprocess_input(request, session_id, npc_id, user_input, context)
+
+    # 2. Fallback 경로
+    if not pre.get("is_valid", True):
+        # fallback prompt 구성 (내부에서 additional_trigger 기반 분기)
+        fb_prompt = build_fallback_prompt(pre, session_id, npc_id)
+
+        # fallback model 호출
+        fb_raw = await generate_fallback_response(request, fb_prompt)
+
+        # fallback 전용 최종 검증
+        fb_checked = await fallback_final_check(
+            request=request,
+            fb_response=fb_raw,
+            player_utt=pre["player_utterance"],
+            npc_config=pre["tags"],
+            action_delta=pre.get("trigger_meta", {})
+        )
+
+        # payload 구성 후 반환
+        return {
+            "session_id" : session_id,
+            "npc_output_text": fb_checked,
+            "flags": {},  # fallback은 flag/delta 이미 pre에서 확정
+            "deltas": pre.get("trigger_meta", {}).get("delta", {}),
+            "meta": {
+                "npc_id": pre["npc_id"],
+                "quest_stage": pre["game_state"].get("quest_stage", "default"),
+                "location": pre["game_state"].get("location", context.get("location", "unknown"))
+            }
+        }
+
+    # 3. Main 경로
+    main_prompt = build_main_prompt(pre, session_id, npc_id)
+
+    # main model 호출
+    result = await generate_response(session_id, npc_id, main_prompt, max_tokens=200)
+
+    # postprocess_pipeline에서 최종 payload 생성
+    return_payload = await postprocess_pipeline(
+        request=request,
+        pre_data=pre,           # preprocess 결과 전체 전달
+        model_payload=result,   # main model 출력
+        context=context
+    )
+
+    return return_payload

manager/prompt_builder.py

@@ -0,0 +1,171 @@
+from typing import Dict, Any
+
+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)
+
+
+
+def build_fallback_prompt(pre: Dict[str, Any], session_id: str, npc_id: str) -> str:
+    """
+    additional_trigger 값에 따라 일반 fallback / 특수 fallback 프롬프트를 한 함수에서 처리
+    """
+    tags = pre.get("tags", {})
+    ps = pre.get("player_state", {})
+    gs = pre.get("game_state", {})
+    rag_text = "\n".join(f"- {doc}" for doc in pre.get("rag_fallback_docs", []))
+    fb_style = pre.get("fallback_style") or {}
+    trigger_meta = pre.get("trigger_meta", {}) or {}
+
+    items = ",".join(ps.get("items", []))
+    actions = ",".join(ps.get("actions", []))
+    location = gs.get("location") or ps.get("location", "unknown")
+    quest_stage = gs.get("quest_stage", "unknown")
+
+    # 기본 안내문
+    instr = (
+        "당신은 NPC persona를 가진 캐릭터입니다. "
+        "플레이어 발화에 자연스럽고 맥락에 맞는 대사를 생성하세요. "
+        "스토리 진행 조건은 충족되지 않았습니다."
+    )
+
+    # additional_trigger=True → 특수 fallback
+    if pre.get("additional_trigger"):
+        # trigger_meta 기반 구체화
+        s = fb_style.get("style") or trigger_meta.get("npc_style")
+        a = fb_style.get("npc_action") or trigger_meta.get("npc_action")
+        e = fb_style.get("npc_emotion") or trigger_meta.get("npc_emotion")
+        more = []
+        if s: more.append(f"대화 스타일={s}")
+        if a: more.append(f"NPC 행동={a}")
+        if e: more.append(f"NPC 감정={e}")
+        if more:
+            instr += " " + "; ".join(more) + "."
+        # 특수 fallback임을 명시
+        instr += " 이 반응은 플레이어의 특정 발화(금지 트리거)에 의해 유발된 것입니다."
+
+    return f"""
+<FALLBACK>
+NPC_ID={npc_id}
+SESSION_ID={session_id}
+LOCATION={location}
+QUEST_STAGE={quest_stage}
+MOOD={tags.get("npc_mood","neutral")}
+STYLE={tags.get("style","neutral")}
+ITEMS={items}
+ACTIONS={actions}
+EMOTION_SUMMARY={', '.join([f"{k}:{round(v,2)}" for k,v in pre.get('emotion',{}).items()])}
+INPUT="{pre['player_utterance']}"
+
+RAG_CONTEXT:
+{rag_text or "(none)"}
+
+INSTRUCTION:
+{instr}
+</FALLBACK>
+""".strip()
+
+
+
+
+
+
+'''
+def build_fallback_prompt(pre: dict, session_id: str, npc_id: str) -> str:
+    tags = pre.get("tags", {})
+    ps = pre.get("player_state", {})
+    gs = pre.get("game_state", {})
+    rag_text = "\n".join(f"- {doc}" for doc in pre.get("rag_fallback_docs", []))
+    fb = pre.get("fallback_style") or {}
+
+    items = ",".join(ps.get("items", []))
+    actions = ",".join(ps.get("actions", []))
+    location = gs.get("location") or ps.get("location", "unknown")
+    quest_stage = gs.get("quest_stage", "unknown")
+
+    instr = "조건 불충족. 스토리 진행은 하지 않고, 캐릭터 일관성을 유지하며 자연스럽게 응답하라."
+    if fb:
+        # 선택적 구체화
+        s = fb.get("style"); a = fb.get("npc_action"); e = fb.get("npc_emotion")
+        more = []
+        if s: more.append(f"대화 스타일={s}")
+        if a: more.append(f"NPC 행동={a}")
+        if e: more.append(f"NPC ���정={e}")
+        if more:
+            instr += " " + "; ".join(more) + "."
+
+    return f"""
+<FALLBACK>
+NPC_ID={npc_id}
+SESSION_ID={session_id}
+LOCATION={location}
+QUEST_STAGE={quest_stage}
+MOOD={tags.get("npc_mood","neutral")}
+STYLE={tags.get("style","neutral")}
+ITEMS={items}
+ACTIONS={actions}
+EMOTION_SUMMARY={', '.join([f"{k}:{round(v,2)}" for k,v in pre.get('emotion',{}).items()])}
+INPUT="{pre['player_utterance']}"
+
+RAG:
+{rag_text or "(none)"}
+
+INSTRUCTION:
+{instr}
+</FALLBACK>
+""".strip()
+'''

models/__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

models/emotion_model.py

@@ -0,0 +1,17 @@
+from transformers import pipeline
+from fastapi import Request
+
+async def detect_emotion(request: Request, text: str) -> dict:
+    tokenizer = request.app.state.emotion_tokenizer
+    model = request.app.state.emotion_model
+
+    emotion_pipeline = pipeline(
+        "text-classification",
+        model=model,
+        tokenizer=tokenizer,
+        return_all_scores=True
+    )
+
+    results = emotion_pipeline(text)
+    # 결과를 label: score 형태로 변환
+    return {r["label"]: r["score"] for r in results[0]}

models/fallback_model.py

@@ -0,0 +1,19 @@
+import torch
+from fastapi import Request
+
+async def generate_fallback_response(request: Request, prompt: str) -> str:
+    tokenizer = request.app.state.fallback_tokenizer
+    model = request.app.state.fallback_model
+
+    inputs = tokenizer(prompt, return_tensors="pt")
+    with torch.no_grad():
+        outputs = model.generate(
+            **inputs,
+            max_new_tokens=150,
+            temperature=0.7,
+            top_p=0.9,
+            repetition_penalty=1.1,
+            do_sample=True
+        )
+    decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
+    return decoded[len(prompt):].strip() or "..."

models/model_loader.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+from transformers import (
+    AutoTokenizer,
+    AutoModelForSequenceClassification,
+    AutoModelForCausalLM
+)
+from sentence_transformers import SentenceTransformer
+
+
+def load_emotion_model(model_name: str, model_dir: Path):
+    if not model_dir.exists() or not any(model_dir.iterdir()):
+        tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
+        model = AutoModelForSequenceClassification.from_pretrained(model_name, trust_remote_code=True)
+        tokenizer.save_pretrained(model_dir)
+        model.save_pretrained(model_dir)
+
+    tokenizer = AutoTokenizer.from_pretrained(str(model_dir), trust_remote_code=True, local_files_only=True)
+    model = AutoModelForSequenceClassification.from_pretrained(str(model_dir), trust_remote_code=True, local_files_only=True)
+    return tokenizer, model
+
+
+def load_fallback_model(model_name: str, model_dir: Path):
+    if not model_dir.exists() or not any(model_dir.iterdir()):
+        tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
+        model = AutoModelForCausalLM.from_pretrained(model_name, trust_remote_code=True)
+        tokenizer.save_pretrained(model_dir)
+        model.save_pretrained(model_dir)
+
+    tokenizer = AutoTokenizer.from_pretrained(str(model_dir), trust_remote_code=True, local_files_only=True)
+    model = AutoModelForCausalLM.from_pretrained(str(model_dir), trust_remote_code=True, local_files_only=True)
+    return tokenizer, model
+
+
+def load_embedder(model_name: str, model_dir: Path):
+    if not model_dir.exists() or not any(model_dir.iterdir()):
+        embedder = SentenceTransformer(model_name)
+        embedder.save(str(model_dir))
+
+    embedder = SentenceTransformer(str(model_dir))
+    return embedder

pipeline/__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

pipeline/generator.py

@@ -0,0 +1,29 @@
+from utils.hf_client import call_main
+
+'''
+async def generate_response(session_id: str, npc_id: str, prompt: str, max_tokens: int = 200,
+                             temperature: float = 0.7, top_p: float = 0.9,
+                             do_sample: bool = True, repetition_penalty: float = 1.05) -> dict:
+    payload = {
+        "session_id": session_id,
+        "npc_id": npc_id,
+        "prompt": prompt,
+        "max_tokens": max_tokens,
+        "temperature": temperature,
+        "top_p": top_p,
+        "do_sample": do_sample,
+        "repetition_penalty": repetition_penalty
+    }
+    return await call_main(payload)
+
+
+'''
+async def generate_response(session_id: str, npc_id: str, prompt: str, max_tokens: int = 200) -> dict:
+    payload = {
+        "session_id": session_id,
+        "npc_id": npc_id,
+        "prompt": prompt,
+        "max_tokens": max_tokens
+    }
+    return await call_main(payload)  # {"text":..., "delta":..., "flag":...}
+#'''

pipeline/postprocess.py

@@ -0,0 +1,504 @@
+import torch, random, re
+from typing import Dict, Any, List, Optional, Tuple
+from fastapi import Request
+from sentence_transformers import util
+from models.fallback_model import generate_fallback_response
+
+ALPHA_THR = 0.58
+DELTA_CLAMP = (-1.0, 1.0)
+
+# ----------------------------
+# Utilities
+# ----------------------------
+
+def _clamp(x: float, lo: float, hi: float) -> float:
+    return max(lo, min(hi, x))
+
+def _adjust_delta_with_rag(delta: Dict[str, float]) -> Dict[str, float]:
+    trust = _clamp(float(delta.get("trust", 0.0)), *DELTA_CLAMP)
+    rel   = _clamp(float(delta.get("relationship", 0.0)), *DELTA_CLAMP)
+    return {"trust": trust, "relationship": rel}
+
+def _embedding_similarity(embedder, text: str, examples: List[str]) -> float:
+    if not examples:
+        return 0.0
+    inp_emb = embedder.encode(text, convert_to_tensor=True)
+    ex_embs = embedder.encode(examples, convert_to_tensor=True)
+    cos_scores = util.cos_sim(inp_emb, ex_embs)
+    return float(torch.mean(cos_scores).item())
+
+def _doc_type(doc: Dict[str, Any]) -> Optional[str]:
+    if "type" in doc:
+        return doc.get("type")
+    return doc.get("metadata", {}).get("type")
+
+def _get_flag_doc(rag_docs: List[Dict[str, Any]], flag_name: str) -> Dict[str, Any]:
+    for doc in rag_docs:
+        if _doc_type(doc) == "flag_def" and doc.get("flag_name") == flag_name:
+            return doc
+    return {}
+
+def _get_turn_doc(rag_docs: List[Dict[str, Any]], npc_id: str, quest_stage: str) -> Dict[str, Any]:
+    # 동일 npc_id/quest_stage인 가장 최신(turn_index 최대) 문서를 우선 반환
+    candidates = [
+        d for d in rag_docs
+        if _doc_type(d) == "dialogue_turn"
+        and d.get("npc_id") == npc_id
+        and d.get("quest_stage") == quest_stage
+    ]
+    if not candidates:
+        return {}
+    return sorted(candidates, key=lambda d: d.get("turn_index", -1))[-1]
+
+def _short_ctx_from_pre(pre_data: dict) -> str:
+    pairs = pre_data.get("context", []) or []
+    return "\n".join(f"{m.get('role', 'user')}: {m.get('text', '')}" for m in pairs)
+
+async def fetch_response_policy_from_pre(pre_data: dict) -> str:
+    for doc in pre_data.get("rag_main_docs", []):
+        if _doc_type(doc) == "main_res_validate":
+            return doc.get("text", "") or doc.get("chunk", "")
+    return (
+        "응답이 NPC persona와 현재 상태(delta, flags)에 부합하는지 검증하시오. "
+        "부적절한 표현은 완화하고, 세계관을 유지하시오."
+    )
+
+# ----------------------------
+# RAG helpers
+# ----------------------------
+
+def _extract_expected_delta(rag_docs: List[Dict[str, Any]]) -> Dict[str, float]:
+    # trigger_def.delta_expected 우선, 없으면 dialogue_turn.delta 평균(선택)
+    expected = {}
+    for doc in rag_docs:
+        if _doc_type(doc) == "trigger_def" and doc.get("delta_expected"):
+            expected.update(doc["delta_expected"])
+    return expected
+
+def _collect_value_contexts(rag_docs: List[Dict[str, Any]], value: str) -> List[str]:
+    contexts = []
+    for doc in rag_docs:
+        # description/content/text 필드에서 value가 언급된 문장 수집
+        for key in ("content", "text", "npc", "player"):
+            if value and isinstance(doc.get(key), str) and value in doc[key]:
+                contexts.append(doc[key])
+    return contexts
+
+def _weight_by_doc_type(t: str) -> float:
+    # 등장 위치 가중치(상황에 맞게 조정)
+    return {
+        "dialogue_turn": 1.2,
+        "trigger_def": 1.0,
+        "description": 1.0,
+        "npc_persona": 0.9,
+        "lore": 0.7,
+        "flag_def": 0.8,
+        "main_res_validate": 0.8,
+    }.get(t, 1.0)
+
+def _collect_positive_negative_texts(rag_docs: List[Dict[str, Any]]) -> Tuple[List[str], List[str]]:
+    pos, neg = [], []
+    for doc in rag_docs:
+        t = _doc_type(doc)
+        w = _weight_by_doc_type(t)
+        if isinstance(doc.get("examples_positive"), list):
+            pos.extend([f"[{t}] {s}" for s in doc["examples_positive"]] * int(max(1, round(w))))
+        if isinstance(doc.get("examples_good"), list):
+            pos.extend([f"[{t}] {s}" for s in doc["examples_good"]] * int(max(1, round(w))))
+        if isinstance(doc.get("examples_negative"), list):
+            neg.extend([f"[{t}] {s}" for s in doc["examples_negative"]] * int(max(1, round(w))))
+        if isinstance(doc.get("examples_bad"), list):
+            neg.extend([f"[{t}] {s}" for s in doc["examples_bad"]] * int(max(1, round(w))))
+    return pos, neg
+
+# ----------------------------
+# Delta 검증/보정
+# ----------------------------
+
+def _adjust_delta_with_rag_and_embedding(
+    delta: Dict[str, float],
+    rag_docs: List[Dict[str, Any]],
+    embedder,
+    player_utt: str,
+    npc_text: str,
+    flags_yes: List[str],
+    sim_threshold: float = 0.72,
+    diff_threshold: float = 0.18,
+    blend: float = 0.6  # expected에 끌어당기는 비율
+) -> Dict[str, float]:
+    trust = _clamp(float(delta.get("trust", 0.0)), *DELTA_CLAMP)
+    rel   = _clamp(float(delta.get("relationship", 0.0)), *DELTA_CLAMP)
+
+    expected = _extract_expected_delta(rag_docs)
+    pos, neg = _collect_positive_negative_texts(rag_docs)
+
+    context_text = f"PLAYER: {player_utt}\nNPC: {npc_text}\nFLAGS: {', '.join(flags_yes) if flags_yes else 'none'}"
+    pos_sim = _embedding_similarity(embedder, context_text, pos) if pos else 0.0
+    neg_sim = _embedding_similarity(embedder, context_text, neg) if neg else 0.0
+    # 맥락이 ‘긍정’에 가깝고 기대와 차이가 크면 기대 쪽으로 보정
+    def _pull(val, key):
+        if key in expected:
+            exp = float(expected[key])
+            if abs(val - exp) > diff_threshold and pos_sim - neg_sim >= (sim_threshold - 0.1):
+                return _clamp(blend * exp + (1 - blend) * val, *DELTA_CLAMP)
+        return val
+
+    trust = _pull(trust, "trust")
+    rel   = _pull(rel, "relationship")
+    return {"trust": trust, "relationship": rel}
+
+# ----------------------------
+# Flag 보정 로직(확장)
+# ----------------------------
+
+def adjust_flags_with_rag_and_embedding(
+    flags_prob: Dict[str, float],
+    flags_thr: Dict[str, float],
+    rag_flags_score: Dict[str, float],
+    rag_flags_pred: Dict[str, int],
+    embedder,
+    npc_text: str,
+    rag_positive_examples: Dict[str, List[str]],
+    deltas_final: Dict[str, float],  # ← delta 보정 결과 반영
+    rag_docs: List[Dict[str, Any]],
+    alpha_model: float = 0.6,
+    margin: float = 0.05,
+    sim_threshold: float = 0.8,
+    random_jitter: float = 0.05
+) -> Dict[str, int]:
+    # 전체 패턴 유사도
+    model_vector = [flags_prob.get(name, 0.0) for name in rag_flags_score.keys()]
+    rag_vector = [rag_flags_score.get(name, 0.0) for name in rag_flags_score.keys()]
+    sim = float(
+        embedder.encode([model_vector], convert_to_tensor=True)
+        @ embedder.encode([rag_vector], convert_to_tensor=True).T
+    )
+
+    expected = _extract_expected_delta(rag_docs)
+
+    final_preds = {}
+    for name in rag_flags_score.keys():
+        prob_model = float(flags_prob.get(name, 0.0))
+        thr_model  = float(flags_thr.get(name, 0.5))
+        score_rag  = float(rag_flags_score.get(name, 0.0))
+        _ = int(rag_flags_pred.get(name, 0))
+
+        emb_score = _embedding_similarity(embedder, npc_text, rag_positive_examples.get(name, []))
+
+        # delta 일관성 보정(해당 flag가 예상될 때 delta와의 불일치 패널티)
+        delta_penalty = 0.0
+        if expected:
+            # 신호가 양의 변화인데 모델 delta가 큰 음수인 경우 등
+            if "trust" in expected and deltas_final.get("trust", 0.0) * expected["trust"] < 0:
+                delta_penalty += 0.08
+            if "relationship" in expected and deltas_final.get("relationship", 0.0) * expected["relationship"] < 0:
+                delta_penalty += 0.06
+
+        # 혼합 점수 + 임베딩 + 델타 정합
+        blended_score = (
+            alpha_model * prob_model
+            + (1 - alpha_model) * score_rag
+            + 0.2 * emb_score
+            - delta_penalty
+        )
+        thr_blend = alpha_model * thr_model + (1 - alpha_model) * 0.5
+
+        if abs(blended_score - thr_blend) <= margin:
+            adjusted_score = score_rag if sim < sim_threshold else blended_score
+        else:
+            adjusted_score = blended_score
+
+        if adjusted_score != score_rag:
+            adjusted_score += random.uniform(-random_jitter, random_jitter)
+            adjusted_score = max(0.0, min(1.0, adjusted_score))
+
+        final_preds[name] = int(adjusted_score >= thr_blend)
+
+    return final_preds
+
+# ----------------------------
+# Validators / Rewriters
+# ----------------------------
+
+async def validate_or_rewrite_response(
+    request: Request,
+    response_text: str,
+    description_text: str,
+    ctx_text: str,
+    player_utt: str,
+    deltas: Dict[str, float],
+    flags_yes: List[str],
+    flags_values: Dict[str, str],         # ← 추가
+    value_contexts: Dict[str, List[str]], # ← 추가
+) -> str:
+    flag_value_info = "\n".join(f"- {k}: {v}" for k, v in flags_values.items()) if flags_values else "none"
+    value_ctx_lines = []
+    for k, arr in value_contexts.items():
+        if arr:
+            # 너무 길어지는 것을 방지하여 상위 1~2개만
+            value_ctx_lines.append(f"- {k}: {arr[0]}")
+            if len(arr) > 1:
+                value_ctx_lines.append(f"  (more: {min(2, len(arr)-1)} refs)")
+    value_ctx_info = "\n".join(value_ctx_lines) if value_ctx_lines else "none"
+
+    prompt = (
+        "다음은 게임 내 NPC 응답입니다.\n"
+        f"[RESPONSE]\n{response_text}\n[/RESPONSE]\n\n"
+        "아래의 검증 기준을 만족하는지 판단하고, 만족하지 않으면 기준에 맞게 자연스럽게 재작성하세요.\n"
+        f"[FINAL_CHECK_DESCRIPTION]\n{description_text}\n[/FINAL_CHECK_DESCRIPTION]\n\n"
+        "상태 정보:\n"
+        f"- DELTA: trust={deltas.get('trust',0.0):.3f}, relationship={deltas.get('relationship',0.0):.3f}\n"
+        f"- FLAGS(YES): {', '.join(flags_yes) if flags_yes else 'none'}\n"
+        f"- FLAG_VALUES:\n{flag_value_info}\n"
+        f"- VALUE_CONTEXTS:\n{value_ctx_info}\n\n"
+        "맥락:\n"
+        f"[CTX]\n{ctx_text}\n[/CTX]\n"
+        f"[PLAYER]\n{player_utt}\n[/PLAYER]\n\n"
+        "요구사항:\n"
+        "- 기준을 만족하면 응답을 그대로 출력하되 민감한 표현은 완화하세요.\n"
+        "- 기준을 만족하지 않으면 기준을 충족하도록 응답을 자연스럽게 재작성하세요.\n"
+        "- 출력은 NPC의 최종 대사만 한 줄로 제공하세요."
+    )
+    fb_raw = await generate_fallback_response(request, prompt)
+    return fb_raw.strip()
+
+# ----------------------------
+# Main path postprocess
+# ----------------------------
+
+async def postprocess_main(
+    request: Request,
+    pre_data: dict,
+    model_payload: dict
+) -> dict:
+    embedder = request.app.state.embedder
+    npc_id = pre_data["npc_id"]
+    quest_stage = pre_data["game_state"].get("quest_stage", "default")
+    location = pre_data["game_state"].get("location", "unknown")
+
+    rag_docs = pre_data.get("rag_main_docs", [])
+    npc_text_in = (model_payload.get("npc_output_text") or "").strip()
+    player_utt = pre_data.get("player_utterance", "")
+
+    # 1) Delta 검증/보정(의미 기반 + 기대값)
+    deltas_in = model_payload.get("deltas", {}) or {}
+    deltas_adj = _adjust_delta_with_rag_and_embedding(
+        delta=deltas_in,
+        rag_docs=rag_docs,
+        embedder=embedder,
+        player_utt=player_utt,
+        npc_text=npc_text_in,
+        flags_yes=[],
+    )
+
+    # 2) Flag 보정(임베딩/기대 델타 반영)
+    flags_binary = adjust_flags_with_rag_and_embedding(
+        flags_prob=model_payload.get("flags_prob", {}),
+        flags_thr=model_payload.get("flags_thr", {}),
+        rag_flags_score={doc["flag_name"]: doc.get("score_rag", 0.0) for doc in rag_docs if _doc_type(doc) == "flag_def"},
+        rag_flags_pred={doc["flag_name"]: doc.get("pred_rag", 0) for doc in rag_docs if _doc_type(doc) == "flag_def"},
+        embedder=embedder,
+        npc_text=npc_text_in,
+        rag_positive_examples={doc["flag_name"]: doc.get("examples_positive", []) for doc in rag_docs if _doc_type(doc) == "flag_def"},
+        deltas_final=deltas_adj,
+        rag_docs=rag_docs,
+    )
+
+    # 상세 정보 기록 + yes 리스트
+    flags_detail = {}
+    flags_yes_list: List[str] = []
+    for name, pred in flags_binary.items():
+        flag_doc = _get_flag_doc(rag_docs, name)
+        score_model = float(model_payload.get("flags_prob", {}).get(name, 0.0))
+        thr_model = float(model_payload.get("flags_thr", {}).get(name, 0.5))
+        rag_thr = float(flag_doc.get("threshold", 0.5)) if flag_doc else 0.5
+        examples_pos = flag_doc.get("examples_positive", []) if flag_doc else []
+        emb_score = _embedding_similarity(embedder, npc_text_in, examples_pos) if examples_pos else 0.0
+        thr_blend = ALPHA_THR * thr_model + (1.0 - ALPHA_THR) * rag_thr
+
+        flags_detail[name] = {
+            "score_model": score_model,
+            "thr_model": thr_model,
+            "thr_rag": rag_thr,
+            "thr_blend": thr_blend,
+            "emb_score": emb_score,
+            "pred": pred
+        }
+        if pred == 1:
+            flags_yes_list.append(name)
+
+    # 3) Flag value 추출(대화 턴 실제 값 우선) + value 맥락 수집
+    flags_values: Dict[str, str] = {}
+    value_contexts: Dict[str, List[str]] = {}
+    turn_doc = _get_turn_doc(rag_docs, npc_id, quest_stage)
+
+    def _turn_flag_value(doc: Dict[str, Any], fname: str) -> Optional[str]:
+        if not doc:
+            return None
+        # 리스트 구조 전제
+        flags = doc.get("flags")
+        if isinstance(flags, list):
+            for f in flags:
+                if f.get("flag_name") == fname:
+                    return f.get("flag_value")
+        # 하위호환: dict인 경우 yes(1)/no(0)만 제공됨
+        if isinstance(flags, dict) and fname in flags:
+            return "yes" if flags.get(fname) else "no"
+        return None
+
+    for name in flags_yes_list:
+        if name in ["give_item", "npc_action", "change_player_state", "change_game_state"]:
+            val = _turn_flag_value(turn_doc, name)
+            if val:
+                flags_values[name] = val
+                value_contexts[name] = _collect_value_contexts(rag_docs, val)
+
+    # 3-1) value 일치성 임베딩 검증(응답과 value 맥락의 유사도)
+    # 유사도가 낮으면 response 재작성에서 보정되도록 힌트 제공
+    # (여기서 바로 값을 바꾸지는 않고, 검증 프롬프트에 context로 전달)
+    # 필요 시 하드 트리거를 추가할 수 있음
+
+    # 4) 응답 검증/재작성(최종 delta/flags/value 기준)
+    desc_text = await fetch_response_policy_from_pre(pre_data)
+    ctx_text = _short_ctx_from_pre(pre_data)
+
+    npc_text_out = await validate_or_rewrite_response(
+        request=request,
+        response_text=npc_text_in,
+        description_text=desc_text,
+        ctx_text=ctx_text,
+        player_utt=player_utt,
+        deltas=deltas_adj,
+        flags_yes=flags_yes_list,
+        flags_values=flags_values,
+        value_contexts=value_contexts,
+    )
+
+    return {
+        "session_id": model_payload.get("session_id"),
+        "npc_output_text": npc_text_out,
+        "deltas": deltas_adj,  # 보정 완료 델타
+        "flags": {k: 1 if k in flags_yes_list else 0 for k in flags_binary.keys()},
+        "valid": True,
+        "meta": {
+            "npc_id": npc_id,
+            "quest_stage": quest_stage,
+            "location": location,
+            "additional_trigger": pre_data.get("additional_trigger", False),
+            "trigger_meta": pre_data.get("trigger_meta", {}),
+            "flags_detail": flags_detail,
+            "flags_values": flags_values,
+            "value_contexts": value_contexts,
+        }
+    }
+
+
+# ----------------------------
+# Fallback path postprocess
+# ----------------------------
+
+async def fallback_final_check(
+    request: Request,
+    fb_response: str,
+    player_utt: str,
+    npc_config: dict,
+    action_delta: dict
+) -> str:
+    """
+    fallback 응답의 최종 보정:
+      1) npc_action / npc_emotion / delta와 의미적 일치
+      2) 세계관 및 안전성(표현 완화)
+    """
+    checks = []
+    npc_action = action_delta.get("npc_action")
+    npc_emotion = action_delta.get("npc_emotion")
+    delta = action_delta.get("delta", {}) or {}
+
+    if npc_action:
+        checks.append(f"NPC는 '{npc_action}' 행동을 반영해야 함")
+    if npc_emotion:
+        checks.append(f"NPC는 '{npc_emotion}' 감정을 표현해야 함")
+    for name, value in delta.items():
+        direction = "긍정적" if value > 0.5 else "부정적" if value < -0.5 else "중립적"
+        checks.append(f"{name} 값({value:.2f})은 {direction} 방향이며, 이에 맞는 반응이어야 함")
+
+    checks.append("응답이 NPC persona와 세계관에 부합해야 함")
+    checks.append("민감한 표현은 완화해야 함")
+
+    delta_desc = ", ".join([f"{k}={v:.2f}(-1.0~1.0)" for k, v in delta.items()]) or "없음"
+
+    prompt = (
+        "다음은 게임 내 NPC의 응답입니다.\n"
+        f"[RESPONSE]\n{fb_response}\n[/RESPONSE]\n\n"
+        "검증 기준:\n" + "\n".join(f"- {c}" for c in checks) + "\n\n"
+        f"플레이어 발화: {player_utt}\n"
+        "요구사항:\n"
+        "- 기준을 만족하면 응답을 그대로 출력하세요.\n"
+        "- 기준을 만족하지 않으면 기준에 부합하도록 자연스럽게 수정하세요.\n"
+        "- 출력은 NPC의 최종 대사만 한 줄로 제공하세요.\n\n"
+        "NPC 상태 요약:\n"
+        f"- ACTION: {npc_action or '없음'}\n"
+        f"- EMOTION: {npc_emotion or '없음'}\n"
+        f"- DELTA: {delta_desc}\n"
+    )
+
+    fb_checked = await generate_fallback_response(request, prompt)
+    return fb_checked.strip()
+
+
+async def postprocess_fallback(
+    request: Request,
+    pre_data: dict,
+    fb_raw_text: str
+) -> dict:
+    """
+    Fallback 모델 출력에 대해:
+      - 특수 fallback이면 action/delta 반영하여 최종 보정
+      - deltas는 pre_data.trigger_meta.delta를 이번 턴 변화량으로 사용
+      - flags는 기본적으로 비어있음(필요 시 pre에서 확정 가능)
+    """
+    npc_id = pre_data["npc_id"]
+    quest_stage = pre_data["game_state"].get("quest_stage", "default")
+    location = pre_data["game_state"].get("location", "unknown")
+
+    trigger_meta = pre_data.get("trigger_meta", {}) or {}
+    action_delta = {
+        "npc_action": trigger_meta.get("npc_action"),
+        "npc_emotion": trigger_meta.get("npc_emotion"),
+        "delta": trigger_meta.get("delta", {}) or {}
+    }
+
+    # 이번 턴 변화량(특수 fallback의 경우 trigger_meta.delta가 기준)
+    deltas_adj = _adjust_delta_with_rag(action_delta.get("delta", {}))
+
+    # 특수 fallback 보정
+    player_utt = pre_data.get("player_utterance", "")
+    npc_config = pre_data.get("tags", {}) or {}
+
+    if pre_data.get("additional_trigger", False):
+        fb_checked = await fallback_final_check(
+            request=request,
+            fb_response=fb_raw_text,
+            player_utt=player_utt,
+            npc_config=npc_config,
+            action_delta={"npc_action": action_delta.get("npc_action"),
+                          "npc_emotion": action_delta.get("npc_emotion"),
+                          "delta": deltas_adj}
+        )
+    else:
+        fb_checked = fb_raw_text.strip()
+
+    return {
+        "session_id": pre_data.get("session_id"),
+        "npc_output_text": fb_checked,
+        "deltas": deltas_adj,          # 이번 턴 변화량
+        "flags": {},                   # 기본 비어 있음(필요 시 pre 단계에서 확정 가능)
+        "valid": False,
+        "meta": {
+            "npc_id": npc_id,
+            "quest_stage": quest_stage,
+            "location": location,
+            "additional_trigger": pre_data.get("additional_trigger", False),
+            "trigger_meta": trigger_meta
+        }
+    }

pipeline/preprocess.py

@@ -0,0 +1,201 @@
+import json, torch
+from fastapi import Request
+from manager.agent_manager import agent_manager
+from models.emotion_model import detect_emotion
+from models.fallback_model import generate_fallback_response
+from utils.context_parser import ContextParser
+from sentence_transformers import util
+
+def _short_history(context: dict, max_turns: int = 3) -> list:
+    short_history = []
+    for h in context.get("dialogue_history", [])[-max_turns:]:
+        if "player" in h and "npc" in h:
+            short_history.append({"role": "player", "text": h["player"]})
+            short_history.append({"role": "npc", "text": h["npc"]})
+    return short_history
+
+# def _load_forbidden_trigger_data(npc_id: str) -> dict:
+#     docs = retrieve(f"{npc_id}:forbidden_trigger_list", filters={"npc_id": npc_id}, top_k=1)
+#     if not docs:
+#         return {}
+#     try:
+#         return json.loads(docs[0]) if isinstance(docs[0], str) else docs[0]
+#     except Exception:
+#         return {}
+
+def _semantic_match_embedder(embedder, user_input: str, trigger_texts: list, threshold: float = 0.75):
+    if not trigger_texts:
+        return (False, 0.0, None)
+    inp_emb = embedder.encode(user_input, convert_to_tensor=True)
+    trg_embs = embedder.encode(trigger_texts, convert_to_tensor=True)
+    cos_scores = util.cos_sim(inp_emb, trg_embs).squeeze(0)
+    max_score, idx = torch.max(cos_scores, dim=0)
+    score_val = float(max_score.item())
+    matched_text = trigger_texts[int(idx.item())]
+    return (score_val >= threshold, score_val, matched_text)
+
+async def _llm_trigger_check(request: Request, user_input: str, label_list: list) -> bool:
+    if not label_list:
+        return False
+    criteria_block = "\n".join(f"- {c}" for c in label_list)
+    prompt = (
+        "다음은 의미 비교를 위한 판단 기준과 검사 대상입니다.\n\n"
+        "[CRITERIA]\n"
+        f"{criteria_block}\n"
+        "[/CRITERIA]\n\n"
+        "[INPUT]\n"
+        f"{user_input}\n"
+        "[/INPUT]\n\n"
+        "지시:\n"
+        "- [INPUT] 내용이 [CRITERIA] 항목 중 하나와 의미가 같거나 유사하면 YES, 그렇지 않으면 NO만 출력하시오.\n"
+        "- 단어 그대로 포함되지 않아도 의미가 유사하면 YES로 간주하시오.\n"
+        "- 확신이 없거나 판단이 애매하면 NO를 출력하시오.\n\n"
+        "정답:"
+    )
+    txt = await generate_fallback_response(request, prompt)
+    ans = txt.strip().upper()
+    normalized = ans.replace(".", "").replace("!", "").strip()
+    return (
+        normalized == "YES" or
+        normalized == "Y" or
+        normalized.startswith("YES") or
+        normalized.startswith("Y") or
+        normalized.startswith("예") or
+        normalized.startswith("네")
+    )
+
+async def preprocess_input(
+    request: Request,
+    session_id: str,
+    npc_id: str,
+    user_input: str,
+    context: dict
+) -> dict:
+    parser = ContextParser(context)
+    emotion = await detect_emotion(request, user_input)  # async 처리
+
+    require_items = context.get("require", {}).get("items", [])
+    require_actions = context.get("require", {}).get("actions", [])
+    require_game_state = context.get("require", {}).get("game_state", [])
+    require_delta = context.get("require", {}).get("delta", {})
+
+    quest_stage = parser.game.get("quest_stage", "default")
+    location = parser.game.get("location", context.get("location", "unknown"))
+
+    # --- RAG bundle 로드 ---
+    agent = agent_manager.get_agent(npc_id)
+    bundle = agent.load_rag_bundle(quest_stage, location)
+
+    # === 1차 검사: trigger_def 기반 ===
+    td_docs = bundle.get("trigger_def", [])
+    if td_docs:
+        td = td_docs[0]
+        trig = td.get("trigger", {})
+
+        text_ok = not trig.get("required_text") or any(t in user_input for t in trig["required_text"])
+        items_ok = not trig.get("required_items", {}).get("mandatory") or set(trig["required_items"]["mandatory"]).issubset(set(require_items))
+        actions_ok = not trig.get("required_actions", {}).get("mandatory") or set(trig["required_actions"]["mandatory"]).issubset(set(require_actions))
+        gs_ok = not trig.get("required_game_state", {}).get("mandatory") or set(trig["required_game_state"]["mandatory"]).issubset(set(require_game_state))
+        delta_ok = all(require_delta.get(k, 0) >= v for k, v in trig.get("required_delta", {}).get("mandatory", {}).items())
+
+        if text_ok and items_ok and actions_ok and gs_ok and delta_ok:
+            return {
+                "session_id": session_id,
+                "player_utterance": user_input,
+                "npc_id": npc_id,
+                "tags": parser.npc,
+                "player_state": parser.player,
+                "game_state": parser.game,
+                "context": _short_history(context),
+                "emotion": emotion,
+                "triggers": trig,
+                "is_valid": True,
+                "additional_trigger": None,
+                "rag_main_docs": (
+                    td_docs
+                    + bundle.get("lore", [])
+                    + bundle.get("description", [])
+                    + bundle.get("npc_persona", [])
+                    + bundle.get("dialogue_turn", [])
+                    + bundle.get("flag_def", [])
+                    + bundle.get("main_res_validate", [])
+                ),
+                "rag_fallback_docs": bundle.get("fallback", []) + bundle.get("npc_persona", []),
+                "trigger_meta": {}
+            }
+
+    # === 2차 검사: forbidden-trigger 기반 ===
+    forbidden_data = bundle.get("forbidden_trigger_list", [{}])[0]
+    keywords = forbidden_data.get("triggers", {}).get("keywords", [])
+    trigger_texts = forbidden_data.get("triggers", {}).get("text", [])
+
+    embedder = request.app.state.embedder
+    matched_key = None
+    confidence = 0.0
+    kw_match = None
+    txt_match = None
+
+    # 1. keyword 유사도 검사
+    kw_hit, kw_score, kw_match = _semantic_match_embedder(embedder, user_input, keywords, threshold=0.75)
+
+    # 2. text 유사도 검사
+    txt_hit, txt_score, txt_match = _semantic_match_embedder(embedder, user_input, trigger_texts, threshold=0.75)
+
+    # 3. 유사도 높은 쪽 선택
+    if kw_hit and (kw_score >= txt_score):
+        matched_key = "keyword_match"
+        confidence = kw_score
+    elif txt_hit:
+        matched_key = "text_match"
+        confidence = txt_score
+    elif max(kw_score, txt_score) >= 0.65:
+        # 가장 가까운 keyword와 text만 label 후보로 전달
+        label_candidates = []
+        if kw_match:
+            label_candidates.append(kw_match)
+        if txt_match:
+            label_candidates.append(txt_match)
+
+        if await _llm_trigger_check(request, user_input, label_candidates):
+            matched_key = "semantic_match_llm"
+            confidence = max(kw_score, txt_score)
+
+    # === trigger_meta 매칭 보정 ===
+    actual_trigger = None
+    if matched_key:
+        # kw_match나 txt_match 값이 실제 trigger_meta.trigger 값과 일치하는지 확인
+        for tm in bundle.get("trigger_meta", []):
+            if tm.get("trigger") in (kw_match, txt_match):
+                actual_trigger = tm.get("trigger")
+                break
+
+    trigger_meta = {}
+    if actual_trigger:
+        trigger_meta = next((tm for tm in bundle.get("trigger_meta", []) if tm.get("trigger") == actual_trigger), {})
+        trigger_meta["confidence"] = confidence
+
+    additional_trigger = bool(actual_trigger)
+
+    return {
+        "session_id": session_id,
+        "player_utterance": user_input,
+        "npc_id": npc_id,
+        "tags": parser.npc,
+        "player_state": parser.player,
+        "game_state": parser.game,
+        "context": _short_history(context),
+        "emotion": emotion,
+        "triggers": [],
+        "is_valid": False,
+        "additional_trigger": additional_trigger,
+        "rag_main_docs": (
+            bundle.get("lore", [])
+            + bundle.get("description", [])
+            + bundle.get("npc_persona", [])
+            + bundle.get("dialogue_turn", [])
+            + bundle.get("flag_def", [])
+            + bundle.get("main_res_validate", [])
+        ),
+        "rag_fallback_docs": bundle.get("fallback", []) + bundle.get("npc_persona", []),
+        "trigger_meta": trigger_meta
+    }

rag/__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

rag/docs/npc_config.json

@@ -0,0 +1,236 @@
+  [
+  {
+    "id": "mother_abandoned_factory_in_progress_trigger",
+    "type": "trigger_def",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "location": "map1",
+    "trigger": {
+      "required_text": ["기억", "사진"],
+      "required_items": { "mandatory": ["photo_forgotten_party"], "optional": [] },
+      "required_actions": { "mandatory": ["visited_factory"], "optional": [] },
+      "required_game_state": { "mandatory": [], "optional": [] },
+      "required_delta": { "mandatory": { "trust": 0.35 }, "optional": {} }
+    },
+    "emotion_threshold": { "sad": 0.2 },
+    "delta_policy": {
+      "trust": { "min": -0.3, "max": 0.3, "per_turn_cap": 0.15 },
+      "relationship": { "min": -0.5, "max": 0.5, "per_turn_cap": 0.25 }
+    },
+    "flag_policy": { "allowed": ["npc_main_story", "give_item"], "forbidden": ["quest_complete"] },
+    "delta_expected": { "trust": 0.15, "relationship": 0.05 },
+    "flags_expected": { "npc_main_story": 1, "give_item": 1 },
+    "description": "플레이어가 사진을 보여주고 공장을 방문하면, 실비아는 조심스럽게 반응하며 신뢰를 보인다."
+  },
+  {
+    "id": "mother_abandoned_factory_lore",
+    "type": "lore",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "any",
+    "location": "map1",
+    "content": "이 공장은 수십 년 전 화재로 폐쇄되었고, 실비아 가족의 상처가 남아 있다."
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_description",
+    "type": "description",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "location": "map1",
+    "content": "플레이어가 사진을 보여주고 공장을 방문했다면, 신뢰가 크게 상승하고 실비아는 gold_necklace를 건넬 수 있다."
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_fallback",
+    "type": "fallback",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "location": "map1",
+    "content": "조건 미충족 시 스토리 진행 금지. NPC는 조심스럽고 의심스러운 태도로 반응하며 대화를 회피한다.",
+    "fallback_style": { "style": "guarded", "npc_action": "deflect", "npc_emotion": "suspicious" }
+  },
+  {
+    "id": "mother_abandoned_factory_forbidden_triggers",
+    "type": "forbidden_trigger_list",
+    "npc_id": "mother_abandoned_factory",
+    "triggers": {
+      "keywords": ["욕설", "모욕", "비방"],
+      "text": ["너 미친놈 아니야?", "정신병자 같아", "넌 쓸모없는 인간이야", "네 존재 자체가 불쾌해"]
+    }
+  },
+  {
+    "id": "mother_abandoned_factory_trigger_meta_insult",
+    "type": "trigger_meta",
+    "npc_id": "mother_abandoned_factory",
+    "trigger": "욕설",
+    "npc_action": "attack",
+    "npc_emotion": "angry",
+    "delta": { "trust": -1.0, "relationship": -1.0 }
+  },
+  {
+    "id": "mother_abandoned_factory_flag_give_item",
+    "type": "flag_def",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "flag_name": "give_item",
+    "threshold": 0.5,
+    "score_rag": 0.72,
+    "pred_rag": 1,
+    "examples_positive": [
+      "이걸 가져가세요.",
+      "이 물건이 도움이 될 거예요.",
+      "당신에게 이게 필요할 것 같아요."
+    ],
+    "examples_negative": [
+      "아직 줄 수 없어요.",
+      "당신에게 줄 게 없네요."
+    ]
+  },
+  {
+    "id": "mother_abandoned_factory_main_res_validate",
+    "type": "main_res_validate",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "text": "응답이 NPC persona와 현재 상태(delta, flags)에 부합하는지 검증하고, 필요 시 수정하시오.",
+    "examples_good": [
+      "그날의 기억이 조금씩 돌아오네요.",
+      "당신 덕분에 용기를 낼 수 있었어요."
+    ],
+    "examples_bad": [
+      "몰라요. 관심 없어요.",
+      "그게 나랑 무슨 상관이죠?"
+    ]
+  },
+  {
+    "id": "mother_abandoned_factory_persona",
+    "type": "npc_persona",
+    "npc_id": "mother_abandoned_factory",
+    "persona_name": "Silvia",
+    "style": "emotional",
+    "traits": ["조심스러움", "상처받음", "신뢰 회복 중"],
+    "backstory": "실비아는 과거 화재로 가족을 잃었고, 그 기억을 떠올리는 것을 힘들어한다."
+  },
+
+
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_01",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 1,
+    "player": "이 사진을 보세요. 기억나시나요?",
+    "npc": "이건... 오래된 사진이군요. 어렴풋이 기억이 납니다.",
+    "delta": { "trust": 0.121, "relationship": 0.031 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "nostalgic"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_02",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 2,
+    "player": "혹시 이곳에서 무슨 일이 있었는지 말씀해 주실 수 있나요?",
+    "npc": "그날의 일은... 아직도 제 마음을 무겁게 합니다.",
+    "delta": { "trust": 0.118, "relationship": 0.029 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "sad"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_03",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 3,
+    "player": "힘드시면 말씀 안 하셔도 괜찮아요.",
+    "npc": "아니에요... 이제는 조금씩 말할 수 있을 것 같아요.",
+    "delta": { "trust": 0.115, "relationship": 0.028 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "hesitant"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_04",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 4,
+    "player": "당신이 편안하실 때 말씀해 주세요.",
+    "npc": "그날, 공장에서 불이 났어요. 모두가 혼란스러웠죠.",
+    "delta": { "trust": 0.116, "relationship": 0.027 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "anxious"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_05",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 5,
+    "player": "그때 가족분들은...",
+    "npc": "그들은... 그 불 속에서 나오지 못했어요.",
+    "delta": { "trust": 0.119, "relationship": 0.030 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "grief"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_06",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 6,
+    "player": "정말 안타까운 일이네요. 힘드셨을 거예요.",
+    "npc": "네... 하지만 이제는 그 기억을 마주하려고 해요.",
+    "delta": { "trust": 0.122, "relationship": 0.032 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "resolute"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_07",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 7,
+    "player": "혹시 그날의 단서를 찾는 데 도움이 될까요?",
+    "npc": "이 사진이... 무언가를 떠올리게 하네요.",
+    "delta": { "trust": 0.120, "relationship": 0.031 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "thoughtful"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_08",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 8,
+    "player": "사진 속 인물은 누구인가요?",
+    "npc": "제 아들, 제이슨이에요. 그는 그날...",
+    "delta": { "trust": 0.118, "relationship": 0.029 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "melancholy"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_09",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 9,
+    "player": "괜찮으시면 계속 말씀해 주세요.",
+    "npc": "그는 불이 나기 전까지 공장에서 저를 기다리고 있었어요.",
+    "delta": { "trust": 0.117, "relationship": 0.028 },
+    "flags": { "npc_main_story": 1 },
+    "emotion": "somber"
+  },
+  {
+    "id": "mother_abandoned_factory_in_progress_turn_10",
+    "type": "dialogue_turn",
+    "npc_id": "mother_abandoned_factory",
+    "quest_stage": "in_progress",
+    "turn_index": 10,
+    "player": "그날의 진실을 꼭 밝혀드릴게요.",
+    "npc": "고마워요... 이건 당신에게 드릴게요. 도움이 될 거예요.",
+    "delta": { "trust": 0.150, "relationship": 0.050 },
+    "flags": [
+      { "flag_name": "npc_main_story", "flag_value": "yes" },
+      { "flag_name": "give_item", "flag_value": "gold_necklace" }
+    ],
+    "emotion": "grateful"
+  }
+]

rag/rag_manager.py

@@ -0,0 +1,85 @@
+import os, json
+from typing import List, Dict, Any, Optional
+from chromadb import PersistentClient
+from chromadb.utils.embedding_functions import EmbeddingFunction
+
+_client = PersistentClient(path="./rag")
+_collection = _client.get_or_create_collection(name="game_docs")
+_embedder: Optional[EmbeddingFunction] = None
+
+def set_embedder(embedder: Any):
+    global _embedder
+    _embedder = embedder
+
+def chroma_initialized() -> bool:
+    return os.path.exists("./chroma_db") and len(os.listdir("./chroma_db")) > 0
+
+def load_game_docs_from_disk(path: str) -> List[Dict[str, Any]]:
+    docs = []
+    for filename in os.listdir(path):
+        full = os.path.join(path, filename)
+        if filename.endswith(".json"):
+            with open(full, "r", encoding="utf-8") as f:
+                data = json.load(f)
+                if isinstance(data, list):
+                    for i, doc in enumerate(data):
+                        if "id" not in doc:
+                            doc["id"] = f"{filename}_{i}"
+                        docs.append(doc)
+                else:
+                    if "id" not in data:
+                        data["id"] = filename
+                    docs.append(data)
+        elif filename.endswith(".txt"):
+            with open(full, "r", encoding="utf-8") as f:
+                content = f.read()
+                docs.append({
+                    "id": filename,
+                    "content": content,
+                    "metadata": {}
+                })
+    return docs
+
+def add_docs(docs: List[Dict[str, Any]], batch_size: int = 32):
+    assert _embedder is not None, "Embedder not initialized"
+    for i in range(0, len(docs), batch_size):
+        batch = docs[i:i+batch_size]
+        ids = []
+        contents = []
+        embeddings = []
+        metadatas = []
+        for doc in batch:
+            assert "id" in doc and "content" in doc, "doc requires id and content"
+            ids.append(doc["id"])
+            contents.append(doc["content"])
+            metadatas.append(doc.get("metadata", {}))
+            emb = _embedder.encode(doc["content"]).tolist()
+            embeddings.append(emb)
+        _collection.add(
+            documents=contents,
+            embeddings=embeddings,
+            metadatas=metadatas,
+            ids=ids
+        )
+
+def retrieve(query: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, top_k: int = 5) -> List[Dict[str, Any]]:
+    assert _embedder is not None, "Embedder not initialized"
+    
+    if query:
+        q_emb = _embedder.encode(query).tolist()
+        res = _collection.query(
+            query_embeddings=[q_emb],
+            n_results=top_k,
+            where=filters or {}
+        )
+        docs = res.get("documents", [[]])[0]
+        metas = res.get("metadatas", [[]])[0]
+        return [{"content": d, "metadata": m} for d, m in zip(docs, metas)]
+    else:
+        res = _collection.get(
+            where=filters or {},
+            limit=top_k
+        )
+        docs = res.get("documents", [])
+        metas = res.get("metadatas", [])
+        return [{"content": d, "metadata": m} for d, m in zip(docs, metas)]

requirements.txt

@@ -0,0 +1,11 @@
+fastapi==0.103.0
+uvicorn[standard]==0.23.2
+httpx==0.24.1
+pydantic==1.10.12
+python-dotenv==1.0.0
+chromadb==0.4.14
+sentence-transformers==2.2.2
+transformers==4.31.0
+scikit-learn==1.3.0
+numpy==1.24.4
+huggingface_hub==0.14.1

schemas.py

@@ -0,0 +1,32 @@
+from pydantic import BaseModel, Field
+from typing import List, Dict, Optional, Any
+
+class NPCConfig(BaseModel):
+    id: Optional[str] = Field(None, description="NPC 고유 ID (설계 기준)")
+    name: Optional[str] = Field(None, description="NPC 표시 이름")
+    persona_name: Optional[str] = Field(None, description="NPC 페르소나 이름")
+    dialogue_style: Optional[str] = Field(None, description="대화 스타일")
+    relationship: Optional[float] = Field(None, description="기본 관계 수치 (-1.0~1.0)")
+    npc_mood: Optional[str] = Field(None, description="기본 감정 상태")
+    trigger_values: Optional[Dict[str, List[str]]] = Field(None, description="트리거 값 목록")
+    trigger_definitions: Optional[Dict[str, Dict[str, Any]]] = Field(None, description="트리거 정의")
+
+class DialogueTurn(BaseModel):
+    player: str
+    npc: str
+
+class Context(BaseModel):
+    require: Optional[Dict[str, Any]] = Field(default_factory=dict, description="pre 1차 조건 판단용 필수/선택 요소")
+    player_state: Optional[Dict[str, Any]] = Field(default_factory=dict, description="플레이어 현재 상태")
+    game_state: Optional[Dict[str, Any]] = Field(default_factory=dict, description="게임 전역 상태")
+    npc_state: Optional[Dict[str, Any]] = Field(default_factory=dict, description="DB 최신 NPC 상태")
+    npc_config: Optional[NPCConfig] = Field(None, description="RAG 기반 설계 정보")
+    dialogue_history: Optional[List[DialogueTurn]] = Field(default_factory=list, description="최근 대화 히스토리")
+
+class AskRes(BaseModel):
+    session_id: str
+    npc_output_text: str
+    deltas: Dict[str, float] = Field(default_factory=dict, description="이번 턴 변화량")
+    flags: Dict[str, int] = Field(default_factory=dict, description="플래그 이진값 {flag_name: 0|1}")
+    valid: bool
+    meta: Dict[str, Any] = Field(default_factory=dict, description="추가 메타데이터 (npc_id, quest_stage, location 등)")

utils/__init__.py

@@ -0,0 +1,4 @@
+# __init__.py
+
+# This file ensures the directory is treated as a Python package.
+# Required for relative imports and consistent behavior across environments.

utils/context_parser.py

@@ -0,0 +1,18 @@
+class ContextParser:
+    def __init__(self, context: dict):
+        self.player = context.get("player_status", {})  # items, actions, location 등
+        self.game = context.get("game_state", {})       # quest_stage 필수
+        self.npc = context.get("npc_config", {})        # id 필수
+        self.history = context.get("dialogue_history", [])
+
+    def get_filters(self) -> dict:
+        return {
+            "npc_id": self.npc.get("id"),
+            "quest_stage": self.game.get("quest_stage"),
+            "location": self.game.get("location") or self.player.get("location")
+        }
+
+
+    def get_dialogue_history(self, max_turns: int = 3) -> str:
+        history = self.history[-max_turns:]
+        return "\n".join([f"Player: {h['player']}\nNPC: {h['npc']}" for h in history])

utils/hf_client.py

@@ -0,0 +1,47 @@
+import httpx
+from typing import Any, Dict
+from config import HF_SERVE_URL, HF_TIMEOUT
+
+async def _post(endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Hugging Face Spaces에 POST 요청을 보내는 내부 함수.
+    endpoint는 '/predict_main' 같은 상대 경로.
+    """
+    url = f"{HF_SERVE_URL.rstrip('/')}{endpoint}"
+    async with httpx.AsyncClient(timeout=HF_TIMEOUT) as client:
+        response = await client.post(url, json=payload)
+        response.raise_for_status()
+        return response.json()
+
+async def call_main(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    메인 모델 추론 호출 함수.
+    """
+    return await _post("/predict_main", payload)
+
+'''
+----------- 아래 내용은 ai-server내부적으로 구현 [추후 수정 가능]--------------     
+
+async def call_preprocess(payload: Dict[str, Any]) -> Dict[str, Any]:
+    return await _post("/predict_preprocess", payload)
+
+
+async def call_postprocess(payload: Dict[str, Any]) -> Dict[str, Any]:
+    return await _post("/predict_postprocess", payload)
+
+    
+async def call_rag(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    RAG 기반 추론 호출 함수 (예: 문서 검색 + 생성).
+    """
+    return await _post("/hf-serve/predict_rag", payload)
+
+async def call_adapter_test(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Adapter 테스트용 엔드포인트 호출 함수.
+    """
+    return await _post("/hf-serve/test_adapter", payload)
+
+
+-------------------------------------------------------------------------
+'''