GoldenPipe

Golden Suite orchestrator -- chains GoldenCheck, GoldenFlow, GoldenMatch.

Related Projects

• GoldenCheck: D:\show_case\goldencheck -- Data validation. Has its own CLAUDE.md.
• GoldenFlow: D:\show_case\goldenflow -- Data transformation. Has its own CLAUDE.md.
• GoldenMatch: D:\show_case\goldenmatch -- Entity resolution. Has its own CLAUDE.md.
• GitHub: benzsevern/goldenpipe, benzsevern/goldencheck, benzsevern/goldenflow, benzsevern/goldenmatch

Branch & Merge SOP (all Golden Suite repos)

• Feature work goes on feature/<name> branches, never directly to main
• Merge via squash merge PR (watchers see PR activity, history stays clean)
• PR title format: feat: <description> or fix: <description>
• Merge when: tests pass, docs updated. Days not weeks.
• After merge: delete remote branch

Environment

• Windows 11, bash shell (Git Bash)
• Python 3.12 at C:\Users\bsevern\AppData\Local\Programs\Python\Python312\python.exe
• Two GitHub accounts: benzsevern (personal) and benzsevern-mjh (work)
• MUST gh auth switch --user benzsevern before push, switch back to benzsevern-mjh after

Architecture

• goldenpipe/pipeline.py -- Pipeline class, run() function. ONLY file that imports from tools.
• goldenpipe/decisions.py -- Adaptive logic (decide_flow, decide_match). NO tool imports. Testable independently.
• goldenpipe/cli/main.py -- Typer CLI
• Tools imported with try/except ImportError guards (HAS_CHECK, HAS_FLOW, HAS_MATCH)
• Data flows as Polars DataFrames in memory between stages

Pipeline Flow

load_file -> GoldenCheck.scan_file(path) -> decide_flow(findings)
  -> if fixable: GoldenFlow.transform_df(df) -> updated df
  -> decide_match(findings, row_count, strategy_override)
  -> GoldenMatch.dedupe_df(df) or AgentSession.deduplicate(path)
  -> PipeResult

Testing

• pytest --tb=short from project root
• test_decisions.py: no tool deps, tests pure decision logic
• test_pipeline.py: requires goldencheck, goldenflow, goldenmatch installed

A2A Port Convention

• GoldenCheck: 8100, GoldenFlow: 8150, GoldenMatch: 8200, GoldenPipe: 8250

Remote MCP Server

Hosted on Railway, registered on Smithery:

• Endpoint: https://goldenpipe-mcp-production.up.railway.app/mcp/
• Smithery: https://smithery.ai/servers/benzsevern/goldenpipe
• Server card: https://goldenpipe-mcp-production.up.railway.app/.well-known/mcp/server-card.json
• Transport: Streamable HTTP (via StreamableHTTPSessionManager)
• Dockerfile: Dockerfile.mcp (Python 3.12-slim, installs .[mcp])
• Railway project: golden-suite-mcp (service: goldenpipe-mcp, port 8250)
• Local HTTP: goldenpipe mcp-serve --transport http --port 8250

API Gotchas

Pipeline.run() does NOT return the DataFrame

Pipeline.run() returns PipeResult which has status, stages, artifacts, errors, reasoning, timing — but NOT the output DataFrame. The PipeContext.df holds the final data but is not exposed in the result.

Workaround for getting the output DataFrame: Run the stages directly:

import goldenflow, goldenmatch, polars as pl

df = pl.read_csv("data.csv")

# Stage 1: Transform
result = goldenflow.transform_df(df)
cleaned = result.df

# Stage 2: Deduplicate
deduped = goldenmatch.dedupe_df(cleaned, fuzzy={"first_name": 0.8}, exact=["email"])

# Output: unique records + golden records
output = deduped.unique  # or deduped.golden for canonical records

PipeResult fields

result.status        # PipeStatus enum: SUCCESS, PARTIAL, FAILED
result.input_rows    # int
result.stages        # dict[str, StageResult] — NOT result.stage_results
result.artifacts     # dict[str, Any] — e.g. {"manifest": Manifest}
result.errors        # list[str]
result.reasoning     # dict[str, str] — why each stage was run/skipped
result.timing        # dict[str, float]
result.skipped       # list[str]

PipelineConfig for selective stages

from goldenpipe import Pipeline, PipelineConfig, StageSpec

config = PipelineConfig(
    pipeline="check-and-flow-only",
    stages=[
        StageSpec(use="goldencheck.scan"),
        StageSpec(use="goldenflow.transform"),
        # omit goldenmatch.dedupe to skip dedup
    ],
)
pipeline = Pipeline(config=config)
result = pipeline.run(source="data.csv")

Stage names

Available stages (discovered via entry points):

• goldencheck.scan — validate data quality
• goldenflow.transform — fix issues
• goldenmatch.dedupe — deduplicate records

GoldenFlow config for pipeline use

Zero-config GoldenFlow may not fix all issues. For pipeline benchmarks, configure explicit transforms:

from goldenflow.config.schema import GoldenFlowConfig, TransformSpec

config = GoldenFlowConfig(transforms=[
    TransformSpec(column="first_name", ops=["strip", "title_case"]),
    TransformSpec(column="last_name", ops=["strip", "title_case"]),
    TransformSpec(column="email", ops=["strip", "lowercase"]),
    TransformSpec(column="phone", ops=["strip", "phone_national"]),
    TransformSpec(column="city", ops=["strip", "title_case"]),
])
result = goldenflow.transform_df(df, config=config)

GoldenMatch config for pipeline use

Use the full config for best results — don't rely on auto-configure for synthetic/benchmark data:

from goldenmatch.config.schemas import (
    GoldenMatchConfig, MatchkeyConfig, MatchkeyField,
    BlockingConfig, BlockingKeyConfig, StandardizationConfig,
)

config = GoldenMatchConfig(
    standardization=StandardizationConfig(
        email=["email"], phone=["phone"],
        first_name=["strip", "name_proper"],
        last_name=["strip", "name_proper"],
    ),
    blocking=BlockingConfig(
        strategy="multi_pass",
        keys=[BlockingKeyConfig(fields=["email"], transforms=["lowercase", "strip"])],
        passes=[
            BlockingKeyConfig(fields=["email"], transforms=["lowercase", "strip"]),
            BlockingKeyConfig(fields=["last_name"], transforms=["soundex"]),
            BlockingKeyConfig(fields=["last_name"], transforms=["substring:0:3"]),
        ],
    ),
    matchkeys=[MatchkeyConfig(
        name="identity", type="weighted", threshold=0.75,
        fields=[
            MatchkeyField(field="first_name", scorer="ensemble", weight=1.0, transforms=["lowercase", "strip"]),
            MatchkeyField(field="last_name", scorer="ensemble", weight=1.0, transforms=["lowercase", "strip"]),
            MatchkeyField(field="email", scorer="jaro_winkler", weight=0.8, transforms=["lowercase", "strip"]),
        ],
    )],
)
result = goldenmatch.dedupe_df(df, config=config)

DQBench Integration

GoldenPipe is benchmarked by DQBench Pipeline category:

• DQBench Pipeline Score: 88.07 (without LLM)
• Adapter: dqbench/adapters/goldenpipe_adapter.py
• Run: pip install dqbench && dqbench run goldenpipe
• The adapter runs GoldenFlow + GoldenMatch directly (not Pipeline.run()) because PipeResult doesn't expose the DataFrame

Examples

• examples/full_suite_demo.py — each tool individually + pipeline
• examples/benchmark_suite.py — DQBench scores for all 4 tools
• examples/custom_pipeline.py — zero-config vs custom PipelineConfig
• examples/basic_pipeline.py — simple pipeline on a CSV