Skip to content

Artifacts Reference

Documentation of the artifact system used for persisting and reusing pipeline outputs.

Artifact System Overview

The artifact system provides persistent storage and retrieval of intermediate pipeline results, enabling efficient reuse and resumption of work.

Core Components

ArtifactCollection

Container for all artifacts produced by a pipeline phase.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/execution.py 
32
33
34
35
36
37
38
@dataclass(slots=True)
class ArtifactCollection:
    artifacts: list[ArtifactEnvelope[ArtifactPayload]]

    def of_kind(self, *kinds: ArtifactKind) -> list[ArtifactEnvelope[ArtifactPayload]]:
        allowed = set(kinds)
        return [artifact for artifact in self.artifacts if artifact.kind in allowed]

ArtifactExecutionContext

Context information for artifact processing.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/execution.py 
285
286
287
288
289
290
@dataclass(slots=True)
class ArtifactExecutionContext:
    run_id: str
    manifest: RunManifest
    pipeline_input: PipelineInput
    previous: ArtifactCollection | None = None

Storage Implementation

JsonArtifactStore

Default artifact storage using JSON serialization.

Bases: ArtifactStoreProtocol

File-based artifact store that persists artifacts as JSON files.

Directory structure

artifacts_dir/ {run_id}/ {artifact_id}.json

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/store.py 
12
13
14
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class JsonArtifactStore(ArtifactStoreProtocol):
    """
    File-based artifact store that persists artifacts as JSON files.

    Directory structure:
        artifacts_dir/
            {run_id}/
                {artifact_id}.json
    """

    def __init__(self, root_dir: str | Path) -> None:
        self.root_dir = Path(root_dir)
        self.root_dir.mkdir(parents=True, exist_ok=True)
        # artifact_id -> path, filled as artifacts are written or looked up, so a
        # repeat lookup does not re-scan every run directory (O-16).
        self._path_index: dict[str, Path] = {}

    async def write_artifact(self, artifact: ArtifactEnvelope[ArtifactPayload]) -> None:
        """Write artifact to JSON file."""
        run_dir = self.root_dir / artifact.run_id
        run_dir.mkdir(parents=True, exist_ok=True)

        path = run_dir / f"{artifact.artifact_id}.json"
        path.write_text(
            json.dumps(artifact.model_dump(mode="json", by_alias=True), indent=2),
            encoding="utf-8",
        )
        self._path_index[artifact.artifact_id] = path

    async def get_artifact(
        self, artifact_id: str
    ) -> ArtifactEnvelope[ArtifactPayload] | None:
        """Read an artifact by id, searching every run directory if need be.

        Artifacts are stored under ``{run_id}/{artifact_id}.json`` and the run id
        is not recoverable from the artifact id, so the first lookup for an id
        written by another process has to scan. The resolved path is cached.
        """
        cached = self._path_index.get(artifact_id)
        if cached is not None and cached.exists():
            return ArtifactEnvelope.model_validate_json(
                cached.read_text(encoding="utf-8")
            )

        for run_dir in self.root_dir.iterdir():
            if not run_dir.is_dir():
                continue
            path = run_dir / f"{artifact_id}.json"
            if path.exists():
                self._path_index[artifact_id] = path
                return ArtifactEnvelope.model_validate_json(
                    path.read_text(encoding="utf-8")
                )
        return None

    async def list_run_artifacts(
        self, run_id: str
    ) -> list[ArtifactEnvelope[ArtifactPayload]]:
        """List all artifacts for a given run."""
        run_dir = self.root_dir / run_id
        if not run_dir.exists():
            return []

        artifacts = []
        for path in run_dir.glob("*.json"):
            try:
                artifact = ArtifactEnvelope.model_validate_json(
                    path.read_text(encoding="utf-8")
                )
                artifacts.append(artifact)
            except Exception:
                # Skip corrupted files
                continue
        return artifacts

    async def list_phase_artifacts(
        self, run_id: str, phase_name: str
    ) -> list[ArtifactEnvelope[ArtifactPayload]]:
        """List all artifacts for a given run and phase."""
        all_artifacts = await self.list_run_artifacts(run_id)
        return [a for a in all_artifacts if a.phase_name == phase_name]

get_artifact(artifact_id) async

Read an artifact by id, searching every run directory if need be.

Artifacts are stored under {run_id}/{artifact_id}.json and the run id is not recoverable from the artifact id, so the first lookup for an id written by another process has to scan. The resolved path is cached.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/store.py 
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
async def get_artifact(
    self, artifact_id: str
) -> ArtifactEnvelope[ArtifactPayload] | None:
    """Read an artifact by id, searching every run directory if need be.

    Artifacts are stored under ``{run_id}/{artifact_id}.json`` and the run id
    is not recoverable from the artifact id, so the first lookup for an id
    written by another process has to scan. The resolved path is cached.
    """
    cached = self._path_index.get(artifact_id)
    if cached is not None and cached.exists():
        return ArtifactEnvelope.model_validate_json(
            cached.read_text(encoding="utf-8")
        )

    for run_dir in self.root_dir.iterdir():
        if not run_dir.is_dir():
            continue
        path = run_dir / f"{artifact_id}.json"
        if path.exists():
            self._path_index[artifact_id] = path
            return ArtifactEnvelope.model_validate_json(
                path.read_text(encoding="utf-8")
            )
    return None

list_phase_artifacts(run_id, phase_name) async

List all artifacts for a given run and phase.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/store.py 
87
88
89
90
91
92
async def list_phase_artifacts(
    self, run_id: str, phase_name: str
) -> list[ArtifactEnvelope[ArtifactPayload]]:
    """List all artifacts for a given run and phase."""
    all_artifacts = await self.list_run_artifacts(run_id)
    return [a for a in all_artifacts if a.phase_name == phase_name]

list_run_artifacts(run_id) async

List all artifacts for a given run.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/store.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def list_run_artifacts(
    self, run_id: str
) -> list[ArtifactEnvelope[ArtifactPayload]]:
    """List all artifacts for a given run."""
    run_dir = self.root_dir / run_id
    if not run_dir.exists():
        return []

    artifacts = []
    for path in run_dir.glob("*.json"):
        try:
            artifact = ArtifactEnvelope.model_validate_json(
                path.read_text(encoding="utf-8")
            )
            artifacts.append(artifact)
        except Exception:
            # Skip corrupted files
            continue
    return artifacts

write_artifact(artifact) async

Write artifact to JSON file.

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/store.py 
29
30
31
32
33
34
35
36
37
38
39
async def write_artifact(self, artifact: ArtifactEnvelope[ArtifactPayload]) -> None:
    """Write artifact to JSON file."""
    run_dir = self.root_dir / artifact.run_id
    run_dir.mkdir(parents=True, exist_ok=True)

    path = run_dir / f"{artifact.artifact_id}.json"
    path.write_text(
        json.dumps(artifact.model_dump(mode="json", by_alias=True), indent=2),
        encoding="utf-8",
    )
    self._path_index[artifact.artifact_id] = path

Artifact Types

Entity Artifacts

Storage format for entity extraction results.

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
70
71
72
73
74
75
76
77
class EntityMentionArtifact(BaseModel):
    mention_id: str
    chunk_id: str
    surface_form: str
    entity_type: str
    start_char: int | None = None
    end_char: int | None = None
    confidence: float | None = None

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
80
81
82
83
84
85
86
87
88
class LinkedEntityArtifact(BaseModel):
    entity_id: str
    canonical_name: str
    entity_type: str
    mention_ids: list[str] = Field(default_factory=list)
    description: str | None = None
    textual_envelope: str | None = None
    is_mature: bool = False
    confidence: float | None = None

Relation Artifacts

Storage format for relationship extraction results.

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
91
92
93
94
95
96
97
98
class LocalRelationArtifact(BaseModel):
    relation_id: str
    subject_entity_id: str
    predicate: str
    object_entity_id: str
    source_chunk_id: str
    confidence: float
    scope: Literal["local"] = "local"

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
101
102
103
104
105
106
107
108
109
class GlobalRelationArtifact(BaseModel):
    relation_id: str
    subject_entity_id: str
    predicate: str
    object_entity_id: str
    supporting_chunk_ids: list[str] = Field(default_factory=list)
    confidence: float
    rerank_score: float | None = None
    scope: Literal["global"] = "global"

Argument Artifacts

Storage format for argument mining results.

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
127
128
129
130
131
132
133
134
135
class TheoryAtomArtifact(BaseModel):
    component_id: str
    chunk_id: str
    text: str
    component_type: str
    confidence: float | None = None
    plausibility: float | None = None
    epistemic_status: str | None = None
    scope_type: str | None = None

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
138
139
140
141
142
143
144
145
class TheoryRelationArtifact(BaseModel):
    relation_id: str
    source_component_id: str
    target_component_id: str
    relation_type: str
    scope: str
    confidence: float
    weight: float | None = None

Fusion Artifacts

Storage format for entity alignment results.

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
112
113
114
115
116
117
class FusionDecisionArtifact(BaseModel):
    decision_id: str
    artifact_ids: list[str] = Field(default_factory=list)
    decision_type: Literal["merge", "keep_separate", "cluster"]
    rationale: str | None = None
    confidence: float | None = None

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/artifacts/models.py 
120
121
122
123
124
class CanonicalizationArtifact(BaseModel):
    canonical_id: str
    original_artifact_ids: list[str] = Field(default_factory=list)
    canonical_label: str
    canonical_type: str

Content Addressing

Artifacts are stored using content-addressed identifiers to enable automatic deduplication.

Fingerprinting

Source code in packages/episteme-pipeline/episteme_pipeline/runs/fingerprints.py 
14
15
16
17
18
def stable_fingerprint(value: Any) -> str:
    payload = json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

Compute a fingerprint for a method object (e.g., LLM, embedding model, extractor).

Prefers an explicit contract: if the object exposes fingerprint() returning a JSON-serialisable value, that value is the fingerprint input. Implement it on any component whose behaviour depends on state this module cannot see — a SentenceTransformer revision, a quantisation setting, a reranker's activation function.

Otherwise this falls back to probing a fixed list of well-known attributes (model name, temperature, max_tokens, dimensions, …). That fallback is best-effort and silently incomplete by construction (O-15): anything not on the list is invisible to invalidation, so a run can be reused after a change that should have invalidated it.

Returns None if the object is None or exposes nothing recognisable.

Source code in packages/episteme-pipeline/episteme_pipeline/runs/fingerprints.py 
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def fingerprint_method(method_obj: Any) -> str | None:
    """Compute a fingerprint for a method object (e.g., LLM, embedding model, extractor).

    Prefers an explicit contract: if the object exposes ``fingerprint()``
    returning a JSON-serialisable value, that value *is* the fingerprint input.
    Implement it on any component whose behaviour depends on state this module
    cannot see — a SentenceTransformer revision, a quantisation setting, a
    reranker's activation function.

    Otherwise this falls back to probing a fixed list of well-known attributes
    (model name, temperature, max_tokens, dimensions, …). That fallback is
    best-effort and silently incomplete by construction (O-15): anything not on
    the list is invisible to invalidation, so a run can be reused after a change
    that should have invalidated it.

    Returns None if the object is None or exposes nothing recognisable.
    """
    if method_obj is None:
        return None

    explicit = getattr(method_obj, "fingerprint", None)
    if callable(explicit):
        try:
            value = explicit()
            if _is_fully_serializable(value):
                return stable_fingerprint(
                    {"class": method_obj.__class__.__name__, "fingerprint": value}
                )
        except Exception as ex:
            logger.exception(ex)
            pass

    fingerprint_data: dict[str, Any] = {"class": method_obj.__class__.__name__}
    model_name = getattr(method_obj, "model_name", None)
    if model_name:
        fingerprint_data["model"] = str(model_name)
    model_name_or_provider = getattr(method_obj, "model_name_or_provider", None)
    if model_name_or_provider:
        fingerprint_data["model_provider"] = str(model_name_or_provider)
    model = getattr(method_obj, "model", None)
    if model is not None:
        fingerprint_data["model_ref"] = str(model)

    # API version detection — important for LLM API changes
    api_version = getattr(method_obj, "api_version", None)
    if api_version is not None and _is_fully_serializable(api_version):
        fingerprint_data["api_version"] = str(api_version)

    # Provider/endpoint configuration
    base_url = getattr(method_obj, "base_url", None)
    if base_url is not None and _is_fully_serializable(base_url):
        fingerprint_data["base_url"] = str(base_url)

    # Runtime parameters that affect model output
    temperature = getattr(method_obj, "temperature", None)
    if temperature is not None and _is_fully_serializable(temperature):
        fingerprint_data["temperature"] = temperature

    max_tokens = getattr(method_obj, "max_tokens", None)
    if max_tokens is not None and _is_fully_serializable(max_tokens):
        fingerprint_data["max_tokens"] = max_tokens

    top_p = getattr(method_obj, "top_p", None)
    if top_p is not None and _is_fully_serializable(top_p):
        fingerprint_data["top_p"] = top_p

    n = getattr(method_obj, "n", None)
    if n is not None and _is_fully_serializable(n):
        fingerprint_data["n"] = n

    # Embedding-specific parameters
    dimensions = getattr(method_obj, "dimensions", None)
    if dimensions is not None and _is_fully_serializable(dimensions):
        fingerprint_data["dimensions"] = dimensions

    embedding_api_version = getattr(method_obj, "embedding_api_version", None)
    if embedding_api_version is not None and _is_fully_serializable(
        embedding_api_version
    ):
        fingerprint_data["embedding_api_version"] = str(embedding_api_version)

    # Catch model_kwargs / kwargs that may contain additional params
    model_kwargs = getattr(method_obj, "model_kwargs", None)
    if (
        model_kwargs
        and isinstance(model_kwargs, dict)
        and _is_fully_serializable(model_kwargs)
    ):
        fingerprint_data["model_kwargs"] = model_kwargs

    return stable_fingerprint(fingerprint_data)
Source code in packages/episteme-pipeline/episteme_pipeline/runs/fingerprints.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
def fingerprint_phase_config(config: Any) -> str:
    if hasattr(config, "model_dump"):
        return stable_fingerprint(config.model_dump(mode="json"))
    phase_dict = getattr(config, "__dict__", None)
    if isinstance(phase_dict, dict):
        serializable = {
            key: value
            for key, value in phase_dict.items()
            if _is_fully_serializable(value)
        }
        if serializable:
            return stable_fingerprint(serializable)
    return stable_fingerprint({"class": config.__class__.__name__})
Source code in packages/episteme-pipeline/episteme_pipeline/runs/fingerprints.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def fingerprint_existing_sources(
    source_paths: list[str], bib_paths: list[str] | None = None
) -> str:
    records: list[dict[str, Any]] = []
    all_paths = [("source", p) for p in sorted(source_paths)]
    if bib_paths:
        all_paths.extend(("bib", p) for p in sorted(bib_paths))

    for kind, source_path in all_paths:
        path = Path(source_path)
        if path.exists():
            stat = path.stat()
            try:
                content_hash = hashlib.sha256(path.read_bytes()).hexdigest()
            except (OSError, PermissionError):
                content_hash = None
            records.append(
                {
                    "kind": kind,
                    "path": source_path,
                    "size": stat.st_size,
                    "content_hash": content_hash,
                }
            )
        else:
            records.append({"kind": kind, "path": source_path, "missing": True})
    return stable_fingerprint(records)

Run Manifests

Tracking execution metadata and artifact lineage.

RunManifest

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/runs/models.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class RunManifest(BaseModel):
    run_id: str
    pipeline_version: str = "0.1.0"
    schema_version: str | None = None
    status: RunStatus = RunStatus.PLANNED
    created_at: datetime = Field(default_factory=utc_now)
    started_at: datetime | None = None
    completed_at: datetime | None = None
    source_fingerprint: str | None = None
    input_fingerprint: str | None = None
    input_fingerprint_inputs: dict[str, object] = Field(default_factory=dict)
    config_fingerprint: str | None = None
    method_fingerprints: dict[str, str] = Field(default_factory=dict)
    prompts_fingerprints: dict[str, str] = Field(default_factory=dict)
    phase_config_fingerprints: dict[str, object] = Field(default_factory=dict)
    config_snapshot: dict[str, object] = Field(default_factory=dict)
    input_sources: list[str] = Field(default_factory=list)
    phase_records: list[RunPhaseRecord] = Field(default_factory=list)
    tags: list[str] = Field(default_factory=list)
    parent_run_id: str | None = None

ExecutionResult

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/runs/models.py 
104
105
106
class ExecutionResult(BaseModel):
    manifest: RunManifest
    report: RunReport

Artifact Reuse

Mechanisms for detecting and reusing equivalent artifacts.

InvalidationDecision

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/runs/models.py 
69
70
71
72
73
class InvalidationDecision(BaseModel):
    resume_point: ResumePoint
    reused_phase_ordinals: list[int] = Field(default_factory=list)
    invalidated_phase_ordinals: list[int] = Field(default_factory=list)
    reason: str = ""

ResumePoint

Bases: BaseModel

Source code in packages/episteme-pipeline/episteme_pipeline/runs/models.py 
63
64
65
66
class ResumePoint(BaseModel):
    run_id: str | None = None
    phase_ordinal: int | None = None
    phase_name: str | None = None