feat(observability): Phoenix 대상 추가 — 같은 span 에 OpenInference 속성 동봉, PHOENIX_HOST 로 /v1/traces. -12 는 도커 불가라 pip 로 뜨는 Phoenix 로 감(deploy/phoenix)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-22 15:39:08 +09:00
parent 340e8e4beb
commit e60c5bc273
8 changed files with 162 additions and 26 deletions
@@ -1,4 +1,9 @@
-"""Langfuse 로 trace/generation 쏘기 — SDK 없이 OTLP/HTTP JSON 한 방 (docs-lib/langfuse.md).
+"""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 …)
+속성은 두 벌을 같은 span 에 같이 실음 — 각자 자기 것만 읽고 나머진 metadata 로 떨어짐.

 v4 는 옛 /api/public/ingestion 이 막혀서(score 만) OTel 엔드포인트로 감. 트레이스 하나 = 루트 span(trace 속성) + 자식 span(generation).
 LANGFUSE_HOST 가 비어 있으면 전부 no-op. 보내는 건 fire-and-forget: 실패해도 채팅엔 영향 0, 로그만.
@@ -28,7 +33,28 @@ _pending: set[asyncio.Task] = set()  # GC 에 안 먹히게 잡아둠


 def enabled() -> bool:
-    return bool(settings.LANGFUSE.get("host"))
+    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:
@@ -97,6 +123,8 @@ def trace(trace_id: str, name: str, *, userId: str = "", sessionId: str = "", in
        "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)

@@ -114,6 +142,9 @@ def tool_span(trace_id: str, name: str, *, parent: dict, input=None, output=None
        "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)

@@ -142,6 +173,8 @@ def generation(trace_id: str, name: str, *, model: str = "", input=None, output=
        "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)
@@ -154,36 +187,45 @@ def usage_of(inp: int | None, out: int | None) -> dict | None:


 def _otlp_body(spans: list[dict]) -> dict:
+    project = settings.LANGFUSE.get("project") or "codeassist"
    return {"resourceSpans": [{
-        "resource": {"attributes": [_attr("service.name", "codeassist-backend")]},
-        "scopeSpans": [{"scope": {"name": "codeassist"}, "spans": spans}],
+        "resource": {"attributes": [_attr("service.name", project), _attr("openinference.project.name", project)]},
+        "scopeSpans": [{"scope": {"name": project}, "spans": spans}],
    }]}


 async def send(spans: list[dict]) -> bool:
-    """span 묶음 하나 전송. OTLP 는 200 + 빈 JSON 이면 성공, partialSuccess 있으면 일부 실패."""
+    """설정된 대상 전부에 전송. 하나라도 성공하면 True. 실패는 로그만."""
    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)
+    if not spans or not enabled():
        return False
+    body = _otlp_body(spans)
+    targets: list[tuple[str, str, dict, tuple | None]] = []
+    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", ""))))
+    if cfg.get("phoenix"):
+        targets.append(("phoenix", cfg["phoenix"].rstrip("/") + "/v1/traces", {}, None))
+    ok = False
+    for name, url, headers, auth in targets:
+        try:
+            async with httpx.AsyncClient(timeout=5.0, transport=TRANSPORT) as client:
+                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: