"""AI 분류 워커 — 도메인/문서타입/태그/요약 생성 + PR-B B-1 tier triage. Legacy 경로 (primary 26B 호출): - app/prompts/classify.txt 기반 classify() - AIClient.summarize() 로 ai_summary 생성 - facet_doctype + ai_suggestion (library 제안) 저장 → ai_domain / ai_sub_group / document_type / ai_confidence / ai_tags / ai_summary / ai_suggestion / facet_doctype / importance 필드 PR-B B-1 tier triage (Mac mini 26B MLX, config.yaml ai.models.triage): - policy.routing.decide_routing 으로 RoutingDecision - policy.prompt_render.render_4b("p3a_short_summary", subject_domain) 로 프롬프트 렌더 - AIClient.call_triage(rendered) 호출 (llm_gate 외부, Mac mini 26B MLX — concurrent 안전성 별 검토) - TriageOutput pydantic validate + JSON 깨짐 시 fallback escalate (R1) - R2 backlog guard: deep_summary 큐 ratio > threshold or pending >= threshold 이면 suppress - R3 head/middle/tail: 260k 초과 시 envelope text_ranges 3조각 → ai_tldr / ai_bullets / ai_analysis_tier='triage' 저장 + analyze_events 기록 → escalate 시 EscalationEnvelope payload 로 deep_summary 큐 enqueue """ from __future__ import annotations import json import re import time from datetime import datetime, timezone import yaml from pydantic import BaseModel, Field, ValidationError from sqlalchemy import text as sql_text from sqlalchemy.ext.asyncio import AsyncSession from ai.client import AIClient, parse_json_response, strip_thinking from ai.envelope import EscalationEnvelope from core.config import settings from core.utils import setup_logger from models.document import Document from models.queue import enqueue_stage from policy.prompt_render import render_4b, policy_version as compute_policy_version from policy.routing import decide_routing from services.document_telemetry import record_analyze_event logger = setup_logger("classify_worker") MAX_CLASSIFY_TEXT = 8000 # legacy classify 입력 상한 TRIAGE_TEXT_LIMIT = 120_000 # p3a summary_triage 입력 상한 (4B) DEEP_PRIMARY_LIMIT = 260_000 # primary context_char_limit (envelope slicing) # settings에서 taxonomy/document_types 로딩 DOCUMENT_TYPES = set(settings.document_types) # facet_doctype 허용값 (실무 문서 유형 — AI 식별 신호, library 자동 분류 제안 트리거) FACET_DOCTYPES = {"발주서", "세금계산서", "명세표", "도면", "증명서", "계획서", "시방서"} # 자료실 자동 분류 제안 대상 (거래 하위) LIBRARY_SUGGESTION_DOCTYPES = {"발주서", "세금계산서", "명세표"} # PR-B prompt_version task 이름 SUMMARY_TRIAGE_TASK = "p3a_short_summary" # R2 — hard escalate 사유 (suppress 되지 않는 critical reason). # 'risk_flag_requires_26b' 는 PR-A domain_policy 가 "이 문서 유형은 26B 필수" 로 지정한 # 신호 (safety_legal_interpretation / chemical_hazard / incident_causation 등). backlog # 가 많아도 억제하지 말 것 — safety 도메인 전수 coverage 보장. HARD_ESCALATE_REASONS = { "long_context", "triage_json_invalid", "low_confidence", "risk_flag_requires_26b", } # ───────────────────────── TriageOutput (R1) ───────────────────────── class TriageOutput(BaseModel): """p3a_short_summary (4B) 응답 스키마. 파싱 실패 시 기본값 + escalate=True fallback.""" tldr: str = "" bullets: list[str] = Field(default_factory=list) tags: list[str] = Field(default_factory=list) doc_type: str = "" time_scope: str | None = None confidence: float = 0.3 high_impact_self_declared: bool = False high_impact_reason: str | None = None recommend_deep_summary: bool = False recommend_entity_pass: bool = False escalate_to_26b: bool = False risk_flags: list[str] = Field(default_factory=list) # Memo Intake Upgrade PR-2B — 메모 의도 분류 hint (선택 응답) # 4B 가 출력하지 않으면 None 유지. AI 자동 events 생성 X (사용자 promote 시점만). event_kind_hint: str | None = None # 'note' | 'task' | 'calendar_event' | 'activity_log' | 'reference' event_kind_confidence: float | None = None # 0.0–1.0 # ───────────────────────── legacy classify (primary) ────────────────── def _get_taxonomy_leaf_paths(taxonomy: dict, prefix: str = "") -> set[str]: """taxonomy dict에서 모든 유효한 경로를 추출""" paths = set() for key, value in taxonomy.items(): current = f"{prefix}/{key}" if prefix else key if isinstance(value, dict): if not value: paths.add(current) else: paths.update(_get_taxonomy_leaf_paths(value, current)) elif isinstance(value, list): if not value: paths.add(current) else: for leaf in value: paths.add(f"{current}/{leaf}") paths.add(current) # 2단계도 허용 (leaf가 없는 경우용) else: paths.add(current) return paths VALID_DOMAIN_PATHS = _get_taxonomy_leaf_paths(settings.taxonomy) def _validate_domain(domain: str) -> str: """domain이 taxonomy에 존재하는지 검증, 없으면 최대한 가까운 경로 찾기""" if domain in VALID_DOMAIN_PATHS: return domain parts = domain.split("/") for i in range(len(parts), 0, -1): partial = "/".join(parts[:i]) if partial in VALID_DOMAIN_PATHS: logger.warning(f"[분류] domain '{domain}' → '{partial}' (부분 매칭)") return partial logger.warning(f"[분류] domain '{domain}' taxonomy에 없음, General/Reading_Notes로 대체") return "General/Reading_Notes" # ───────────────────────── B-1 subject_domain 매칭 ───────────────────── def _match_subject_domain(doc: Document, text: str) -> str: """본문/메타/제목 기반으로 domain_policy.yaml 의 subject_domain 이름 결정. 매칭 우선순위 (feedback_category_vs_ai_domain_axis.md — category 매칭 금지): 1. source_channel (news / law_monitor / memo) 2. **제목(title) 기반 매칭** — 본문에 키워드가 한두 번 스치는 문서까지 잘못 잡지 않도록 제목의 의도를 우선 (예: PPE 선정기준.md 가 본문 MSDS 언급 한 번으로 msds 분류되는 문제 방지) 3. 본문 keywords (첫 8k, 대소문자 무시, 키워드 **2회 이상**) 4. ai_domain prefix — legacy classify 가 채운 taxonomy path 5. generic (fallback) """ sc = (doc.source_channel or "").lower() if sc in ("news", "news_collector"): return "news_item" if sc == "law_monitor": return "safety_reference" if sc == "memo": return "generic" # 제목 기반 매칭 — 우선순위 가장 높음. title 은 사용자 의도 시그널이라 정확. title = (doc.title or "").lower() def _title_hit(words: list[str]) -> bool: return any(w.lower() in title for w in words if w) if _title_hit(["산업안전보건법", "산안법", "중처법", "안전보건법", "시행규칙", "시행령"]): return "safety_reference" if _title_hit(["사고", "재해", "재발방지", "원인분석"]): return "incident_report" if _title_hit(["건강검진", "작업환경측정", "특수건강", "보건관리"]): return "health_record" if _title_hit(["밀폐공간", "추락", "끼임", "감전", "화재", "폭발", "화기작업"]): return "hazard_specific" if _title_hit(["msds", "sds", "물질안전보건자료", "화학물질"]): return "msds" if _title_hit(["위험성평가", "작업허가", "jsa", "sop", "ptw", "ppe", "보호구", "안전작업"]): return "safety_operational" # 본문 keyword — 2회 이상 등장해야 도메인 매칭 (single-mention 오분류 방지) head = (text or "")[:8000].lower() def _body_count(keywords: list[str]) -> int: return sum(head.count(kw.lower()) for kw in keywords if kw) # 우선순위: 구체 먼저 if _body_count(["msds", "sds", "물질안전보건자료"]) >= 2: return "msds" if _body_count(["사고보고", "재해조사", "원인분석", "재발방지", "중대재해"]) >= 2: return "incident_report" if _body_count(["건강검진", "작업환경측정", "보건관리", "특수건강진단"]) >= 2: return "health_record" if _body_count(["밀폐공간", "추락", "끼임", "감전", "화재", "폭발", "중독", "질식"]) >= 2: return "hazard_specific" if _body_count(["위험성평가", "작업허가서", "jsa", "sop", "보호구"]) >= 2: return "safety_operational" if _body_count(["산업안전보건법", "안전보건관리체계", "유해위험방지계획서"]) >= 2: return "safety_reference" if (doc.ai_domain or "").startswith("Industrial_Safety"): return "safety_reference" return "generic" def _slice_text_ranges(doc_id: int, total_len: int) -> dict: """R3 — 26B 에 전달할 원문 슬라이스. total_len ≤ primary limit 이면 단일 range, 초과 시 head/mid/tail 3조각.""" if total_len <= DEEP_PRIMARY_LIMIT: return { "doc_ids": [doc_id], "text_ranges": [{"doc_id": doc_id, "start": 0, "end": total_len}], } head_end = 120_000 mid_start = max(head_end, total_len // 2 - 10_000) mid_end = mid_start + 20_000 tail_start = max(mid_end, total_len - 120_000) return { "doc_ids": [doc_id], "text_ranges": [ {"doc_id": doc_id, "start": 0, "end": head_end}, {"doc_id": doc_id, "start": mid_start, "end": mid_end}, {"doc_id": doc_id, "start": tail_start, "end": total_len}, ], } async def _check_backlog_guard(session: AsyncSession) -> str | None: """R2 — deep_summary 큐 적체 시 soft escalate 억제. 억제 사유 문자열 반환, 아니면 None.""" backlog = settings.ai.deep_summary_backlog window = backlog.window_minutes ratio_threshold = backlog.ratio_threshold pending_threshold = backlog.pending_threshold row = (await session.execute(sql_text(""" SELECT COUNT(*) FILTER (WHERE stage = 'classify') AS classify_n, COUNT(*) FILTER (WHERE stage = 'deep_summary') AS deep_n FROM processing_queue WHERE created_at > NOW() - make_interval(mins => :w) """), {"w": window})).one() classify_n = row.classify_n or 0 deep_n = row.deep_n or 0 ratio = (deep_n / classify_n) if classify_n > 0 else 0.0 pending_deep = int(await session.scalar(sql_text(""" SELECT COUNT(*) FROM processing_queue WHERE stage = 'deep_summary' AND status IN ('pending', 'processing') """)) or 0) if ratio > ratio_threshold or pending_deep > pending_threshold: return f"backlog_guard(ratio={ratio:.2f},pending={pending_deep})" return None def _classify_escalation_reason( triage_out: TriageOutput, input_chars: int, triage_limit: int, confidence_floor: float, routing_decision=None, ) -> str | None: """escalate_to_26b 가 True 이도록 만든 근본 사유 하나를 반환. 아니면 None. 우선순위 (높은 것부터 먼저 매칭): 1. long_context (입력 초과) — hard escalate 2. low_confidence (4B 자체 신뢰도 낮음) — hard escalate 3. self_declare (4B 가 자기 상한 선언) — soft 4. deep_requested (4B 가 recommend_deep_summary=true) — soft 5. PR-A routing policy (high_impact / risk_flag_requires_26b / multi_doc) — soft """ if input_chars > triage_limit: return "long_context" if triage_out.confidence < confidence_floor: return "low_confidence" if triage_out.escalate_to_26b: return "self_declare" if triage_out.recommend_deep_summary: return "deep_requested" # PR-A 정책 반영 — domain_policy.yaml 의 high_impact=true / default_risk_flags 의 # requires_26b=true 때문에 decide_routing 이 escalate_to_26b=True 로 판정하면 # 그 첫 reason 을 사용. safety_reference/msds/hazard_specific/incident_report 등 # safety/health 도메인은 여기서 escalate 됨. if routing_decision and routing_decision.escalate_to_26b: reasons = routing_decision.escalation_reasons if reasons: # high_impact 는 모든 safety 도메인에 공통이라 너무 흔함. 더 구체적인 사유를 우선. for r in reasons: if r != "high_impact": return r return reasons[0] return "policy_escalate" return None def _is_hard_escalate(reason: str | None) -> bool: """R2 — hard escalate 는 suppress 되지 않는다.""" return reason in HARD_ESCALATE_REASONS def _distill(triage_out: TriageOutput, limit: int = 2000) -> str: """26B envelope.distilled_context 용 4B 출력 압축 텍스트.""" parts: list[str] = [] if triage_out.tldr: parts.append(f"TL;DR: {triage_out.tldr}") if triage_out.bullets: parts.append("핵심:") parts.extend(f"- {b}" for b in triage_out.bullets) if triage_out.doc_type: parts.append(f"doc_type: {triage_out.doc_type}") if triage_out.tags: parts.append(f"tags: {', '.join(triage_out.tags)}") return "\n".join(parts)[:limit] # ───────────────────── frontmatter 파싱 (옵션 C) ────────────────────── # YAML frontmatter (--- ... ---) + body 분리. body 가 없거나 frontmatter 가 형식 오류여도 안전하게 fallback. _FM_PATTERN = re.compile("^---\\s*\\n(.*?)\\n---\\s*\\n?(.*)$", re.DOTALL) def _parse_frontmatter(extracted_text: str) -> tuple[dict, str]: """extracted_text 시작에 YAML frontmatter 가 있으면 (frontmatter_dict, body) 반환. 없으면 ({}, extracted_text). YAML 파싱 실패 시도 ({}, extracted_text) 로 안전 fallback. """ if not extracted_text or not extracted_text.startswith("---"): return {}, extracted_text m = _FM_PATTERN.match(extracted_text) if not m: return {}, extracted_text fm_text, body = m.group(1), m.group(2) try: fm = yaml.safe_load(fm_text) if not isinstance(fm, dict): return {}, extracted_text return fm, body except yaml.YAMLError: return {}, extracted_text # frontmatter 우선 인식: code/section/source_pdf/source_pages/source_basis/verified_level/verification_pending # 등 원문 추적 메타데이터는 LLM 이 절대 덮어쓰지 못하게 차단. _FRONTMATTER_PRESERVED_KEYS = { "code", "section", "source_pdf", "source_pages", "source_basis", "verified_level", "verification_pending", "source_type", "kgs_code", } # ───────────────────────── main process ──────────────────────────────── async def process(document_id: int, session: AsyncSession) -> None: """문서 분류 + 요약 + tier triage. 1) Legacy: classify() → ai_domain/document_type/ai_tags/ai_confidence/ai_suggestion 2) Legacy: summarize() → ai_summary 3) PR-B B-1: summary_triage (4B) → ai_tldr/ai_bullets/ai_analysis_tier='triage' 예외 — source_channel='law_monitor': 법령은 외부 source-of-truth (law.go.kr) 보유 + immutable + 자동 재수집. AI 분류는 무가치 + 본문 해석 환각 위험. 26B legacy + 4B triage 전부 skip. 최소 필드만 세팅 후 return → queue_consumer 가 embed/chunk 자동 chain. 참고: feedback_category_vs_ai_domain_axis.md, plan stateless-churning-raccoon.md. """ doc = await session.get(Document, document_id) if not doc: raise ValueError(f"문서 ID {document_id}를 찾을 수 없음") if doc.source_channel == "law_monitor": if not doc.ai_domain: doc.ai_domain = "법령" if not doc.ai_tags: doc.ai_tags = ["법령"] if not doc.importance: doc.importance = "medium" await session.commit() logger.info(f"doc {document_id}: law_monitor → classify skip") return # Web/Blog ingest (devonagent 트랙) — plan db-snuggly-petal.md # queue_consumer override 가 classify 를 skip 시키지만, 우회 경로 (예: 수동 enqueue) # 로 들어왔을 때 안전망. ai_tldr/ai_bullets 같은 LLM 가공은 별 PR (Mac mini derived-worker). if doc.source_channel == "devonagent": from urllib.parse import urlparse if not doc.ai_domain: doc.ai_domain = "Web" if not doc.ai_tags: host = (urlparse(doc.edit_url or "").hostname or "web").lower() doc.ai_tags = [f"Web/{host}"] if not doc.importance: doc.importance = "medium" await session.commit() logger.info(f"doc {document_id}: devonagent → classify skip") return if not doc.extracted_text: raise ValueError(f"문서 ID {document_id}: extracted_text가 비어있음") # ─── 옵션 C: markdown frontmatter 우선 인식 ─────────────────────────── # KGS Code 등 외부 작성 마크다운은 frontmatter 에 정확한 메타가 있다. # title / tags / ai_summary / ai_domain 은 frontmatter 에 있으면 그대로 사용, # 없는 필드만 LLM 호출. code/section/source_pages/verified_level 등 원문 # 추적 메타는 documents.md_frontmatter JSONB 에 보존하고 LLM 이 덮어쓰지 못하게 한다. fm, body = _parse_frontmatter(doc.extracted_text) if fm: # frontmatter 전체를 md_frontmatter JSONB 에 저장 (원문 추적용 single source) doc.md_frontmatter = fm # 우선 반영 (LLM 보다 신뢰도 높음, frontmatter 가 authoritative) if fm.get("title"): doc.title = str(fm["title"]) fm_tags = fm.get("tags") if isinstance(fm_tags, list) and fm_tags: # ai_tags 에 frontmatter 태그 우선 적재 (LLM 이 추가만 가능) doc.ai_tags = [str(t) for t in fm_tags] if fm.get("ai_domain"): doc.ai_domain = str(fm["ai_domain"]) parts = doc.ai_domain.split("/") if len(parts) > 1 and not doc.ai_sub_group: doc.ai_sub_group = parts[1] if fm.get("ai_sub_group"): doc.ai_sub_group = str(fm["ai_sub_group"]) if fm.get("document_type"): doc.document_type = str(fm["document_type"]) if fm.get("ai_summary"): doc.ai_summary = str(fm["ai_summary"]) if fm.get("importance") in ("high", "medium", "low"): doc.importance = fm["importance"] # 핵심 메타 (title + ai_domain + ai_summary) 가 모두 frontmatter 로 채워졌으면 # LLM classify/summarize 스킵. tier triage 도 스킵 (frontmatter 가 더 정확). # frontmatter 미커버 필드는 그대로 두어 향후 필요 시 manual UI 채움. if doc.title and doc.ai_domain and doc.ai_summary: if not doc.ai_confidence: doc.ai_confidence = 1.0 # frontmatter 는 사람이 작성한 단정값 doc.ai_processed_at = datetime.now(timezone.utc) doc.ai_model_version = "frontmatter@manual" await session.commit() logger.info(f"doc {document_id}: frontmatter 옵션 C → classify/summarize/triage 전부 skip") return # 일부만 frontmatter 에 있을 때는 LLM 으로 미설정 필드 보완. 단 _FRONTMATTER_PRESERVED_KEYS # 는 이미 md_frontmatter 에 있으므로 LLM 이 ai_domain/document_type 등에 영향 못 준다. logger.info(f"doc {document_id}: frontmatter 부분 인식 → LLM 으로 미설정 필드 보완") client = AIClient() try: # ─── 1. Legacy classify (primary 26B) ─── truncated = doc.extracted_text[:MAX_CLASSIFY_TEXT] raw_response = await client.classify(truncated) parsed = parse_json_response(raw_response) if not parsed: raise ValueError(f"AI 응답에서 JSON 추출 실패: {raw_response[:200]}") # domain 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시) domain = _validate_domain(parsed.get("domain", "")) if not doc.ai_domain: doc.ai_domain = domain # sub_group은 domain 경로에서 추출 (호환성) parts = domain.split("/") doc.ai_sub_group = parts[1] if len(parts) > 1 else "" # document_type 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시) doc_type = parsed.get("document_type", "") if not doc.document_type: doc.document_type = doc_type if doc_type in DOCUMENT_TYPES else "Note" # confidence confidence = parsed.get("confidence", 0.5) doc.ai_confidence = max(0.0, min(1.0, float(confidence))) # importance importance = parsed.get("importance", "medium") doc.importance = importance if importance in ("high", "medium", "low") else "medium" # tags doc.ai_tags = parsed.get("tags", [])[:5] # source/origin if parsed.get("sourceChannel") and not doc.source_channel: doc.source_channel = parsed["sourceChannel"] # data_origin enum 검증 — AI 가 'knowledge' 같은 doc_purpose 값을 잘못 던지면 # asyncpg InvalidTextRepresentationError → 같은 session 후속 호출 cascade fail. # 허용값(work/external) 외는 skip (NULL 유지). if parsed.get("dataOrigin") and not doc.data_origin: origin = parsed["dataOrigin"] if origin in ("work", "external"): doc.data_origin = origin # 용도 (AI는 빈 값만 채움 — 수동/업로드 명시값 우선) if parsed.get("docPurpose") and not doc.doc_purpose: purpose = parsed["docPurpose"] if purpose in ("business", "knowledge"): doc.doc_purpose = purpose # ─── facet_doctype 식별 (§1 실무 문서 유형 신호) ─── ai_doctype_raw = parsed.get("facet_doctype") ai_doctype = ai_doctype_raw if ai_doctype_raw in FACET_DOCTYPES else None if ai_doctype and not doc.facet_doctype: doc.facet_doctype = ai_doctype # ─── ai_suggestion 저장 (자료실 승인 대기함 제안, §1) ─── if ai_doctype in LIBRARY_SUGGESTION_DOCTYPES: year = doc.facet_year or datetime.now(timezone.utc).year doc.ai_suggestion = { "proposed_category": "library", "proposed_path": f"@library/거래/{year}/{ai_doctype}", "proposed_doctype": ai_doctype, "confidence": doc.ai_confidence, "source_updated_at": ( doc.updated_at.isoformat() if doc.updated_at else None ), "reason": "classify pipeline", } # ─── 2. Legacy 요약 (primary 26B) ─── summary = await client.summarize(doc.extracted_text[:50000]) doc.ai_summary = strip_thinking(summary) # ─── 메타데이터 (legacy 완료) ─── doc.ai_model_version = settings.ai.primary.model doc.ai_processed_at = datetime.now(timezone.utc) logger.info( f"[분류] document_id={document_id}: " f"domain={domain}, type={doc.document_type}, " f"confidence={doc.ai_confidence:.2f}, tags={doc.ai_tags}" ) # ─── 3. PR-B B-1 — tier triage (4B, 실패는 legacy 결과 보존) ─── try: await _run_tier_triage(client, doc, session) except Exception as exc: logger.exception(f"[triage] id={document_id} 전체 실패 — legacy 유지: {exc}") finally: await client.close() async def _run_tier_triage(client: AIClient, doc: Document, session: AsyncSession) -> None: """summary_triage (p3a_short_summary) 경로.""" document_id = doc.id text = doc.extracted_text or "" input_chars = len(text) subject_domain = _match_subject_domain(doc, text) triage_start = time.perf_counter() parse_error: str | None = None triage_out = TriageOutput() # 입력이 triage 한도 초과면 호출 생략하고 long_context 로 escalate if input_chars > TRIAGE_TEXT_LIMIT: # routing_decision 은 long_context 경로에선 content_chars 로 바로 판정 rd = None try: rd = decide_routing( subject_domain=subject_domain, content_chars=input_chars, deterministic_keyword_hits=(), self_declared_high_impact=False, self_declared_risk_flags=(), confidence=0.3, evidence_doc_count=0, ) except Exception: pass await _apply_triage_result( doc=doc, session=session, triage_out=triage_out, subject_domain=subject_domain, input_chars=input_chars, latency_ms=0, escalation_reason="long_context", parse_error=None, routing_decision=rd, ) return try: rendered = render_4b(SUMMARY_TRIAGE_TASK, subject_domain) except Exception as exc: logger.exception(f"[triage] render_4b 실패 subject={subject_domain}: {exc}") return # 템플릿의 {extracted_text} 는 단일중괄호로 format 후 남아있음. str.replace 로 주입. prompt = rendered.replace("{extracted_text}", text[:TRIAGE_TEXT_LIMIT]) try: raw_triage = await client.call_triage(prompt) except Exception as exc: logger.warning(f"[triage] 4B 호출 실패 id={document_id}: {exc}") parse_error = "call_failed" raw_triage = "" latency_ms = int((time.perf_counter() - triage_start) * 1000) # R1 — JSON parsing + schema validation + fallback if raw_triage: try: parsed_triage = parse_json_response(raw_triage) or {} triage_out = TriageOutput.model_validate(parsed_triage) except (ValidationError, ValueError, TypeError) as exc: parse_error = f"parse:{type(exc).__name__}" logger.warning(f"[triage] JSON 파싱/검증 실패 → fallback: {exc}") if parse_error: triage_out = triage_out.model_copy(update={ "escalate_to_26b": True, "confidence": 0.3, }) # decide_routing 먼저 계산 — _classify_escalation_reason 이 PR-A high_impact / risk_flag # 결정을 존중하도록 routing_decision 을 전달. routing_decision = None try: routing_decision = decide_routing( subject_domain=subject_domain, content_chars=input_chars, deterministic_keyword_hits=(), self_declared_high_impact=triage_out.high_impact_self_declared, self_declared_risk_flags=tuple(triage_out.risk_flags), confidence=triage_out.confidence, evidence_doc_count=0, ) except Exception as exc: logger.warning(f"[triage] decide_routing 실패 id={document_id}: {exc}") if parse_error: escalation_reason: str | None = "triage_json_invalid" else: escalation_reason = _classify_escalation_reason( triage_out, input_chars=input_chars, triage_limit=TRIAGE_TEXT_LIMIT, confidence_floor=0.6, routing_decision=routing_decision, ) await _apply_triage_result( doc=doc, session=session, triage_out=triage_out, subject_domain=subject_domain, input_chars=input_chars, latency_ms=latency_ms, escalation_reason=escalation_reason, parse_error=parse_error, routing_decision=routing_decision, ) async def _apply_triage_result( *, doc: Document, session: AsyncSession, triage_out: TriageOutput, subject_domain: str, input_chars: int, latency_ms: int, escalation_reason: str | None, parse_error: str | None, routing_decision=None, ) -> None: """TriageOutput → Document 필드 + R2 suppression + envelope enqueue + audit. routing_decision 은 호출자(_run_tier_triage) 가 이미 계산해 전달. 여기서 다시 계산하지 않는다 (escalation_reason 판정이 routing_decision 결과에 의존하므로 양쪽이 같은 값을 봐야 함). """ document_id = doc.id # Document 필드 (파싱 실패 시에는 legacy ai_summary 가 노출되도록 tldr 비움) if not parse_error: doc.ai_tldr = (triage_out.tldr or "").strip() or None doc.ai_bullets = triage_out.bullets or [] # Memo Intake Upgrade PR-2B — event kind hint (4B 가 출력했을 때만) # 허용 enum 외 값이면 무시 (DB enum 제약). AI worker 는 events row 직접 생성 X. valid_kinds = {"note", "task", "calendar_event", "activity_log", "reference"} hint = (triage_out.event_kind_hint or "").strip().lower() or None if hint in valid_kinds: doc.ai_event_kind = hint try: conf = triage_out.event_kind_confidence if conf is not None and 0.0 <= float(conf) <= 1.0: doc.ai_event_confidence = float(conf) except (TypeError, ValueError): pass doc.ai_analysis_tier = "triage" # R2 — backlog guard (hard 제외 soft escalate 만 억제) suppressed_reason: str | None = None escalate = escalation_reason is not None if escalate and not _is_hard_escalate(escalation_reason): suppressed_reason = await _check_backlog_guard(session) if suppressed_reason: escalate = False escalation_reason = None # enqueue deep_summary if escalate: try: # PR-A envelope.ValidFromStage 기준 — P3a 에서 에스컬레이션은 'summarize_short'. # 내부 task 이름 'summary_triage' 는 analyze_events.prompt_version 에만 쓰고, # envelope.from_stage 는 PR-A 가 정의한 enum 값을 따른다. envelope = EscalationEnvelope( from_stage="summarize_short", escalation_reasons=tuple([escalation_reason] if escalation_reason else []), risk_flags=tuple(routing_decision.risk_flags) if routing_decision else tuple(triage_out.risk_flags), distilled_context=_distill(triage_out), original_pointers=_slice_text_ranges(document_id, input_chars), synthesis_directives=( tuple(routing_decision.synthesis_directives) if routing_decision else () ), user_intent=None, draft_hint=(triage_out.tldr or None), ) await enqueue_stage( session, document_id, "deep_summary", payload={ "envelope": json.loads(envelope.to_json()), "subject_domain": subject_domain, }, ) except Exception as exc: logger.exception(f"[triage] envelope enqueue 실패 id={document_id}: {exc}") # analyze_events try: pv = compute_policy_version(SUMMARY_TRIAGE_TASK) except Exception: pv = None escalation_reasons_list = ( list(routing_decision.escalation_reasons) if routing_decision else ([escalation_reason] if escalation_reason else []) ) shadow_target = ( "primary" if (routing_decision and routing_decision.escalate_to_26b) else "triage" ) await record_analyze_event( doc_id=document_id, user_id=None, mode="summary_triage", text_limit=TRIAGE_TEXT_LIMIT, truncated=input_chars > TRIAGE_TEXT_LIMIT, layers_returned=["tldr", "bullets"] if not parse_error else [], cached=False, latency_ms=latency_ms, model_name=settings.ai.triage.model, prompt_version=(f"{SUMMARY_TRIAGE_TASK}@{pv}" if pv else SUMMARY_TRIAGE_TASK), error_code=parse_error, source="document_server", subject_domain=subject_domain, risk_flags=(list(routing_decision.risk_flags) if routing_decision else list(triage_out.risk_flags)), high_impact_task=(routing_decision.high_impact_task if routing_decision else None), escalation_reasons=escalation_reasons_list, confidence=triage_out.confidence, policy_version=pv, shadow_would_route_to=shadow_target, tier="triage", escalated_to_26b=escalate, suppressed_reason=suppressed_reason, ) logger.info( f"[triage] id={document_id} subject={subject_domain} " f"escalate={escalate} reason={escalation_reason} " f"suppressed={bool(suppressed_reason)} " f"confidence={triage_out.confidence:.2f} " f"tldr_len={len(doc.ai_tldr or '')} bullets={len(doc.ai_bullets or [])}" )