Skip to content

Agent Observability (Four-Layer Observation Model)

This page systematizes Eagle-RAG's observability design with a four-layer model: how to quickly locate an "agent execution failure" online — i.e. a call that ran the full RAG chain but produced an abnormal result. All four layers are implemented; the section at the end gives a Langfuse optional integration design (not implemented).

Boundary statement

Eagle-RAG is a RAG data layer (ADR-008), not an Agent application platform. "Agent execution failure" here means: (a) Eagle-RAG's own chain failed (the downstream agent observes an error / empty result / timeout); (b) Eagle-RAG provides a trace_id context that downstream agents can correlate. Operational endpoints and commands live in Observability (operations).

1. Four-layer model overview

Layer Question it answers Implementation Typical failure-location question
L1 Trace "Which step did the failure happen in?" OpenTelemetry tracing Some span ERROR / long latency
L2 Metrics "Systemic or one-off?" Prometheus + metric_sample circuit breaker, backlog, latency spike
L3 Input/Output "What were the exact IO this time?" structlog ai_telemetry.jsonl prompt/completion/hits abnormal
L4 State "How did decisions/state evolve?" PluginAudit + task_audit + documents.status routing error, stuck state, dedupe anomaly

Core correlation key: trace_id threads through all four layers — L1 spans, L3 JSONL entries, and L4 PluginAuditEvent all carry trace_id/span_id (injected by telemetry/context.py + logging_setup.add_open_telemetry_span). A single failure is stitched across the four layers with one trace_id.

2. Layer 1 — Trace execution chain

Code: eagle_rag/telemetry/tracing.py

Capability Function Behavior
Tracing bootstrap configure_tracing (L56) Builds TracerProvider (resource with service.name/version/environment); picks exporter by tracing_enabled/otlp_endpoint: OTLP gRPC + BatchSpanProcessor (L92-97) / ConsoleSpanExporter / NoOp
Open span trace_span (L109) Dual form (context manager + parameterless decorator); on enter bind_context(trace_id, span_id); on exception record_exception + set_status(ERROR)
HTTP SERVER span TelemetryMiddleware (L215) Opens a SERVER span per request, extracts W3C traceparent to continue the upstream trace, binds request_id; status>=500 marks ERROR
Celery CONSUMER span register_celery_signals (L277) task_prerun extracts the parent span from task.request.headers and opens a CONSUMER span (named {task.name}:{task_id}), binds job_id/document_id/kb_name; task_failure records the exception
API→Celery propagation send_task_with_trace (L366) injects the current span context into Celery headers, continuing the trace across processes
GenAI semantic conventions set_llm_span_attributes (L179) Annotates gen_ai.system/gen_ai.request.model/gen_ai.prompt/gen_ai.completion/gen_ai.usage.*; prompt/completion truncated by prompt_truncate(512)/completion_truncate(1024)

Typical /query trace tree

SERVER  POST /query                        (TelemetryMiddleware)
├─ span  route                              (QueryRouteClassifier)
├─ span  retrieve                           (RetrieverOrchestrator)
│   ├─ span  retrieve.text (per plan)
│   └─ span  retrieve.visual (per plan)
├─ span  rerank                             (rerank_merged)
└─ span  generate (gen_ai.*)                (EagleMultimodalQueryEngine)

Async ingest runs on a separate trace: API SERVER → send_task_with_trace dispatch → Celery CONSUMER (ingest_router/knowhere_parse/pixelrag_build).

Tracing is off by default

telemetry.tracing_enabled defaults to false (settings.yaml). When off, a NoOp TracerProvider is still installed; trace_span degrades to a no-op but still produces a valid trace_id for log correlation (the fallback in logging_setup.add_open_telemetry_span). To enable, set OTEL_TRACING_ENABLED=true + OTEL_EXPORTER_OTLP_ENDPOINT.

3. Layer 2 — Metrics

Code: eagle_rag/metrics.py (MCP/Prometheus) + eagle_rag/admin/metrics.py (metric_sample time-series table)

Prometheus metrics (/metrics endpoint)

Metric Type Labels Meaning
mcp_tool_calls_total Counter tool, status MCP tool call count; status = success/cache_hit/circuit_open/timeout/error
mcp_tool_duration_seconds Histogram tool Tool call latency
mcp_active_requests Gauge tool In-flight requests
mcp_circuit_state Gauge tool Circuit-breaker state 0=closed/1=half-open/2=open
plugin_audit_decisions_total Counter category, plugin_namespace, outcome Plugin decision count; outcome = ok/error
plugin_audit_rrf_dedupe_total Counter plugin_namespace Cross-collection RRF dedupe events (G32 dual-write monitoring)

metric_sample time-series table (PG)

eagle_rag/db/models/metric_sample.py defines a generic sampling table, periodically sampled by the Celery beat job in admin/metrics.py:

metric_name Source Purpose
queue_size sample_queue_lengths (beat) Depth of each Celery queue
vlm_latency_ms generation path sampling VLM latency
vlm_tokens generation path sampling token usage
vlm_error generation path sampling VLM errors

GET /health exposes queue time-series (_read_queue_sizes, api/health.py L476); get_metric_aggregate(name, agg, hours) queries aggregates (e.g. 24h avg of vlm_latency_ms, L1154).

Locating with metrics: mcp_circuit_state=2 → some tool circuit-broken; queue_size spike → worker backlog; mcp_tool_calls_total{status="error"} rate rising → systemic failure, not one-off.

4. Layer 3 — Input/Output

Code: eagle_rag/telemetry/logging_setup.py

Capability Function Behavior
AI events logger get_ai_logger(name) (L255) structlog BoundLogger, bound to component=name, writes logs/ai_telemetry.jsonl (telemetry.ai_log_file); lazy proxy, falls back to stdlib when telemetry is off
trace injection add_open_telemetry_span (L196) structlog processor: injects trace_id/span_id/parent_span_id from the OTel span; when no span is recording, reads from contextvars or generates a random hex, so every log line is correlatable
Truncation truncate(text, limit) (L348) prompt/completion/query/hits beyond prompt_truncate(512)/completion_truncate(1024) are truncated and marked ...<truncated>, preventing JSONL bloat and full persistence of sensitive data
Ops logger get_logger(name) (L266) loguru logger.bind, reads contextvars on each call to keep trace_id current; multi-sink: stderr (pretty) + rotating file (JSON) + Redis pubsub

Hot-path instrumentation events cover route / retrieve / rerank / generate / ingest / mcp_call / query (see Observability (operations) §AI events logger). Each JSONL line carries trace_id/span_id/session_id/query_id/kb_name (bound via bind_context).

Redis pubsub sink (_make_redis_sink, L314) publishes {level, message, timestamp, trace_id?} to redis_log_channel (default logs), consumed by the /admin/logs SSE for real-time subscription.

5. Layer 4 — State changes

Code: eagle_rag/plugins/audit.py (PluginAudit) + eagle_rag/db/repositories/task_audit.py + documents.status transitions

PluginAudit: multi-sink decision audit

PluginAudit.log_decision (audit.py L200) is the stable call point, fanning out to four sinks (all best-effort, never throwing into the caller):

Sink Implementation Purpose
In-memory ring deque(maxlen=ring_cap) (L145), default 1000 In-process recent window, fallback when Redis is unavailable
Redis LIST LPUSH + LTRIM (L231), key eagle:plugin_audit:{ns}:recent Cross-process recent window
AI JSONL _emit_ai_log (L246) via get_ai_logger Persistent, aggregatable
Prometheus _emit_metrics (L275) Aggregate dashboards

PluginAuditEvent (L97) fields: category / target_collection / confidence / reason / plugin_namespace / kb_name / document_id / error / trace_id / span_id / extra. When reason=="rrf_dedupe", it additionally increments plugin_audit_rrf_dedupe_total (connects to the dedupe audit in Evidence aggregation).

Other state sources

  • task_audit repository: Celery task audit persistence;
  • documents.status transitions: ingest state machine PENDING → RENDERING → EMBEDDING → INDEXING → SUCCESS (Reliability);
  • collections_used catalog: the actually-hit collections.

Exposed endpoint

GET /health/plugins exposes recent_decisions (PluginAudit.recent(), newest-last) and audit_stats() ({buffer_size, source, enabled, redis_enabled}).

6. Failure-location playbook

Given a failed agent / RAG call (e.g. the downstream agent reports "query returned empty" or a timeout):

1. Get the trace_id
   - If the downstream agent passed a W3C traceparent, Eagle-RAG continues it (TelemetryMiddleware L237 extract)
   - Otherwise grab the trace_id Eagle-RAG generated from /admin/logs SSE, response headers, or logs

2. L1 Trace — find the ERROR span to locate the failing stage
   - Filter by trace_id in the OTel backend; find spans with status=ERROR or abnormal latency
   - Without a backend: grep trace_id through L3 JSONL (below) and reconstruct stages by event order

3. L3 Input/Output — inspect the exact IO to tell input problem vs processing problem
   grep <trace_id> logs/ai_telemetry.jsonl | jq .
   - Check whether the retrieve event hits is empty (recall failure)
   - Check the generate event prompt/completion (truncated to 512/1024)

4. L2 Metrics — check whether systemic to rule out one-off jitter
   - /metrics for mcp_circuit_state = 2 (circuit broken)
   - /admin/celery for queue_size backlog
   - GET /health for vlm_latency_ms / vlm_error trends

5. L4 State — inspect decisions and state to find routing/classification errors
   - GET /health/plugins for recent_decisions
   - Check category / target_collection / reason for mis-routing
   - Check documents.status for a stuck intermediate state

Operational commands (consistent with existing endpoints):

# Reconstruct all AI events of one call by trace_id
grep <trace_id> logs/ai_telemetry.jsonl | jq .

# Queue and time-series
curl /admin/celery        # queue depth (_read_queue_sizes)
curl /health/plugins      # recent_decisions + audit_stats
curl /metrics             # raw Prometheus metrics

7. Langfuse optional integration design (not implemented)

This section is an unimplemented design. Langfuse is not integrated: grep -ri langfuse only hits the "non-goals" section of .trae/specs/integrate-ai-telemetry-tracing/spec.md. This design gives zero-intrusion integration points based on the existing OTel setup; landing it needs a new ADR.

7.1 Integration principle

Langfuse natively supports OTLP HTTP ingestion (https://cloud.langfuse.com/api/public/otel, Basic Auth = base64(public_key:secret_key), with an extra x-langfuse-ingestion-version=4 header). Key mappings:

  • spans with gen_ai.* attributes → Langfuse generation;
  • other spans → ordinary Langfuse observation (trace/span tree).

Eagle-RAG already annotates generation spans with gen_ai.* via set_llm_span_attributes (tracing.py L179), and the whole chain has OTel spans. So integrating Langfuse needs no business-code change — only export the spans to Langfuse's OTLP endpoint.

7.2 Integration point (pick one)

configure_tracing (tracing.py L56) currently uses the gRPC exporter:

# Current (tracing.py L92-96)
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(endpoint=tel.otlp_endpoint, insecure=tel.otlp_insecure)

The pyproject.toml dependency opentelemetry-exporter-otlp package bundles both gRPC and HTTP exporters, so the HTTP branch needs no new dependency:

# Design (not implemented): Langfuse branch
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
    endpoint=tel.langfuse_endpoint,  # https://cloud.langfuse.com/api/public/otel
    headers={
        "Authorization": f"Basic {base64(f'{pk}:{sk}')}",
        "x-langfuse-ingestion-version": "4",
    },
)

Option B: OTel Collector sidecar

Deploy an OTel Collector; Eagle-RAG still sends gRPC to the Collector, which does gRPC→HTTP forwarding and injects the Langfuse Auth header. Suits a unified egress for multiple services.

7.3 Eagle-RAG span → Langfuse observation mapping

Eagle-RAG span Langfuse observation Notes
POST /query (SERVER) trace root
route / retrieve / rerank span
generate (with gen_ai.*) generation auto-carries model/tokens/prompt/completion
ingest_router / knowhere_parse (CONSUMER) span async ingest trace

7.4 Langfuse score integration point (connects to citationware)

Langfuse's score mechanism can carry quality evaluation, connecting to the Citationware RAG roadmap:

  • faithfulness: whether each claim is backed by a cited source;
  • citation coverage: proportion of uncited claims in the answer;
  • abstain rate: proportion of explicit refusals when evidence is insufficient.

After generation you may use the Langfuse Python SDK's @observe / start_as_current_observation to explicitly enrich input/output and score() report (optional, not required — pure OTLP ingestion already shows the trace tree).

7.5 Required config knobs (design level, no code)

# settings.yaml telemetry section (design, not implemented)
telemetry:
  langfuse:
    enabled: ${LANGFUSE_ENABLED:-false}
    public_key: ${LANGFUSE_PUBLIC_KEY:-}
    secret_key: ${LANGFUSE_SECRET_KEY:-}
    endpoint: ${LANGFUSE_OTLP_ENDPOINT:-https://cloud.langfuse.com/api/public/otel}

When landing, add ADR-009 to record decisions like "why OTLP HTTP direct vs Collector sidecar" and "score evaluation scope."

8. References