AI가 카탈로그를 거버넌스한다 — 제안 워크플로·PII 안전장치·셀프-텔레메트리

def _submit(catalog, cfg, dataset, items, apply_fields=None) -> str:
    """결과 처리 — suggest 면 반입, apply 면 직접 수정 (PII 제외)."""
    if cfg.mode == "apply" and apply_fields:
        catalog.update_dataset(dataset["id"], apply_fields)
        items = [i for i in items if i.get("_apply_field") is None]
    if items:
        catalog.import_suggestions(dataset["id"], items, provider="agent", model=cfg.model)
    ...

def run_pii(catalog, llm, cfg, dataset) -> str:
    """PII 컬럼 감지 — 민감 정보라 모드와 무관하게 항상 제안만 (사람 승인 강제)."""
    parsed, _ = llm.generate_json(build_pii_prompt(dataset, sample), SYSTEM_PROMPT)
    # 실제 존재하는 컬럼만 통과 (모델 환각 방어)
    known = {f["field_path"].lower() for f in dataset.get("schema_fields", [])}
    findings = [p for p in parsed
                if isinstance(p, dict) and str(p.get("name", "")).lower() in known]
    ...
    # suggest 강제 — apply 모드여도 직접 적용하지 않는다
    catalog.import_suggestions(dataset["id"], [item], provider="agent", model=cfg.model)
    return f"PII 제안 {len(findings)}건 (사람 승인 필요)"

SYSTEM_PROMPT = (
    "당신은 데이터 카탈로그의 메타데이터 작성 전문가입니다. "
    "데이터셋 스키마와 샘플을 근거로 정확하고 간결한 한국어 메타데이터를 작성합니다. "
    "근거 없는 추측을 하지 않으며, 지시된 출력 형식을 엄격히 지킵니다."
)

def _worker_loop(catalog, llm, cfg, args) -> None:
    processed: set[int] = set()
    while True:
        summaries = catalog.list_datasets_by_datasource(args.datasource_id)
        targets = [s for s in summaries
                   if not (s.get("description") or "").strip()   # 설명 없음
                   and s["id"] not in processed]                  # 미처리
        for s in targets:
            ds = catalog.get_dataset(s["id"])
            run_for_dataset(catalog, llm, cfg, ds, list(TASK_NAMES))
            processed.add(s["id"])
        if args.once:
            return
        time.sleep(cfg.poll_interval)

def _registration_payload(self, tool_names: list[str]) -> dict:
    return {
        "name": self.name,
        "display_name": "Argus Catalog 어시스턴트",
        "description": "카탈로그 도구를 호출해 근거 기반으로 답하는 tool-use 채팅 어시스턴트.",
        "base_model": self.model,
        "model_provider": self.provider,       # llm_url 로 추정 (ollama/openai/...)
        "protocol": "sse", "streaming": True,
        "pii_handling": "read-only", "hitl_required": False,
        "capabilities": tool_names,            # 노출한 도구 9종
        "use_cases": ["메타데이터 Q&A", "Text-to-SQL", "품질·리니지 설명", "표준 용어 준수 점검"],
    }

telemetry.record(
    token, tool_names=tool_names, status=status,
    latency_ms=int((time.monotonic() - started) * 1000),
    tokens_in=tokens_in, tokens_out=tokens_out, session_id=conv_id,
)

def record(self, token, *, status, latency_ms, tokens_in, tokens_out, ...) -> None:
    """채팅 1건의 호출 텔레메트리를 push — 베스트에포트(예외를 던지지 않음)."""
    if not self.enabled or not token:
        return
    try:
        self._ensure_registered(token, tool_names)
        self._post(token, f"/ai-agents/{self.name}/invocations", {...})
    except (urllib.error.URLError, OSError) as e:
        logger.warning("텔레메트리 적재 연결 실패: %s — 무시", e)