From 55f2e8fb60a5951f50624871f4e88d72540bc242 Mon Sep 17 00:00:00 2001 From: lee-hyeon-cheol Date: Tue, 22 Sep 2026 13:53:20 +0900 Subject: [PATCH] =?UTF-8?q?fix(langfuse):=20v4=20=EA=B0=80=20=EC=98=9B=20i?= =?UTF-8?q?ngestion=20=EC=9D=84=20=EA=B1=B0=EB=B6=80=20=E2=80=94=20OTLP/HT?= =?UTF-8?q?TP=20JSON(/api/public/otel/v1/traces)=EC=9C=BC=EB=A1=9C=20?= =?UTF-8?q?=EC=A0=84=ED=99=98,=20=EB=A1=9C=EC=BB=AC=20Langfuse=20=EC=8B=A4?= =?UTF-8?q?=EC=B8=A1=20=ED=86=B5=EA=B3=BC.=20=EC=82=AC=EC=9A=A9=EC=9E=90?= =?UTF-8?q?=C2=B7=EC=84=B8=EC=85=98=20=EC=86=8D=EC=84=B1=EC=9D=80=20?= =?UTF-8?q?=EC=9E=90=EC=8B=9D=20span=20=EC=97=90=EB=8F=84=20=EB=B3=B5?= =?UTF-8?q?=EC=82=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- 5_django_backend/.gitignore | 2 + 5_django_backend/apps/chat/stream.py | 11 +- 5_django_backend/apps/gateway/langfuse.py | 123 ++++++++++++++++++---- 5_django_backend/apps/gateway/views.py | 8 +- 5_django_backend/docs-lib/langfuse.md | 40 +++---- 5_django_backend/tests/test_gateway.py | 9 +- 5_django_backend/tests/test_langfuse.py | 50 +++++---- 7 files changed, 175 insertions(+), 68 deletions(-) diff --git a/5_django_backend/.gitignore b/5_django_backend/.gitignore index 45ff25f..a15dcd3 100644 --- a/5_django_backend/.gitignore +++ b/5_django_backend/.gitignore @@ -8,3 +8,5 @@ staticfiles/ # 고객사 소스 위키(ABAP_INDEXING 결과물) — 고객 데이터라 git 밖 opencode/wiki/ +deploy/langfuse/.env +deploy/langfuse/docker-compose.override.yml diff --git a/5_django_backend/apps/chat/stream.py b/5_django_backend/apps/chat/stream.py index b0e3e8f..80fb018 100644 --- a/5_django_backend/apps/chat/stream.py +++ b/5_django_backend/apps/chat/stream.py @@ -256,13 +256,14 @@ async def _finalize(session: ChatSession, state: TurnState, started: float, *, f tid = f"{session.id}-{int(started * 1000)}" user_email = await sync_to_async(lambda: session.user.email)() meta = {"failed": failed} if failed else {} + t0 = datetime.fromtimestamp(started, timezone.utc).isoformat() + root = langfuse.trace(tid, "chat", userId=user_email, sessionId=session.id, input=state.user_text, + output=content, metadata=meta, tags=["codeassist"], start=t0, end=langfuse.now_iso()) langfuse.send_later([ - langfuse.trace(tid, "chat", userId=user_email, sessionId=session.id, input=state.user_text, - output=content, metadata=meta, tags=["codeassist"]), + root, langfuse.generation( - tid, "opencode-turn", - startTime=datetime.fromtimestamp(started, timezone.utc).isoformat(), - endTime=langfuse.now_iso(), + tid, "opencode-turn", parent=root, + startTime=t0, endTime=langfuse.now_iso(), input=state.user_text, output=content, usage=langfuse.usage_of(usage["input"], usage["output"]) if usage else None, level="ERROR" if failed else "DEFAULT", statusMessage=failed or "", diff --git a/5_django_backend/apps/gateway/langfuse.py b/5_django_backend/apps/gateway/langfuse.py index f590ab8..cb99011 100644 --- a/5_django_backend/apps/gateway/langfuse.py +++ b/5_django_backend/apps/gateway/langfuse.py @@ -1,5 +1,6 @@ -"""Langfuse 로 trace/generation 쏘기 — SDK 없이 HTTP 한 방 (docs-lib/langfuse.md). +"""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(누가·어느 세션·질문·최종 답·토큰·시간) @@ -10,7 +11,10 @@ LANGFUSE_HOST 가 비어 있으면 전부 no-op. 보내는 건 fire-and-forget: from __future__ import annotations import asyncio +import hashlib +import json import logging +import time import uuid from datetime import datetime, timezone @@ -31,16 +35,86 @@ def now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") -def _event(kind: str, body: dict) -> dict: - return {"id": str(uuid.uuid4()), "timestamp": now_iso(), "type": kind, "body": body} +def _trace_hex(trace_id: str) -> str: + """아무 문자열 → OTel traceId(16바이트 hex). 같은 문자열이면 같은 trace 로 묶임.""" + return hashlib.sha256(trace_id.encode()).hexdigest()[:32] -def trace(trace_id: str, name: str, **body) -> dict: - return _event("trace-create", {"id": trace_id, "name": name, **body}) +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 generation(trace_id: str, name: str, **body) -> dict: - return _event("generation-create", {"id": str(uuid.uuid4()), "traceId": trace_id, "name": name, **body}) +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.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: @@ -49,36 +123,45 @@ def usage_of(inp: int | None, out: int | None) -> dict | None: return {"input": inp or 0, "output": out or 0, "total": (inp or 0) + (out or 0)} -async def send(events: list[dict]) -> bool: - """배치 하나 전송. 207 이면 성공(부분 실패는 로그).""" +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 events: + 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/ingestion", - json={"batch": events}, + 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 not in (200, 207): + if resp.status_code != 200: log.warning("langfuse ← %s %s", resp.status_code, resp.text[:200]) return False - errors = (resp.json() or {}).get("errors") or [] - if errors: - log.warning("langfuse 일부 실패: %s", errors[:3]) - return not errors + 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(events: list[dict]) -> None: +def send_later(spans: list[dict]) -> None: """지금 흐름 안 막고 백그라운드로. 이벤트 루프 없으면(동기 테스트) 조용히 버림.""" - if not enabled() or not events: + if not enabled() or not spans: return try: - task = asyncio.get_running_loop().create_task(send(events)) + task = asyncio.get_running_loop().create_task(send(spans)) except RuntimeError: return _pending.add(task) diff --git a/5_django_backend/apps/gateway/views.py b/5_django_backend/apps/gateway/views.py index 85fbf80..09612e5 100644 --- a/5_django_backend/apps/gateway/views.py +++ b/5_django_backend/apps/gateway/views.py @@ -101,11 +101,13 @@ def _observe(model_id: str, body: dict, col: _Collect, started: float, status: i return u = col.usage or {} tid = str(uuid.uuid4()) + t0 = datetime.fromtimestamp(started, timezone.utc).isoformat() + root = langfuse.trace(tid, "fabrix", tags=["gateway"], metadata={"parts": kinds}, start=t0, end=langfuse.now_iso()) langfuse.send_later([ - langfuse.trace(tid, "fabrix", tags=["gateway"], metadata={"parts": kinds}), + root, langfuse.generation( - tid, "fabrix.chat", model=model_id, - startTime=datetime.fromtimestamp(started, timezone.utc).isoformat(), endTime=langfuse.now_iso(), + tid, "fabrix.chat", model=model_id, parent=root, + startTime=t0, endTime=langfuse.now_iso(), input=body.get("messages"), output="".join(col.text), usage=langfuse.usage_of(u.get("prompt_tokens"), u.get("completion_tokens")), level="ERROR" if status >= 400 else "DEFAULT", statusMessage="" if status < 400 else f"upstream {status}", diff --git a/5_django_backend/docs-lib/langfuse.md b/5_django_backend/docs-lib/langfuse.md index 3d21748..80ed2e8 100644 --- a/5_django_backend/docs-lib/langfuse.md +++ b/5_django_backend/docs-lib/langfuse.md @@ -2,30 +2,32 @@ 출처: `https://cloud.langfuse.com/generated/api/openapi.yml`, `langfuse.com/self-hosting`. SDK 안 쓰고 HTTP 로 직접 쏨(오프라인 wheels 반입 줄이려고). -## POST /api/public/ingestion +## v4 는 OTLP 로 (2026-09-22 로컬 실측) -- 인증: Basic auth — user = `pk-lf-…`(public key), password = `sk-lf-…`(secret key) -- 응답: **207** (배치 부분 성공). 각 이벤트 결과가 `successes[]`/`errors[]` 로 옴 -- Langfuse Cloud 에선 2026-11-16 폐기 예정이지만 **self-host 는 계속 지원**. OTel 엔드포인트(`/api/public/otel/v1/traces`)는 v3+ 만. +옛 `POST /api/public/ingestion` 은 v4 기본(`LANGFUSE_MIGRATION_V4_WRITE_MODE=events_only`)에서 **trace/generation 을 400 으로 거부**함(score 만 받음). +`dual` 로 열 수 있지만 다음 메이저에서 없어짐 → 우리는 OTLP 로 감. + +### POST /api/public/otel/v1/traces +- OTLP/HTTP **JSON** 도 받음(protobuf 안 써도 됨). gRPC 는 없음. +- 헤더: `Authorization: Basic base64(pk-lf-…:sk-lf-…)`, `x-langfuse-ingestion-version: 4`(직접 쓰기, 15분 지연 없음) +- 응답 200 `{}`. 일부 거부면 `partialSuccess.rejectedSpans`. +- traceId = 16바이트 hex(32자), spanId = 8바이트 hex(16자). 시간은 `startTimeUnixNano`/`endTimeUnixNano` 문자열. ```json -{ - "batch": [ - {"id": "evt-1", "timestamp": "2026-09-22T01:00:00Z", "type": "trace-create", - "body": {"id": "trace-1", "name": "chat", "userId": "u@x.com", "sessionId": "ses_…", "input": "…", "output": "…", "metadata": {}, "tags": []}}, - {"id": "evt-2", "timestamp": "2026-09-22T01:00:05Z", "type": "generation-create", - "body": {"id": "gen-1", "traceId": "trace-1", "name": "fabrix", "model": "581", - "startTime": "…", "endTime": "…", "input": [...messages], "output": "…", - "usage": {"input": 120, "output": 30, "total": 150}, "level": "DEFAULT", "statusMessage": "", "metadata": {}}} - ] -} +{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"codeassist-backend"}}]}, + "scopeSpans":[{"scope":{"name":"codeassist"},"spans":[ + {"traceId":"<32hex>","spanId":"<16hex>","name":"chat","kind":1,"startTimeUnixNano":"…","endTimeUnixNano":"…", + "attributes":[{"key":"langfuse.trace.name","value":{"stringValue":"chat"}},{"key":"langfuse.user.id","value":{"stringValue":"u@x.com"}}], + "status":{"code":1}}, + {"traceId":"<같은>","spanId":"<16hex>","parentSpanId":"<루트 spanId>","name":"fabrix.chat","kind":1,…, + "attributes":[{"key":"langfuse.observation.type","value":{"stringValue":"generation"}},…]}]}]}]} ``` -- envelope: `id`(중복 제거용, 유일), `timestamp`(ISO 8601), `type`, `body` -- 이벤트 타입: `trace-create`, `generation-create`, `span-create`, `event-create`, `score-create`, `*-update` -- TraceBody: id·name·userId·sessionId·input·output·metadata·tags·version·release·environment·public -- CreateGenerationBody: id·traceId·name·startTime·endTime·completionStartTime·model·modelParameters·input·output·usage{input,output,total,unit,inputCost,outputCost,totalCost}·usageDetails·costDetails·level(DEBUG/DEFAULT/WARNING/ERROR)·statusMessage·metadata·parentObservationId -- 같은 traceId 로 trace-create 를 여러 번 보내도 됨(upsert). generation 먼저 와도 trace 가 나중에 붙음. +### 속성 → Langfuse 필드 +- 관측(span) 단위: `langfuse.observation.type`(generation/span), `.model.name`, `.input`, `.output`, `.usage_details`(JSON `{"input","output","total"}`), `.model_parameters`, `.level`(DEBUG/DEFAULT/WARNING/ERROR), `.status_message`, `.metadata.<키>`, `.cost_details` + (대안: `gen_ai.request.model`, `gen_ai.prompt`/`gen_ai.completion`, `gen_ai.usage.*`) +- 트레이스 단위(아무 span 에나 실으면 트레이스 전체에 적용, 필터 쓰려면 루트에): `langfuse.trace.name`(없으면 루트 span 이름), `langfuse.user.id`, `langfuse.session.id`, `langfuse.trace.tags`(JSON 배열), `langfuse.trace.input`/`.output`, `langfuse.trace.metadata.<키>`, `langfuse.environment`, `langfuse.version` +- 값 인코딩: 문자열은 stringValue, 객체/배열은 JSON 문자열로 넣으면 Langfuse 가 파싱해서 보여줌. ## 자체 호스팅 (v4 docker-compose, 2026-09 기준) diff --git a/5_django_backend/tests/test_gateway.py b/5_django_backend/tests/test_gateway.py index 2d65e09..737cb24 100644 --- a/5_django_backend/tests/test_gateway.py +++ b/5_django_backend/tests/test_gateway.py @@ -133,13 +133,16 @@ async def test_stream_reports_generation_to_langfuse(upstream, monkeypatch): from apps.gateway import langfuse got = [] - monkeypatch.setattr(langfuse, "TRANSPORT", httpx.MockTransport(lambda r: (got.append(json.loads(r.content)), httpx.Response(207, json={}))[1])) + monkeypatch.setattr(langfuse, "TRANSPORT", httpx.MockTransport(lambda r: (got.append(json.loads(r.content)), httpx.Response(200, json={}))[1])) upstream(lambda r: _sse(b'data: {"choices":[{"delta":{"content":"hi"}}],"usage":{"prompt_tokens":2,"completion_tokens":1}}\n\ndata: [DONE]\n\n')) resp = await _post({"model": "581", "stream": True, "messages": [{"role": "user", "content": "q"}]}) b"".join([c async for c in resp.streaming_content]) await asyncio.gather(*list(langfuse._pending)) - gen = got[0]["batch"][1]["body"] - assert gen["model"] == "581" and gen["output"] == "hi" and gen["usage"]["total"] == 3 and gen["input"][0]["content"] == "q" + spans = got[0]["resourceSpans"][0]["scopeSpans"][0]["spans"] + ga = {a["key"]: a["value"].get("stringValue") for a in spans[1]["attributes"]} + assert ga["langfuse.observation.model.name"] == "581" and ga["langfuse.observation.output"] == "hi" + assert json.loads(ga["langfuse.observation.usage_details"])["total"] == 3 + assert json.loads(ga["langfuse.observation.input"])[0]["content"] == "q" @override_settings(FABRIX_ENV=FULL_ENV) diff --git a/5_django_backend/tests/test_langfuse.py b/5_django_backend/tests/test_langfuse.py index 69d01cf..08b8c65 100644 --- a/5_django_backend/tests/test_langfuse.py +++ b/5_django_backend/tests/test_langfuse.py @@ -1,4 +1,4 @@ -"""Langfuse 전송 — 이벤트 모양·인증·게이트웨이/턴 마무리 훅. 서버는 MockTransport.""" +"""Langfuse 전송(OTLP/HTTP JSON) — span 모양·인증·게이트웨이/턴 마무리 훅. 서버는 MockTransport.""" import asyncio import base64 @@ -19,27 +19,38 @@ def lf_server(monkeypatch): def handler(req: httpx.Request) -> httpx.Response: got.append({"url": str(req.url), "auth": req.headers.get("authorization"), "body": json.loads(req.content)}) - return httpx.Response(207, json={"successes": [], "errors": []}) + return httpx.Response(200, json={}) monkeypatch.setattr(langfuse, "TRANSPORT", httpx.MockTransport(handler)) return got +def _attrs(span: dict) -> dict: + out = {} + for a in span["attributes"]: + v = a["value"] + out[a["key"]] = v.get("stringValue", v.get("intValue", v.get("boolValue"))) + return out + + @override_settings(LANGFUSE=LF) -async def test_send_batch_shape_and_basic_auth(lf_server): - ok = await langfuse.send([ - langfuse.trace("t1", "chat", userId="u@x.com", sessionId="s1"), - langfuse.generation("t1", "gen", model="581", usage=langfuse.usage_of(10, 5)), - ]) - assert ok and len(lf_server) == 1 +async def test_send_otlp_shape_and_basic_auth(lf_server): + root = langfuse.trace("t1", "chat", userId="u@x.com", sessionId="s1", tags=["x"]) + gen = langfuse.generation("t1", "gen", model="581", usage=langfuse.usage_of(10, 5), parent=root) + assert await langfuse.send([root, gen]) and len(lf_server) == 1 req = lf_server[0] - assert req["url"] == "http://lf.test/api/public/ingestion" + assert req["url"] == "http://lf.test/api/public/otel/v1/traces" assert req["auth"] == "Basic " + base64.b64encode(b"pk-lf-a:sk-lf-b").decode() - batch = req["body"]["batch"] - assert [e["type"] for e in batch] == ["trace-create", "generation-create"] - assert batch[0]["body"] == {"id": "t1", "name": "chat", "userId": "u@x.com", "sessionId": "s1"} - assert batch[1]["body"]["traceId"] == "t1" and batch[1]["body"]["usage"] == {"input": 10, "output": 5, "total": 15} - assert batch[0]["id"] != batch[1]["id"] and batch[0]["timestamp"].endswith("Z") + spans = req["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"] + assert len(spans) == 2 and spans[0]["traceId"] == spans[1]["traceId"] and len(spans[0]["traceId"]) == 32 + assert spans[1]["parentSpanId"] == spans[0]["spanId"] and "parentSpanId" not in spans[0] + ra, ga = _attrs(spans[0]), _attrs(spans[1]) + assert ra["langfuse.trace.name"] == "chat" and ra["langfuse.user.id"] == "u@x.com" and ra["langfuse.session.id"] == "s1" + assert json.loads(ra["langfuse.trace.tags"]) == ["x"] + assert ga["langfuse.observation.type"] == "generation" and ga["langfuse.observation.model.name"] == "581" + assert ga["langfuse.user.id"] == "u@x.com" and ga["langfuse.session.id"] == "s1" # 자식에도 복사 + assert json.loads(ga["langfuse.observation.usage_details"]) == {"input": 10, "output": 5, "total": 15} + assert int(spans[1]["endTimeUnixNano"]) >= int(spans[1]["startTimeUnixNano"]) @override_settings(LANGFUSE={"host": "", "public_key": "", "secret_key": ""}) @@ -80,8 +91,11 @@ async def test_finalize_sends_user_trace(lf_server, django_user_model, monkeypat await _finalize(session, TurnState("ses_1", "질문"), 1.0) await asyncio.gather(*list(langfuse._pending)) assert len(lf_server) == 1 - batch = lf_server[0]["body"]["batch"] - t, g = batch[0]["body"], batch[1]["body"] - assert t["userId"] == "u@x.com" and t["sessionId"] == "ses_1" and t["input"] == "질문" and t["output"] == "답" - assert g["traceId"] == t["id"] and g["usage"] == {"input": 3, "output": 5, "total": 8} and g["level"] == "DEFAULT" + spans = lf_server[0]["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"] + t, g = _attrs(spans[0]), _attrs(spans[1]) + assert t["langfuse.user.id"] == "u@x.com" and t["langfuse.session.id"] == "ses_1" + assert t["langfuse.trace.input"] == "질문" and t["langfuse.trace.output"] == "답" + assert spans[1]["parentSpanId"] == spans[0]["spanId"] + assert json.loads(g["langfuse.observation.usage_details"]) == {"input": 3, "output": 5, "total": 8} + assert g["langfuse.observation.level"] == "DEFAULT" and spans[1]["status"]["code"] == 1 assert stream_mod is not None