---
name: pickit-llm-cache
description: PICKIT 백엔드를 구현합니다. Gemini API로 넛지를 생성하고, Bandit 상태는 JSON 파일에 저장합니다. Redis는 선택사항입니다. server/llm.py, server/main.py, requirements.txt를 프로젝트에 직접 생성합니다.
license: MIT
metadata:
  author: pickit
  version: "2.0"
---

# PICKIT LLM Cache

## MVP 설계 원칙

| 항목 | MVP 지금 | Phase 2 나중에 |
|---|---|---|
| LLM 카피 캐시 | Redis 있으면 사용, 없으면 매번 Gemini 호출 | Redis 필수화 |
| Bandit 상태 저장 | `data/bandit_state.json` 파일 | PostgreSQL |
| 이벤트 로그 | 없음 (콘솔 출력만) | PostgreSQL `events` 테이블 |

**캐싱 원칙** (변하지 않음):
- LLM 카피(텍스트)만 캐시 — Bandit arm 선택은 절대 캐시 금지
- 캐시 키: `nudge:{shop_id}:{rule_id}:{product_no}`
- 로그인 개인화 카피는 캐시 금지

---

## 실행 절차

### 1. 프로젝트 파일 읽기

- `server/llm_prompts.py` — NUDGE_PROMPTS format 변수 목록 확인
- `rules/rules.json` — Rule 구조 확인 (conditions[].field 목록)
- `rules/nudge_copy.json` — 폴백 카피 구조 확인
- `requirements.txt` 있으면 읽어서 기존 패키지 파악

### 2. `data/` 디렉토리 생성

`data/` 폴더가 없으면 생성. `data/bandit_state.json`이 없으면 빈 파일 `{}` 생성.

### 3. `server/llm.py` 생성

```python
import asyncio
import json
import os

from google import genai
from google.genai import types

from server.llm_prompts import NUDGE_PROMPTS

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

# Redis는 선택사항 — 없어도 동작함
_redis = None
try:
    import redis as _redis_lib
    _r = _redis_lib.Redis(
        host=os.environ.get("REDIS_HOST", "localhost"),
        port=int(os.environ.get("REDIS_PORT", 6379)),
        decode_responses=True,
        socket_connect_timeout=1,
    )
    _r.ping()
    _redis = _r
    print("✅ Redis 연결됨 — LLM 응답 캐싱 활성화")
except Exception:
    print("⚠️  Redis 없음 — 캐싱 없이 동작 (매번 Gemini 호출)")

_fallback_copy: dict | None = None


def _load_fallback() -> dict:
    global _fallback_copy
    if _fallback_copy is None:
        with open("rules/nudge_copy.json", encoding="utf-8") as f:
            _fallback_copy = json.load(f)
    return _fallback_copy


def _fallback(rule_id: str) -> dict:
    return _load_fallback().get(rule_id, {
        "icon": "✨",
        "tag": "PICKIT",
        "title": "지금이 구매 타이밍",
        "body": "이 상품을 고려 중이신 분들이 많아요",
        "cta": "장바구니 담기",
        "cta_target": "#actionCart",
    })


async def _call_gemini(prompt: str) -> dict:
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(
        None,
        lambda: _client.models.generate_content(
            model="gemini-2.0-flash",
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                max_output_tokens=256,
                temperature=0.7,
            ),
        ),
    )
    return json.loads(response.text)


def _build_prompt_vars(product_ctx: dict) -> dict:
    parts = []
    if product_ctx.get("product_name"):
        parts.append(f"상품명: {product_ctx['product_name']}")
    if product_ctx.get("review_count"):
        parts.append(f"리뷰 수: {product_ctx['review_count']}개")
    if product_ctx.get("purchase_count_7d"):
        parts.append(f"최근 7일 구매: {product_ctx['purchase_count_7d']}건")
    if product_ctx.get("stock_number", 999) <= 10:
        parts.append(f"잔여 재고: {product_ctx['stock_number']}개")
    return {
        "product_price": product_ctx.get("product_price", 0),
        "review_count": product_ctx.get("review_count", 0),
        "llm_context_fields_desc": "\n".join(parts) if parts else "추가 데이터 없음",
        "brand_voice": product_ctx.get("brand_voice", "친근함"),
        "tone_keywords": product_ctx.get("tone_keywords", ""),
    }


async def generate_nudge(
    rule_id: str,
    shop_id: str,
    product_no: str,
    product_ctx: dict,
    is_personalized: bool,
) -> dict:
    cache_key = f"nudge:{shop_id}:{rule_id}:{product_no}"

    # Redis 캐시 확인 (비로그인 + Redis 있을 때만)
    if _redis and not is_personalized:
        try:
            cached = _redis.get(cache_key)
            if cached:
                return json.loads(cached)
        except Exception:
            pass  # Redis 장애 시 그냥 Gemini 호출

    prompt_template = NUDGE_PROMPTS.get(rule_id)
    if not prompt_template:
        return _fallback(rule_id)

    try:
        prompt = prompt_template.format(**_build_prompt_vars(product_ctx))
        result = await asyncio.wait_for(_call_gemini(prompt), timeout=1.5)
    except Exception:
        return _fallback(rule_id)

    # 캐시 저장 (비로그인 + Redis 있을 때만)
    if _redis and not is_personalized:
        try:
            _redis.setex(cache_key, 3600, json.dumps(result, ensure_ascii=False))
        except Exception:
            pass

    return result
```

### 4. `server/main.py` 생성

Bandit 상태를 `data/bandit_state.json`에 저장해 서버 재시작해도 유지한다.  
`rules/rules.json`에서 확인한 `conditions[].field` 값들을 `_eval_condition()`의 `field_map`에 모두 반영한다.

```python
import json
import os
import random
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from slowapi import Limiter
from slowapi.util import get_remote_address

from server.llm import generate_nudge

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"https://.*\.cafe24\.com",
    allow_origins=["http://localhost:3000", "http://localhost:8080"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

ALLOWED_EVENTS = {
    "page_view", "scroll_up", "option_click", "cart_add",
    "nudge_shown", "nudge_clicked", "purchase_complete", "exit_intent",
}

# Rules 로드
with open("rules/rules.json", encoding="utf-8") as f:
    _RULES: list[dict] = json.load(f)

# Bandit 상태 — JSON 파일로 영속 저장
_BANDIT_PATH = Path("data/bandit_state.json")
_BANDIT_PATH.parent.mkdir(exist_ok=True)

def _load_bandit() -> dict:
    if _BANDIT_PATH.exists():
        try:
            return json.loads(_BANDIT_PATH.read_text(encoding="utf-8"))
        except Exception:
            pass
    return {}

def _save_bandit(state: dict) -> None:
    _BANDIT_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")

_bandit: dict = _load_bandit()


# --- 스키마 ---

class CollectEvent(BaseModel):
    shop_id: str
    event: str
    product_no: str | None = None
    payload: dict = {}

class NudgeRequest(BaseModel):
    shop_id: str
    product_no: str
    trigger_id: str
    product_price: int = 0
    stock_number: int = 999
    review_count: int = 0
    is_logged_in: bool = False
    product_view_count: int = 1
    exit_intent: bool = False
    product_name: str | None = None
    purchase_count_7d: int | None = None
    brand_voice: str = "친근함"
    tone_keywords: str = ""
    member_id_crypt: str | None = None

class FeedbackRequest(BaseModel):
    shop_id: str
    rule_id: str
    nudge_type: str
    clicked: bool


# --- 엔드포인트 ---

@app.get("/health")
async def health():
    return {"ok": True, "rules": len(_RULES), "bandit_keys": len(_bandit)}


@app.post("/collect")
@limiter.limit("120/minute")
async def collect(event: CollectEvent, request: Request):
    if event.event not in ALLOWED_EVENTS:
        raise HTTPException(400, {"error": "INVALID_EVENT", "message": "허용되지 않은 이벤트입니다"})
    # MVP: 콘솔 출력만 (추후 PostgreSQL 저장)
    print(f"[collect] {event.shop_id} | {event.event} | {event.product_no}")
    return {"ok": True}


@app.post("/nudge")
@limiter.limit("60/minute")
async def nudge(req: NudgeRequest, request: Request):
    matched = _match_rules(req)
    if not matched:
        raise HTTPException(404, {"error": "NO_RULE", "message": "발동 조건을 만족하는 Rule이 없습니다"})

    rule = _thompson_select(matched, req.shop_id)
    is_personalized = bool(req.is_logged_in and req.member_id_crypt)

    copy = await generate_nudge(
        rule_id=rule["id"],
        shop_id=req.shop_id,
        product_no=req.product_no,
        product_ctx={
            "product_price": req.product_price,
            "stock_number": req.stock_number,
            "review_count": req.review_count,
            "brand_voice": req.brand_voice,
            "tone_keywords": req.tone_keywords,
            "product_name": req.product_name,
            "purchase_count_7d": req.purchase_count_7d,
        },
        is_personalized=is_personalized,
    )

    return {"rule_id": rule["id"], "nudge_type": rule["nudge_type"], **copy}


@app.post("/nudge/{nudge_id}/feedback")
async def feedback(nudge_id: str, req: FeedbackRequest):
    key = f"{req.shop_id}:{req.rule_id}:{req.nudge_type}"
    state = _bandit.setdefault(key, {"alpha": 1, "beta": 1})
    if req.clicked:
        state["alpha"] += 1
    else:
        state["beta"] += 1
    _save_bandit(_bandit)  # 파일에 즉시 저장
    return {"ok": True, "alpha": state["alpha"], "beta": state["beta"]}


# --- 내부 헬퍼 ---

def _eval_condition(cond: dict, req: NudgeRequest) -> bool:
    field_map = {
        "pageType": "product",
        "exitIntent": req.exit_intent,
        "isLoggedIn": req.is_logged_in,
        "productViewCount": req.product_view_count,
        "product_price": req.product_price,
        "stock_number": req.stock_number,
        "review_count": req.review_count,
    }
    val = field_map.get(cond["field"])
    if val is None:
        return True
    op, target = cond["op"], cond["value"]
    if op == "eq":  return val == target
    if op == "lt":  return val < target
    if op == "lte": return val <= target
    if op == "gte": return val >= target
    if op == "gt":  return val > target
    return False


def _match_rules(req: NudgeRequest) -> list[dict]:
    result = [
        r for r in _RULES
        if r.get("enabled")
        and r["trigger_id"] == req.trigger_id
        and all(_eval_condition(c, req) for c in r.get("conditions", []))
    ]
    result.sort(key=lambda r: r.get("priority", 0), reverse=True)
    return result[:5]


def _thompson_select(rules: list[dict], shop_id: str) -> dict:
    best_rule, best_sample = rules[0], -1.0
    for rule in rules:
        key = f"{shop_id}:{rule['id']}:{rule['nudge_type']}"
        state = _bandit.get(key, {"alpha": 1, "beta": 1})
        sample = random.betavariate(state["alpha"], state["beta"])
        if sample > best_sample:
            best_rule, best_sample = rule, sample
    return best_rule
```

### 5. `requirements.txt` 생성 또는 업데이트

기존 파일이 있으면 읽어서 누락된 패키지만 추가. 없으면 새로 생성:

```
fastapi
uvicorn[standard]
slowapi
pydantic
google-genai
redis        # 선택사항 — 없어도 서버 동작함
```

### 6. `.gitignore` 확인 및 업데이트

`.gitignore`에 다음이 없으면 추가:

```
.env
data/bandit_state.json
__pycache__/
*.pyc
```

`data/bandit_state.json`은 로컬 학습 데이터라 git에 올리지 않는다.

### 7. 동작 확인

파일 생성 후 실행:

```bash
python -c "from server.llm import generate_nudge; from server.main import app; print('✅ import OK')"
```

에러 있으면 수정 후 재확인.

### 8. 완료 안내

```
✅ PICKIT 백엔드 생성 완료

생성된 파일:
  server/llm.py          — Gemini API 넛지 생성 (Redis 있으면 캐싱, 없어도 동작)
  server/main.py         — FastAPI 서버 (collect / nudge / feedback)
  data/bandit_state.json — Bandit 학습 데이터 (서버 재시작해도 유지)
  requirements.txt       — 패키지 목록

시작하기:
  1. GEMINI_API_KEY 환경변수 설정 필요 (아래 참고)
  2. pip install -r requirements.txt
  3. uvicorn server.main:app --reload

환경변수 설정:
  export GEMINI_API_KEY=여기에_키_입력   # Mac/Linux
  # 또는 .env 파일에 GEMINI_API_KEY=... 작성 후 python-dotenv 사용

동작 테스트:
  curl http://localhost:8000/health
  curl -X POST http://localhost:8000/nudge \
    -H "Content-Type: application/json" \
    -d '{"shop_id":"teonteon20","product_no":"12345","trigger_id":"T07","product_price":25000,"exit_intent":true}'

Redis 없어도 됩니다 — 설치하면 자동으로 캐싱이 켜집니다.
```