Skip to content

Migrating a Pipeline

This guide moves an existing research pipeline onto sci-etl-core, using one real migration as the worked example: udg-catalogue, which builds a catalogue of ultra-diffuse galaxies (UDGs) from arXiv papers with an LLM.

Every "before" snippet is taken from udg-catalogue as it was before the migration. Every "after" snippet in Steps 1 to 9 is taken from the migrated project on sci-etl-core 0.2, at commit 8cd9471. Some of those 0.2 calls were renamed or removed since. Where you'll end up shows the project on 0.6, and Upgrading the project shows how the code changed on the project's main branch as it moved to 0.4, 0.5, and 0.6. The migration guide lists every change by release. The science is astronomy, but nothing in the steps depends on it: swap the prompts, fields, and domain rules for your own.

The project before

udg-catalogue searches arXiv for cat:astro-ph.GA AND abs:ultra-diffuse. For each paper it asks an LLM whether the abstract reports real observations, downloads the LaTeX source or PDF, asks the LLM to extract every galaxy as JSON, and merges the galaxies into a CSV. A post-processing step then removes duplicates, scores completeness, assigns constellations and 3D clusters, and writes the sorted catalogue behind a Streamlit dashboard.

The ETL machinery lived in flat modules at the project root:

Module Lines Responsibility
arxiv_client.py 198 arXiv search and Atom parsing, LaTeX and PDF download, table extraction, reference trimming, both LLM calls
data_processor.py 314 processed-id file, galaxy validation, CSV upsert, deduplication, completeness, constellations, clustering, quality flags
main.py 94 paging loop over a ThreadPoolExecutor
config.py, logger.py, incremental.py 97 YAML read into module constants, logging setup, resume offset

The orchestration loop in main.py looked like this:

while papers_processed < MAX_PAPERS:
    xml_data = search_arxiv(SEARCH_QUERY, max_results=MAX_PAPERS, start_index=start_index)
    if not xml_data:
        break
    papers, total_in_xml = parse_arxiv_xml(xml_data, processed_ids)
    if total_in_xml == 0 or not papers:
        break

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_single_paper_task, paper, processed_ids): paper for paper in papers}
        for future in as_completed(futures):
            try:
                success = future.result()
                if success:
                    papers_processed += 1
                if papers_processed >= MAX_PAPERS:
                    break
            except Exception as exc:
                logger.error(f"Paper processing generated an exception: {exc}")

    start_index += MAX_PAPERS
    save_pipeline_metadata(start_index)
    time.sleep(SLEEP_BETWEEN)

Where you'll end up

After the migration, the whole ingestion side is a few functions that wire library components together. This is udg_catalogue/pipeline.py on sci-etl-core 0.6, trimmed to the ingestion pipeline and without the wrappers that log progress:

def build_catalogue_exporter(config: CatalogueConfig) -> AsyncCsvExporter:
    return AsyncCsvExporter(config.paths.raw_catalogue, [KEY_COLUMN, *MEASUREMENT_FIELDS])


def build_entity_extractor(
    config: CatalogueConfig,
    llm_client: AsyncLLMClient,
    rejections: AsyncRejectionStore | None = None,
) -> AsyncLLMEntityExtractor[dict[str, Any]]:
    return AsyncLLMEntityExtractor(
        llm_client,
        EXTRACTION_PROMPT,
        result_key=EXTRACTION_RESULT_KEY,
        timeout=config.llm.timeout,
        validator=build_galaxy_validator(),
        rejections=rejections,
        label_field=KEY_COLUMN,
    )


def build_pipeline(config, logger, http_client, llm_client, library, shutdown=None):
    cache = AsyncSqliteLLMResponseCache(config.paths.llm_cache)
    cached_llm = CachingLLMClient(llm_client, cache, model=config.llm.model)
    rejections = AsyncSqliteRejectionStore(config.paths.rejections)
    extractor = AsyncArxivExtractor.from_config(
        config.http,
        config.pipeline,
        client=http_client,
        pdf_parser=PdfPlumberParser(),
        latex_parser=LatexTarballParser(),
        full_text=config.full_text,
    )
    return AsyncETLPipeline.from_config(
        config.pipeline,
        extractor=extractor,
        relevance_filter=AsyncLLMRelevanceFilter(llm_client=cached_llm, system_prompt=RELEVANCE_PROMPT),
        entity_extractor=build_entity_extractor(config, cached_llm, rejections),
        exporter=build_catalogue_exporter(config),
        state_manager=AsyncFileStateManager(config.paths.processed_ids, config.paths.pipeline_metadata),
        closeables=[http_client, llm_client, cache, rejections, *library.closeables],
        memory_ingestor=library.memory_ingestor(build_chunker(config.embeddings)),
        shutdown=shutdown,
        on_event=PipelineEventLogger(logger.info, logger.warning),
        usage_sources=[llm_client, *library.usage_sources],
    )


async def run_ingestion(config, logger, start_index=None, shutdown=None) -> int:
    library = open_paper_library(config)
    http_client = build_http_client(config)
    llm_client = build_llm_client(config)
    async with build_pipeline(config, logger, http_client, llm_client, library, shutdown) as pipeline:
        return await pipeline.run(**run_arguments(config.pipeline, start_index))

build_pipeline takes the HTTP and LLM clients as arguments, so the project's tests pass in an httpx.MockTransport that serves a fake Atom feed and e-print, plus a scripted AsyncLLMClient, and run the real pipeline offline.

main.py shrinks to loading the config, running ingestion, and building the outputs:

config = load_catalogue_config(arguments.config)
log = configure_run_logging(config.paths.log_file)
processed = asyncio.run(run_ingestion(config, log, start_index, ShutdownSignal()))
catalogue = build_sorted_catalogue(config, log.info, replace=arguments.replace_catalogue)
write_manifest(config, log.info)

What stays in the project is the part only an astronomer can write: prompts, the rules for naming and validating galaxies, sky-position matching, the features used for clustering, and the dashboard.

Map your pipeline onto the library

Start by sorting every function in the old pipeline into one of three groups: replaced by a library component, replaced by a library component plus a small plug-in, or kept.

Before (udg-catalogue) After What you still write
search_arxiv, parse_arxiv_xml AsyncArxivExtractor nothing
fetch_paper_text, extract_tables_from_pdf, trim_references AsyncArxivExtractor with LatexTarballParser and PdfPlumberParser nothing
requests session with Retry build_async_client nothing
is_paper_relevant AsyncLLMRelevanceFilter the prompt
extract_udg_data AsyncLLMEntityExtractor(result_key="galaxies") the prompt
OpenAI client pointed at DeepSeek AsyncOpenAICompatibleClient base URL and model
upsert_to_csv AsyncCsvExporter, then DeduplicationStep in post-processing the column list
load_processed_ids, save_processed_id, incremental.py AsyncFileStateManager file paths
main.py loop AsyncETLPipeline the wiring above
logger.py the standard logging module handlers for the run log
config.py BaseAppConfig subclass and load_config project settings
universal_normalize_name KeyNormalizer subclass name-matching rules
is_valid_galaxy RecordValidators passed to AsyncLLMEntityExtractor(validator=) field rules
clean_duplicates NormalizationStep and DeduplicationStep a NeighborMatcher for sky positions
calculate_completeness, assign_quality_flag CompletenessStep, QualityFlagStep the field list
assign_3d_clusters ClusteringStep a FeatureExtractor for 3D positions
assign_constellations, plots, dashboard kept domain code, as Processors where it fits

The table names the components of sci-etl-core 0.6. Steps 1 to 9 show the 0.2 components udg-catalogue used at the time, such as an upserting CSV exporter and a validating extractor wrapper, which later releases replaced.

You will change both codebases while migrating, so install the library in editable mode into the project's environment. udg-catalogue sits two folders away from its sci-etl-core checkout:

python -m venv .venv
source .venv/bin/activate
pip install -e "../../sci-etl-core[async,llm,pdf,cluster]"
python -c "import sci_etl_core; print(sci_etl_core.__file__)"

On Windows, activate with .venv\Scripts\activate. The last command should print a path inside the checkout's src folder. Choose extras for the components you use: udg-catalogue needs async for the arXiv extractor and CSV exporter, llm for the OpenAI-compatible client, pdf for PdfPlumberParser, and cluster for ClusteringStep.

An editable install records the checkout's absolute path. If you move or rename the library folder, import sci_etl_core fails until you reinstall; this is exactly what happened when the udg-catalogue workspace moved into a synced OneDrive folder.

Deployed environments can't see a sibling checkout, and pip can't install one package from two sources in the same resolve. udg-catalogue therefore keeps its pinned third-party packages in requirements-app.txt and chooses the library's source in two thin files. For local development, requirements-local.txt:

-r requirements-app.txt
-e ../../sci-etl-core[async,llm,pdf,cluster]

For Docker, CI, and new users, requirements.txt installs the published release from PyPI:

-r requirements-app.txt
sci-etl-core[async,llm,pdf,cluster]>=0.2.0,<0.3

Pin a version range rather than one exact version, so bug-fix releases arrive without a change to the project, and raise the upper bound deliberately after checking a new minor release against your tests. Before the library was on PyPI, udg-catalogue committed a wheel built with pip wheel --no-deps -w vendor path/to/sci-etl-core and installed it from vendor/; that still works for a build that can't reach PyPI.

Step 2: Configuration and secrets

Before (config.py): the YAML was read into module-level constants when the module was imported.

load_dotenv()
API_KEY: str | None = os.getenv("DEEPSEEK_API_KEY")
SCRIPT_DIR: str = os.path.dirname(os.path.abspath(__file__))

CONFIG_PATH = os.path.join(SCRIPT_DIR, "config.yaml")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    cfg: dict = yaml.safe_load(f)

MODEL: str = cfg.get("model", "deepseek-v4-flash")
CSV_FILE: str = os.path.join(SCRIPT_DIR, cfg.get("csv_file", "udg_database.csv"))
MAX_PAPERS: int = cfg.get("max_papers", 500)

After: config.yaml uses the library's sections (llm, http, pipeline) plus the project's own (paths, clustering, deduplication):

llm:
  base_url: "https://api.deepseek.com"
  model: "deepseek-v4-flash"
  timeout: 120

http:
  user_agent: "UDG-ResearchScript/1.0 (lanhua1122333@gmail.com)"
  max_retries: 4
  backoff_factor: 5.0
  timeout: 25

pipeline:
  search_query: "cat:astro-ph.GA AND abs:ultra-diffuse"
  max_records: 500
  page_size: 100
  search_delay: 3.0
  sleep_between: 5.0
  max_workers: 6

paths:
  raw_catalogue: "udg_database.csv"
  sorted_catalogue: "udg_database_sorted.csv"
  processed_ids: "processed_arxiv_ids.txt"
  pipeline_metadata: "pipeline_meta.json"

clustering:
  max_distance_mpc: 5.0
  min_samples: 2

deduplication:
  max_separation_arcsec: 3.0

udg_catalogue/config.py subclasses the library models:

class CataloguePipelineConfig(PipelineConfig):
    page_size: int = 100
    search_delay: float = 3.0


class CatalogueConfig(BaseAppConfig):
    pipeline: CataloguePipelineConfig = Field(default_factory=CataloguePipelineConfig)
    paths: PathsConfig = Field(default_factory=PathsConfig)
    clustering: ClusteringConfig = Field(default_factory=ClusteringConfig)
    deduplication: DeduplicationConfig = Field(default_factory=DeduplicationConfig)


def load_catalogue_config(config_path: Path = DEFAULT_CONFIG_PATH) -> CatalogueConfig:
    resolved = Path(config_path).resolve()
    config = load_config(
        CatalogueConfig,
        resolved,
        resolved.parent / ".env",
        api_key_env_var=API_KEY_ENV_VAR,
    )
    return config.model_copy(update={"paths": config.paths.anchored_at(resolved.parent)})

Three choices here are worth copying:

  • Keep your secret's name. api_key_env_var="DEEPSEEK_API_KEY" means the existing .env file and the Docker Compose file keep working unchanged. The key is held as a SecretStr, so it never shows up in reprs or logs.
  • Extend sections by subclassing. CataloguePipelineConfig adds page_size and search_delay while keeping every field the library reads.
  • Resolve paths from the config file. Passing the .env beside the config explicitly, and anchoring relative paths to the config's folder, means the pipeline behaves the same whichever directory you start it from. The old code resolved most paths from the script folder but pipeline_meta.json and analysis/ from the working directory.

Config values aren't applied automatically: pass them to constructors and to run(), as build_pipeline does.

Step 3: Source and full text

Before (arxiv_client.py, trimmed):

def search_arxiv(query: str, max_results: int = 5, start_index: int = 0) -> bytes | None:
    ...
    for attempt in range(MAX_RETRIES):
        try:
            response = session.get(base_url, params=params, headers={"User-Agent": USER_AGENT}, timeout=(10, 60))
            if response.status_code == 429:
                time.sleep(20)
                continue
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            if attempt < MAX_RETRIES - 1:
                time.sleep(5 * (attempt + 1))
            else:
                logger.error(f"arXiv search failed after {MAX_RETRIES} attempts: {e}")
    return None


def fetch_paper_text(entry: dict) -> str:
    ...
    with session.get(source_url, headers={"User-Agent": USER_AGENT}, timeout=25, verify=False, stream=True) as r:
        ...

After:

extractor = AsyncArxivExtractor(
    client=http_client,
    pdf_parser=PdfPlumberParser(),
    latex_parser=LatexTarballParser(),
    max_retries=config.http.max_retries,
    backoff_factor=config.http.backoff_factor,
    sleep_before_search=config.pipeline.search_delay,
    logger=logger.info,
)

What changed in behavior:

  • A failed search is an error, not the end of the data. search_arxiv returned None, and the loop in main.py treated that as "no more papers" and went on to report success. The old log shows 9 runs that stopped this way. AsyncETLPipeline.run() now raises PipelineAborted, and main.py exits with status 1.
  • Retries wait longer than the defaults. The extractor waits backoff_factor ** attempt seconds between attempts, so the default factor of 2 waits 1 s and then 2 s. arXiv throttles with 429 for longer than that, and the old code waited 20 s after a 429, so udg-catalogue sets backoff_factor: 5.0 and max_retries: 4 (waits of 1, 5, and 25 s).
  • TLS verification is back on. The old e-print download used verify=False.
  • More submissions yield LaTeX. A single gzipped .tex file is read instead of failing over to the PDF, and multi-file sources are assembled in \input order.
  • Everything else is the same. The reference-trimming patterns are the same seven expressions, and the PDF parser appends tables under the same --- EXTRACTED TABLES --- marker. For three recent papers, old and new code returned byte-identical text.

Step 4: The LLM steps

Before (arxiv_client.py, trimmed):

def is_paper_relevant(title: str, abstract: str) -> bool:
    if not abstract:
        return True
    try:
        response = client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": STRICT_SIMULATION_PROMPT},
                {"role": "user", "content": f"Title: {title}\nAbstract: {abstract}"}
            ],
            temperature=0.0,
            response_format={"type": "json_object"},
            timeout=20
        )
        data = json.loads(response.choices[0].message.content.strip())
        return bool(data.get("relevant", False))
    except Exception as e:
        logger.warning(f"Filter error: {e}. Proceeding to download.")
        return True


def extract_udg_data(text: str | bytes) -> list[dict]:
    try:
        ...
    except Exception as e:
        logger.error(f"DeepSeek error: {e}")
        return []

After (the same calls as in run_ingestion and build_pipeline, pulled out into variables):

llm_client = AsyncOpenAICompatibleClient(
    api_key=config.llm.api_key,
    base_url=config.llm.base_url,
    model=config.llm.model,
    default_timeout=config.llm.timeout,
)
relevance_filter = AsyncLLMRelevanceFilter(llm_client=llm_client, system_prompt=RELEVANCE_PROMPT)
entity_extractor = AsyncLLMEntityExtractor(
    llm_client=llm_client,
    system_prompt=EXTRACTION_PROMPT,
    result_key="galaxies",
    timeout=config.llm.timeout,
)

The prompts moved into udg_catalogue/prompts.py character for character. They already mentioned JSON, which JSON mode requires, and the extraction prompt already asked for {"galaxies": [...]}, so result_key="galaxies" reads that shape and the prompt didn't have to change. The 120,000-character cap, the HTML stripping, and the temperature of 0 are the library's defaults too.

What changed in behavior:

  • Relevance still fails open. A failed call lets the paper through, as before, and the timeout is still 20 s. One difference: a reply with no clear verdict now also lets the paper through, where the old code read a missing relevant key as False. Pass default_on_error=False to hold such papers back instead; since 0.5.1 they are then retried on the next run.
  • Extraction failures are retried. extract_udg_data returned [] when DeepSeek failed, so the paper was marked processed with nothing extracted; the old log shows 8 such papers that will never be revisited. AsyncLLMEntityExtractor raises LLMError, the pipeline leaves the paper unmarked, and the next run tries it again.

Step 5: Domain rules as plug-ins

The rules that decide what counts as the same galaxy, and what counts as a real one, are science, and they stay in the project. The library gives them a place to plug in.

Name matching: a KeyNormalizer

Before (data_processor.py):

def universal_normalize_name(name: str) -> str:
    if not name or pd.isna(name):
        return ""
    s = str(name).strip().lower()
    s = re.sub(r"[^a-z0-9]", "", s)
    digits_match = re.search(r"\d+", s)
    if digits_match:
        digits = str(int(digits_match.group()))
        if s.startswith("vcc"):
            return f"vcc{digits}"
        return f"dragonfly{digits}"

    return s

Port a function like this unchanged first, as a KeyNormalizer subclass, so the parity check in Step 8 compares plumbing and nothing else. udg-catalogue did exactly that.

This function turned out to merge different galaxies (see What the migration uncovered), so it was replaced once parity was confirmed. After (udg_catalogue/naming.py):

DEFAULT_PREFIX_ALIASES: dict[str, str] = {"dragonfly": "df"}
_NAME_TOKEN = re.compile(r"[^\W\d_]+|\d+")
_NUMBER_SEPARATOR = "."


class GalaxyNameNormalizer(KeyNormalizer):
    def __init__(self, prefix_aliases: Mapping[str, str] | None = None) -> None:
        self._prefix_aliases = dict(DEFAULT_PREFIX_ALIASES if prefix_aliases is None else prefix_aliases)
        self._missing_value_guard = DefaultKeyNormalizer()

    def normalize(self, raw_value: Any) -> str:
        if not self._missing_value_guard.normalize(raw_value):
            return ""
        tokens = _NAME_TOKEN.findall(unicodedata.normalize("NFKC", str(raw_value)).casefold())
        if not tokens:
            return ""
        tokens[0] = self._prefix_aliases.get(tokens[0], tokens[0])
        key: list[str] = []
        previous_is_number = False
        for token in tokens:
            is_number = token.isdecimal()
            if is_number and previous_is_number:
                key.append(_NUMBER_SEPARATOR)
            key.append(str(int(token)) if is_number else token)
            previous_is_number = is_number
        return "".join(key)

DF 44, DF044, and Dragonfly 44 still share the key df44, while KDG 44, NGC 1052-DF2, and NGC 1052-DF4 now keep keys of their own. Delegating the missing-value check to DefaultKeyNormalizer means None, NaN, and non-scalar values are handled the way every library component expects. The CSV exporter and the post-processing deduplication both use GalaxyNameNormalizer, so the two stages always agree on identity.

Validation: RecordValidators and a wrapper

Before (data_processor.py): validation was buried inside upsert_to_csv.

def is_valid_galaxy(galaxy: dict) -> bool:
    ...
    if FORBIDDEN_PATTERN.search(name):
        logger.info(f"Object '{name}' filtered out as simulation/model.")
        return False

    ra, dec = galaxy.get("ra"), galaxy.get("dec")
    if ra is not None:
        try:
            if not (0.0 <= float(ra) <= 360.0):
                return False
        except (ValueError, TypeError):
            return False
    ...
    return any(galaxy.get(f) is not None for f in KEY_FIELDS)

After (udg_catalogue/validation.py): the library's validators cover the keyword and range rules, and one small class covers the rest.

class HasAnyMeasurement(RecordValidator):
    def __init__(self, fields: Iterable[str]) -> None:
        self._fields = tuple(fields)

    def is_valid(self, record: dict[str, Any]) -> bool:
        return any(record.get(field) is not None for field in self._fields)


def build_galaxy_validator() -> RecordValidator:
    return CompositeValidator(
        [
            KeywordExclusionValidator(KEY_COLUMN, list(SIMULATION_KEYWORDS)),
            NumericRangeValidator(SKY_COORDINATE_RANGES),
            HasAnyMeasurement(MEASUREMENT_FIELDS),
        ]
    )

AsyncETLPipeline doesn't call validators itself, so a thin AsyncEntityExtractor applies them between extraction and export and logs what it drops:

class ValidatedEntityExtractor(AsyncEntityExtractor):
    def __init__(
        self,
        inner: AsyncEntityExtractor,
        validator: RecordValidator,
        logger: Callable[[str], None] | None = None,
    ) -> None:
        self._inner = inner
        self._validator = validator
        self._log = logger or (lambda _message: None)

    async def extract(self, text: str | bytes) -> list[dict[str, Any]]:
        accepted: list[dict[str, Any]] = []
        for entity in await self._inner.extract(text):
            if self._validator.is_valid(entity):
                accepted.append(entity)
            else:
                self._log(f"Entity rejected by validation: {entity.get(KEY_COLUMN)!r}")
        return accepted

Before relying on KeywordExclusionValidator in place of the old regular expression, check the two against your existing data. For udg-catalogue they agreed on all 1,285 stored names and on edge cases such as TNG50-1, illustris_galaxy_1, and Firefly 7.

Step 6: Export and state

Before (data_processor.py and incremental.py, trimmed): each worker thread read the whole CSV, changed it, and wrote it back, with no lock between the six threads.

def upsert_to_csv(records: list[dict]) -> None:
    ...
    df = pd.read_csv(CSV_FILE) if os.path.isfile(CSV_FILE) and os.path.getsize(CSV_FILE) > 0 else pd.DataFrame(columns=fieldnames)
    ...
    df.drop(columns=["_norm_name"]).to_csv(CSV_FILE, index=False, encoding="utf-8")


def save_processed_id(arxiv_id: str) -> None:
    if arxiv_id:
        with open(PROCESSED_FILE, "a", encoding="utf-8") as f:
            f.write(arxiv_id + "\n")


def save_pipeline_metadata(start_index: int) -> None:
    meta = {"last_run_date": datetime.now().isoformat(), "last_start_index": start_index}
    with open(META_FILE, "w", encoding="utf-8") as f:
        json.dump(meta, f, indent=4)

After (build_catalogue_exporter and the state manager from build_pipeline, with the constants from udg_catalogue/config.py and the default paths written out):

exporter = AsyncCsvUpsertExporter(
    key_column="galaxy_name",
    value_columns=["ra", "dec", "distance_mpc", "effective_radius_kpc", "stellar_mass_solar", "dark_matter_fraction"],
    normalizer=GalaxyNameNormalizer(),
    numeric_clip={"dark_matter_fraction": (0.0, 1.0)},
)
state_manager = AsyncFileStateManager("processed_arxiv_ids.txt", "pipeline_meta.json")

The exporter keeps the old merge rule: one row per normalized name, later records only fill empty cells, values are converted to floats, and numeric_clip replaces the hand-written clamp on the dark-matter fraction. It also serializes concurrent exports and publishes each snapshot with an atomic rename.

Check whether your existing state files can be reused as they are. udg-catalogue's could: processed_arxiv_ids.txt already held bare versioned ids such as 2607.14209v1, the format AsyncArxivExtractor produces, and pipeline_meta.json already had the last_run_date and last_start_index keys AsyncFileStateManager reads. Pointing the state manager at the old files carried all 542 processed papers over with no import script. If your ids are stored as URLs or without the version suffix, convert them first, or every paper is processed again.

Two resume details changed:

  • Newest-first listings. arXiv lists the newest submissions first, so a saved offset drifts as new papers arrive. main.py now passes start_index=0 by default, rescanning from the newest submission while skipping processed papers by id; a rescan costs listing requests, not LLM calls. python main.py --resume uses the saved offset instead.
  • Offsets only move past settled pages. The old loop added 500 to the offset after every page, even when papers on it had failed. The library only advances past a page once every paper on it is processed or ruled irrelevant.

Step 7: Post-processing

Before (data_processor.py, trimmed): deduplication rewrote the raw CSV in place after taking a .bak copy, and process_database ran a sequence of functions.

def process_database() -> None:
    df = pd.read_csv(CSV_FILE)
    ...
    if "dark_matter_fraction" in df.columns:
        df["dark_matter_fraction"] = df["dark_matter_fraction"].clip(0.0, 1.0)

    df = calculate_completeness(df)
    df = assign_constellations(df)
    df = assign_3d_clusters(df)
    df = assign_quality_flag(df)
    ...
    df_sorted.to_csv(SORTED_CSV_FILE, index=False, encoding="utf-8")

After (udg_catalogue/postprocess.py): the same sequence as a ProcessorChain, with the raw CSV left untouched and only the sorted catalogue written.

def build_catalogue_chain(config: CatalogueConfig, normalizer: KeyNormalizer | None = None) -> ProcessorChain:
    return ProcessorChain(
        [
            NormalizationStep(KEY_COLUMN, normalizer or GalaxyNameNormalizer(), NORMALIZED_KEY_COLUMN),
            DeduplicationStep(
                NORMALIZED_KEY_COLUMN,
                matcher=SkyPositionMatcher(),
                match_threshold=config.deduplication.max_separation_arcsec,
            ),
            ValueClipStep(FRACTION_BOUNDS),
            CompletenessStep(list(MEASUREMENT_FIELDS)),
            ConstellationStep(),
            ClusteringStep(
                CartesianDistanceFeatures(),
                eps=config.clustering.max_distance_mpc,
                min_samples=config.clustering.min_samples,
            ),
            QualityFlagStep(),
            CatalogueLayoutStep(),
        ]
    )

CompletenessStep, QualityFlagStep, NormalizationStep, DeduplicationStep, and ClusteringStep come from the library. The science plugs in through two small interfaces. SkyPositionMatcher tells DeduplicationStep which rows are the same object on the sky:

class SkyPositionMatcher(NeighborMatcher):
    def find_matches(self, frame: pd.DataFrame, threshold: float) -> list[tuple[int, int]]:
        located = frame[located_rows(frame, self._ra_column, self._dec_column)]
        if len(located) < 2:
            return []
        coordinates = sky_coordinates(located, self._ra_column, self._dec_column)
        neighbours, separations, _ = coordinates.match_to_catalog_sky(coordinates, nthneighbor=2)
        separation_arcsec = separations.to_value(u.arcsec)
        labels = located.index.to_numpy()
        return [
            (int(labels[position]), int(labels[neighbour]))
            for position, neighbour in enumerate(neighbours)
            if separation_arcsec[position] <= threshold and labels[position] < labels[neighbour]
        ]

CartesianDistanceFeatures gives ClusteringStep 3D positions built from right ascension, declination, and distance. ConstellationStep, ValueClipStep, and CatalogueLayoutStep are ordinary Processors kept in the project: a Processor only has to return a new DataFrame without mutating its input.

Keeping the raw CSV as the exporter's source of truth, and deriving the sorted file from it, makes post-processing repeatable: python main.py --skip-ingestion rebuilds every output without touching arXiv or the LLM.

Step 8: Check parity before changing behavior

Keep the old code runnable next to the new code and feed both the same inputs. You don't need the old project installed: export the old modules from Git into a scratch folder with git show <commit>:data_processor.py, and import them from there.

udg-catalogue ran four checks.

Export replay. Each row of the existing catalogue was split at random into two partial extraction records, a few hand-written invalid records were added, and the shuffled records were fed in batches through both the old upsert and the new validator and exporter:

for batch in batches:
    data_processor.upsert_to_csv([dict(record) for record in batch])

for batch in batches:
    await exporter.export([record for record in batch if validator.is_valid(record)], "new.csv")

Post-processing. The old clean_duplicates and process_database and the new chain ran on copies of the same raw catalogue, and the sorted outputs were compared cell by cell. Compare cluster membership, not cluster ids: DBSCAN numbers clusters by the order it meets them.

Full text. Old and new retrieval fetched the same three recent papers.

Live run. The assembled pipeline ran against arXiv and DeepSeek with a small page_size and total_limit, into a fresh state folder.

Check Result
Export replay: 2,581 records in 735 batches identical CSV
Export replay with a galaxy repeated within one paper identical after merging 2 duplicate rows the old upsert created
Post-processing on the 1,285-galaxy catalogue identical cell for cell, including row order and cluster ids, except 3 rows whose RA exceeds 360°
Full text for 3 recent papers byte-identical LaTeX text
Live run arXiv answered 429 to every listing attempt, and run() raised PipelineAborted with nothing written, instead of reporting success; this is what prompted the longer backoff in Step 3
Project test suite 84 offline tests, 100% coverage, passing both with the editable library and in a clean environment installed from the vendored wheel

Only after these matched did the normalizer fix go in, as a separate change. Regenerating the sorted catalogue with it, the only differences were the three invalid-RA rows, renumbered clusters with unchanged membership, and the dropped filled_fields column.

When you run the migrated pipeline:

  • Search the log for Record processing failed. Those records weren't marked processed, and the next run retries them.
  • If run() raises PipelineAborted, read the exception's __cause__: a rejected API key, an unreadable CSV, or a throttled listing request each show up there.
  • If you wrote your own components, check them against the contracts in Adding a New Component.

Step 9: Delete the old code

Once the checks pass, delete what the library replaced. udg-catalogue removed arxiv_client.py, data_processor.py, config.py, logger.py, incremental.py, the duplicated simulation-keyword pattern, and the unused filter prompt, along with the tests that mocked requests and the thread pool. What remains is a udg_catalogue package of domain modules (config, prompts, naming, validation, astrometry, post-processing, maps, analytics) and a test suite that exercises the real pipeline offline.

visualization.py and the dashboard used to build the same Plotly figure twice; the migration was a good moment to give them one shared figure builder. AsyncPlotly3DExporter wasn't used, because udg-catalogue's map needs custom hover text and a fixed color range.

Upgrading the project

Step 1 pins a version range and raises its upper bound only after checking a new minor release against the project's tests. The migration guide lists what each release changes; this section records what those changes meant for udg-catalogue.

Moving to 0.4

udg-catalogue skipped 0.3: its build_pipeline passed no memory_ingestor, and AsyncFileStateManager was unchanged, so nothing in 0.3 affected it. It moved from >=0.2.0,<0.3 straight to >=0.4.0,<0.5, adding the embeddings, embeddings-local, and search extras:

sci-etl-core[async,llm,pdf,cluster,embeddings,embeddings-local,search]>=0.4.0,<0.5

The 0.4 upgrade retired several pieces of code the steps above had the project write:

  • Renamed pipeline settings. The 0.2 build_pipeline and run_ingestion warn on 0.4, because max_records and max_workers became total_limit and max_concurrency. udg-catalogue renamed both keys in config.yaml.
  • No pipeline subclass. PipelineConfig gained page_size, search_delay, and newest_first, so CataloguePipelineConfig from Step 2 was deleted and CatalogueConfig uses the library's pipeline section as it is.
  • Validation without a wrapper. udg-catalogue passes build_galaxy_validator() and label_field=KEY_COLUMN to AsyncLLMEntityExtractor and deleted the ValidatedEntityExtractor from Step 5.
  • Newest-first resume. udg-catalogue sets pipeline.newest_first: true, so python main.py now resumes by default; the --resume flag from Step 6 is gone, and --rescan passes start_index=0 with newest_first=False to page the whole listing.
  • Caching, shutdown, and run summaries. udg-catalogue wraps its LLM client in a CachingLLMClient backed by an AsyncSqliteLLMResponseCache, so a rerun after a crash doesn't pay for the same relevance and extraction calls twice. It passes a ShutdownSignal, and main.py exits with code 130 on PipelineInterrupted. An on_event callback logs each PageFinished, and the RunFinished event's RunMetrics, including token usage from usage_sources, becomes the run summary in the log.
  • Plots. ScatterPlotConfig gained the custom hover text and fixed color range that kept udg-catalogue off AsyncPlotly3DExporter in Step 9.
  • Clamping and table layout. udg-catalogue deleted its own ValueClipStep and CatalogueLayoutStep from Step 7; the chain now imports ValueClipStep from the library and ends with TableLayoutStep(sort_by=SORT_ORDER, leading_columns=LEADING_COLUMNS, hidden_prefixes=("_",)).

After the upgrade, udg-catalogue's pipeline wiring reads its settings from the config and indexes every relevant paper for search. Trimmed to the ingestion branch, udg_catalogue/pipeline.py builds the pipeline like this:

def build_pipeline(config, logger, http_client, llm_client, library, shutdown=None):
    cache = AsyncSqliteLLMResponseCache(config.paths.llm_cache)
    cached_llm = CachingLLMClient(llm_client, cache, model=config.llm.model, logger=logger.warning)
    extractor = AsyncArxivExtractor.from_config(
        config.http,
        config.pipeline,
        client=http_client,
        pdf_parser=PdfPlumberParser(),
        latex_parser=LatexTarballParser(),
        logger=logger.info,
    )
    return AsyncETLPipeline.from_config(
        config.pipeline,
        extractor=extractor,
        relevance_filter=AsyncLLMRelevanceFilter(llm_client=cached_llm, system_prompt=RELEVANCE_PROMPT),
        entity_extractor=build_entity_extractor(config, cached_llm, logger),
        exporter=build_catalogue_exporter(),
        state_manager=AsyncFileStateManager(config.paths.processed_ids, config.paths.pipeline_metadata),
        destination=str(config.paths.raw_catalogue),
        logger=logger.warning,
        closeables=[http_client, llm_client, cache, *library.closeables],
        memory_ingestor=library.memory_ingestor(build_chunker(config.embeddings), logger.warning),
        shutdown=shutdown,
        on_event=progress_logger(logger.info),
        usage_sources=[llm_client, *library.usage_sources],
    )

library.memory_ingestor returns an AsyncCompositeIngestor that stores embedded chunks in an AsyncSqliteEmbeddingStore and indexes the paper in an AsyncSqliteFts5Store, or just the AsyncSearchIndexer when embeddings.enabled is false. The run itself shrinks to one async with block:

async def _run(build, config, logger, start_index, shutdown):
    library = open_paper_library(config)
    http_client = build_http_client(config)
    llm_client = build_llm_client(config)
    async with build(config, logger, http_client, llm_client, library, shutdown) as pipeline:
        return await pipeline.run(**run_arguments(config.pipeline, start_index))

Papers screened before the upgrade were never indexed, and udg-catalogue had no vector memory to backfill from. python main.py --index-papers fetches them again through the same pipeline with an entity extractor that returns nothing, a relevance filter that skips papers already in the text index, and a separate state manager (indexed_arxiv_ids.txt, indexing_meta.json), so the galaxy catalogue and its processed ids are left alone.

Moving to 0.5

udg-catalogue moved to >=0.5.1,<0.6. Two changes touched it:

  • Strict config. The library's config sections reject unknown keys, and udg-catalogue made its own sections (paths, embeddings, clustering, deduplication) strict as well, so a typo in config.yaml fails at startup with the key's name. The 0.2 names max_records and max_workers no longer load.
  • Parsed listing pages. Extractors return a ListingPage from fetch_page(query, cursor, page_size) instead of raw bytes from search and parse_listing, so the project's wrapper that logs each listing request forwards cursor_for_offset and fetch_page instead.

Moving to 0.6

udg-catalogue moved to >=0.6.0,<0.7. A base install needs only Pydantic since 0.6, so the project lists every extra it imports, adding config (YAML and .env), arxiv (Atom parsing), html, and processors (pandas):

sci-etl-core[config,async,arxiv,html,llm,pdf,processors,cluster,embeddings,embeddings-local,search]>=0.6.0,<0.7

The rest of the upgrade changed what the catalogue records:

  • One row per galaxy per paper. AsyncCsvUpsertExporter is gone. AsyncCsvExporter writes one row per galaxy, tagged with the paper's record_id, never merges, clips, or coerces a value, and replaces a paper's rows when the paper is extracted again. All merging moved into post-processing. The raw catalogue written by 0.5 has no record_id column, so the catalogue is rebuilt from scratch once, and post-processing refuses an old raw file with a message that says so.
  • The papers behind each galaxy. The post-processing chain starts with a small RawRowsStep that hides record_id and extra as _record_id and _extra and reads the measurements as numbers. Then DeduplicationStep(source_column="_record_id", sources_column="source_papers") lists the papers merged into each galaxy in the published catalogue.
  • Rejections with reasons, kept for review. GalaxyValidator.validate returns a Violation whose code names the broken rule (no-name, paper-local-name, simulation-keyword, not-a-number, not-positive, out-of-range, or no-measurement), where is_valid only said False. The extractor logs each rejection with its reason and stores it in an AsyncSqliteRejectionStore, where a reviewer can list and resolve it.
  • Standard logging. The logger= arguments and configure_logging are gone. The project's configure_run_logging attaches the run log's handlers to its own logger and to sci_etl_core, and a logging.Filter prefixes the library's lines with the arXiv id of the paper being processed.

  • Indexing never touches the catalogue. The --index-papers run passed the catalogue exporter along with an extractor that returns nothing. Since 0.6 the pipeline writes every processed record, including one without entities, and such a write clears that paper's rows. The indexing pipeline therefore gets an exporter that keeps nothing:

    class DiscardingExporter(AsyncExporter[Any]):
    async def write(self, record: RawRecord, entities: Sequence[Any]) -> None:
        return None

  • A run manifest. After each build, write_manifest records the sci-etl-core version, the model and base URL, hashes of both prompts, the query, the processed arXiv ids, and the row count and SHA-256 of both catalogues in data/run_manifest.json, which is committed next to the published catalogue.

What the migration uncovered

Moving code onto shared components forces you to state every rule precisely. udg-catalogue's migration surfaced these problems, most of them invisible in the old output:

  1. Name matching merged different galaxies. The old normalizer keyed any name containing digits, other than VCC names, as dragonfly<first number>: 1,201 of the 1,285 stored names (93%). KDG 44 matched DF 44, and NGC 1052-DF2 matched NGC 1052-DF4, so the upsert silently filled one galaxy's gaps with another's measurements. The fixed normalizer keeps all 1,285 stored names distinct, but rows merged by earlier runs can only be separated by re-extracting the catalogue.
  2. Failures looked like success. Nine arXiv search failures ended runs as if the listing were exhausted, and 8 DeepSeek errors marked papers processed with nothing extracted.
  3. Concurrent CSV writes had no lock. Six threads rewrote the same CSV, and the log records 8 worker exceptions in that loop.
  4. TLS verification was disabled for e-print downloads.
  5. The upsert duplicated galaxies named twice in one paper's extraction, found only by the export replay.
  6. Three stored galaxies have RA above 360°. The old constellation step wrapped them silently; they are now reported as Unknown.
  7. The dependency pins couldn't be installed on the Python version the README named: numpy==1.22.0 has no Python 3.11 wheels, and pandas==2.0.0 needs a newer numpy there.
  8. An editable install broke after the workspace moved to another folder.

Adapting this to your field

  • [ ] List every function in your pipeline and sort it into the three groups in Map your pipeline onto the library.
  • [ ] Install the library in editable mode, and decide how deployments will get it (a version range from PyPI, or a vendored wheel where PyPI is out of reach).
  • [ ] Move settings into BaseAppConfig sections; keep your API key's environment-variable name with api_key_env_var.
  • [ ] Move prompts over unchanged, and set result_key to the list key your extraction prompt already asks for.
  • [ ] Port your name-matching rule as a KeyNormalizer and your record rules as RecordValidators, unchanged at first.
  • [ ] Check whether your processed-id and offset files already match the state manager's format before writing an import script.
  • [ ] Express post-processing as a ProcessorChain, with your domain logic in NeighborMatcher, FeatureExtractor, and Processor plug-ins.
  • [ ] Replay real data through the old and new code and compare the outputs.
  • [ ] Only then fix the rules you've found wanting, one commit at a time.
  • [ ] Delete the old code and the tests that only covered it.

Questions or a rough edge in your migration? Open an issue — we're happy to help.