feat(backend): Gemini 기반 소개문 요약 함수 추가

캔버스 미리보기에서 intro/room_intro 원문이 길 때 CSS로 자르는 대신
실제 문장을 줄여 보여주기 위한 첫 단계. DB에는 저장하지 않고 프로세스
메모리 캐시(sha256 키)로 같은 원문의 재호출만 막는다.
This commit is contained in:
김성경 2026-09-14 17:01:00 +09:00
parent c07e2bddf4
commit 3342981bd2
2 changed files with 123 additions and 0 deletions

View File

@ -10,6 +10,7 @@
한때 이 네 가지가 한 파일 500줄에 뭉쳐 있었다. "FAQ 답이 이상하다" 를 고치러 와도
어디를 봐야 할지가 파일 안에서 갈리지 않았다.
"""
import hashlib
import json
from dataclasses import dataclass, field
from typing import Optional
@ -196,6 +197,67 @@ async def generate_copy(
return result
# ── 요약(summarize_text) ──────────────────────────────────────────────────
# ★ generate_copy 와 다르다: 여기서 압축하는 문장은 **이미 승인된 값**이다(fact 로 저장된 intro·
# room_intro). 새 사실을 만드는 게 아니라 같은 내용을 짧게 쓰는 것뿐이라 ground_check 를 다시
# 걸지 않는다 — "사실을 더하지 마라"는 프롬프트 지시로 충분하다.
_SUMMARY_CACHE: dict[str, str] = {}
_SUMMARY_CACHE_MAX = 500
async def summarize_text(
text: str,
*,
model: str = DEFAULT_TEXT_MODEL,
max_retries: int = 2,
client: Optional[httpx.AsyncClient] = None,
) -> Optional[str]:
"""캔버스 미리보기용 축약문. 실패해도 예외를 올리지 않는다 — 호출측은 None 이면 원문을 쓴다.
★ DB 에 남기지 않는다. 같은 원문은 프로세스 메모리 캐시(sha256 키)로 재호출을 막는다
(서버 재시작하면 비워진다 — 요구사항: "DB 저장은 생략하고 프론트 응답에만 실어준다").
"""
stripped = text.strip()
if not stripped:
return None
if not is_configured():
return None
cache_key = hashlib.sha256(stripped.encode("utf-8")).hexdigest()
cached = _SUMMARY_CACHE.get(cache_key)
if cached is not None:
return cached
body = {
"contents": [{"role": "user", "parts": [{
"text": (
"다음 문장을 사실을 더하거나 빼지 말고 한국어 2~3문장, 150자 내외로 "
"간결하게 줄여줘. 요약문만 출력하고 다른 말은 붙이지 마.\n\n" + stripped
),
}]}],
"generationConfig": {"temperature": 0.2},
}
owns_client = client is None
client = client or httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0))
try:
payload = await call(client, model, body, max_retries)
summary = extract_text(payload).strip()
except GeminiError as ex:
LOG.w(f"[gemini-text] 요약 실패: {ex}")
return None
finally:
if owns_client:
await client.aclose()
if not summary:
return None
if len(_SUMMARY_CACHE) >= _SUMMARY_CACHE_MAX:
_SUMMARY_CACHE.clear() # 간단한 캐시 상한 — 관리 도구 트래픽 규모에는 LRU 가 과하다.
_SUMMARY_CACHE[cache_key] = summary
return summary
@dataclass
class GeneratedSong:
"""가사 생성 결과. 곡은 여기서 만들지 않는다 — 작곡은 services/external/suno 다."""

View File

@ -361,3 +361,64 @@ def test_ground_check_grounds_numbers_from_unit_summaries():
grounding = FACTS + gt._unit_facts([{"name": "A동", "facts": {"max_capacity": "4"}}])
ok, _reasons = grounding_copy.ground_check("A동은 최대 4명까지 이용하실 수 있습니다.", grounding)
assert ok is True
# ── 요약(summarize_text) ──────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def _clear_summary_cache():
"""모듈 전역 캐시가 테스트끼리 새어 들어가지 않게 매번 비운다."""
gt._SUMMARY_CACHE.clear()
async def test_summarize_returns_shortened_text():
"""검증: 긴 문장을 요약 API 로 축약한다.
기대결과: 응답 텍스트가 그대로 반환되고, 원문이 요청 프롬프트에 실린다."""
payload = {
"candidates": [{"content": {"parts": [{"text": "짧게 줄인 문장입니다."}]}, "finishReason": "STOP"}],
"usageMetadata": {"promptTokenCount": 300, "candidatesTokenCount": 20},
}
calls = []
async with _client(_ok(payload, calls)) as c:
result = await gt.summarize_text("첫 번째 테스트용 원문입니다. " * 20, client=c)
assert result == "짧게 줄인 문장입니다."
prompt = json.loads(calls[0].content)["contents"][0]["parts"][0]["text"]
assert "첫 번째 테스트용 원문입니다." in prompt
async def test_summarize_skips_when_not_configured():
"""검증: GEMINI_API_KEY 미설정.
기대결과: 호출 자체를 안 하고 None — 캔버스는 원문으로 폴백한다."""
llm.external_api_config.gemini_api_key = ""
result = await gt.summarize_text("두 번째 테스트용 원문입니다. " * 20)
assert result is None
async def test_summarize_caches_repeated_calls():
"""검증: 같은 원문을 두 번 요약 요청한다.
기대결과: 두 번째는 API 를 다시 부르지 않고 캐시된 값을 돌려준다."""
payload = {"candidates": [{"content": {"parts": [{"text": "캐시 확인용 요약"}]}, "finishReason": "STOP"}]}
calls = []
text = "세 번째 테스트용 원문입니다. " * 20
async with _client(_ok(payload, calls)) as c:
first = await gt.summarize_text(text, client=c)
second = await gt.summarize_text(text, client=c)
assert first == second == "캐시 확인용 요약"
assert len(calls) == 1, "같은 원문인데 API 를 두 번 불렀다"
async def test_summarize_returns_none_on_repeated_failure():
"""검증: 재시도까지 전부 5xx 로 실패한다.
기대결과: 예외를 올리지 않고 None — 요약 실패가 캔버스를 깨면 안 된다."""
async with _client(lambda r: httpx.Response(503, text="unavailable")) as c:
result = await gt.summarize_text("네 번째 테스트용 원문입니다. " * 20, client=c, max_retries=1)
assert result is None
async def test_summarize_empty_text_returns_none():
"""검증: 빈 문자열.
기대결과: 호출 없이 None."""
result = await gt.summarize_text(" ")
assert result is None