lee-hyeon-cheolandClaude Fable 5.1
68b55907f0
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""LLM 관측 전송 — Langfuse 또는 Phoenix 로, SDK 없이 OTLP/HTTP JSON 한 방 (docs-lib/langfuse.md, docs-lib/phoenix.md).
|
|
|
|
대상은 설정으로 고름(둘 다 켜도 됨):
|
|
LANGFUSE_HOST → {host}/api/public/otel/v1/traces (Basic 인증, langfuse.* 속성)
|
|
PHOENIX_HOST → {host}/v1/traces (인증 없음, OpenInference 속성: openinference.span.kind, input.value …)
|
|
Phoenix 는 JSON 을 안 받아서(415, 2026-09-22 실측) protobuf 로 바꿔 보냄 — opentelemetry-proto 필요
|
|
속성은 두 벌을 같은 span 에 같이 실음 — 각자 자기 것만 읽고 나머진 metadata 로 떨어짐.
|
|
|
|
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") or settings.LANGFUSE.get("phoenix"))
|
|
|
|
|
|
def _oi(kind: str, *, input=None, output=None, model: str = "", usage: dict | None = None,
|
|
userId: str = "", sessionId: str = "", tool: str = "", metadata: dict | None = None) -> dict:
|
|
"""OpenInference(Phoenix) 속성 한 벌. 값이 없으면 _attr 이 걸러줌."""
|
|
js = lambda v: v if v is None or isinstance(v, str) else json.dumps(v, ensure_ascii=False) # noqa: E731
|
|
return {
|
|
"openinference.span.kind": kind,
|
|
"input.value": js(input),
|
|
"input.mime_type": None if input is None or isinstance(input, str) else "application/json",
|
|
"output.value": js(output),
|
|
"output.mime_type": None if output is None or isinstance(output, str) else "application/json",
|
|
"llm.model_name": model,
|
|
"llm.token_count.prompt": (usage or {}).get("input"),
|
|
"llm.token_count.completion": (usage or {}).get("output"),
|
|
"llm.token_count.total": (usage or {}).get("total"),
|
|
"session.id": sessionId,
|
|
"user.id": userId,
|
|
"tool.name": tool,
|
|
"metadata": js(metadata) if metadata else None,
|
|
}
|
|
|
|
|
|
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,
|
|
usage: dict | None = None, model: str = "", error: str | None = None) -> dict:
|
|
"""루트 span. trace 속성(이름·사용자·세션·태그)은 여기 실림.
|
|
usage 를 주면 루트 자체가 generation — 목록 한 줄 = 질문/답/토큰, 밑에 도구 호출만 트리로."""
|
|
attrs = {
|
|
"langfuse.trace.name": name,
|
|
"langfuse.observation.type": "generation" if usage else "span",
|
|
"langfuse.observation.model.name": model,
|
|
"langfuse.observation.usage_details": usage,
|
|
"langfuse.observation.level": "ERROR" if error else "DEFAULT",
|
|
"langfuse.observation.status_message": error or "",
|
|
"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()},
|
|
**_oi("LLM" if usage else "CHAIN", input=input, output=output, model=model, usage=usage,
|
|
userId=userId, sessionId=sessionId, metadata={**(metadata or {}), "tags": tags} if (metadata or tags) else None),
|
|
}
|
|
return _span(trace_id, name, attrs, start=start, end=end, parent=None, error=error)
|
|
|
|
|
|
def tool_span(trace_id: str, name: str, *, parent: dict, input=None, output=None, title: str = "",
|
|
start: str | None = None, end: str | None = None, error: str | None = None) -> dict:
|
|
"""도구 호출 하나(SAP 조회·위키 read·grep…). 트리에서 루트 밑에 순서대로 보임."""
|
|
inherited = {a["key"]: a["value"].get("stringValue") for a in parent.get("attributes", [])
|
|
if a["key"] in ("langfuse.user.id", "langfuse.session.id", "langfuse.trace.name", "langfuse.trace.tags")}
|
|
attrs = {
|
|
**inherited,
|
|
"langfuse.observation.type": "tool",
|
|
"langfuse.observation.input": input,
|
|
"langfuse.observation.output": output,
|
|
"langfuse.observation.level": "ERROR" if error else "DEFAULT",
|
|
"langfuse.observation.status_message": error or "",
|
|
"langfuse.observation.metadata.title": title,
|
|
**_oi("TOOL", input=input, output=output, tool=name,
|
|
userId=inherited.get("langfuse.user.id", ""), sessionId=inherited.get("langfuse.session.id", ""),
|
|
metadata={"title": title} if title else None),
|
|
}
|
|
return _span(trace_id, name, attrs, start=start, end=end, parent=parent["spanId"], error=error)
|
|
|
|
|
|
def ms_iso(ms: int | float | None) -> str | None:
|
|
"""OpenCode 의 epoch ms → ISO. 없으면 None(=지금)."""
|
|
return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat() if ms else 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()},
|
|
**_oi("LLM", input=input, output=output, model=model, usage=usage,
|
|
userId=inherited.get("langfuse.user.id", ""), sessionId=inherited.get("langfuse.session.id", ""), metadata=metadata),
|
|
}
|
|
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 _to_protobuf(body: dict) -> bytes:
|
|
"""OTLP JSON → protobuf 바이트. traceId/spanId 는 JSON 이 hex, proto-JSON 은 base64 라 바꿔 넣음."""
|
|
import base64
|
|
import copy
|
|
|
|
from google.protobuf.json_format import ParseDict
|
|
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
|
|
|
|
b = copy.deepcopy(body)
|
|
for rs in b["resourceSpans"]:
|
|
for ss in rs["scopeSpans"]:
|
|
for sp in ss["spans"]:
|
|
for k in ("traceId", "spanId", "parentSpanId"):
|
|
if k in sp:
|
|
sp[k] = base64.b64encode(bytes.fromhex(sp[k])).decode()
|
|
return ParseDict(b, ExportTraceServiceRequest()).SerializeToString()
|
|
|
|
|
|
def _otlp_body(spans: list[dict]) -> dict:
|
|
project = settings.LANGFUSE.get("project") or "codeassist"
|
|
return {"resourceSpans": [{
|
|
"resource": {"attributes": [_attr("service.name", project), _attr("openinference.project.name", project)]},
|
|
"scopeSpans": [{"scope": {"name": project}, "spans": spans}],
|
|
}]}
|
|
|
|
|
|
async def send(spans: list[dict]) -> bool:
|
|
"""설정된 대상 전부에 전송. 하나라도 성공하면 True. 실패는 로그만."""
|
|
cfg = settings.LANGFUSE
|
|
if not spans or not enabled():
|
|
return False
|
|
body = _otlp_body(spans)
|
|
# (이름, url, headers, auth, protobuf 여부)
|
|
targets: list[tuple[str, str, dict, tuple | None, bool]] = []
|
|
if cfg.get("host"):
|
|
targets.append(("langfuse", cfg["host"].rstrip("/") + "/api/public/otel/v1/traces",
|
|
{"x-langfuse-ingestion-version": "4"}, (cfg.get("public_key", ""), cfg.get("secret_key", "")), False))
|
|
if cfg.get("phoenix"):
|
|
targets.append(("phoenix", cfg["phoenix"].rstrip("/") + "/v1/traces", {"Content-Type": "application/x-protobuf"}, None, True))
|
|
ok = False
|
|
for name, url, headers, auth, pb in targets:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0, transport=TRANSPORT) as client:
|
|
if pb:
|
|
resp = await client.post(url, content=_to_protobuf(body), headers=headers, auth=auth)
|
|
else:
|
|
resp = await client.post(url, json=body, headers=headers, auth=auth)
|
|
if resp.status_code != 200:
|
|
log.warning("%s ← %s %s", name, resp.status_code, resp.text[:200])
|
|
continue
|
|
partial = {}
|
|
try:
|
|
partial = (resp.json() or {}).get("partialSuccess") or {}
|
|
except ValueError:
|
|
pass
|
|
if partial.get("rejectedSpans"):
|
|
log.warning("%s 일부 거부: %s", name, partial)
|
|
continue
|
|
ok = True
|
|
except Exception as e: # noqa: BLE001 — 관측용이라 절대 본 흐름 안 깨뜨림
|
|
log.warning("%s 전송 실패: %s: %s", name, type(e).__name__, e)
|
|
return ok
|
|
|
|
|
|
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)
|