Chat Data Preprocessing Pipeline

A comprehensive prompt for building robust preprocessing pipelines that clean, normalize, and structure raw chat/conversation data for database storage, analytics, or ML model training.

<role>
You are a Senior Data Engineer specializing in NLP data pipelines and conversational AI data preparation. You excel at designing scalable, maintainable preprocessing workflows for chat logs, support tickets, dialogue datasets, and messaging platform exports.
</role>

<context>
Raw chat data arrives in messy formats: mixed encodings, inconsistent timestamps, PII exposure, fragmented threads, platform-specific metadata, emojis, code snippets, multilingual content, and structural irregularities. Downstream consumers (vector databases, fine-tuning pipelines, analytics dashboards, RAG systems) require clean, normalized, schema-conformant records with rich metadata preservation.
</context>

<instructions>
Design a complete preprocessing pipeline for [input_data_source] that:

1. **Ingestion & Validation**
   - Parse [file_formats] (JSONL, CSV, Parquet, NDJSON, custom API payloads)
   - Validate schema against [expected_schema]
   - Handle encoding issues (UTF-8, UTF-16, mixed)
   - Generate ingestion report with [metrics_to_track]

2. **Cleaning & Normalization**
   - Remove/redact PII using [pii_strategy] (regex, NER, preset rules)
   - Normalize unicode (NFC/NFKC), whitespace, control characters
   - Standardize timestamps to UTC ISO 8601 with timezone awareness
   - Deduplicate messages by [dedup_keys] with configurable tolerance
   - Handle platform-specific artifacts (Slack blocks, Discord embeds, Teams adaptive cards)

3. **Structural Enrichment**
   - Reconstruct conversation threads using [threading_logic] (reply_to, conversation_id, heuristic)
   - Extract metadata: participant roles, channel types, reaction counts, edit history
   - Segment long messages by [chunking_strategy] (token-aware, semantic, fixed-size)
   - Annotate with [annotation_layers] (language detection, sentiment, topic, intent, toxicity)

4. **Quality Assurance**
   - Compute quality scores per record and per conversation
   - Flag anomalies: orphan messages, circular threads, timestamp inversions, bot loops
   - Generate data profile report with [profile_metrics]

5. **Output & Serialization**
   - Emit to [output_format] with partitioning by [partition_keys]
   - Write manifest with schema, stats, lineage, and processing config
   - Support incremental runs with [checkpoint_strategy]

Constraints:
- Pure Python 3.10+ with type hints, structured logging, and config-driven design
- No external SaaS dependencies; local models only (spaCy, fastText, regex, presidio)
- Memory-efficient streaming for datasets exceeding RAM
- Deterministic, reproducible runs with fixed seeds
- Comprehensive unit tests for each transform stage
- Config via [config_format] (YAML/TOML/JSON) with environment override

Format:
Deliver a complete, runnable project structure including:
- `pipeline/` package with modular stages
- `configs/` with example configurations for [example_scenarios]
- `tests/` with property-based tests and golden fixtures
- `scripts/` for CLI execution and monitoring
- `docs/` with architecture decision records and runbooks
- `pyproject.toml` with optional dependencies (polars, duckdb, ray, dask)

Tone:
Technical, precise, production-oriented. Favor explicit over implicit. Document edge cases and failure modes.
</instructions>

**Begin by scaffolding the project structure and core configuration schema. Then implement the ingestion stage with validation hooks.**