feat: make indexing and compile concurrency configurable

config.yaml.example

@@ -2,6 +2,12 @@ model: gpt-5.4                   # LLM model (any LiteLLM-supported provider)
 language: en                     # Wiki output language
 pageindex_threshold: 20          # PDF pages threshold for PageIndex
 
+# Optional: cap concurrent LLM calls during ingest (PageIndex indexing and
+# concept/entity compilation — they never overlap, so one setting covers
+# both). Lower it if you hit provider rate limits or "too many open files" on
+# large PDFs. Omit to let each stage apply its own default.
+# concurrency: 5
+
 # Optional: override the entity-type vocabulary used for entity pages.
 # Omit this key to use the default 7 types
 # (person, organization, place, product, work, event, other).

examples/configuration/README.md

@@ -70,6 +70,12 @@ model: gpt-5.4                   # LLM model (any LiteLLM-supported provider)
 language: en                     # Wiki output language
 pageindex_threshold: 20          # PDF pages threshold for PageIndex
 
+# Optional: cap concurrent LLM calls during ingest (PageIndex indexing and
+# concept/entity compilation — they never overlap, so one setting covers
+# both). Lower it if you hit provider rate limits or "too many open files" on
+# large PDFs. Omit to let each stage apply its own default.
+# concurrency: 5
+
 # Optional: override the entity-type vocabulary used for entity pages.
 # Omit this key to use the default 7 types
 # (person, organization, place, product, work, event, other).
@@ -95,6 +101,7 @@ pageindex_threshold: 20          # PDF pages threshold for PageIndex
 | `model` | `gpt-5.4` | LLM used for all compile/query/chat work. |
 | `language` | `en` | Language the wiki is written in. |
 | `pageindex_threshold` | `20` | PDFs with this many pages **or more** take the long-doc (PageIndex) path; shorter ones go through the short-doc path. See [`pageindex-cloud/`](../pageindex-cloud/). |
+| `concurrency` | `null` | Caps concurrent LLM calls OpenKB makes during ingest — both PageIndex's indexing of a long document and OpenKB's own concept/entity compilation. The two never run at once for the same document, so one setting covers both. Lower it if you hit provider rate limits or "too many open files" on large PDFs. `null` lets each stage apply its own default. |
 | `entity_types` | 7 defaults | Custom vocabulary for entity pages. `other` is always kept. |
 | `litellm:` | – | A pass-through block for LiteLLM. See below. |
 

openkb/cli.py

@@ -47,13 +47,14 @@ def filter(self, record: logging.LogRecord) -> bool:
 litellm.suppress_debug_info = True
 from dotenv import load_dotenv
 
-from openkb.agent.compiler import compile_long_doc
+from openkb.agent.compiler import DEFAULT_COMPILE_CONCURRENCY, compile_long_doc
 from openkb.config import (
     DEFAULT_CONFIG,
     load_config,
     save_config,
     load_global_config,
     register_kb,
+    resolve_concurrency,
     resolve_extra_headers,
     set_extra_headers,
     resolve_timeout,
@@ -555,6 +556,7 @@ def commit_body(snapshot) -> None:
                     kb_dir,
                     model,
                     doc_description=index_result.description,
+                    max_concurrency=resolve_concurrency(config) or DEFAULT_COMPILE_CONCURRENCY,
                 ),
                 label=f"Compiling long doc (doc_id={index_result.doc_id})",
             )
@@ -563,7 +565,13 @@ def commit_body(snapshot) -> None:
                 raise RuntimeError(f"Converted document has no source artifact: {file_path.name}")
             source_path = result.source_path
             _run_compile_with_retry(
-                lambda: compile_short_doc(doc_name, source_path, kb_dir, model),
+                lambda: compile_short_doc(
+                    doc_name,
+                    source_path,
+                    kb_dir,
+                    model,
+                    max_concurrency=resolve_concurrency(config) or DEFAULT_COMPILE_CONCURRENCY,
+                ),
                 label="Compiling short doc",
             )
 
@@ -684,6 +692,7 @@ def commit_body(_snapshot) -> None:
                         kb_dir,
                         model,
                         doc_description=cloud.description,
+                        max_concurrency=resolve_concurrency(config) or DEFAULT_COMPILE_CONCURRENCY,
                     ),
                     label=f"Compiling imported doc (doc_id={doc_id})",
                 )
@@ -1636,6 +1645,7 @@ def _classify(meta: dict) -> str:
     _setup_llm_key(kb_dir)
     config = load_config(openkb_dir / "config.yaml")
     model: str = config.get("model", DEFAULT_CONFIG["model"])
+    max_concurrency = resolve_concurrency(config) or DEFAULT_COMPILE_CONCURRENCY
 
     # Import lazily and reference via the module so tests can patch
     # ``openkb.agent.compiler.compile_*`` and see the call.
@@ -1671,7 +1681,16 @@ def _classify(meta: dict) -> str:
             click.echo(f"[{i}/{total}] Recompiling long doc {name}...")
             start = time.time()
             try:
-                asyncio.run(compiler.compile_long_doc(name, summary_path, doc_id, kb_dir, model))
+                asyncio.run(
+                    compiler.compile_long_doc(
+                        name,
+                        summary_path,
+                        doc_id,
+                        kb_dir,
+                        model,
+                        max_concurrency=max_concurrency,
+                    )
+                )
             except Exception as exc:
                 click.echo(f"  [ERROR] Compilation failed: {exc}")
                 logging.getLogger(__name__).debug("Recompile traceback:", exc_info=True)
@@ -1691,7 +1710,15 @@ def _classify(meta: dict) -> str:
             click.echo(f"[{i}/{total}] Recompiling short doc {name}...")
             start = time.time()
             try:
-                asyncio.run(compiler.compile_short_doc(name, source_path, kb_dir, model))
+                asyncio.run(
+                    compiler.compile_short_doc(
+                        name,
+                        source_path,
+                        kb_dir,
+                        model,
+                        max_concurrency=max_concurrency,
+                    )
+                )
             except Exception as exc:
                 click.echo(f"  [ERROR] Compilation failed: {exc}")
                 logging.getLogger(__name__).debug("Recompile traceback:", exc_info=True)

openkb/config.py

@@ -17,6 +17,13 @@
     "model": "gpt-5.4",
     "language": "en",
     "pageindex_threshold": 20,
+    # Cap on concurrent LLM calls OpenKB makes during ingest — both PageIndex's
+    # indexing of a long document and OpenKB's own concept/entity compilation.
+    # These never run concurrently with each other for the same document, so
+    # one knob covers both. None = each stage applies its own built-in default.
+    # Lower it if you hit provider rate limits or "too many open files"; raise
+    # it to go faster.
+    "concurrency": None,
 }
 
 # Default entity-type vocabulary. Overridable per-KB via the optional
@@ -176,6 +183,30 @@ def resolve_timeout(config: dict) -> float | None:
     return value
 
 
+def resolve_concurrency(config: dict) -> int | None:
+    """Resolve the optional ``concurrency:`` key — the cap on concurrent LLM
+    calls OpenKB makes during ingest (PageIndex indexing and concept/entity
+    compilation alike; they never run at the same time for one document, so
+    one setting covers both).
+
+    Returns ``None`` when absent, explicitly ``null``, or invalid (rejecting
+    bools and non-positive values with a warning when present but unusable —
+    an explicit ``null``/absent key is the normal "unset" case and stays
+    silent, matching ``resolve_timeout``). Callers apply their own default (or
+    omit the setting entirely) when this returns ``None``.
+    """
+    value = config.get("concurrency")
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        logger.warning(
+            "config: 'concurrency' must be a positive integer, got %r — ignoring it.",
+            value,
+        )
+        return None
+    return value
+
+
 def resolve_litellm_settings(config: dict) -> dict[str, Any]:
     """Resolve the optional ``litellm:`` mapping of LiteLLM module settings.
 

openkb/indexer.py

@@ -11,7 +11,7 @@
 
 from pageindex import IndexConfig, PageIndexClient
 
-from openkb.config import load_config
+from openkb.config import load_config, resolve_concurrency
 from openkb.tree_renderer import render_summary_md
 
 logger = logging.getLogger(__name__)
@@ -153,6 +153,33 @@ def _write_long_doc_artifacts(
     return summary_path
 
 
+def _build_index_config(config: dict[str, Any]) -> IndexConfig:
+    """Build the PageIndex ``IndexConfig`` for local indexing.
+
+    Forwards the KB's ``concurrency`` setting to PageIndex, which caps how many
+    indexing LLM calls run at once (guarding against "too many open files" fd
+    exhaustion on large documents). The value is only passed when set *and* the
+    installed PageIndex's ``IndexConfig`` declares the field, so OpenKB keeps
+    working against a pinned PageIndex that predates it (``IndexConfig``
+    forbids unknown kwargs).
+    """
+    kwargs: dict[str, Any] = {
+        "if_add_node_text": True,
+        "if_add_node_summary": True,
+        "if_add_doc_description": True,
+    }
+    concurrency = resolve_concurrency(config)
+    if concurrency is not None:
+        if "max_concurrency" in IndexConfig.model_fields:
+            kwargs["max_concurrency"] = concurrency
+        else:
+            logger.warning(
+                "config: 'concurrency' is set but the installed PageIndex "
+                "version does not support it yet — ignoring it."
+            )
+    return IndexConfig(**kwargs)
+
+
 def index_long_document(pdf_path: Path, kb_dir: Path, doc_name: str | None = None) -> IndexResult:
     """Index a long PDF document using PageIndex and write wiki pages.
 
@@ -166,11 +193,7 @@ def index_long_document(pdf_path: Path, kb_dir: Path, doc_name: str | None = Non
     model: str = config.get("model", "gpt-5.4")
     pageindex_api_key = os.environ.get("PAGEINDEX_API_KEY", "")
 
-    index_config = IndexConfig(
-        if_add_node_text=True,
-        if_add_node_summary=True,
-        if_add_doc_description=True,
-    )
+    index_config = _build_index_config(config)
 
     client = PageIndexClient(
         api_key=pageindex_api_key or None,

pyproject.toml

@@ -32,7 +32,7 @@ keywords = ["ai", "rag", "retrieval", "knowledge-base", "llm", "pageindex", "age
 # returning empty Responses output (BerriAI/litellm#25429) and
 # auto-injects GitHub Copilot IDE-auth headers.
 dependencies = [
-    "pageindex==0.3.0.dev1",
+    "pageindex==0.3.0.dev2",
     "markitdown[docx,pptx,xlsx,xls]==0.1.5",
     "trafilatura==2.0.0",
     "click==8.4.0",

tests/test_add_command.py

@@ -97,6 +97,29 @@ def test_add_single_file_compile_failure_rolls_back_converted_artifacts(self, tm
         assert not (kb_dir / "wiki" / "sources" / "notes.md").exists()
         assert HashRegistry(kb_dir / ".openkb" / "hashes.json").all_entries() == {}
 
+    def test_add_forwards_concurrency_from_config(self, tmp_path):
+        from unittest.mock import AsyncMock
+
+        from openkb.cli import add_single_file
+
+        kb_dir = self._setup_kb(tmp_path)
+        (kb_dir / ".openkb" / "config.yaml").write_text(
+            "model: gpt-4o-mini\nconcurrency: 3\n", encoding="utf-8"
+        )
+        doc = tmp_path / "notes.md"
+        doc.write_text("# Notes\n\nBody", encoding="utf-8")
+
+        with (
+            patch(
+                "openkb.agent.compiler.compile_short_doc", new_callable=AsyncMock
+            ) as mock_compile,
+            patch("openkb.cli._setup_llm_key"),
+        ):
+            outcome = add_single_file(doc, kb_dir)
+
+        assert outcome == "added"
+        assert mock_compile.call_args.kwargs["max_concurrency"] == 3
+
     def test_add_single_file_uses_add_mutation_coordinator(self, tmp_path):
         from openkb.cli import add_single_file
         from openkb.converter import ConvertResult

tests/test_config.py

@@ -5,6 +5,7 @@
     get_extra_headers,
     get_timeout,
     load_config,
+    resolve_concurrency,
     resolve_extra_headers,
     resolve_litellm_settings,
     resolve_timeout,
@@ -26,6 +27,53 @@ def test_default_config_values():
     assert DEFAULT_CONFIG["pageindex_threshold"] == 20
 
 
+def test_concurrency_defaults_to_none():
+    assert DEFAULT_CONFIG["concurrency"] is None
+
+
+def test_load_concurrency_override(tmp_path):
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text("concurrency: 12\n", encoding="utf-8")
+    assert load_config(config_path)["concurrency"] == 12
+
+
+def test_resolve_concurrency_absent_is_none():
+    assert resolve_concurrency({}) is None
+
+
+def test_resolve_concurrency_valid_value():
+    assert resolve_concurrency({"concurrency": 3}) == 3
+
+
+def test_resolve_concurrency_rejects_bool(caplog):
+    with caplog.at_level(logging.WARNING, logger="openkb.config"):
+        result = resolve_concurrency({"concurrency": True})
+    assert result is None
+    assert "concurrency" in caplog.text
+
+
+def test_resolve_concurrency_rejects_non_positive(caplog):
+    with caplog.at_level(logging.WARNING, logger="openkb.config"):
+        assert resolve_concurrency({"concurrency": 0}) is None
+    assert "concurrency" in caplog.text
+    caplog.clear()
+    with caplog.at_level(logging.WARNING, logger="openkb.config"):
+        assert resolve_concurrency({"concurrency": -1}) is None
+    assert "concurrency" in caplog.text
+
+
+def test_resolve_concurrency_rejects_non_int():
+    assert resolve_concurrency({"concurrency": "3"}) is None
+
+
+def test_resolve_concurrency_none_is_silent(caplog):
+    # Explicit null / absent is the normal "unset — caller applies its own
+    # default, or omits the setting entirely" case — no warning.
+    with caplog.at_level(logging.WARNING, logger="openkb.config"):
+        assert resolve_concurrency({"concurrency": None}) is None
+    assert caplog.text == ""
+
+
 def test_load_missing_file_returns_defaults(tmp_path):
     missing = tmp_path / "nonexistent" / "config.yaml"
     config = load_config(missing)