Utilities API
Supporting utilities for theme discovery, LLM management, and graph operations.
Theme Registry
Scans models/ recursively to discover theme folders and build a theme tree.
A folder is a theme iff it contains a catalog.py. Nested theme folders (sub-themes) are supported to any depth.
Build the plain-text model catalog block to inject into LLM decision prompts.
Format
Available annotation models:
-
ClassName — First line of docstring Fields: field1: str, field2: int | None
-
ContainerClass [list container] — Description Fields: items: list[ItemClass] Each item (ItemClass): field1: str, field2: str
This format is readable by all LLM families (Claude, OpenAI, Kimi, GLM, etc.)
and replaces the previous XML
Builds the
Called once at extraction startup; the result is static for the duration of the run (themes do not change at runtime).
Return the most specific available ThemeNode for detected_path.
Resolution order (most to least specific): 1. Exact match for the full detected path 2. Progressively shorter path prefixes 3. "default" theme (if available) 4. First registered theme (emergency fallback)
Examples:
detected_path="structural_specs/pharmaceutical/ema" → tries "structural_specs/pharmaceutical/ema" → not found → tries "structural_specs/pharmaceutical" → not found → tries "structural_specs" → FOUND, returns it
detected_path="unknown_theme" → tries "unknown_theme" → not found → falls back to "default"
Returns all registered theme paths sorted alphabetically.
Used by the extraction pipeline to build a dynamic Literal type for structured output validation.
Return sorted list of ALL valid model class names across ALL themes.
Return the full theme tree as a list of dicts for Neo4j ingestion.
Each dict has
path — str, e.g. "structural_specs/nta" name — str, e.g. "nta" parent_path — str | None (parent theme path, or None if top-level) model_names — list[str] of selectable class names
Build a formatted list of available themes for injection into the classification prompt.
Example output
- default: Generic fallback for content that does not fit a specific domain
- pharmaceutical: Pharmaceutical drug development documents following ICH CTD Module 3
- structural_specs: Documents that prescribe how other documents must be structured
LLM Utilities
utils/llm_factory.py — Deprecated. Use scinr_config instead.
This module exists for backward compatibility only.
Retry Utilities
utils/llm_retry.py — Generic LLM retry with exponential backoff.
Handles rate-limiting and transient errors from any LangChain-compatible LLM provider (AWS Bedrock, OpenAI, Anthropic, Ollama, etc.).
async
with_bedrock_retry(
coro_fn: Callable[[], Awaitable[Any]],
max_retries: int = _MAX_RETRIES,
) -> Any
Deprecated alias for with_llm_retry. Use with_llm_retry instead.
async
Execute an async LLM call with exponential backoff on retryable errors.
Parameters
coro_fn: Zero-argument async callable that invokes the LLM. max_retries: Maximum number of retry attempts (default 6).
Returns
Any The return value of coro_fn() on success.
Raises
The last exception if all retries are exhausted.
utils/neo4j_retry.py — Two-phase retry wrapper for transient Neo4j errors.
Neo4j can raise transient errors (e.g. DeadlockDetected, LockClientStopped, ServiceUnavailable / defunct connection) under write contention or parallel read overload. Without a retry mechanism, concurrent document ingestion would fail immediately on the first lock conflict or socket reset.
Public API
result = await with_neo4j_retry(lambda: session.run(...))
result = with_neo4j_retry_sync(lambda: session.run(...)) # sync mirror,
# for use inside asyncio.to_thread() worker threads
The wrapper uses a two-phase retry strategy:
Phase 1 — Exponential backoff with full jitter, capped at 60 s (7 attempts). Phase 2 — Fixed 60 s plateau, 5 additional attempts.
Total: 12 retry attempts maximum (~6.8 min worst-case wait). Non-transient exceptions are re-raised immediately without retrying.
module-attribute
Initial wait in seconds; doubles each exponential attempt.
module-attribute
Number of exponential-backoff attempts (attempts 0–6, delays 1 s → 60 s).
module-attribute
Upper cap on the exponential back-off delay and fixed plateau delay (seconds).
module-attribute
Number of fixed-plateau attempts after the exponential phase (60 s each).
module-attribute
Maximum total retry attempts after the initial failure (12).
async
with_neo4j_retry(
coro_fn: Callable[[], Coroutine[Any, Any, T]],
*,
exp_retries: int = EXP_RETRIES,
plateau_retries: int = PLATEAU_RETRIES,
base_delay: float = BASE_DELAY,
max_delay_exp: float = MAX_DELAY_EXP,
) -> T
Await coro_fn() with two-phase retry on transient Neo4j errors.
Parameters
coro_fn:
A zero-argument callable that returns an awaitable (e.g.
lambda: session.run(...)). It is called fresh on every attempt
so that a new coroutine object is created each time.
exp_retries:
Number of exponential-backoff retry attempts (Phase 1).
plateau_retries:
Number of fixed-plateau retry attempts after the exponential phase
(Phase 2).
base_delay:
Base wait time in seconds for the exponential phase.
max_delay_exp:
Cap on the exponential delay and the fixed plateau delay (seconds).
Returns
T Whatever coro_fn() returns on success.
Raises
Exception Re-raises the last transient exception once all retries are exhausted, or immediately re-raises any non-transient exception.
with_neo4j_retry_sync(
fn: Callable[[], T],
*,
exp_retries: int = EXP_RETRIES,
plateau_retries: int = PLATEAU_RETRIES,
base_delay: float = BASE_DELAY,
max_delay_exp: float = MAX_DELAY_EXP,
) -> T
Synchronous mirror of with_neo4j_retry(): uses time.sleep() instead of asyncio.sleep(). Designed to run inside an asyncio.to_thread() worker thread (no event loop of its own) — must never await.
Parameters
fn:
A zero-argument callable that returns a value synchronously (e.g.
lambda: session.run(...)). It is called fresh on every attempt.
exp_retries:
Number of exponential-backoff retry attempts (Phase 1).
plateau_retries:
Number of fixed-plateau retry attempts after the exponential phase
(Phase 2).
base_delay:
Base wait time in seconds for the exponential phase.
max_delay_exp:
Cap on the exponential delay and the fixed plateau delay (seconds).
Returns
T Whatever fn() returns on success.
Raises
Exception Re-raises the last transient exception once all retries are exhausted, or immediately re-raises any non-transient exception.
Concurrency
utils/neo4j_concurrency.py — Compatibility shim for Neo4j concurrency semaphore.
The semaphore logic has moved to config.py alongside get_llm_semaphore() and reset_llm_semaphore(). This module re-exports those functions for backward compatibility so that existing callers (annotation/nodes.py, entity_extraction/nodes.py) do not need to change their imports.
To configure concurrency, call configure(neo4j_concurrency=N) followed by reset_neo4j_semaphore() before running annotation or entity extraction stages.
Return (creating if needed) the global asyncio.Semaphore for Neo4j session concurrency.
The semaphore bounds the number of concurrent Neo4j sessions during annotation (Stage 3) and entity extraction (Stage 4). It is created lazily on the first call using the neo4j_concurrency value from ScinrConfig.
To change the concurrency at runtime, call configure(neo4j_concurrency=N) followed by reset_neo4j_semaphore() before running the relevant stages.
Logging
utils/logging_config.py — Logging setup for scinr-ingest.
Library mode (default)
When called without arguments, only a console handler is added — no files are created, no directories are touched. This is the correct behaviour for a library: the application that uses scinr-ingest is responsible for configuring file logging if it wants it.
CLI mode
When log_dir is provided (the CLI passes Path("logs")), two
rotating daily-folder file handlers are added in addition to the console
handler:
<log_dir>/
└── YYYY-MM-DD/
├── scinr.log ← INFO+, retained 30 days
└── scinr.errors.log ← ERROR+ only, retained 90 days
Usage::
# Library — console only, no files:
from scinr.newton.utils.logging_config import setup_logging
setup_logging()
# CLI — console + daily file rotation under ./logs/:
setup_logging(log_dir=Path("logs"))
Configure logging for scinr-ingest.
Parameters
log_dir:
Directory under which dated sub-folders and log files are created.
When None (the default), only a console handler is configured —
no files or directories are created. Pass an explicit path (e.g.
Path("logs")) to enable file logging.