feat(agent-office): 텔레그램 자연어 의도 분류

agent-office/app/agents/classify_intent.py

@@ -0,0 +1,75 @@
+"""텔레그램 사용자 응답 자연어 분류 — 화이트리스트 우선, 모호 시 LLM."""
+import os
+import json
+import logging
+import httpx
+
+logger = logging.getLogger("agent-office.classify_intent")
+
+CLAUDE_HAIKU_DEFAULT = "claude-haiku-4-5-20251001"
+
+APPROVE_WORDS = {
+    "승인", "시작", "진행", "ok", "okay", "agree",
+    "네", "예", "좋아", "좋아요", "go", "yes", "y",
+}
+REJECT_WORDS = {"반려", "거절", "취소", "no", "nope", "n"}
+
+
+def _get_api_key() -> str:
+    return os.getenv("ANTHROPIC_API_KEY", "")
+
+
+def _get_model() -> str:
+    return os.getenv("CLAUDE_HAIKU_MODEL", CLAUDE_HAIKU_DEFAULT)
+
+
+def classify(text: str) -> tuple[str, str | None]:
+    """returns (intent, feedback) — intent ∈ {approve, reject, unclear}"""
+    if not text:
+        return ("unclear", None)
+    t = text.strip().lower()
+    if t in APPROVE_WORDS:
+        return ("approve", None)
+    if t in REJECT_WORDS:
+        return ("reject", None)
+    # 반려 단어로 시작 + 추가 텍스트
+    for w in REJECT_WORDS:
+        if t.startswith(w):
+            rest = text.strip()[len(w):].lstrip(" ,.-:").strip()
+            if rest:
+                return ("reject", rest)
+    # 승인 단어로 시작 (긍정 의도면 추가 텍스트 무시)
+    for w in APPROVE_WORDS:
+        if t.startswith(w + " ") or t == w:
+            return ("approve", None)
+    return _llm_classify(text)
+
+
+def _llm_classify(text: str) -> tuple[str, str | None]:
+    api_key = _get_api_key()
+    if not api_key:
+        return ("unclear", None)
+    prompt = (
+        "사용자 응답을 분류하세요. JSON으로만 응답.\n"
+        f'응답: "{text}"\n\n'
+        '출력: {"intent":"approve|reject|unclear","feedback":"반려면 수정 방향, 아니면 빈 문자열"}'
+    )
+    try:
+        resp = httpx.post(
+            "https://api.anthropic.com/v1/messages",
+            headers={"x-api-key": api_key, "anthropic-version": "2023-06-01"},
+            json={"model": _get_model(), "max_tokens": 200,
+                  "messages": [{"role": "user", "content": prompt}]},
+            timeout=15,
+        )
+        resp.raise_for_status()
+        text_out = resp.json()["content"][0]["text"]
+        start = text_out.find("{")
+        end = text_out.rfind("}") + 1
+        if start < 0 or end <= start:
+            return ("unclear", None)
+        data = json.loads(text_out[start:end])
+        return (data.get("intent", "unclear"), data.get("feedback") or None)
+    except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, json.JSONDecodeError) as e:
+        logger.warning("LLM 분류 실패: %s", e)
+        return ("unclear", None)

agent-office/requirements.txt

@@ -3,5 +3,6 @@ uvicorn[standard]==0.30.6
 apscheduler==3.10.4
 websockets>=12.0
 httpx>=0.27
+respx>=0.21
 google-api-python-client>=2.100.0
 pytrends>=4.9.2

agent-office/tests/test_classify_intent.py

@@ -0,0 +1,48 @@
+import pytest
+import respx
+from httpx import Response
+from app.agents import classify_intent as ci
+
+
+def test_clear_approve_no_llm(monkeypatch):
+    # Patch _llm_classify so we can assert it wasn't called
+    called = {"n": 0}
+    def fake(text):
+        called["n"] += 1
+        return ("unclear", None)
+    monkeypatch.setattr(ci, "_llm_classify", fake)
+    assert ci.classify("승인") == ("approve", None)
+    assert ci.classify("OK") == ("approve", None)
+    assert ci.classify("진행") == ("approve", None)
+    assert ci.classify("agree") == ("approve", None)
+    assert called["n"] == 0
+
+
+def test_clear_reject_only_no_llm(monkeypatch):
+    monkeypatch.setattr(ci, "_llm_classify", lambda t: ("unclear", None))
+    assert ci.classify("반려") == ("reject", None)
+    assert ci.classify("거절") == ("reject", None)
+
+
+def test_reject_with_text_split(monkeypatch):
+    monkeypatch.setattr(ci, "_llm_classify", lambda t: ("unclear", None))
+    intent, fb = ci.classify("반려, 제목 짧게")
+    assert intent == "reject"
+    assert "제목 짧게" in fb
+
+
+@respx.mock
+def test_ambiguous_calls_llm(monkeypatch):
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "k")
+    respx.post("https://api.anthropic.com/v1/messages").mock(
+        return_value=Response(200, json={"content": [{"type": "text",
+            "text": '{"intent":"reject","feedback":"좀 더 화려하게"}'}]})
+    )
+    intent, fb = ci.classify("음... 좀 더 화려한 분위기가 좋겠어")
+    assert intent == "reject"
+    assert "화려하게" in fb
+
+
+def test_empty_text_returns_unclear():
+    assert ci.classify("") == ("unclear", None)
+    assert ci.classify(None) == ("unclear", None)