crosswatch

Unified observability for DATA and AI products.

One plugin-driven API, one config format, one local store, and one CLI across data quality and LLM/agent observability — by wrapping and composing best-in-class OSS rather than reinventing it.

🚧 Alpha. See docs/the architecture notes for the full design.

Table of contents

• Features
• What's unique: data → AI impact correlation
• Why
• Install
• Connectors
• Checks
• Evals
• Alerts
• Exporters
• 5-minute quickstart
• Configuration — YAML, JSON, or Python
• Examples
• Local demo services
• Running in the cloud
• Architecture at a glance
• Known limitations / roadmap
• Development
• License

Features

• Data: freshness, volume, schema-change, null/duplicate, and drift checks as portable Ibis expressions that run unchanged on DuckDB, SQLite, Postgres, MySQL, ClickHouse, Snowflake, BigQuery, Databricks — plus dbt artifact ingestion (Great Expectations / Soda providers planned). Failures carry the reason, a suggested fix, and a sample of the exact offending rows/keys/bins.
• AI: OpenTelemetry-native tracing for OpenAI/Anthropic (and LangChain/LlamaIndex via OpenInference), with token / cost / latency on every span using GenAI semantic conventions.
• MCP: trace every tool call and evaluate tool-selection accuracy, argument correctness, and task success — turn recorded sessions into regression suites.
• Evals: evaluator plugin API + CI mode that fails the build on a score regression. Ships an LLM-as-judge evaluator (deterministic or real-LLM grading) and the MCP tool-use evals above (DeepEval/Ragas providers planned, not yet built — see Evals).
• Store / alerts / UI: default DuckDB/Parquet store, OTLP export to any backend (Phoenix/Langfuse/Jaeger/Datadog), Slack/webhook alerts, and a minimal FastAPI dashboard.

What's unique: data → AI impact correlation

Data-observability tools (Great Expectations, Soda, Monte Carlo) stop at the warehouse. LLM-observability tools (Langfuse, Phoenix, LangSmith) start at the prompt. Nothing joins them — so when a RAG eval regresses, you can't tell from one tool whether a data problem upstream caused it.

crosswatch records both data-quality results and AI evals in one store, so it correlates them. Tag an eval with the data assets it depends on, and crosswatch links a failing check on that asset to the AI output it degraded:

crosswatch impact -c crosswatch.yml

Data → AI impact (1 link):
  [warehouse.customers] null_duplicate FAILED ⇒ IMPACTED
      rag:answer_quality=0.42 (12 min later)
      data cause: email null_fraction=0.100 > 0.0; 1 duplicate rows over key ['id']

from crosswatch.models import EvalResult
from crosswatch.impact import correlate

# Declare the data an AI output depends on:
EvalResult(evaluator="rag", metric="answer_quality", score=0.42, threshold=0.8,
           assets=["warehouse.customers"])

links = correlate(checks, evals)   # failing check on an asset ⟶ the evals that use it

This answers a question no single-domain tool can: which data problem is degrading my AI, and which AI outputs are at risk from this data problem? It shows up as a dedicated card on the dashboard, and crosswatch impact exits non-zero when a data failure impacted a failing eval — a CI gate across the data/AI boundary.

Why

Teams stitch together Great Expectations for data quality, OpenTelemetry for tracing, DeepEval/Ragas for evals, and bespoke MCP logging — each with its own config, storage, and alerting. crosswatch is the composition layer: it doesn't re-implement expectation engines or tracers, it gives them one surface.

Install

pip install crosswatch                              # thin core (DuckDB + native checks)
pip install 'crosswatch[postgres,anthropic,server]' # add only what you need

Extras: sqlite postgres mysql clickhouse snowflake bigquery databricks openai anthropic langchain llama-index mcp gx soda evidently deepeval ragas otlp phoenix langfuse server slack. Managed with uv.

gx, soda, deepeval, and ragas currently only install the underlying library — see Known limitations / roadmap.

Connectors

One adapter per source, all exposing the same interface (get_schema / profile / run_check / get_query_history) — so a check definition runs unchanged on any of them. Referenced in config by type:

pip install 'crosswatch[mysql,clickhouse]'

Third parties can publish their own connector by registering a crosswatch.connectors entry point — no change to the core.

Checks

Every check is a Pydantic model discovered via the crosswatch.checks entry-point group; set it in config with type. List what's available (built-ins plus any third-party checks installed) at any time:

from crosswatch import plugins

print(plugins.available(plugins.CHECKS))
# ['drift', 'freshness', 'null_duplicate', 'schema_change', 'volume']

python -c "from crosswatch import plugins; print(plugins.available(plugins.CHECKS))"

Every check also takes the common fields table, severity (error/warn/info, default error), and an optional name:

Third parties can publish their own check by registering a crosswatch.checks entry point — no change to the core.

Evals

An evaluator scores an EvalCase (an AI input/output/context — e.g. a RAG answer or an agent's tool call) and returns an EvalResult. Every evaluator returns the same result shape, so a rule-based score, an LLM judge, and a replayed MCP session all flow into the same store, CI mode, and alerts. Evaluators are discovered via the crosswatch.evaluators entry-point group:

from crosswatch import plugins

print(plugins.available(plugins.EVALUATORS))
# ['llm_judge']

LLM-as-judge

LLMJudge scores a case 0.0–1.0 against a rubric. Inject a deterministic score_fn for CI/tests (no LLM call needed), or pass a real client to grade with a model:

from crosswatch.evals import LLMJudge, run_dataset
from crosswatch.models import EvalCase

cases = [
    EvalCase(id="1", input="What is the capital of France?",
             context=["Paris is the capital of France."], output="Paris"),
]

# CI-friendly: deterministic, no LLM call
judge = LLMJudge(metric="answer_quality", threshold=0.8,
                  score_fn=lambda c: 1.0 if "Paris" in c.output else 0.0)
results = run_dataset(judge, cases)

# Or grade with a real LLM:
import anthropic
judge = LLMJudge(metric="answer_quality", threshold=0.8,
                  client=anthropic.Anthropic(), model="claude-opus-4-8")
results = run_dataset(judge, cases)

CLI note: crosswatch eval -c crosswatch.yml currently only dispatches mcp_tool_selection/mcp evaluators configured in crosswatch.yml's evals: block (see Examples below) — any other evaluator name, including llm_judge, is skipped with a warning. LLMJudge is Python-API-only for now.

MCP tool-use evals

tool_selection_accuracy, argument_correctness, and task_success_rate score recorded MCP sessions against ground truth, and run_regression replays them through a live agent to catch drift between model/prompt versions — see MCP — trace tool calls + eval tool selection for a worked example.

gx/soda/deepeval/ragas evaluator providers aren't built yet — see Known limitations / roadmap.

Alerts

An alert channel delivers a failing check/eval result somewhere (Slack, webhook). dispatch turns results into AlertEvents and fans them out to whichever channels you pass it — only failures by default, or every result with on="always". Channels are discovered via the crosswatch.alerts entry-point group:

from crosswatch import plugins

print(plugins.available(plugins.ALERTS))
# ['slack', 'webhook']

from crosswatch.alerts.base import dispatch
from crosswatch.alerts.webhook import WebhookChannel

channel = WebhookChannel(url="https://example.com/hooks/crosswatch")
sent = dispatch(results, [channel])              # failures only
sent = dispatch(results, [channel], on="always")  # every result

Config note: crosswatch.yml's alerts: block parses into AlertConfig objects, but neither crosswatch check nor crosswatch eval call dispatch yet — configuring alerts: today is a no-op. Call dispatch yourself in the Python API until the CLI wires it in; see Known limitations / roadmap.

Third parties can publish their own channel by registering a crosswatch.alerts entry point — no change to the core.

Exporters

Span exporters send OpenTelemetry traces to a tracing backend — set via telemetry.exporters in config (see Configuration) or crosswatch.ai.configure(exporters=[...]) directly. Unlike alerts, this one is fully wired: both the CLI and the Python API call into it. Discover what's available:

from crosswatch import plugins

print(plugins.available(plugins.EXPORTERS))
# ['langfuse', 'otlp', 'phoenix']

Third parties can publish their own exporter by registering a crosswatch.exporters entry point — no change to the core.

5-minute quickstart

1. Describe what to watch in crosswatch.yml:

store: { type: duckdb, path: .crosswatch/crosswatch.duckdb }
telemetry: { exporters: [] }            # add 'phoenix' / 'langfuse' / 'otlp'
connectors:
  warehouse: { type: duckdb, path: data.duckdb }   # or postgres: { dsn: ${POSTGRES_DSN} }
checks:
  - { type: volume,         connector: warehouse, table: orders, min_rows: 1 }
  - { type: freshness,      connector: warehouse, table: orders, column: created_at, max_age: 24h }
  - { type: null_duplicate, connector: warehouse, table: orders, columns: [email], max_null_fraction: 0.0, key: [id] }

2. Run the checks (writes results to the store; non-zero exit on failure — drop it straight into CI):

crosswatch check -c crosswatch.yml

  ✓ volume [orders] pass: row_count=1240
  ✗ freshness [orders] fail: newest created_at = 2026-10-18T09:00:00+00:00
  ✓ null_duplicate [orders] pass: no nulls/duplicates beyond threshold

2/3 checks passed; 1 error-severity failure(s)

3. See it in the dashboard:

crosswatch serve -c crosswatch.yml      # http://127.0.0.1:8000

Or use the Python API

import crosswatch
from crosswatch.connectors.duckdb import DuckDBConnector
from crosswatch.data.checks.freshness import FreshnessCheck

conn = DuckDBConnector(path="data.duckdb")
result = conn.run_check(FreshnessCheck(table="orders", column="created_at", max_age="24h"))
print(result.status, result.message)

cfg = crosswatch.load("crosswatch.yml")   # same config, programmatic access

Configuration — YAML, JSON, or Python

The same config (store, telemetry exporters, connectors, checks, evals, alerts) can be expressed three ways. ${ENV_VAR} references are expanded in all of them.

YAML (crosswatch.yml)

store: { type: duckdb, path: .crosswatch/crosswatch.duckdb }
telemetry: { service_name: shop, exporters: [otlp] }   # or [], [phoenix], [langfuse]
connectors:
  warehouse: { type: postgres, dsn: ${POSTGRES_DSN} }
  events:    { type: clickhouse, host: localhost, port: 8123, user: crosswatch, password: ${CH_PW}, database: crosswatch }
checks:
  - { type: volume,         connector: warehouse, table: orders, min_rows: 1000 }
  - { type: freshness,      connector: warehouse, table: orders, column: created_at, max_age: 24h }
  - { type: null_duplicate, connector: warehouse, table: customers, columns: [email], max_null_fraction: 0.0, key: [id] }
  - { type: drift,          connector: events,    table: events_today, column: latency_ms, reference_table: events_baseline, threshold: 0.2, severity: warn }

crosswatch check -c crosswatch.yml       # exits non-zero on any error-severity failure

JSON (crosswatch.json)

JSON is valid input to the same loader (YAML is a JSON superset), so the CLI and crosswatch.load accept a .json file directly:

{
  "store": { "type": "duckdb", "path": ".crosswatch/crosswatch.duckdb" },
  "connectors": {
    "warehouse": { "type": "mysql", "host": "127.0.0.1", "port": 3306,
                   "user": "root", "password": "${MYSQL_PW}", "database": "shop" }
  },
  "checks": [
    { "type": "volume", "connector": "warehouse", "table": "orders", "min_rows": 1000 },
    { "type": "null_duplicate", "connector": "warehouse", "table": "customers",
      "columns": ["email"], "max_null_fraction": 0.0, "key": ["id"] }
  ]
}

crosswatch check -c crosswatch.json

Python

Load either file format, build from a dict, or construct the objects directly:

import json
import crosswatch
from crosswatch.config import Config
from crosswatch.data import run_checks

# (a) load a YAML or JSON file
cfg = crosswatch.load("crosswatch.yml")            # == Config.from_yaml("crosswatch.yml" | "crosswatch.json")

# (b) build from a Python dict (e.g. parsed JSON)
cfg = Config.from_dict(json.load(open("crosswatch.json")))

# run everything the config describes
store = cfg.build_store()
connectors = cfg.build_connectors()
checks = cfg.build_checks()
results = run_checks(checks, connectors, check_targets=[c.connector for c in cfg.checks])
store.write_check_results(results)
failed = [r for r in results if not r.ok]

# (c) or skip config entirely — construct a connector + checks by hand
from crosswatch.connectors.duckdb import DuckDBConnector
from crosswatch.data.checks.volume import VolumeCheck

conn = DuckDBConnector(path="data.duckdb")
r = conn.run_check(VolumeCheck(table="orders", min_rows=1000))
print(r.status, r.message, r.samples)   # samples pinpoint the offending rows

Examples

dbt — ingest run_results.json

dbt build                                  # produces target/run_results.json
python examples/dbt_quality.py target/run_results.json

Every dbt model/test outcome becomes a crosswatch CheckResult in the same store, dashboard, and alert path as native checks (see examples/dbt_quality.py).

RAG — trace an LLM pipeline with token/cost

from crosswatch.ai import configure, instrument_anthropic, observe

configure(service_name="rag", exporters=["phoenix"])   # OTLP to local Phoenix

@observe(kind="chain")
def answer(q: str) -> str:
    import anthropic
    client = instrument_anthropic(anthropic.Anthropic())   # auto-trace every call
    resp = client.messages.create(model="claude-opus-4-8", max_tokens=512,
                                  messages=[{"role": "user", "content": q}])
    return resp.content[0].text

Each call emits a GenAI-convention span with gen_ai.usage.input_tokens, output_tokens, and crosswatch.cost_usd. Full example: examples/rag_tracing.py.

MCP — trace tool calls + eval tool selection

from crosswatch.ai.mcp import TracedMCPClient, tool_selection_accuracy, MCPSessionRecord

client = TracedMCPClient(mcp_session)          # every call_tool is traced + recorded
await client.call_tool("get_weather", {"city": "Paris"})

# Score recorded sessions against ground-truth expected tools (CI regression gate):
score = tool_selection_accuracy(recorded_sessions, threshold=0.9)

Run crosswatch eval -c crosswatch.yml to gate CI on tool-selection accuracy. Full example: examples/mcp_eval.py.

Local demo services

docker compose up -d        # Postgres + MySQL + ClickHouse + Phoenix
export CROSSWATCH_TEST_PG_DSN=postgresql://crosswatch:crosswatch@localhost:5433/crosswatch
pytest                      # unit tests run on DuckDB/SQLite; -m integration hits the servers

Running in the cloud (Spark, Dataproc, Airflow, …)

crosswatch is a library + CLI, not a server — it runs wherever Python runs: an Airflow/Dagster/Prefect task, a Dataproc/EMR PySpark job, a Cloud Run or Kubernetes job, or CI. Two things change versus a laptop:

1. Use a shared, durable store (not the local DuckDB file). On ephemeral cloud workers, point the store at a managed Postgres so every job writes to one place the dashboard and crosswatch impact can read:

store: { type: postgres, path: ${CROSSWATCH_STORE_DSN} }   # e.g. Cloud SQL / RDS
telemetry: { exporters: [otlp] }                       # spans → your OTel collector

(DuckDB/Parquet = local default; Postgres = shared/cloud; a ClickHouse results store is on the roadmap. Traces always export over OTLP regardless of store.)

2. Checks run on the cluster via the connector. The checks are Ibis expressions, so they execute in the engine you point at — Spark included.

Spark / Dataproc / EMR / Databricks — attach to the SparkSession your job already has (nothing is pulled to the driver except the aggregates):

# inside a Dataproc / EMR PySpark job (or a Databricks notebook)
from crosswatch.connectors.spark import SparkConnector
from crosswatch.data.checks.null_duplicate import NullDuplicateCheck

conn = SparkConnector.from_session(spark)                      # your SparkSession
r = conn.run_check(NullDuplicateCheck(table="events", columns=["user_id"],
                                      max_null_fraction=0.0, key=["event_id"]))

or declaratively (pip install 'crosswatch[spark]'):

connectors:
  lake: { type: spark, master: "local[*]", app_name: crosswatch }   # omit master to attach

Cloud warehouses use the same interface — snowflake, bigquery, databricks connectors (install the matching extra, pass credentials/DSN). A typical pattern: run crosswatch check / crosswatch impact as a step in your orchestrator after the load job, and fail the task on non-zero exit.

Verified locally: DuckDB, SQLite, Postgres, MySQL, ClickHouse. Spark and the cloud warehouses are wired to the same Ibis backends but need a live cluster/account to exercise — treat them as integration-tested-by-you.

Architecture at a glance

connectors/   one adapter per source (Ibis): get_schema/profile/run_check/get_query_history
data/         native checks (Ibis exprs) + dbt ingestion (GE/Soda/Evidently providers planned)
ai/           @observe + OpenAI/Anthropic auto-instrumentation (OTel GenAI conventions)
ai/mcp/       trace tool calls + tool-selection / argument / task-success evals
evals/        evaluator plugin API, LLM-as-judge, MCP tool-use evals, CI mode
store/         DuckDB/Parquet default; ClickHouse/Postgres optional; OTLP exporter
alerts/        Slack / webhook channels, dispatch on failure or always
cli + server/  crosswatch check | eval | serve (FastAPI dashboard)

Everything pluggable is discovered via importlib.metadata entry points (groups: crosswatch.connectors, crosswatch.checks, crosswatch.check_providers, crosswatch.evaluators, crosswatch.alerts, crosswatch.exporters), so third parties ship compatible plugins without touching the core.

Known limitations / roadmap

crosswatch is alpha software. One honest list of what's actually implemented vs. scaffolded-but-not-wired, instead of caveats scattered across the doc:

Hit a gap not listed here? Please open an issue.

Development

uv venv && uv pip install -e '.[dev,server,postgres]'
pytest
ruff check src tests

License

Apache-2.0.