hbyang
8ae52b0a82
요약 성능지표(No.7, ROUGE 65점) 실험 일습을 커밋한다. - app/engine/summary_consensus.py: 후보 5개 생성 후 합의 선택(select_consensus). 문자/어절 2-gram 가중 합의로 고르며, 유효 후보가 없으면 valid=False 를 낸다. - app/engine/summary_grounded.py: 근거 추출 후 압축하는 2단계 생성. - summarizer.py: 추출 전략 3종(textrank/coverage/lead). coverage 는 MMR 로 중복 문장을 눌러 문서 전체를 넓게 담는다. 기본값은 textrank 로 유지한다. 스크립트는 생성·평가·검증을 분리했다. verify_* 는 API 호출 없이 저장된 산출물만 재계산하는 독립 검증기라 지표를 공용 함수로 합치지 않는다. 합치면 검증이 성립하지 않는다. tune_summarizer 는 사람 검수 참조(build_summary_annotation_packet -> export_summary_annotations)를 입력으로 받아 선택과 최종 보고를 분리한다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""스토리 요약/분석 모듈 (계획서 과제2 ②, p.13 / p.21 고도화).
|
|
|
|
계획서 요약 시스템 흐름:
|
|
입력 ⇒ 추출적 요약(중요 정보 식별) ⇒ 추상적 요약(내용 재구성) ⇒ 최종 요약
|
|
|
|
본 모듈은 데이터(요약 정답셋) 없이도 동작하는 부분까지 구현한다.
|
|
|
|
① 추출적 요약 (extractive) — 정답셋 불필요. 비지도 TextRank 변형.
|
|
문장 분할 → 문장 간 lemma 코사인 유사도 그래프 → PageRank 중심성 →
|
|
상위 문장 선택. (계획서: "구성 요소 추출 모델 활용해 핵심 내용 추출")
|
|
② 추상적 요약 (abstractive) — LLM 훅(옵션). 키 없으면 추출적 결과로 폴백.
|
|
자체 sLLM(고려대 2차저작 생성지원 모델) 으로 교체할 자리.
|
|
③ 통합 요약 (hybrid) — ①의 핵심 문장을 ②의 입력으로 (계획서 통합 요약 시스템).
|
|
|
|
사용자 맞춤형 옵션(요약 상세도/강조 주제)은 계획서 2단계
|
|
'사용자 맞춤형 요약' 반영.
|
|
|
|
평가지표 No.7(N-gram ROUGE)는 scripts/eval_rouge.py 로 측정한다 (정답셋 들어오면).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import re
|
|
from collections import Counter
|
|
from dataclasses import dataclass
|
|
|
|
from app.engine.structural import extract_lemmas
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DETAIL_RATIOS = {"brief": 0.2, "standard": 0.3, "detailed": 0.45}
|
|
EXTRACTIVE_STRATEGIES = {"textrank", "coverage", "lead"}
|
|
|
|
# 문장 분할 — 종결부호 기준 (한국어 '다./요./까?/!' + 줄바꿈)
|
|
_SENT_SPLIT = re.compile(r"(?<=[.!?。…])\s+|\n+")
|
|
|
|
|
|
def split_sentences(text: str) -> list[str]:
|
|
raw = _SENT_SPLIT.split(text.strip())
|
|
return [s.strip() for s in raw if s.strip()]
|
|
|
|
|
|
def _lemma_vector(sentence: str) -> Counter:
|
|
return Counter(extract_lemmas(sentence))
|
|
|
|
|
|
def _cosine(a: Counter, b: Counter) -> float:
|
|
if not a or not b:
|
|
return 0.0
|
|
common = set(a) & set(b)
|
|
if not common:
|
|
return 0.0
|
|
dot = sum(a[t] * b[t] for t in common)
|
|
na = math.sqrt(sum(v * v for v in a.values()))
|
|
nb = math.sqrt(sum(v * v for v in b.values()))
|
|
return dot / (na * nb) if na and nb else 0.0
|
|
|
|
|
|
def _textrank_scores(vectors: list[Counter], damping: float = 0.85, iters: int = 30) -> list[float]:
|
|
"""문장 그래프 PageRank. 정답셋 불필요한 비지도 중심성."""
|
|
n = len(vectors)
|
|
if n == 0:
|
|
return []
|
|
if n == 1:
|
|
return [1.0]
|
|
|
|
# 유사도 인접행렬 (자기 자신 제외)
|
|
sim = [[0.0] * n for _ in range(n)]
|
|
for i in range(n):
|
|
for j in range(i + 1, n):
|
|
s = _cosine(vectors[i], vectors[j])
|
|
sim[i][j] = sim[j][i] = s
|
|
|
|
# 행 정규화
|
|
row_sum = [sum(sim[i]) for i in range(n)]
|
|
scores = [1.0 / n] * n
|
|
for _ in range(iters):
|
|
new = [(1 - damping) / n] * n
|
|
for i in range(n):
|
|
for j in range(n):
|
|
if i == j or row_sum[j] == 0:
|
|
continue
|
|
new[i] += damping * scores[j] * sim[j][i] / row_sum[j]
|
|
scores = new
|
|
return scores
|
|
|
|
|
|
def _normalize(scores: list[float]) -> list[float]:
|
|
"""0~1 정규화. 모든 값이 같으면 동률로 취급한다."""
|
|
if not scores:
|
|
return []
|
|
low, high = min(scores), max(scores)
|
|
if high - low < 1e-12:
|
|
return [1.0] * len(scores)
|
|
return [(score - low) / (high - low) for score in scores]
|
|
|
|
|
|
def _coverage_scores(sentences: list[str], vectors: list[Counter]) -> list[float]:
|
|
"""문서의 고유 내용어를 넓게 담는 문장을 우선하는 점수.
|
|
|
|
TextRank만 쓰면 같은 사건을 반복하는 문장이 상위를 독점할 수 있다. 이 점수는
|
|
문서 전체에서 드문 내용어(인물·사건·시간 등)가 든 문장을 높게 평가하고, 선택
|
|
단계에서 이미 선택한 문장과의 중복을 감점한다. 외부 모델이나 정답셋을 보지 않는
|
|
순수 추출 방식이다.
|
|
"""
|
|
document_frequency: Counter = Counter()
|
|
for vector in vectors:
|
|
document_frequency.update(vector.keys())
|
|
n = max(1, len(sentences))
|
|
lexical = [
|
|
sum(count * math.log((n + 1) / (document_frequency[token] + 0.5))
|
|
for token, count in vector.items()) / max(1, sum(vector.values()))
|
|
for vector in vectors
|
|
]
|
|
# 회고록·에피소드 형식에서는 첫 문장이 사건의 맥락을 잡는 경우가 많다. 단,
|
|
# 위치만으로 결정되지 않게 작은 사전 등록 가중치만 둔다.
|
|
position = [1.0 - (index / max(1, n - 1)) * 0.35 for index in range(n)]
|
|
centrality = _normalize(_textrank_scores(vectors))
|
|
lexical = _normalize(lexical)
|
|
return [0.50 * c + 0.38 * l + 0.12 * p
|
|
for c, l, p in zip(centrality, lexical, position)]
|
|
|
|
|
|
def _coverage_select(vectors: list[Counter], base_scores: list[float], k: int) -> list[int]:
|
|
"""MMR 방식으로 중요하면서 서로 다른 정보를 담는 문장 k개를 고른다."""
|
|
chosen: list[int] = []
|
|
remaining = set(range(len(vectors)))
|
|
while remaining and len(chosen) < k:
|
|
def candidate_score(index: int) -> tuple[float, float, int]:
|
|
redundancy = max((_cosine(vectors[index], vectors[other]) for other in chosen), default=0.0)
|
|
# 중복 문장은 낮추되, 낮은 중요도의 문장이 끼어들 정도로 과도하게 벌주진 않는다.
|
|
value = base_scores[index] - 0.22 * redundancy
|
|
return (value, base_scores[index], -index)
|
|
|
|
best = max(remaining, key=candidate_score)
|
|
chosen.append(best)
|
|
remaining.remove(best)
|
|
return chosen
|
|
|
|
|
|
@dataclass
|
|
class SummaryResult:
|
|
extractive: str # 추출적 요약 (선택된 원문 문장)
|
|
abstractive: str | None # 추상적 요약 (LLM, 없으면 None)
|
|
final: str # 최종 요약 (abstractive 우선, 없으면 extractive)
|
|
selected_indices: list[int] # 선택된 문장 인덱스 (원문 순서)
|
|
mode: str # "extractive" | "hybrid"
|
|
num_sentences_in: int
|
|
num_sentences_out: int
|
|
detail: str = "standard"
|
|
emphasis: list[str] | None = None
|
|
|
|
|
|
def extractive_summary(
|
|
text: str,
|
|
ratio: float = 0.3,
|
|
max_sentences: int | None = None,
|
|
emphasis: list[str] | None = None,
|
|
detail: str = "standard",
|
|
strategy: str = "textrank",
|
|
) -> SummaryResult:
|
|
"""비지도 추출적 요약 — 정답셋/LLM/외부호출 불필요."""
|
|
sentences = split_sentences(text)
|
|
n = len(sentences)
|
|
clean_emphasis = [term.strip() for term in (emphasis or []) if term.strip()]
|
|
if n == 0:
|
|
return SummaryResult("", None, "", [], "extractive", 0, 0, detail, clean_emphasis)
|
|
if n <= 2:
|
|
joined = " ".join(sentences)
|
|
return SummaryResult(
|
|
joined, None, joined, list(range(n)), "extractive", n, n,
|
|
detail, clean_emphasis,
|
|
)
|
|
|
|
k = max(1, math.ceil(n * ratio))
|
|
if max_sentences is not None:
|
|
k = min(k, max_sentences)
|
|
|
|
if strategy not in EXTRACTIVE_STRATEGIES:
|
|
raise ValueError("strategy must be one of %s" % sorted(EXTRACTIVE_STRATEGIES))
|
|
|
|
vectors = [_lemma_vector(s) for s in sentences]
|
|
scores = _textrank_scores(vectors) if strategy == "textrank" else _coverage_scores(sentences, vectors)
|
|
if strategy == "lead":
|
|
scores = [float(n - index) for index in range(n)]
|
|
|
|
# 사용자가 지정한 주제를 포함한 문장에 명시적 가중치를 준다.
|
|
# TextRank 중심성은 유지하되 강조 요청이 상위 선택에 반영되도록 한다.
|
|
if clean_emphasis:
|
|
lowered_terms = [term.casefold() for term in clean_emphasis]
|
|
for i, sentence in enumerate(sentences):
|
|
matched = sum(term in sentence.casefold() for term in lowered_terms)
|
|
scores[i] += 2.0 * matched / len(lowered_terms)
|
|
|
|
# 상위 k개 문장 선택 → 원문 등장 순서로 재정렬 (가독성)
|
|
top = (_coverage_select(vectors, scores, k) if strategy == "coverage"
|
|
else sorted(range(n), key=lambda i: scores[i], reverse=True)[:k])
|
|
top_sorted = sorted(top)
|
|
summary = " ".join(sentences[i] for i in top_sorted)
|
|
return SummaryResult(
|
|
extractive=summary,
|
|
abstractive=None,
|
|
final=summary,
|
|
selected_indices=top_sorted,
|
|
mode="extractive",
|
|
num_sentences_in=n,
|
|
num_sentences_out=len(top_sorted),
|
|
detail=detail,
|
|
emphasis=clean_emphasis,
|
|
)
|
|
|
|
|
|
_ABSTRACTIVE_PROMPT = """다음은 어떤 글에서 추출한 핵심 문장들이다. 이 내용을 바탕으로
|
|
간결하고 자연스러운 한국어 요약문을 작성하라. 원문에 없는 사실을 지어내지 말 것.
|
|
|
|
[핵심 문장]
|
|
"""
|
|
|
|
|
|
class Summarizer:
|
|
"""통합 요약기 — 추출적(항상) + 추상적(LLM 옵션).
|
|
|
|
계획서 통합 요약 시스템: 추출적으로 중요 문장을 뽑은 뒤 추상적으로 재구성.
|
|
use_llm=False 이거나 키가 없으면 추출적 결과를 최종 요약으로 사용.
|
|
"""
|
|
|
|
def __init__(self, settings=None):
|
|
from app.core.config import get_settings
|
|
self.settings = settings or get_settings()
|
|
|
|
def summarize(
|
|
self,
|
|
text: str,
|
|
ratio: float | None = None,
|
|
max_sentences: int | None = None,
|
|
use_abstractive: bool = True,
|
|
detail: str = "standard",
|
|
emphasis: list[str] | None = None,
|
|
strategy: str = "textrank",
|
|
) -> SummaryResult:
|
|
effective_ratio = ratio if ratio is not None else DETAIL_RATIOS.get(detail, 0.3)
|
|
base = extractive_summary(
|
|
text,
|
|
ratio=effective_ratio,
|
|
max_sentences=max_sentences,
|
|
emphasis=emphasis,
|
|
detail=detail,
|
|
strategy=strategy,
|
|
)
|
|
if not base.extractive:
|
|
return base
|
|
|
|
if use_abstractive and self.settings.use_llm_extractor and self.settings.has_openai:
|
|
abstractive = self._abstractive(base.extractive, detail, emphasis or [])
|
|
if abstractive:
|
|
return SummaryResult(
|
|
extractive=base.extractive,
|
|
abstractive=abstractive,
|
|
final=abstractive,
|
|
selected_indices=base.selected_indices,
|
|
mode="hybrid",
|
|
num_sentences_in=base.num_sentences_in,
|
|
num_sentences_out=base.num_sentences_out,
|
|
detail=detail,
|
|
emphasis=base.emphasis,
|
|
)
|
|
return base
|
|
|
|
def _abstractive(
|
|
self, extractive_text: str, detail: str, emphasis: list[str]
|
|
) -> str | None:
|
|
try:
|
|
from openai import OpenAI
|
|
client = OpenAI(api_key=self.settings.openai_api_key)
|
|
resp = client.chat.completions.create(
|
|
model=self.settings.openai_extraction_model,
|
|
temperature=0.2,
|
|
messages=[
|
|
{"role": "system", "content": "You are a concise Korean summarizer. Never hallucinate."},
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
_ABSTRACTIVE_PROMPT
|
|
+ extractive_text
|
|
+ f"\n\n[\uC0C1세도] {detail}"
|
|
+ (f"\n[\uAC15조 주제] {', '.join(emphasis)}" if emphasis else "")
|
|
),
|
|
},
|
|
],
|
|
)
|
|
return (resp.choices[0].message.content or "").strip() or None
|
|
except Exception as exc: # pragma: no cover - 네트워크/키 의존
|
|
logger.warning("Abstractive summary failed, using extractive: %s", exc)
|
|
return None
|
|
|
|
|
|
_default_summarizer: Summarizer | None = None
|
|
|
|
|
|
def get_summarizer(settings=None) -> Summarizer:
|
|
global _default_summarizer
|
|
if _default_summarizer is None:
|
|
_default_summarizer = Summarizer(settings)
|
|
return _default_summarizer
|