cristiano.hoshikawa/agent_platform_oci
1
2
Fork 0
mirror of https://github.com/hoshikawa2/agent_platform_oci.git synced 2026-07-09 14:04:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

bugfix Alex 2026-06-24

Browse Source
This commit is contained in:
Cristiano Hoshikawa
2026-06-24 14:15:44 -03:00
parent 853c286313
commit dc618bfe47
11 changed files with 337 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
libs/agent_framework/src/agent_framework/config/settings.py
View File

@@ -99,6 +99,7 @@ class Settings(BaseSettings):
OCI_EMBEDDING_MODEL: str = 'cohere.embed-multilingual-v3.0'
ENABLE_LANGFUSE: bool = False
LANGFUSE_TRACE_MODE: Literal['verbose','compact'] = 'verbose'
LANGFUSE_PUBLIC_KEY: str | None = None
LANGFUSE_SECRET_KEY: str | None = None
LANGFUSE_HOST: str = 'https://cloud.langfuse.com'

28
libs/agent_framework/src/agent_framework/observability/context.py
View File

@@ -21,6 +21,7 @@ _workflow_id: ContextVar[str | None] = ContextVar("workflow_id", default=None)
_message_id: ContextVar[str | None] = ContextVar("message_id", default=None)
_trace_id: ContextVar[str | None] = ContextVar("trace_id", default=None)
_current_observation_id: ContextVar[str | None] = ContextVar("current_observation_id", default=None)
_current_span_events: ContextVar[list[dict[str, Any]] | None] = ContextVar("current_span_events", default=None)
@dataclass(slots=True)
class ObservabilityContext:
@@ -66,6 +67,31 @@ def reset_current_observation_id(token) -> None:
_current_observation_id.set(None)
def get_current_span_events() -> list[dict[str, Any]] | None:
"""Return the mutable aggregate-event bucket for the active macro span."""
return _current_span_events.get()
def set_current_span_events(events: list[dict[str, Any]] | None):
"""Set aggregate-event bucket and return ContextVar token."""
return _current_span_events.set(events)
def reset_current_span_events(token) -> None:
"""Restore previous aggregate-event bucket."""
try:
_current_span_events.reset(token)
except Exception:
_current_span_events.set(None)
def record_current_span_event(event: dict[str, Any]) -> None:
"""Append an event summary to the active macro span, if one exists."""
events = _current_span_events.get()
if events is not None:
events.append(event)
def set_observability_context(**kwargs: Any) -> ObservabilityContext:
if not kwargs.get("request_id") and not _request_id.get():
kwargs["request_id"] = str(uuid4())
@@ -82,7 +108,7 @@ def set_observability_context(**kwargs: Any) -> ObservabilityContext:
def clear_observability_context() -> None:
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id):
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id, _current_span_events):
var.set(None)

41
libs/agent_framework/src/agent_framework/observability/langgraph_telemetry.py
View File

@@ -3,6 +3,36 @@ import time
from contextlib import asynccontextmanager
from typing import Any
_LANGGRAPH_STEP_ORDER = {
"__start__": 0,
"input_guardrails": 1,
"routing_decision": 2,
"billing_agent": 3,
"product_agent": 3,
"orders_agent": 3,
"support_agent": 3,
"handoff": 3,
"supervisor_agent": 3,
"output_supervisor": 4,
"output_guardrails": 5,
"judge": 6,
"supervisor_review": 7,
"persist": 8,
"__end__": 9,
}
def _langgraph_step(name: str, state: dict[str, Any]) -> int:
explicit_steps = state.get("langgraph_steps")
if isinstance(explicit_steps, dict) and name in explicit_steps:
try:
return int(explicit_steps[name])
except (TypeError, ValueError):
pass
return _LANGGRAPH_STEP_ORDER.get(name, 50)
class LangGraphDeepTelemetry:
"""Eventos profundos do LangGraph no padrão FIRST.
@@ -16,7 +46,16 @@ class LangGraphDeepTelemetry:
async def node(self, name: str, state: dict[str, Any] | None = None):
state=state or {}
session_id=state.get('conversation_key') or state.get('session_id')
payload={'node': name, 'session_id': session_id, 'agent_id': state.get('agent_id'), 'tenant_id': state.get('tenant_id'), 'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or ''))}
payload={
'node': name,
'langgraph_node': name,
'langgraph_step': _langgraph_step(name, state),
'framework': 'langgraph',
'session_id': session_id,
'agent_id': state.get('agent_id'),
'tenant_id': state.get('tenant_id'),
'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or '')),
}
start=time.time()
await self.telemetry.event('langgraph.node.started', payload, kind='langgraph')
async with self.telemetry.span(f'langgraph.node.{name}', **payload):

90
libs/agent_framework/src/agent_framework/observability/telemetry.py
View File

@@ -21,9 +21,13 @@ from uuid import uuid4
from .context import (
context_metadata,
get_current_observation_id,
get_current_span_events,
get_observability_context,
record_current_span_event,
reset_current_observation_id,
reset_current_span_events,
set_current_observation_id,
set_current_span_events,
set_observability_context,
)
from .event_bus import TelemetryEventBus
@@ -42,6 +46,18 @@ def _langfuse_type(kind: str | None) -> str:
_LANGFUSE_TRACE_ID_RE = re.compile(r"^[0-9a-f]{32}$")
_COMPACT_SUPPRESSED_SPAN_PREFIXES = (
"llm.chat_completion",
"workflow.agent.",
"workflow.handoff",
"workflow.input_guardrails",
"workflow.judge",
"workflow.output_guardrails",
"workflow.output_supervisor",
"workflow.persist",
"workflow.routing_decision",
"workflow.supervisor_review",
)
def _raw_correlation_id(attrs: dict[str, Any] | None = None) -> str | None:
@@ -100,6 +116,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
which grouped everything in one trace while flattening the tree.
"""
attrs = attrs or kwargs.get("metadata") or {}
ignore_current_parent = bool(attrs.get("_ignore_current_parent") or kwargs.get("_ignore_current_parent"))
raw_id = _raw_correlation_id(attrs)
trace_id = _langfuse_trace_id(raw_id)
parent_id = (
@@ -107,7 +124,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
or attrs.get("parent_span_id")
or kwargs.get("parent_observation_id")
or kwargs.get("parent_span_id")
or get_current_observation_id()
or (None if ignore_current_parent else get_current_observation_id())
)
if trace_id:
trace_context = dict(kwargs.get("trace_context") or {})
@@ -121,6 +138,8 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
metadata.setdefault("langfuse_trace_id", trace_id)
if parent_id:
metadata.setdefault("parent_observation_id", str(parent_id))
metadata.pop("_ignore_current_parent", None)
kwargs.pop("_ignore_current_parent", None)
return kwargs
@@ -188,6 +207,15 @@ class Telemetry:
def is_enabled(self) -> bool:
return bool(self.enabled and self.langfuse)
def is_compact_mode(self) -> bool:
mode = getattr(self.settings, "LANGFUSE_TRACE_MODE", "verbose") or "verbose"
return str(mode).lower() == "compact"
def _should_emit_langfuse_span(self, name: str) -> bool:
if not self.is_compact_mode():
return True
return not str(name).startswith(_COMPACT_SUPPRESSED_SPAN_PREFIXES)
def bind_context(self, **kwargs: Any):
return set_observability_context(**kwargs)
@@ -199,6 +227,9 @@ class Telemetry:
"""Cria span correlacionado em logs, Langfuse e OpenTelemetry."""
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"):
attrs["_ignore_current_parent"] = True
if not attrs.get("request_id"):
attrs["request_id"] = str(uuid4())
if not attrs.get("trace_id"):
@@ -207,19 +238,27 @@ class Telemetry:
observation_cm = None
observation = None
observation_token = None
parent_observation_id = attrs.get("parent_observation_id") or get_current_observation_id()
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:
parent_observation_id = get_current_observation_id()
if parent_observation_id:
attrs.setdefault("parent_observation_id", str(parent_observation_id))
logger.info("span.start %s %s", name, _safe(attrs))
otel_cm = self.otel.span(name, attrs)
otel_span = otel_cm.__enter__()
if self.is_enabled():
emit_langfuse_span = self.is_enabled() and self._should_emit_langfuse_span(name)
span_events: list[dict[str, Any]] | None = [] if emit_langfuse_span and self.is_compact_mode() else None
span_events_token = set_current_span_events(span_events) if span_events is not None else None
observation_metadata = {k: v for k, v in attrs.items() if k != "input" and not str(k).startswith("_")}
if emit_langfuse_span:
observation_cm = self._start_observation(
name=name,
as_type="span",
input=attrs.get("input"),
metadata={k: v for k, v in attrs.items() if k != "input"},
metadata=observation_metadata,
_ignore_current_parent=attrs.get("_ignore_current_parent"),
)
try:
if observation_cm is not None:
@@ -228,7 +267,8 @@ class Telemetry:
if observation_id:
observation_token = set_current_observation_id(observation_id)
attrs.setdefault("observation_id", observation_id)
self._update_trace_from_attrs(observation, attrs)
if not attrs.get("parent_observation_id"):
self._update_trace_from_attrs(observation, attrs)
# 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.
@@ -236,7 +276,11 @@ class Telemetry:
yield observation
duration_ms = int((time.time() - start) * 1000)
out = {"status": "ok", "duration_ms": duration_ms}
self._update_observation(observation, output=out, metadata={"duration_ms": duration_ms})
metadata = {**observation_metadata, "duration_ms": duration_ms}
if span_events:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
self._update_observation(observation, 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")
@@ -244,7 +288,11 @@ class Telemetry:
except Exception as exc:
duration_ms = int((time.time() - start) * 1000)
out = {"status": "error", "error": str(exc), "duration_ms": duration_ms}
self._update_observation(observation, level="ERROR", status_message=str(exc), output=out, metadata={"duration_ms": duration_ms})
metadata = {**observation_metadata, "duration_ms": duration_ms}
if span_events:
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)
if otel_span is not None:
try:
otel_span.record_exception(exc)
@@ -260,6 +308,8 @@ class Telemetry:
except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name)
if observation_token is not None:
reset_current_observation_id(observation_token)
if span_events_token is not None:
reset_current_span_events(span_events_token)
try: otel_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True)
@@ -267,6 +317,14 @@ class Telemetry:
payload = context_metadata(payload or {})
logger.info("event %s %s", name, _safe(payload))
await self.event_bus.publish(name, payload, kind=kind)
if self.is_compact_mode():
if get_current_span_events() is not None:
record_current_span_event({
"name": name,
"kind": kind,
"payload": payload,
})
return
if not self.is_enabled():
return
# IMPORTANT: do not call ``langfuse.event(...)`` directly here. In SDK
@@ -274,7 +332,10 @@ class Telemetry:
# a new trace row for every telemetry event. We create a correlated
# observation instead, using request_id/trace_id as the stable trace id.
try:
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata={**payload, "event_kind": kind})
metadata = {**payload, "event_kind": kind}
if self.is_compact_mode():
metadata["_ignore_current_parent"] = True
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata=metadata)
if cm is not None:
with cm: pass
except Exception:
@@ -305,7 +366,8 @@ class Telemetry:
# 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}
clean = _inject_langfuse_trace_context(clean, metadata)
if not self.is_compact_mode():
clean = _inject_langfuse_trace_context(clean, metadata)
try:
with self.langfuse.start_as_current_generation(**clean) as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
@@ -367,7 +429,13 @@ class Telemetry:
clean = {k: v for k, v in kwargs.items() if v is not None}
if "as_type" in clean:
clean["as_type"] = _langfuse_type(clean.get("as_type"))
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
if self.is_compact_mode():
clean.pop("_ignore_current_parent", None)
else:
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
metadata = clean.get("metadata")
if isinstance(metadata, dict):
clean["metadata"] = {k: v for k, v in metadata.items() if not str(k).startswith("_")}
try:
return self.langfuse.start_as_current_observation(**clean)
except (TypeError, ValueError):
@@ -416,6 +484,8 @@ class Telemetry:
def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]):
if observation is None: return
trace_attrs = {}
if attrs.get("_span_name"):
trace_attrs["name"] = attrs["_span_name"]
for key in ("session_id", "user_id"):
if attrs.get(key): trace_attrs[key] = attrs[key]
if attrs.get("input"): trace_attrs["input"] = attrs["input"]