Add pluggable ingest bundle pipeline

Author: specode

examples/ingest-plugin/README.md

@@ -0,0 +1,27 @@
+# OpenKB Ingest Plugin Example
+
+This directory shows the minimal shape of an external package that contributes
+bundle ingest components through Python entry points.
+
+Install a package like this in the same environment as OpenKB, then enable its
+components in `.openkb/config.yaml`:
+
+```yaml
+ingest:
+  pipeline: bundle
+  importers:
+    enabled:
+      - example_text
+  normalizers:
+    enabled:
+      - example_text
+```
+
+The entry point groups are:
+
+- `openkb.ingest.importers`
+- `openkb.ingest.normalizers`
+- `openkb.ingest.enrichers`
+
+Each entry point should resolve to a class or factory returning an object that
+matches the corresponding OpenKB ingest protocol.

examples/ingest-plugin/openkb_example_ingest.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from openkb.ingest.models import DocumentBundle, IngestInput, ProvenanceRecord, TextBlock
+
+
+class ExampleTextImporter:
+    name = "example_text"
+
+    def can_handle(self, target: str, context) -> bool:
+        del context
+        return target.startswith("example-text:")
+
+    def import_source(self, target: str, context) -> IngestInput:
+        path = context.staging_dir / "raw" / "example.txt"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(target.removeprefix("example-text:").strip() + "\n", encoding="utf-8")
+        return IngestInput(
+            target=target,
+            path=path,
+            source_uri=target,
+            media_type="text/x-example",
+            metadata={"display_name": "example.txt"},
+        )
+
+
+class ExampleTextNormalizer:
+    name = "example_text"
+
+    def supports(self, input_: IngestInput, context) -> bool:
+        del context
+        return input_.media_type == "text/x-example"
+
+    def normalize(self, input_: IngestInput, context) -> DocumentBundle:
+        del context
+        if input_.path is None:
+            raise ValueError("Example normalizer requires a local path.")
+        source_uri = input_.source_uri or input_.target
+        text = input_.path.read_text(encoding="utf-8")
+        return DocumentBundle(
+            id=source_uri,
+            title="Example Text",
+            source_uri=source_uri,
+            blocks=[TextBlock(text)],
+            metadata={
+                "display_name": input_.metadata.get("display_name", input_.path.name),
+                "source_path": input_.path.as_posix(),
+                "source_identity": source_uri,
+            },
+            provenance=[ProvenanceRecord(source_uri=source_uri)],
+        )

examples/ingest-plugin/pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "openkb-example-ingest-plugin"
+version = "0.1.0"
+requires-python = ">=3.10"
+dependencies = ["openkb"]
+
+[project.entry-points."openkb.ingest.importers"]
+example_text = "openkb_example_ingest:ExampleTextImporter"
+
+[project.entry-points."openkb.ingest.normalizers"]
+example_text = "openkb_example_ingest:ExampleTextNormalizer"

openkb/cli.py

@@ -240,6 +240,17 @@ def _setup_llm_key(kb_dir: Path | None = None) -> None:
     ".csv",
 }
 
+BUNDLE_ONLY_EXTENSIONS = {
+    ".png",
+    ".jpg",
+    ".jpeg",
+    ".gif",
+    ".webp",
+    ".bmp",
+}
+
+BUNDLE_SUPPORTED_EXTENSIONS = SUPPORTED_EXTENSIONS | BUNDLE_ONLY_EXTENSIONS
+
 # Map raw doc types to display types
 _TYPE_DISPLAY_MAP = {
     "long_pdf": "pageindex",
@@ -987,9 +998,16 @@ def init(model, language):
     help="Import an already-indexed PageIndex Cloud document by its doc-id "
     "(no local file). Mutually exclusive with PATH.",
 )
+@click.option(
+    "--ingest-pipeline",
+    "ingest_pipeline",
+    type=click.Choice(["legacy", "bundle"]),
+    default=None,
+    help="Select the add ingest pipeline. Defaults to .openkb/config.yaml ingest.pipeline, then legacy.",
+)
 @click.pass_context
 @_with_kb_lock(exclusive=True)
-def add(ctx, path, from_pageindex_cloud):
+def add(ctx, path, from_pageindex_cloud, ingest_pipeline):
     """Add a document or directory of documents at PATH to the knowledge base.
 
     PATH may be a local file, a local directory (which is walked
@@ -1021,13 +1039,23 @@ def add(ctx, path, from_pageindex_cloud):
         click.echo("Provide a PATH or use --from-pageindex-cloud <DOC_ID>.")
         return
 
-    # URL ingest: download into raw/ first, then call add_single_file explicitly.
-    # Keep staged conversion enabled so converted source artifacts do not touch
-    # the live KB before the mutation snapshot exists. The tri-state outcome
-    # still lets us clean up the just-downloaded raw file on dedup.
+    from openkb.ingest.add import add_bundle_target, resolve_ingest_pipeline
+
+    config = load_config(kb_dir / ".openkb" / "config.yaml")
+    pipeline = resolve_ingest_pipeline(config, ingest_pipeline)
+    supported_extensions = (
+        BUNDLE_SUPPORTED_EXTENSIONS if pipeline == "bundle" else SUPPORTED_EXTENSIONS
+    )
+
+    # Legacy URL ingest downloads into raw/ first, then calls add_single_file.
+    # Bundle URL ingest keeps downloads inside the bundle staging directory and
+    # commits through the same mutation boundary as local bundle files.
     from openkb.url_ingest import looks_like_url, fetch_url_to_raw
 
     if looks_like_url(path):
+        if pipeline == "bundle":
+            add_bundle_target(path, kb_dir, legacy_fallback=add_single_file)
+            return
         fetched = fetch_url_to_raw(path, kb_dir)
         if fetched is None:
             return
@@ -1049,7 +1077,7 @@ def add(ctx, path, from_pageindex_cloud):
         files = [
             f
             for f in sorted(target.rglob("*"))
-            if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS
+            if f.is_file() and f.suffix.lower() in supported_extensions
         ]
         if not files:
             click.echo(f"No supported files found in {path}.")
@@ -1058,15 +1086,21 @@ def add(ctx, path, from_pageindex_cloud):
         click.echo(f"Found {total} supported file(s) in {path}.")
         for i, f in enumerate(files, 1):
             click.echo(f"\n[{i}/{total}] ", nl=False)
-            add_single_file(f, kb_dir)
+            if pipeline == "bundle":
+                add_bundle_target(f, kb_dir, legacy_fallback=add_single_file)
+            else:
+                add_single_file(f, kb_dir)
     else:
-        if target.suffix.lower() not in SUPPORTED_EXTENSIONS:
+        if target.suffix.lower() not in supported_extensions:
             click.echo(
                 f"Unsupported file type: {target.suffix}. "
-                f"Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
+                f"Supported: {', '.join(sorted(supported_extensions))}"
             )
             return
-        add_single_file(target, kb_dir)
+        if pipeline == "bundle":
+            add_bundle_target(target, kb_dir, legacy_fallback=add_single_file)
+        else:
+            add_single_file(target, kb_dir)
 
 
 def _stream_to_tty() -> bool:
@@ -1309,6 +1343,13 @@ def remove(ctx, identifier, keep_raw, keep_empty, dry_run, yes):
             )
         )
 
+    bundle_path = None
+    if meta.get("bundle_path"):
+        candidate = kb_dir / meta["bundle_path"]
+        if candidate.exists():
+            bundle_path = candidate
+            actions.append(("DELETE", str(candidate.relative_to(kb_dir))))
+
     # Scan concept pages to predict which will be edited vs. deleted.
     # Only frontmatter ``sources:`` membership drives the plan — body-only
     # references (e.g. a stray ``See also:`` line a user added by hand
@@ -1424,6 +1465,8 @@ def remove(ctx, identifier, keep_raw, keep_empty, dry_run, yes):
     source_json.unlink(missing_ok=True)
     if images_dir.is_dir():
         shutil.rmtree(images_dir, ignore_errors=True)
+    if bundle_path is not None:
+        bundle_path.unlink(missing_ok=True)
 
     concept_result = remove_doc_from_concept_pages(
         wiki_dir,

openkb/ingest/__init__.py

@@ -0,0 +1,7 @@
+"""Pluggable ingest pipeline for OpenKB."""
+
+from __future__ import annotations
+
+from openkb.ingest.add import add_bundle_target, resolve_ingest_pipeline
+
+__all__ = ["add_bundle_target", "resolve_ingest_pipeline"]