diff --git a/5_django_backend/apps/chat/stream.py b/5_django_backend/apps/chat/stream.py index 80fb018..d4c8139 100644 --- a/5_django_backend/apps/chat/stream.py +++ b/5_django_backend/apps/chat/stream.py @@ -192,6 +192,21 @@ class TurnState: return out +def _tool_spans(tid: str, root: dict, tool_parts: list[dict]) -> list[dict]: + """OpenCode 도구 파트 → Langfuse 자식 span. 출력은 앞 4KB 만(원문은 OpenCode 에 있음).""" + out = [] + for p in tool_parts: + st = p.get("state") or {} + t = st.get("time") or {} + out.append(langfuse.tool_span( + tid, p.get("tool") or "tool", parent=root, input=st.get("input"), title=st.get("title") or "", + output=(st.get("output") or "")[:4000] or None, + start=langfuse.ms_iso(t.get("start")), end=langfuse.ms_iso(t.get("end")), + error=st.get("error") if st.get("status") == "error" else None, + )) + return out + + def _real_title(raw) -> str | None: """OpenCode 기본 제목("New session - 2026-…")은 제목이 아님 → None.""" t = (raw or "").strip() @@ -212,12 +227,15 @@ async def _finalize(session: ChatSession, state: TurnState, started: float, *, f content = state.accumulated() usage: dict | None = None title = state.title + tool_parts: list[dict] = [] try: msgs = await opencode_service.list_messages_a(session.id) or [] assistant = [m for m in msgs if (m.get("info") or {}).get("role") == "assistant"] if assistant: info = assistant[-1]["info"] parts = assistant[-1].get("parts") or [] + # 이 턴의 assistant 메시지가 여러 개(도구 왕복마다 하나)라 도구 파트는 전부 모음 + tool_parts = [p for m in assistant for p in (m.get("parts") or []) if p.get("type") == "tool"] full = "".join(p.get("text", "") for p in parts if p.get("type") == "text" and not p.get("synthetic")) if full: content = full @@ -251,25 +269,18 @@ async def _finalize(session: ChatSession, state: TurnState, started: float, *, f session.title_llm = title[:200] await session.asave(update_fields=["is_generating", "title_llm", "updated_at"]) - # 관측: 사용자 단위 trace 하나 + 답변 generation 하나. 실패해도 여기까진 이미 저장됨. + # 관측: 목록 한 줄 = 이 채팅(질문·답·토큰). 트리 = 답 만들며 부른 도구들(SAP 조회·위키 read…) 순서대로. 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 {} 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([ - root, - langfuse.generation( - 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 "", - metadata={"elapsed_ms": usage["elapsed_ms"]} if usage else {}, - ), - ]) + root = langfuse.trace( + tid, "chat", userId=user_email, sessionId=session.id, input=state.user_text, output=content, + tags=["codeassist"], start=t0, end=langfuse.now_iso(), + usage=langfuse.usage_of(usage["input"], usage["output"]) if usage else None, + metadata={"elapsed_ms": usage["elapsed_ms"]} if usage else {}, error=failed, + ) + langfuse.send_later([root, *_tool_spans(tid, root, tool_parts)]) return usage, title diff --git a/5_django_backend/apps/gateway/langfuse.py b/5_django_backend/apps/gateway/langfuse.py index 1cb6291..03531b8 100644 --- a/5_django_backend/apps/gateway/langfuse.py +++ b/5_django_backend/apps/gateway/langfuse.py @@ -78,11 +78,17 @@ def _span(trace_id: str, name: str, attrs: dict, *, start: str | None, end: str 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 속성(이름·사용자·세션·태그)은 여기 실림.""" + metadata: dict | None = None, tags: list[str] | None = None, start: str | None = None, end: str | None = None, + usage: dict | None = None, model: str = "", error: str | None = None) -> dict: + """루트 span. trace 속성(이름·사용자·세션·태그)은 여기 실림. + usage 를 주면 루트 자체가 generation — 목록 한 줄 = 질문/답/토큰, 밑에 도구 호출만 트리로.""" attrs = { "langfuse.trace.name": name, - "langfuse.observation.type": "span", + "langfuse.observation.type": "generation" if usage else "span", + "langfuse.observation.model.name": model, + "langfuse.observation.usage_details": usage, + "langfuse.observation.level": "ERROR" if error else "DEFAULT", + "langfuse.observation.status_message": error or "", "langfuse.user.id": userId, "langfuse.session.id": sessionId, "langfuse.trace.input": input, @@ -92,7 +98,29 @@ def trace(trace_id: str, name: str, *, userId: str = "", sessionId: str = "", in "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) + return _span(trace_id, name, attrs, start=start, end=end, parent=None, error=error) + + +def tool_span(trace_id: str, name: str, *, parent: dict, input=None, output=None, title: str = "", + start: str | None = None, end: str | None = None, error: str | None = None) -> dict: + """도구 호출 하나(SAP 조회·위키 read·grep…). 트리에서 루트 밑에 순서대로 보임.""" + inherited = {a["key"]: a["value"].get("stringValue") for a in parent.get("attributes", []) + if a["key"] in ("langfuse.user.id", "langfuse.session.id", "langfuse.trace.name", "langfuse.trace.tags")} + attrs = { + **inherited, + "langfuse.observation.type": "tool", + "langfuse.observation.input": input, + "langfuse.observation.output": output, + "langfuse.observation.level": "ERROR" if error else "DEFAULT", + "langfuse.observation.status_message": error or "", + "langfuse.observation.metadata.title": title, + } + return _span(trace_id, name, attrs, start=start, end=end, parent=parent["spanId"], error=error) + + +def ms_iso(ms: int | float | None) -> str | None: + """OpenCode 의 epoch ms → ISO. 없으면 None(=지금).""" + return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat() if ms else None def generation(trace_id: str, name: str, *, model: str = "", input=None, output=None, usage: dict | None = None, diff --git a/5_django_backend/tests/test_langfuse.py b/5_django_backend/tests/test_langfuse.py index 08b8c65..d522f06 100644 --- a/5_django_backend/tests/test_langfuse.py +++ b/5_django_backend/tests/test_langfuse.py @@ -78,7 +78,10 @@ async def test_finalize_sends_user_trace(lf_server, django_user_model, monkeypat 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": "답"}]}] + 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 조회"} @@ -92,10 +95,15 @@ async def test_finalize_sends_user_trace(lf_server, django_user_model, monkeypat await asyncio.gather(*list(langfuse._pending)) assert len(lf_server) == 1 spans = lf_server[0]["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"] - t, g = _attrs(spans[0]), _attrs(spans[1]) + 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 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 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