Skip to content

Architecture

Jump to bottom
mohit-nagaraj edited this page Jan 13, 2026 · 13 revisions

X-Ray is a decision-reasoning observability system for multi-step pipelines. It captures why decisions were made (candidates, filters, scores, reasoning), not just what functions ran.

Database Schema

View interactive diagram on dbdiagram.io

XRAY DIAGRAM (2)

Tables

Table Purpose
Run Pipeline execution (id, pipeline_name, status, timestamps, metadata)
Step Single processing step within a run (input/output counts, reasoning, timing)
Payload Externalized large data (lists ≥100 items, strings ≥1KB)

Computed Fields

  • removed_ratio = (input_count - output_count) / input_count — stored in Step (indexed for fast filtering)
  • duration_ms — stored in Step, computed at step end in SDK

Flow Diagram

Screenshot 2026-01-09 at 3 10 37 PM

Type System

Enums (shared between SDK & API)

Located in shared/types.py:

StepType

Value Description
filter Removes candidates by criteria
rank Orders candidates by score
llm LLM/AI model call
retrieval Fetches external data
transform Data format changes
other Catch-all

RunStatus / StepStatus: running, success, error

DetailLevel: summary (counts + samples), full (complete data with truncation)

SDK Components

XRayClient

Entry point for the SDK. Manages Transport lifecycle and provides start_run() context manager.

from xray_logger import init_xray

client = init_xray(base_url="http://localhost:8000")
with client.start_run("pipeline") as run:
    # Steps created here

Run (sdk/_internal/run.py)

Represents a pipeline execution. Creates Steps, manages context via contextvars.

Step (sdk/_internal/step.py)

Single processing step. Captures input/output, timing, and reasoning via PayloadCollector and summarize_payload().

Transport (sdk/_internal/transport.py)

Async buffered HTTP sender with fail-open semantics—network errors are logged but never crash the application.

Decorators (sdk/decorators.py)

  • @step(name, step_type) — Instrument a function
  • @instrument_class(step_type) — Instrument all public methods of a class
  • attach_reasoning(data) — Add reasoning to current step
  • attach_candidates(candidates) — Add candidate list to current step

Middleware (sdk/middleware.py)

  • XRayMiddleware — FastAPI/Starlette middleware for automatic HTTP request instrumentation

Payload Summarization

Large data is automatically summarized to keep event sizes manageable:

Data Type Threshold Behavior
Lists < 100 items Stored inline
Lists ≥ 100 items Externalized with preview
Strings < 1KB Stored inline
Strings ≥ 1KB Externalized with preview
Candidate lists Any size Always inline (id + score + reason only)

Externalization Flow

  1. SDK summarize_payload() detects large data
  2. Creates reference: {"_ref": "p-000", "_type": "list", "_count": 500, "_preview": [...]}
  3. Event includes _payloads: {"p-000": [full data]}
  4. API extracts payloads, stores in Payload table
  5. Step stores summary only; Payload stores full data

Event Types

run_start / run_end

{
  "event_type": "run_start",
  "id": "uuid",
  "pipeline_name": "my-pipeline",
  "status": "running",
  "started_at": "2024-01-01T00:00:00Z",
  "input_summary": {...},
  "metadata": {"user_id": "123"},
  "_payloads": {...}
}

step_start / step_end

{
  "event_type": "step_end",
  "id": "uuid",
  "run_id": "parent-run-uuid",
  "step_name": "filter_candidates",
  "step_type": "filter",
  "status": "success",
  "duration_ms": 150,
  "input_count": 100,
  "output_count": 25,
  "reasoning": {"threshold": 0.7, "removed": 75},
  "_payloads": null
}

Configuration

SDK (xray.config.yaml)

sdk:
  base_url: http://localhost:8000
  api_key: your-api-key          # Optional, or set XRAY_API_KEY env var
  buffer_size: 1000              # Max events to buffer
  flush_interval: 5.0            # Seconds between flushes
  batch_size: 100                # Events per batch
  http_timeout: 30.0             # HTTP request timeout

Programmatic Config

from xray_logger import init_xray, XRayConfig

client = init_xray(XRayConfig(
    base_url="http://localhost:8000",
    api_key="your-key",
    buffer_size=500,
    flush_interval=2.0
))

API (Environment Variables)

XRAY_DATABASE_URL=postgresql+asyncpg://localhost:5432/xray
XRAY_API_KEY=your-secret-key  # Optional, for authentication

Config discovery searches upward from current directory for xray.config.yaml.

Key Design Decisions

  1. Counts as columns — input_count/output_count are indexed for efficient filtering
  2. Stored ratios — removed_ratio computed on step end and stored for indexed querying
  3. Payload externalization — Large data stored separately, summaries inline (thresholds: 100 items, 1KB strings)
  4. Candidate preservation — All candidate IDs captured, never truncated
  5. Fail-open SDK — Network errors never crash the application
  6. Context managers — Automatic cleanup on exceptions
  7. No separate Candidate table — Candidates embedded in Step.metadata to avoid join complexity

Project Structure

xray-logger/
├── shared/
│   ├── types.py          # StepType, RunStatus, DetailLevel
│   └── config.py         # Config discovery utilities
├── sdk/
│   ├── __init__.py       # Public API exports
│   ├── client.py         # XRayClient, init_xray()
│   ├── config.py         # XRayConfig, load_config()
│   ├── types.py          # SDK-specific types
│   ├── decorators.py     # @step, @instrument_class, attach_*
│   ├── middleware.py     # XRayMiddleware for FastAPI/Starlette
│   └── _internal/
│       ├── run.py        # Run class implementation
│       ├── step.py       # Step, PayloadCollector, summarize_payload()
│       └── transport.py  # Async buffered HTTP transport
├── api/
│   ├── __init__.py
│   ├── main.py           # FastAPI app
│   ├── routes.py         # /health, /ingest, /xray/runs, /xray/steps
│   ├── models.py         # SQLAlchemy ORM (Run, Step, Payload)
│   ├── schemas.py        # Pydantic request/response schemas
│   ├── config.py         # API configuration
│   ├── auth.py           # API key authentication
│   └── _internal/
│       ├── database.py   # Database session management
│       └── store.py      # Database CRUD operations
└── examples/
    ├── competitor_pipeline/        # RAG pipeline with retrieval + LLM
    ├── recommendation-pipeline/    # Product recommendation filtering + ranking
    ├── fastapi-middleware/         # HTTP instrumentation example
    └── product-categorization/     # Multi-step categorization pipeline

Data Flow

image

Authentication

Optional API key authentication via:

  • SDK: api_key in config or XRAY_API_KEY env var
  • API: XRAY_API_KEY env var (when set, all endpoints except /health require auth)
  • Header: Authorization: Bearer <api-key>

Deployment Options

  1. Docker Compose (Recommended) - PostgreSQL + X-Ray API
  2. Docker Run - Single container with SQLite
  3. Development - Build from source with docker-compose.dev.yml

See Quick Start for deployment instructions.

Additional Resources

Clone this wiki locally