diff --git a/.env.example b/.env.example
index 0b0ac9a..874b1c2 100644
--- a/.env.example
+++ b/.env.example
@@ -72,6 +72,8 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
+LANGFUSE_ROOT_SPAN_NAME=agent.gateway_message
+LANGFUSE_LEGACY_IO_FALLBACK=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_HOST=http://localhost:3005
diff --git a/.idea/workspace.xml b/.idea/workspace.xml
index 8feba4f..189ce06 100644
--- a/.idea/workspace.xml
+++ b/.idea/workspace.xml
@@ -4,9 +4,7 @@
-
-
-
+
diff --git a/libs/agent_framework/src/agent_framework/config/settings.py b/libs/agent_framework/src/agent_framework/config/settings.py
index 599d39f..247381c 100644
--- a/libs/agent_framework/src/agent_framework/config/settings.py
+++ b/libs/agent_framework/src/agent_framework/config/settings.py
@@ -100,6 +100,8 @@ class Settings(BaseSettings):
ENABLE_LANGFUSE: bool = False
LANGFUSE_TRACE_MODE: Literal['verbose','compact'] = 'verbose'
+ LANGFUSE_ROOT_SPAN_NAME: str = 'agent.gateway_message'
+ LANGFUSE_LEGACY_IO_FALLBACK: bool = True
LANGFUSE_PUBLIC_KEY: str | None = None
LANGFUSE_SECRET_KEY: str | None = None
LANGFUSE_HOST: str = 'https://cloud.langfuse.com'
diff --git a/libs/agent_framework/src/agent_framework/llm/providers.py b/libs/agent_framework/src/agent_framework/llm/providers.py
index ecd4235..610d2d5 100644
--- a/libs/agent_framework/src/agent_framework/llm/providers.py
+++ b/libs/agent_framework/src/agent_framework/llm/providers.py
@@ -74,20 +74,23 @@ class MockLLMProvider(LLMProvider):
profile_found = kwargs.get("profile_found")
profiles_enabled = kwargs.get("profiles_enabled")
profiles_path = kwargs.get("profiles_path")
- last = messages[-1].get("content", "") if messages else ""
- answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
- usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
+ llm_metadata = {"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}
+ async with _maybe_generation(
+ self.telemetry,
+ name=generation_name,
+ model=model,
+ input=messages,
+ metadata=llm_metadata,
+ model_parameters={},
+ ) as generation:
+ last = messages[-1].get("content", "") if messages else ""
+ answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
+ usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
+ generation.set_output(answer)
+ generation.set_usage(usage)
+ generation.set_metadata(**usage)
if self.usage_repository:
- await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, {"provider":"mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}))
- if self.telemetry:
- await self.telemetry.generation(
- name=generation_name,
- model=model,
- input=messages,
- output=answer,
- metadata={"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path, **usage},
- usage=usage,
- )
+ await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, llm_metadata))
return answer
@@ -259,6 +262,22 @@ class OCICompatibleOpenAIProvider(LLMProvider):
for optional_key in ("top_p", "frequency_penalty", "presence_penalty"):
if effective.get(optional_key) is not None:
request_kwargs[optional_key] = effective[optional_key]
+ model_parameters = {
+ key: value
+ for key, value in request_kwargs.items()
+ if key not in {"model", "messages"} and value is not None
+ }
+ llm_metadata = {
+ "provider": provider,
+ "model": model,
+ "component": component_name,
+ "profile_name": resolved_profile_name,
+ "requested_profile_name": requested_profile_name,
+ "profile_source": profile_source,
+ "profile_found": profile_found,
+ "profiles_enabled": bool(effective.get("profiles_enabled")),
+ "profiles_path": effective.get("profiles_path"),
+ }
async with _maybe_span(
self.telemetry,
@@ -275,47 +294,35 @@ class OCICompatibleOpenAIProvider(LLMProvider):
profiles_enabled=bool(effective.get("profiles_enabled")),
):
try:
- resp = await client.chat.completions.create(**request_kwargs)
- answer = resp.choices[0].message.content or ""
+ async with _maybe_generation(
+ self.telemetry,
+ name=generation_name,
+ model=model,
+ input=messages,
+ metadata=llm_metadata,
+ model_parameters=model_parameters,
+ ) as generation:
+ resp = await client.chat.completions.create(**request_kwargs)
+ answer = resp.choices[0].message.content or ""
- usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
- usage_metadata.update({
- "profile_name": resolved_profile_name,
- "requested_profile_name": requested_profile_name,
- "profile_source": profile_source,
- "profile_found": profile_found,
- "component": component_name,
- "model": model,
- "provider": provider,
- "temperature": temperature,
- "max_tokens": max_tokens,
- })
- llm_metadata = {
- "provider": provider,
- "model": model,
- "component": component_name,
- "profile_name": resolved_profile_name,
- "requested_profile_name": requested_profile_name,
- "profile_source": profile_source,
- "profile_found": profile_found,
- "profiles_enabled": bool(effective.get("profiles_enabled")),
- "profiles_path": effective.get("profiles_path"),
- "temperature": temperature,
- "max_tokens": max_tokens,
- }
+ usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
+ usage_metadata.update({
+ "profile_name": resolved_profile_name,
+ "requested_profile_name": requested_profile_name,
+ "profile_source": profile_source,
+ "profile_found": profile_found,
+ "component": component_name,
+ "model": model,
+ "provider": provider,
+ **model_parameters,
+ })
+ generation.set_output(answer)
+ generation.set_usage(usage_metadata)
+ generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
UsageRecord.from_usage(provider, model, generation_name, usage_metadata, llm_metadata)
)
- if self.telemetry:
- await self.telemetry.generation(
- name=generation_name,
- model=model,
- input=messages,
- output=answer,
- metadata={**llm_metadata, **usage_metadata},
- usage=usage_metadata,
- )
return answer
except Exception as exc:
@@ -550,6 +557,19 @@ class OCISDKProvider(LLMProvider):
)
service_endpoint = self._resolve_endpoint(self.settings, endpoint)
+ model_parameters = {
+ "temperature": temperature,
+ "max_tokens": max_tokens,
+ }
+ llm_metadata = {
+ "provider": "oci_sdk",
+ "model": model,
+ "endpoint_id": endpoint_id,
+ "service_endpoint": service_endpoint,
+ "component": component_name,
+ "profile_name": profile_name,
+ "auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
+ }
async with _maybe_span(
self.telemetry,
@@ -575,26 +595,30 @@ class OCISDKProvider(LLMProvider):
max_tokens=max_tokens,
)
- response = await asyncio.to_thread(client.chat, details)
- answer = self._extract_answer(response)
+ async with _maybe_generation(
+ self.telemetry,
+ name=generation_name,
+ model=model,
+ input=messages,
+ metadata=llm_metadata,
+ model_parameters=model_parameters,
+ ) as generation:
+ response = await asyncio.to_thread(client.chat, details)
+ answer = self._extract_answer(response)
- usage_metadata = {
- "prompt_tokens": max(1, len(str(messages)) // 4),
- "completion_tokens": max(1, len(answer) // 4),
- "total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
- "cost_usd": 0.0,
- "cost_brl": 0.0,
- "estimated_usage": True,
- "provider": "oci_sdk",
- "model": model,
- "endpoint_id": endpoint_id,
- "service_endpoint": service_endpoint,
- "component": component_name,
- "profile_name": profile_name,
- "auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
- }
-
- llm_metadata = dict(usage_metadata)
+ usage_metadata = {
+ "prompt_tokens": max(1, len(str(messages)) // 4),
+ "completion_tokens": max(1, len(answer) // 4),
+ "total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
+ "cost_usd": 0.0,
+ "cost_brl": 0.0,
+ "estimated_usage": True,
+ **llm_metadata,
+ **model_parameters,
+ }
+ generation.set_output(answer)
+ generation.set_usage(usage_metadata)
+ generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
@@ -607,16 +631,6 @@ class OCISDKProvider(LLMProvider):
)
)
- if self.telemetry:
- await self.telemetry.generation(
- name=generation_name,
- model=model,
- input=messages,
- output=answer,
- metadata=llm_metadata,
- usage=usage_metadata,
- )
-
return answer
@@ -656,3 +670,36 @@ class _maybe_span:
if self.cm:
return await self.cm.__aexit__(exc_type, exc, tb)
return False
+
+
+class _NoopGeneration:
+ def set_output(self, output: Any) -> None:
+ pass
+
+ def set_usage(self, usage: dict[str, Any] | None) -> None:
+ pass
+
+ def set_metadata(self, **metadata: Any) -> None:
+ pass
+
+ def set_model_parameters(self, **model_parameters: Any) -> None:
+ pass
+
+
+class _maybe_generation:
+ def __init__(self, telemetry, **attrs: Any):
+ self.telemetry = telemetry
+ self.attrs = attrs
+ self.cm = None
+ self.noop = _NoopGeneration()
+
+ async def __aenter__(self):
+ if not self.telemetry or not hasattr(self.telemetry, "generation_span"):
+ return self.noop
+ self.cm = self.telemetry.generation_span(**self.attrs)
+ return await self.cm.__aenter__()
+
+ async def __aexit__(self, exc_type, exc, tb):
+ if self.cm:
+ return await self.cm.__aexit__(exc_type, exc, tb)
+ return False
diff --git a/libs/agent_framework/src/agent_framework/observability/telemetry.py b/libs/agent_framework/src/agent_framework/observability/telemetry.py
index efeb1b9..7d56239 100644
--- a/libs/agent_framework/src/agent_framework/observability/telemetry.py
+++ b/libs/agent_framework/src/agent_framework/observability/telemetry.py
@@ -15,6 +15,7 @@ import logging
import re
import time
from contextlib import asynccontextmanager
+from datetime import datetime, timezone
from typing import Any
from uuid import uuid4
@@ -36,6 +37,24 @@ from .otel import OpenTelemetryProvider
logger = logging.getLogger("agent_framework.telemetry")
_LANGFUSE_OBSERVATION_TYPES = {"span", "generation", "agent", "tool", "chain", "retriever", "embedding", "evaluator", "guardrail"}
+_LANGFUSE_START_OBSERVATION_KWARGS = {
+ "trace_context",
+ "name",
+ "as_type",
+ "input",
+ "output",
+ "metadata",
+ "version",
+ "level",
+ "status_message",
+ "completion_start_time",
+ "model",
+ "model_parameters",
+ "usage_details",
+ "cost_details",
+ "prompt",
+ "end_on_exit",
+}
def _langfuse_type(kind: str | None) -> str:
# Langfuse SDKs do not accept arbitrary event types such as "event"; FIRST pattern
@@ -58,6 +77,7 @@ _COMPACT_SUPPRESSED_SPAN_PREFIXES = (
"workflow.routing_decision",
"workflow.supervisor_review",
)
+_COMPACT_VISIBLE_EVENT_PREFIXES = ("AGA.", "NOC.")
def _raw_correlation_id(attrs: dict[str, Any] | None = None) -> str | None:
@@ -169,6 +189,128 @@ def _extract_observation_id(observation: Any) -> str | None:
pass
return None
+
+def _is_compact_visible_event(name: str) -> bool:
+ return str(name or "").startswith(_COMPACT_VISIBLE_EVENT_PREFIXES)
+
+
+class _SpanHandle:
+ """Mutable handle yielded by Telemetry.span for setting final output."""
+
+ def __init__(self, observation: Any | None = None) -> None:
+ self.observation = observation
+ self.output: Any = None
+ self.has_output = False
+ self.metadata: dict[str, Any] = {}
+
+ def set_observation(self, observation: Any | None) -> None:
+ self.observation = observation
+
+ def set_output(self, output: Any) -> None:
+ self.output = output
+ self.has_output = True
+
+ def set_metadata(self, **metadata: Any) -> None:
+ self.metadata.update({k: v for k, v in metadata.items() if v is not None})
+
+ def __getattr__(self, name: str) -> Any:
+ if self.observation is None:
+ raise AttributeError(name)
+ return getattr(self.observation, name)
+
+
+class _GenerationHandle:
+ """Mutable handle yielded by Telemetry.generation_span."""
+
+ def __init__(self, observation: Any | None = None) -> None:
+ self.observation = observation
+ self.output: Any = None
+ self.has_output = False
+ self.metadata: dict[str, Any] = {}
+ self.usage: dict[str, Any] | None = None
+ self.model_parameters: dict[str, Any] = {}
+
+ def set_observation(self, observation: Any | None) -> None:
+ self.observation = observation
+
+ def set_output(self, output: Any) -> None:
+ self.output = output
+ self.has_output = True
+
+ def set_usage(self, usage: dict[str, Any] | None) -> None:
+ self.usage = dict(usage or {})
+
+ def set_metadata(self, **metadata: Any) -> None:
+ self.metadata.update({k: v for k, v in metadata.items() if v is not None})
+
+ def set_model_parameters(self, **model_parameters: Any) -> None:
+ self.model_parameters.update({k: v for k, v in model_parameters.items() if v is not None})
+
+ def __getattr__(self, name: str) -> Any:
+ if self.observation is None:
+ raise AttributeError(name)
+ return getattr(self.observation, name)
+
+
+def _usage_details_from_usage(usage: dict[str, Any] | None) -> dict[str, int] | None:
+ if not isinstance(usage, dict):
+ return None
+
+ def int_value(*keys: str) -> int | None:
+ for key in keys:
+ value = usage.get(key)
+ if value is None:
+ continue
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ continue
+ return None
+
+ input_tokens = int_value("input", "input_tokens", "prompt_tokens")
+ output_tokens = int_value("output", "output_tokens", "completion_tokens")
+ total_tokens = int_value("total", "total_tokens")
+
+ # Langfuse self-hosted versions may sum all custom usage keys into totalUsage.
+ # Send split fields only when available; send total only when there is no split.
+ details: dict[str, int] = {}
+ if input_tokens is not None:
+ details["input"] = input_tokens
+ if output_tokens is not None:
+ details["output"] = output_tokens
+ if not details and total_tokens is not None:
+ details["total"] = total_tokens
+ return details or None
+
+
+def _cost_details_from_usage(usage: dict[str, Any] | None) -> dict[str, float] | None:
+ if not isinstance(usage, dict):
+ return None
+ details: dict[str, float] = {}
+ if usage.get("cost_usd") is not None:
+ try:
+ details["total"] = float(usage["cost_usd"])
+ except (TypeError, ValueError):
+ pass
+ if usage.get("cost_brl") is not None:
+ try:
+ details["total_brl"] = float(usage["cost_brl"])
+ except (TypeError, ValueError):
+ pass
+ return details or None
+
+
+def _clean_mapping(value: dict[str, Any] | None) -> dict[str, Any] | None:
+ if not isinstance(value, dict):
+ return None
+ clean = {k: v for k, v in value.items() if v is not None}
+ return clean or None
+
+
+def _utc_iso_ms() -> str:
+ return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
class Telemetry:
def __init__(self, settings):
self.settings = settings
@@ -228,7 +370,8 @@ class Telemetry:
start = time.time()
attrs = context_metadata(attrs)
attrs.setdefault("_span_name", name)
- if self.is_compact_mode() and name == "agent.gateway_message" and not attrs.get("parent_observation_id"):
+ is_root_span = bool(attrs.get("_root_span")) or name == "agent.gateway_message"
+ if self.is_compact_mode() and is_root_span and not attrs.get("parent_observation_id"):
attrs["_ignore_current_parent"] = True
if not attrs.get("request_id"):
attrs["request_id"] = str(uuid4())
@@ -237,7 +380,10 @@ class Telemetry:
set_observability_context(request_id=attrs.get("request_id"), trace_id=attrs.get("trace_id"))
observation_cm = None
observation = None
+ handle = _SpanHandle()
observation_token = None
+ propagation_cm = None
+ legacy_io_update: dict[str, Any] | None = None
ignore_current_parent = bool(attrs.get("_ignore_current_parent"))
parent_observation_id = attrs.get("parent_observation_id")
if not parent_observation_id and not ignore_current_parent:
@@ -263,36 +409,61 @@ class Telemetry:
try:
if observation_cm is not None:
observation = observation_cm.__enter__()
+ handle.set_observation(observation)
observation_id = _extract_observation_id(observation)
if observation_id:
observation_token = set_current_observation_id(observation_id)
attrs.setdefault("observation_id", observation_id)
- if not attrs.get("parent_observation_id"):
+ if is_root_span:
self._update_trace_from_attrs(observation, attrs)
+ self._set_trace_io(observation, input=attrs.get("input"))
+ propagation_cm = self._start_trace_attribute_propagation(name, attrs)
+ if propagation_cm is not None:
+ propagation_cm.__enter__()
# Publish span.started only after the Langfuse observation is current,
# so secondary analytics/exporters can attach it as a child instead
# of creating a sibling/root entry.
await self.event_bus.publish(f"{name}.started", attrs, kind="span")
- yield observation
+ yield handle
duration_ms = int((time.time() - start) * 1000)
- out = {"status": "ok", "duration_ms": duration_ms}
- metadata = {**observation_metadata, "duration_ms": duration_ms}
- if span_events:
+ status = {"status": "ok", "duration_ms": duration_ms}
+ out = handle.output if handle.has_output else status
+ metadata = {**observation_metadata, **status, **handle.metadata}
+ if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
- self._update_observation(observation, output=out, metadata=metadata)
+ self._update_observation(observation, input=attrs.get("input"), output=out, metadata=metadata)
+ if is_root_span:
+ self._set_trace_io(observation, input=attrs.get("input"), output=out)
+ legacy_io_update = {
+ "input": attrs.get("input"),
+ "output": out,
+ "metadata": metadata,
+ }
if otel_span is not None:
otel_span.set_attribute("duration_ms", duration_ms)
- await self.event_bus.publish(f"{name}.completed", {**attrs, **out}, kind="span")
+ completed_payload = {**attrs, **status}
+ if handle.has_output:
+ completed_payload["output"] = out
+ await self.event_bus.publish(f"{name}.completed", completed_payload, kind="span")
logger.info("span.end %s duration_ms=%s", name, duration_ms)
except Exception as exc:
duration_ms = int((time.time() - start) * 1000)
out = {"status": "error", "error": str(exc), "duration_ms": duration_ms}
metadata = {**observation_metadata, "duration_ms": duration_ms}
- if span_events:
+ if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
- self._update_observation(observation, level="ERROR", status_message=str(exc), output=out, metadata=metadata)
+ self._update_observation(observation, level="ERROR", status_message=str(exc), input=attrs.get("input"), output=out, metadata=metadata)
+ if is_root_span:
+ self._set_trace_io(observation, input=attrs.get("input"), output=out)
+ legacy_io_update = {
+ "input": attrs.get("input"),
+ "output": out,
+ "metadata": metadata,
+ "level": "ERROR",
+ "status_message": str(exc),
+ }
if otel_span is not None:
try:
otel_span.record_exception(exc)
@@ -303,9 +474,19 @@ class Telemetry:
logger.exception("span.error %s %s", name, exc)
raise
finally:
+ if propagation_cm is not None:
+ try: propagation_cm.__exit__(None, None, None)
+ except Exception: logger.debug("Falha ao encerrar propagação Langfuse", exc_info=True)
if observation_cm is not None:
try: observation_cm.__exit__(None, None, None)
except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name)
+ if legacy_io_update is not None:
+ self._legacy_observation_update(
+ observation,
+ observation_type="span",
+ name=name,
+ **legacy_io_update,
+ )
if observation_token is not None:
reset_current_observation_id(observation_token)
if span_events_token is not None:
@@ -324,6 +505,16 @@ class Telemetry:
"kind": kind,
"payload": payload,
})
+ if not _is_compact_visible_event(name) or not self.is_enabled():
+ return
+ try:
+ metadata = {**payload, "event_kind": kind}
+ cm = self._start_observation(name=name, as_type="span", input=payload, metadata=metadata)
+ if cm is not None:
+ with cm as obs:
+ self._update_observation(obs, input=payload, output={"status": "ok"}, metadata=metadata)
+ except Exception:
+ logger.exception("Falha ao enviar event compacto via observation")
return
if not self.is_enabled():
return
@@ -341,49 +532,170 @@ class Telemetry:
except Exception:
logger.exception("Falha ao enviar event via observation")
- async def generation(self, name: str, model: str, input: list | dict | str, output: str,
- metadata: dict[str, Any] | None = None, usage: dict[str, Any] | None = None):
+ @asynccontextmanager
+ async def generation_span(
+ self,
+ name: str,
+ model: str,
+ input: list | dict | str,
+ *,
+ metadata: dict[str, Any] | None = None,
+ usage: dict[str, Any] | None = None,
+ model_parameters: dict[str, Any] | None = None,
+ ):
metadata = context_metadata(metadata or {})
# Keep the actual LLM model visible both in Langfuse's generation.model field
# and in metadata for filtering/debugging across SDK versions.
metadata.setdefault("model", model)
metadata.setdefault("llm_model", model)
metadata.setdefault("component", metadata.get("profile_name") or name)
- if usage:
- metadata["usage"] = usage
- logger.info("generation %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
- await self.event_bus.publish(name, {"model": model, "llm_model": model, "output_chars": len(output or ""), **metadata}, kind="generation")
- if not self.is_enabled():
- return
+ clean_model_parameters = _clean_mapping(model_parameters)
+ if clean_model_parameters:
+ metadata.setdefault("model_parameters", clean_model_parameters)
+ handle = _GenerationHandle()
+ observation_cm = None
+ observation = None
+ observation_token = None
+ legacy_io_update: dict[str, Any] | None = None
+ logger.info("generation.start %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
try:
- kwargs = dict(name=name, as_type="generation", input=input, output=output, model=model, metadata=metadata)
- if usage:
- kwargs["usage"] = usage
- kwargs["usage_details"] = {k: usage.get(k) for k in ("prompt_tokens", "completion_tokens", "total_tokens", "cached_tokens", "reasoning_tokens") if k in usage}
-
- # Prefer current/correlated generation APIs. Avoid raw
- # ``langfuse.generation(...)`` first because it can create a separate
- # trace row per LLM call when no current observation exists.
- if hasattr(self.langfuse, "start_as_current_generation"):
- clean = {k: v for k, v in kwargs.items() if k != "as_type" and v is not None}
- if not self.is_compact_mode():
- clean = _inject_langfuse_trace_context(clean, metadata)
+ if self.is_enabled():
try:
- with self.langfuse.start_as_current_generation(**clean) as obs:
- self._update_observation(obs, output=output, model=model, metadata=metadata)
- return
- except TypeError:
- clean.pop("trace_context", None)
- with self.langfuse.start_as_current_generation(**clean) as obs:
- self._update_observation(obs, output=output, model=model, metadata=metadata)
- return
+ observation_cm = self._start_observation(
+ name=name,
+ as_type="generation",
+ input=input,
+ model=model,
+ model_parameters=clean_model_parameters,
+ usage_details=_usage_details_from_usage(usage),
+ cost_details=_cost_details_from_usage(usage),
+ metadata=metadata,
+ )
+ if observation_cm is not None:
+ observation = observation_cm.__enter__()
+ handle.set_observation(observation)
+ observation_id = _extract_observation_id(observation)
+ if observation_id:
+ observation_token = set_current_observation_id(observation_id)
+ except Exception:
+ observation_cm = None
+ observation = None
+ logger.exception("Falha ao iniciar generation Langfuse %s", name)
+ yield handle
+ final_usage = handle.usage if handle.usage is not None else usage
+ final_model_parameters = {
+ **(clean_model_parameters or {}),
+ **handle.model_parameters,
+ } or None
+ final_metadata = {**metadata, **handle.metadata}
+ if final_usage:
+ final_metadata["usage"] = final_usage
+ output = handle.output if handle.has_output else None
+ usage_details = _usage_details_from_usage(final_usage)
+ cost_details = _cost_details_from_usage(final_usage)
+ self._update_observation(
+ observation,
+ input=input,
+ output=output,
+ model=model,
+ metadata=final_metadata,
+ model_parameters=final_model_parameters,
+ usage_details=usage_details,
+ cost_details=cost_details,
+ )
+ legacy_io_update = {
+ "input": input,
+ "output": output,
+ "model": model,
+ "metadata": final_metadata,
+ "model_parameters": final_model_parameters,
+ "usage_details": usage_details,
+ "cost_details": cost_details,
+ }
+ await self.event_bus.publish(
+ name,
+ {
+ "model": model,
+ "llm_model": model,
+ "output_chars": len(output or "") if isinstance(output, str) else 0,
+ **final_metadata,
+ },
+ kind="generation",
+ )
+ logger.info("generation.end %s model=%s", name, model)
+ except Exception as exc:
+ final_usage = handle.usage if handle.usage is not None else usage
+ final_model_parameters = {
+ **(clean_model_parameters or {}),
+ **handle.model_parameters,
+ } or None
+ final_metadata = {**metadata, **handle.metadata}
+ if final_usage:
+ final_metadata["usage"] = final_usage
+ usage_details = _usage_details_from_usage(final_usage)
+ cost_details = _cost_details_from_usage(final_usage)
+ output = handle.output if handle.has_output else None
+ self._update_observation(
+ observation,
+ level="ERROR",
+ status_message=str(exc),
+ input=input,
+ output=output,
+ model=model,
+ metadata=final_metadata,
+ model_parameters=final_model_parameters,
+ usage_details=usage_details,
+ cost_details=cost_details,
+ )
+ legacy_io_update = {
+ "input": input,
+ "output": output,
+ "model": model,
+ "metadata": final_metadata,
+ "model_parameters": final_model_parameters,
+ "usage_details": usage_details,
+ "cost_details": cost_details,
+ "level": "ERROR",
+ "status_message": str(exc),
+ }
+ await self.event_bus.publish(f"{name}.failed", {"model": model, "llm_model": model, "error": str(exc), **final_metadata}, kind="generation")
+ logger.exception("generation.error %s model=%s exc=%s", name, model, exc)
+ raise
+ finally:
+ if observation_cm is not None:
+ try: observation_cm.__exit__(None, None, None)
+ except Exception: logger.exception("Falha ao finalizar generation Langfuse %s", name)
+ if legacy_io_update is not None:
+ self._legacy_observation_update(
+ observation,
+ observation_type="generation",
+ name=name,
+ **legacy_io_update,
+ )
+ if observation_token is not None:
+ reset_current_observation_id(observation_token)
- cm = self._start_observation(**kwargs)
- if cm is not None:
- with cm as obs:
- self._update_observation(obs, output=output, model=model, metadata=metadata)
- except Exception:
- logger.exception("Falha ao registrar generation no Langfuse")
+ async def generation(
+ self,
+ name: str,
+ model: str,
+ input: list | dict | str,
+ output: str,
+ metadata: dict[str, Any] | None = None,
+ usage: dict[str, Any] | None = None,
+ model_parameters: dict[str, Any] | None = None,
+ ):
+ async with self.generation_span(
+ name=name,
+ model=model,
+ input=input,
+ metadata=metadata,
+ usage=usage,
+ model_parameters=model_parameters,
+ ) as generation:
+ generation.set_output(output)
+ if usage:
+ generation.set_usage(usage)
async def rag_event(self, name: str, query: str, results_count: int, metadata: dict[str, Any] | None = None):
await self.event(f"rag.{name}", {"query": query, "results_count": results_count, **(metadata or {})}, kind="rag")
@@ -426,7 +738,7 @@ class Telemetry:
def _start_observation(self, **kwargs):
if not self.is_enabled(): return None
if hasattr(self.langfuse, "start_as_current_observation"):
- clean = {k: v for k, v in kwargs.items() if v is not None}
+ clean = {k: v for k, v in kwargs.items() if v is not None and k in _LANGFUSE_START_OBSERVATION_KWARGS}
if "as_type" in clean:
clean["as_type"] = _langfuse_type(clean.get("as_type"))
if self.is_compact_mode():
@@ -481,6 +793,61 @@ class Telemetry:
if hasattr(observation, "update"): observation.update(**clean)
except Exception: logger.debug("Observation update não suportado", exc_info=True)
+ def _legacy_observation_update(self, observation, *, observation_type: str, name: str, **kwargs):
+ """Compatibility fallback for Langfuse servers that drop OTEL observation I/O."""
+ if not self.is_enabled() or not bool(getattr(self.settings, "LANGFUSE_LEGACY_IO_FALLBACK", True)):
+ return
+ if observation is None:
+ return
+ obs_id = _extract_observation_id(observation)
+ trace_id = getattr(observation, "trace_id", None)
+ if not obs_id or not trace_id:
+ return
+ api = getattr(self.langfuse, "api", None)
+ ingestion = getattr(api, "ingestion", None)
+ if ingestion is None or not hasattr(ingestion, "batch"):
+ return
+
+ clean = {k: v for k, v in kwargs.items() if v is not None}
+ if not any(k in clean for k in ("input", "output", "metadata")):
+ return
+ try:
+ if hasattr(self.langfuse, "flush"):
+ self.langfuse.flush()
+
+ if observation_type == "generation":
+ from langfuse.api.ingestion.types import (
+ IngestionEvent_GenerationUpdate,
+ UpdateGenerationBody,
+ )
+
+ body = UpdateGenerationBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
+ event = IngestionEvent_GenerationUpdate(
+ id=str(uuid4()),
+ timestamp=_utc_iso_ms(),
+ body=body,
+ metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
+ )
+ else:
+ from langfuse.api.ingestion.types import IngestionEvent_SpanUpdate, UpdateSpanBody
+
+ body = UpdateSpanBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
+ event = IngestionEvent_SpanUpdate(
+ id=str(uuid4()),
+ timestamp=_utc_iso_ms(),
+ body=body,
+ metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
+ )
+
+ response = ingestion.batch(
+ batch=[event],
+ metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
+ )
+ if getattr(response, "errors", None):
+ logger.debug("Langfuse legacy I/O fallback retornou erros: %s", response.errors)
+ except Exception:
+ logger.debug("Falha no fallback legado de input/output Langfuse", exc_info=True)
+
def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]):
if observation is None: return
trace_attrs = {}
@@ -497,6 +864,43 @@ class Telemetry:
if hasattr(observation, "update_trace"): observation.update_trace(**trace_attrs)
except Exception: logger.debug("Trace update não suportado", exc_info=True)
+ def _set_trace_io(self, observation, *, input: Any | None = None, output: Any | None = None):
+ if observation is None: return
+ try:
+ if hasattr(observation, "set_trace_io"):
+ observation.set_trace_io(input=input, output=output)
+ return
+ if hasattr(observation, "update_trace"):
+ payload = {}
+ if input is not None:
+ payload["input"] = input
+ if output is not None:
+ payload["output"] = output
+ if payload:
+ observation.update_trace(**payload)
+ 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"):
+ 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
+ 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,
+ )
+ except Exception:
+ logger.debug("Trace attribute propagation não suportada", exc_info=True)
+ return None
+
class _LegacyObservationContext:
def __init__(self, observation): self.observation = observation
def __enter__(self): return self.observation
diff --git a/templates/agent_template_backend/.env b/templates/agent_template_backend/.env
index 1d80741..7745db8 100644
--- a/templates/agent_template_backend/.env
+++ b/templates/agent_template_backend/.env
@@ -70,14 +70,7 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
-LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
-# Nome customizado do trace pai, ex.: backoffice.checklist.workflow ou backoffice.emulador.workflow
-LANGFUSE_ROOT_TRACE_NAME=agent.gateway_message
-# No modo compact, manter visíveis apenas os eventos de negócio/operação prioritários
-LANGFUSE_COMPACT_SPAN_ALLOWLIST=AGA.,NOC.,agent.gateway_message,workflow.langgraph.ainvoke,workflow.input_guardrails,workflow.routing,workflow.agent.,workflow.output_guardrails,workflow.judge,mcp.tool_call,llm.
-# Evita traces/eventos de healthcheck e endpoints técnicos no Langfuse
-LANGFUSE_IGNORE_HEALTHCHECKS=true
-LANGFUSE_IGNORED_PATHS=/health,/ready,/metrics
+LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
diff --git a/templates/agent_template_backend/app/main.py b/templates/agent_template_backend/app/main.py
index d7c4ab1..06d1bd1 100644
--- a/templates/agent_template_backend/app/main.py
+++ b/templates/agent_template_backend/app/main.py
@@ -111,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
+def _metadata_value(payload: dict, key: str):
+ metadata = payload.get("metadata")
+ if isinstance(metadata, dict):
+ return metadata.get(key)
+ return None
+
+
+def _extract_workflow_id(payload: dict) -> str | None:
+ return (
+ payload.get("workflow_id")
+ or payload.get("workflowId")
+ or _metadata_value(payload, "workflow_id")
+ or _metadata_value(payload, "workflowId")
+ )
+
+
+def _format_root_span_name(template: str | None, values: dict) -> str:
+ template = template or "agent.gateway_message"
+ try:
+ return template.format(**{k: v or "unknown" for k, v in values.items()})
+ except Exception:
+ logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
+ return "agent.gateway_message"
+
+
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -146,9 +171,11 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
+ payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
- message_id = (req.payload or {}).get("message_id") or str(uuid4())
+ message_id = payload.get("message_id") or str(uuid4())
+ workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -156,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
- ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
+ workflow_id=workflow_id,
+ ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -212,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
- trace_input = {
+ cms_input = {
+ "channel": req.channel,
+ "tenant_id": req.tenant_id,
+ "agent_id": req.agent_id,
+ "payload": payload,
+ }
+ trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
+ root_span_name = _format_root_span_name(
+ getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
+ {
+ "workflow_id": workflow_id,
+ "channel": msg.channel,
+ "agent_id": identity.agent_id,
+ "tenant_id": identity.tenant_id,
+ },
+ )
+ root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
+ if workflow_id:
+ root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
- "agent.gateway_message",
+ root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
- input=trace_input,
- tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
- ):
- await telemetry.event("gateway.message.received", trace_input)
- await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
+ workflow_id=workflow_id,
+ input=cms_input,
+ tags=root_tags,
+ _root_span=True,
+ ) as root_span:
+ await telemetry.event("gateway.message.received", trace_context)
+ await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -249,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -259,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
- await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
- await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
+ await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
+ await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
- answer = result.get("final_answer") or result.get("answer") or ""
- await memory.append(
- agent_session_id,
- ChatMessage(
- role="assistant",
- content=answer,
- metadata={
+ answer = result.get("final_answer") or result.get("answer") or ""
+ await memory.append(
+ agent_session_id,
+ ChatMessage(
+ role="assistant",
+ content=answer,
+ metadata={
+ "tenant_id": identity.tenant_id,
+ "agent_id": identity.agent_id,
+ "message_id": f"assistant-{message_id}",
+ "route": result.get("route"),
+ "intent": result.get("intent"),
+ "route_decision": result.get("route_decision"),
+ "judges": result.get("judge_results"),
+ },
+ ),
+ )
+
+ await telemetry.event(
+ "gateway.message.responded",
+ {
+ "session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
- "message_id": f"assistant-{message_id}",
+ "route": result.get("route"),
+ "intent": result.get("intent"),
+ "answer_chars": len(answer),
+ },
+ )
+
+ response = ChannelResponse(
+ channel=msg.channel,
+ session_id=agent_session_id,
+ text=answer,
+ metadata={
+ "channel_id": msg.channel_id,
+ "tenant_id": identity.tenant_id,
+ "agent_id": identity.agent_id,
+ "original_session_id": msg.session_id,
+ "conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
+ "message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
+ "domain": result.get("domain"),
+ "mcp_tools": result.get("mcp_tools"),
+ "mcp_results": result.get("mcp_results"),
+ "business_context": business_context.model_dump(),
+ "identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
+ "guardrails": result.get("guardrail_decisions"),
},
- ),
- )
-
- await telemetry.event(
- "gateway.message.responded",
- {
- "session_id": agent_session_id,
- "tenant_id": identity.tenant_id,
- "agent_id": identity.agent_id,
- "route": result.get("route"),
- "intent": result.get("intent"),
- "answer_chars": len(answer),
- },
- )
-
- response = ChannelResponse(
- channel=msg.channel,
- session_id=agent_session_id,
- text=answer,
- metadata={
- "channel_id": msg.channel_id,
- "tenant_id": identity.tenant_id,
- "agent_id": identity.agent_id,
- "original_session_id": msg.session_id,
- "conversation_key": agent_session_id,
- "message_id": message_id,
- "route": result.get("route"),
- "intent": result.get("intent"),
- "route_decision": result.get("route_decision"),
- "domain": result.get("domain"),
- "mcp_tools": result.get("mcp_tools"),
- "mcp_results": result.get("mcp_results"),
- "business_context": business_context.model_dump(),
- "identity_missing": missing_identity_keys,
- "judges": result.get("judge_results"),
- "guardrails": result.get("guardrail_decisions"),
- },
- )
- rendered = await gateway.render(response)
- await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
- await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
- return rendered
+ )
+ rendered = await gateway.render(response)
+ root_span.set_output(rendered)
+ await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
+ await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
+ return rendered
@app.get("/health")
diff --git a/templates/agent_template_backend/app/state.py b/templates/agent_template_backend/app/state.py
index 3620e4e..db5369b 100644
--- a/templates/agent_template_backend/app/state.py
+++ b/templates/agent_template_backend/app/state.py
@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
+ workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str
diff --git a/templates/agent_template_backend/data/agent_framework.db b/templates/agent_template_backend/data/agent_framework.db
index 0cea93f..d1d18fd 100644
Binary files a/templates/agent_template_backend/data/agent_framework.db and b/templates/agent_template_backend/data/agent_framework.db differ
diff --git a/templates/agent_template_backend_day_zero/.env b/templates/agent_template_backend_day_zero/.env
index 1d80741..f62a8f5 100644
--- a/templates/agent_template_backend_day_zero/.env
+++ b/templates/agent_template_backend_day_zero/.env
@@ -70,14 +70,7 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
-LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
-# Nome customizado do trace pai, ex.: backoffice.checklist.workflow ou backoffice.emulador.workflow
-LANGFUSE_ROOT_TRACE_NAME=agent.gateway_message
-# No modo compact, manter visíveis apenas os eventos de negócio/operação prioritários
-LANGFUSE_COMPACT_SPAN_ALLOWLIST=AGA.,NOC.,agent.gateway_message,workflow.langgraph.ainvoke,workflow.input_guardrails,workflow.routing,workflow.agent.,workflow.output_guardrails,workflow.judge,mcp.tool_call,llm.
-# Evita traces/eventos de healthcheck e endpoints técnicos no Langfuse
-LANGFUSE_IGNORE_HEALTHCHECKS=true
-LANGFUSE_IGNORED_PATHS=/health,/ready,/metrics
+LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
@@ -129,9 +122,6 @@ PROMPT_POLICY_PATH=./config/prompt_policy.yaml
# Gateway de canais
###############################################################################
DEFAULT_CHANNEL=web
-# embedded = backend may parse simple/native channel payloads.
-# external = backend only accepts GatewayRequest normalized by an external Channel Gateway.
-FRAMEWORK_CHANNEL_INPUT_MODE=embedded
ENABLE_VOICE_ADAPTER=true
ENABLE_WHATSAPP_ADAPTER=true
ENABLE_TEXT_ADAPTER=true
diff --git a/templates/agent_template_backend_day_zero/app/main.py b/templates/agent_template_backend_day_zero/app/main.py
index d7c4ab1..06d1bd1 100644
--- a/templates/agent_template_backend_day_zero/app/main.py
+++ b/templates/agent_template_backend_day_zero/app/main.py
@@ -111,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
+def _metadata_value(payload: dict, key: str):
+ metadata = payload.get("metadata")
+ if isinstance(metadata, dict):
+ return metadata.get(key)
+ return None
+
+
+def _extract_workflow_id(payload: dict) -> str | None:
+ return (
+ payload.get("workflow_id")
+ or payload.get("workflowId")
+ or _metadata_value(payload, "workflow_id")
+ or _metadata_value(payload, "workflowId")
+ )
+
+
+def _format_root_span_name(template: str | None, values: dict) -> str:
+ template = template or "agent.gateway_message"
+ try:
+ return template.format(**{k: v or "unknown" for k, v in values.items()})
+ except Exception:
+ logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
+ return "agent.gateway_message"
+
+
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -146,9 +171,11 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
+ payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
- message_id = (req.payload or {}).get("message_id") or str(uuid4())
+ message_id = payload.get("message_id") or str(uuid4())
+ workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -156,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
- ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
+ workflow_id=workflow_id,
+ ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -212,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
- trace_input = {
+ cms_input = {
+ "channel": req.channel,
+ "tenant_id": req.tenant_id,
+ "agent_id": req.agent_id,
+ "payload": payload,
+ }
+ trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
+ root_span_name = _format_root_span_name(
+ getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
+ {
+ "workflow_id": workflow_id,
+ "channel": msg.channel,
+ "agent_id": identity.agent_id,
+ "tenant_id": identity.tenant_id,
+ },
+ )
+ root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
+ if workflow_id:
+ root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
- "agent.gateway_message",
+ root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
- input=trace_input,
- tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
- ):
- await telemetry.event("gateway.message.received", trace_input)
- await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
+ workflow_id=workflow_id,
+ input=cms_input,
+ tags=root_tags,
+ _root_span=True,
+ ) as root_span:
+ await telemetry.event("gateway.message.received", trace_context)
+ await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -249,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -259,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
- await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
- await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
+ await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
+ await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
- answer = result.get("final_answer") or result.get("answer") or ""
- await memory.append(
- agent_session_id,
- ChatMessage(
- role="assistant",
- content=answer,
- metadata={
+ answer = result.get("final_answer") or result.get("answer") or ""
+ await memory.append(
+ agent_session_id,
+ ChatMessage(
+ role="assistant",
+ content=answer,
+ metadata={
+ "tenant_id": identity.tenant_id,
+ "agent_id": identity.agent_id,
+ "message_id": f"assistant-{message_id}",
+ "route": result.get("route"),
+ "intent": result.get("intent"),
+ "route_decision": result.get("route_decision"),
+ "judges": result.get("judge_results"),
+ },
+ ),
+ )
+
+ await telemetry.event(
+ "gateway.message.responded",
+ {
+ "session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
- "message_id": f"assistant-{message_id}",
+ "route": result.get("route"),
+ "intent": result.get("intent"),
+ "answer_chars": len(answer),
+ },
+ )
+
+ response = ChannelResponse(
+ channel=msg.channel,
+ session_id=agent_session_id,
+ text=answer,
+ metadata={
+ "channel_id": msg.channel_id,
+ "tenant_id": identity.tenant_id,
+ "agent_id": identity.agent_id,
+ "original_session_id": msg.session_id,
+ "conversation_key": agent_session_id,
+ "workflow_id": workflow_id,
+ "message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
+ "domain": result.get("domain"),
+ "mcp_tools": result.get("mcp_tools"),
+ "mcp_results": result.get("mcp_results"),
+ "business_context": business_context.model_dump(),
+ "identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
+ "guardrails": result.get("guardrail_decisions"),
},
- ),
- )
-
- await telemetry.event(
- "gateway.message.responded",
- {
- "session_id": agent_session_id,
- "tenant_id": identity.tenant_id,
- "agent_id": identity.agent_id,
- "route": result.get("route"),
- "intent": result.get("intent"),
- "answer_chars": len(answer),
- },
- )
-
- response = ChannelResponse(
- channel=msg.channel,
- session_id=agent_session_id,
- text=answer,
- metadata={
- "channel_id": msg.channel_id,
- "tenant_id": identity.tenant_id,
- "agent_id": identity.agent_id,
- "original_session_id": msg.session_id,
- "conversation_key": agent_session_id,
- "message_id": message_id,
- "route": result.get("route"),
- "intent": result.get("intent"),
- "route_decision": result.get("route_decision"),
- "domain": result.get("domain"),
- "mcp_tools": result.get("mcp_tools"),
- "mcp_results": result.get("mcp_results"),
- "business_context": business_context.model_dump(),
- "identity_missing": missing_identity_keys,
- "judges": result.get("judge_results"),
- "guardrails": result.get("guardrail_decisions"),
- },
- )
- rendered = await gateway.render(response)
- await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
- await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
- return rendered
+ )
+ rendered = await gateway.render(response)
+ root_span.set_output(rendered)
+ await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
+ await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
+ return rendered
@app.get("/health")
diff --git a/templates/agent_template_backend_day_zero/app/state.py b/templates/agent_template_backend_day_zero/app/state.py
index 3620e4e..db5369b 100644
--- a/templates/agent_template_backend_day_zero/app/state.py
+++ b/templates/agent_template_backend_day_zero/app/state.py
@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
+ workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str
diff --git a/templates/agent_template_backend_day_zero/data/agent_framework.db b/templates/agent_template_backend_day_zero/data/agent_framework.db
index 0cea93f..d1d18fd 100644
Binary files a/templates/agent_template_backend_day_zero/data/agent_framework.db and b/templates/agent_template_backend_day_zero/data/agent_framework.db differ