# pyright: reportMissingImports=false """증분 업데이트 — 새 카톡 데이터로 기존 지식 보강""" from __future__ import annotations import logging from .knowledge_graph import KnowledgeGraph from .vector_store import VectorStore logger = logging.getLogger(__name__) # confidence 단계 정의 (낮은 순서) _CONFIDENCE_LEVELS = ["low", "medium", "high"] class IncrementalUpdater: """새 인사이트를 기존 지식 베이스와 비교하여 추가/보강/충돌 처리""" def __init__(self, vector_store: VectorStore, graph: KnowledgeGraph): """ Args: vector_store: VectorStore 인스턴스 graph: KnowledgeGraph 인스턴스 """ self._store = vector_store self._graph = graph def update( self, new_insights: list[dict], similarity_threshold: float = 0.85 ) -> dict: """새 인사이트를 기존 DB와 비교하여 업데이트. 각 new_insight에 대해: 1. vector_store.search_similar()로 유사 인사이트 검색 2. 가장 유사한 것의 distance 확인: - distance > threshold → 신규 (new): 그대로 추가 - distance <= threshold → 유사 인사이트 존재 - 같은 카테고리 + 같은 타입 → 보강 (augment): key_points 병합, 참여자 추가 - 같은 카테고리 + 다른 타입 → 신규로 취급 - 다른 카테고리 → 신규로 취급 Returns: { "new": int, "augmented": int, "total_processed": int, "new_ids": [str], "augmented_ids": [str] } """ stats: dict = { "new": 0, "augmented": 0, "total_processed": 0, "new_ids": [], "augmented_ids": [], } if not new_insights: return stats # 모든 new_insights를 처리하기 위한 내부 상태 # augment 후 최종 rebuild를 위해 모든 인사이트 수집 all_current_insights: list[dict] = [] for new_insight in new_insights: stats["total_processed"] += 1 new_id = new_insight.get("id", "") new_category = new_insight.get("category", "") new_type = str(new_insight.get("type", "")) # 스토어가 비어 있으면 바로 신규 if self._store.count() == 0: self._store.add_insights([new_insight]) stats["new"] += 1 stats["new_ids"].append(new_id) all_current_insights.append(new_insight) continue # 유사 인사이트 검색 (top_k=1 → 가장 유사한 것만) similar = self._store.search_similar( new_insight.get("title", "") + " " + new_insight.get("summary", ""), top_k=1, ) if not similar: # 검색 결과 없음 → 신규 self._store.add_insights([new_insight]) stats["new"] += 1 stats["new_ids"].append(new_id) all_current_insights.append(new_insight) continue top = similar[0] distance = top["distance"] if distance > similarity_threshold: # 충분히 다름 → 신규 self._store.add_insights([new_insight]) stats["new"] += 1 stats["new_ids"].append(new_id) all_current_insights.append(new_insight) logger.info("신규 인사이트 추가: %s (distance=%.4f)", new_id, distance) else: # 유사 인사이트 존재 → 카테고리/타입 비교 existing_id = top["id"] existing_category = top.get("category", "") # 기존 인사이트의 타입을 가져오기 위해 get_by_id 호출 existing_meta = self._store.get_by_id(existing_id) existing_type = ( str(existing_meta.get("type", "")) if existing_meta else "" ) same_category = new_category == existing_category same_type = new_type == existing_type if same_category and same_type: # 보강 (augment) existing_full = self._get_full_insight(existing_id, existing_meta) augmented = self._augment_insight(existing_full, new_insight) self._store.add_insights([augmented]) stats["augmented"] += 1 stats["augmented_ids"].append(existing_id) all_current_insights.append(augmented) logger.info( "인사이트 보강: %s ← %s (distance=%.4f)", existing_id, new_id, distance, ) else: # 카테고리/타입 불일치 → 신규로 취급 self._store.add_insights([new_insight]) stats["new"] += 1 stats["new_ids"].append(new_id) all_current_insights.append(new_insight) logger.info( "신규 인사이트 추가 (카테고리/타입 불일치): %s (distance=%.4f)", new_id, distance, ) # KnowledgeGraph 재빌드 (현재 그래프 노드 + 새/갱신된 인사이트 병합) if all_current_insights: self._rebuild_graph(all_current_insights) logger.info( "증분 업데이트 완료 — 신규: %d, 보강: %d, 총: %d", stats["new"], stats["augmented"], stats["total_processed"], ) return stats def _augment_insight(self, existing: dict, new_insight: dict) -> dict: """기존 인사이트에 새 정보를 보강. - key_points: 기존 + 새 것 (중복 제거) - participants: 합집합 - related_topics: 합집합 - tags: 합집합 - summary: 기존 유지 (자동 수정 위험) - confidence: 보강 횟수에 따라 상향 (low→medium→high) """ augmented = dict(existing) # key_points 병합 (순서 유지, 중복 제거) existing_kp = list(existing.get("key_points", [])) new_kp = list(new_insight.get("key_points", [])) merged_kp = existing_kp[:] seen_kp = set(existing_kp) for kp in new_kp: if kp not in seen_kp: merged_kp.append(kp) seen_kp.add(kp) augmented["key_points"] = merged_kp # participants 합집합 existing_participants = set(existing.get("participants", [])) new_participants = set(new_insight.get("participants", [])) augmented["participants"] = list(existing_participants | new_participants) # related_topics 합집합 existing_rt = set(existing.get("related_topics", [])) new_rt = set(new_insight.get("related_topics", [])) augmented["related_topics"] = list(existing_rt | new_rt) # tags 합집합 existing_tags = set(existing.get("tags", [])) new_tags = set(new_insight.get("tags", [])) augmented["tags"] = list(existing_tags | new_tags) # summary: 기존 유지 # augmented["summary"] = existing["summary"] # (이미 복사됨) # confidence 상향 (low → medium → high) current_conf = existing.get("confidence", "medium") augmented["confidence"] = self._upgrade_confidence(current_conf) return augmented # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _upgrade_confidence(self, current: str) -> str: """confidence를 한 단계 상향한다: low → medium → high → high.""" try: idx = _CONFIDENCE_LEVELS.index(current) new_idx = min(idx + 1, len(_CONFIDENCE_LEVELS) - 1) return _CONFIDENCE_LEVELS[new_idx] except ValueError: return "medium" def _get_full_insight(self, insight_id: str, meta: dict | None) -> dict: """VectorStore 메타데이터로부터 insight dict를 재구성한다. VectorStore.get_by_id()가 리스트 필드를 자동 역직렬화하므로 key_points, participants 등이 이미 list로 복원됨. """ if meta is None: return {"id": insight_id} result = dict(meta) result.setdefault("key_points", []) result.setdefault("participants", []) result.setdefault("related_topics", []) result.setdefault("tags", []) return result def _rebuild_graph(self, updated_insights: list[dict]) -> None: """KnowledgeGraph를 갱신된 인사이트로 재빌드한다.""" try: self._graph.build_from_insights(updated_insights) logger.info("KnowledgeGraph 재빌드 완료: %d개 노드", len(updated_insights)) except Exception as exc: logger.warning("KnowledgeGraph 재빌드 실패: %s", exc)