Skip to content

Architecture

Jump to bottom
Mohit Nagaraj edited this page Jan 9, 2026 · 13 revisions

X-Ray SDK & API - Architecture Documentation

A decision-reasoning observability system for multi-step pipelines. Captures why decisions were made (candidates, filters, scores, reasoning), not just what functions ran.

Table of Contents

Database Schema

The system uses 3 tables with clear relationships:

Link: https://dbdiagram.io/d/XRAY-DIAGRAM-6960bf2fd6e030a0248dae71 Untitled

Computed Fields (Query Time)

Field Formula Notes
removed_ratio (input_count - output_count) / NULLIF(input_count, 0) Computed at query time
duration_ms Stored in Step table Computed in SDK at step end

Type Locations

Enums (shared between SDK and API)

Located in shared/types.py:

class StepType(str, Enum):
    filter = "filter"       # Removes candidates based on criteria
    rank = "rank"           # Orders candidates by score
    llm = "llm"             # LLM/AI model call
    retrieval = "retrieval" # Fetches data from external source
    transform = "transform" # Transforms data format
    other = "other"         # Catch-all

class RunStatus(str, Enum):
    running = "running"     # In progress
    success = "success"     # Completed successfully
    error = "error"         # Failed with error

class StepStatus(str, Enum):
    running = "running"
    success = "success"
    error = "error"

class DetailLevel(str, Enum):
    summary = "summary"     # Counts + samples only
    full = "full"           # Complete data with truncation

Classes (SDK only)

Class Location Purpose
Run sdk/run.py Pipeline execution with Step factory
Step sdk/step.py Single decision step lifecycle
PayloadCollector sdk/step.py Collects large data during summarization
Transport sdk/transport.py Async buffered HTTP sender
XRayConfig sdk/config.py SDK configuration

Package Structure

shared/
└── types.py
    ├── StepType (enum)      ─┐
    ├── RunStatus (enum)      │  Used by both SDK and API
    ├── StepStatus (enum)     │
    └── DetailLevel (enum)   ─┘

sdk/
├── __init__.py              # Public exports
├── config.py
│   └── XRayConfig (class)
│   └── load_config (function)
├── transport.py
│   └── Transport (class)
├── run.py
│   └── Run (class)
└── step.py
    ├── PayloadCollector (class)
    ├── Step (class)
    ├── summarize_payload (function)
    ├── infer_count (function)
    ├── is_candidate_list (function)
    └── extract_candidate (function)

api/  (to be implemented)
├── models.py                # SQLAlchemy ORM models
├── schemas.py               # Pydantic request/response
├── store.py                 # Database operations
└── routes.py                # FastAPI endpoints

SDK Components

PayloadCollector

Location: sdk/step.py:31-63

Collects large data during summarization and assigns reference IDs.

class PayloadCollector:
    _payloads: dict[str, Any]  # {"p-000": [...], "p-001": "..."}
    _counter: int              # Sequential counter

    def add(self, data: Any) -> str
        # Stores data, returns "p-000", "p-001", etc.

    def get_payloads(self) -> dict[str, Any] | None
        # Returns payloads dict or None if empty

Step

Location: sdk/step.py:331-521

Represents a single decision step within a Run.

class Step:
    # Properties
    id: str                   # UUID
    run_id: str               # Parent run UUID
    name: str                 # "price_filter"
    step_type: StepType       # filter/rank/llm/etc.
    status: StepStatus        # running/success/error

    # Methods
    def attach_reasoning(self, reasoning: dict | str) -> None
        # Adds reasoning info to step

    def end(self, output: Any, status: StepStatus = success) -> None
        # Ends step with output data

    def end_with_error(self, error: BaseException | str) -> None
        # Ends step with error

Lifecycle:

  1. Created via Run.start_step() → sends step_start event
  2. Execute business logic
  3. Call step.end(output) or step.end_with_error(error) → sends step_end event

Run

Location: sdk/run.py:15-237

Represents a complete pipeline execution.

class Run:
    # Properties
    id: str                   # UUID
    pipeline_name: str        # "recommendation_pipeline"
    status: RunStatus         # running/success/error
    metadata: dict[str, Any]  # request_id, user_id, etc.

    # Methods
    def start_step(self, name, step_type, input_data, metadata) -> Step
        # Creates and starts a new Step

    def end(self, output=None, status=success) -> None
        # Ends the run

    def end_with_error(self, error, output=None) -> None
        # Ends the run with error

    # Context manager support
    def __enter__(self) -> Run
    def __exit__(self, exc_type, exc_val, exc_tb) -> bool

Lifecycle:

  1. Created directly or via XRayClient.start_run() → sends run_start event
  2. Create steps via run.start_step()
  3. Call run.end() or use context manager → sends run_end event

Transport

Location: sdk/transport.py:16-183

Async buffered transport with fail-open semantics.

class Transport:
    # Properties
    is_started: bool
    queue_size: int

    # Methods
    async def start(self) -> None
        # Creates HTTP client, starts background worker

    def send(self, event: dict) -> bool
        # Non-blocking queue put (drops if full)

    async def shutdown(self, timeout: float = 5.0) -> None
        # Graceful shutdown with queue drain

Fail-Open Behavior:

  • Queue full → event dropped (logged)
  • Network error → logged, retry with backoff
  • SDK never crashes the application

Payload Summarization

Constants

# Truncation limits
MAX_STRING_LENGTH = 1024       # Truncate inline strings at 1KB
MAX_DICT_KEYS = 50             # Max keys to extract from dicts
MAX_PAYLOAD_DEPTH = 5          # Max recursion depth

# Externalization thresholds
LARGE_LIST_THRESHOLD = 100     # Lists ≥100 items → externalize
LARGE_STRING_THRESHOLD = 2048  # Strings ≥2KB → externalize
PREVIEW_SIZE = 5               # Items in preview for large lists
STRING_PREVIEW_SIZE = 100      # Chars in preview for large strings

# ID detection
ID_FIELDS = ("id", "_id", "candidate_id", "item_id", "product_id", "doc_id")

Functions

infer_count(obj: Any) -> int | None

Returns count for list-like objects:

  • Lists, tuples, sets → len()
  • Dicts with keys items, results, data, records, candidates → length of that value
  • Otherwise → None

is_candidate_list(obj: Any) -> bool

Returns True if obj is a list of dicts where every dict has an ID field. Checks ALL items (no sampling) to ensure correctness.

extract_candidate(item: dict) -> dict

Extracts {id, score, reason} from a candidate dict:

  • ID: looks in id, _id, candidate_id, item_id, product_id, doc_id
  • Score: looks in score, rank, relevance, confidence, weight
  • Reason: looks in reason, explanation, rationale, why, filter_reason

summarize_payload(obj, depth=0, collector=None) -> dict

Main summarization function. Handles all types recursively:

Input Type Condition Output
None - {"_type": "null", "_value": None}
bool - {"_type": "bool", "_value": True}
int/float - {"_type": "int", "_value": 42}
str < 2KB {"_type": "str", "_length": N, "_value": "..."}
str ≥ 2KB {"_type": "str", "_length": N, "_ref": "p-000", "_preview": "..."}
bytes - {"_type": "bytes", "_length": N}
list is_candidate_list {"_type": "candidates", "_count": N, "_candidates": [...]}
list < 100 items {"_type": "list", "_count": N, "_values": [...]}
list ≥ 100 items {"_type": "list", "_count": N, "_ref": "p-000", "_preview": [...]}
dict - {"_type": "dict", "_key_count": N, "_keys": [...], "_values": {...}}
object - {"_type": "ClassName", "_id": "..."}
any depth ≥ 5 {"_type": "...", "_truncated": true}

Payload Externalization

Large data is stored separately to keep summaries compact while preserving full data.

When Does Externalization Happen?

In the SDK, during summarize_payload() execution — before data is sent to the API.

Thresholds

Data Type Threshold Small (Inline) Large (Externalized)
Lists 100 items _values: [all items] _ref: "p-000" + _preview
Strings 2KB _value: "full string" _ref: "p-001" + _preview
Candidates Never Always inline (id+score+reason) N/A

Flow Diagram

Untitled-2025-11-28-1402

Linking Mechanism

Layer What Happens
SDK: summarize_payload() Large data → collector.add(data) → returns "p-000"
SDK: Summary Contains {"_ref": "p-000", "_preview": [...]}
SDK: Event Contains both summary AND _payloads: {"p-000": [...]}
API: /ingest Extracts _payloads, stores in Payload table
DB: Step Stores summary with _ref pointers only
DB: Payload Stores full data, linked by (step_id, ref_id, event_type)
Query Time Join Step ↔ Payload to get full data

Event Structures

Events sent from SDK to API via Transport.

run_start

{
  "event_type": "run_start",
  "id": "run-uuid",
  "pipeline_name": "recommendation_pipeline",
  "status": "running",
  "started_at": "2024-01-15T10:30:00Z",
  "input_summary": { ... },
  "metadata": { "request_id": "req-123", "user_id": "u-456" },
  "request_id": "req-123",
  "user_id": "u-456",
  "environment": "prod",
  "_payloads": { "p-000": [...] }
}

run_end

{
  "event_type": "run_end",
  "id": "run-uuid",
  "status": "success",
  "ended_at": "2024-01-15T10:30:05Z",
  "output_summary": { ... },
  "error_message": null,
  "_payloads": { "p-000": [...] }
}

step_start

{
  "event_type": "step_start",
  "id": "step-uuid",
  "run_id": "run-uuid",
  "step_name": "price_filter",
  "step_type": "filter",
  "index": 0,
  "started_at": "2024-01-15T10:30:01Z",
  "input_summary": {
    "_type": "candidates",
    "_count": 500,
    "_candidates": [
      {"id": "p-1", "score": 0.9, "reason": null},
      {"id": "p-2", "score": 0.85, "reason": null}
    ]
  },
  "input_count": 500,
  "metadata": { "threshold": 100 },
  "_payloads": null
}

step_end

{
  "event_type": "step_end",
  "id": "step-uuid",
  "run_id": "run-uuid",
  "status": "success",
  "ended_at": "2024-01-15T10:30:02Z",
  "duration_ms": 1200,
  "output_summary": {
    "_type": "candidates",
    "_count": 50,
    "_candidates": [
      {"id": "p-5", "score": 0.95, "reason": "best match"},
      {"id": "p-8", "score": 0.92, "reason": null}
    ]
  },
  "output_count": 50,
  "reasoning": { "threshold": 100, "explanation": "Filtered by price" },
  "error_message": null,
  "_payloads": null
}

Complete Data Flow

┌─────────────────────────────────────────────────────────────────────────────┐
│                              USER CODE                                       │
│                                                                              │
│  from sdk import Run, Transport, XRayConfig                                 │
│                                                                              │
│  transport = Transport(XRayConfig(base_url="http://localhost:8000"))        │
│  await transport.start()                                                    │
│                                                                              │
│  with Run(transport, "rec_pipeline", input_data=query) as run:              │
│      # Step 1: Retrieve candidates                                          │
│      step1 = run.start_step("retrieve", "retrieval", query)                 │
│      candidates = db.search(query)  # Returns 500 items                     │
│      step1.end(candidates)                                                  │
│                                                                              │
│      # Step 2: Filter by price                                              │
│      step2 = run.start_step("price_filter", "filter", candidates)           │
│      filtered = [c for c in candidates if c["price"] < 100]                 │
│      step2.attach_reasoning({"threshold": 100})                             │
│      step2.end(filtered)  # 50 items remain                                 │
│                                                                              │
│      # Step 3: Rank by relevance                                            │
│      step3 = run.start_step("rank", "rank", filtered)                       │
│      ranked = model.rank(filtered)                                          │
│      step3.end(ranked)                                                      │
│                                                                              │
│  await transport.shutdown()                                                 │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘
                                     │
                                     │ Events queued in Transport
                                     ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              TRANSPORT                                       │
│                                                                              │
│  Queue: [run_start, step1_start, step1_end, step2_start, ...]              │
│                                                                              │
│  Background worker:                                                          │
│  • Collects batch (up to batch_size or flush_interval)                      │
│  • POST /ingest with JSON array                                             │
│  • Fail-open: network errors logged, not raised                             │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘
                                     │
                                     │ HTTP POST /ingest
                                     ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              API SERVER                                      │
│                                                                              │
│  POST /ingest:                                                               │
│  for event in events:                                                        │
│      payloads = event.pop("_payloads", None)                                │
│                                                                              │
│      if event["event_type"] == "run_start":                                 │
│          db.insert(Run(...))                                                │
│      elif event["event_type"] == "step_start":                              │
│          step = db.insert(Step(...))                                        │
│          if payloads:                                                        │
│              for ref_id, data in payloads.items():                          │
│                  db.insert(Payload(step_id=step.id, ref_id=ref_id, ...))    │
│      # ... handle other event types                                         │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘
                                     │
                                     ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              DATABASE                                        │
│                                                                              │
│  Run:     id="run-123", pipeline_name="rec_pipeline", status="success"      │
│                                                                              │
│  Step:    id="step-1", run_id="run-123", step_name="retrieve", index=0      │
│           input_summary={"_type": "dict", ...}                              │
│           output_summary={"_type": "candidates", "_count": 500, ...}        │
│                                                                              │
│  Step:    id="step-2", run_id="run-123", step_name="price_filter", index=1  │
│           input_count=500, output_count=50                                  │
│           reasoning={"threshold": 100}                                      │
│                                                                              │
│  Payload: step_id="step-1", ref_id="p-000", data=[...large data...]         │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Configuration

SDK Configuration (xray.config.yaml)

sdk:
  base_url: http://localhost:8000   # API endpoint
  api_key: your-api-key             # Auth token (optional)
  buffer_size: 1000                 # Max events in queue
  flush_interval: 5.0               # Seconds between flushes
  batch_size: 100                   # Events per HTTP request
  http_timeout: 30.0                # HTTP timeout in seconds

API Configuration (xray.config.yaml)

api:
  database_url: postgresql+asyncpg://localhost:5432/xray
  debug: false

Config File Discovery

The config file is found by searching from the current directory upward:

/home/user/myproject/src/main.py  (running here)
         ↓ searches upward
/home/user/myproject/xray.config.yaml  (found!)

Key Design Decisions

  1. Counts as columns: input_count/output_count indexed for efficient filtering
  2. removed_ratio computed at query time: Allows formula changes without migration
  3. Payload externalization: Large data stored separately, linked by _ref
  4. All candidate IDs captured: Never truncate candidate lists (essential for debugging)
  5. Fail-open SDK: Network errors never crash the application
  6. Context manager support: with Run(...) as run: for automatic cleanup

Clone this wiki locally