"""Langfuse 로 trace/generation 쏘기 — SDK 없이 OTLP/HTTP JSON 한 방 (docs-lib/langfuse.md). v4 는 옛 /api/public/ingestion 이 막혀서(score 만) OTel 엔드포인트로 감. 트레이스 하나 = 루트 span(trace 속성) + 자식 span(generation). LANGFUSE_HOST 가 비어 있으면 전부 no-op. 보내는 건 fire-and-forget: 실패해도 채팅엔 영향 0, 로그만. 두 군데서 부름: - apps/chat/stream.py _finalize → 사용자 단위 trace(누가·어느 세션·질문·최종 답·토큰·시간) - apps/gateway/views.py → FabriX 호출 단위 generation(모델·프롬프트 원문·응답·상태) 둘은 서로 모름(OpenCode 가 사이에 있어 사용자 정보가 게이트웨이까지 안 옴). 나중에 잇고 싶으면 sessionId 로 대조. """ from __future__ import annotations import asyncio import hashlib import json import logging import time import uuid from datetime import datetime, timezone import httpx from django.conf import settings log = logging.getLogger(__name__) TRANSPORT: httpx.AsyncBaseTransport | None = None # 테스트가 MockTransport 꽂는 자리 _pending: set[asyncio.Task] = set() # GC 에 안 먹히게 잡아둠 def enabled() -> bool: return bool(settings.LANGFUSE.get("host")) def now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _trace_hex(trace_id: str) -> str: """아무 문자열 → OTel traceId(16바이트 hex). 같은 문자열이면 같은 trace 로 묶임.""" return hashlib.sha256(trace_id.encode()).hexdigest()[:32] def _nanos(iso_or_none: str | None) -> str: if not iso_or_none: return str(time.time_ns()) dt = datetime.fromisoformat(iso_or_none.replace("Z", "+00:00")) return str(int(dt.timestamp() * 1_000_000_000)) def _attr(k: str, v) -> dict | None: if v is None or v == "" or v == {} or v == []: return None if isinstance(v, bool): return {"key": k, "value": {"boolValue": v}} if isinstance(v, int): return {"key": k, "value": {"intValue": str(v)}} if isinstance(v, float): return {"key": k, "value": {"doubleValue": v}} if isinstance(v, str): return {"key": k, "value": {"stringValue": v}} return {"key": k, "value": {"stringValue": json.dumps(v, ensure_ascii=False)}} def _span(trace_id: str, name: str, attrs: dict, *, start: str | None, end: str | None, parent: str | None, error: str | None) -> dict: span = { "traceId": _trace_hex(trace_id), "spanId": uuid.uuid4().hex[:16], "name": name, "kind": 1, # INTERNAL "startTimeUnixNano": _nanos(start), "endTimeUnixNano": _nanos(end), "attributes": [a for a in (_attr(k, v) for k, v in attrs.items()) if a], "status": {"code": 2, "message": error} if error else {"code": 1}, } if parent: span["parentSpanId"] = parent return span def trace(trace_id: str, name: str, *, userId: str = "", sessionId: str = "", input=None, output=None, metadata: dict | None = None, tags: list[str] | None = None, start: str | None = None, end: str | None = None) -> dict: """루트 span. trace 속성(이름·사용자·세션·태그)은 여기 실림.""" attrs = { "langfuse.trace.name": name, "langfuse.observation.type": "span", "langfuse.user.id": userId, "langfuse.session.id": sessionId, "langfuse.trace.input": input, "langfuse.trace.output": output, "langfuse.observation.input": input, # 루트 span 본문에도 — 트레이스 화면이 루트 span 의 입출력을 보여줌(실측) "langfuse.observation.output": output, "langfuse.trace.tags": tags, **{f"langfuse.trace.metadata.{k}": v for k, v in (metadata or {}).items()}, } return _span(trace_id, name, attrs, start=start, end=end, parent=None, error=None) def generation(trace_id: str, name: str, *, model: str = "", input=None, output=None, usage: dict | None = None, level: str = "DEFAULT", statusMessage: str = "", metadata: dict | None = None, modelParameters: dict | None = None, startTime: str | None = None, endTime: str | None = None, parent: dict | None = None) -> dict: """generation span. parent 로 trace() 결과를 주면 그 밑에 붙고, 없으면 같은 traceId 의 루트. 사용자·세션·이름·태그는 자식에도 복사 — Langfuse 가 필터·집계할 때 span 마다 보기 때문(로컬 실측: 안 하면 빈 값).""" inherited = {a["key"]: a["value"].get("stringValue") for a in (parent or {}).get("attributes", []) if a["key"] in ("langfuse.user.id", "langfuse.session.id", "langfuse.trace.name", "langfuse.trace.tags")} attrs = { **inherited, "langfuse.observation.type": "generation", "langfuse.observation.model.name": model, "langfuse.observation.input": input, "langfuse.observation.output": output, "langfuse.observation.usage_details": usage, "langfuse.observation.model_parameters": modelParameters, "langfuse.observation.level": level, "langfuse.observation.status_message": statusMessage, **{f"langfuse.observation.metadata.{k}": v for k, v in (metadata or {}).items()}, } return _span(trace_id, name, attrs, start=startTime, end=endTime, parent=parent["spanId"] if parent else None, error=statusMessage if level == "ERROR" else None) def usage_of(inp: int | None, out: int | None) -> dict | None: if inp is None and out is None: return None return {"input": inp or 0, "output": out or 0, "total": (inp or 0) + (out or 0)} def _otlp_body(spans: list[dict]) -> dict: return {"resourceSpans": [{ "resource": {"attributes": [_attr("service.name", "codeassist-backend")]}, "scopeSpans": [{"scope": {"name": "codeassist"}, "spans": spans}], }]} async def send(spans: list[dict]) -> bool: """span 묶음 하나 전송. OTLP 는 200 + 빈 JSON 이면 성공, partialSuccess 있으면 일부 실패.""" cfg = settings.LANGFUSE if not cfg.get("host") or not spans: return False try: async with httpx.AsyncClient(timeout=5.0, transport=TRANSPORT) as client: resp = await client.post( cfg["host"].rstrip("/") + "/api/public/otel/v1/traces", json=_otlp_body(spans), headers={"x-langfuse-ingestion-version": "4"}, auth=(cfg.get("public_key", ""), cfg.get("secret_key", "")), ) if resp.status_code != 200: log.warning("langfuse ← %s %s", resp.status_code, resp.text[:200]) return False partial = (resp.json() or {}).get("partialSuccess") or {} if partial.get("rejectedSpans"): log.warning("langfuse 일부 거부: %s", partial) return False return True except Exception as e: # noqa: BLE001 — 관측용이라 절대 본 흐름 안 깨뜨림 log.warning("langfuse 전송 실패: %s: %s", type(e).__name__, e) return False def send_later(spans: list[dict]) -> None: """지금 흐름 안 막고 백그라운드로. 이벤트 루프 없으면(동기 테스트) 조용히 버림.""" if not enabled() or not spans: return try: task = asyncio.get_running_loop().create_task(send(spans)) except RuntimeError: return _pending.add(task) task.add_done_callback(_pending.discard)