admin/0pt
5
0
Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(er): route production pipeline through canonical TS ER engine

Browse Source
This commit is contained in:
admin
2026-09-07 17:20:52 +02:00
parent ebb9521801
commit 21ff9af769
1 changed files with 47 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/api/workers/pipeline.py
+47 -74
View File
@@ -1,23 +1,9 @@
"""
Production pipeline (docs/06): sourcing -> extraction -> structuring ->
enrichment/fusion -> QA. Compliant connectors only; every field gets provenance;
rows below the order's confidence target are dropped (deliver fewer, never fabricate).
"""Production pipeline: sourcing -> extraction -> canonical ER -> QA -> quality.
P010 additions (both flagged in BEFORE_STATE.md, both additive — neither
changes which rows get delivered by the Python filter above):
* Stage 6 (advisory, feature-flagged): after QA, `delivered` rows are
optionally handed to the TS QualityService via
`workers.quality_bridge.run_quality_bridge` — see that module's
docstring for the full contract, the fail-open guarantee, and why
per-row gating isn't implemented yet. Disabled unless
`OPT_QUALITY_BRIDGE_ENABLED=true`.
* Every delivered record's provenance rows are stamped with `producer`,
`connector_version` (both already set at fetch time in
sources/connectors.py), and an `artifact_ref` identifying this
delivery (`0pt:pipeline_delivery:<order_id>`) — plus, when the quality
bridge ran successfully, a `quality` block. BEFORE_STATE.md §3 found
these fields absent from the actual dataflow entirely.
Identity decisions are owned by src/er/entity-resolution-service.ts. The
Python worker only applies canonical `match` decisions returned by its
internal bridge. If tenant context or the canonical bridge is unavailable,
records remain unmerged: false identity merges are never used as fallback.
"""
from __future__ import annotations
@@ -26,7 +12,6 @@ from typing import Any
from workers.resolve import resolve
from workers.quality_bridge import run_quality_bridge
from sources.connectors import (
CompanyResult,
SourceBlockedError,
@@ -39,53 +24,38 @@ from sources.connectors import (
from workers.adaptive_router import default_router, plan_for_market
logger = logging.getLogger(__name__)
PIPELINE_VERSION = "0pt-pipeline@2"
PIPELINE_VERSION = "0pt-pipeline@1"
# Minimal built-in seed so a job is demonstrable without customer input.
DEMO_SEEDS: dict[str, list[str]] = {
"NO": ["923609016"], # brreg orgnr (example)
"GB": ["00000006"], # companies house number (example)
"NO": ["923609016"],
"GB": ["00000006"],
}
async def _fetch(source_id: str, key: str) -> CompanyResult | None:
if source_id == "gleif":
return await gleif_by_lei(key)
if source_id == "brreg":
return await brreg_by_orgnr(key)
if source_id == "companies_house":
return await companies_house_by_number(key)
if source_id == "prh":
return await prh_by_businessid(key)
if source_id == "cvr":
return await cvr_by_number(key)
if source_id == "gleif": return await gleif_by_lei(key)
if source_id == "brreg": return await brreg_by_orgnr(key)
if source_id == "companies_house": return await companies_house_by_number(key)
if source_id == "prh": return await prh_by_businessid(key)
if source_id == "cvr": return await cvr_by_number(key)
return None
async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
markets: list[str] = config.get("markets", []) or ["NO"]
threshold = float(config.get("confidence", 92)) / 100.0
# Use caller-supplied seeds when present (an explicit empty {} is honoured so a
# real job is never silently backfilled with DEMO_SEEDS), but treat an explicit
# null the same as "missing" -- `config.get("seeds") is not None` avoids a 500
# (None.get) on client-controlled `{"seeds": null}`, which the old `... or ...`
# idiom tolerated and the `"seeds" in config` form regressed.
seeds: dict[str, list[str]] = config["seeds"] if config.get("seeds") is not None else DEMO_SEEDS
tenant_id = config.get("tenant_id")
project_id = config.get("project_id", "p_default")
correlation_id = config.get("correlation_id")
records: list[dict[str, Any]] = []
per_market: dict[str, int] = {}
# 1) sourcing + 2) extraction
for market in markets:
keys = seeds.get(market, [])
got = 0
for key in keys:
# Recomputed per key (not hoisted above this loop): a source
# blocked while fetching an earlier key in this same market must
# be excluded from the plan for every key still to come in this
# run, not just retried forever with no memory (P014 -- see
# adaptive_router.py's module docstring).
plan = plan_for_market(market)
for source_id in plan:
try:
@@ -97,57 +67,55 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
res = None
if res and res.name:
records.append({
"market": market, "name": res.name, "status": res.status,
"lei": res.lei, "org_nr": res.org_nr, "source": res.source,
"market": market,
"name": res.name,
"status": res.status,
"lei": res.lei,
"org_nr": res.org_nr,
"source": res.source,
"provenance": [p.__dict__ for p in res.provenance],
"confidence": max((p.confidence for p in res.provenance), default=0.0),
})
got += 1
break # first compliant source that answers wins
break
per_market[market] = got
# 3) structuring — entity resolution (blocking + fuzzy match + corroboration)
deduped = resolve(records)
# Canonical ER is tenant-bound. Missing tenant context must never be
# papered over with a synthetic tenant or the removed Python scorer.
if tenant_id:
deduped = await resolve(
records,
tenant_id=tenant_id,
project_id=project_id,
correlation_id=correlation_id,
)
er_mode = "canonical_ts"
else:
deduped = [dict(r) for r in records]
er_mode = "fail_closed_unmerged_missing_tenant"
# 4) enrichment/fusion already folded into confidence above.
# 5) QA — drop rows below the order's confidence threshold. THIS is the
# filter that actually governs delivery; nothing below is allowed to
# change its outcome.
delivered = [r for r in deduped if r["confidence"] >= threshold]
dropped = len(deduped) - len(delivered)
avg_conf = round(sum(r["confidence"] for r in delivered) / len(delivered), 3) if delivered else 0.0
# Stamp the delivery artifact reference on every provenance row of every
# delivered record (BEFORE_STATE.md §3: `artifact_ref` did not exist
# anywhere). This identifies WHICH delivery a field's value shipped in;
# it does not assert the row was ever written to ST0RE (not wired yet).
artifact_ref = f"0pt:pipeline_delivery:{order_id}"
for r in delivered:
for p in r.get("provenance", []):
p["artifact_ref"] = artifact_ref
# 6) quality bridge (advisory, feature-flagged, fail-open) — see
# workers/quality_bridge.py's docstring for the full contract. Runs
# AFTER the QA gate above and can never change `delivered`.
tenant_id = config.get("tenant_id")
quality_bridge_metrics: dict[str, Any]
if not tenant_id:
# The engine has no tenant context yet for orders created through
# this API (orchestrator/main.py never threads tenant_id into
# `config` — a pre-existing, separately-tracked gap). Rather than
# fabricate a tenant to make the bridge call "work", skip it
# honestly and say why.
quality_bridge_metrics = {
"status": "not_configured",
"detail": "config.tenant_id missing — engine has no tenant context for this order",
"detail": "config.tenant_id missing — quality bridge not called",
}
else:
bridge_result = await run_quality_bridge(
tenant_id=tenant_id,
project_id=config.get("project_id", "p_default"),
project_id=project_id,
dataset_id=f"0pt-order-{order_id}",
rows=delivered,
correlation_id=config.get("correlation_id"),
correlation_id=correlation_id,
)
quality_bridge_metrics = bridge_result.as_metrics()
if bridge_result.status == "ok":
@@ -167,11 +135,16 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
)
return {
"stage": "qa", "pct": 100, "state": "done",
"stage": "qa",
"pct": 100,
"state": "done",
"metrics": {
"rows": len(delivered), "dropped_below_threshold": dropped,
"per_market": per_market, "avg_confidence": avg_conf,
"rows": len(delivered),
"dropped_below_threshold": dropped,
"per_market": per_market,
"avg_confidence": avg_conf,
"pipeline_version": PIPELINE_VERSION,
"entity_resolution": er_mode,
"quality_bridge": quality_bridge_metrics,
},
"records": delivered,