lee-hyeon-cheolandClaude Fable 5.1
68b55907f0
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
134 lines
7.5 KiB
Python
134 lines
7.5 KiB
Python
"""Langfuse 전송(OTLP/HTTP JSON) — span 모양·인증·게이트웨이/턴 마무리 훅. 서버는 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", "gateway": False}
|
|
|
|
|
|
@pytest.fixture
|
|
def lf_server(monkeypatch):
|
|
got: list[dict] = []
|
|
|
|
def handler(req: httpx.Request) -> httpx.Response:
|
|
if req.headers.get("content-type", "").startswith("application/x-protobuf"):
|
|
from google.protobuf.json_format import MessageToDict
|
|
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
|
|
|
|
m = ExportTraceServiceRequest(); m.ParseFromString(req.content)
|
|
body = MessageToDict(m) # bytes 필드는 base64 로 나옴
|
|
else:
|
|
body = json.loads(req.content)
|
|
got.append({"url": str(req.url), "auth": req.headers.get("authorization"), "body": body, "ctype": req.headers.get("content-type", "")})
|
|
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_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/otel/v1/traces"
|
|
assert req["auth"] == "Basic " + base64.b64encode(b"pk-lf-a:sk-lf-b").decode()
|
|
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": ""})
|
|
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):
|
|
tool = {"type": "tool", "tool": "sap-icf_get_program_source", "state": {"status": "completed", "input": {"program_name": "ZTEST"}, "output": "REPORT ztest.", "title": "ZTEST", "time": {"start": 1100, "end": 1300}}}
|
|
bad = {"type": "tool", "tool": "read", "state": {"status": "error", "input": {"filePath": "x"}, "error": "no such file", "time": {"start": 1300, "end": 1310}}}
|
|
return [{"info": {"role": "assistant"}, "parts": [tool, bad]},
|
|
{"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
|
|
spans = lf_server[0]["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"]
|
|
assert [sp["name"] for sp in spans] == ["chat", "sap-icf_get_program_source", "read"]
|
|
t = _attrs(spans[0])
|
|
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 t["langfuse.observation.type"] == "generation" # 루트가 곧 답변 — 토큰은 여기
|
|
assert json.loads(t["langfuse.observation.usage_details"]) == {"input": 3, "output": 5, "total": 8}
|
|
a1, a2 = _attrs(spans[1]), _attrs(spans[2])
|
|
assert spans[1]["parentSpanId"] == spans[0]["spanId"] and a1["langfuse.observation.type"] == "tool"
|
|
assert json.loads(a1["langfuse.observation.input"]) == {"program_name": "ZTEST"} and a1["langfuse.observation.output"] == "REPORT ztest."
|
|
assert spans[1]["startTimeUnixNano"] == "1100000000" and spans[1]["endTimeUnixNano"] == "1300000000"
|
|
assert spans[2]["status"] == {"code": 2, "message": "no such file"} and a2["langfuse.observation.level"] == "ERROR"
|
|
assert stream_mod is not None
|
|
|
|
|
|
@override_settings(LANGFUSE={"host": "", "public_key": "", "secret_key": "", "gateway": False, "phoenix": "http://px.test", "project": "codeassist"})
|
|
async def test_phoenix_target_no_auth_and_openinference_attrs(lf_server):
|
|
root = langfuse.trace("t9", "chat", userId="u@x.com", sessionId="s9", input="질문", output="답", usage=langfuse.usage_of(7, 3), model="581")
|
|
tool = langfuse.tool_span("t9", "read", parent=root, input={"filePath": "a.md"}, output="본문")
|
|
assert await langfuse.send([root, tool]) and len(lf_server) == 1
|
|
req = lf_server[0]
|
|
assert req["url"] == "http://px.test/v1/traces" and req["auth"] is None and req["ctype"].startswith("application/x-protobuf")
|
|
res = {a["key"]: a["value"]["stringValue"] for a in req["body"]["resourceSpans"][0]["resource"]["attributes"]}
|
|
assert res["openinference.project.name"] == "codeassist"
|
|
r, tl = _attrs(req["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"][0]), _attrs(req["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"][1])
|
|
assert r["openinference.span.kind"] == "LLM" and r["input.value"] == "질문" and r["output.value"] == "답"
|
|
assert r["llm.model_name"] == "581" and r["llm.token_count.total"] == "10" and r["session.id"] == "s9" and r["user.id"] == "u@x.com"
|
|
assert tl["openinference.span.kind"] == "TOOL" and tl["tool.name"] == "read" and json.loads(tl["input.value"]) == {"filePath": "a.md"}
|
|
|