ABAP-Indexing/summarize/llm_client.py

"""LLM 클라이언트 추상화. 백엔드 3종 — api / file / fake.

실제 모델 정보(LLM_BASE_URL/LLM_API_KEY/LLM_MODEL)는 환경변수에서만 읽는다 (계획서 §11.5).

## 백엔드

- `api`  : OpenAI 호환 `/chat/completions`. 429/5xx 는 지수 백오프로 재시도한다.
- `file` : **키 없이 돌리는 입구.** 프롬프트를 `data/llm_jobs/` 에 파일로 내놓고, 누군가(사람이든
           에이전트든)가 응답 JSON 을 채워 넣으면 그걸 읽어 파이프라인을 계속한다.
           키가 없는 상태에서 몇 건만 실제 품질로 돌려보려면 이 경로를 쓴다 → `summarize/jobs.py`
- `fake` : 스키마만 맞는 더미. 파이프라인 배선 검증용(품질 검증 아님).
"""
from __future__ import annotations

import hashlib
import json
import random
import re
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

from config.settings import settings


class LLMClient:
    def complete_json(self, system: str, user: str) -> dict:
        raise NotImplementedError


class PendingResponse(Exception):
    """file 백엔드에서 아직 응답이 채워지지 않은 프롬프트 — 실패가 아니라 '대기'다.

    runner 는 이걸 failed 로 기록하지 않고 대기 건수만 집계한다.
    """

    def __init__(self, job_id: str, prompt_path: Path, response_path: Path) -> None:
        super().__init__(f"응답 대기: {job_id} → {response_path}")
        self.job_id = job_id
        self.prompt_path = prompt_path
        self.response_path = response_path


def _extract_json(content: str) -> dict:
    """모델이 코드펜스/서문을 붙이는 경우까지 감안해 JSON 을 뽑아낸다."""
    try:
        return json.loads(content)
    except json.JSONDecodeError:
        s, e = content.find("{"), content.rfind("}")
        if s >= 0 and e > s:
            return json.loads(content[s : e + 1])
        raise


class OpenAICompatClient(LLMClient):
    """OpenAI 호환 `/chat/completions` 호출. 규격(LLM_PROVIDER)에 따라 헤더만 다르다.

    - openai : Authorization: Bearer <LLM_API_KEY>
    - fabrix : 고객사 사내 LLM. x-openapi-token(Bearer ...) / x-generative-ai-client /
               x-llm-model-id / x-generative-ai-user-email. body 의 model 은 LLM_BODY_MODEL.
    """

    def __init__(self) -> None:
        if not settings.llm_base_url:
            raise RuntimeError("LLM_BASE_URL 환경변수가 필요합니다")
        self.provider = settings.llm_provider
        if self.provider == "fabrix":
            if not (settings.fabrix_client_key and settings.fabrix_openapi_token):
                raise RuntimeError("LLM_PROVIDER=fabrix 는 FABRIX_CLIENT_KEY / FABRIX_OPENAPI_TOKEN 이 필요합니다")
        elif self.provider == "openai":
            if not settings.llm_api_key:
                raise RuntimeError("LLM_BASE_URL / LLM_API_KEY 환경변수가 필요합니다")
        else:
            raise RuntimeError(f"알 수 없는 LLM_PROVIDER: {self.provider} (openai | fabrix)")
        self.base = settings.llm_base_url.rstrip("/")
        self.url = self.base + settings.llm_chat_path
        self.key = settings.llm_api_key
        self.model = settings.llm_model
        # 호출 누적 사용량 — runner가 stats에 실어 보고한다
        self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0,
                      "retries": 0}
        self._lock = threading.Lock()  # 병렬 호출에서 usage 집계가 어긋나지 않게

    def headers(self) -> dict:
        h = {"Content-Type": "application/json"}
        if self.provider == "fabrix":
            token = settings.fabrix_openapi_token.strip()
            if not token.lower().startswith("bearer "):
                token = "Bearer " + token   # 날것으로 보내면 401 (팀원 실측)
            h.update({
                "x-openapi-token": token,
                "x-generative-ai-client": settings.fabrix_client_key,
                "x-llm-model-id": self.model,
                "x-generative-ai-user-email": settings.fabrix_user_email,
            })
        else:
            h["Authorization"] = f"Bearer {self.key}"
        return h

    def body(self, system: str, user: str) -> dict:
        if settings.llm_merge_system:
            messages = [{"role": "user", "content": system + "\n\n" + user}]
        else:
            messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
        body = {
            "model": settings.llm_body_model or self.model,
            "temperature": 0.1,
            "messages": messages,
        }
        if settings.llm_json_mode:
            body["response_format"] = {"type": "json_object"}
        if settings.llm_max_tokens > 0:
            body["max_tokens"] = settings.llm_max_tokens
        if "openrouter" in self.base:
            body["usage"] = {"include": True}  # OpenRouter 확장 — usage.cost(USD 크레딧) 포함
        return body

    def _post_once(self, system: str, user: str) -> dict:
        req = urllib.request.Request(
            self.url,
            data=json.dumps(self.body(system, user), ensure_ascii=False).encode("utf-8"),
            headers=self.headers(),
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=settings.llm_timeout_s) as res:
            return json.loads(res.read().decode("utf-8"))

    def complete_json(self, system: str, user: str) -> dict:
        """429/5xx 재시도 포함. 동시성을 올리면 429 가 늘어나므로 백오프가 필수다.

        (실측: 무료 티어에서 순차 실행만으로도 429 로 unit 6건이 failed 로 굳었다.)
        """
        delay = settings.llm_backoff_base
        last: Exception | None = None
        for attempt in range(settings.llm_max_retries + 1):
            try:
                data = self._post_once(system, user)
                break
            except urllib.error.HTTPError as e:
                last = e
                retryable = e.code == 429 or 500 <= e.code < 600
                if not retryable or attempt >= settings.llm_max_retries:
                    raise
                wait = delay
                hdr = (e.headers or {}).get("Retry-After") if hasattr(e, "headers") else None
                if hdr:
                    try:
                        wait = max(wait, float(hdr))
                    except ValueError:
                        pass
                time.sleep(min(wait, settings.llm_backoff_max) * (1 + random.random() * 0.25))
                delay *= 2
                with self._lock:
                    self.usage["retries"] += 1
            except (urllib.error.URLError, TimeoutError) as e:
                last = e
                if attempt >= settings.llm_max_retries:
                    raise
                time.sleep(min(delay, settings.llm_backoff_max))
                delay *= 2
                with self._lock:
                    self.usage["retries"] += 1
        else:  # pragma: no cover — 위 for 는 break/raise 로만 끝난다
            raise last or RuntimeError("LLM 호출 실패")

        u = data.get("usage") or {}
        with self._lock:
            self.usage["calls"] += 1
            self.usage["prompt_tokens"] += u.get("prompt_tokens", 0)
            self.usage["completion_tokens"] += u.get("completion_tokens", 0)
            self.usage["cost_usd"] += u.get("cost") or 0.0
        return _extract_json(data["choices"][0]["message"]["content"])


# --------------------------------------------------------------------- file 백엔드

JOB_INDEX = "index.jsonl"


def job_id(system: str, user: str) -> str:
    """프롬프트 내용으로 결정되는 id — 같은 프롬프트는 재실행해도 같은 파일을 가리킨다."""
    return hashlib.sha256((system + "\n\x00\n" + user).encode("utf-8")).hexdigest()[:16]


def _meta_from_prompt(user: str) -> dict:
    """프롬프트 본문에서 작업 종류·프로그램·unit 을 긁어 index.jsonl 에 남긴다."""
    def grab(pattern: str) -> str:
        m = re.search(pattern, user, re.M)
        return m.group(1).strip() if m else ""

    return {
        "task": grab(r"^\[작업\]\s*(\S+)"),
        "program": grab(r"^\[프로그램\]\s*(\S+)"),
        # UNIT_PROMPT 는 '[단위] unit_id: ...' 형태라 줄 앞에 [단위] 가 붙는다
        "unit_id": grab(r"^\[단위\]\s*unit_id:\s*(\S+)"),
        "unit_type": grab(r"^unit_type:\s*(\w+)"),
        "window": grab(r"지금은 (\d+)번째 창"),
    }


class FileQueueLLM(LLMClient):
    """프롬프트를 파일로 내놓고 응답 파일을 기다리는 백엔드 (키 불필요).

    complete_json 은
      - `<id>.response.json` 이 있으면 그 JSON 을 돌려준다 (정상 경로)
      - 없으면 `<id>.prompt.md` 를 쓰고 PendingResponse 를 던진다

    응답 파일은 사람이 직접 채워도 되고, 코딩 에이전트가 프롬프트를 읽고 채워도 된다.
    `python -m summarize.jobs` 가 목록·조회·검증 저장을 돕는다.
    """

    def __init__(self, jobs_dir: Path | None = None, *, emit: bool = True) -> None:
        self.dir = Path(jobs_dir or settings.data_llm_jobs)
        self.emit = emit
        self.dir.mkdir(parents=True, exist_ok=True)
        self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0,
                      "retries": 0}
        self.pending: list[str] = []
        self.answered: list[str] = []
        self._lock = threading.Lock()

    def prompt_path(self, jid: str) -> Path:
        return self.dir / f"{jid}.prompt.md"

    def response_path(self, jid: str) -> Path:
        return self.dir / f"{jid}.response.json"

    def _write_prompt(self, jid: str, system: str, user: str) -> None:
        meta = _meta_from_prompt(user)
        header = [
            "<!-- abap-indexing LLM 작업. 이 파일은 읽기용이다.",
            f"     응답은 {jid}.response.json 에 JSON 만 써라 (프롬프트 맨 끝 스키마 그대로).",
            "     저장·검증: python -m summarize.jobs answer " + jid + " --file <응답.json>",
            f"     task={meta['task']} program={meta['program']} unit={meta['unit_id']} "
            f"window={meta['window'] or '1'} -->",
            "",
            "## SYSTEM", "", system, "", "## USER", "", user, "",
        ]
        self.prompt_path(jid).write_text("\n".join(header), encoding="utf-8")
        line = json.dumps({"job_id": jid, **meta,
                           "prompt": self.prompt_path(jid).name,
                           "response": self.response_path(jid).name}, ensure_ascii=False)
        idx = self.dir / JOB_INDEX
        existing = idx.read_text(encoding="utf-8").splitlines() if idx.exists() else []
        if not any(f'"job_id": "{jid}"' in ln for ln in existing):
            with idx.open("a", encoding="utf-8") as f:
                f.write(line + "\n")

    def complete_json(self, system: str, user: str) -> dict:
        jid = job_id(system, user)
        rp = self.response_path(jid)
        if rp.exists():
            data = _extract_json(rp.read_text(encoding="utf-8"))
            with self._lock:
                self.usage["calls"] += 1
                self.answered.append(jid)
            return data
        with self._lock:
            if self.emit:
                self._write_prompt(jid, system, user)
            self.pending.append(jid)
        raise PendingResponse(jid, self.prompt_path(jid), rp)


class FakeLLM(LLMClient):
    """테스트/드라이런용 — 프롬프트 종류([작업] 표식)에 맞는 최소 JSON 을 돌려준다.

    extract_chunks 는 코드의 첫 실행문 한 줄을 조각 하나로 만든다 (파이프라인 검증용).
    """

    def __init__(self) -> None:
        self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0,
                      "retries": 0}

    def complete_json(self, system: str, user: str) -> dict:
        self.usage["calls"] += 1
        task = ""
        for line in user.splitlines():
            if line.startswith("[작업]"):
                task = line.split("]", 1)[1].strip()
                break
        if task == "program_summary":
            return {"program": "", "business_purpose_ko": "(fake) 프로그램 요약", "confidence": 0.0}
        # extract_chunks — '  12| code' 형식의 첫 코드 줄을 조각으로
        m = re.search(r"^\s*(\d+)\| (?!FORM |ENDFORM|METHOD |ENDMETHOD|FUNCTION |ENDFUNCTION|MODULE |ENDMODULE)(\S.*)$",
                      user, re.M)
        chunks = []
        if m:
            no, text = int(m.group(1)), m.group(2)
            chunks.append({"line_start": no, "line_end": no, "first_line": text, "kind": "other",
                           "purpose_ko": "(fake) 조각", "confidence": 0.0})
        return {"unit_purpose_ko": "(fake) unit 요약", "chunks": chunks}


BACKENDS = ("api", "file", "fake")


def create_llm(fake: bool = False, backend: str | None = None, **kwargs) -> LLMClient:
    """백엔드 선택. `backend` 가 우선이고, 하위호환으로 `fake=True` 도 받는다."""
    name = backend or ("fake" if fake else "api")
    if name not in BACKENDS:
        raise ValueError(f"알 수 없는 LLM 백엔드: {name} (가능: {', '.join(BACKENDS)})")
    if name == "fake":
        return FakeLLM()
    if name == "file":
        return FileQueueLLM(**kwargs)
    return OpenAICompatClient()