From 0321050053fdb322cd3682a121f3308096adb343 Mon Sep 17 00:00:00 2001 From: lee-hyeon-cheol Date: Tue, 22 Sep 2026 13:44:12 +0900 Subject: [PATCH] =?UTF-8?q?feat(observability):=20Langfuse=20=EC=A0=84?= =?UTF-8?q?=EC=86=A1=20=E2=80=94=20=ED=84=B4=20=EB=A7=88=EB=AC=B4=EB=A6=AC?= =?UTF-8?q?(=EC=82=AC=EC=9A=A9=EC=9E=90=20trace)=C2=B7=EA=B2=8C=EC=9D=B4?= =?UTF-8?q?=ED=8A=B8=EC=9B=A8=EC=9D=B4(FabriX=20generation)=20=EB=91=90=20?= =?UTF-8?q?=ED=9B=85,=20SDK=20=EC=97=86=EC=9D=B4=20HTTP.=20deploy/langfuse?= =?UTF-8?q?=20=EC=97=90=20compose+env+=EB=B0=98=EC=9E=85=20=EC=A0=88?= =?UTF-8?q?=EC=B0=A8?= 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/.env.example | 5 + 5_django_backend/apps/chat/stream.py | 21 ++ 5_django_backend/apps/gateway/langfuse.py | 85 +++++++ 5_django_backend/apps/gateway/views.py | 66 +++++ 5_django_backend/config/settings.py | 6 + 5_django_backend/deploy.sh | 1 + 5_django_backend/deploy/langfuse/.env.example | 24 ++ 5_django_backend/deploy/langfuse/README.md | 41 ++++ .../deploy/langfuse/docker-compose.yml | 225 ++++++++++++++++++ 5_django_backend/docs-lib/README.md | 1 + 5_django_backend/docs-lib/langfuse.md | 39 +++ 5_django_backend/tests/test_gateway.py | 16 ++ 5_django_backend/tests/test_langfuse.py | 87 +++++++ 13 files changed, 617 insertions(+) create mode 100644 5_django_backend/apps/gateway/langfuse.py create mode 100644 5_django_backend/deploy/langfuse/.env.example create mode 100644 5_django_backend/deploy/langfuse/README.md create mode 100644 5_django_backend/deploy/langfuse/docker-compose.yml create mode 100644 5_django_backend/docs-lib/langfuse.md create mode 100644 5_django_backend/tests/test_langfuse.py diff --git a/5_django_backend/.env.example b/5_django_backend/.env.example index 31b28b2..f096df1 100644 --- a/5_django_backend/.env.example +++ b/5_django_backend/.env.example @@ -62,3 +62,8 @@ SAP_API_BASE_ICF= # ── ABAP 소스 인덱스(ABAP_INDEXING, 별도 서버 :8100) — OpenCode 툴 abap_index_* 가 씀 ── # 비우면 기본 127.0.0.1:8100. 서버가 없어도 앱은 돌아감(툴이 "없음"으로 답하고 생성 코드로). ABAP_INDEX_URL=http://127.0.0.1:8100 + +# ── Langfuse(LLM 관측, 자체 호스팅 deploy/langfuse/) — HOST 비우면 안 보냄 ── +LANGFUSE_HOST= +LANGFUSE_PUBLIC_KEY= +LANGFUSE_SECRET_KEY= diff --git a/5_django_backend/apps/chat/stream.py b/5_django_backend/apps/chat/stream.py index 1b96ea3..b0e3e8f 100644 --- a/5_django_backend/apps/chat/stream.py +++ b/5_django_backend/apps/chat/stream.py @@ -20,12 +20,14 @@ import asyncio import json import logging import time +from datetime import datetime, timezone from django.conf import settings from django.http import StreamingHttpResponse from django.views.decorators.csrf import csrf_exempt from apps.accounts.authentication import user_from_token +from apps.gateway import langfuse from asgiref.sync import sync_to_async from common.envelope import json_error from common.opencode_service import opencode_service @@ -248,6 +250,25 @@ async def _finalize(session: ChatSession, state: TurnState, started: float, *, f if title: session.title_llm = title[:200] await session.asave(update_fields=["is_generating", "title_llm", "updated_at"]) + + # 관측: 사용자 단위 trace 하나 + 답변 generation 하나. 실패해도 여기까진 이미 저장됨. + if langfuse.enabled(): + tid = f"{session.id}-{int(started * 1000)}" + user_email = await sync_to_async(lambda: session.user.email)() + meta = {"failed": failed} if failed else {} + langfuse.send_later([ + langfuse.trace(tid, "chat", userId=user_email, sessionId=session.id, input=state.user_text, + output=content, metadata=meta, tags=["codeassist"]), + langfuse.generation( + tid, "opencode-turn", + startTime=datetime.fromtimestamp(started, timezone.utc).isoformat(), + 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 "", + metadata={"elapsed_ms": usage["elapsed_ms"]} if usage else {}, + ), + ]) return usage, title diff --git a/5_django_backend/apps/gateway/langfuse.py b/5_django_backend/apps/gateway/langfuse.py new file mode 100644 index 0000000..f590ab8 --- /dev/null +++ b/5_django_backend/apps/gateway/langfuse.py @@ -0,0 +1,85 @@ +"""Langfuse 로 trace/generation 쏘기 — SDK 없이 HTTP 한 방 (docs-lib/langfuse.md). + +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 logging +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 _event(kind: str, body: dict) -> dict: + return {"id": str(uuid.uuid4()), "timestamp": now_iso(), "type": kind, "body": body} + + +def trace(trace_id: str, name: str, **body) -> dict: + return _event("trace-create", {"id": trace_id, "name": name, **body}) + + +def generation(trace_id: str, name: str, **body) -> dict: + return _event("generation-create", {"id": str(uuid.uuid4()), "traceId": trace_id, "name": name, **body}) + + +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)} + + +async def send(events: list[dict]) -> bool: + """배치 하나 전송. 207 이면 성공(부분 실패는 로그).""" + cfg = settings.LANGFUSE + if not cfg.get("host") or not events: + 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}, + auth=(cfg.get("public_key", ""), cfg.get("secret_key", "")), + ) + if resp.status_code not in (200, 207): + 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 + except Exception as e: # noqa: BLE001 — 관측용이라 절대 본 흐름 안 깨뜨림 + log.warning("langfuse 전송 실패: %s: %s", type(e).__name__, e) + return False + + +def send_later(events: list[dict]) -> None: + """지금 흐름 안 막고 백그라운드로. 이벤트 루프 없으면(동기 테스트) 조용히 버림.""" + if not enabled() or not events: + return + try: + task = asyncio.get_running_loop().create_task(send(events)) + except RuntimeError: + return + _pending.add(task) + task.add_done_callback(_pending.discard) diff --git a/5_django_backend/apps/gateway/views.py b/5_django_backend/apps/gateway/views.py index dbd2b58..85fbf80 100644 --- a/5_django_backend/apps/gateway/views.py +++ b/5_django_backend/apps/gateway/views.py @@ -11,6 +11,9 @@ from __future__ import annotations import asyncio import json +import time +import uuid +from datetime import datetime, timezone import logging from typing import AsyncIterator @@ -19,6 +22,7 @@ from django.conf import settings from django.http import HttpRequest, HttpResponse, JsonResponse, StreamingHttpResponse from django.views.decorators.csrf import csrf_exempt +from . import langfuse from .fabrix import FabrixConfig, Variant, auth_variants log = logging.getLogger(__name__) @@ -60,6 +64,57 @@ def _sse_error(status: int, detail: bytes) -> bytes: return b"data: " + json.dumps(payload, ensure_ascii=False).encode() + b"\n\ndata: [DONE]\n\n" +class _Collect: + """상류 응답에서 답변 텍스트·usage 만 긁어 모음(관측용). 스트림은 줄 단위 `data: {json}`.""" + + def __init__(self) -> None: + self.text: list[str] = [] + self.usage: dict | None = None + self.buf = b"" + + def feed(self, chunk: bytes) -> None: + self.buf += chunk + while b"\n" in self.buf: + line, self.buf = self.buf.split(b"\n", 1) + self._line(line.strip()) + + def _line(self, line: bytes) -> None: + if not line.startswith(b"data:") or line.endswith(b"[DONE]"): + return + try: + d = json.loads(line[5:]) + except ValueError: + return + self.json(d) + + def json(self, d: dict) -> None: + for c in d.get("choices") or []: + t = (c.get("delta") or c.get("message") or {}).get("content") + if isinstance(t, str): + self.text.append(t) + if d.get("usage"): + self.usage = d["usage"] + + +def _observe(model_id: str, body: dict, col: _Collect, started: float, status: int, *, kinds: list[str]) -> None: + if not langfuse.enabled(): + return + u = col.usage or {} + tid = str(uuid.uuid4()) + langfuse.send_later([ + langfuse.trace(tid, "fabrix", tags=["gateway"], metadata={"parts": kinds}), + langfuse.generation( + tid, "fabrix.chat", model=model_id, + startTime=datetime.fromtimestamp(started, timezone.utc).isoformat(), 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}", + modelParameters={k: v for k, v in body.items() if k in ("temperature", "max_completion_tokens", "max_tokens")}, + metadata={"stream": bool(body.get("stream")), "tools": len(body.get("tools") or [])}, + ), + ]) + + async def healthcheck(_request: HttpRequest) -> JsonResponse: return JsonResponse({"success": True}) @@ -95,6 +150,9 @@ async def chat_completions(request: HttpRequest) -> HttpResponse: # FabriX 가 맞는 조합에도 가끔 401 을 뱉음(2026-09-21 고객사 실측: curl 10번 중 1~2번). # 4개 조합 다 돌고 나서 제일 유력한 조합(맨 앞)을 잠깐 쉬고 한 번 더 — 그래도 401 이면 진짜 인증 문제 variants = variants + variants[:1] + model_id = headers.get("x-llm-model-id", "") + started = time.time() + col = _Collect() if body.get("stream"): @@ -119,9 +177,11 @@ async def chat_completions(request: HttpRequest) -> HttpResponse: return _last_ok["variant"] = vkey async for chunk in resp.aiter_raw(): + col.feed(chunk) yield chunk finally: await resp.aclose() + _observe(model_id, body, col, started, resp.status_code, kinds=kinds or ["text"]) out = StreamingHttpResponse(gen(), content_type="text/event-stream") out["Cache-Control"] = "no-cache" @@ -141,4 +201,10 @@ async def chat_completions(request: HttpRequest) -> HttpResponse: break except httpx.HTTPError as e: return _err(502, f"FabriX 호출 실패: {type(e).__name__}: {e}", "upstream_error") + if resp.status_code < 400: + try: + col.json(resp.json()) + except ValueError: + pass + _observe(model_id, body, col, started, resp.status_code, kinds=kinds or ["text"]) return HttpResponse(resp.content, status=resp.status_code, content_type="application/json") diff --git a/5_django_backend/config/settings.py b/5_django_backend/config/settings.py index 64b8e28..76c0be5 100644 --- a/5_django_backend/config/settings.py +++ b/5_django_backend/config/settings.py @@ -88,6 +88,12 @@ STREAM_FIRST_EVENT_TIMEOUT_S = int(_env("STREAM_FIRST_EVENT_TIMEOUT_S", "300")) from apps.gateway.fabrix import ENV_KEYS as _FABRIX_KEYS # noqa: E402 FABRIX_ENV = {k: _env(k) for k in _FABRIX_KEYS} +# Langfuse(LLM 관측) — 자체 호스팅 주소 + 프로젝트 키. HOST 비우면 전송 안 함 +LANGFUSE = { + "host": _env("LANGFUSE_HOST"), + "public_key": _env("LANGFUSE_PUBLIC_KEY"), + "secret_key": _env("LANGFUSE_SECRET_KEY"), +} # 세션 컨텍스트 하드 한도 — usage.limit 로 프론트 게이지에 감 CONTEXT_LIMIT_TOKENS = int(_env("CONTEXT_LIMIT_TOKENS", "128000") or "128000") diff --git a/5_django_backend/deploy.sh b/5_django_backend/deploy.sh index 701df83..df07923 100644 --- a/5_django_backend/deploy.sh +++ b/5_django_backend/deploy.sh @@ -19,6 +19,7 @@ grep -q '^BACKEND_PORT=' .env || echo 'BACKEND_PORT=8080' >> .env grep -q '^OPENCODE_VISION_MODEL=' .env || echo 'OPENCODE_VISION_MODEL=gateway/605' >> .env # 고객사 FabriX 는 Bearer 접두 필수(9/18 실측). 첫 시도부터 맞게 — 날것으로 보내면 401 뒤 재시도에 기댐 grep -q '^AAF_FABRIX_TOKEN_PREFIX=' .env || echo 'AAF_FABRIX_TOKEN_PREFIX=bearer' >> .env # OpenCode 가 붙을 Django 포트. 없으면 8001 로 렌더돼 'Unable to connect' +for k in LANGFUSE_HOST LANGFUSE_PUBLIC_KEY LANGFUSE_SECRET_KEY; do grep -q "^$k=" .env || echo "$k=" >> .env; done # 비우면 관측 안 함 sed -i 's/^AAF_FABRIX_MODEL_ID=.*/AAF_FABRIX_MODEL_ID=581/; s/^AAF_FABRIX_MODELS=.*/AAF_FABRIX_MODELS=581:GaussO Think,339:GaussO Flash,605:Gemma4/' .env .venv/bin/python manage.py migrate --noinput [ -f seed/snippets.db ] && .venv/bin/python manage.py import_snippets seed/snippets.db diff --git a/5_django_backend/deploy/langfuse/.env.example b/5_django_backend/deploy/langfuse/.env.example new file mode 100644 index 0000000..ce7e380 --- /dev/null +++ b/5_django_backend/deploy/langfuse/.env.example @@ -0,0 +1,24 @@ +# Langfuse 자체 호스팅 (.env — docker-compose.yml 이 읽음). 전부 CHANGEME 는 바꿔서 쓸 것. +# 값 만드는 법: openssl rand -hex 32 (SALT·NEXTAUTH_SECRET·ENCRYPTION_KEY 각각 다르게) + +# 웹 UI 가 열리는 주소 — 고객사는 -12 남는 포트(예: 8915→3000)로. 로컬은 http://localhost:3000 +NEXTAUTH_URL=http://localhost:3000 +NEXTAUTH_SECRET=CHANGEME +SALT=CHANGEME +ENCRYPTION_KEY=CHANGEME_HEX64 + +# 컨테이너 안 서비스(postgres/clickhouse/redis/minio) 비번은 compose 기본값 그대로 둠 — +# 밖으로 안 열리는 내부 네트워크라 위험 없고, 바꾸면 DATABASE_URL·LANGFUSE_S3_* 도 같이 맞춰야 해서 실수 잦음. + +# 헤드리스 초기화 — 첫 기동 때 조직/프로젝트/키/관리자 자동 생성. 이 pk/sk 를 백엔드 .env 의 LANGFUSE_* 에 그대로. +LANGFUSE_INIT_ORG_ID=codeassist +LANGFUSE_INIT_ORG_NAME=CodeAssist +LANGFUSE_INIT_PROJECT_ID=codeassist +LANGFUSE_INIT_PROJECT_NAME=CodeAssist +LANGFUSE_INIT_PROJECT_PUBLIC_KEY=pk-lf-CHANGEME +LANGFUSE_INIT_PROJECT_SECRET_KEY=sk-lf-CHANGEME +LANGFUSE_INIT_USER_EMAIL=admin@example.com +LANGFUSE_INIT_USER_NAME=admin +LANGFUSE_INIT_USER_PASSWORD=CHANGEME + +TELEMETRY_ENABLED=false diff --git a/5_django_backend/deploy/langfuse/README.md b/5_django_backend/deploy/langfuse/README.md new file mode 100644 index 0000000..36ce120 --- /dev/null +++ b/5_django_backend/deploy/langfuse/README.md @@ -0,0 +1,41 @@ +# Langfuse 자체 호스팅 — -12 에 올리기 + +LLM 호출 관측(누가·언제·어떤 모델·프롬프트·답·토큰·시간). 백엔드가 `LANGFUSE_HOST` 있으면 자동으로 쏨(`apps/gateway/langfuse.py`). +공식 `docker-compose.yml` 그대로(2026-09-22 main). 컨테이너 6개, 밖으로 여는 건 웹 3000 하나. + +## 1. 이미지 반입 (인터넷 되는 PC) + +```bash +cd 5_django_backend/deploy/langfuse +docker compose pull +docker save $(grep -oE 'image: \S+' docker-compose.yml | cut -d' ' -f2 | sed 's/\${POSTGRES_VERSION:-17}/17/') | gzip > langfuse-images.tar.gz +ls -la langfuse-images.tar.gz # 3~4GB. 메일 안 됨 — 인프라팀 반입 통로로 +``` + +## 2. -12 에서 띄우기 + +```bash +docker load < langfuse-images.tar.gz +cd /www/abap-ito/code-assistant/5_django_backend/deploy/langfuse +cp .env.example .env && vi .env # CHANGEME 전부 교체. NEXTAUTH_URL 은 사용자가 여는 주소 +docker compose up -d +docker compose ps # 6개 healthy 까지 1~2분 +curl -s http://127.0.0.1:3000/api/public/health +``` + +## 3. 백엔드 연결 + +`5_django_backend/.env` 에 (`.env` 의 INIT 키와 같은 값): +``` +LANGFUSE_HOST=http://127.0.0.1:3000 +LANGFUSE_PUBLIC_KEY=pk-lf-… +LANGFUSE_SECRET_KEY=sk-lf-… +``` +`bash deploy.sh` 로 재기동 → 채팅 하나 보내고 Langfuse 웹(Traces)에 `chat`·`fabrix` trace 뜨면 끝. + +## 운영 + +- 웹 UI 접속은 -12 의 3000 → 인프라팀이 외부 포트(8915 등) 매핑해줘야 고객사 PC 에서 열림 +- 중지 `docker compose down` (데이터 볼륨 남음) / 완전 삭제 `docker compose down -v` +- 로그 `docker compose logs -f langfuse-web` +- 백엔드는 Langfuse 죽어 있어도 멀쩡(전송 실패는 warning 로그만) diff --git a/5_django_backend/deploy/langfuse/docker-compose.yml b/5_django_backend/deploy/langfuse/docker-compose.yml new file mode 100644 index 0000000..40293a8 --- /dev/null +++ b/5_django_backend/deploy/langfuse/docker-compose.yml @@ -0,0 +1,225 @@ +# Make sure to update the credential placeholders with your own secrets. +# We mark them with # CHANGEME in the file below. +# In addition, we recommend to restrict inbound traffic on the host to langfuse-web (port 3000) and minio (port 9090) only. +# All other components are bound to localhost (127.0.0.1) to only accept connections from the local machine. +# External connections from other machines will not be able to reach these services directly. +services: + langfuse-worker: + image: docker.langfuse.com/langfuse/langfuse-worker:4 + restart: always + depends_on: &langfuse-depends-on + postgres: + condition: service_healthy + minio: + condition: service_healthy + redis: + condition: service_healthy + clickhouse: + condition: service_healthy + ports: + - 127.0.0.1:3030:3030 + environment: &langfuse-worker-env + NEXTAUTH_URL: ${NEXTAUTH_URL:-http://localhost:3000} + DATABASE_URL: ${DATABASE_URL:-postgresql://postgres:postgres@postgres:5432/postgres} # CHANGEME + SALT: ${SALT:-mysalt} # CHANGEME + ENCRYPTION_KEY: ${ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000} # CHANGEME: generate via `openssl rand -hex 32` + TELEMETRY_ENABLED: ${TELEMETRY_ENABLED:-true} + LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES: ${LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES:-false} + # Lets the worker reach the web service internally for in-app agent MCP + # calls without redirecting NEXTAUTH_URL, which also builds links for + # users. Pair it with LANGFUSE_MCP_ALLOWED_HOSTS on langfuse-web. + LANGFUSE_MCP_BASE_URL: ${LANGFUSE_MCP_BASE_URL:-} + LANGFUSE_LLM_CONNECTION_WHITELISTED_HOST: ${LANGFUSE_LLM_CONNECTION_WHITELISTED_HOST:-} + LANGFUSE_LLM_CONNECTION_WHITELISTED_IPS: ${LANGFUSE_LLM_CONNECTION_WHITELISTED_IPS:-} + LANGFUSE_LLM_CONNECTION_WHITELISTED_IP_SEGMENTS: ${LANGFUSE_LLM_CONNECTION_WHITELISTED_IP_SEGMENTS:-} + CLICKHOUSE_MIGRATION_URL: ${CLICKHOUSE_MIGRATION_URL:-clickhouse://clickhouse:9000} + CLICKHOUSE_URL: ${CLICKHOUSE_URL:-http://clickhouse:8123} + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} # CHANGEME + CLICKHOUSE_CLUSTER_ENABLED: ${CLICKHOUSE_CLUSTER_ENABLED:-false} + CLICKHOUSE_CLUSTER_NAME: ${CLICKHOUSE_CLUSTER_NAME:-default} + LANGFUSE_JSON_BAD_UNICODE_ESCAPE: ${LANGFUSE_JSON_BAD_UNICODE_ESCAPE:-} + LANGFUSE_USE_AZURE_BLOB: ${LANGFUSE_USE_AZURE_BLOB:-false} + LANGFUSE_USE_OCI_NATIVE_OBJECT_STORAGE: ${LANGFUSE_USE_OCI_NATIVE_OBJECT_STORAGE:-false} + LANGFUSE_OCI_AUTH_TYPE: ${LANGFUSE_OCI_AUTH_TYPE:-workload_identity} + LANGFUSE_S3_EVENT_UPLOAD_BUCKET: ${LANGFUSE_S3_EVENT_UPLOAD_BUCKET:-langfuse} + LANGFUSE_S3_EVENT_UPLOAD_REGION: ${LANGFUSE_S3_EVENT_UPLOAD_REGION:-auto} + LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY:-miniosecret} # CHANGEME + LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: ${LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT:-http://minio:9000} + LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE:-true} + LANGFUSE_S3_EVENT_UPLOAD_PREFIX: ${LANGFUSE_S3_EVENT_UPLOAD_PREFIX:-events/} + LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: ${LANGFUSE_S3_MEDIA_UPLOAD_BUCKET:-langfuse} + LANGFUSE_S3_MEDIA_UPLOAD_REGION: ${LANGFUSE_S3_MEDIA_UPLOAD_REGION:-auto} + LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY:-miniosecret} # CHANGEME + LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://minio:9000} + LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE:-true} + LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: ${LANGFUSE_S3_MEDIA_UPLOAD_PREFIX:-media/} + LANGFUSE_OBSERVATION_FIELD_OVERFLOW_ENABLED: ${LANGFUSE_OBSERVATION_FIELD_OVERFLOW_ENABLED:-false} + LANGFUSE_OBSERVATION_FIELD_SIZE_LIMIT_BYTES: ${LANGFUSE_OBSERVATION_FIELD_SIZE_LIMIT_BYTES:-2097152} + LANGFUSE_S3_BATCH_EXPORT_ENABLED: ${LANGFUSE_S3_BATCH_EXPORT_ENABLED:-false} + LANGFUSE_S3_BATCH_EXPORT_BUCKET: ${LANGFUSE_S3_BATCH_EXPORT_BUCKET:-langfuse} + LANGFUSE_S3_BATCH_EXPORT_PREFIX: ${LANGFUSE_S3_BATCH_EXPORT_PREFIX:-exports/} + LANGFUSE_S3_BATCH_EXPORT_REGION: ${LANGFUSE_S3_BATCH_EXPORT_REGION:-auto} + LANGFUSE_S3_BATCH_EXPORT_ENDPOINT: ${LANGFUSE_S3_BATCH_EXPORT_ENDPOINT:-http://minio:9000} + LANGFUSE_S3_BATCH_EXPORT_EXTERNAL_ENDPOINT: ${LANGFUSE_S3_BATCH_EXPORT_EXTERNAL_ENDPOINT:-http://localhost:9090} + LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID: ${LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID:-minio} + LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY: ${LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY:-miniosecret} # CHANGEME + LANGFUSE_S3_BATCH_EXPORT_FORCE_PATH_STYLE: ${LANGFUSE_S3_BATCH_EXPORT_FORCE_PATH_STYLE:-true} + LANGFUSE_INGESTION_QUEUE_DELAY_MS: ${LANGFUSE_INGESTION_QUEUE_DELAY_MS:-} + LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS: ${LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS:-} + REDIS_HOST: ${REDIS_HOST:-redis} + REDIS_PORT: ${REDIS_PORT:-6379} + REDIS_AUTH: ${REDIS_AUTH:-myredissecret} # CHANGEME + LANGFUSE_BULLMQ_SKIP_REDIS_VERSION_CHECK: ${LANGFUSE_BULLMQ_SKIP_REDIS_VERSION_CHECK:-false} + REDIS_TLS_ENABLED: ${REDIS_TLS_ENABLED:-false} + REDIS_TLS_CA: ${REDIS_TLS_CA:-/certs/ca.crt} + REDIS_TLS_CERT: ${REDIS_TLS_CERT:-/certs/redis.crt} + REDIS_TLS_KEY: ${REDIS_TLS_KEY:-/certs/redis.key} + EMAIL_FROM_ADDRESS: ${EMAIL_FROM_ADDRESS:-} + SMTP_CONNECTION_URL: ${SMTP_CONNECTION_URL:-} + # Shared AI features configuration. These apply to AI features that run + # in both web and worker. + LANGFUSE_IN_APP_AGENT_ENABLED: ${LANGFUSE_IN_APP_AGENT_ENABLED:-} + LANGFUSE_EVALUATOR_MEDIA_TRANSPORT: ${LANGFUSE_EVALUATOR_MEDIA_TRANSPORT:-} + LANGFUSE_EVALUATOR_MEDIA_INLINE_MAX_BYTES: ${LANGFUSE_EVALUATOR_MEDIA_INLINE_MAX_BYTES:-} + LANGFUSE_AI_PROVIDER: ${LANGFUSE_AI_PROVIDER:-} + LANGFUSE_AI_MODEL: ${LANGFUSE_AI_MODEL:-} + LANGFUSE_AI_SMALL_MODEL: ${LANGFUSE_AI_SMALL_MODEL:-} + LANGFUSE_AI_API_KEY: ${LANGFUSE_AI_API_KEY:-} + LANGFUSE_AI_BASE_URL: ${LANGFUSE_AI_BASE_URL:-} + LANGFUSE_AI_USE_RESPONSES_API: ${LANGFUSE_AI_USE_RESPONSES_API:-} + LANGFUSE_AI_EXTRA_HEADERS: ${LANGFUSE_AI_EXTRA_HEADERS:-} + LANGFUSE_AI_AWS_BEDROCK_REGION: ${LANGFUSE_AI_AWS_BEDROCK_REGION:-} + LANGFUSE_AI_VERTEX_LOCATION: ${LANGFUSE_AI_VERTEX_LOCATION:-} + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN:-} + AWS_PROFILE: ${AWS_PROFILE:-} + # In-app agent worker settings (ignored by the web service). + LANGFUSE_IN_APP_AGENT_RUN_QUEUE_PROCESSING_CONCURRENCY: ${LANGFUSE_IN_APP_AGENT_RUN_QUEUE_PROCESSING_CONCURRENCY:-} + LANGFUSE_IN_APP_AGENT_AWS_PROFILE: ${LANGFUSE_IN_APP_AGENT_AWS_PROFILE:-} + LANGFUSE_IN_APP_AGENT_SANDBOX_PROVIDER: ${LANGFUSE_IN_APP_AGENT_SANDBOX_PROVIDER:-} + LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_IMAGE_IDENTIFIER: ${LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_IMAGE_IDENTIFIER:-} + LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_EXECUTION_ROLE_ARN: ${LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_EXECUTION_ROLE_ARN:-} + LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_EGRESS_NETWORK_CONNECTOR_ARN: ${LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_EGRESS_NETWORK_CONNECTOR_ARN:-} + LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_REGION: ${LANGFUSE_IN_APP_AGENT_SANDBOX_AWS_LAMBDA_MICROVM_REGION:-} + + langfuse-web: + image: docker.langfuse.com/langfuse/langfuse:4 + restart: always + depends_on: *langfuse-depends-on + ports: + - 3000:3000 + environment: + <<: *langfuse-worker-env + NEXTAUTH_SECRET: ${NEXTAUTH_SECRET:-mysecret} # CHANGEME + LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://localhost:9090} + LANGFUSE_S3_MEDIA_UPLOAD_INTERNAL_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_INTERNAL_ENDPOINT:-http://minio:9000} + LANGFUSE_INIT_ORG_ID: ${LANGFUSE_INIT_ORG_ID:-} + LANGFUSE_INIT_ORG_NAME: ${LANGFUSE_INIT_ORG_NAME:-} + LANGFUSE_INIT_PROJECT_ID: ${LANGFUSE_INIT_PROJECT_ID:-} + LANGFUSE_INIT_PROJECT_NAME: ${LANGFUSE_INIT_PROJECT_NAME:-} + LANGFUSE_INIT_PROJECT_PUBLIC_KEY: ${LANGFUSE_INIT_PROJECT_PUBLIC_KEY:-} + LANGFUSE_INIT_PROJECT_SECRET_KEY: ${LANGFUSE_INIT_PROJECT_SECRET_KEY:-} + LANGFUSE_INIT_USER_EMAIL: ${LANGFUSE_INIT_USER_EMAIL:-} + LANGFUSE_INIT_USER_NAME: ${LANGFUSE_INIT_USER_NAME:-} + LANGFUSE_INIT_USER_PASSWORD: ${LANGFUSE_INIT_USER_PASSWORD:-} + LANGFUSE_MARKDOWN_RENDER_CHARACTER_LIMIT: ${LANGFUSE_MARKDOWN_RENDER_CHARACTER_LIMIT:-} + # Hostnames allowed to reach the MCP endpoint, beyond NEXTAUTH_URL's own + # host. Required when the worker uses LANGFUSE_MCP_BASE_URL. + LANGFUSE_MCP_ALLOWED_HOSTS: ${LANGFUSE_MCP_ALLOWED_HOSTS:-} + # In-app agent settings. + LANGFUSE_IN_APP_AGENT_MAX_ACTIVE_RUNS_PER_USER: ${LANGFUSE_IN_APP_AGENT_MAX_ACTIVE_RUNS_PER_USER:-} + LANGFUSE_IN_APP_AGENT_MAX_ACTIVE_RUNS_PER_ORG: ${LANGFUSE_IN_APP_AGENT_MAX_ACTIVE_RUNS_PER_ORG:-} + + clickhouse: + image: docker.io/clickhouse/clickhouse-server:25.12 + restart: always + user: "101:101" + environment: + CLICKHOUSE_DB: default + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} # CHANGEME + volumes: + - langfuse_clickhouse_data:/var/lib/clickhouse + - langfuse_clickhouse_logs:/var/log/clickhouse-server + ports: + - 127.0.0.1:8123:8123 + - 127.0.0.1:9000:9000 + healthcheck: + test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1 + interval: 5s + timeout: 5s + retries: 10 + start_period: 1s + + minio: + image: cgr.dev/chainguard/minio + restart: always + entrypoint: sh + # create the 'langfuse' bucket before starting the service + command: -c 'mkdir -p /data/langfuse && minio server --address ":9000" --console-address ":9001" /data' + environment: + MINIO_ROOT_USER: ${MINIO_ROOT_USER:-minio} + MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-miniosecret} # CHANGEME + ports: + - 9090:9000 + - 127.0.0.1:9091:9001 + volumes: + - langfuse_minio_data:/data + healthcheck: + test: ["CMD", "mc", "ready", "local"] + interval: 1s + timeout: 5s + retries: 5 + start_period: 1s + + redis: + image: docker.io/redis:7 + restart: always + # CHANGEME: row below to secure redis password + command: > + --requirepass ${REDIS_AUTH:-myredissecret} + --maxmemory-policy noeviction + ports: + - 127.0.0.1:6379:6379 + volumes: + - langfuse_redis_data:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 3s + timeout: 10s + retries: 10 + + postgres: + image: docker.io/postgres:${POSTGRES_VERSION:-17} + restart: always + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 3s + timeout: 3s + retries: 10 + environment: + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} # CHANGEME + POSTGRES_DB: ${POSTGRES_DB:-postgres} + TZ: UTC + PGTZ: UTC + ports: + - 127.0.0.1:5432:5432 + volumes: + - langfuse_postgres_data:/var/lib/postgresql/data + +volumes: + langfuse_postgres_data: + driver: local + langfuse_clickhouse_data: + driver: local + langfuse_clickhouse_logs: + driver: local + langfuse_minio_data: + driver: local + langfuse_redis_data: + driver: local diff --git a/5_django_backend/docs-lib/README.md b/5_django_backend/docs-lib/README.md index f3afd18..258aad7 100644 --- a/5_django_backend/docs-lib/README.md +++ b/5_django_backend/docs-lib/README.md @@ -4,6 +4,7 @@ | 파일 | 뭐냐 | 출처 | |---|---|---| +| `langfuse.md` | 자체 호스팅 Langfuse ingestion API(배치 이벤트 모양·인증·207) + docker-compose env·헤드리스 초기화 키 | openapi.yml, langfuse.com/self-hosting (2026-09-22) | | `opencode-sdk-1.18.6-types.d.ts` | OpenCode 서버(`opencode serve`) HTTP API 전체 타입 — 세션·메시지·파트·이벤트·엔드포인트 url. `url: "/session…"` 로 grep 하면 엔드포인트, `export type Event…` 로 grep 하면 SSE 이벤트 | `@opencode-ai/sdk@1.18.6` `dist/gen/types.gen.d.ts` (ABAP_OPENCODE `.opencode/node_modules`) | 갱신: OpenCode 버전 올리면 같은 경로에서 다시 복사하고 파일명 버전 바꿈. diff --git a/5_django_backend/docs-lib/langfuse.md b/5_django_backend/docs-lib/langfuse.md new file mode 100644 index 0000000..3d21748 --- /dev/null +++ b/5_django_backend/docs-lib/langfuse.md @@ -0,0 +1,39 @@ +# Langfuse — 자체 호스팅 ingestion API (2026-09-22 정리) + +출처: `https://cloud.langfuse.com/generated/api/openapi.yml`, `langfuse.com/self-hosting`. SDK 안 쓰고 HTTP 로 직접 쏨(오프라인 wheels 반입 줄이려고). + +## POST /api/public/ingestion + +- 인증: 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+ 만. + +```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": {}}} + ] +} +``` + +- 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 가 나중에 붙음. + +## 자체 호스팅 (v4 docker-compose, 2026-09 기준) + +이미지 6개: `langfuse/langfuse:4`, `langfuse/langfuse-worker:4`, `clickhouse/clickhouse-server:25.12`, `cgr.dev/chainguard/minio`, `redis:7`, `postgres:17`. +web 포트 3000. 필수 env: `DATABASE_URL`, `NEXTAUTH_URL`, `NEXTAUTH_SECRET`, `SALT`, `ENCRYPTION_KEY`(hex 64자), CLICKHOUSE_*, REDIS_*, LANGFUSE_S3_*. + +헤드리스 초기화(첫 기동 때 조직·프로젝트·키·관리자 자동 생성, UI 클릭 없이): +`LANGFUSE_INIT_ORG_ID`, `LANGFUSE_INIT_ORG_NAME`, `LANGFUSE_INIT_PROJECT_ID`, `LANGFUSE_INIT_PROJECT_NAME`, +`LANGFUSE_INIT_PROJECT_PUBLIC_KEY`(pk-lf-…), `LANGFUSE_INIT_PROJECT_SECRET_KEY`(sk-lf-…), +`LANGFUSE_INIT_USER_EMAIL`, `LANGFUSE_INIT_USER_NAME`, `LANGFUSE_INIT_USER_PASSWORD`. +→ 우리 `deploy/langfuse/.env` 가 이걸 채우고, 같은 pk/sk 를 백엔드 `.env` 의 LANGFUSE_* 에 넣음. diff --git a/5_django_backend/tests/test_gateway.py b/5_django_backend/tests/test_gateway.py index a39bff1..2d65e09 100644 --- a/5_django_backend/tests/test_gateway.py +++ b/5_django_backend/tests/test_gateway.py @@ -126,6 +126,22 @@ async def test_401_retries_other_auth_shape_and_remembers(upstream): assert len(calls) == n + 1 and calls[-1].headers["x-openapi-token"] == "Bearer tok" +@override_settings(FABRIX_ENV=FULL_ENV, LANGFUSE={"host": "http://lf.test", "public_key": "pk", "secret_key": "sk"}) +async def test_stream_reports_generation_to_langfuse(upstream, monkeypatch): + import asyncio + + 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])) + 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" + + @override_settings(FABRIX_ENV=FULL_ENV) async def test_flaky_401_on_right_auth_is_retried_once(upstream, monkeypatch): """FabriX 가 맞는 조합에도 가끔 401 — 4개 다 돈 뒤 맨 앞 조합 한 번 더.""" diff --git a/5_django_backend/tests/test_langfuse.py b/5_django_backend/tests/test_langfuse.py new file mode 100644 index 0000000..69d01cf --- /dev/null +++ b/5_django_backend/tests/test_langfuse.py @@ -0,0 +1,87 @@ +"""Langfuse 전송 — 이벤트 모양·인증·게이트웨이/턴 마무리 훅. 서버는 MockTransport.""" + +import asyncio +import base64 +import json + +import httpx +import pytest +from django.test import override_settings + +from apps.gateway import langfuse + +LF = {"host": "http://lf.test", "public_key": "pk-lf-a", "secret_key": "sk-lf-b"} + + +@pytest.fixture +def lf_server(monkeypatch): + got: list[dict] = [] + + 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": []}) + + monkeypatch.setattr(langfuse, "TRANSPORT", httpx.MockTransport(handler)) + return got + + +@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 + req = lf_server[0] + assert req["url"] == "http://lf.test/api/public/ingestion" + 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") + + +@override_settings(LANGFUSE={"host": "", "public_key": "", "secret_key": ""}) +async def test_disabled_is_noop(lf_server): + assert not langfuse.enabled() + assert await langfuse.send([langfuse.trace("t", "x")]) is False + langfuse.send_later([langfuse.trace("t", "x")]) + assert lf_server == [] + + +def test_collect_parses_sse_text_and_usage(): + from apps.gateway.views import _Collect + + c = _Collect() + c.feed(b'data: {"choices":[{"delta":{"content":"SEL"}}]}\n\ndata: {"choices":[{"delta":{"co') + c.feed(b'ntent":"ECT"}}],"usage":{"prompt_tokens":7,"completion_tokens":2}}\n\ndata: [DONE]\n\n') + assert "".join(c.text) == "SELECT" and c.usage == {"prompt_tokens": 7, "completion_tokens": 2} + + +@pytest.mark.django_db(transaction=True) +async def test_finalize_sends_user_trace(lf_server, django_user_model, monkeypatch): + from apps.chat import stream as stream_mod + from apps.chat.models import ChatSession + from apps.chat.stream import TurnState, _finalize + from common.opencode_service import opencode_service + + async def _msgs(_sid): + return [{"info": {"role": "assistant", "tokens": {"input": 3, "output": 4, "reasoning": 1}, "time": {"created": 1000, "completed": 1500}}, "parts": [{"type": "text", "text": "답"}]}] + + async def _sess(_sid): + return {"title": "MARA 조회"} + + monkeypatch.setattr(opencode_service, "list_messages_a", _msgs) + monkeypatch.setattr(opencode_service, "get_session_a", _sess) + user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com") + session = await ChatSession.objects.acreate(id="ses_1", user=user, is_generating=True) + with override_settings(LANGFUSE=LF): + 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" + assert stream_mod is not None