RAG¶
Auto-generated from source docstrings (mkdocstrings). The complete public contract is agentmaker.__all__ plus each subpackage __all__.
agentmaker.rag: retrieval-augmented generation (RAG) subsystem, built on top of the agentmaker.retrieval base.
Reads files of various formats into a Document, splits them into Chunks, ingests them (source-of-truth store + retrieval index), performs retrieval-based question answering, and optionally applies Contextual Retrieval enhancement. - Document / Chunk: source document and text chunk - DocumentLoader / load_file: extension-dispatched file loading (txt/md/json/csv/pdf/docx/html) - Splitter / split_document: format-dispatched chunking (Markdown heading-aware / structured by record / plain text by token) - SourceStore / IngestionPipeline: chunk source-of-truth store and ingestion orchestrator (doc_id upsert dedup) - RagRetriever / RAGTool: retrieval-based question answering and agentic RAG tool - Contextualizer: adds context to chunks before ingestion (used for retrieval only) - QueryTransformer: query expansion before retrieval (MQE / HyDE, opt-in, off by default) - ChunkExpander / NeighborWindowExpander: post-retrieval chunk expansion (small-to-big, opt-in; lives in retriever.py alongside QueryTransformer)
Bases: ABC
Turn a chunk into "enhanced text for retrieval" (original chunk + context).
abstractmethod
¶
Return the enhanced retrieval text; does not modify the chunk itself. On LLM failure it should fall back to the original text / heading path on its own; but governance-class exceptions (RunLimitExceeded / RunCancelled) must propagate and not be swallowed, otherwise RunPolicy's limit / cancel stops applying to per-chunk enhancement.
The identity of this enhancer (goes into the doc ingestion fingerprint): defaults to the class name. A subclass whose behavior is affected by prompt / model / params (like LLMContextualizer) should override this and include them, otherwise re-ingesting a document after swapping the prompt / model would be wrongly skipped as "fingerprint unchanged" and the index would keep the stale enhanced text.
Bases: Contextualizer
Prepend the chunk's heading path to its body (zero LLM cost).
Example: heading_path="Expense Policy > Lodging" + body "tier-1 cities capped at 500 per night"
-> "Expense Policy > Lodging
tier-1 cities capped at 500 per night" This way the keyword in the heading (Lodging) enters the searched text, directly mitigating "chunking lost the heading".
LLMContextualizer(llm: LLMClient, *, max_doc_chars: int = 4000, context_prompt: Optional[str] = None, prompts=None, tracer=None)
Bases: Contextualizer
Use an LLM to generate one sentence of context per chunk (the original Anthropic version, stronger but one LLM call per chunk).
Note: cost = document chunk count x one LLM call. For large documents, pair with caching / a cheap model (e.g. deepseek).
max_doc_chars: The maximum number of characters of the whole document fed to the LLM
(truncated if too long, to control cost).
context_prompt: The system prompt for generating the chunk-context annotation; if omitted,
use the framework default (registry key rag.contextualize). Pass your own to switch
language.
prompts: Optional prompt registry (PromptRegistry); defaults to DEFAULT_PROMPTS if omitted.
context_prompt is a local shortcut override of it.
Call the LLM to generate one sentence of context and prepend it to the body; on failure fall back to the heading path / original text (without losing searchability).
The LLM sub-step during ingestion: driven via aio.run_sync over async governed_chat (keeping contextualize a synchronous interface, so as not to make the whole IngestionPipeline chain async; ingestion is an offline batch-processing path, consistent with the synchronous embedder).
Include class name + prompt + model + max_doc_chars: changing any one counts as "changed", so a re-ingest is not wrongly skipped (overrides the base class's plain class name).
dataclass
¶
RAG question-answering result (returned by RagRetriever.ask): answer text + list of sources.
Attributes:
| Name | Type | Description |
|---|---|---|
answer |
str
|
The answer text the model produced based on the retrieved material. |
sources |
list
|
list[SourceRef], the sources the answer relies on; [] when there are no hits / no chunks relied upon. |
dataclass
¶
Chunking knobs (used on the ingestion write path). Mirrors the splitter's _CHUNK_TOKENS/_OVERLAP_TOKENS defaults.
Attributes:
| Name | Type | Description |
|---|---|---|
chunk_tokens |
int
|
Target token count for a chunk. |
overlap_tokens |
int
|
Overlap token count between adjacent chunks (must satisfy 0 <= overlap < chunk). |
dataclass
¶
Chunk(content: str, chunk_id: str = (lambda: hex)(), doc_id: str = '', heading_path: str = '', index: int = 0, metadata: Dict[str, Any] = dict())
One text chunk.
Attributes:
| Name | Type | Description |
|---|---|---|
content |
str
|
The chunk body text. |
chunk_id |
str
|
Unique chunk id, auto-generated by default. |
doc_id |
str
|
The doc_id of the owning document (for tracing back / deleting by document). |
heading_path |
str
|
The heading path, e.g. "Chapter 1 > 1.1 Introduction" (only present for Markdown chunking). |
index |
int
|
The position within the document (0-based). |
metadata |
Dict[str, Any]
|
Accompanying information (source, format, etc.), useful for tracing provenance after retrieval. |
dataclass
¶
Document(content: str, doc_id: str = (lambda: hex)(), title: str = '', source: str = '', format: str = '', metadata: Dict[str, Any] = dict(), records: Optional[List[str]] = None)
One source document.
Attributes:
| Name | Type | Description |
|---|---|---|
content |
str
|
The body text (already converted by the loader into plain text / Markdown). |
doc_id |
str
|
Unique document id, auto-generated by default; used for dedup / upsert. |
title |
str
|
Document-level title (defaults to the file name), used as the leading prefix of each chunk's heading path so that "which file it belongs to" becomes part of the retrievable context. |
source |
str
|
The origin (file path or name). |
format |
str
|
The original format (md / json / pdf, etc.); the splitter picks its chunking strategy from this. |
metadata |
Dict[str, Any]
|
Accompanying information (upload time, author, etc.). |
records |
Optional[List[str]]
|
For structured data (json/csv), a list of "per-record texts" pre-split by the loader; RecordSplitter uses it directly, one record per chunk, without a round trip through "blank-line separated" text (which avoids mis-splitting records whose body contains blank lines / CSV multi-line cells). For unstructured or hand-constructed documents this is None, and RecordSplitter falls back to splitting on blank lines. |
dataclass
¶
The result of one ingestion (returned by IngestionPipeline.ingest_file / ingest_text).
Attributes:
| Name | Type | Description |
|---|---|---|
doc_id |
str
|
The doc_id of the ingested document (used for dedup / deleting by document). |
chunks |
int
|
The number of chunks ingested (or already present when short-circuited). |
skipped |
bool
|
True means the content fingerprint was unchanged and the whole document was short-circuited (chunking / embedding / LLM enhancement were all skipped). |
dataclass
¶
RAG multi-query expansion knobs (used on the retrieval read path, only relevant when a query_transformer is attached).
Attributes:
| Name | Type | Description |
|---|---|---|
mq_pool_factor |
int
|
The multiplier for how many extra candidates to fetch per query when doing multi-query. |
mq_max_queries |
int
|
The maximum fan-out count for query expansion. |
dataclass
¶
One source of a RAG answer (the number n corresponds to the [n] citation in the answer body).
Attributes:
| Name | Type | Description |
|---|---|---|
n |
int
|
The source number (1-based). |
content |
str
|
The original chunk content of this source. |
heading_path |
str
|
The chunk's heading path (only present for Markdown chunking). |
doc_id |
str
|
The doc_id of the owning document. |
IngestionPipeline(retriever: HybridRetriever, source_store: SourceStore, *, contextualizer: Optional[Contextualizer] = None, scope: Optional[Scope] = None, config: Optional[ChunkingConfig] = None, index_sync: Optional[IndexSync] = None, token_counter: TokenCounter = count_tokens)
Document ingestion orchestrator: coordinates loader -> splitter -> SourceStore + HybridRetriever.
source_store: The RAG source-of-truth store.
contextualizer: Optional Contextual Retrieval enhancer; if provided, the "enhanced text"
is pushed into the index while the original chunks are still stored in the source-of-truth
store. If omitted, the original chunks are used directly (backward compatible).
scope: The default ownership, isolating RAG data from memory; defaults to Scope(base="rag").
config: Optional chunking knobs (chunk_tokens / overlap_tokens); defaults to ChunkingConfig()
if omitted. The per-call kwargs of ingest_file / ingest_text can override these
(three-level resolution).
index_sync: Optional derived-index sync seam (IndexSync, shared with memory, see
agentmaker.retrieval.index_sync); ingestion goes through it for the atomic-replace
plus reconciliation. Defaults to SyncIndexSync(retriever) if omitted. For async /
distributed use, implement and inject your own.
token_counter: Pluggable token counter (defaults to count_tokens); chunking budgets are
estimated with it, so it is recommended to use the same ruler as the context budget.
classmethod
¶
from_config(config, *, embedder=None, retriever: Optional[HybridRetriever] = None, source_store: Optional[SourceStore] = None, db_path: str = ':memory:', reranker=None, contextualizer: Optional[Contextualizer] = None, index_sync: Optional[IndexSync] = None, token_counter: TokenCounter = count_tokens) -> IngestionPipeline
Assemble an IngestionPipeline from an AgentmakerConfig in one line: defaults to the sqlite backend; pass retriever / source_store to inject a custom backend.
Must share the same base with RagRetriever (the assembly root is in the app). Typical usage: first rag = RagRetriever.from_config(...), then IngestionPipeline.from_config(config, retriever=rag.retriever, source_store=rag.source_store), so the two read and write the same data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
AgentmakerConfig (reads config.chunking). |
required | |
embedder
|
Required when using the default sqlite base; not needed if a retriever is injected. |
None
|
|
retriever / source_store
|
Inject a custom backend (typically reuse the same instances built by RagRetriever); if omitted, a default sqlite one is built. |
required |
ingest_file(path: str, *, doc_id: Optional[str] = None, chunk_tokens: Optional[int] = None, overlap_tokens: Optional[int] = None) -> IngestReport
Read file -> chunk -> ingest. Returns IngestReport(doc_id / chunks / skipped); skipped=True when the content is unchanged and the run short-circuits. chunk/overlap default to self.cfg if omitted.
If doc_id is omitted, it is stably derived from the file's absolute path: re-ingesting the same file yields the same doc_id, which combined with the content fingerprint means an unchanged file short-circuits the whole run (no chunking, no per-chunk LLM enhancement, no embedding); a changed file atomically replaces the old version (upsert dedup).
ingest_text(text: str, *, source: str = '', title: Optional[str] = None, doc_id: Optional[str] = None, fmt: str = 'txt', chunk_tokens: Optional[int] = None, overlap_tokens: Optional[int] = None) -> IngestReport
Ingest a piece of text directly (no file). If title is omitted it is derived from source (its filename); if doc_id is omitted one is auto-generated. chunk/overlap default to self.cfg if omitted.
The concept of "re-ingest" only applies when doc_id is passed: unchanged content short-circuits the whole run (skipped=True), changed content atomically replaces the old version. Returns IngestReport.
Delete a whole document (first the retrieval index, then the source-of-truth store, and clear the ingestion fingerprint); returns the number of chunks deleted.
rebuild_index(*, scope: Optional[Scope] = None, batch_size: int = 256) -> int
Fully rebuild the retrieval index from the source-of-truth store (symmetric with Memory.rebuild_index): iterate over all chunks in the scope and, through the seam's reconcile, "delete orphans + force re-push in batches". Returns the number of chunks re-pushed.
Uses: migrating data when switching retrieval backends (sqlite -> pgvector), fully re-embedding after switching embedding models, and recovering from index corruption / a pending-repair set.
Boundary: re-pushing uses the original chunk text from the source-of-truth store. For a store with a Contextualizer attached, the enhanced text is not in the source of truth (intentional: the source of truth stays clean), so after a rebuild the index reverts to un-enhanced text; to rebuild with enhancement, re-ingest the source document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scope
|
Optional[Scope]
|
Which ownership scope to rebuild; defaults to this pipeline's scope. |
None
|
batch_size
|
int
|
The batch size for re-pushing. |
256
|
verify(*, scope: Optional[Scope] = None) -> dict
Cross-check consistency between the source-of-truth store and the retrieval index: return a divergence report without auto-repairing.
After a normal ingest, the chunk_id sets of the two are equal. If clearing old chunks fails or a crash leaves stale rows in the source of truth (the retrieval index was already atomically replaced but the source of truth was not cleaned), you get "source of truth is a superset of the indexed set"; those stale rows would be revived from the source of truth by rebuild_index. This method surfaces the divergence (and logs a warning), leaving the app to decide to re-ingest that document to fix it (one clean ingest converges it via replace).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scope
|
Optional[Scope]
|
Which ownership scope to check; defaults to this pipeline's scope. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
{"scope", "source_only" (ids present in the source of truth but not tracked by the index), "index_only" (ids tracked by the index but no longer in the source of truth), "consistent" (True if the two are equal; None if the seam does not support enumeration)}. |
Close the index-sync seam _sync (including the SqliteBookkeeping connection installed by default in from_config). source_store is owned by the caller and released per ownership rules.
async
¶
aingest_file(path: str, **kwargs) -> IngestReport
Async version of ingest_file (to_thread).
async
¶
aingest_text(text: str, **kwargs) -> IngestReport
Async version of ingest_text (to_thread).
async
¶
Async version of delete_document (to_thread).
RAGTool(pipeline: IngestionPipeline, rag_retriever: RagRetriever, *, top_k: int = 5, filter_fields: tuple = (), prompts=None)
Bases: Tool
Tool that lets an Agent manage a knowledge base and answer questions: add_text / add_document / search / ask / stats.
rag_retriever: Retrieval and question answering (search / ask).
top_k: How many chunks search / ask retrieve.
filter_fields: Optional; which metadata filter fields to expose to the model (e.g.
("doc_id", "tag"), contract in retrieval/types.py). These fields become optional
parameters of search / ask, and the model fills their values via function calling,
which is equivalent to self-query (natural language -> structured filter). The
retrieval backend must have declared filterable columns of the same name at index
build time (build_sqlite_hybrid's metadata_columns=). Not exposed by default.
Only add_document (reads disk, a red-line action) needs human confirmation; add_text / search / ask / stats do not touch disk and are allowed through.
The confirmation gate (ToolRegistry.execute_tool and HITL's harness.exec_tool) always reads this: without a confirm passed, it defaults to reject, so the Agent cannot read disk on its own. The safe default is guaranteed by the unified confirmation gate, and this tool need not hold its own callback.
get_parameters() -> List[ToolParameter]
Declare parameters; each filter field in filter_fields becomes an optional parameter of search / ask (model fills it = self-query).
run(parameters: dict) -> ToolResponse
Dispatch by action and return a result for the LLM to read (errors have status="error").
async
¶
arun(parameters: dict) -> ToolResponse
Native async version of run: ask (which calls the LLM) awaits the async ask directly; other actions run the synchronous logic via to_thread without blocking the event loop.
Note: add_text/add_document/search involve embedding (network) but are not multi-turn LLM calls, so they still go through the synchronous run, just dispatched to a thread pool (same as the base Tool.arun default), avoiding a synchronous embedding request on the event loop.
Bases: ABC
Abstract base for post-retrieval expansion: expand a hit's small chunk into a fuller context.
abstractmethod
¶
expand(results: List[RetrievalResult], *, source_store, scope=None) -> List[RetrievalResult]
Expand a batch of hits and return the expanded results (preserving the relevance-ordered input order).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
results
|
List[RetrievalResult]
|
Retrieval hits from RagRetriever (metadata contains doc_id / index). |
required |
source_store
|
RAG source-of-truth store (SourceStore, fetches neighbor chunks by doc_id + idx). |
required | |
scope
|
Retrieval scope (same scope as the hit, to avoid fetching chunks from a sibling scope). |
None
|
HyDETransformer(llm: LLMClient, *, include_original: bool = True, hyde_prompt: Optional[str] = None, prompts=None, tracer=None)
Bases: QueryTransformer
HyDE (hypothetical document embeddings): have the LLM first write a "hypothetical answer" and search with it. Matching an answer chunk with "answer wording" is more precise than matching with the question.
include_original: Whether to also retrieve with the original query (default True; in
hybrid retrieval the original query preserves the keyword-based recall path).
hyde_prompt: System prompt for generating the hypothetical document; if omitted the
framework default DEFAULT_HYDE_PROMPT is used. Pass your own to switch languages.
Generate a hypothetical answer text as the retrieval query (plus the optional original); fall back to [query] on LLM failure.
The LLM sub-step during retrieval is driven via aio.run_sync over async governed_chat (same as MQE, keeping the synchronous interface).
MultiQueryExpander(llm: LLMClient, *, n: int = 3, include_original: bool = True, expand_prompt: Optional[str] = None, prompts=None, tracer=None)
Bases: QueryTransformer
MQE (multi-query expansion): have the LLM rewrite one question into several phrasings, search each, and fuse with RRF. Addresses the mismatch between the user's wording and the document's wording.
n: How many rewrites to expect (>= 1; take whatever the LLM gives, up to n).
include_original: Whether to also retrieve the original query (default True: the
original query preserves the keyword-based recall path).
expand_prompt: System prompt for multi-query rewriting; if omitted the framework
default DEFAULT_MQE_PROMPT is used. Pass your own to switch languages.
Generate several rewritten queries (plus the optional original); fall back to [query] on LLM failure.
An LLM sub-step during retrieval: driven via aio.run_sync over async governed_chat (keeps transform's synchronous interface, so retrieve's whole chain does not become async: retrieval is a synchronous IO path, consistent with the synchronous embedder).
Bases: ChunkExpander
Default battery: neighbor-window expansion. A hit chunk together with the window chunks before/after it (same document, ordered by idx) is merged into one result.
Deduplication rule: each (doc_id, idx) is used at most once (assigned by relevance from highest to lowest); if a hit's window is already fully covered by an earlier, higher-ranked hit, that hit is skipped (to avoid duplicate content eating the budget). Hits without doc_id / index (e.g. non-RAG sources) are kept as-is.
For each hit, take neighbor chunks [idx-window, idx+window] and merge by idx; deduplicate across hits by (doc_id, idx).
Bases: ABC
Pre-retrieval query expansion (disabled by default): transform the user query into "queries that search better".
transform returns one or more retrieval queries; RagRetriever retrieves each one and fuses them by rank with RRF. See MultiQueryExpander (MQE) / HyDETransformer (HyDE) for implementations; both call the LLM and add one extra LLM cost per retrieval, hence opt-in.
abstractmethod
¶
Transform the original query into one or more retrieval queries (at least one).
On LLM failure it should fall back to [query] itself and not raise; however governance exceptions (RunLimitExceeded / RunCancelled) must propagate and must not be swallowed, otherwise RunPolicy limits / cancellation would be bypassed on the retrieval side path.
RagRetriever(retriever: HybridRetriever, source_store: SourceStore, llm: LLMClient, *, scope: Optional[Scope] = None, system_prompt: Optional[str] = None, query_transformer: Optional[QueryTransformer] = None, config: Optional[RetrievalConfig] = None, rag_config: Optional[RagConfig] = None, prompts=None, expander=None, tracer=None)
RAG retriever: retrieve document chunks and generate answers grounded in those chunks.
source_store: RAG source-of-truth store; fetch complete chunks from it after a hit.
llm: The LLM used to generate answers (a cheap model such as deepseek is recommended).
scope: Retrieval scope, defaults to Scope(base="rag"), isolated from memory.
system_prompt: Anti-hallucination system prompt for ask / ask_stream; if omitted the
framework default (DEFAULT_ASK_PROMPT) is used. The framework only provides the
mechanism; the specific wording / tone / business rules belong to the app: pass
your own to customize.
query_transformer: Optional query expander (MQE / HyDE); None by default (disabled),
retrieving with the original query directly. When provided, the query is rewritten
/ expanded into several queries before retrieval, each is retrieved, and results
are fused with RRF. Adds one extra LLM cost per retrieval, so enable on demand.
expander: Optional post-retrieval chunk expander (ChunkExpander, see above); None by
default (disabled). Pass NeighborWindowExpander(window=N) for small-to-big: small
chunks retrieve precisely, and after a hit the context is expanded into a neighbor
window to give the LLM fuller context.
tracer: Optional tracer (duck-typed emit); once attached, retrieve emits a rag_retrieve
event, and the LLM calls of ask and query expansion enter trace and RunPolicy
governance. Zero overhead when not attached.
classmethod
¶
from_config(config, *, embedder=None, retriever: Optional[HybridRetriever] = None, source_store: Optional[SourceStore] = None, llm: Optional[LLMClient] = None, db_path: str = ':memory:', reranker=None, query_transformer: Optional[QueryTransformer] = None, prompts=None) -> RagRetriever
Assemble a RagRetriever (read-only) from an AgentmakerConfig in one line: uses the sqlite backend by default; pass retriever / source_store to inject a custom backend.
The backend is pluggable, same as Memory.from_config (the assembly root is in the app). Ingestion uses a separate IngestionPipeline, and the two must share the same backend. Typical usage: build rag first, then IngestionPipeline.from_config(config, retriever=rag.retriever, source_store=rag.source_store).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
AgentmakerConfig (reads config.retrieval / config.rag). |
required | |
embedder
|
Required when using the default sqlite backend; not needed if retriever is injected. |
None
|
|
retriever / source_store
|
Inject a custom backend; if omitted a default sqlite one is built. |
required | |
llm
|
Optional[LLMClient]
|
Used to generate answers; can be omitted if you only retrieve without asking. |
None
|
Example
rag = RagRetriever.from_config(AgentmakerConfig(retrieval=RetrievalConfig(top_k=8)), embedder=emb, llm=llm)
retrieve(query: str, *, top_k: Optional[int] = None, scope: Optional[Scope] = None, filters=None) -> List[RetrievalResult]
Retrieve the most relevant chunks; after a hit, go back to the source-of-truth store to fill in complete information such as heading_path / doc_id.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
The query text. |
required |
top_k
|
Optional[int]
|
How many chunks to return; if omitted, uses self.cfg.top_k (None uses the default, and does not swallow 0). |
None
|
scope
|
Optional[Scope]
|
Retrieval scope; defaults to self.scope (fixed at construction). If provided, retrieval follows it (so context engineering can thread the run scope through). |
None
|
filters
|
Optional hard metadata filter (a list of MetadataFilter, see agentmaker.retrieval.types), passed to the backend for pre-filtering (e.g. search only a given doc_id / tag); the corresponding fields must have been declared via metadata_columns= when building the index. |
None
|
Returns:
| Type | Description |
|---|---|
List[RetrievalResult]
|
List[RetrievalResult]: The hit chunks; metadata contains heading_path / doc_id. |
async
¶
aretrieve(query: str, **kwargs) -> List[RetrievalResult]
Async version of retrieve (to_thread; the embedding network call does not block the event loop). Same parameters as retrieve.
async
¶
ask(query: str, *, top_k: Optional[int] = None, scope: Optional[Scope] = None, filters=None) -> AskResult
RAG question answering (non-streaming, async): retrieve -> assemble context -> LLM generation. Returns AskResult(answer, sources).
Retrieval goes through aretrieve (does not block the event loop), then awaits chat generation. Synchronous callers use agentmaker.core.aio.run_sync.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
The user's question. |
required |
top_k
|
Optional[int]
|
How many chunks to use as grounding. |
None
|
scope
|
Optional[Scope]
|
Retrieval scope; defaults to self.scope, retrieval follows it when provided (specify as needed in multi-user / multi-app scenarios). |
None
|
filters
|
Optional hard metadata filter, passed through to retrieve (e.g. answer only within a given doc_id / tag). |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
AskResult |
AskResult
|
answer text plus sources (list[SourceRef]; [] when there are no hits, which can be used to branch programmatically). |
async
¶
ask_stream(query: str, *, top_k: Optional[int] = None, scope: Optional[Scope] = None, filters=None)
RAG question answering (streaming, async): yield the answer text piece by piece (consume with async for). Sources can be obtained separately via retrieve before / after.
Retrieval goes through aretrieve, then yields piece by piece with async for (using llm.stream). Synchronous consumption uses agentmaker.core.aio.iter_sync.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
The user's question. |
required |
top_k
|
Optional[int]
|
How many chunks to use as grounding. |
None
|
scope
|
Optional[Scope]
|
Retrieval scope; defaults to self.scope. |
None
|
filters
|
Optional hard metadata filter, passed through to retrieve. |
None
|
RAG source-of-truth store: full Chunks stored into SQLite by (chunk_id, scope), recording doc_id to make per-document add/delete easy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
db_path
|
str
|
SQLite file path; the default ":memory:" is for self-tests only, use a file path in production to persist. |
':memory:'
|
Store chunks in batch (overwrites if the same (chunk_id, scope) already exists; the same chunk_id under a different scope is unaffected).
Fetch the full chunk by (chunk_id, scope); returns None if it does not exist.
scope uses B semantics (only filters non-empty dimensions), consistent with the retrieval base's search scope filtering: after a retrieval hit, fetch back with the same scope to avoid fetching a sibling chunk with the same chunk_id from a different scope.
chunk_ids_of_doc(doc_id: str, *, scope: Optional[Scope] = None) -> List[str]
List all chunk_ids of a document (within scope) (read-only, no delete). Lets the layer above get the ids to delete for "delete the index first, then the source of truth".
delete_chunks(chunk_ids: List[str], *, scope: Optional[Scope] = None) -> None
Precisely delete the given chunks by (chunk_id, scope) (all-dimension match, does not touch sibling rows with the same id under a different scope). Idempotent.
list_docs(*, scope: Optional[Scope] = None) -> dict
Return {doc_id: chunk count} (within the scope-limited range), for use by stats.
Fetch all chunks in the scope (ordered by doc_id, idx): the realization of "the index can be rebuilt from the authoritative copy", for rebuild_index's full re-push.
Mirrors MemoryStore.all: load all at once (local SQLite, manageable scale); batching happens on the index-write side of the re-push (reconcile batch_size).
get_doc_chunks(doc_id: str, *, index_range: Optional[tuple] = None, scope: Optional[Scope] = None) -> List[Chunk]
Fetch a document's chunks (ordered by idx); with index_range=(lo, hi), fetch only those whose index is in [lo, hi]: for neighbor-chunk / parent-chunk expansion.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
doc_id
|
str
|
The document id. |
required |
index_range
|
Optional[tuple]
|
Optional closed interval (lo, hi); if omitted, fetch the whole document. |
None
|
scope
|
Optional[Scope]
|
Ownership filter (B semantics); defaults to Scope(). |
None
|
Returns:
| Type | Description |
|---|---|
List[Chunk]
|
List[Chunk]: Ascending by in-document index. |
get_doc_hash(doc_id: str, *, scope: Optional[Scope] = None) -> Optional[str]
Read a document's ingestion fingerprint; returns None if there is none.
set_doc_hash(doc_id: str, content_hash: str, *, scope: Optional[Scope] = None) -> None
Register / overwrite a document's ingestion fingerprint (called after a successful ingest).
Splitter(chunk_tokens: int = _CHUNK_TOKENS, overlap_tokens: int = _OVERLAP_TOKENS, token_counter: TokenCounter = count_tokens)
load_file(path: str, *, max_bytes: int = 10 * 1024 * 1024) -> Document
Automatically pick a loader by extension to read the file; unknown extensions are treated as plain text.
All loaders dispatch through here, and the various file read / parse errors (IO, encoding, CSV, JSON, etc.) are normalized here to RetrievalError for a consistent external contract; a RetrievalError already raised inside a loader is passed through unchanged and not re-wrapped.
Before reading, set an upper bound on file size (max_bytes, default 10MB): exceeding it fails loud, to avoid one oversized file being read entirely into memory and blowing it up.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The file path. |
required |
max_bytes
|
int
|
The upper bound on file size (bytes); exceeding it raises RetrievalError. |
10 * 1024 * 1024
|
Returns:
| Name | Type | Description |
|---|---|---|
Document |
Document
|
The document after reading and normalization. |
register_loader(ext: str, loader: DocumentLoader) -> None
Register / override the loader for an extension (lets users extend custom formats).
split_document(doc: Document, *, chunk_tokens: int = _CHUNK_TOKENS, overlap_tokens: int = _OVERLAP_TOKENS, token_counter: TokenCounter = count_tokens) -> List[Chunk]
Automatically pick a splitter by doc.format and chunk.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
doc
|
Document
|
The source document. |
required |
chunk_tokens
|
int
|
Target token count per chunk. |
_CHUNK_TOKENS
|
overlap_tokens
|
int
|
Overlap token count between adjacent chunks. |
_OVERLAP_TOKENS
|
token_counter
|
TokenCounter
|
Pluggable token counter (default count_tokens); the chunking budget is estimated with it, and using the same ruler as the context budget is more accurate. |
count_tokens
|
Returns:
| Type | Description |
|---|---|
List[Chunk]
|
List[Chunk]: The resulting chunks; an empty document returns an empty list. |