Promote PoC scripts to the AI service (item 2.2)

ai-service/ wraps generate_draft/escalation_filter/retrieve_style
behind a single POST /draft endpoint that the Go core will call
internally. poc/tone-corpus/ stays untouched for corpus experiments
and blind-eval; this is the promoted copy meant for the real service.

Verified with TestClient: style_examples path, history/retrieval
path (confirms the earlier scoring fix still ranks the on-topic
exemplar first), escalation short-circuit, and 422 validation when
zero or both of style_examples/history are given.

Still missing: the Go core's actual HTTP client calling this service.

ai-service/README.md

@@ -0,0 +1,50 @@
+# 분신 AI service (Python)
+
+Phase 1 AI 서비스 (`docs/roadmap.md` Phase 1 §2.2). `poc/tone-corpus/`의 세 스크립트
+(`generate_draft.py`·`escalation_filter.py`·`retrieve_style.py`)를 그대로 승격한 내부 API —
+`core-backend/`(Go)가 이 서비스를 내부망 HTTP로 호출한다 (`tech-design.md` §8).
+
+`poc/tone-corpus/`는 그대로 둔다 — 거긴 코퍼스 실험/블라인드 평가용 PoC 도구로 계속 쓰고,
+여기 코드가 실제로 서비스에 쓰이는 "승격된" 버전이다. 두 곳의 로직은 지금 동일하지만, 앞으로
+갈라질 수 있다(예: 여기는 프로덕션 안정성 위주로만 바뀌고, PoC 쪽은 계속 실험적으로 바뀌는 식).
+
+## 실행
+
+```bash
+pip install -r requirements.txt
+export GEMINI_API_KEY=...   # 또는 이 디렉토리에 .env 파일
+uvicorn app.main:app --reload --port 8001
+```
+
+## API
+
+### `POST /draft`
+
+```json
+{
+  "context_lines": ["상대: 오늘 저녁에 뭐 먹을래?"],
+  "style_examples": ["ㅇㅇ 좋지", "나도 궁금하네ㅋㅋ"]
+}
+```
+
+`style_examples`(직접 큐레이션) 또는 `history`(과거 발화 전체, 자동 검색 — `k`로 개수 조절) 중
+**정확히 하나만** 넣는다. 둘 다 넣거나 둘 다 안 넣으면 422.
+
+응답:
+
+```json
+{ "status": "ok" | "escalate" | "no_key", "text": "..." }
+```
+
+- `ok`: `text`는 생성된 답장 초안
+- `escalate`: `text`는 에스컬레이션 사유 (금전/약속 확정/감정적으로 무거운 주제) — 이 경우
+  LLM은 호출되지 않는다 (`escalation_filter.py`가 하드 게이트)
+- `no_key`: `GEMINI_API_KEY`가 없어서 실제 전송될 프롬프트만 `text`에 담아 반환
+
+TestClient로 style_examples/history 두 경로, 에스컬레이션 케이스, 검증 오류(422) 전부 확인함.
+
+## 아직 없는 것
+
+- `core-backend/`에서 이 서비스를 실제로 호출하는 클라이언트 코드 (지금은 이 서비스 자체만 있음)
+- 온디바이스 말투 이력 저장 (이건 클라이언트/코어 백엔드 쪽 책임 — `tech-design.md` §2 참고)
+- 사후 알림 + 되돌리기 로그 (코어 백엔드의 `escalation_logs` 테이블과 연동 필요)

ai-service/app/escalation_filter.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""Rule-based escalation gate -- the first step of the autonomy engine in
+`docs/tech-design.md` §3. Money, appointment confirmation, and emotionally
+heavy content always escalate to the human, at every autonomy level, with
+no exception (`AGENTS.md` absolute safety invariants).
+
+This runs BEFORE any LLM call. `generation.py`'s own [ESCALATE]
+instruction in its system prompt is a second line of defense for whatever
+this misses, not a replacement for it -- a keyword miss must not be the
+only thing standing between a user and an auto-sent money confirmation.
+
+v1 is keyword/regex only, per tech-design.md §3: "100% 정확도를 목표하지
+않는다 -- 애매하면 항상 에스컬레이션 쪽으로 fail-safe." Tune the pattern
+lists against real false positive/negative rates once PoC data comes in.
+
+Identical to poc/tone-corpus/escalation_filter.py -- promoted here
+verbatim per roadmap.md Phase 1 §2.2 ("PoC 스크립트를 FastAPI로 승격").
+"""
+import re
+from dataclasses import dataclass
+
+MONEY_PATTERNS = [
+    r"\d[\d,]*\s*(원|만원|천원)",
+    r"(계좌|입금|송금|이체|환불|결제|대출|카드번호|계좌번호)",
+]
+APPOINTMENT_PATTERNS = [
+    r"(그럼|그러면).{0,10}(맞지|확정|콜)",
+    r"(약속|만나|보자).{0,10}(확정|잡자|정하자)",
+    r"\d{1,2}시.{0,10}(맞지|확정|괜찮|어때)",
+]
+EMOTIONAL_KEYWORDS = [
+    "힘들어", "힘들다", "슬퍼", "슬프다", "우울", "죽고싶", "죽고 싶",
+    "아파", "이별", "헤어졌", "헤어지자", "싸웠어", "화나", "짜증나", "속상",
+]
+
+_MONEY = [re.compile(p) for p in MONEY_PATTERNS]
+_APPOINTMENT = [re.compile(p) for p in APPOINTMENT_PATTERNS]
+
+
+@dataclass
+class EscalationResult:
+    escalate: bool
+    reason: str = ""
+
+
+def check(text):
+    for pat in _MONEY:
+        if pat.search(text):
+            return EscalationResult(True, "금전")
+    for pat in _APPOINTMENT:
+        if pat.search(text):
+            return EscalationResult(True, "약속 확정")
+    for kw in EMOTIONAL_KEYWORDS:
+        if kw in text:
+            return EscalationResult(True, "감정적으로 무거운 주제")
+    return EscalationResult(False)

ai-service/app/generation.py

@@ -0,0 +1,85 @@
+"""Tone-matched reply drafting -- the core of `generate_draft.py`
+(`poc/tone-corpus/`), promoted here as a FastAPI-callable library per
+roadmap.md Phase 1 §2.2. This is the server-fallback path described in
+`docs/tech-design.md` §2 ("온디바이스 부족시 서버 LLM 호출, 최소 컨텍스트만
+전송") -- a real on-device model would replace the API call later, but the
+prompt contract (exemplars + context in, one draft out) stays the same.
+"""
+import os
+from pathlib import Path
+
+from .escalation_filter import check as check_escalation
+
+
+def load_dotenv_if_present():
+    """Load KEY=VALUE pairs from the nearest .env (service root preferred)."""
+    if os.environ.get("GEMINI_API_KEY"):
+        return
+    here = Path(__file__).resolve()
+    candidates = [here.parent.parent / ".env", Path.cwd() / ".env"]
+    for env_path in candidates:
+        if not env_path.is_file():
+            continue
+        for raw in env_path.read_text(encoding="utf-8").splitlines():
+            line = raw.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, _, value = line.partition("=")
+            key, value = key.strip(), value.strip().strip("'").strip('"')
+            if key and key not in os.environ:
+                os.environ[key] = value
+        break
+
+
+SYSTEM_PROMPT = """너는 어떤 사람의 '분신'이다. 아래 예시 발화들의 말투(어휘, 문장 길이, 이모티콘 습관, 격식 정도)를 \
+그대로 따라서, 대화의 마지막 메시지에 대한 답장 '초안 하나만' 자연스러운 한국어로 작성해라.
+
+지켜야 할 것:
+- 답장 초안만 출력한다. 설명, 인사말, 따옴표를 덧붙이지 않는다.
+- 금전, 약속 시간 확정, 감정적으로 무거운 주제라고 판단되면 초안 대신 정확히 이 문장만 출력한다: \
+[ESCALATE] 이 내용은 본인 확인이 필요합니다.
+- 예시에 없는 존댓말/반말을 새로 만들지 말고, 예시의 격식 수준을 그대로 유지한다.
+
+(이 지침은 2차 방어선이다 -- 1차는 escalation_filter.py의 규칙 기반 하드 게이트로, 이미 걸러진
+내용은 여기까지 오지 않는다. 이 지침이 남아있는 이유는 규칙이 놓친 케이스를 위한 것이다.)"""
+
+
+def build_user_prompt(style_examples, context_lines):
+    examples = "\n".join(f"- {s}" for s in style_examples)
+    context = "\n".join(context_lines)
+    return f"""[말투 예시]\n{examples}\n\n[최근 대화]\n{context}\n\n위 대화의 마지막 메시지에 대한 답장 초안:"""
+
+
+def last_incoming_text(context_lines):
+    """The message needing a reply -- last line, minus its '상대: '/'나: ' prefix."""
+    if not context_lines:
+        return ""
+    last = context_lines[-1]
+    return last.split(": ", 1)[1] if ": " in last else last
+
+
+def draft_reply(style_examples, context_lines, model="gemini-2.5-flash", api_key=None):
+    """Returns (status, text). status is one of "escalate" | "no_key" | "ok".
+
+    "escalate": text is the escalation reason (금전/약속 확정/감정적으로 무거운 주제).
+    "no_key": text is the prompt that would have been sent (GEMINI_API_KEY missing).
+    "ok": text is the generated draft.
+    """
+    gate = check_escalation(last_incoming_text(context_lines))
+    if gate.escalate:
+        return "escalate", gate.reason
+
+    api_key = api_key or os.environ.get("GEMINI_API_KEY")
+    if not api_key:
+        return "no_key", build_user_prompt(style_examples, context_lines)
+
+    from google import genai  # pip install google-genai
+    from google.genai import types
+
+    client = genai.Client(api_key=api_key)
+    resp = client.models.generate_content(
+        model=model,
+        contents=build_user_prompt(style_examples, context_lines),
+        config=types.GenerateContentConfig(system_instruction=SYSTEM_PROMPT, max_output_tokens=300),
+    )
+    return "ok", resp.text.strip()

ai-service/app/main.py

@@ -0,0 +1,51 @@
+from typing import List, Optional
+
+from fastapi import FastAPI
+from pydantic import BaseModel, model_validator
+
+from .generation import draft_reply, last_incoming_text, load_dotenv_if_present
+from .retrieve_style import retrieve as retrieve_style_examples
+
+app = FastAPI(title="분신 AI service")
+
+
+@app.on_event("startup")
+def startup():
+    load_dotenv_if_present()
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok"}
+
+
+class DraftRequest(BaseModel):
+    context_lines: List[str]
+    style_examples: Optional[List[str]] = None
+    history: Optional[List[str]] = None
+    k: int = 6
+    model: str = "gemini-2.5-flash"
+
+    @model_validator(mode="after")
+    def check_exactly_one_style_source(self):
+        if bool(self.style_examples) == bool(self.history):
+            raise ValueError("provide exactly one of style_examples or history")
+        return self
+
+
+class DraftResponse(BaseModel):
+    status: str
+    text: str
+
+
+@app.post("/draft", response_model=DraftResponse)
+def draft(req: DraftRequest):
+    if req.style_examples is not None:
+        style_examples = req.style_examples
+    else:
+        style_examples = retrieve_style_examples(
+            req.history, last_incoming_text(req.context_lines), k=req.k
+        )
+
+    status, text = draft_reply(style_examples, req.context_lines, model=req.model)
+    return DraftResponse(status=status, text=text)

ai-service/app/retrieve_style.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Keyword/recency-based retrieval -- the search step of the personalization
+layer in `docs/tech-design.md` §2-1: "지금부터 임베딩 인프라를 먼저 만들지
+않는다" -- so this is plain token overlap + recency, not embeddings.
+
+Identical to poc/tone-corpus/retrieve_style.py -- promoted here verbatim
+per roadmap.md Phase 1 §2.2.
+"""
+import re
+
+TOKEN_RE = re.compile(r"[가-힣A-Za-z0-9]+")
+
+
+def tokenize(text):
+    return set(TOKEN_RE.findall(text))
+
+
+def _jaccard(a, b):
+    if not a or not b:
+        return 0.0
+    return len(a & b) / len(a | b)
+
+
+def retrieve(history_messages, query_text, k=6):
+    """history_messages: oldest-first list of past messages by the person.
+    Ranked by keyword overlap with query_text first; recency only breaks
+    ties. A weighted-sum of the two was tried and dropped -- Korean short
+    messages rarely share more than one or two tokens (no particle
+    stripping here, so "핀란드" and "핀란드는" don't match each other),
+    so any nonzero recency weight ended up drowning out real overlap and
+    just returning the most recent messages regardless of topic."""
+    query_tokens = tokenize(query_text)
+    n = len(history_messages)
+    scored = []
+    for i, msg in enumerate(history_messages):
+        overlap = _jaccard(tokenize(msg), query_tokens)
+        recency = i / max(n - 1, 1)
+        scored.append(((overlap, recency), msg))
+    scored.sort(key=lambda pair: pair[0], reverse=True)
+    return [msg for _, msg in scored[:k]]

ai-service/requirements.txt

@@ -0,0 +1,4 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.0.0
+google-genai>=1.0.0

docs/roadmap.md

@@ -43,8 +43,10 @@
 `backend/`(Python 프로토타입)는 그대로 참고용으로 남겨둔다 — `core-backend/`(Go)가 실제로 쓰는 것.
 
 **2.2 AI 서비스** (Python, PoC 스크립트 → 내부 API로 승격)
-- [ ] `poc/tone-corpus/generate_draft.py`·`escalation_filter.py`·`retrieve_style.py`를 감싸는
+- [x] `poc/tone-corpus/generate_draft.py`·`escalation_filter.py`·`retrieve_style.py`를 감싸는
-  FastAPI 서비스로 승격 (Go 코어가 내부망 HTTP로 호출)
+  FastAPI 서비스로 승격 — `ai-service/` (`POST /draft`, style_examples/history 두 경로 +
+  에스컬레이션 하드게이트 + 검증 오류 전부 실제 테스트로 확인함)
+- [ ] Go 코어가 이 서비스를 실제로 호출하는 클라이언트 코드 (`core-backend/`에서 `AI_SERVICE_URL` 사용)
 - [ ] 자율성 엔진(L0~L2) 오케스트레이션: 에스컬레이션 게이트 → 검색 → 초안 생성 → 승인/자동발송 분기
   (`tech-design.md` §3 흐름 그대로) — 이 오케스트레이션이 Go 코어와 Python AI 서비스 중 어디
   책임인지는 구현 시작 시 정할 것 (에스컬레이션 하드게이트는 Go 코어에 두는 게 안전선 원칙상 더 맞을 수 있음)
@@ -93,8 +95,8 @@
 2. [x] 2.1 코어 백엔드 Go 구현 — `core-backend/` (가입·메시지·WebSocket 릴레이 완료, 푸시 알림만 남음).
    병행하려던 2.3 Flutter 채팅 UI 뼈대는 **이 작업 환경에 Flutter/Dart SDK가 없어 빌드 검증이
    불가능**해서 보류 — Flutter는 Windows에 SDK가 설치된 환경(본인 로컬)에서 시작
-3. [~] 2.2 AI 서비스 (Python, PoC 스크립트를 FastAPI로 승격) — **순서를 앞당김**(원래 2.3과 병행
+3. [~] 2.2 AI 서비스 — `ai-service/`(Python) 완료. Go 코어에서 실제 호출하는 연동 코드는 아직 —
-   예정이었으나 2.3이 막혀서), 이 환경에서 계속 진행 가능 — **다음 작업**
+   **다음 작업**. 2.3 Flutter는 여전히 로컬 환경 대기 중
 4. [ ] 2.3 나머지 UX(온보딩·설정·뱃지)
 5. [ ] 2.4/2.5 안전장치·QA
 6. [ ] §3 확정 (PoC 결과 필요 — 1~5번 전부 끝난 뒤에만) → 2.6 베타 오픈