LLM¶

Every name on this page is importable from sci_etl_core.llm.

sci_etl_core.llm.async_base ¶

AsyncLLMClient ¶

Bases: ABC

Contract for a chat-completion backend that answers with a JSON object.

The relevance filter and the entity extractor depend only on this interface, so a provider, a cache such as :class:~sci_etl_core.llm.cache_async.CachingLLMClient, or a test double can be injected in its place.

response_format `property` ¶

response_format: dict[str, Any]

The response_format requested with each completion, {"type": "json_object"} by default.

A subclass that requests another format overrides this property. :class:~sci_etl_core.llm.cache_async.CachingLLMClient keys its cache on the value, so answers requested in different formats never share a cache entry.

usage `property` ¶

usage: TokenUsage | None

Tokens this client has used so far, or None when it does not track usage.

complete_json `abstractmethod` `async` ¶

complete_json(
    system_prompt: str,
    user_content: str,
    timeout: int | None = None,
) -> dict[str, Any]

Send a chat completion request and return the parsed JSON object.

Raises:

Type	Description
`LLMError`	The request failed, or the body is not a JSON object.

complete_structured `async` ¶

complete_structured(
    system_prompt: str,
    user_content: str,
    schema: Mapping[str, Any],
    timeout: int | None = None,
) -> dict[str, Any]

Request a JSON object that should match the JSON Schema schema.

A provider that supports JSON-schema structured output is asked for it. The default calls :meth:complete_json, so the answer comes in JSON mode and the caller validates it against schema itself; :class:~sci_etl_core.llm.extraction_async.AsyncLLMEntityExtractor always does.

Raises:

Type	Description
`LLMError`	The request failed, or the body is not a JSON object.

invalidate `async` ¶

invalidate(
    system_prompt: str,
    user_content: str,
    *,
    schema: Mapping[str, Any] | None = None,
) -> None

Report that the response to this request was rejected as unusable.

schema is the one passed to :meth:complete_structured, or None for a :meth:complete_json request. :class:~sci_etl_core.llm.cache_async.CachingLLMClient removes the cached response, so the next request reaches the LLM again. A client that keeps no responses does nothing, which is the default.

sci_etl_core.llm.openai_compatible_async ¶

AsyncOpenAICompatibleClient ¶

AsyncOpenAICompatibleClient(
    api_key: str | SecretStr,
    base_url: str,
    model: str,
    default_timeout: int = 120,
    temperature: float = 0.0,
    max_retries: int = 3,
    backoff_factor: float = 2.0,
    sleep: Any = sleep,
    max_retry_after: float = 60.0,
    rate_limiter: RateLimiting | None = None,
    structured_output: bool = False,
)

Bases: AsyncLLMClient

JSON-mode chat completions from any OpenAI-compatible endpoint, with retries and token accounting.

The client owns its HTTP connection pool; close it with :meth:aclose.

Configure the client.

This client is the only retry layer: the OpenAI SDK's own retries are turned off, so max_retries is the total number of attempts. Between attempts it waits backoff_factor ** attempt seconds, or longer when the server's retry-after-ms or Retry-After header asks for it, up to max_retry_after seconds.

Every attempt first enters rate_limiter, an :class:~sci_etl_core.rate_limiter.AsyncRateLimiter or a :class:~sci_etl_core.rate_limiter.HostRateLimiter matched against base_url, and releases it once the response arrives. Share one limiter between a chat client and an embedder that call the same provider to keep both inside one budget.

Raises:

Type	Description
`ValueError`	`max_retries` is less than 1, which would fail every completion without making a single attempt, or `max_retry_after` is negative.

model `property` ¶

model: str

The model completions are requested from.

base_url `property` ¶

base_url: str

The API endpoint completions are requested from.

temperature `property` ¶

temperature: float

The sampling temperature sent with every completion.

structured_output `property` ¶

structured_output: bool

Whether :meth:complete_structured requests JSON-schema structured output.

usage `property` ¶

usage: TokenUsage

Tokens reported across every response received so far, as a snapshot.

from_config `classmethod` ¶

from_config(
    llm: LLMConfig, **options: Any
) -> AsyncOpenAICompatibleClient

Build a client from the llm config section.

The section supplies api_key, base_url, model, structured_output, and timeout as default_timeout. options pass any other constructor argument, such as rate_limiter, and override a value taken from the config.

complete_json `async` ¶

complete_json(
    system_prompt: str,
    user_content: str,
    timeout: int | None = None,
) -> dict[str, Any]

Request a JSON-mode completion and return the parsed object.

Every response the API returns counts toward :attr:usage, including one whose body is then rejected.

Raises:

Type	Description
`LLMError`	The request failed after retries, or the completion is empty, is not valid JSON, or is JSON other than an object. An empty completion is not evidence of an empty answer, so it fails the record instead of settling it.

complete_structured `async` ¶

complete_structured(
    system_prompt: str,
    user_content: str,
    schema: Mapping[str, Any],
    timeout: int | None = None,
) -> dict[str, Any]

Request an answer matching schema: as structured output when enabled, in JSON mode otherwise.

Raises:

Type	Description
`LLMError`	As for :meth:`complete_json`.

aclose `async` ¶

aclose() -> None

Close the underlying HTTP connection pool.

sci_etl_core.llm.relevance_async ¶

AsyncRelevanceFilter ¶

Bases: ABC

Contract for the gate that decides, from a listing entry alone, whether a record is worth its full text.

is_relevant `abstractmethod` `async` ¶

is_relevant(record: RawRecord) -> bool

Decide whether a record should proceed through the pipeline.

AsyncLLMRelevanceFilter ¶

AsyncLLMRelevanceFilter(
    llm_client: AsyncLLMClient,
    system_prompt: str,
    timeout: int = 20,
    default_on_empty_abstract: bool = True,
    default_on_error: bool = True,
)

Bases: AsyncRelevanceFilter

Ask an LLM whether a record's title and abstract match system_prompt.

Configure the filter.

system_prompt must ask for a JSON object with a relevant key. timeout is passed to every completion. A record without an abstract is not sent and reads as default_on_empty_abstract, which defaults to True.

default_on_error decides what an :class:~sci_etl_core.exceptions.LLMError or an unclear verdict does. True, the default, lets the record through, so a fault costs an extraction call rather than a record. False fails closed: the fault raises, the record is neither extracted nor marked processed, and the pipeline counts a failed attempt and retries it on the next run. A fault never reads as a verdict of irrelevance, since that would mark the record processed for good.

is_relevant `async` ¶

is_relevant(record: RawRecord) -> bool

Ask the LLM for a verdict read from the relevant key.

A boolean is used as-is; 0/1 and the strings "true", "false", "yes", "no", "1" and "0" (any case) are accepted too. A failed call, a response that is not a JSON object, and a missing or unrecognized verdict all count as a fault, so a garbled answer such as the string "false" read as truthy can never pass for a confident verdict. A fault returns True when default_on_error is True and raises otherwise. A response without a clear verdict is reported through :meth:~sci_etl_core.llm.async_base.AsyncLLMClient.invalidate, so a caching client does not replay it.

Raises:

Type	Description
`LLMError`	`default_on_error` is `False` and the call failed or gave no clear verdict.

sci_etl_core.llm.relevance_embedding_async ¶

AsyncEmbeddingRelevanceFilter ¶

AsyncEmbeddingRelevanceFilter(
    embedder: AsyncEmbedder,
    reference_texts: Sequence[str],
    threshold: float = 0.35,
    record_to_text: Callable[
        [RawRecord], str
    ] = _default_record_text,
    default_on_empty_abstract: bool = True,
    default_on_error: bool = True,
)

Bases: AsyncRelevanceFilter

Keep records whose meaning is close to a set of reference concepts.

Reference concepts are embedded once, lazily, on first use. Each record is then embedded and kept when its cosine similarity to the nearest reference reaches threshold, so semantically related work is matched even when it shares no keywords with the query.

Only a real similarity score can reject a record. A failed embedding call, a record vector that is missing, all zero or non-finite, and a vector whose dimension differs from the references are faults, not verdicts: none of them is evidence that the record is irrelevant, and a negative verdict would mark it processed for good. With default_on_error=True, the default, a fault lets the record through. With False it raises :class:~sci_etl_core.exceptions.EmbeddingError, so the record is neither extracted nor marked processed, and the pipeline counts a failed attempt and retries it on the next run.

is_relevant `async` ¶

is_relevant(record: RawRecord) -> bool

sci_etl_core.llm.extraction_async ¶

E `module-attribute` ¶

E = TypeVar('E', covariant=True)

M `module-attribute` ¶

M = TypeVar('M', bound=BaseModel)

AsyncEntityExtractor ¶

Bases: ABC, Generic[E]

Contract for turning a record's full text into structured entities.

E is the entity type: dict[str, Any] for untyped extraction, or a Pydantic model or dataclass for typed extraction. The pipeline calls :meth:extract_record, whose default calls :meth:extract, so an extractor that overrides only :meth:extract works unchanged.

An extractor that cannot work without the record, such as :class:~sci_etl_core.claims.extractor_async.AsyncLLMClaimExtractor, sets requires_record to True and overrides :meth:extract_record. An extractor that wraps another calls the inner one's :meth:extract_record and copies its requires_record.

requires_record `class-attribute` ¶

requires_record: bool = False

extract `abstractmethod` `async` ¶

extract(text: str | bytes) -> Sequence[E]

Extract structured entities from text alone.

An empty result means the text holds no entities, and the pipeline marks the record processed. Raising any exception instead fails the record, which stays unmarked and is retried on the next run.

Raises:

Type	Description
`TypeError`	`requires_record` is `True`; call :meth:`extract_record` instead.

extract_record `async` ¶

extract_record(
    record: RawRecord, text: str | bytes
) -> Sequence[E]

Extract entities from text, knowing which record it belongs to.

This is the entry point for every caller, the pipeline and wrapping extractors included. The default ignores record and calls :meth:extract.

AsyncLLMEntityExtractor ¶

AsyncLLMEntityExtractor(
    llm_client: AsyncLLMClient,
    system_prompt: str,
    *,
    schema: None = None,
    html_parser: Parser | None = None,
    result_key: str = "items",
    max_chars: int = 120000,
    timeout: int = 120,
    max_tokens: int | None = None,
    encoding_name: str = "cl100k_base",
    validator: RecordValidator | None = None,
    rejections: AsyncRejectionStore | None = None,
    label_field: str | None = None,
    now: Callable[[], datetime] | None = None,
)

AsyncLLMEntityExtractor(
    llm_client: AsyncLLMClient,
    system_prompt: str,
    *,
    schema: type[M],
    html_parser: Parser | None = None,
    result_key: str = "items",
    max_chars: int = 120000,
    timeout: int = 120,
    max_tokens: int | None = None,
    encoding_name: str = "cl100k_base",
    validator: RecordValidator | None = None,
    rejections: AsyncRejectionStore | None = None,
    label_field: str | None = None,
    now: Callable[[], datetime] | None = None,
)

AsyncLLMEntityExtractor(
    llm_client: AsyncLLMClient,
    system_prompt: str,
    *,
    schema: type[BaseModel] | None = None,
    html_parser: Parser | None = None,
    result_key: str = "items",
    max_chars: int = 120000,
    timeout: int = 120,
    max_tokens: int | None = None,
    encoding_name: str = "cl100k_base",
    validator: RecordValidator | None = None,
    rejections: AsyncRejectionStore | None = None,
    label_field: str | None = None,
    now: Callable[[], datetime] | None = None,
)

Bases: AsyncEntityExtractor[E]

Extract entities from full text with one LLM call per record.

Text that starts with markup is stripped with html_parser. The text is then truncated to max_tokens tokens when that is set and tiktoken is installed, or to max_chars characters otherwise.

Without schema, entities are the dict objects the model returned. With a Pydantic model class as schema, the request asks for JSON-schema structured output through :meth:~sci_etl_core.llm.async_base.AsyncLLMClient.complete_structured, every entity is validated against the model, and entities are model instances. A client that cannot request structured output answers in JSON mode, and the validation still applies, so the system prompt should still describe the fields.

Configure the extractor.

An entity that fails schema validation, or that validator rejects, is dropped and logged at INFO as Entity rejected by validation: <label> (<reasons>). The label is the entity's label_field value when that is given, and its position in the response otherwise. validator sees the entity as a dict, after the schema check when there is one. With rejections, each rejected entity is also stored there with its :class:~sci_etl_core.processors.validation.Violation list and the :attr:stamp, at the time now returns. A record whose every entity is rejected exports nothing and is still marked processed, as for a response with no entities.

answer_schema `property` ¶

answer_schema: dict[str, Any] | None

The JSON Schema requested for the whole answer, or None without schema.

stamp `property` ¶

stamp: ExtractionStamp

What produces this extractor's entities: model, prompt and schema hashes, and library version.

extract `async` ¶

extract(text: str | bytes) -> Sequence[E]

Extract entities from text with a single LLM call.

Rejected entities are stored with an empty record_id; the pipeline calls :meth:extract_record, which stores the record's id.

extract_record `async` ¶

extract_record(
    record: RawRecord, text: str | bytes
) -> Sequence[E]

Extract entities from text and file any rejections under record's id.

The entity list is read from result_key, or from the only value when the response has exactly one key. null reads as no entities and a lone object as a one-entity list. Preparing the text runs in a worker thread, since stripping and token counting are CPU-bound.

Raises:

Type Description

LLMError

The completion failed; the response holds no entity list, because it is empty or has several keys and none is result_key; or the entity list is not a list of objects. The error propagates instead of reading as "no entities", so the pipeline leaves the record unmarked and retries it on the next run. A rejected response is reported through :meth:~sci_etl_core.llm.async_base.AsyncLLMClient.invalidate, so a caching client does not replay it.

ClaimStoreError

The rejection store could not be written.

prepare ¶

prepare(text: str | bytes) -> str

Return the text the model is sent: markup stripped, then truncated.

Bytes are decoded as UTF-8, ignoring undecodable bytes. Text whose first 200 characters start with < is parsed with html_parser, by default an :class:~sci_etl_core.parsers.html.HtmlTextParser, which needs the html extra. Preparing already prepared plain text returns it unchanged.

entity_list_schema ¶

entity_list_schema(
    entity_schema: Any, result_key: str
) -> dict[str, Any]

Return the JSON Schema of an answer holding a list of entities under result_key.

entity_schema is a Pydantic model class or a JSON Schema mapping for one entity. Its $defs move to the root so that every $ref still resolves.

sci_etl_core.llm.cache_async ¶

AsyncLLMResponseCache ¶

Bases: ABC

Storage for parsed completions, keyed by :func:response_cache_key.

get `abstractmethod` `async` ¶

get(key: str) -> dict[str, Any] | None

Return the cached response for key, or None when there is none.

set `abstractmethod` `async` ¶

set(key: str, response: dict[str, Any]) -> None

Store response under key, replacing any earlier one.

clear `abstractmethod` `async` ¶

clear() -> None

Remove every cached response.

delete `async` ¶

delete(key: str) -> None

Remove the response cached under key, if there is one.

:class:CachingLLMClient calls this for a response a component rejected, so the next request reaches the LLM again. Both bundled backends implement it; a custom backend should too.

Raises:

Type	Description
`NotImplementedError`	The backend does not implement it. :class:`CachingLLMClient` logs this as a cache fault, and the rejected response stays cached.

InMemoryLLMResponseCache ¶

InMemoryLLMResponseCache(max_entries: int | None = None)

Bases: AsyncLLMResponseCache

A cache that lives as long as the process, optionally bounded.

With max_entries, the least recently used response is evicted once the cache is full. Responses are copied in and out, so a caller that changes a returned dict cannot change what is cached.

Create an empty cache.

Raises:

Type	Description
`ValueError`	`max_entries` is less than 1.

get `async` ¶

get(key: str) -> dict[str, Any] | None

set `async` ¶

set(key: str, response: dict[str, Any]) -> None

delete `async` ¶

delete(key: str) -> None

clear `async` ¶

clear() -> None

AsyncSqliteLLMResponseCache ¶

AsyncSqliteLLMResponseCache(
    path: str | Path,
    now: Callable[[], datetime] | None = None,
)

Bases: AsyncLLMResponseCache

A cache kept in a SQLite file, so responses survive between runs.

The file and its parent folder are created on first use, and responses are stored as JSON. Every SQLite failure, including a file that is not a database or a file written by a newer sci-etl-core, raises :class:~sci_etl_core.exceptions.LLMCacheError. The file records its schema version in PRAGMA user_version. Close it with :meth:aclose, for example by listing it in the pipeline's closeables. Use each instance from one event loop.

get `async` ¶

get(key: str) -> dict[str, Any] | None

set `async` ¶

set(key: str, response: dict[str, Any]) -> None

delete `async` ¶

delete(key: str) -> None

clear `async` ¶

clear() -> None

count `async` ¶

count() -> int

Return how many responses are cached.

aclose `async` ¶

aclose() -> None

Close the connection; a later call reopens it.

CacheStats `dataclass` ¶

CacheStats(hits: int = 0, misses: int = 0, faults: int = 0)

How many completions a :class:CachingLLMClient served from its cache.

hits `class-attribute` `instance-attribute` ¶

hits: int = 0

misses `class-attribute` `instance-attribute` ¶

misses: int = 0

faults `class-attribute` `instance-attribute` ¶

faults: int = 0

CachingLLMClient ¶

CachingLLMClient(
    client: AsyncLLMClient,
    cache: AsyncLLMResponseCache,
    model: str | None = None,
    *,
    variant: str = "",
)

Bases: AsyncLLMClient

Serve repeated completions from a cache instead of calling the LLM again.

A request is keyed on model, base_url, temperature, response_format, the schema of a :meth:complete_structured request, variant, and both prompts (:func:response_cache_key). base_url, temperature, and response_format are read from the wrapped client's attributes of those names, as :class:~sci_etl_core.llm.openai_compatible_async.AsyncOpenAICompatibleClient exposes them, and are left out of the key for a client without them. The timeout is not part of the key.

Only responses a component accepts stay cached. A failed call is never cached, and a response that :class:~sci_etl_core.llm.extraction_async.AsyncLLMEntityExtractor or :class:~sci_etl_core.llm.relevance_async.AsyncLLMRelevanceFilter rejects is removed through :meth:invalidate, so the next request reaches the LLM again instead of replaying the rejected answer.

The cache never fails a completion: an exception from the cache is logged as LLM cache <get|set|delete> failed: <error>, counted in :attr:stats, and the call goes to the LLM as on a miss. :attr:usage is the wrapped client's, so cache hits cost no tokens.

Wrap client.

model defaults to the wrapped client's model attribute. variant joins every key, so caching clients with different variants over one cache never serve one another's answers.

Raises:

Type	Description
`ValueError`	`model` is not given and the client has no `model` string to key the cache by.

model `property` ¶

model: str

The model name responses are cached under.

base_url `property` ¶

base_url: str | None

The endpoint responses are cached under, or None when the wrapped client has none.

temperature `property` ¶

temperature: float | None

The sampling temperature responses are cached under, or None when the wrapped client has none.

response_format `property` ¶

response_format: dict[str, Any]

The wrapped client's response_format, which responses are cached under.

stats `property` ¶

stats: CacheStats

Cache hits, misses, and faults so far, as a snapshot.

usage `property` ¶

usage: TokenUsage | None

complete_json `async` ¶

complete_json(
    system_prompt: str,
    user_content: str,
    timeout: int | None = None,
) -> dict[str, Any]

Return the cached response, or ask the wrapped client and cache its answer.

Raises:

Type	Description
`LLMError`	The wrapped client failed; nothing is cached.

complete_structured `async` ¶

complete_structured(
    system_prompt: str,
    user_content: str,
    schema: Mapping[str, Any],
    timeout: int | None = None,
) -> dict[str, Any]

Return the answer cached for this schema, or ask the wrapped client and cache its answer.

Raises:

Type	Description
`LLMError`	The wrapped client failed; nothing is cached.

invalidate `async` ¶

invalidate(
    system_prompt: str,
    user_content: str,
    *,
    schema: Mapping[str, Any] | None = None,
) -> None

Remove the cached response to this request and tell the wrapped client.

A cache failure is logged as LLM cache delete failed: <error> and counted in :attr:stats; it never raises.

response_cache_key ¶

response_cache_key(
    model: str,
    system_prompt: str,
    user_content: str,
    *,
    base_url: str | None = None,
    temperature: float | None = None,
    response_format: Mapping[str, Any] | None = None,
    schema: Mapping[str, Any] | None = None,
    variant: str = "",
) -> str

Return the cache key for a completion.

The key is a SHA-256 hex digest of the model, the endpoint's base_url, the sampling temperature, the requested response_format, the JSON Schema the answer must match, a variant label, and both prompts, so an answer cached for one provider, temperature, format, or schema is never served for another. The schema is serialized with sorted keys, so equal schemas give one key. variant keeps apart answers a caller wants kept apart, such as repeated samples of one request. Without a schema and with an empty variant, the key is the one sci-etl-core 0.5.1 computed.

LLM¶

sci_etl_core.llm.async_base ¶

AsyncLLMClient ¶

response_format property ¶

usage property ¶

complete_json abstractmethod async ¶

complete_structured async ¶

invalidate async ¶

sci_etl_core.llm.openai_compatible_async ¶

AsyncOpenAICompatibleClient ¶

model property ¶

base_url property ¶

temperature property ¶

structured_output property ¶

usage property ¶

from_config classmethod ¶

complete_json async ¶

complete_structured async ¶

aclose async ¶

sci_etl_core.llm.relevance_async ¶

AsyncRelevanceFilter ¶

is_relevant abstractmethod async ¶

AsyncLLMRelevanceFilter ¶

is_relevant async ¶

sci_etl_core.llm.relevance_embedding_async ¶

AsyncEmbeddingRelevanceFilter ¶

is_relevant async ¶

sci_etl_core.llm.extraction_async ¶

E module-attribute ¶

M module-attribute ¶

AsyncEntityExtractor ¶

requires_record class-attribute ¶

extract abstractmethod async ¶

extract_record async ¶

AsyncLLMEntityExtractor ¶

answer_schema property ¶

stamp property ¶

extract async ¶

extract_record async ¶

prepare ¶

entity_list_schema ¶

sci_etl_core.llm.cache_async ¶

AsyncLLMResponseCache ¶

get abstractmethod async ¶

set abstractmethod async ¶

clear abstractmethod async ¶

delete async ¶

InMemoryLLMResponseCache ¶

get async ¶

set async ¶

delete async ¶

clear async ¶

AsyncSqliteLLMResponseCache ¶

get async ¶

set async ¶

delete async ¶

clear async ¶

count async ¶

aclose async ¶

CacheStats dataclass ¶

hits class-attribute instance-attribute ¶

misses class-attribute instance-attribute ¶

faults class-attribute instance-attribute ¶

CachingLLMClient ¶

model property ¶

base_url property ¶

temperature property ¶

response_format property ¶

stats property ¶

usage property ¶

complete_json async ¶

complete_structured async ¶

invalidate async ¶

response_cache_key ¶

response_format `property` ¶

usage `property` ¶

complete_json `abstractmethod` `async` ¶

complete_structured `async` ¶

invalidate `async` ¶

model `property` ¶

base_url `property` ¶

temperature `property` ¶

structured_output `property` ¶

usage `property` ¶

from_config `classmethod` ¶

complete_json `async` ¶

complete_structured `async` ¶

aclose `async` ¶

is_relevant `abstractmethod` `async` ¶

is_relevant `async` ¶

is_relevant `async` ¶

E `module-attribute` ¶

M `module-attribute` ¶

requires_record `class-attribute` ¶

extract `abstractmethod` `async` ¶

extract_record `async` ¶

answer_schema `property` ¶

stamp `property` ¶

extract `async` ¶

extract_record `async` ¶

get `abstractmethod` `async` ¶

set `abstractmethod` `async` ¶

clear `abstractmethod` `async` ¶

delete `async` ¶

get `async` ¶

set `async` ¶

delete `async` ¶

clear `async` ¶

get `async` ¶

set `async` ¶

delete `async` ¶

clear `async` ¶

count `async` ¶

aclose `async` ¶

CacheStats `dataclass` ¶

hits `class-attribute` `instance-attribute` ¶

misses `class-attribute` `instance-attribute` ¶

faults `class-attribute` `instance-attribute` ¶

model `property` ¶

base_url `property` ¶

temperature `property` ¶

response_format `property` ¶

stats `property` ¶

usage `property` ¶

complete_json `async` ¶

complete_structured `async` ¶

invalidate `async` ¶