MINERVA Phase C-8 — 지식 문서 생명주기 (수집·전처리·인덱싱·갱신·폐기)

1 왜 지식 문서에도 생명주기인가

지금까지 C26 에이전트 생명주기·C30 스킬 게이트는 코드·프롬프트 단위 생명주기였다. 그러나 RAG 기반 시스템에서 문서·인덱스도 같은 위험을 가진다.

문서 측 위험	결과
새 문서가 즉시 100% query에 노출	잘못된 청크·메타데이터로 인한 hallucination 폭증
오래된 문서를 즉시 제거	기존 사용자에게 인용 깨짐 (URL 유효 X)
외부 source 변경 추적 불가	stale knowledge — 옛 정책 답변
같은 정보가 N개 문서에 중복	검색 점수 분산·일관성 깨짐
권한 다른 컬렉션이 섞임	permission leak — 권한 없는 사용자에 노출

해법은 코드·스킬과 같다 — 단계 게이트로 변경 위험을 분산하고, 자동 검증으로 사람 인지 부담을 줄인다.

2 6단계 생명주기

collected → indexed → canary → active → deprecated → retired
   ↓          ↓         ↓        ↓          ↓            ↓
 수집됨    전처리·    소수      전체     polluted·     검색에서
          청킹·     검색에     query    오래됨        제외 (코드만)
          embedded   노출       대상      sunset 예정

각 단계 전이에 게이트 — 자동 검증 또는 사람 결정. C26·C30과 같은 거버넌스 분류 (auto·pending·governance review).

단계	검색 노출	다음 단계 게이트
collected	0%	전처리 통과 + PII 마스킹
indexed	0%	청킹 품질 + 임베딩 검증 + duplicate 점검
canary	5% (특정 사용자·세그먼트)	7일 모니터링 (인용률·정확도)
active	100%	(수정 시 새 indexed → canary)
deprecated	100% (sunset 예고)	후속 문서 active 또는 sunset 도달
retired	0%	(코드 cleanup PR)

3 단계 1 — Collected (수집)

문서가 raw 상태로 들어옴. 출처·메타·timestamp 기록.

# app/knowledge/collect.py
from pydantic import BaseModel
from datetime import datetime


class RawDocument(BaseModel):
    doc_id: str                          # UUID
    source: str                           # "confluence", "sharepoint", "github_wiki"
    source_url: str
    source_modified: datetime             # 외부에서의 수정 시각
    collected_at: datetime
    title: str
    content_raw: str                      # 원본 (HTML/Markdown/PDF text 등)
    content_hash: str                     # 변경 감지용
    sensitivity: str = "internal"         # public/internal/confidential
    owners: list[str] = []                # 부서·담당자 from source metadata

3.1 수집 채널 — Source 어댑터

# app/knowledge/sources/
class ConfluenceSource:
    async def fetch_changes(self, since: datetime) -> AsyncIterator[RawDocument]:
        async for page in self._api_pages_modified(since):
            yield RawDocument(
                doc_id=f"conf-{page.id}",
                source="confluence",
                source_url=page.url,
                source_modified=page.modified,
                collected_at=datetime.utcnow(),
                title=page.title,
                content_raw=page.body,
                content_hash=hash(page.body),
                sensitivity=page.space_meta.get("sensitivity", "internal"),
                owners=page.space_meta.get("owners", []),
            )


class GitHubWikiSource:
    async def fetch_changes(self, since: datetime) -> AsyncIterator[RawDocument]:
        ...

각 source가 같은 RawDocument 형식을 반환 — 다운스트림 처리 통일. 새 source 추가는 어댑터 한 개.

3.2 변경 감지

webhook이 우선 (실시간), polling은 fallback:

# 매 시간 cron 또는 webhook
async def sync_loop():
    last_sync = db.kb_state.get("last_sync_confluence")
    async for raw_doc in confluence.fetch_changes(since=last_sync):
        existing = db.raw_docs.get(raw_doc.doc_id)
        if existing and existing.content_hash == raw_doc.content_hash:
            continue                      # 변경 없음
        db.raw_docs.upsert(raw_doc)
        await enqueue_for_preprocessing(raw_doc.doc_id)
    db.kb_state.set("last_sync_confluence", datetime.utcnow())

4 단계 2 — Indexed (전처리·청킹·임베딩)

# app/knowledge/preprocess.py
async def preprocess(raw_doc_id: str):
    raw = db.raw_docs.get(raw_doc_id)

    # 1. 형식별 파싱
    text = parse_by_format(raw.content_raw, source=raw.source)

    # 2. PII 마스킹 (C20 패턴)
    text_masked = mask_pii(text)

    # 3. 중복 검사 (similarity hash)
    if has_near_duplicate(text_masked):
        flag_duplicate(raw_doc_id)
        return

    # 4. 청킹 (C32 다음 편에서 깊이)
    chunks = chunk_text(text_masked, strategy="recursive_with_metadata")

    # 5. 임베딩
    embeddings = await embed_batch([c.text for c in chunks])

    # 6. 인덱스 저장
    indexed = IndexedDocument(
        doc_id=raw.doc_id,
        version=str(uuid4()),             # 매 indexing 새 version
        chunks=chunks,
        embeddings=embeddings,
        indexed_at=datetime.utcnow(),
        metadata={
            "source_modified": raw.source_modified,
            "sensitivity": raw.sensitivity,
            "owners": raw.owners,
            "language": detect_language(text),
            "topic_tags": auto_tag(text_masked),    # C21 의도·토픽 분류
        },
        stage="indexed",
    )
    db.indexed_docs.insert(indexed)

4.1 인덱스 검증

def validate_indexed(doc: IndexedDocument) -> tuple[bool, list[str]]:
    issues = []

    if len(doc.chunks) == 0:
        issues.append("no chunks")
    if any(len(c.text) < 50 for c in doc.chunks):
        issues.append("chunk too short")
    if any(np.isnan(e).any() for e in doc.embeddings):
        issues.append("nan in embeddings")
    if not doc.metadata.get("topic_tags"):
        issues.append("auto-tagging failed")

    return (not issues, issues)

검증 실패 시 — indexed 단계로 진입 못 함. 수동 fix 필요한 alert.

5 단계 3 — Canary (소수 노출)

# app/knowledge/promotion.py
def promote_to_canary(doc_id: str, audience_pct: float = 5.0):
    doc = db.indexed_docs.get(doc_id)
    assert doc.stage == "indexed"

    doc.stage = "canary"
    doc.canary_started_at = datetime.utcnow()
    doc.canary_audience_pct = audience_pct
    db.indexed_docs.update(doc)

    schedule_canary_monitor(doc_id, duration_days=7)

5.1 Canary 모니터링

7일간 — retrieval에서 어떻게 동작하는지.

def canary_monitor(doc_id: str):
    snapshot = collect_metrics(doc_id, period_days=7)

    checks = {
        "retrieval_count": snapshot["citations"] >= 5,                        # 사용됨
        "thumbs_up_rate": snapshot["citing_responses_thumbs_up"] >= 0.5,
        "no_hallucination_signal": snapshot["hallucination_flags"] < 2,
        "no_pii_leak": snapshot["pii_in_responses"] == 0,
        "language_consistent": snapshot["lang_mismatch_rate"] < 0.05,
    }

    if all(checks.values()):
        return "ready_for_active"
    elif snapshot["pii_in_responses"] > 0 or snapshot["hallucination_flags"] > 5:
        return "rollback"                   # indexed로 역행 + alert
    else:
        return "continue_canary"

C22 응답 품질에서 정의한 신호가 모두 활용됨 — canary 모니터링은 그 인프라 그대로.

6 단계 4 — Active (전체 검색)

100% 검색 대상. 같은 doc_id의 이전 active 자동 deprecated:

def promote_to_active(doc_id: str):
    doc = db.indexed_docs.get(doc_id)
    assert doc.stage == "canary"

    # 이전 버전 자동 deprecated (sunset 30일)
    for prev in db.indexed_docs.find(
        doc_id=doc.doc_id, stage="active", version_ne=doc.version
    ):
        deprecate(prev.id, sunset_days=30, reason=f"replaced by version {doc.version}")

    doc.stage = "active"
    db.indexed_docs.update(doc)

7 단계 5 — Deprecated (sunset 예고)

여전히 검색되지만 만료 예정:

def deprecate(indexed_id: str, sunset_days: int, reason: str):
    doc = db.indexed_docs.get(indexed_id)
    doc.stage = "deprecated"
    doc.deprecated_at = datetime.utcnow()
    doc.sunset_date = datetime.utcnow() + timedelta(days=sunset_days)
    doc.deprecation_reason = reason
    db.indexed_docs.update(doc)

deprecated 문서가 인용에 포함되면 응답에 sunset 안내 자동 추가:

def render_with_sunset_warning(response: str, citations: list) -> str:
    deprecated_cites = [c for c in citations if doc_stage(c.doc_id) == "deprecated"]
    if deprecated_cites:
        warning = f"\n\n참고: 인용된 문서 일부는 곧 만료됩니다 (sunset {min_sunset(deprecated_cites)})."
        return response + warning
    return response

8 단계 6 — Retired (검색 제거)

sunset 도달 시 자동:

def retire(indexed_id: str):
    doc = db.indexed_docs.get(indexed_id)
    assert doc.stage == "deprecated"
    assert datetime.utcnow() > doc.sunset_date

    doc.stage = "retired"
    doc.retired_at = datetime.utcnow()
    db.indexed_docs.update(doc)

    # 인덱스에서 제거 (vector store, search engine)
    remove_from_indices(indexed_id)

    # raw_docs는 audit·legal 목적으로 보존 (단 마스킹)
    archive_raw(doc.doc_id)

retired는 검색 결과에서 제외되지만 metadata는 audit 목적으로 보존. 진짜 삭제는 retention 정책 만료 시.

9 갱신 — Delta Indexing

같은 source URL의 같은 문서가 변경되면 — 전체 재인덱싱이 아니라 변경 청크만.

async def update_incremental(doc_id: str, new_content: str):
    old = db.indexed_docs.get_active(doc_id)
    new_chunks = chunk_text(new_content)

    # chunk-level diff
    old_hashes = {c.hash for c in old.chunks}
    new_hashes = {c.hash for c in new_chunks}

    added = [c for c in new_chunks if c.hash not in old_hashes]
    removed = [c for c in old.chunks if c.hash not in new_hashes]
    unchanged = [c for c in new_chunks if c.hash in old_hashes]

    # 새 indexed_doc — version은 새로
    new_indexed = IndexedDocument(
        doc_id=doc_id,
        version=str(uuid4()),
        chunks=new_chunks,
        embeddings=await embed_batch([c.text for c in added]) + [
            old.embeddings_by_hash[c.hash] for c in unchanged
        ],
        ...
        stage="indexed",
    )
    db.indexed_docs.insert(new_indexed)

    # canary → active 패턴으로 진행

unchanged chunk의 임베딩 재사용 — 비용 절감 (큰 문서면 90%+ 절감).

10 C24 하네싱과 통합

문서 검색이 Tool Guard와 직접 결합:

def search_with_permission(query: str, user_segment: dict) -> list[Citation]:
    # 1. 사용자가 접근 가능한 collection
    allowed = allowed_collections(user_segment)

    # 2. 검색 — collection·sensitivity 필터
    candidates = vector_store.search(
        query=query,
        filter={
            "collection": {"$in": allowed},
            "sensitivity": {"$lte": user_max_sensitivity(user_segment)},
            "stage": {"$in": ["active", "deprecated"]},
        },
    )
    return candidates

retired는 검색에서 제외, deprecated는 포함되지만 응답에 sunset 안내. sensitivity 분류는 source에서 가져온 metadata 활용.

11 C28 Registry와 연결

스킬이 어느 collection에 의존하는지 registry에 명시:

# skills/finance_summarize/SKILL.md
collections_required:
  - finance_internal
  - finance_external_filings

collections_optional:
  - finance_research

collection이 deprecated → 의존 스킬 owner에게 알림 → 후속 collection으로 마이그레이션 또는 스킬 deprecate.

12 MINERVA 적용

app/knowledge/
├── sources/                     # 어댑터
│   ├── confluence.py
│   ├── sharepoint.py
│   └── github_wiki.py
├── collect.py                   # 수집 + 변경 감지
├── preprocess.py                 # 파싱·PII·dedup·청킹·임베딩
├── promotion.py                  # 단계 전이
├── monitor.py                    # canary 모니터링
├── retention.py                  # sunset·retired·archive
├── search_filtered.py            # 권한·stage 필터 검색
└── audit.py                      # 모든 변경 기록

scripts/
├── collect_all.py               # 모든 source sync (cron)
├── reindex.py                    # 단일 문서 재인덱싱
├── promote_canary.py             # CLI promote
├── retire_due.py                 # cron — sunset 도달 자동 retire
└── coverage_audit.py             # 미발견·중복·갭 탐지

config/
├── sources.yaml                 # source 목록·접근정보
├── collections.yaml             # collection·권한 매트릭스
└── retention.yaml                # 카테고리별 보존 기간

이 구조가 C26 lifecycle·C28 registry·C20 logging 위에 자연스럽게 얹힘.

13 자주 발생하는 함정

13.1 Stale Knowledge

source 변경을 못 잡아 옛 정보로 응답. 정책·가격·인사·기술 표준이 자주 변하는 도메인에서 위험.

해법: - webhook 우선·polling fallback - 매일 freshness check — now - source_modified > threshold인 active 문서 알림 - 응답에 source_modified 표시 (“이 정보는 2025-12 기준”)

13.2 Near-Duplicate Drift

같은 정보가 약간씩 다른 문장으로 N개 문서에 — 검색 점수 분산, 일관성 깨짐.

해법: - MinHash·SimHash로 유사도 99% 이상은 자동 dedupe - 사람 review 큐 (90~99% 유사) — duplicate vs 의도된 중복 판단 - duplicate detection을 indexed 단계 게이트에 강제

13.3 Permission Leak

한 collection에 권한 없는 사용자가 그 문서를 인용한 응답을 받음 — 검색 필터 누락.

해법: - collection·sensitivity 필터를 검색 쿼리 자체에 포함 (post-filter는 위험) - 응답 생성 후 Output Guard가 인용 문서 재검증 - 분기 audit — “권한 없는 collection 인용 0건” 회귀 보장

13.4 Embedding Model Drift

embedding 모델 upgrade → 기존 임베딩과 새 임베딩이 호환 안 됨. 검색 점수 의미 깨짐.

해법: - 모델 변경은 전체 재인덱싱 + canary로 점진 (사용자별 sticky 할당) - 새 모델 인덱스 별도 컬렉션 → 점진 전환 - 절대 부분만 새 모델로 인덱싱 X

13.5 Sensitivity Misclassification

source metadata의 sensitivity가 부정확 — public 표시인데 실제 confidential 정보.

해법: - sensitivity classifier (regex + LLM) 후처리 - 사용자 신고 채널 (“이 답변에 민감 정보 포함”) + 즉시 격리 - sensitivity는 항상 상향 조정만 자동, 하향은 사람 결정

13.6 Forced Retire의 Citation Break

retired 문서를 인용한 과거 응답이 sunset 후 dead link 됨. 사용자 신뢰 ↓.

해법: - citation은 raw_docs (archive) 참조 유지 — retired 문서도 archive에서 조회 가능 - retired 응답에 “이 인용 문서는 만료됨 (archive에서 확인)” 표시 - retention.yaml에 archive 기간 명시 (1~2년)

13.7 Knowledge Graph Cycle

문서 A가 B를 참조, B가 A를 참조 — 갱신 시 무한 루프 위험.

해법: - 의존성 그래프 (C28 패턴)로 cycle 감지 - 갱신은 topological order - 재귀 갱신 limit (5 depth)

14 정리

영역	핵심
6단계	collected·indexed·canary·active·deprecated·retired
수집	source 어댑터 통일 + webhook + polling fallback
전처리	parsing·PII 마스킹·dedupe·청킹·임베딩·자동 tagging
Canary	5% 7일 + retrieval 메트릭·hallucination 신호·PII leak 점검
Active 전이	이전 버전 자동 deprecated (sunset 30일)
Delta indexing	chunk-level diff로 unchanged 임베딩 재사용 (90%+ 비용 절감)
결합	C24 권한 필터·C28 registry collections·C22 신호·C20 retention
함정	stale·duplicate·permission leak·embedding drift·sensitivity·citation break·cycle

15 응용 분야

시나리오	활용
정책 문서 변경 빠른 반영	webhook + 즉시 indexed → canary → active
외부 파일링 (재무 보고서)	source 어댑터 신규 + sensitivity 분류
부서별 collection 분리	C24 권한 매트릭스 + sensitivity 메타
옛 버전 인용 유지	archive_raw + citation 폴백
신규 도입 검증	canary 5% 7일 + 회귀 자동
외부 source down 대응	last cached version으로 fallback (deprecated 표시)

16 관련 주제

선행 학습 (선수)

• C26 에이전트 생명주기 — 6단계 패턴 토대
• C20 대화 로깅 — raw·structured·feature·retention 구조 토대
• C22 응답 품질 평가 — canary 모니터링의 신호
• C24 하네싱 — collection·sensitivity 권한 통합

후속 (Phase C-8)

• C32 청킹 전략 — 본 편 indexed 단계의 청킹 깊이
• C33 지식 품질 모니터링 — Phase C-8 클로저, drift·gap 탐지

Cross-reference

• C28 스킬 레지스트리 — collection 의존 추적
• 03편 RAG 파이프라인 — 검색 단계 토대
• 11-1 Reproducible Build — 임베딩 모델 버전 영구 기록