From eaf6ea85b1197fe43a78f12c18a901d5f67cfc03 Mon Sep 17 00:00:00 2001 From: "cristiano.hoshikawa" Date: Fri, 14 Aug 2026 17:22:36 -0300 Subject: [PATCH] =?UTF-8?q?bugfix:=20Propaga=C3=A7=C3=A3o=20de=20sessionId?= =?UTF-8?q?=20no=20langfuse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md | 62 +++++++++++++++++++ .../observability/telemetry.py | 50 ++++++++++++--- tests/unit/test_telemetry_langfuse_compact.py | 60 ++++++++++++++++++ 3 files changed, 163 insertions(+), 9 deletions(-) create mode 100644 libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md diff --git a/libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md b/libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md new file mode 100644 index 0000000..37a2d76 --- /dev/null +++ b/libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md @@ -0,0 +1,62 @@ +# Langfuse native `sessionId` fix + +## Problema + +O Agent Framework já carregava `session_id` no contexto e no metadata de spans, +porém traces criados com Langfuse Python SDK v4 podiam aparecer com +`trace.sessionId = null`. Como consequência, `/api/public/sessions` não retornava +as conversas, embora `metadata.session_id` estivesse presente nas observations. + +## Causa + +No Langfuse Python SDK v4, atributos correlacionais como `session_id`, `user_id`, +tags e metadata devem ser aplicados por `propagate_attributes`, que é uma função +**de nível de módulo** (`from langfuse import propagate_attributes`). + +O framework tinha duas tentativas: + +1. `observation.update_trace(session_id=...)` — mantido por compatibilidade, mas + descontinuado no SDK v4; +2. `self.langfuse.propagate_attributes(...)` — formato incompatível com o SDK v4, + pois `propagate_attributes` não é método do client. + +## Correção + +`Telemetry` agora importa e mantém o callable de módulo do Langfuse v4 e o usa +imediatamente dentro do root span: + +```text +agent.gateway_message root span + -> propagate_attributes( + session_id=, + user_id=, + metadata=, + tags=, + trace_name= + ) + -> workflow / child observations +``` + +O fallback por método do client permanece para compatibilidade com SDKs ou +wrappers anteriores. + +## Resultado esperado + +Para uma sessão de negócio: + +```text +default:telecom_contas:f2a6e957-2c74-49ba-882e-ad14131cf1cc +``` + +o trace retornado pelo Langfuse deve apresentar: + +```json +{ + "sessionId": "default:telecom_contas:f2a6e957-2c74-49ba-882e-ad14131cf1cc" +} +``` + +e `/api/public/sessions` deve materializar/agrupar a sessão. + +O `session_id` continua também no metadata do framework para diagnóstico e +retrocompatibilidade. diff --git a/libs/agent_framework/src/agent_framework/observability/telemetry.py b/libs/agent_framework/src/agent_framework/observability/telemetry.py index 6605d1f..8f6c925 100644 --- a/libs/agent_framework/src/agent_framework/observability/telemetry.py +++ b/libs/agent_framework/src/agent_framework/observability/telemetry.py @@ -318,6 +318,11 @@ class Telemetry: def __init__(self, settings): self.settings = settings self.langfuse = None + # Langfuse SDK v4 exposes propagate_attributes as a module-level + # context manager (from langfuse import propagate_attributes), not as + # a Langfuse client method. Keep the callable on the Telemetry instance + # so the framework can support v4 while preserving legacy fallbacks. + self._langfuse_propagate_attributes = None self.enabled = bool(getattr(settings, "ENABLE_LANGFUSE", False)) self.event_bus = TelemetryEventBus() self.otel = OpenTelemetryProvider(settings) @@ -342,7 +347,12 @@ class Telemetry: return try: from langfuse import Langfuse + try: + from langfuse import propagate_attributes as langfuse_propagate_attributes + except ImportError: + langfuse_propagate_attributes = None self.langfuse = Langfuse(public_key=public_key, secret_key=secret_key, host=host) + self._langfuse_propagate_attributes = langfuse_propagate_attributes logger.info("Langfuse habilitado host=%s", host) except Exception as exc: logger.exception("Falha ao inicializar Langfuse: %s", exc) @@ -884,25 +894,47 @@ class Telemetry: except Exception: logger.debug("Trace input/output update não suportado", exc_info=True) def _start_trace_attribute_propagation(self, name: str, attrs: dict[str, Any]): - if not self.is_enabled() or not hasattr(self.langfuse, "propagate_attributes"): + """Propagate native Langfuse trace attributes, including session_id. + + Langfuse Python SDK v4 moved ``propagate_attributes`` to a module-level + context manager. Calling ``observation.update_trace(session_id=...)`` is + not sufficient/recommended in v4 and, in practice, left ``sessionId`` + unset on traces even though the framework metadata contained + ``session_id``. + + Prefer the v4 module-level callable imported during initialization. A + client-method fallback is retained for older/custom SDK versions. + """ + if not self.is_enabled(): return None + metadata = { k: attrs.get(k) for k in ("request_id", "trace_id", "agent_id", "tenant_id", "channel", "message_id", "ura_call_id", "workflow_id") if attrs.get(k) } tags = attrs.get("tags") if isinstance(attrs.get("tags"), list) else None + kwargs = { + "user_id": str(attrs["user_id"]) if attrs.get("user_id") is not None else None, + "session_id": str(attrs["session_id"]) if attrs.get("session_id") is not None else None, + "metadata": metadata or None, + "tags": [str(tag) for tag in tags] if tags else None, + "trace_name": name, + } + try: - return self.langfuse.propagate_attributes( - user_id=str(attrs["user_id"]) if attrs.get("user_id") is not None else None, - session_id=str(attrs["session_id"]) if attrs.get("session_id") is not None else None, - metadata=metadata or None, - tags=[str(tag) for tag in tags] if tags else None, - trace_name=name, - ) + # Langfuse SDK v4: ``from langfuse import propagate_attributes``. + if callable(self._langfuse_propagate_attributes): + return self._langfuse_propagate_attributes(**kwargs) + + # Backward compatibility for SDK builds/wrappers that exposed the + # propagation context manager on the client instance. + legacy_propagate = getattr(self.langfuse, "propagate_attributes", None) + if callable(legacy_propagate): + return legacy_propagate(**kwargs) except Exception: logger.debug("Trace attribute propagation não suportada", exc_info=True) - return None + return None class _LegacyObservationContext: def __init__(self, observation): self.observation = observation diff --git a/tests/unit/test_telemetry_langfuse_compact.py b/tests/unit/test_telemetry_langfuse_compact.py index 785c4f6..152cfa4 100644 --- a/tests/unit/test_telemetry_langfuse_compact.py +++ b/tests/unit/test_telemetry_langfuse_compact.py @@ -227,3 +227,63 @@ async def test_legacy_io_fallback_updates_same_root_and_generation_observations( assert root_event.body.input == {"request": "cms"} assert root_event.body.output == {"answer": "ok"} assert len(telemetry.langfuse.observations) == 2 + +@pytest.mark.asyncio +async def test_langfuse_v4_module_level_propagation_sets_native_session_context(): + """SDK v4 propagation must receive the business session id natively.""" + clear_observability_context() + telemetry = telemetry_with_fake_langfuse() + calls = [] + + def v4_propagate_attributes(**kwargs): + calls.append(kwargs) + return FakePropagationContext(telemetry.langfuse, kwargs) + + # Simulates ``from langfuse import propagate_attributes`` from SDK v4. + telemetry._langfuse_propagate_attributes = v4_propagate_attributes + + async with telemetry.span( + "agent.gateway_message", + session_id="default:telecom_contas:session-123", + user_id="11999999999", + agent_id="telecom_contas", + tenant_id="default", + input={"message": "hello"}, + tags=["agent:telecom_contas"], + _root_span=True, + ): + await telemetry.event("IC.TEST", {"ok": True}, kind="ic") + + assert len(calls) == 1 + assert calls[0]["session_id"] == "default:telecom_contas:session-123" + assert calls[0]["user_id"] == "11999999999" + assert calls[0]["trace_name"] == "agent.gateway_message" + assert calls[0]["metadata"]["agent_id"] == "telecom_contas" + assert calls[0]["metadata"]["tenant_id"] == "default" + assert calls[0]["tags"] == ["agent:telecom_contas"] + + # Keep the legacy update as a compatibility fallback, but the v4 path above + # is now the authoritative way to materialize native sessionId. + root = telemetry.langfuse.observations[0] + assert any( + update.get("session_id") == "default:telecom_contas:session-123" + for update in root.trace_updates + ) + + +def test_trace_attribute_propagation_keeps_legacy_client_method_fallback(): + telemetry = telemetry_with_fake_langfuse() + telemetry._langfuse_propagate_attributes = None + + cm = telemetry._start_trace_attribute_propagation( + "agent.gateway_message", + { + "session_id": "legacy-session", + "user_id": "legacy-user", + "agent_id": "legacy-agent", + }, + ) + assert cm is not None + with cm: + pass + assert telemetry.langfuse.propagations[-1]["session_id"] == "legacy-session"