Skip to content

API Reference

compare

normpare

normpare — compare two editions of a technical standard.

Public facade of the package: compare() builds a configuration and runs the full pipeline in one call.

compare(old, new, out_dir, *, provider='anthropic', model=None, language='de', use_llm=True, batch=False, embed_fallback=False, embed_model=None, config=None)

Compare two versions of a standard and write all outputs to out_dir.

One-call facade: builds a :class:~normpare.config.Config and runs the :class:~normpare.pipeline.Pipeline.

Parameters:

Name Type Description Default
old str | Path

Path to the old version of the standard (PDF or DOCX).

 required
new str | Path

Path to the new version of the standard (PDF or DOCX).

 required
out_dir str | Path

Target directory for all output.

 required
provider str

LLM provider (anthropic | openai | ollama | chat).

 'anthropic'
model str | None

Model name at the provider (provider default if omitted).

 None
language str

Language of the compared standard (default "de").

 'de'
use_llm bool

If False, produce only the deterministic outputs (skip interpretation).

 True
batch bool

If True, submit the interpretation as one Anthropic message batch (~50% cheaper, asynchronous).

 False
embed_fallback bool

If True, run the optional embedding rescue pass (ADR-0001) after the deterministic alignment; needs the embeddings extra. Off by default.

 False
embed_model str | None

Embedding model for the rescue pass (default if omitted).

 None
config str | Path | None

Optional configuration file (JSON or TOML); when given it supplies the document metadata.

 None

Returns:

Type Description

A RunResult with the paths of the produced artifacts.
Source code in src/normpare/__init__.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def compare(
    old: str | Path,
    new: str | Path,
    out_dir: str | Path,
    *,
    provider: str = "anthropic",
    model: str | None = None,
    language: str = "de",
    use_llm: bool = True,
    batch: bool = False,
    embed_fallback: bool = False,
    embed_model: str | None = None,
    config: str | Path | None = None,
):
    """Compare two versions of a standard and write all outputs to ``out_dir``.

    One-call facade: builds a :class:`~normpare.config.Config` and runs the
    :class:`~normpare.pipeline.Pipeline`.

    Args:
        old: Path to the old version of the standard (PDF or DOCX).
        new: Path to the new version of the standard (PDF or DOCX).
        out_dir: Target directory for all output.
        provider: LLM provider (``anthropic`` | ``openai`` | ``ollama`` | ``chat``).
        model: Model name at the provider (provider default if omitted).
        language: Language of the compared standard (default ``"de"``).
        use_llm: If ``False``, produce only the deterministic outputs (skip interpretation).
        batch: If ``True``, submit the interpretation as one Anthropic message batch
            (~50% cheaper, asynchronous).
        embed_fallback: If ``True``, run the optional embedding rescue pass (ADR-0001) after
            the deterministic alignment; needs the ``embeddings`` extra. Off by default.
        embed_model: Embedding model for the rescue pass (default if omitted).
        config: Optional configuration file (JSON or TOML); when given it supplies the
            document metadata.

    Returns:
        A ``RunResult`` with the paths of the produced artifacts.
    """
    from .config import Config
    from .pipeline import Pipeline

    if config is not None:
        cfg = Config.from_file(config, out_dir)
        if embed_fallback:
            cfg.embed_fallback = True
        if embed_model:
            cfg.embed_model = embed_model
        if batch:
            cfg.llm.batch = True
    else:
        cfg = Config.for_compare(old, new, out_dir, provider=provider, model=model,
                                 language=language, embed_fallback=embed_fallback,
                                 embed_model=embed_model, batch=batch)
    return Pipeline(cfg).run("all", use_llm=use_llm)

Configuration

normpare.config

Configuration model for a comparison run.

Built either from explicit arguments (:meth:Config.for_compare, used by compare() and the CLI) or loaded from a JSON/TOML configuration file (:meth:Config.from_file).

Config dataclass

Everything a single comparison run needs.

Source code in src/normpare/config.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass
class Config:
    """Everything a single comparison run needs."""

    old: SourceSpec
    new: SourceSpec
    out_dir: str
    run_name: str = "run"
    pair_label: str = ""
    similarity: str = "tfidf"
    tau: float = 0.62
    tau_moved: float = 0.72
    language: str = "de"
    embed_fallback: bool = False
    embed_model: str = "jinaai/jina-embeddings-v2-base-de"
    tau_embed: float = 0.82
    llm: LlmConfig = field(default_factory=LlmConfig)

    @classmethod
    def for_compare(cls, old, new, out_dir, *, provider: str = "anthropic",
                    model: str | None = None, language: str = "de",
                    run_name: str | None = None, pair_label: str | None = None,
                    embed_fallback: bool = False, embed_model: str | None = None,
                    tau_embed: float = 0.82, batch: bool = False) -> "Config":
        """Build a config from explicit paths, deriving document metadata from the filenames."""
        o, n = Path(old), Path(new)
        return cls(
            old=SourceSpec(str(old), o.stem, o.stem, "old"),
            new=SourceSpec(str(new), n.stem, n.stem, "new"),
            out_dir=str(out_dir),
            run_name=run_name or (Path(out_dir).name or "run"),
            pair_label=pair_label or f"{o.stem} <-> {n.stem}",
            language=language,
            embed_fallback=embed_fallback,
            embed_model=embed_model or "jinaai/jina-embeddings-v2-base-de",
            tau_embed=tau_embed,
            llm=LlmConfig(provider=provider, model=model or DEFAULT_MODEL, batch=batch),
        )

    @classmethod
    def from_file(cls, path, out_dir) -> "Config":
        """Load a config from a JSON or TOML file (see the documentation for the layout)."""
        p = Path(path)
        text = p.read_text(encoding="utf-8")
        if p.suffix.lower() == ".toml":
            import tomllib
            data = tomllib.loads(text)
        else:
            data = json.loads(text)

        def _spec(d: dict) -> SourceSpec:
            return SourceSpec(d["path"], d["doc_id"], d["title"], d["version_label"])

        return cls(
            old=_spec(data["old"]), new=_spec(data["new"]), out_dir=str(out_dir),
            run_name=data.get("run_name", "run"), pair_label=data.get("pair_label", ""),
            similarity=data.get("similarity", "tfidf"),
            tau=data.get("tau", 0.62), tau_moved=data.get("tau_moved", 0.72),
            language=data.get("language", "de"),
            embed_fallback=data.get("embed_fallback", False),
            embed_model=data.get("embed_model", "jinaai/jina-embeddings-v2-base-de"),
            tau_embed=data.get("tau_embed", 0.82),
            llm=LlmConfig(
                provider=data.get("llm_provider", "anthropic"),
                model=data.get("llm_model", DEFAULT_MODEL),
                base_url=data.get("llm_base_url"), key_env=data.get("llm_key_env"),
                api_version=data.get("llm_api_version"), scope=data.get("deutung_scope", "core"),
                batch=data.get("llm_batch", False),
            ),
        )

for_compare(old, new, out_dir, *, provider='anthropic', model=None, language='de', run_name=None, pair_label=None, embed_fallback=False, embed_model=None, tau_embed=0.82, batch=False) classmethod

Build a config from explicit paths, deriving document metadata from the filenames.

Source code in src/normpare/config.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@classmethod
def for_compare(cls, old, new, out_dir, *, provider: str = "anthropic",
                model: str | None = None, language: str = "de",
                run_name: str | None = None, pair_label: str | None = None,
                embed_fallback: bool = False, embed_model: str | None = None,
                tau_embed: float = 0.82, batch: bool = False) -> "Config":
    """Build a config from explicit paths, deriving document metadata from the filenames."""
    o, n = Path(old), Path(new)
    return cls(
        old=SourceSpec(str(old), o.stem, o.stem, "old"),
        new=SourceSpec(str(new), n.stem, n.stem, "new"),
        out_dir=str(out_dir),
        run_name=run_name or (Path(out_dir).name or "run"),
        pair_label=pair_label or f"{o.stem} <-> {n.stem}",
        language=language,
        embed_fallback=embed_fallback,
        embed_model=embed_model or "jinaai/jina-embeddings-v2-base-de",
        tau_embed=tau_embed,
        llm=LlmConfig(provider=provider, model=model or DEFAULT_MODEL, batch=batch),
    )

from_file(path, out_dir) classmethod

Load a config from a JSON or TOML file (see the documentation for the layout).

Source code in src/normpare/config.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def from_file(cls, path, out_dir) -> "Config":
    """Load a config from a JSON or TOML file (see the documentation for the layout)."""
    p = Path(path)
    text = p.read_text(encoding="utf-8")
    if p.suffix.lower() == ".toml":
        import tomllib
        data = tomllib.loads(text)
    else:
        data = json.loads(text)

    def _spec(d: dict) -> SourceSpec:
        return SourceSpec(d["path"], d["doc_id"], d["title"], d["version_label"])

    return cls(
        old=_spec(data["old"]), new=_spec(data["new"]), out_dir=str(out_dir),
        run_name=data.get("run_name", "run"), pair_label=data.get("pair_label", ""),
        similarity=data.get("similarity", "tfidf"),
        tau=data.get("tau", 0.62), tau_moved=data.get("tau_moved", 0.72),
        language=data.get("language", "de"),
        embed_fallback=data.get("embed_fallback", False),
        embed_model=data.get("embed_model", "jinaai/jina-embeddings-v2-base-de"),
        tau_embed=data.get("tau_embed", 0.82),
        llm=LlmConfig(
            provider=data.get("llm_provider", "anthropic"),
            model=data.get("llm_model", DEFAULT_MODEL),
            base_url=data.get("llm_base_url"), key_env=data.get("llm_key_env"),
            api_version=data.get("llm_api_version"), scope=data.get("deutung_scope", "core"),
            batch=data.get("llm_batch", False),
        ),
    )

SourceSpec dataclass

One source document (old or new) with its metadata.

Source code in src/normpare/config.py 
15
16
17
18
19
20
21
22
@dataclass
class SourceSpec:
    """One source document (old or new) with its metadata."""

    path: str
    doc_id: str
    title: str
    version_label: str

LlmConfig dataclass

LLM backend settings for the interpretation stage.

Source code in src/normpare/config.py 
25
26
27
28
29
30
31
32
33
34
35
@dataclass
class LlmConfig:
    """LLM backend settings for the interpretation stage."""

    provider: str = "anthropic"
    model: str = DEFAULT_MODEL
    base_url: str | None = None
    key_env: str | None = None
    api_version: str | None = None
    scope: str = "core"
    batch: bool = False

Pipeline

normpare.pipeline

Pipeline orchestrator: runs the stages of a comparison and writes the JSON artifacts.

Orchestrates the stages in order, operating on the norm_doc.json dicts that the migrated stage functions expect. Heavy dependencies (lxml/PyMuPDF for ingest, numpy/scipy for alignment, python-docx/ python-pptx for reports) are imported lazily inside each stage, so importing this module stays cheap and dependency-free.

Pipeline

Runs the comparison stages for a :class:Config and writes artifacts to out_dir.

Source code in src/normpare/pipeline.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class Pipeline:
    """Runs the comparison stages for a :class:`Config` and writes artifacts to ``out_dir``."""

    def __init__(self, config: Config):
        self.cfg = config
        self.out = Path(config.out_dir)
        self.out.mkdir(parents=True, exist_ok=True)

    def run(self, stages: str | list[str] = "all", use_llm: bool = True) -> RunResult:
        """Run the given stage(s) in order. ``use_llm=False`` skips the interpretation stage."""
        names = STAGES if stages == "all" else ([stages] if isinstance(stages, str) else list(stages))
        for name in names:
            if name == "deutung" and not use_llm:
                continue
            if name not in STAGES:
                raise ValueError(f"unknown stage: {name}")
            getattr(self, name)()
        run = self.cfg.run_name
        return RunResult(
            out_dir=str(self.out), run_name=run,
            synopse_json=str(self.out / "synopse.json"),
            statistics_json=str(self.out / "statistics.json"),
            chapters_json=str(self.out / "chapters.json"),
            html=str(self.out / f"annotiert_{run}.html"),
            synopse_det=str(self.out / f"Synopse_deterministisch_{run}.docx"),
            synopse_final=str(self.out / f"Synopse_final_{run}.docx"),
            pptx=str(self.out / f"Aenderungen_{run}.pptx"),
        )

    # -- stages (heavy deps imported lazily, mirroring run.py) ------------------------

    def ingest(self) -> None:
        from .stages.ingest import read_document
        for side, spec in (("alt", self.cfg.old), ("neu", self.cfg.new)):
            src = Path(spec.path)
            if not src.exists():
                raise MissingInputError(f"source document missing: {src}")
            read_document(src, self.out / side, spec.doc_id, spec.title, spec.version_label)

    def enrich(self) -> None:
        from .stages.enrich import enrich_doc
        enrich_doc(self.out / "alt" / "norm_doc.json")
        enrich_doc(self.out / "neu" / "norm_doc.json")

    def map(self) -> None:
        from .stages.align.sections import build_section_mapping
        from .text.similarity import load_backend
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        backend = load_backend(self.cfg.similarity)
        records = build_section_mapping(old, new, backend)
        _write(self.out / "mapping.json", {"pair": self.cfg.pair_label, "records": records})
        _write(self.out / "alt" / "norm_doc.json", old)  # synthetic titles were added in place

    def align(self) -> None:
        from .stages.align.paras import align_document
        from .text.similarity import load_backend
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        mapping = _load(self.out / "mapping.json", "map")
        backend = load_backend(self.cfg.similarity)
        mapping["moved"] = align_document(old, new, mapping["records"], backend,
                                          tau=self.cfg.tau, tau_moved=self.cfg.tau_moved)
        if self.cfg.embed_fallback:
            from .stages.align.paras import embed_rescue_pass
            from .text.similarity import load_embed_backend
            embed = load_embed_backend(self.cfg.embed_model,
                                       cache_path=self.out / ".vec_cache.npz")
            mapping["embed_rescue"] = embed_rescue_pass(
                old, new, mapping["records"], embed, tau_embed=self.cfg.tau_embed)
            if getattr(embed, "cache", None) is not None:
                embed.cache.flush()
        _write(self.out / "mapping.json", mapping)

    def synopse(self) -> None:
        from .stages.diff import build_synopse
        from .text.similarity import load_backend
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        mapping = _load(self.out / "mapping.json", "align")
        backend = load_backend(self.cfg.similarity)
        build_synopse(old, new, mapping["records"], backend,
                      self.out / "synopse.json", self.cfg.pair_label)

    def keywords(self) -> None:
        from .stages.keywords import assign_keywords
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        assign_keywords(new, self.out / "keywords.json")
        assign_keywords(old)
        _write(self.out / "neu" / "norm_doc.json", new)
        _write(self.out / "alt" / "norm_doc.json", old)

    def deutung(self) -> None:
        from .stages.deutung import run_deutung
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        syn = _load(self.out / "synopse.json", "synopse")
        llm = self.cfg.llm
        run_deutung(syn, old, new, Path.cwd(), self.out,
                    model=llm.model, scope=llm.scope, provider=llm.provider,
                    base_url=llm.base_url, key_env=llm.key_env, api_version=llm.api_version,
                    batch=llm.batch, language=self.cfg.language,
                    title=self.cfg.new.title or self.cfg.pair_label)

    def report(self) -> None:
        from .report.csv_export import export_tables_csv
        from .report.dossier_report import build_dossier
        from .report.html import build_annotated_html
        from .report.pptx import build_pptx
        from .report.stats import build_statistics
        from .report.synopse_det import build_docx_synopse
        from .report.synopse_final import build_final_synopse
        run, pair = self.cfg.run_name, self.cfg.pair_label
        old = _load(self.out / "alt" / "norm_doc.json", "ingest")
        new = _load(self.out / "neu" / "norm_doc.json", "ingest")
        syn = _load(self.out / "synopse.json", "synopse")
        deut_path = self.out / "deutung.json"
        deut = json.loads(deut_path.read_text(encoding="utf-8")) if deut_path.exists() else None
        stats = build_statistics(old, new, syn, self.out / "statistics.json")
        build_dossier(new, old, syn, deut, self.out / "chapters.json")
        # tables as CSV assets (both versions)
        export_tables_csv(old, self.out / "alt")
        export_tables_csv(new, self.out / "neu")
        build_annotated_html(new, syn, deut, self.out / f"annotiert_{run}.html", pair,
                             old_doc=old, out_dir=self.out)
        # two synopses: deterministic (always) + final (LLM-interpreted, only when available)
        build_docx_synopse(syn, None, self.out / f"Synopse_deterministisch_{run}.docx", pair)
        if deut is not None:
            build_final_synopse(syn, deut, self.out / f"Synopse_final_{run}.docx", pair)
        build_pptx(syn, deut, stats, self.out / f"Aenderungen_{run}.pptx", pair)

run(stages='all', use_llm=True)

Run the given stage(s) in order. use_llm=False skips the interpretation stage.

Source code in src/normpare/pipeline.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def run(self, stages: str | list[str] = "all", use_llm: bool = True) -> RunResult:
    """Run the given stage(s) in order. ``use_llm=False`` skips the interpretation stage."""
    names = STAGES if stages == "all" else ([stages] if isinstance(stages, str) else list(stages))
    for name in names:
        if name == "deutung" and not use_llm:
            continue
        if name not in STAGES:
            raise ValueError(f"unknown stage: {name}")
        getattr(self, name)()
    run = self.cfg.run_name
    return RunResult(
        out_dir=str(self.out), run_name=run,
        synopse_json=str(self.out / "synopse.json"),
        statistics_json=str(self.out / "statistics.json"),
        chapters_json=str(self.out / "chapters.json"),
        html=str(self.out / f"annotiert_{run}.html"),
        synopse_det=str(self.out / f"Synopse_deterministisch_{run}.docx"),
        synopse_final=str(self.out / f"Synopse_final_{run}.docx"),
        pptx=str(self.out / f"Aenderungen_{run}.pptx"),
    )

RunResult dataclass

Paths of the artifacts produced by a run.

Source code in src/normpare/pipeline.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
@dataclass
class RunResult:
    """Paths of the artifacts produced by a run."""

    out_dir: str
    run_name: str
    synopse_json: str
    statistics_json: str
    chapters_json: str
    html: str
    synopse_det: str
    synopse_final: str
    pptx: str