Telecom-native ML framework for Ray clusters.
PU Learning · Distributed Training · Batch Embedding & ONNX Serving
Tributo tackles three problems that no existing open-source framework solves end-to-end on Ray:
① PU Learning — train classifiers when you only have positive labels
Real-world fraud detection, churn prediction, and identity resolution all share the same data problem: confirmed positives are scarce, but unlabeled samples are plentiful and not truly negative. Tributo's PU Learning module runs nnPU/uPU loss training on Ray (single-worker; DDP planned for v2), estimates class priors automatically, and exports ONNX models for production serving — all with a single JSON config.
from tributo.training.pu_trainer import run_pu_training_from_json
# Single-worker PU training on Ray → ONNX export
result = run_pu_training_from_json("pu_config.json")
print(result["onnx_path"]) # Ready for serving② Behavioral Sequence Pre-training — model irregular time-series at scale (coming in v2.0)
Call records, SMS logs, and app usage traces have irregular intervals and multi-granularity patterns that standard Transformers cannot handle. Tributo's planned Temporal Transformer will encode continuous timestamps and support three self-supervised objectives (masked span, interval prediction, next-activity) for pre-training on billion-row event tables. See the roadmap for timeline.
# Coming in v2.0 — stay tuned
# from tributo.pretrain import TemporalTransformer, SequenceConfig
#
# config = SequenceConfig.from_json("pretrain_config.json")
# model = TemporalTransformer(config)
# model.pretrain(data_path="s3://bucket/events/*.parquet")③ Large-scale Embedding Pipeline — declarative ETL → distributed inference → Lance storage
A declarative pipeline that chains Daft (ETL) → Ray (distributed inference) → Lance (columnar storage) for embedding workflows. Turn raw tables into embedding indices and run similarity queries — all on CPU, no vector database required.
Note: Lance datasets are currently loaded into driver memory via
ds.to_table(), making this pipeline unsuitable for datasets exceeding available RAM. Streaming and distributed reads are planned for a future release.
from tributo.data.registry import get_connector
# Read from Parquet, write to Lance — both via unified connector API
reader = get_connector("parquet")
ds = reader.read(path="s3://bucket/data.parquet")
writer = get_connector("lance")
writer.write(ds, path="s3://bucket/index.lance") # Auto-detects vector columnspip install tributofrom tributo import TributoClient
client = TributoClient("http://127.0.0.1:8265")
job_id = client.submit(entrypoint="python my_script.py")
print(client.get_status(job_id))┌─────────────────────────────────────────────────────────┐
│ CLI / Python API │
│ tributo submit / embed / serve │
├──────────┬──────────┬───────────┬───────────────────────┤
│ training │embeddings│ serving │ inference │
│ XGBoost │ BGE ONNX │ ONNX HTTP │ XGBoost+ONNX batch │
│ PU Learn │ Daft ETL │ gRPC │ Ray Data │
│ DNN │ │ streaming │ │
│ Ray Train│ Ray Data │ Ray Serve │ Ray Data │
├──────────┴──────────┴───────────┴───────────────────────┤
│ data (Parquet / Lance / Iceberg, unified S3) │
├─────────────────────────────────────────────────────────┤
│ _common (runtime_env / io / logging) │
├─────────────────────────────────────────────────────────┤
│ Ray Cluster (≥ 2.9.0) │
└─────────────────────────────────────────────────────────┘
git clone https://github.com/jiangxt2/tributo.git
cd tributo
# Core install
uv sync
# With XGBoost training + ONNX export
uv sync --extra training
# With data connectors (Lance / Iceberg)
uv sync --extra data
# With text embeddings (BGE models)
uv sync --extra embeddings
# Development dependencies
uv sync --extra devXGBoost on Ray Train with S3 data sources and automatic ONNX export.
uv run python examples/xgboost_s3_training.pynnPU/uPU training with automatic class prior estimation and PU-specific metrics (single-worker; DDP planned for v2). See PU Learning guide.
Distributed embedding with BGE models, output to Lance or Parquet.
uv run tributo embed batch \
--input s3://bucket/data.parquet \
--output s3://bucket/embedded.lance \
--model bge-small-zh \
--concurrency 2
uv run tributo embed listRay Serve deployment for ONNX models with HTTP API.
uv run tributo serve start --model-path /path/to/model.onnx
uv run tributo serve status
uv run tributo serve stopXGBoost + ONNX distributed batch inference.
from tributo.inference.pipeline import InferenceConfig, run_batch_inference
config = InferenceConfig(
s3_input_path="s3://bucket/input.parquet",
s3_output_path="s3://bucket/output/",
model_uri="s3://bucket/model.onnx",
)
run_batch_inference(config)SSE-based streaming inference for LLMs on Ray Serve.
uv run tributo serve streaming start --model-path /path/to/model --tokenizer-path /path/to/tokenizer
uv run tributo serve streaming statusRandom search / BayesOpt with FIFO / ASHA / HyperBand schedulers.
uv run tributo tune run \
--trainer xgboost \
--config train_config.json \
--space search_space.json \
--output ./tune_results \
--num-samples 50 \
--search-alg bayesoptsrc/tributo/
├── __init__.py # Public API (TributoClient, JobConfig, exceptions)
├── job.py # TributoClient core
├── config.py # JobConfig (Pydantic models)
├── exceptions.py # Exception hierarchy
├── cli.py # CLI entry point (Click)
├── _common/ # Shared utilities (runtime_env, IO, logging, Serve)
├── data/ # Data connectors (Parquet / Lance / Iceberg)
├── training/ # XGBoost, DNN, PU Learning, Tune on Ray Train
├── embeddings/ # Text embeddings (batch + online)
├── serving/ # ONNX inference service (HTTP / gRPC / streaming)
├── inference/ # Distributed batch inference
├── registry/ # Model registry (MLflow integration)
└── util/ # @PublicAPI decorator, stability annotations
# Run tests
uv run pytest
# Format
uv run ruff check --fix src/ tests/
uv run ruff format src/ tests/
# Type checking
uv run mypy src/tributoApache 2.0 — see LICENSE.