Compare commits

...

2 Commits

Author SHA1 Message Date
5b0bf3f826 bugfix Alex 2026-06-24 2026-06-24 14:24:45 -03:00
dc618bfe47 bugfix Alex 2026-06-24 2026-06-24 14:15:44 -03:00
12 changed files with 345 additions and 105 deletions

View File

@@ -71,6 +71,7 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade # Observabilidade
############################################################################### ###############################################################################
ENABLE_LANGFUSE=true ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3 LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20 LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_HOST=http://localhost:3005 LANGFUSE_HOST=http://localhost:3005
@@ -85,7 +86,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo. # Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop # Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente. # Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC= AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH= GCP_PUBSUB_TOPIC_PATH=

30
.idea/workspace.xml generated
View File

@@ -4,12 +4,7 @@
<option name="autoReloadType" value="SELECTIVE" /> <option name="autoReloadType" value="SELECTIVE" />
</component> </component>
<component name="ChangeListManager"> <component name="ChangeListManager">
<list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="Ajustes na documentação e remanejamento dos folders"> <list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="bugfix Alex 2026-06-24" />
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/README.md" beforeDir="false" afterPath="$PROJECT_DIR$/README.md" afterDir="false" />
<change beforePath="$PROJECT_DIR$/README_en.md" beforeDir="false" afterPath="$PROJECT_DIR$/README_en.md" afterDir="false" />
<change beforePath="$PROJECT_DIR$/specs/SPEC-003-AI-Gateway.md" beforeDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" /> <option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" /> <option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" /> <option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
@@ -86,7 +81,7 @@
<workItem from="1782038604623" duration="6486000" /> <workItem from="1782038604623" duration="6486000" />
<workItem from="1782047166074" duration="27000" /> <workItem from="1782047166074" duration="27000" />
<workItem from="1782047194363" duration="961000" /> <workItem from="1782047194363" duration="961000" />
<workItem from="1782048494672" duration="230000" /> <workItem from="1782048494672" duration="2790000" />
</task> </task>
<task id="LOCAL-00001" summary="Ajustes na documentação e remanejamento dos folders"> <task id="LOCAL-00001" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" /> <option name="closed" value="true" />
@@ -112,7 +107,23 @@
<option name="project" value="LOCAL" /> <option name="project" value="LOCAL" />
<updated>1782048304579</updated> <updated>1782048304579</updated>
</task> </task>
<option name="localTasksCounter" value="4" /> <task id="LOCAL-00004" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
<created>1782048747421</created>
<option name="number" value="00004" />
<option name="presentableId" value="LOCAL-00004" />
<option name="project" value="LOCAL" />
<updated>1782048747421</updated>
</task>
<task id="LOCAL-00005" summary="bugfix Alex 2026-06-24">
<option name="closed" value="true" />
<created>1782321346355</created>
<option name="number" value="00005" />
<option name="presentableId" value="LOCAL-00005" />
<option name="project" value="LOCAL" />
<updated>1782321346355</updated>
</task>
<option name="localTasksCounter" value="6" />
<servers /> <servers />
</component> </component>
<component name="TypeScriptGeneratedFilesManager"> <component name="TypeScriptGeneratedFilesManager">
@@ -120,6 +131,7 @@
</component> </component>
<component name="VcsManagerConfiguration"> <component name="VcsManagerConfiguration">
<MESSAGE value="Ajustes na documentação e remanejamento dos folders" /> <MESSAGE value="Ajustes na documentação e remanejamento dos folders" />
<option name="LAST_COMMIT_MESSAGE" value="Ajustes na documentação e remanejamento dos folders" /> <MESSAGE value="bugfix Alex 2026-06-24" />
<option name="LAST_COMMIT_MESSAGE" value="bugfix Alex 2026-06-24" />
</component> </component>
</project> </project>

View File

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

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) _message_id: ContextVar[str | None] = ContextVar("message_id", default=None)
_trace_id: ContextVar[str | None] = ContextVar("trace_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_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) @dataclass(slots=True)
class ObservabilityContext: class ObservabilityContext:
@@ -66,6 +67,31 @@ def reset_current_observation_id(token) -> None:
_current_observation_id.set(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: def set_observability_context(**kwargs: Any) -> ObservabilityContext:
if not kwargs.get("request_id") and not _request_id.get(): if not kwargs.get("request_id") and not _request_id.get():
kwargs["request_id"] = str(uuid4()) kwargs["request_id"] = str(uuid4())
@@ -82,7 +108,7 @@ def set_observability_context(**kwargs: Any) -> ObservabilityContext:
def clear_observability_context() -> None: 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) var.set(None)

View File

@@ -3,6 +3,36 @@ import time
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Any 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: class LangGraphDeepTelemetry:
"""Eventos profundos do LangGraph no padrão FIRST. """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): async def node(self, name: str, state: dict[str, Any] | None = None):
state=state or {} state=state or {}
session_id=state.get('conversation_key') or state.get('session_id') 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() start=time.time()
await self.telemetry.event('langgraph.node.started', payload, kind='langgraph') await self.telemetry.event('langgraph.node.started', payload, kind='langgraph')
async with self.telemetry.span(f'langgraph.node.{name}', **payload): async with self.telemetry.span(f'langgraph.node.{name}', **payload):

View File

@@ -21,9 +21,13 @@ from uuid import uuid4
from .context import ( from .context import (
context_metadata, context_metadata,
get_current_observation_id, get_current_observation_id,
get_current_span_events,
get_observability_context, get_observability_context,
record_current_span_event,
reset_current_observation_id, reset_current_observation_id,
reset_current_span_events,
set_current_observation_id, set_current_observation_id,
set_current_span_events,
set_observability_context, set_observability_context,
) )
from .event_bus import TelemetryEventBus 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}$") _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: 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. which grouped everything in one trace while flattening the tree.
""" """
attrs = attrs or kwargs.get("metadata") or {} 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) raw_id = _raw_correlation_id(attrs)
trace_id = _langfuse_trace_id(raw_id) trace_id = _langfuse_trace_id(raw_id)
parent_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 attrs.get("parent_span_id")
or kwargs.get("parent_observation_id") or kwargs.get("parent_observation_id")
or kwargs.get("parent_span_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: if trace_id:
trace_context = dict(kwargs.get("trace_context") or {}) 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) metadata.setdefault("langfuse_trace_id", trace_id)
if parent_id: if parent_id:
metadata.setdefault("parent_observation_id", str(parent_id)) metadata.setdefault("parent_observation_id", str(parent_id))
metadata.pop("_ignore_current_parent", None)
kwargs.pop("_ignore_current_parent", None)
return kwargs return kwargs
@@ -188,6 +207,15 @@ class Telemetry:
def is_enabled(self) -> bool: def is_enabled(self) -> bool:
return bool(self.enabled and self.langfuse) 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): def bind_context(self, **kwargs: Any):
return set_observability_context(**kwargs) return set_observability_context(**kwargs)
@@ -199,6 +227,9 @@ class Telemetry:
"""Cria span correlacionado em logs, Langfuse e OpenTelemetry.""" """Cria span correlacionado em logs, Langfuse e OpenTelemetry."""
start = time.time() start = time.time()
attrs = context_metadata(attrs) 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"): if not attrs.get("request_id"):
attrs["request_id"] = str(uuid4()) attrs["request_id"] = str(uuid4())
if not attrs.get("trace_id"): if not attrs.get("trace_id"):
@@ -207,19 +238,27 @@ class Telemetry:
observation_cm = None observation_cm = None
observation = None observation = None
observation_token = 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: if parent_observation_id:
attrs.setdefault("parent_observation_id", str(parent_observation_id)) attrs.setdefault("parent_observation_id", str(parent_observation_id))
logger.info("span.start %s %s", name, _safe(attrs)) logger.info("span.start %s %s", name, _safe(attrs))
otel_cm = self.otel.span(name, attrs) otel_cm = self.otel.span(name, attrs)
otel_span = otel_cm.__enter__() 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( observation_cm = self._start_observation(
name=name, name=name,
as_type="span", as_type="span",
input=attrs.get("input"), 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: try:
if observation_cm is not None: if observation_cm is not None:
@@ -228,6 +267,7 @@ class Telemetry:
if observation_id: if observation_id:
observation_token = set_current_observation_id(observation_id) observation_token = set_current_observation_id(observation_id)
attrs.setdefault("observation_id", observation_id) attrs.setdefault("observation_id", observation_id)
if not attrs.get("parent_observation_id"):
self._update_trace_from_attrs(observation, attrs) self._update_trace_from_attrs(observation, attrs)
# Publish span.started only after the Langfuse observation is current, # Publish span.started only after the Langfuse observation is current,
# so secondary analytics/exporters can attach it as a child instead # so secondary analytics/exporters can attach it as a child instead
@@ -236,7 +276,11 @@ class Telemetry:
yield observation yield observation
duration_ms = int((time.time() - start) * 1000) duration_ms = int((time.time() - start) * 1000)
out = {"status": "ok", "duration_ms": duration_ms} 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: if otel_span is not None:
otel_span.set_attribute("duration_ms", duration_ms) otel_span.set_attribute("duration_ms", duration_ms)
await self.event_bus.publish(f"{name}.completed", {**attrs, **out}, kind="span") await self.event_bus.publish(f"{name}.completed", {**attrs, **out}, kind="span")
@@ -244,7 +288,11 @@ class Telemetry:
except Exception as exc: except Exception as exc:
duration_ms = int((time.time() - start) * 1000) duration_ms = int((time.time() - start) * 1000)
out = {"status": "error", "error": str(exc), "duration_ms": duration_ms} 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: if otel_span is not None:
try: try:
otel_span.record_exception(exc) otel_span.record_exception(exc)
@@ -260,6 +308,8 @@ class Telemetry:
except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name) except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name)
if observation_token is not None: if observation_token is not None:
reset_current_observation_id(observation_token) 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) try: otel_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True) except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True)
@@ -267,6 +317,14 @@ class Telemetry:
payload = context_metadata(payload or {}) payload = context_metadata(payload or {})
logger.info("event %s %s", name, _safe(payload)) logger.info("event %s %s", name, _safe(payload))
await self.event_bus.publish(name, payload, kind=kind) 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(): if not self.is_enabled():
return return
# IMPORTANT: do not call ``langfuse.event(...)`` directly here. In SDK # 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 # a new trace row for every telemetry event. We create a correlated
# observation instead, using request_id/trace_id as the stable trace id. # observation instead, using request_id/trace_id as the stable trace id.
try: 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: if cm is not None:
with cm: pass with cm: pass
except Exception: except Exception:
@@ -305,6 +366,7 @@ class Telemetry:
# trace row per LLM call when no current observation exists. # trace row per LLM call when no current observation exists.
if hasattr(self.langfuse, "start_as_current_generation"): 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 = {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) clean = _inject_langfuse_trace_context(clean, metadata)
try: try:
with self.langfuse.start_as_current_generation(**clean) as obs: with self.langfuse.start_as_current_generation(**clean) as obs:
@@ -367,7 +429,13 @@ class Telemetry:
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}
if "as_type" in clean: if "as_type" in clean:
clean["as_type"] = _langfuse_type(clean.get("as_type")) clean["as_type"] = _langfuse_type(clean.get("as_type"))
if self.is_compact_mode():
clean.pop("_ignore_current_parent", None)
else:
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {}) 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: try:
return self.langfuse.start_as_current_observation(**clean) return self.langfuse.start_as_current_observation(**clean)
except (TypeError, ValueError): except (TypeError, ValueError):
@@ -416,6 +484,8 @@ class Telemetry:
def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]): def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]):
if observation is None: return if observation is None: return
trace_attrs = {} trace_attrs = {}
if attrs.get("_span_name"):
trace_attrs["name"] = attrs["_span_name"]
for key in ("session_id", "user_id"): for key in ("session_id", "user_id"):
if attrs.get(key): trace_attrs[key] = attrs[key] if attrs.get(key): trace_attrs[key] = attrs[key]
if attrs.get("input"): trace_attrs["input"] = attrs["input"] if attrs.get("input"): trace_attrs["input"] = attrs["input"]

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6 OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID= OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles # OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência # Persistência
############################################################################### ###############################################################################
# Opções: memory, autonomous, mongodb # Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=autonomous MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=autonomous CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database # Autonomous Database
ADB_USER=admin ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
############################################################################### ###############################################################################
# RAG / Vector / Graph # RAG / Vector / Graph
############################################################################### ###############################################################################
VECTOR_STORE_PROVIDER=memory VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=memory GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5 RAG_TOP_K=5
EMBEDDING_PROVIDER=mock EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0 OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade # Observabilidade
############################################################################### ###############################################################################
ENABLE_LANGFUSE=true ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3 LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20 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 LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT= OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo. # Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop # Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente. # Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC= AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH= GCP_PUBSUB_TOPIC_PATH=
@@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router ROUTING_MODE=router
# Usage/cost accounting # Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -74,6 +74,7 @@ logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http") @app.middleware("http")
async def observability_context_middleware(request: Request, call_next): async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4()) request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context( set_observability_context(
request_id=request_id, request_id=request_id,
@@ -99,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000), "duration_ms": int((time.time() - started) * 1000),
}, kind="http") }, kind="http")
raise raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel): class GatewayRequest(BaseModel):

View File

@@ -320,7 +320,6 @@ class AgentWorkflow:
} }
async def billing_agent(self, state): async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.billing", "workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -329,7 +328,6 @@ class AgentWorkflow:
return await self.billing.run(state) return await self.billing.run(state)
async def product_agent(self, state): async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.product", "workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -338,7 +336,6 @@ class AgentWorkflow:
return await self.product.run(state) return await self.product.run(state)
async def orders_agent(self, state): async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.orders", "workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -346,9 +343,7 @@ class AgentWorkflow:
): ):
return await self.orders.run(state) return await self.orders.run(state)
async def support_agent(self, state): async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.support", "workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6 OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID= OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles # OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência # Persistência
############################################################################### ###############################################################################
# Opções: memory, autonomous, mongodb # Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=autonomous MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=autonomous CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database # Autonomous Database
ADB_USER=admin ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
############################################################################### ###############################################################################
# RAG / Vector / Graph # RAG / Vector / Graph
############################################################################### ###############################################################################
VECTOR_STORE_PROVIDER=memory VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=memory GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5 RAG_TOP_K=5
EMBEDDING_PROVIDER=mock EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0 OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade # Observabilidade
############################################################################### ###############################################################################
ENABLE_LANGFUSE=true ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3 LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20 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 LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT= OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo. # Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop # Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente. # Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC= AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH= GCP_PUBSUB_TOPIC_PATH=
@@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router ROUTING_MODE=router
# Usage/cost accounting # Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -4,7 +4,7 @@ import logging
from uuid import uuid4 from uuid import uuid4
import time import time
from fastapi import FastAPI, Request from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
from pydantic import BaseModel from pydantic import BaseModel
@@ -52,7 +52,7 @@ summary_memory = create_conversation_summary_memory(settings, message_history=me
sessions = create_session_repository(settings) sessions = create_session_repository(settings)
checkpoints = create_checkpoint_repository(settings) checkpoints = create_checkpoint_repository(settings)
cache = create_cache(settings, telemetry=telemetry) cache = create_cache(settings, telemetry=telemetry)
gateway = ChannelGateway() gateway = ChannelGateway(input_mode=settings.FRAMEWORK_CHANNEL_INPUT_MODE)
analytics = create_analytics_publisher(settings) analytics = create_analytics_publisher(settings)
observer = TelemetryBackedAgentObserver(telemetry=telemetry) observer = TelemetryBackedAgentObserver(telemetry=telemetry)
configure_global_observer({ configure_global_observer({
@@ -70,9 +70,11 @@ logger.info("LLM provider carregado: %s", llm.__class__.__name__)
logger.info("Langfuse habilitado: %s host=%s", telemetry.is_enabled(), settings.LANGFUSE_HOST) logger.info("Langfuse habilitado: %s host=%s", telemetry.is_enabled(), settings.LANGFUSE_HOST)
logger.info("Analytics habilitado: %s providers=%s", getattr(settings, "ENABLE_ANALYTICS", False), getattr(settings, "ANALYTICS_PROVIDERS", "")) logger.info("Analytics habilitado: %s providers=%s", getattr(settings, "ENABLE_ANALYTICS", False), getattr(settings, "ANALYTICS_PROVIDERS", ""))
logger.info("Agentes disponíveis: %s", [p.agent_id for p in agent_profiles.list_profiles()]) logger.info("Agentes disponíveis: %s", [p.agent_id for p in agent_profiles.list_profiles()])
logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http") @app.middleware("http")
async def observability_context_middleware(request: Request, call_next): async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4()) request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context( set_observability_context(
request_id=request_id, request_id=request_id,
@@ -98,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000), "duration_ms": int((time.time() - started) * 1000),
}, kind="http") }, kind="http")
raise raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel): class GatewayRequest(BaseModel):
@@ -138,7 +142,10 @@ def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, Bu
async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) -> dict: async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) -> dict:
try:
msg = await gateway.normalize(req.channel, req.payload) msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg) identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key() agent_session_id = identity.conversation_key()
message_id = (req.payload or {}).get("message_id") or str(uuid4()) message_id = (req.payload or {}).get("message_id") or str(uuid4())
@@ -331,6 +338,8 @@ async def health():
"usage_repository": settings.USAGE_REPOSITORY_PROVIDER, "usage_repository": settings.USAGE_REPOSITORY_PROVIDER,
"identity_config_path": settings.IDENTITY_CONFIG_PATH, "identity_config_path": settings.IDENTITY_CONFIG_PATH,
"mcp_parameter_mapping_path": settings.MCP_PARAMETER_MAPPING_PATH, "mcp_parameter_mapping_path": settings.MCP_PARAMETER_MAPPING_PATH,
"framework_channel_input_mode": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"legacy_channel_gateway_mode": settings.CHANNEL_GATEWAY_MODE,
} }
@@ -354,6 +363,8 @@ async def debug_env():
"AGENTS_CONFIG_PATH": settings.AGENTS_CONFIG_PATH, "AGENTS_CONFIG_PATH": settings.AGENTS_CONFIG_PATH,
"ROUTING_CONFIG_PATH": settings.ROUTING_CONFIG_PATH, "ROUTING_CONFIG_PATH": settings.ROUTING_CONFIG_PATH,
"ROUTING_MODE": settings.ROUTING_MODE, "ROUTING_MODE": settings.ROUTING_MODE,
"FRAMEWORK_CHANNEL_INPUT_MODE": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"CHANNEL_GATEWAY_MODE": settings.CHANNEL_GATEWAY_MODE,
} }

View File

@@ -187,6 +187,16 @@ class AgentWorkflow:
input=state.get("user_text"), input=state.get("user_text"),
): ):
history_texts = [m.get("content", "") for m in state.get("history", [])] history_texts = [m.get("content", "") for m in state.get("history", [])]
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
},
component="workflow.input_guardrails.start",
)
sanitized, decisions = await self.guardrails.run_input( sanitized, decisions = await self.guardrails.run_input(
state["user_text"], state["user_text"],
{ {
@@ -199,6 +209,19 @@ class AgentWorkflow:
) )
for _decision in decisions: for _decision in decisions:
await self.guardrail_telemetry.evaluated("input", _decision) await self.guardrail_telemetry.evaluated("input", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.input_guardrails.decision",
)
if not _decision.allowed: if not _decision.allowed:
await self.guardrail_telemetry.blocked("input", _decision) await self.guardrail_telemetry.blocked("input", _decision)
await self.telemetry.event( await self.telemetry.event(
@@ -210,6 +233,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions], "decisions": [d.model_dump() for d in decisions],
}, },
) )
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.input_guardrails.final",
)
if any(not d.allowed for d in decisions): if any(not d.allowed for d in decisions):
return { return {
"sanitized_input": sanitized, "sanitized_input": sanitized,
@@ -285,7 +320,6 @@ class AgentWorkflow:
} }
async def billing_agent(self, state): async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.billing", "workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -294,7 +328,6 @@ class AgentWorkflow:
return await self.billing.run(state) return await self.billing.run(state)
async def product_agent(self, state): async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.product", "workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -303,7 +336,6 @@ class AgentWorkflow:
return await self.product.run(state) return await self.product.run(state)
async def orders_agent(self, state): async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.orders", "workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -311,9 +343,7 @@ class AgentWorkflow:
): ):
return await self.orders.run(state) return await self.orders.run(state)
async def support_agent(self, state): async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span( async with self.telemetry.span(
"workflow.agent.support", "workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
@@ -419,6 +449,21 @@ class AgentWorkflow:
}, },
) )
await self.observer.emit_ic(
"IC.OUTPUT_SUPERVISOR_COMPLETED",
{
"session_id": context["session_id"],
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"route": state.get("route"),
"intent": state.get("intent"),
"action": action,
"approved": decision.approved,
"result_count": len(decision.results),
},
component="workflow.output_supervisor",
)
if decision.action in {RailAction.ALLOW, RailAction.SANITIZE, RailAction.OBSERVE}: if decision.action in {RailAction.ALLOW, RailAction.SANITIZE, RailAction.OBSERVE}:
final_answer = decision.candidate final_answer = decision.candidate
elif decision.action == RailAction.HANDOVER: elif decision.action == RailAction.HANDOVER:
@@ -457,11 +502,36 @@ class AgentWorkflow:
session_id=state.get("conversation_key") or state.get("session_id"), session_id=state.get("conversation_key") or state.get("session_id"),
input=state.get("answer"), input=state.get("answer"),
): ):
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"route": state.get("route"),
"intent": state.get("intent"),
},
component="workflow.output_guardrails.start",
)
final, decisions = await self.guardrails.run_output( final, decisions = await self.guardrails.run_output(
state["answer"], state.get("context", {}) state["answer"], state.get("context", {})
) )
for _decision in decisions: for _decision in decisions:
await self.guardrail_telemetry.evaluated("output", _decision) await self.guardrail_telemetry.evaluated("output", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.output_guardrails.decision",
)
if not _decision.allowed: if not _decision.allowed:
await self.guardrail_telemetry.blocked("output", _decision) await self.guardrail_telemetry.blocked("output", _decision)
await self.telemetry.event( await self.telemetry.event(
@@ -473,6 +543,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions], "decisions": [d.model_dump() for d in decisions],
}, },
) )
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.output_guardrails.final",
)
return { return {
"final_answer": final, "final_answer": final,
"guardrail_decisions": state.get("guardrail_decisions", []) "guardrail_decisions": state.get("guardrail_decisions", [])