feat(observability): Langfuse 전송 — 턴 마무리(사용자 trace)·게이트웨이(FabriX generation) 두 훅, SDK 없이 HTTP. deploy/langfuse 에 compose+env+반입 절차

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
lee-hyeon-cheol
2026-09-22 13:44:12 +09:00
co-authored by Claude Fable 5.1
parent 95b169a7b9
commit 0321050053
13 changed files with 617 additions and 0 deletions
+5
View File
@@ -62,3 +62,8 @@ SAP_API_BASE_ICF=
# ── ABAP 소스 인덱스(ABAP_INDEXING, 별도 서버 :8100) — OpenCode 툴 abap_index_* 가 씀 ── # ── ABAP 소스 인덱스(ABAP_INDEXING, 별도 서버 :8100) — OpenCode 툴 abap_index_* 가 씀 ──
# 비우면 기본 127.0.0.1:8100. 서버가 없어도 앱은 돌아감(툴이 "없음"으로 답하고 생성 코드로). # 비우면 기본 127.0.0.1:8100. 서버가 없어도 앱은 돌아감(툴이 "없음"으로 답하고 생성 코드로).
ABAP_INDEX_URL=http://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=
+21
View File
@@ -20,12 +20,14 @@ import asyncio
import json import json
import logging import logging
import time import time
from datetime import datetime, timezone
from django.conf import settings from django.conf import settings
from django.http import StreamingHttpResponse from django.http import StreamingHttpResponse
from django.views.decorators.csrf import csrf_exempt from django.views.decorators.csrf import csrf_exempt
from apps.accounts.authentication import user_from_token from apps.accounts.authentication import user_from_token
from apps.gateway import langfuse
from asgiref.sync import sync_to_async from asgiref.sync import sync_to_async
from common.envelope import json_error from common.envelope import json_error
from common.opencode_service import opencode_service from common.opencode_service import opencode_service
@@ -248,6 +250,25 @@ async def _finalize(session: ChatSession, state: TurnState, started: float, *, f
if title: if title:
session.title_llm = title[:200] session.title_llm = title[:200]
await session.asave(update_fields=["is_generating", "title_llm", "updated_at"]) 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 return usage, title
+85
View File
@@ -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)
+66
View File
@@ -11,6 +11,9 @@ from __future__ import annotations
import asyncio import asyncio
import json import json
import time
import uuid
from datetime import datetime, timezone
import logging import logging
from typing import AsyncIterator from typing import AsyncIterator
@@ -19,6 +22,7 @@ from django.conf import settings
from django.http import HttpRequest, HttpResponse, JsonResponse, StreamingHttpResponse from django.http import HttpRequest, HttpResponse, JsonResponse, StreamingHttpResponse
from django.views.decorators.csrf import csrf_exempt from django.views.decorators.csrf import csrf_exempt
from . import langfuse
from .fabrix import FabrixConfig, Variant, auth_variants from .fabrix import FabrixConfig, Variant, auth_variants
log = logging.getLogger(__name__) 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" 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: async def healthcheck(_request: HttpRequest) -> JsonResponse:
return JsonResponse({"success": True}) return JsonResponse({"success": True})
@@ -95,6 +150,9 @@ async def chat_completions(request: HttpRequest) -> HttpResponse:
# FabriX 가 맞는 조합에도 가끔 401 을 뱉음(2026-09-21 고객사 실측: curl 10번 중 1~2번). # FabriX 가 맞는 조합에도 가끔 401 을 뱉음(2026-09-21 고객사 실측: curl 10번 중 1~2번).
# 4개 조합 다 돌고 나서 제일 유력한 조합(맨 앞)을 잠깐 쉬고 한 번 더 — 그래도 401 이면 진짜 인증 문제 # 4개 조합 다 돌고 나서 제일 유력한 조합(맨 앞)을 잠깐 쉬고 한 번 더 — 그래도 401 이면 진짜 인증 문제
variants = variants + variants[:1] variants = variants + variants[:1]
model_id = headers.get("x-llm-model-id", "")
started = time.time()
col = _Collect()
if body.get("stream"): if body.get("stream"):
@@ -119,9 +177,11 @@ async def chat_completions(request: HttpRequest) -> HttpResponse:
return return
_last_ok["variant"] = vkey _last_ok["variant"] = vkey
async for chunk in resp.aiter_raw(): async for chunk in resp.aiter_raw():
col.feed(chunk)
yield chunk yield chunk
finally: finally:
await resp.aclose() 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 = StreamingHttpResponse(gen(), content_type="text/event-stream")
out["Cache-Control"] = "no-cache" out["Cache-Control"] = "no-cache"
@@ -141,4 +201,10 @@ async def chat_completions(request: HttpRequest) -> HttpResponse:
break break
except httpx.HTTPError as e: except httpx.HTTPError as e:
return _err(502, f"FabriX 호출 실패: {type(e).__name__}: {e}", "upstream_error") 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") return HttpResponse(resp.content, status=resp.status_code, content_type="application/json")
+6
View File
@@ -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 from apps.gateway.fabrix import ENV_KEYS as _FABRIX_KEYS # noqa: E402
FABRIX_ENV = {k: _env(k) for k in _FABRIX_KEYS} 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 로 프론트 게이지에 감 # 세션 컨텍스트 하드 한도 — usage.limit 로 프론트 게이지에 감
CONTEXT_LIMIT_TOKENS = int(_env("CONTEXT_LIMIT_TOKENS", "128000") or "128000") CONTEXT_LIMIT_TOKENS = int(_env("CONTEXT_LIMIT_TOKENS", "128000") or "128000")
+1
View File
@@ -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 grep -q '^OPENCODE_VISION_MODEL=' .env || echo 'OPENCODE_VISION_MODEL=gateway/605' >> .env
# 고객사 FabriX 는 Bearer 접두 필수(9/18 실측). 첫 시도부터 맞게 — 날것으로 보내면 401 뒤 재시도에 기댐 # 고객사 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' 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 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 .venv/bin/python manage.py migrate --noinput
[ -f seed/snippets.db ] && .venv/bin/python manage.py import_snippets seed/snippets.db [ -f seed/snippets.db ] && .venv/bin/python manage.py import_snippets seed/snippets.db
@@ -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
@@ -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 로그만)
@@ -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
+1
View File
@@ -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-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 버전 올리면 같은 경로에서 다시 복사하고 파일명 버전 바꿈. 갱신: OpenCode 버전 올리면 같은 경로에서 다시 복사하고 파일명 버전 바꿈.
+39
View File
@@ -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_* 에 넣음.
+16
View File
@@ -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" 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) @override_settings(FABRIX_ENV=FULL_ENV)
async def test_flaky_401_on_right_auth_is_retried_once(upstream, monkeypatch): async def test_flaky_401_on_right_auth_is_retried_once(upstream, monkeypatch):
"""FabriX 가 맞는 조합에도 가끔 401 — 4개 다 돈 뒤 맨 앞 조합 한 번 더.""" """FabriX 가 맞는 조합에도 가끔 401 — 4개 다 돈 뒤 맨 앞 조합 한 번 더."""
+87
View File
@@ -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