diff --git a/.idea/workspace.xml b/.idea/workspace.xml index 8f0894d..f4cc08f 100644 --- a/.idea/workspace.xml +++ b/.idea/workspace.xml @@ -6,9 +6,11 @@ - - - + + + + + diff --git a/libs/agent_framework/src/agent_framework/config/settings.py b/libs/agent_framework/src/agent_framework/config/settings.py index 48524fc..599d39f 100644 --- a/libs/agent_framework/src/agent_framework/config/settings.py +++ b/libs/agent_framework/src/agent_framework/config/settings.py @@ -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' diff --git a/libs/agent_framework/src/agent_framework/observability/context.py b/libs/agent_framework/src/agent_framework/observability/context.py index 22d9879..28c0ad3 100644 --- a/libs/agent_framework/src/agent_framework/observability/context.py +++ b/libs/agent_framework/src/agent_framework/observability/context.py @@ -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) diff --git a/libs/agent_framework/src/agent_framework/observability/langgraph_telemetry.py b/libs/agent_framework/src/agent_framework/observability/langgraph_telemetry.py index 57e4b7c..c2e5f5c 100644 --- a/libs/agent_framework/src/agent_framework/observability/langgraph_telemetry.py +++ b/libs/agent_framework/src/agent_framework/observability/langgraph_telemetry.py @@ -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): diff --git a/libs/agent_framework/src/agent_framework/observability/telemetry.py b/libs/agent_framework/src/agent_framework/observability/telemetry.py index 6ed9d8f..efeb1b9 100644 --- a/libs/agent_framework/src/agent_framework/observability/telemetry.py +++ b/libs/agent_framework/src/agent_framework/observability/telemetry.py @@ -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"] diff --git a/templates/agent_template_backend/.env b/templates/agent_template_backend/.env index 9e908e2..6b58c39 100644 --- a/templates/agent_template_backend/.env +++ b/templates/agent_template_backend/.env @@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1 OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6 OCI_GENAI_PROJECT_OCID= -# OCI_AUTH_MODE=config_file|instance_principal|resource_principal -OCI_AUTH_MODE=config_file # OCI SDK / signer / profiles OCI_CONFIG_FILE=~/.oci/config OCI_PROFILE=DEFAULT @@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1 # Persistência ############################################################################### # Opções: memory, autonomous, mongodb -SESSION_REPOSITORY_PROVIDER=autonomous -MEMORY_REPOSITORY_PROVIDER=autonomous -CHECKPOINT_REPOSITORY_PROVIDER=autonomous +SESSION_REPOSITORY_PROVIDER=sqlite +MEMORY_REPOSITORY_PROVIDER=sqlite +CHECKPOINT_REPOSITORY_PROVIDER=sqlite +SQLITE_DB_PATH=./data/agent_framework.db # Autonomous Database ADB_USER=admin @@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false ############################################################################### # RAG / Vector / Graph ############################################################################### -VECTOR_STORE_PROVIDER=memory -GRAPH_STORE_PROVIDER=memory +VECTOR_STORE_PROVIDER=sqlite +GRAPH_STORE_PROVIDER=sqlite RAG_TOP_K=5 EMBEDDING_PROVIDER=mock OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0 @@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json # Observabilidade ############################################################################### ENABLE_LANGFUSE=true -LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3 -LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20 +LANGFUSE_TRACE_MODE=verbose # 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 ENABLE_OTEL=false 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. ENABLE_ANALYTICS=false # Providers aceitos: oci_streaming,pubsub,noop -ANALYTICS_PROVIDERS=oci_streaming +ANALYTICS_PROVIDERS=pubsub # Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente. AGENT_PUBSUB_TOPIC= GCP_PUBSUB_TOPIC_PATH= @@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30 ROUTING_MODE=router # Usage/cost accounting -USAGE_REPOSITORY_PROVIDER=autonomous +USAGE_REPOSITORY_PROVIDER=sqlite IDENTITY_CONFIG_PATH=./config/identity.yaml MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml diff --git a/templates/agent_template_backend/app/main.py b/templates/agent_template_backend/app/main.py index 9a76977..d7c4ab1 100644 --- a/templates/agent_template_backend/app/main.py +++ b/templates/agent_template_backend/app/main.py @@ -74,6 +74,7 @@ logger.info("Framework channel input mode: %s", gateway.input_mode) @app.middleware("http") async def observability_context_middleware(request: Request, call_next): + clear_observability_context() request_id = request.headers.get("x-request-id") or str(uuid4()) set_observability_context( request_id=request_id, @@ -99,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next): "duration_ms": int((time.time() - started) * 1000), }, kind="http") raise + finally: + clear_observability_context() class GatewayRequest(BaseModel): diff --git a/templates/agent_template_backend/app/workflows/agent_graph.py b/templates/agent_template_backend/app/workflows/agent_graph.py index 2492b56..0e58ed3 100644 --- a/templates/agent_template_backend/app/workflows/agent_graph.py +++ b/templates/agent_template_backend/app/workflows/agent_graph.py @@ -320,41 +320,36 @@ class AgentWorkflow: } async def billing_agent(self, state): - async with self.langgraph_telemetry.node("billing_agent", state): - async with self.telemetry.span( - "workflow.agent.billing", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.billing.run(state) + async with self.telemetry.span( + "workflow.agent.billing", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.billing.run(state) async def product_agent(self, state): - async with self.langgraph_telemetry.node("product_agent", state): - async with self.telemetry.span( - "workflow.agent.product", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.product.run(state) + async with self.telemetry.span( + "workflow.agent.product", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.product.run(state) async def orders_agent(self, state): - async with self.langgraph_telemetry.node("orders_agent", state): - async with self.telemetry.span( - "workflow.agent.orders", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.orders.run(state) - + async with self.telemetry.span( + "workflow.agent.orders", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.orders.run(state) async def support_agent(self, state): - async with self.langgraph_telemetry.node("support_agent", state): - async with self.telemetry.span( - "workflow.agent.support", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.support.run(state) + async with self.telemetry.span( + "workflow.agent.support", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.support.run(state) async def supervisor_agent(self, state): """Executa um ou mais agentes no modo supervisor e consolida a resposta. diff --git a/templates/agent_template_backend_day_zero/.env b/templates/agent_template_backend_day_zero/.env index 9e908e2..6b58c39 100644 --- a/templates/agent_template_backend_day_zero/.env +++ b/templates/agent_template_backend_day_zero/.env @@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1 OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6 OCI_GENAI_PROJECT_OCID= -# OCI_AUTH_MODE=config_file|instance_principal|resource_principal -OCI_AUTH_MODE=config_file # OCI SDK / signer / profiles OCI_CONFIG_FILE=~/.oci/config OCI_PROFILE=DEFAULT @@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1 # Persistência ############################################################################### # Opções: memory, autonomous, mongodb -SESSION_REPOSITORY_PROVIDER=autonomous -MEMORY_REPOSITORY_PROVIDER=autonomous -CHECKPOINT_REPOSITORY_PROVIDER=autonomous +SESSION_REPOSITORY_PROVIDER=sqlite +MEMORY_REPOSITORY_PROVIDER=sqlite +CHECKPOINT_REPOSITORY_PROVIDER=sqlite +SQLITE_DB_PATH=./data/agent_framework.db # Autonomous Database ADB_USER=admin @@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false ############################################################################### # RAG / Vector / Graph ############################################################################### -VECTOR_STORE_PROVIDER=memory -GRAPH_STORE_PROVIDER=memory +VECTOR_STORE_PROVIDER=sqlite +GRAPH_STORE_PROVIDER=sqlite RAG_TOP_K=5 EMBEDDING_PROVIDER=mock OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0 @@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json # Observabilidade ############################################################################### ENABLE_LANGFUSE=true -LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3 -LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20 +LANGFUSE_TRACE_MODE=verbose # 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 ENABLE_OTEL=false 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. ENABLE_ANALYTICS=false # Providers aceitos: oci_streaming,pubsub,noop -ANALYTICS_PROVIDERS=oci_streaming +ANALYTICS_PROVIDERS=pubsub # Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente. AGENT_PUBSUB_TOPIC= GCP_PUBSUB_TOPIC_PATH= @@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30 ROUTING_MODE=router # Usage/cost accounting -USAGE_REPOSITORY_PROVIDER=autonomous +USAGE_REPOSITORY_PROVIDER=sqlite IDENTITY_CONFIG_PATH=./config/identity.yaml MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml diff --git a/templates/agent_template_backend_day_zero/app/main.py b/templates/agent_template_backend_day_zero/app/main.py index 2fad940..d7c4ab1 100644 --- a/templates/agent_template_backend_day_zero/app/main.py +++ b/templates/agent_template_backend_day_zero/app/main.py @@ -4,7 +4,7 @@ import logging from uuid import uuid4 import time -from fastapi import FastAPI, Request +from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from pydantic import BaseModel @@ -52,7 +52,7 @@ summary_memory = create_conversation_summary_memory(settings, message_history=me sessions = create_session_repository(settings) checkpoints = create_checkpoint_repository(settings) cache = create_cache(settings, telemetry=telemetry) -gateway = ChannelGateway() +gateway = ChannelGateway(input_mode=settings.FRAMEWORK_CHANNEL_INPUT_MODE) analytics = create_analytics_publisher(settings) observer = TelemetryBackedAgentObserver(telemetry=telemetry) 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("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("Framework channel input mode: %s", gateway.input_mode) @app.middleware("http") async def observability_context_middleware(request: Request, call_next): + clear_observability_context() request_id = request.headers.get("x-request-id") or str(uuid4()) set_observability_context( request_id=request_id, @@ -98,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next): "duration_ms": int((time.time() - started) * 1000), }, kind="http") raise + finally: + clear_observability_context() 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: - msg = await gateway.normalize(req.channel, req.payload) + try: + 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) agent_session_id = identity.conversation_key() message_id = (req.payload or {}).get("message_id") or str(uuid4()) @@ -331,6 +338,8 @@ async def health(): "usage_repository": settings.USAGE_REPOSITORY_PROVIDER, "identity_config_path": settings.IDENTITY_CONFIG_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, "ROUTING_CONFIG_PATH": settings.ROUTING_CONFIG_PATH, "ROUTING_MODE": settings.ROUTING_MODE, + "FRAMEWORK_CHANNEL_INPUT_MODE": settings.FRAMEWORK_CHANNEL_INPUT_MODE, + "CHANNEL_GATEWAY_MODE": settings.CHANNEL_GATEWAY_MODE, } diff --git a/templates/agent_template_backend_day_zero/app/workflows/agent_graph.py b/templates/agent_template_backend_day_zero/app/workflows/agent_graph.py index b49cf52..0e58ed3 100644 --- a/templates/agent_template_backend_day_zero/app/workflows/agent_graph.py +++ b/templates/agent_template_backend_day_zero/app/workflows/agent_graph.py @@ -187,6 +187,16 @@ class AgentWorkflow: input=state.get("user_text"), ): 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( state["user_text"], { @@ -199,6 +209,19 @@ class AgentWorkflow: ) for _decision in decisions: 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: await self.guardrail_telemetry.blocked("input", _decision) await self.telemetry.event( @@ -210,6 +233,18 @@ class AgentWorkflow: "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): return { "sanitized_input": sanitized, @@ -285,41 +320,36 @@ class AgentWorkflow: } async def billing_agent(self, state): - async with self.langgraph_telemetry.node("billing_agent", state): - async with self.telemetry.span( - "workflow.agent.billing", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.billing.run(state) + async with self.telemetry.span( + "workflow.agent.billing", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.billing.run(state) async def product_agent(self, state): - async with self.langgraph_telemetry.node("product_agent", state): - async with self.telemetry.span( - "workflow.agent.product", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.product.run(state) + async with self.telemetry.span( + "workflow.agent.product", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.product.run(state) async def orders_agent(self, state): - async with self.langgraph_telemetry.node("orders_agent", state): - async with self.telemetry.span( - "workflow.agent.orders", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.orders.run(state) - + async with self.telemetry.span( + "workflow.agent.orders", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.orders.run(state) async def support_agent(self, state): - async with self.langgraph_telemetry.node("support_agent", state): - async with self.telemetry.span( - "workflow.agent.support", - session_id=state.get("conversation_key") or state.get("session_id"), - input={"intent": state.get("intent")}, - ): - return await self.support.run(state) + async with self.telemetry.span( + "workflow.agent.support", + session_id=state.get("conversation_key") or state.get("session_id"), + input={"intent": state.get("intent")}, + ): + return await self.support.run(state) async def supervisor_agent(self, state): """Executa um ou mais agentes no modo supervisor e consolida a resposta. @@ -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}: final_answer = decision.candidate elif decision.action == RailAction.HANDOVER: @@ -457,11 +502,36 @@ class AgentWorkflow: session_id=state.get("conversation_key") or state.get("session_id"), 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( state["answer"], state.get("context", {}) ) for _decision in decisions: 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: await self.guardrail_telemetry.blocked("output", _decision) await self.telemetry.event( @@ -473,6 +543,18 @@ class AgentWorkflow: "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 { "final_answer": final, "guardrail_decisions": state.get("guardrail_decisions", [])