---
title: "Golden Dataset"
description: "Golden dataset lifecycle patterns for curation, versioning, quality validation, and CI integration. Use when building evaluation datasets, managing dataset versions, validating quality scores, or integrating golden tests into pipelines."
canonical: "https://orchestkit.yonyon.ai/docs/reference/skills/golden-dataset"
---
# Golden Dataset
Golden dataset lifecycle patterns for curation, versioning, quality validation, and CI integration. Use when building evaluation datasets, managing dataset versions, validating quality scores, or integrating golden tests into pipelines.
Reference medium
> **Not directly invocable** — no slash command and no model auto-selection. An agent loads it explicitly via `Read()`.
> **Golden Dataset** Golden dataset lifecycle patterns for curation, versioning, quality validation, and CI integration. Use when building evaluation datasets, managing dataset versions, validating quality scores, or integrating golden tests into pipelines.
# Golden Dataset
Comprehensive patterns for building, managing, and validating golden datasets for AI/ML evaluation. Each category has individual rule files in `rules/` loaded on-demand.
## Quick Reference
| Category | Rules | Impact | When to Use |
| -------- | ----- | ------ | ----------- |
| [Curation](#curation) | 2 | HIGH | Content collection, annotation pipelines |
| [Management](#management) | 2 | HIGH | Versioning, backup/restore |
| [Validation](#validation) | 1 | CRITICAL | Regression testing |
| [Add Workflow](#add-workflow) | 1 | HIGH | 9-phase curation, quality scoring, bias detection, silver-to-gold |
Total: 6 rules across 4 categories. House thresholds and scars: `references/ork-delta.md`.
## Curation
Content collection, multi-agent annotation, and diversity analysis for golden datasets.
| Rule | File | Key Pattern |
| ---- | ---- | ----------- |
| Collection | `rules/curation-collection.md` | Content type classification, quality thresholds, duplicate prevention |
| Annotation | `rules/curation-annotation.md` | Multi-agent pipeline, consensus aggregation, Langfuse tracing |
Difficulty ladder, coverage floors, and duplicate thresholds: `references/ork-delta.md`.
## Management
Versioning, storage, and CI/CD automation for golden datasets.
| Rule | File | Key Pattern |
| ---- | ---- | ----------- |
| Versioning | `rules/management-versioning.md` | JSON backup format, embedding regeneration, disaster recovery |
| Storage | `rules/management-storage.md` | Backup strategies, URL contract, data integrity checks |
CI automation for backups is upstream's job; see "Upstream coverage" below.
## Validation
Quality scoring, drift detection, and regression testing for golden datasets.
| Rule | File | Key Pattern |
| ---- | ---- | ----------- |
| Regression | `rules/validation-regression.md` | Difficulty distribution, pre-commit hooks, full dataset validation |
Schema validation and duplicate detection are upstream's job (see "Upstream coverage"
below); the house thresholds they must enforce live in `references/ork-delta.md`.
## Add Workflow
Structured workflow for adding new documents to the golden dataset.
| Rule | File | Key Pattern |
| ---- | ---- | ----------- |
| Add Document | `rules/curation-add-workflow.md` | 9-phase curation, parallel quality analysis, bias detection |
## Quick Start Example
```python
async def validate_before_add(document: dict, source_url_map: dict) -> dict:
"""Pre-addition validation for golden dataset entries."""
errors = []
# 1. URL contract check
if "placeholder" in document.get("source_url", ""):
errors.append("URL must be canonical, not a placeholder")
# 2. Content quality
if len(document.get("title", "")) < 10:
errors.append("Title too short (min 10 chars)")
# 3. Tag requirements
if len(document.get("tags", [])) < 2:
errors.append("At least 2 domain tags required")
return {"valid": len(errors) == 0, "errors": errors}
```
## Key Decisions
| Decision | Recommendation |
| -------- | -------------- |
| Backup format | JSON (version controlled, portable) |
| Embedding storage | Exclude from backup (regenerate on restore) |
| Quality threshold | >= 0.70 quality score for inclusion |
| Confidence threshold | >= 0.65 for auto-include |
| Duplicate threshold | >= 0.90 similarity blocks, >= 0.85 warns |
| Min tags per entry | 2 domain tags |
| Min test queries | 3 per document |
| Difficulty balance | Trivial 3, Easy 3, Medium 5, Hard 3 minimum |
| CI frequency | Weekly automated backup (Sunday 2am UTC) |
## Common Mistakes
1. Using placeholder URLs instead of canonical source URLs
2. Skipping embedding regeneration after restore
3. Not validating referential integrity between documents and queries
4. Over-indexing on articles (neglecting tutorials, research papers)
5. Missing difficulty distribution balance in test queries
6. Not running verification after backup/restore operations
7. Testing restore procedures in production instead of staging
8. Committing SQL dumps instead of JSON (not version-control friendly)
## Running a dataset as an experiment
Curating a dataset is half the job; the other half is running something against it and scoring the
result. Both Langfuse SDKs ship a runner, and their shapes differ.
**Python (SDK 4.x):** see `monitoring-observability/references/experiments-api.md`.
**JS/TS (SDK 5.x):** `@langfuse/client` exposes the runner directly on a fetched dataset.
```typescript
import { LangfuseClient } from "@langfuse/client";
const langfuse = new LangfuseClient();
const dataset = await langfuse.dataset.get("my-evaluation-dataset");
const result = await dataset.runExperiment({
name: "Retrieval quality",
task: myTask, // (params) => Promise
evaluators: [myEvaluator], // per-item: (params) => Promise
});
```
| Type | Scores | Use for |
|---|---|---|
| `Evaluator` | one item | Per-example quality (faithfulness, relevance) |
| `RunEvaluator` | the whole run | Aggregate assertions — pass rate, mean score, regression checks |
| `Evaluation` | — | `\{ name, value, comment?, metadata?, dataType?, configId? \}` |
A per-item `Evaluator` cannot see the other items, so anything comparative belongs in a
`RunEvaluator`. `createEvaluatorFromAutoevals` wraps an autoevals scorer instead of hand-writing
one, and `RegressionError` is thrown when a run regresses against a configured baseline — catch it
to fail CI on a quality drop rather than only on an exception.
Full JS surface: `monitoring-observability/references/langfuse-js-v5.md`.
## Evaluations
See `test-cases.json` for 9 test cases across all categories.
## Upstream coverage (do not restate)
| Topic | First-party source |
| ----- | ------------------ |
| Dataset schema validation (JSON Schema, field constraints) | https://json-schema.org and https://zod.dev |
| Duplicate detection via embeddings, cosine similarity | https://github.com/pgvector/pgvector |
| Scheduled backup automation (cron workflows, commit bots) | https://docs.github.com/actions/using-workflows/events-that-trigger-workflows#schedule |
| Dataset runs, experiment scoring, annotation queues | https://langfuse.com/docs/datasets |
| Backup and restore mechanics for postgres datasets | https://www.postgresql.org/docs/current/backup.html |
House thresholds these must enforce: `references/ork-delta.md`.
## Related Skills
- `ork:rag-retrieval` - Retrieval evaluation using golden dataset
- `ork:monitoring-observability` - Langfuse tracing patterns for curation workflows
- `ork:testing-llm` - Evaluation harnesses that consume golden datasets
- `ork:testing-unit` - Unit testing patterns and strategies
## Capability Details
### curation
**Keywords:** golden dataset, curation, content collection, annotation, quality criteria
**Solves:**
- Classify document content types for golden dataset
- Run multi-agent quality analysis pipelines
- Generate test queries for new documents
### management
**Keywords:** golden dataset, backup, restore, versioning, disaster recovery
**Solves:**
- Backup and restore golden datasets with JSON
- Regenerate embeddings after restore
- Automate backups with CI/CD
### validation
**Keywords:** golden dataset, validation, schema, duplicate detection, quality metrics
**Solves:**
- Validate entries against document schema
- Detect duplicate or near-duplicate entries
- Analyze dataset coverage and distribution gaps
---
## Rules (6)
### Follow the full curation pipeline when adding entries to the golden dataset — HIGH
## Add to Golden Dataset Workflow
Multi-agent curation pipeline with quality scoring, bias detection, and silver-to-gold promotion.
**Incorrect — adding documents without validation:**
```python
# No quality check, no bias detection, no dedup
dataset.append({"url": url, "content": content})
```
**Correct — 9-phase curation workflow:**
**Phase 1-2: Input and extraction**
```python
# Detect content type and extract structure
content_type = classify(url) # article, tutorial, documentation, research_paper
structured = extract(url) # title, sections, code blocks, key terms, metadata
```
**Phase 3: Parallel quality analysis (4 agents)**
```python
# Launch ALL quality agents in parallel
# Agent 1: Accuracy, coherence, depth, relevance scores
# Agent 2: Keyword directness, difficulty level
# Agent 3: Domain tags, skill level classification
# Agent 4: Test query generation (direct, paraphrased, multi-hop)
```
**Phase 4: Quality scoring formula**
```python
quality_score = (
accuracy * 0.25 +
coherence * 0.20 +
depth * 0.25 +
relevance * 0.30
)
```
**Phase 5-6: Bias detection and diversity check**
| Bias Score | Action |
|------------|--------|
| 0-2 | Proceed normally |
| 3-5 | Add disclaimer |
| 6-8 | Require user review |
| 9-10 | Recommend against inclusion |
**Phase 7-8: Validation and classification**
| Status | Quality Score | Action |
|--------|--------------|--------|
| GOLD | >= 0.75 | Add to main dataset |
| SILVER | 0.55-0.74 | Add to silver tier, track |
| REJECT | < 0.55 | Do not add |
**Promotion criteria:** 7+ days in silver, quality >= 0.75, no negative feedback.
**Phase 9: Version tracking**
```json
{
"version": "1.2.3",
"change_type": "ADD",
"document_id": "doc-123",
"quality_score": 0.82,
"rollback_available": true
}
```
| Update Type | Version Bump |
|-------------|--------------|
| Add/Update document | Patch (0.0.X) |
| Remove document | Minor (0.X.0) |
| Schema change | Major (X.0.0) |
**Key rules:**
- Never skip the quality analysis phase — it prevents low-quality entries from degrading evaluations
- Run bias detection on every addition — dataset contamination is hard to reverse
- Use the two-tier system (silver/gold) to let borderline documents prove themselves
- Always validate URL is canonical (not a placeholder) and check for >80% duplicate similarity
- Minimum requirements: 2+ domain tags, 3+ test queries per document
### Use multi-agent annotation for consistent and thorough curation quality decisions — HIGH
## Multi-Agent Annotation
Multi-agent analysis pipeline with consensus aggregation for golden dataset curation.
**Pipeline Architecture:**
```
INPUT: URL/Content
|
v
+------------------+
| FETCH AGENT | WebFetch or file read
| (sequential) | Extract structure, detect type
+--------+---------+
|
v
+-----------------------------------------------+
| PARALLEL ANALYSIS AGENTS |
| +----------+ +----------+ +--------+ +------+ |
| | Quality | |Difficulty| | Domain | |Query | |
| |Evaluator | |Classifier| | Tagger | |Gen | |
| +----+-----+ +----+-----+ +---+----+ +--+---+ |
+-------+------------+-----------+---------+-----+
|
v
+-----------------------------------------------+
| CONSENSUS AGGREGATOR |
| - Weighted quality score |
| - Confidence level (agent agreement) |
| - Final recommendation: include/review/exclude|
+--------+--------------------------------------+
|
v
+------------------+
| USER APPROVAL | Show scores, get confirmation
+------------------+
```
**Quality Evaluator Agent:**
```python
Agent(
subagent_type="ork:code-quality-reviewer",
prompt="""GOLDEN DATASET QUALITY EVALUATION
Evaluate this content for golden dataset inclusion:
Content: {content_preview}
Source: {source_url}
Type: {content_type}
Score these dimensions (0.0-1.0):
1. ACCURACY (weight 0.25)
- Technical correctness
- Code validity
- Up-to-date information
2. COHERENCE (weight 0.20)
- Logical structure
- Clear flow
- Consistent terminology
3. DEPTH (weight 0.25)
- Comprehensive coverage
- Edge cases mentioned
- Appropriate detail level
4. RELEVANCE (weight 0.30)
- Alignment with AI/ML, backend, frontend, DevOps
- Practical applicability
- Technical value
Output JSON:
{
"accuracy": {"score": 0.X, "rationale": "..."},
"coherence": {"score": 0.X, "rationale": "..."},
"depth": {"score": 0.X, "rationale": "..."},
"relevance": {"score": 0.X, "rationale": "..."},
"weighted_total": 0.X,
"recommendation": "include|review|exclude"
}
""",
run_in_background=True
)
```
**Consensus Aggregation Logic:**
```python
from dataclasses import dataclass
from typing import Literal
@dataclass
class CurationConsensus:
"""Aggregated result from multi-agent analysis."""
quality_score: float # Weighted average (0-1)
confidence: float # Agent agreement (0-1)
decision: Literal["include", "review", "exclude"]
content_type: str
difficulty: str
tags: list[str]
suggested_queries: list[dict]
warnings: list[str]
def aggregate_results(
quality_result: dict,
difficulty_result: dict,
domain_result: dict,
query_result: dict,
) -> CurationConsensus:
"""Aggregate multi-agent results into consensus."""
# Calculate weighted quality score
q = quality_result
quality_score = (
q["accuracy"]["score"] * 0.25 +
q["coherence"]["score"] * 0.20 +
q["depth"]["score"] * 0.25 +
q["relevance"]["score"] * 0.30
)
# Calculate confidence (variance-based)
scores = [
q["accuracy"]["score"],
q["coherence"]["score"],
q["depth"]["score"],
q["relevance"]["score"],
]
variance = sum((s - quality_score)**2 for s in scores) / len(scores)
confidence = 1.0 - min(variance * 4, 1.0)
# Decision thresholds
if quality_score >= 0.75 and confidence >= 0.7:
decision = "include"
elif quality_score >= 0.55:
decision = "review"
else:
decision = "exclude"
return CurationConsensus(
quality_score=quality_score,
confidence=confidence,
decision=decision,
content_type=difficulty_result.get("content_type", "article"),
difficulty=difficulty_result["difficulty"],
tags=domain_result["tags"],
suggested_queries=query_result["queries"],
warnings=[],
)
```
**Langfuse Integration (v3):**
```python
from langfuse import observe, get_client
@observe(name="golden-dataset-curation")
async def curate_with_tracing(url: str, doc_id: str, consensus: CurationConsensus) -> dict:
"""Trace curation decisions to Langfuse for audit trail."""
get_client().update_current_trace(
metadata={"source_url": url, "document_id": doc_id}
)
# Log individual dimension scores against the current trace
lf = get_client()
trace_id = lf.get_current_trace_id()
lf.score(trace_id=trace_id, name="accuracy", value=0.85)
lf.score(trace_id=trace_id, name="coherence", value=0.90)
lf.score(trace_id=trace_id, name="depth", value=0.78)
lf.score(trace_id=trace_id, name="relevance", value=0.92)
# Final aggregated score
lf.score(trace_id=trace_id, name="quality_total", value=consensus.quality_score)
get_client().update_current_observation(
metadata={"curation_decision": consensus.decision}
)
return {"decision": consensus.decision, "score": consensus.quality_score}
```
**Incorrect — Sequential agent execution:**
```python
# Sequential - 4x slower
quality_result = await analyze_quality(content)
difficulty_result = await analyze_difficulty(content)
domain_result = await analyze_domain(content)
query_result = await generate_queries(content)
```
**Correct — Parallel agent execution:**
```python
# Parallel - all agents run concurrently
quality_task = Agent(subagent_type="ork:code-quality-reviewer", prompt=quality_prompt, run_in_background=True)
difficulty_task = Agent(subagent_type="", prompt=difficulty_prompt, run_in_background=True)
domain_task = Agent(subagent_type="", prompt=domain_prompt, run_in_background=True)
query_task = Agent(subagent_type="", prompt=query_prompt, run_in_background=True)
# Wait for all results
results = await gather_task_results([quality_task, difficulty_task, domain_task, query_task])
```
**Key rules:**
- Run all 4 analysis agents in parallel for throughput
- Use weighted scoring (accuracy 0.25, coherence 0.20, depth 0.25, relevance 0.30)
- Require user approval before final inclusion
- Log all scores to Langfuse for audit trail
### Apply systematic collection criteria to maintain consistent golden dataset quality — HIGH
## Content Collection
Systematic patterns for collecting and classifying content for golden dataset inclusion.
**Content Type Classification:**
| Type | Description | Quality Focus |
|------|-------------|---------------|
| `article` | Technical articles, blog posts | Depth, accuracy, actionability |
| `tutorial` | Step-by-step guides | Completeness, clarity, code quality |
| `research_paper` | Academic papers, whitepapers | Rigor, citations, methodology |
| `documentation` | API docs, reference materials | Accuracy, completeness, examples |
| `video_transcript` | Transcribed video content | Structure, coherence, key points |
| `code_repository` | README, code analysis | Code quality, documentation |
**Classification Decision Tree:**
```python
def classify_content_type(content: str, source_url: str) -> str:
"""Classify content type based on structure and source."""
# URL-based hints
if "arxiv.org" in source_url or "papers" in source_url:
return "research_paper"
if "docs." in source_url or "/api/" in source_url:
return "documentation"
if "github.com" in source_url:
return "code_repository"
# Content-based analysis
if has_step_by_step_structure(content):
return "tutorial"
if has_academic_structure(content): # Abstract, methodology, results
return "research_paper"
# Default
return "article"
```
**Quality Thresholds:**
```yaml
# Recommended thresholds for golden dataset inclusion
minimum_quality_score: 0.70
minimum_confidence: 0.65
required_tags: 2 # At least 2 domain tags
required_queries: 3 # At least 3 test queries
```
**Quality Dimensions:**
| Dimension | Weight | Perfect | Acceptable | Failing |
|-----------|--------|---------|------------|---------|
| **Accuracy** | 0.25 | 0.95-1.0 | 0.70-0.94 | <0.70 |
| **Coherence** | 0.20 | 0.90-1.0 | 0.60-0.89 | <0.60 |
| **Depth** | 0.25 | 0.90-1.0 | 0.55-0.89 | <0.55 |
| **Relevance** | 0.30 | 0.95-1.0 | 0.70-0.94 | <0.70 |
**Decision Thresholds:**
| Quality Score | Confidence | Decision |
|---------------|------------|----------|
| >= 0.75 | >= 0.70 | **include** |
| >= 0.55 | any | **review** |
| < 0.55 | any | **exclude** |
**Duplicate Prevention Checklist:**
1. Check URL against existing `source_url_map.json`
2. Run semantic similarity against existing document embeddings
3. Warn if >80% similar to existing document
**Provenance Tracking -- always record:**
- Source URL (canonical)
- Curation date
- Agent scores (for audit trail)
- Langfuse trace ID
**Incorrect — Placeholder URL:**
```python
# Missing real source URL
analysis = Analysis(
url="https://orchestkit.dev/placeholder/123",
content_type="article",
)
```
**Correct — Real canonical URL:**
```python
# Real source for re-fetching and validation
analysis = Analysis(
url="https://docs.python.org/3/library/asyncio.html",
content_type="documentation",
)
```
**Key rules:**
- Never use placeholder URLs -- always store real canonical source URLs
- Require minimum 2 domain tags and 3 test queries per entry
- Score all 4 quality dimensions before inclusion decision
- Track provenance for full audit trail
### Choose the right backup strategy and URL contract for golden dataset storage — HIGH
## Storage Patterns
Backup strategies, URL contract enforcement, and data integrity checks.
**Backup Strategy Comparison:**
| Strategy | Version Control | Restore Speed | Portability | Inspection |
|----------|-----------------|---------------|-------------|------------|
| **JSON** (recommended) | Yes | Slower (regen embeddings) | High | Easy |
| **SQL Dump** | No (binary) | Fast | DB-version dependent | Hard |
**The URL Contract:**
Golden dataset analyses MUST store **real canonical URLs**, not placeholders.
```python
# WRONG - Placeholder URL (breaks restore)
analysis.url = "https://orchestkit.dev/placeholder/123"
# CORRECT - Real canonical URL (enables re-fetch if needed)
analysis.url = "https://docs.python.org/3/library/asyncio.html"
```
**Why this matters:**
- Enables re-fetching content if embeddings need regeneration
- Allows validation that source content hasn't changed
- Provides audit trail for data provenance
**URL Validation:**
```python
FORBIDDEN_URL_PATTERNS = [
"orchestkit.dev",
"placeholder",
"example.com",
"localhost",
"127.0.0.1",
]
def validate_url(url: str) -> tuple[bool, str]:
"""Validate URL is not a placeholder."""
for pattern in FORBIDDEN_URL_PATTERNS:
if pattern in url.lower():
return False, f"URL contains forbidden pattern: {pattern}"
if not url.startswith("https://"):
if not url.startswith("http://arxiv.org"): # arXiv redirects
return False, "URL must use HTTPS"
return True, "OK"
```
**Data Integrity Checks:**
| Check | Error/Warning | Description |
|-------|---------------|-------------|
| Count mismatch | Error | Analysis/chunk count differs from metadata |
| Placeholder URLs | Error | URLs containing orchestkit.dev or placeholder |
| Missing embeddings | Error | Chunks without embeddings after restore |
| Orphaned chunks | Warning | Chunks with no parent analysis |
**Verification Implementation:**
```python
async def verify_golden_dataset() -> dict:
"""Verify golden dataset integrity."""
errors = []
warnings = []
async with get_session() as session:
# 1. Check counts
analysis_count = await session.scalar(select(func.count(Analysis.id)))
chunk_count = await session.scalar(select(func.count(Chunk.id)))
expected = load_metadata()
if analysis_count != expected["total_analyses"]:
errors.append(f"Analysis count mismatch: {analysis_count} vs {expected['total_analyses']}")
# 2. Check URL contract
query = select(Analysis).where(
Analysis.url.like("%orchestkit.dev%") |
Analysis.url.like("%placeholder%")
)
result = await session.execute(query)
invalid_urls = result.scalars().all()
if invalid_urls:
errors.append(f"Found {len(invalid_urls)} analyses with placeholder URLs")
# 3. Check embeddings exist
query = select(Chunk).where(Chunk.embedding.is_(None))
result = await session.execute(query)
missing_embeddings = result.scalars().all()
if missing_embeddings:
errors.append(f"Found {len(missing_embeddings)} chunks without embeddings")
# 4. Check orphaned chunks
query = select(Chunk).outerjoin(Analysis).where(Analysis.id.is_(None))
result = await session.execute(query)
orphaned = result.scalars().all()
if orphaned:
warnings.append(f"Found {len(orphaned)} orphaned chunks")
return {"valid": len(errors) == 0, "errors": errors, "warnings": warnings}
```
**Best Practices:**
1. **Version control backups** -- commit JSON to git for history and diffs
2. **Validate before deployment** -- run verify before production changes
3. **Test restore in staging** -- never test restore in production first
4. **Document changes** -- track additions/removals in metadata
**Incorrect — Missing URL validation:**
```python
# No URL contract enforcement
analysis.url = url # Could be placeholder
session.add(analysis)
await session.commit()
```
**Correct — Enforcing URL contract:**
```python
# Validate before saving
valid, msg = validate_url(url)
if not valid:
raise ValueError(f"Invalid URL: {msg}")
analysis.url = url # Guaranteed to be real canonical URL
session.add(analysis)
await session.commit()
```
**Key rules:**
- Always use JSON backup for version control and portability
- Never store placeholder URLs -- enforce the URL contract
- Run all 4 integrity checks (counts, URLs, embeddings, orphans) after every restore
- SQL dumps for local snapshots only, not version control
### Version golden datasets for reproducible evaluation across environments and recovery — HIGH
## Dataset Versioning
JSON backup format, embedding regeneration, and disaster recovery patterns.
**Backup Format:**
```json
{
"version": "1.0",
"created_at": "2025-12-19T10:30:00Z",
"metadata": {
"total_analyses": 98,
"total_chunks": 415,
"total_artifacts": 98
},
"analyses": [
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"url": "https://docs.python.org/3/library/asyncio.html",
"content_type": "documentation",
"status": "completed",
"created_at": "2025-11-15T08:20:00Z",
"chunks": [
{
"id": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"content": "asyncio is a library...",
"section_title": "Introduction to asyncio"
}
]
}
]
}
```
**Key design decisions:**
- Embeddings excluded (regenerate on restore with current model)
- Nested structure (analyses -> chunks -> artifacts)
- Metadata for validation
- ISO timestamps for reproducibility
**Restore with Embedding Regeneration:**
```python
async def restore_golden_dataset(replace: bool = False):
"""Restore golden dataset from JSON backup."""
with open(BACKUP_FILE) as f:
backup_data = json.load(f)
async with get_session() as session:
if replace:
await session.execute(delete(Chunk))
await session.execute(delete(Artifact))
await session.execute(delete(Analysis))
await session.commit()
from app.shared.services.embeddings import embed_text
for analysis_data in backup_data["analyses"]:
analysis = Analysis(
id=UUID(analysis_data["id"]),
url=analysis_data["url"],
)
session.add(analysis)
for chunk_data in analysis_data["chunks"]:
# Regenerate embedding using CURRENT model
embedding = await embed_text(chunk_data["content"])
chunk = Chunk(
id=UUID(chunk_data["id"]),
analysis_id=analysis.id,
content=chunk_data["content"],
embedding=embedding, # Freshly generated!
)
session.add(chunk)
if idx % 10 == 0:
await session.commit()
await session.commit()
```
**Why regenerate embeddings?**
- Embedding models improve over time (Voyage AI v1 -> v2)
- Ensures consistency with current production model
- Smaller backup files (exclude large vectors)
**Disaster Recovery Scenarios:**
| Scenario | Steps |
|----------|-------|
| Accidental deletion | `restore --replace` -> `verify` -> run tests |
| Migration failure | `alembic downgrade -1` -> `restore --replace` -> fix migration |
| New environment | Clone repo -> setup DB -> `restore` -> run tests |
**CLI Commands:**
```bash
cd backend
# Backup golden dataset
poetry run python scripts/backup_golden_dataset.py backup
# Verify backup integrity
poetry run python scripts/backup_golden_dataset.py verify
# Restore from backup (WARNING: Deletes existing data)
poetry run python scripts/backup_golden_dataset.py restore --replace
# Restore without deleting (adds to existing)
poetry run python scripts/backup_golden_dataset.py restore
```
**Incorrect — Storing embeddings in backup:**
```python
# Embedding vectors bloat backup file
backup_data = {
"chunks": [{
"content": "...",
"embedding": [0.123, 0.456, ...], # 1024 floats!
}]
}
```
**Correct — Regenerate embeddings on restore:**
```python
# Exclude embeddings from backup
backup_data = {
"chunks": [{
"content": "...",
# No embedding field
}]
}
# Regenerate during restore
embedding = await embed_text(chunk_data["content"])
chunk.embedding = embedding # Fresh with current model
```
**Key rules:**
- Always regenerate embeddings on restore -- never store them in backup
- Commit backups every 10 analyses to avoid huge transactions
- Verify counts match metadata after every restore
- Test restore procedures in staging before production
### Run regression tests and enforce difficulty distribution to maintain evaluation reliability — CRITICAL
## Regression Testing
Difficulty distribution enforcement, pre-commit hooks, and full dataset validation.
**Difficulty Distribution Validation:**
```python
def validate_difficulty_distribution(queries: list[dict]) -> list[str]:
"""Ensure balanced difficulty distribution."""
warnings = []
# Count by difficulty
distribution = {}
for query in queries:
diff = query.get("difficulty", "unknown")
distribution[diff] = distribution.get(diff, 0) + 1
# Minimum requirements
requirements = {
"trivial": 3,
"easy": 3,
"medium": 5, # Most common real-world case
"hard": 3,
}
for level, min_count in requirements.items():
actual = distribution.get(level, 0)
if actual < min_count:
warnings.append(
f"Insufficient {level} queries: {actual}/{min_count}"
)
return warnings
```
**Query Schema:**
```json
{
"type": "object",
"required": ["id", "query", "difficulty", "expected_chunks", "min_score"],
"properties": {
"id": {"type": "string", "pattern": "^q-[a-z0-9-]+$"},
"query": {"type": "string", "minLength": 5, "maxLength": 500},
"modes": {"type": "array", "items": {"enum": ["semantic", "keyword", "hybrid"]}},
"category": {"enum": ["specific", "broad", "negative", "edge", "coarse-to-fine"]},
"difficulty": {"enum": ["trivial", "easy", "medium", "hard", "adversarial"]},
"expected_chunks": {"type": "array", "items": {"type": "string"}, "minItems": 1},
"min_score": {"type": "number", "minimum": 0, "maximum": 1}
}
}
```
**Full Dataset Validation:**
```python
async def validate_full_dataset() -> dict:
"""Run comprehensive validation on entire dataset.
Use this for:
- Pre-commit hooks
- CI/CD validation
- Periodic integrity checks
"""
from backend.tests.smoke.retrieval.fixtures.loader import FixtureLoader
loader = FixtureLoader(use_expanded=True)
documents = loader.load_documents()
queries = loader.load_queries()
source_url_map = loader.load_source_url_map()
all_errors = []
all_warnings = []
# 1. Schema validation for all documents
for doc in documents:
errors = validate_schema(doc)
all_errors.extend([f"[{doc['id']}] {e}" for e in errors])
# 2. Unique ID validation
id_errors = validate_unique_ids(documents, queries)
all_errors.extend(id_errors)
# 3. Referential integrity
ref_errors = validate_references(documents, queries)
all_errors.extend(ref_errors)
# 4. URL validation
for doc in documents:
valid, msg = validate_url(doc.get("source_url", ""))
if not valid:
all_errors.append(f"[{doc['id']}] {msg}")
# 5. Difficulty distribution
dist_warnings = validate_difficulty_distribution(queries)
all_warnings.extend(dist_warnings)
# 6. Coverage analysis
coverage = analyze_coverage_gaps(documents, queries)
all_warnings.extend(coverage["gaps"])
return {
"valid": len(all_errors) == 0,
"errors": all_errors,
"warnings": all_warnings,
"coverage": coverage,
"stats": {
"documents": len(documents),
"queries": len(queries),
"sections": sum(len(d.get("sections", [])) for d in documents),
}
}
```
**Pre-Commit Hook:**
```bash
#!/bin/bash
# .claude/hooks/pretool/bash/validate-golden-dataset.sh
# Only run if golden dataset files changed
CHANGED_FILES=$(git diff --cached --name-only)
if echo "$CHANGED_FILES" | grep -q "fixtures/documents_expanded.json\|fixtures/queries.json\|fixtures/source_url_map.json"; then
echo "Validating golden dataset changes..."
cd backend
poetry run python scripts/data/add_to_golden_dataset.py validate-all
if [ $? -ne 0 ]; then
echo "Golden dataset validation failed!"
echo "Fix errors before committing."
exit 1
fi
echo "Golden dataset validation passed"
fi
```
**CLI Validation Commands:**
```bash
# Validate specific document
poetry run python scripts/data/add_to_golden_dataset.py validate \
--document-id "new-doc-id"
# Validate full dataset
poetry run python scripts/data/add_to_golden_dataset.py validate-all
# Check for duplicates
poetry run python scripts/data/add_to_golden_dataset.py check-duplicate \
--url "https://example.com/article"
# Analyze coverage gaps
poetry run python scripts/data/add_to_golden_dataset.py coverage
```
**Incorrect — Unbalanced difficulty distribution:**
```python
# All queries marked "easy"
queries = [
{"id": "q-1", "difficulty": "easy"},
{"id": "q-2", "difficulty": "easy"},
{"id": "q-3", "difficulty": "easy"},
]
```
**Correct — Balanced difficulty distribution:**
```python
# Mix of difficulty levels
queries = [
{"id": "q-1", "difficulty": "trivial"}, # 3+ trivial
{"id": "q-2", "difficulty": "easy"}, # 3+ easy
{"id": "q-3", "difficulty": "medium"}, # 5+ medium
{"id": "q-4", "difficulty": "hard"}, # 3+ hard
]
# Validate distribution
validate_difficulty_distribution(queries) # Checks minimums
```
**Key rules:**
- Run full dataset validation before every commit that modifies golden dataset files
- Enforce minimum difficulty distribution (trivial 3, easy 3, medium 5, hard 3)
- Run all 6 validation steps: schema, IDs, references, URLs, distribution, coverage
- Block commits that introduce schema errors or referential integrity violations
- Treat difficulty distribution and coverage gaps as warnings that should be addressed
---
## References (4)
### Ork Delta
# ork delta: golden-dataset
House numbers and scars kept after the 2026-07-31 wrap-plus-delta thinning of
src/skills/golden-dataset. Generic dataset-management tutorials were removed; the
"Upstream coverage (do not restate)" table in SKILL.md maps every removed topic to its
first-party source. The operational thresholds (quality 0.70, confidence 0.65, tag and
query minimums) live in SKILL.md's Key Decisions table, single-sourced there.
## Store canonical source URLs, never placeholders, in every dataset entry
Why: The retired validation-contracts.md recorded the original scar: entries saved with
placeholder URLs (docs.orchestkit.dev/placeholder/...) could not be re-fetched when
embeddings needed regeneration, broke restore, and left no provenance trail. Canonical
URLs are what make "exclude embeddings from backup, regenerate on restore" (Key
Decisions) possible at all.
Upstream: https://langfuse.com/docs/datasets (dataset item shape and provenance)
## Grade retrieval difficulty on the five-level ladder with expected-score floors
Why: House rubric rescued from the retired curation-diversity.md; without it, difficulty
labels regress to guesses. trivial expects >0.85 retrieval score (direct keyword match),
easy >0.70 (synonyms), medium >0.55 (paraphrased intent), hard >0.40 (multi-hop),
adversarial expects graceful degradation only. Coverage floors from the same file:
tutorials at least 15% of documents, research papers at least 5%, at least 5 documents
per expected domain, hard queries at least 10% and adversarial at least 5% of queries.
Upstream: https://langfuse.com/docs/datasets and ork:testing-llm for evaluation harnesses
## Block additions at 0.90 cosine similarity, warn at 0.85, note at 0.80
Why: House duplicate policy rescued from the retired validation-drift.md, tuned on the
original 98-document dataset: 0.90 blocks, 0.85 warns, 0.80 is informational. Compare
normalized URLs too (lowercase, strip www, trailing slashes, and query strings), and
truncate content to 8000 chars before embedding for the comparison.
Upstream: pgvector cosine-distance docs, https://github.com/pgvector/pgvector
## Do not document infrastructure this repo does not ship
Why: The retired orchestkit-dataset-workflow.md (726 lines), backup-restore-checklist.md
(547 lines), and management-ci.md described a poetry backend at
coding/OrchestKit/backend with a port-5437 postgres, 98 analyses, 415 chunks, and a
91.6% / 0.777 MRR baseline. That app is the pre-plugin OrchestKit and does not exist in
this repository; the numbers are historical, not current truth. Same failure class as
the 2026-06 theater-vs-reality audit. Verify a described path exists at HEAD before
documenting a workflow around it.
Upstream: src/skills/CONTRIBUTING-SKILLS.md (house authoring standard; no vendor owns this rule)
### Quality Metrics
# Quality Metrics and Coverage Analysis
Metrics and analysis patterns for golden dataset quality.
## Coverage Analysis
### Gap Detection
```python
def analyze_coverage_gaps(
documents: list[dict],
queries: list[dict],
) -> dict:
"""Analyze dataset coverage and identify gaps."""
# Content type distribution
content_types = {}
for doc in documents:
ct = doc.get("content_type", "unknown")
content_types[ct] = content_types.get(ct, 0) + 1
# Domain/tag distribution
all_tags = []
for doc in documents:
all_tags.extend(doc.get("tags", []))
tag_counts = {}
for tag in all_tags:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
# Difficulty distribution
difficulties = {}
for query in queries:
diff = query.get("difficulty", "unknown")
difficulties[diff] = difficulties.get(diff, 0) + 1
# Identify gaps
gaps = []
# Check content type balance
total_docs = len(documents)
if content_types.get("tutorial", 0) / total_docs < 0.15:
gaps.append("Under-represented: tutorials (<15%)")
if content_types.get("research_paper", 0) / total_docs < 0.05:
gaps.append("Under-represented: research papers (<5%)")
# Check domain coverage
expected_domains = ["ai-ml", "backend", "frontend", "devops", "security"]
for domain in expected_domains:
if tag_counts.get(domain, 0) < 5:
gaps.append(f"Under-represented domain: {domain} (<5 docs)")
# Check difficulty balance
total_queries = len(queries)
if difficulties.get("hard", 0) / total_queries < 0.10:
gaps.append("Under-represented: hard queries (<10%)")
if difficulties.get("adversarial", 0) / total_queries < 0.05:
gaps.append("Under-represented: adversarial queries (<5%)")
return {
"content_type_distribution": content_types,
"tag_distribution": dict(sorted(tag_counts.items(), key=lambda x: -x[1])[:20]),
"difficulty_distribution": difficulties,
"gaps": gaps,
"total_documents": total_docs,
"total_queries": total_queries,
}
```
## Validation Workflow
### Pre-Addition Validation
```python
async def validate_before_add(
document: dict,
existing_documents: list[dict],
existing_queries: list[dict],
source_url_map: dict[str, str],
embedding_service,
) -> dict:
"""Run full validation before adding document.
Returns:
{
"valid": bool,
"errors": list[str], # Blocking issues
"warnings": list[str], # Non-blocking issues
"duplicate_check": {
"is_duplicate": bool,
"similar_to": str | None,
"similarity": float | None,
}
}
"""
errors = []
warnings = []
# 1. Schema validation
schema_errors = validate_schema(document)
errors.extend(schema_errors)
# 2. URL validation
url_valid, url_msg = validate_url(document.get("source_url", ""))
if not url_valid:
errors.append(url_msg)
# 3. URL duplicate check
url_dup = check_url_duplicate(document.get("source_url", ""), source_url_map)
if url_dup:
errors.append(f"URL already exists in dataset as: {url_dup}")
# 4. Content quality
quality_warnings = validate_content_quality(document)
warnings.extend(quality_warnings)
# 5. Semantic duplicate check
content = " ".join(
s.get("content", "") for s in document.get("sections", [])
)
existing_embeddings = await load_existing_embeddings(existing_documents)
dup_result = await check_duplicate(
content, existing_embeddings, embedding_service
)
duplicate_check = {
"is_duplicate": dup_result is not None,
"similar_to": dup_result[0] if dup_result else None,
"similarity": dup_result[1] if dup_result else None,
}
if dup_result and dup_result[1] >= 0.90:
errors.append(
f"Content too similar to existing document: {dup_result[0]} "
f"(similarity: {dup_result[1]:.2f})"
)
elif dup_result and dup_result[1] >= 0.80:
warnings.append(
f"Content similar to existing document: {dup_result[0]} "
f"(similarity: {dup_result[1]:.2f})"
)
return {
"valid": len(errors) == 0,
"errors": errors,
"warnings": warnings,
"duplicate_check": duplicate_check,
}
```
### Full Dataset Validation
```python
async def validate_full_dataset() -> dict:
"""Run comprehensive validation on entire dataset.
Use this for:
- Pre-commit hooks
- CI/CD validation
- Periodic integrity checks
"""
from backend.tests.smoke.retrieval.fixtures.loader import FixtureLoader
loader = FixtureLoader(use_expanded=True)
documents = loader.load_documents()
queries = loader.load_queries()
source_url_map = loader.load_source_url_map()
all_errors = []
all_warnings = []
# 1. Schema validation for all documents
for doc in documents:
errors = validate_schema(doc)
all_errors.extend([f"[{doc['id']}] {e}" for e in errors])
# 2. Unique ID validation
id_errors = validate_unique_ids(documents, queries)
all_errors.extend(id_errors)
# 3. Referential integrity
ref_errors = validate_references(documents, queries)
all_errors.extend(ref_errors)
# 4. URL validation
for doc in documents:
valid, msg = validate_url(doc.get("source_url", ""))
if not valid:
all_errors.append(f"[{doc['id']}] {msg}")
# 5. Difficulty distribution
dist_warnings = validate_difficulty_distribution(queries)
all_warnings.extend(dist_warnings)
# 6. Coverage analysis
coverage = analyze_coverage_gaps(documents, queries)
all_warnings.extend(coverage["gaps"])
return {
"valid": len(all_errors) == 0,
"errors": all_errors,
"warnings": all_warnings,
"coverage": coverage,
"stats": {
"documents": len(documents),
"queries": len(queries),
"sections": sum(len(d.get("sections", [])) for d in documents),
}
}
```
## CLI Integration
### Validation Commands
```bash
# Validate specific document
poetry run python scripts/data/add_to_golden_dataset.py validate \
--document-id "new-doc-id"
# Validate full dataset
poetry run python scripts/data/add_to_golden_dataset.py validate-all
# Check for duplicates
poetry run python scripts/data/add_to_golden_dataset.py check-duplicate \
--url "https://example.com/article"
# Analyze coverage gaps
poetry run python scripts/data/add_to_golden_dataset.py coverage
```
## Pre-Commit Hook
```bash
#!/bin/bash
# .claude/hooks/pretool/bash/validate-golden-dataset.sh
# Only run if golden dataset files changed
CHANGED_FILES=$(git diff --cached --name-only)
if echo "$CHANGED_FILES" | grep -q "fixtures/documents_expanded.json\|fixtures/queries.json\|fixtures/source_url_map.json"; then
echo "Validating golden dataset changes..."
cd backend
poetry run python scripts/data/add_to_golden_dataset.py validate-all
if [ $? -ne 0 ]; then
echo "Golden dataset validation failed!"
echo "Fix errors before committing."
exit 1
fi
echo "Golden dataset validation passed"
fi
```
### Storage Patterns
# Storage Patterns
Backup strategies and storage formats for golden datasets.
## Backup Strategies
### Strategy 1: JSON Backup (Recommended)
**Pros:**
- Version controlled (commit to git)
- Human-readable (easy to inspect)
- Portable (works across DB versions)
- Incremental diffs (see what changed)
**Cons:**
- Must regenerate embeddings on restore
- Larger file size than SQL dump
**OrchestKit uses JSON backup.**
### Strategy 2: SQL Dump
**Pros:**
- Fast restore (includes embeddings)
- Exact replica (binary-identical)
- Native PostgreSQL format
**Cons:**
- Not version controlled (binary format)
- DB version dependent
- No easy inspection
**Use case:** Local snapshots, not version control.
## Backup Format
```json
{
"version": "1.0",
"created_at": "2025-12-19T10:30:00Z",
"metadata": {
"total_analyses": 98,
"total_chunks": 415,
"total_artifacts": 98
},
"analyses": [
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"url": "https://docs.python.org/3/library/asyncio.html",
"content_type": "documentation",
"status": "completed",
"created_at": "2025-11-15T08:20:00Z",
"findings": [
{
"agent": "security_agent",
"category": "best_practices",
"content": "Always use asyncio.run() for top-level entry point",
"confidence": 0.92
}
],
"chunks": [
{
"id": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"content": "asyncio is a library to write concurrent code...",
"section_title": "Introduction to asyncio",
"section_path": "docs/python/asyncio/intro.md",
"content_type": "paragraph",
"chunk_index": 0
// Note: embedding NOT included (regenerated on restore)
}
],
"artifact": {
"id": "a1b2c3d4-e5f6-4a5b-8c7d-9e8f7a6b5c4d",
"summary": "Comprehensive guide to asyncio...",
"key_findings": ["..."],
"metadata": {}
}
}
]
}
```
**Key Design Decisions:**
- Embeddings excluded (regenerate on restore with current model)
- Nested structure (analyses -> chunks -> artifacts)
- Metadata for validation
- ISO timestamps for reproducibility
## Backup Implementation
```python
# backend/scripts/backup_golden_dataset.py
import asyncio
import json
from datetime import datetime, UTC
from pathlib import Path
from sqlalchemy import select
from app.db.session import get_session
from app.db.models import Analysis, Chunk, Artifact
BACKUP_DIR = Path("backend/data")
BACKUP_FILE = BACKUP_DIR / "golden_dataset_backup.json"
METADATA_FILE = BACKUP_DIR / "golden_dataset_metadata.json"
async def backup_golden_dataset():
"""Backup golden dataset to JSON."""
async with get_session() as session:
# Fetch all completed analyses
query = (
select(Analysis)
.where(Analysis.status == "completed")
.order_by(Analysis.created_at)
)
result = await session.execute(query)
analyses = result.scalars().all()
# Serialize to JSON
backup_data = {
"version": "1.0",
"created_at": datetime.now(UTC).isoformat(),
"metadata": {
"total_analyses": len(analyses),
"total_chunks": sum(len(a.chunks) for a in analyses),
"total_artifacts": len([a for a in analyses if a.artifact])
},
"analyses": [
serialize_analysis(a) for a in analyses
]
}
# Write backup file
BACKUP_DIR.mkdir(exist_ok=True)
with open(BACKUP_FILE, "w") as f:
json.dump(backup_data, f, indent=2, default=str)
# Write metadata file (quick stats)
with open(METADATA_FILE, "w") as f:
json.dump(backup_data["metadata"], f, indent=2)
print(f"Backup completed: {BACKUP_FILE}")
print(f" Analyses: {backup_data['metadata']['total_analyses']}")
print(f" Chunks: {backup_data['metadata']['total_chunks']}")
def serialize_analysis(analysis: Analysis) -> dict:
"""Serialize analysis to dict."""
return {
"id": str(analysis.id),
"url": analysis.url,
"content_type": analysis.content_type,
"status": analysis.status,
"created_at": analysis.created_at.isoformat(),
"findings": [serialize_finding(f) for f in analysis.findings],
"chunks": [serialize_chunk(c) for c in analysis.chunks],
"artifact": serialize_artifact(analysis.artifact) if analysis.artifact else None
}
def serialize_chunk(chunk: Chunk) -> dict:
"""Serialize chunk (WITHOUT embedding)."""
return {
"id": str(chunk.id),
"content": chunk.content,
"section_title": chunk.section_title,
"section_path": chunk.section_path,
"content_type": chunk.content_type,
"chunk_index": chunk.chunk_index
# embedding excluded (regenerate on restore)
}
```
## CLI Usage
```bash
cd backend
# Backup golden dataset
poetry run python scripts/backup_golden_dataset.py backup
# Verify backup integrity
poetry run python scripts/backup_golden_dataset.py verify
# Restore from backup (WARNING: Deletes existing data)
poetry run python scripts/backup_golden_dataset.py restore --replace
# Restore without deleting (adds to existing)
poetry run python scripts/backup_golden_dataset.py restore
```
## CI/CD Integration
### Automated Backups
```yaml
# .github/workflows/backup-golden-dataset.yml
name: Backup Golden Dataset
on:
schedule:
- cron: '0 2 * * 0' # Weekly on Sunday at 2am
workflow_dispatch: # Manual trigger
jobs:
backup:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Setup Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
cd backend
poetry install
- name: Run backup
env:
DATABASE_URL: ${{ secrets.PROD_DATABASE_URL }}
run: |
cd backend
poetry run python scripts/backup_golden_dataset.py backup
- name: Commit backup
run: |
git config user.name "GitHub Actions"
git config user.email "actions@github.com"
git add backend/data/golden_dataset_backup.json
git add backend/data/golden_dataset_metadata.json
git commit -m "chore: automated golden dataset backup"
git push
```
### Versioning
# Versioning and Recovery
Restore procedures, validation, and disaster recovery patterns.
## Restore Implementation
### Process Overview
1. **Load JSON backup**
2. **Validate structure** (version, required fields)
3. **Create analyses** (without embeddings yet)
4. **Create chunks** (without embeddings yet)
5. **Generate embeddings** (using current embedding model)
6. **Create artifacts**
7. **Verify integrity** (counts, URL contract)
### Regenerating Embeddings
```python
async def restore_golden_dataset(replace: bool = False):
"""Restore golden dataset from JSON backup."""
# Load backup
with open(BACKUP_FILE) as f:
backup_data = json.load(f)
async with get_session() as session:
if replace:
# Delete existing data
await session.execute(delete(Chunk))
await session.execute(delete(Artifact))
await session.execute(delete(Analysis))
await session.commit()
# Restore analyses and chunks
from app.shared.services.embeddings import embed_text
for analysis_data in backup_data["analyses"]:
# Create analysis
analysis = Analysis(
id=UUID(analysis_data["id"]),
url=analysis_data["url"],
# ... other fields ...
)
session.add(analysis)
# Create chunks with regenerated embeddings
for chunk_data in analysis_data["chunks"]:
# Regenerate embedding using CURRENT model
embedding = await embed_text(chunk_data["content"])
chunk = Chunk(
id=UUID(chunk_data["id"]),
analysis_id=analysis.id,
content=chunk_data["content"],
embedding=embedding, # Freshly generated!
# ... other fields ...
)
session.add(chunk)
await session.commit()
print("Restore completed")
```
**Why regenerate embeddings?**
- Embedding models improve over time
- Ensures consistency with current model
- Smaller backup files (exclude large vectors)
## Validation
### Validation Checklist
```python
async def verify_golden_dataset() -> dict:
"""Verify golden dataset integrity."""
errors = []
warnings = []
async with get_session() as session:
# 1. Check counts
analysis_count = await session.scalar(select(func.count(Analysis.id)))
chunk_count = await session.scalar(select(func.count(Chunk.id)))
artifact_count = await session.scalar(select(func.count(Artifact.id)))
expected = load_metadata()
if analysis_count != expected["total_analyses"]:
errors.append(f"Analysis count mismatch: {analysis_count} vs {expected['total_analyses']}")
# 2. Check URL contract
query = select(Analysis).where(
Analysis.url.like("%orchestkit.dev%") |
Analysis.url.like("%placeholder%")
)
result = await session.execute(query)
invalid_urls = result.scalars().all()
if invalid_urls:
errors.append(f"Found {len(invalid_urls)} analyses with placeholder URLs")
# 3. Check embeddings exist
query = select(Chunk).where(Chunk.embedding.is_(None))
result = await session.execute(query)
missing_embeddings = result.scalars().all()
if missing_embeddings:
errors.append(f"Found {len(missing_embeddings)} chunks without embeddings")
# 4. Check orphaned chunks
query = select(Chunk).outerjoin(Analysis).where(Analysis.id.is_(None))
result = await session.execute(query)
orphaned = result.scalars().all()
if orphaned:
warnings.append(f"Found {len(orphaned)} orphaned chunks")
return {
"valid": len(errors) == 0,
"errors": errors,
"warnings": warnings,
"stats": {
"analyses": analysis_count,
"chunks": chunk_count,
"artifacts": artifact_count
}
}
```
## Best Practices
### 1. Version Control Backups
```bash
# Commit backups to git
git add backend/data/golden_dataset_backup.json
git commit -m "chore: golden dataset backup (98 analyses, 415 chunks)"
```
### 2. Validate Before Deployment
```bash
# Pre-deployment check
poetry run python scripts/backup_golden_dataset.py verify
# Should output:
# Validation passed
# Analyses: 98
# Chunks: 415
# Artifacts: 98
# No errors found
```
### 3. Test Restore in Staging
```bash
# Never test restore in production first!
# Staging environment
export DATABASE_URL=$STAGING_DATABASE_URL
poetry run python scripts/backup_golden_dataset.py restore --replace
# Run tests to verify
poetry run pytest tests/integration/test_retrieval_quality.py
```
### 4. Document Changes
```json
// backend/data/golden_dataset_metadata.json
{
"total_analyses": 98,
"total_chunks": 415,
"last_updated": "2025-12-19T10:30:00Z",
"changes": [
{
"date": "2025-12-19",
"action": "added",
"count": 5,
"description": "Added 5 new LangGraph tutorial analyses"
},
{
"date": "2025-12-10",
"action": "removed",
"count": 2,
"description": "Removed 2 outdated React 17 analyses"
}
]
}
```
## Disaster Recovery
### Scenario 1: Accidental Deletion
```bash
# Oh no! Someone ran DELETE FROM analyses WHERE 1=1
# 1. Restore from backup
poetry run python scripts/backup_golden_dataset.py restore --replace
# 2. Verify
poetry run python scripts/backup_golden_dataset.py verify
# 3. Run tests
poetry run pytest tests/integration/test_retrieval_quality.py
```
### Scenario 2: Database Migration Gone Wrong
```bash
# Migration corrupted data
# 1. Rollback migration
alembic downgrade -1
# 2. Restore from backup
poetry run python scripts/backup_golden_dataset.py restore --replace
# 3. Re-run migration (fixed)
alembic upgrade head
```
### Scenario 3: New Environment Setup
```bash
# Fresh dev environment, need golden dataset
# 1. Clone repo (includes backup)
git clone https://github.com/your-org/orchestkit
cd orchestkit/backend
# 2. Setup DB
docker compose up -d postgres
alembic upgrade head
# 3. Restore golden dataset
poetry run python scripts/backup_golden_dataset.py restore
# 4. Verify
poetry run pytest tests/integration/test_retrieval_quality.py
```
## Data Integrity Contracts
### The URL Contract
Golden dataset analyses MUST store **real canonical URLs**, not placeholders.
```python
# WRONG - Placeholder URL (breaks restore)
analysis.url = "https://orchestkit.dev/placeholder/123"
# CORRECT - Real canonical URL (enables re-fetch if needed)
analysis.url = "https://docs.python.org/3/library/asyncio.html"
```
**Why this matters:**
- Enables re-fetching content if embeddings need regeneration
- Allows validation that source content hasn't changed
- Provides audit trail for data provenance
**Verification:**
```python
# Check for placeholder URLs
def verify_url_contract(analyses: list[Analysis]) -> list[str]:
"""Find analyses with placeholder URLs."""
invalid = []
for analysis in analyses:
if "orchestkit.dev" in analysis.url or "placeholder" in analysis.url:
invalid.append(analysis.id)
return invalid
```