Skip to content

feat(langfuse): attribute traces to the messaging end-user (userId) #43491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kamonspecial wants to merge 1 commit into from
+328 −36
Closed

feat(langfuse): attribute traces to the messaging end-user (userId) #43491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 83 additions & 36 deletions plugins/observability/langfuse/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@
HERMES_LANGFUSE_SAMPLE_RATE - sampling rate 0.0–1.0 (default: 1.0)
HERMES_LANGFUSE_MAX_CHARS - max chars per field (default: 12000)
HERMES_LANGFUSE_DEBUG - set to "true" for verbose logging
HERMES_LANGFUSE_END_USER_ATTRIBUTION
- set to "true" to attribute the trace userId to
the messaging end-user (sender_id); default off
"""
from __future__ import annotations

Expand Down Expand Up @@ -50,6 +53,7 @@ class TraceState:
pending_tools_by_name: Dict[str, list] = field(default_factory=dict)
turn_tool_calls: list[dict[str, Any]] = field(default_factory=list)
last_updated_at: float = field(default_factory=time.time)
session_id: str = ""


_STATE_LOCK = threading.Lock()
Expand All @@ -63,6 +67,10 @@ class TraceState:
# is far above any realistic concurrent-live-turn working set; it exists only
# to bound the leak from non-finalizing turns, not to limit concurrency.
_MAX_TRACE_STATE = 256
# session_id -> messaging end-user id (sender_id). Captured from the turn-scoped
# pre_llm_call hook so the root trace (created later from pre_api_request, which
# carries no user identity) can attribute itself to that user via Langfuse user_id.
_SENDER_BY_SESSION: Dict[str, str] = {}
_LANGFUSE_CLIENT = None
_READ_FILE_LINE_RE = re.compile(r"^\s*(\d+)\|(.*)$")
_READ_FILE_HEAD_LINES = 25
Expand Down Expand Up @@ -96,6 +104,14 @@ def _debug_enabled() -> bool:
return _env_bool("HERMES_LANGFUSE_DEBUG")


def _end_user_attribution_enabled() -> bool:
# Opt-in (default off): attribute the trace userId to the messaging
# end-user (sender_id). Kept off by default so it never conflicts with
# other userId semantics (e.g. assigning the active profile, #26455) — when
# unset the trace carries no user, exactly like the pre-existing behavior.
return _env_bool("HERMES_LANGFUSE_END_USER_ATTRIBUTION")


def _debug(message: str) -> None:
if _debug_enabled():
logger.info("Langfuse tracing: %s", message)
Expand Down Expand Up @@ -602,7 +618,7 @@ def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str,

def _start_root_trace(task_key: str, *, task_id: str, session_id: str, platform: str, provider: str, model: str,
api_mode: str, messages: Any, client: Langfuse,
turn_id: str = "", api_request_id: str = "") -> TraceState:
turn_id: str = "", api_request_id: str = "", user_id: str = "") -> TraceState:
trace_id = client.create_trace_id(seed=f"{session_id or 'sessionless'}::{task_id or task_key}")
trace_input = _extract_last_user_message(messages)
metadata = {
Expand All @@ -615,56 +631,62 @@ def _start_root_trace(task_key: str, *, task_id: str, session_id: str, platform:
"model": model,
"api_mode": api_mode,
}
# Also record the end-user on trace metadata as a fallback: an older SDK
# whose propagate_attributes drops the user_id kwarg (see the retry below)
# still surfaces the user here.
if user_id:
metadata["user_id"] = user_id

# session_id must be passed in trace_context for Langfuse session grouping.
trace_ctx: Dict[str, Any] = {"trace_id": trace_id}
if session_id:
trace_ctx["session_id"] = session_id

if propagate_attributes is not None:
try:
with propagate_attributes(
session_id=session_id or task_key,
trace_name="Hermes turn",
tags=["hermes", "langfuse"],
):
root_ctx = client.start_as_current_observation(
trace_context=trace_ctx,
name="Hermes turn",
as_type="chain",
input=trace_input,
metadata=metadata,
end_on_exit=False,
)
root_span = root_ctx.__enter__()
except Exception:
root_ctx = client.start_as_current_observation(
trace_context=trace_ctx,
name="Hermes turn",
as_type="chain",
input=trace_input,
metadata=metadata,
end_on_exit=False,
)
root_span = root_ctx.__enter__()
else:
root_ctx = client.start_as_current_observation(
def _open_root() -> tuple[Any, Any]:
ctx = client.start_as_current_observation(
trace_context=trace_ctx,
name="Hermes turn",
as_type="chain",
input=trace_input,
metadata=metadata,
end_on_exit=False,
)
root_span = root_ctx.__enter__()
return ctx, ctx.__enter__()

if propagate_attributes is not None:
attr_kwargs: Dict[str, Any] = {
"session_id": session_id or task_key,
"trace_name": "Hermes turn",
"tags": ["hermes", "langfuse"],
}
if user_id:
attr_kwargs["user_id"] = user_id
try:
try:
attr_cm = propagate_attributes(**attr_kwargs)
except TypeError:
# Older Langfuse SDKs don't accept user_id on
# propagate_attributes. Drop only that kwarg and retry so
# session grouping / trace_name / tags are still applied —
# without this, an unsupported user_id would fall through to
# the bare-observation fallback below and silently lose all
# three.
attr_kwargs.pop("user_id", None)
attr_cm = propagate_attributes(**attr_kwargs)
with attr_cm:
root_ctx, root_span = _open_root()
except Exception:
root_ctx, root_span = _open_root()
else:
root_ctx, root_span = _open_root()

try:
root_span.set_trace_io(input=trace_input)
except Exception:
pass

_debug(f"started trace {trace_id} for {task_key}")
return TraceState(trace_id=trace_id, root_ctx=root_ctx, root_span=root_span)
return TraceState(trace_id=trace_id, root_ctx=root_ctx, root_span=root_span, session_id=session_id)


def _start_child_observation(state: TraceState, *, client: Langfuse, name: str, as_type: str,
Expand Down Expand Up @@ -728,6 +750,11 @@ def _evict_stale_locked() -> None:
stale = sorted(_TRACE_STATE.items(), key=lambda kv: kv[1].last_updated_at)[:over]
for key, state in stale:
_TRACE_STATE.pop(key, None)
# Bound _SENDER_BY_SESSION by the same eviction that bounds _TRACE_STATE:
# a turn that never reaches _finish_trace (interrupted / tool-only final
# step) would otherwise leave its sender entry behind forever.
if state.session_id:
_SENDER_BY_SESSION.pop(state.session_id, None)
try:
state.root_span.end()
except Exception as exc: # pragma: no cover - fail-open
Expand All @@ -741,6 +768,8 @@ def _finish_trace(task_key: str, *, output: Any = None) -> None:

with _STATE_LOCK:
state = _TRACE_STATE.pop(task_key, None)
if state is not None and state.session_id:
_SENDER_BY_SESSION.pop(state.session_id, None)
if state is None:
return

Expand Down Expand Up @@ -778,7 +807,27 @@ def on_pre_llm_call(*, task_id: str = "", session_id: str = "", platform: str =
provider: str = "", base_url: str = "", api_mode: str = "",
api_call_count: int = 0, messages: Any = None, turn_type: str = "user",
conversation_history: Any = None, user_message: Any = None,
turn_id: str = "", api_request_id: str = "", **_: Any) -> None:
turn_id: str = "", api_request_id: str = "",
sender_id: str = "", **_: Any) -> None:
client = _get_langfuse()
if client is None:
return

# End-user attribution is opt-in (default off): when disabled, drop the
# sender so every downstream path (stash, request-shaped trace creation,
# and the pre_api_request read) sees no user and behaves like upstream.
if sender_id and not _end_user_attribution_enabled():
sender_id = ""

# Capture the messaging end-user (emitted on the turn-scoped pre_llm_call)
# before the early return below, keyed by session_id, so the root trace
# created later from pre_api_request — which carries no user identity — can
# attribute itself to that user. Gated on an active client so it stays
# symmetric with the cleanup in _finish_trace / _evict_stale_locked.
if session_id and sender_id:
with _STATE_LOCK:
_SENDER_BY_SESSION[session_id] = sender_id

# Older Hermes branches used pre_llm_call for request-scoped tracing and
# passed the actual API messages. Current Hermes also has a turn-scoped
# pre_llm_call used for context injection; tracing that hook creates an
Expand All @@ -787,10 +836,6 @@ def on_pre_llm_call(*, task_id: str = "", session_id: str = "", platform: str =
if not isinstance(messages, list):
return

client = _get_langfuse()
if client is None:
return

# messages is a list only for legacy Hermes branches that fired
# pre_llm_call with API messages directly. Current Hermes fires
# pre_llm_call for context injection (conversation_history/user_message,
Expand All @@ -817,6 +862,7 @@ def on_pre_llm_call(*, task_id: str = "", session_id: str = "", platform: str =
client=client,
turn_id=turn_id,
api_request_id=api_request_id,
user_id=sender_id,
)
_evict_stale_locked()
_TRACE_STATE[task_key] = state
Expand Down Expand Up @@ -881,6 +927,7 @@ def on_pre_llm_request(
client=client,
turn_id=turn_id,
api_request_id=api_request_id,
user_id=_SENDER_BY_SESSION.get(session_id, ""),
)
_evict_stale_locked()
_TRACE_STATE[task_key] = state
Expand Down
Loading
Loading