mirror of
https://github.com/hoshikawa2/agent_platform_oci.git
synced 2026-09-07 10:13:46 +00:00
bugfix: Propagação de sessionId no langfuse
This commit is contained in:
62
libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md
Normal file
62
libs/agent_framework/docs/LANGFUSE_NATIVE_SESSION_ID_FIX.md
Normal file
@@ -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=<agent_session_id>,
|
||||
user_id=<user_id>,
|
||||
metadata=<correlation metadata>,
|
||||
tags=<root tags>,
|
||||
trace_name=<root span 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.
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user