Spaces:
Sleeping
Sleeping
File size: 5,515 Bytes
f0ad703 349c949 f0ad703 782a588 f0ad703 782a588 f0ad703 782a588 f0ad703 782a588 f0ad703 782a588 f0ad703 782a588 f0ad703 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | ---
title: Dataclean Env
emoji: 🧹
colorFrom: blue
colorTo: green
sdk: docker
pinned: false
tags:
- openenv
- data-cleaning
- etl
- real-world
- tabular
- data-drift
- streaming
---
# DataClean OpenEnv
A **real-world data cleaning and ETL environment** for AI agent RL training, built on the [OpenEnv](https://meta-pytorch.org/OpenEnv/) framework by Meta-PyTorch and Hugging Face.
Agents learn to fix messy tabular data — filling nulls, normalising inconsistent values, removing duplicates, filtering outliers, performing multi-table ETL joins, and handling **live data drift** (novel: fresh dirty rows injected mid-episode every 5 steps).
---
## Why This Environment
Data cleaning is one of the most common, time-consuming tasks in real data engineering. Every company with a data pipeline does this daily. Training an RL agent on this task has immediate real-world value — unlike game-based environments.
**What makes this different from other OpenEnv submissions:**
- Task 4 (Data Drift) is genuinely novel — no existing OpenEnv environment simulates live streaming row injection mid-episode
- Parallel baseline using `concurrent.futures.ThreadPoolExecutor` — all 4 tasks run simultaneously
- Deterministic graders using seeded dataset generation — perfectly reproducible scores
- Dense reward signal: reward = grader(new_state) − grader(prev_state) at every step
---
## Tasks
| Task | Difficulty | Description | Max Steps |
|------|-----------|-------------|-----------|
| `task1` | Easy | Fix nulls + dtypes in 50-row customer CSV | 10 |
| `task2` | Medium | Dedup + normalize strings + fix dates + fill nulls | 20 |
| `task3` | Hard | Multi-table merge + outlier removal + derived column | 30 |
| `task4_data_drift` | **Expert** | Live streaming table — 7 dirty rows injected every 5 steps | 40 |
---
## Action Space
Every action is a JSON object with an `operation` field:
```json
{"operation": "fill_nulls", "column": "age", "strategy": "median"}
{"operation": "cast_column", "column": "age", "dtype": "int"}
{"operation": "remove_duplicates"}
{"operation": "normalize_values", "column": "country", "method": "upper"}
{"operation": "cast_column", "column": "order_date", "dtype": "datetime"}
{"operation": "filter_outliers", "column": "amount", "method": "iqr", "threshold": 1.5}
{"operation": "merge_tables", "left_table": "orders", "right_table": "customers", "on": "customer_id", "output_table": "merged"}
{"operation": "add_derived_column", "column_name": "order_year", "source_column": "order_date", "transform": "year_from_date", "table_name": "merged"}
{"operation": "submit"}
```
## Observation Space
After each `reset()` / `step()` the agent receives:
- `task_id`, `task_description`, `step_count`, `max_steps`, `message`
- `tables` — dict of `{table_name → JSON string of df.head(10)}`
- `column_dtypes`, `null_counts`, `duplicate_count`, `row_count`
- `schema_errors` — list of detected problems to guide the agent
- `reward`, `done`, `partial_score` — RL signals
---
## Reward Function
```
step_reward = grader(current_state) − grader(previous_state) # dense delta signal
invalid_op = −0.02 # bad operation penalty
terminal = final grader score on submit or max_steps
```
Partial credit per sub-dimension. Score range: `[0.0, 1.0]`.
---
## Baseline Scores (llama-3.3-70b-versatile, seed=42, parallel run)
| Task | Score | Time |
|------|-------|------|
| task1 (easy) | 1.0000 | 11.8s |
| task2 (medium) | 1.0000 | 34.1s |
| task3 (hard) | 0.8000 | 22.4s |
| task4_data_drift (expert) | 0.9297 | 26.6s |
| **mean** | **0.9324** | **34.3s wall** |
All 4 tasks run in parallel — wall time = slowest task, not sum.
---
## Setup & Local Run
```bash
pip install -r requirements.txt
uvicorn server.app:app --host 0.0.0.0 --port 7860 --reload
```
```bash
curl http://localhost:7860/health
curl http://localhost:7860/tasks
```
```bash
export OPENAI_API_KEY=your_key
export OPENAI_BASE_URL=https://api.groq.com/openai/v1
export BASELINE_MODEL=llama-3.3-70b-versatile
python baseline.py
```
## Docker
```bash
docker build -t dataclean-env .
docker run -p 7860:7860 -e OPENAI_API_KEY=your_key dataclean-env
```
---
## API Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/reset` | Start new episode |
| POST | `/step` | Execute one cleaning operation |
| GET | `/state` | Current episode metadata |
| GET | `/tasks` | All tasks + action schema |
| GET | `/grader` | Score current episode state |
| GET | `/baseline` | Run baseline agent on all tasks |
| GET | `/health` | Liveness probe |
| GET | `/docs` | Interactive Swagger UI |
---
## Project Structure
```
dataCleaningProject/
├── server/
│ ├── app.py # FastAPI server — all endpoints
│ ├── environment.py # Core env logic — reset/step/state + drift injection
│ ├── graders.py # Deterministic scoring for all 4 tasks
│ └── dataset_factory.py # Seeded dirty+expected dataset generation + drift batches
├── models.py # Pydantic Action + Observation models
├── baseline.py # Parallel baseline (ThreadPoolExecutor, 4 tasks at once)
├── client.py # HTTP client
├── openenv.yaml # OpenEnv manifest
├── Dockerfile # Port 7860, Python 3.11-slim
└── requirements.txt
```
|