Skip to content

teff.rag

teff.rag

Retrieval-augmented generation primitives.

Modules:

Name Description
base

Vector store abstract base and similarity utilities.
chunker

Text chunking strategies for RAG document splitting.
embedder

Embedding service using provider APIs.
image_tool

Image extraction tool — OCR via an OpenAI-compatible vision model.
pdf_tool

PDF extraction tool — turn a PDF into per-page text for RAG.
stores

Vector store implementations.
tool

RAG tool — retrieve context from a vector store for LLM use.

Classes:

Name Description
Chunker

Split text into chunks for embedding and retrieval.
Embedder

Convert text to vector embeddings using a provider API.
ImageTool

Extract text from an image with an OpenAI-compatible vision model.
PDFTool

Extract text from a PDF file, one section per page.
RAGTool

Tool that searches a vector store and returns ranked results.
VectorStore

Abstract interface for vector storage and similarity search.

Chunker dataclass

Split text into chunks for embedding and retrieval.

Supports three strategies:

  • token — Split on whitespace into token windows.
  • sentence — Split on sentence boundaries.
  • fixed — Split by fixed character count.

Attributes:

Name Type Description
strategy str

Chunking strategy name.
chunk_size int

Target chunk size (tokens, sentences, or chars).
overlap int

Overlap between consecutive chunks.

Methods:

Name Description
chunk

Split text into chunks using the configured strategy.
Source code in teff/rag/chunker.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class Chunker:
    """Split text into chunks for embedding and retrieval.

    Supports three strategies:

    - ``token`` — Split on whitespace into token windows.
    - ``sentence`` — Split on sentence boundaries.
    - ``fixed`` — Split by fixed character count.

    Attributes:
        strategy: Chunking strategy name.
        chunk_size: Target chunk size (tokens, sentences, or chars).
        overlap: Overlap between consecutive chunks.
    """

    strategy: str = "token"
    chunk_size: int = 500
    overlap: int = 50

    def chunk(self, text: str) -> list[str]:
        """Split *text* into chunks using the configured strategy."""
        if self.strategy == "token":
            return self._chunk_token(text)
        if self.strategy == "sentence":
            return self._chunk_sentence(text)
        if self.strategy == "fixed":
            return self._chunk_fixed(text)
        raise ValueError(f"unknown chunk strategy: {self.strategy}")

    def _chunk_token(self, text: str) -> list[str]:
        tokens = text.split()
        chunks = []
        start = 0
        while start < len(tokens):
            end = start + self.chunk_size
            chunk = " ".join(tokens[start:end])
            chunks.append(chunk)
            start += self.chunk_size - self.overlap
            if self.chunk_size - self.overlap <= 0:
                break
        return chunks

    def _chunk_sentence(self, text: str) -> list[str]:
        import re

        sentences = re.split(r"(?<=[.!?])\s+", text)
        chunks = []
        current = []
        for s in sentences:
            current.append(s)
            if len(current) >= self.chunk_size:
                chunks.append(" ".join(current))
                overlap_start = max(0, len(current) - self.overlap)
                current = current[overlap_start:]
        if current:
            chunks.append(" ".join(current))
        return chunks

    def _chunk_fixed(self, text: str) -> list[str]:
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunks.append(text[start:end])
            start += self.chunk_size - self.overlap
            if self.chunk_size - self.overlap <= 0:
                break
        return chunks

chunk 

chunk(text)

Split text into chunks using the configured strategy.

Source code in teff/rag/chunker.py 
26
27
28
29
30
31
32
33
34
def chunk(self, text: str) -> list[str]:
    """Split *text* into chunks using the configured strategy."""
    if self.strategy == "token":
        return self._chunk_token(text)
    if self.strategy == "sentence":
        return self._chunk_sentence(text)
    if self.strategy == "fixed":
        return self._chunk_fixed(text)
    raise ValueError(f"unknown chunk strategy: {self.strategy}")

Embedder dataclass

Convert text to vector embeddings using a provider API.

Attributes:

Name Type Description
provider str

Provider name (openai, ollama, mistral, voyage, jina, together, groq, ...). Any OpenAI-compatible /v1/embeddings endpoint works.
model str

Embedding model name; defaults to a per-provider model.
base_url str | None

Optional custom API base URL.
api_key_env str | None

Env var holding the API key; defaults to the per-provider env var.

Methods:

Name Description
embed

Embed a single text string.
embed_many

Embed multiple texts in a single API call.
Source code in teff/rag/embedder.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass
class Embedder:
    """Convert text to vector embeddings using a provider API.

    Attributes:
        provider: Provider name (``openai``, ``ollama``, ``mistral``,
            ``voyage``, ``jina``, ``together``, ``groq``, ...). Any
            OpenAI-compatible ``/v1/embeddings`` endpoint works.
        model: Embedding model name; defaults to a per-provider model.
        base_url: Optional custom API base URL.
        api_key_env: Env var holding the API key; defaults to the
            per-provider env var.
    """

    provider: str = "openai"
    model: str = ""
    base_url: str | None = None
    api_key_env: str | None = None

    def __post_init__(self):
        default_url, default_env, default_model = _EMBEDDER_DEFAULTS.get(
            self.provider, _EMBEDDER_DEFAULTS["openai"]
        )
        self._base_url = self.base_url or os.environ.get(
            f"{self.provider.upper()}_BASE_URL", default_url
        )
        env_key = default_env if self.api_key_env is None else self.api_key_env
        self._api_key = os.environ.get(
            f"{self.provider.upper()}_API_KEY",
            os.environ.get(env_key, ""),
        )
        if env_key and not self._api_key:
            raise ValueError(f"API key not found for provider {self.provider}")
        if not self.model:
            self.model = default_model

    async def embed(self, text: str) -> list[float]:
        """Embed a single text string."""
        results = await self.embed_many([text])
        return results[0]

    async def embed_many(self, texts: list[str]) -> list[list[float]]:
        """Embed multiple texts in a single API call."""
        headers = {"Content-Type": "application/json"}
        if self._api_key:
            headers["Authorization"] = f"Bearer {self._api_key}"
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                f"{self._base_url}/embeddings",
                headers=headers,
                json={
                    "model": self.model,
                    "input": texts,
                },
            )
            response.raise_for_status()
            data = response.json()
            return [
                item["embedding"]
                for item in sorted(data["data"], key=lambda x: x["index"])
            ]

embed async 

embed(text)

Embed a single text string.

Source code in teff/rag/embedder.py 
59
60
61
62
async def embed(self, text: str) -> list[float]:
    """Embed a single text string."""
    results = await self.embed_many([text])
    return results[0]

embed_many async 

embed_many(texts)

Embed multiple texts in a single API call.

Source code in teff/rag/embedder.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
async def embed_many(self, texts: list[str]) -> list[list[float]]:
    """Embed multiple texts in a single API call."""
    headers = {"Content-Type": "application/json"}
    if self._api_key:
        headers["Authorization"] = f"Bearer {self._api_key}"
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(
            f"{self._base_url}/embeddings",
            headers=headers,
            json={
                "model": self.model,
                "input": texts,
            },
        )
        response.raise_for_status()
        data = response.json()
        return [
            item["embedding"]
            for item in sorted(data["data"], key=lambda x: x["index"])
        ]

ImageTool

Bases: Tool

Extract text from an image with an OpenAI-compatible vision model.

The image is base64-encoded and sent to a chat-completions vision endpoint (default ollama/llava; openai/gpt-4o-mini is the API alternative). Use it for OCR on screenshots, scans, charts and photos.

Parameters:

Name Type Description Default
config dict | None

Optional dict with provider, model, base_url, api_key (falls back to the <PROVIDER>_API_KEY env var) and a default prompt.

 None

Methods:

Name Description
arun

OCR the image at path and return the transcribed text.
Source code in teff/rag/image_tool.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ImageTool(Tool):
    """Extract text from an image with an OpenAI-compatible vision model.

    The image is base64-encoded and sent to a chat-completions vision
    endpoint (default ``ollama``/``llava``; ``openai``/``gpt-4o-mini`` is
    the API alternative).  Use it for OCR on screenshots, scans, charts
    and photos.

    Args:
        config: Optional dict with ``provider``, ``model``, ``base_url``,
            ``api_key`` (falls back to the ``<PROVIDER>_API_KEY`` env var)
            and a default ``prompt``.
    """

    name = "image"
    description = "Extract text from an image using a vision model (OCR)"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        provider = cfg.get("provider", "ollama")
        default_url, default_env, default_model = _VISION_DEFAULTS.get(
            provider, _VISION_DEFAULTS["ollama"]
        )
        self.provider: str = provider
        self.model: str = cfg.get("model") or default_model
        self.base_url: str = cfg.get("base_url") or os.environ.get(
            f"{provider.upper()}_BASE_URL", default_url
        )
        self.api_key: str = (
            cfg.get("api_key")
            or os.environ.get(f"{provider.upper()}_API_KEY", "")
            or os.environ.get(default_env, "")
        )
        self.prompt: str = cfg.get(
            "prompt", "Extract all text visible in this image. Return only the text."
        )

    async def arun(  # type: ignore[override]
        self,
        path: str,
        prompt: str | None = None,
        max_chars: int = 50000,
    ) -> str:
        """OCR the image at *path* and return the transcribed text."""
        if not path:
            raise ValueError("path is required")
        if not os.path.exists(path):
            raise FileNotFoundError(path)
        with open(path, "rb") as f:
            data_url = (
                f"data:{_guess_mime(path)};base64,"
                f"{base64.b64encode(f.read()).decode('ascii')}"
            )

        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt or self.prompt},
                        {
                            "type": "image_url",
                            "image_url": {"url": data_url},
                        },
                    ],
                }
            ],
        }

        async with httpx.AsyncClient(timeout=120) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
            )
            response.raise_for_status()
            data = response.json()

        text = data["choices"][0]["message"].get("content") or ""
        if max_chars and max_chars > 0:
            return text[:max_chars]
        return text

arun async 

arun(path, prompt=None, max_chars=50000)

OCR the image at path and return the transcribed text.

Source code in teff/rag/image_tool.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def arun(  # type: ignore[override]
    self,
    path: str,
    prompt: str | None = None,
    max_chars: int = 50000,
) -> str:
    """OCR the image at *path* and return the transcribed text."""
    if not path:
        raise ValueError("path is required")
    if not os.path.exists(path):
        raise FileNotFoundError(path)
    with open(path, "rb") as f:
        data_url = (
            f"data:{_guess_mime(path)};base64,"
            f"{base64.b64encode(f.read()).decode('ascii')}"
        )

    headers = {"Content-Type": "application/json"}
    if self.api_key:
        headers["Authorization"] = f"Bearer {self.api_key}"

    payload = {
        "model": self.model,
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt or self.prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": data_url},
                    },
                ],
            }
        ],
    }

    async with httpx.AsyncClient(timeout=120) as client:
        response = await client.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
        )
        response.raise_for_status()
        data = response.json()

    text = data["choices"][0]["message"].get("content") or ""
    if max_chars and max_chars > 0:
        return text[:max_chars]
    return text

PDFTool

Bases: Tool

Extract text from a PDF file, one section per page.

Text-based PDFs are read with pypdf (extra teff[rag-pdf]). Scanned / image-only pages yield no text — feed those pages to :class:~teff.rag.image_tool.ImageTool instead.

Parameters:

Name Type Description Default
config dict | None

Optional dict. max_chars sets the default output limit (default 50000). Kept for config parity with other tools in a workflow tools: block.

 None

Methods:

Name Description
run

Return the PDF text as --- page N --- sections.
Source code in teff/rag/pdf_tool.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class PDFTool(Tool):
    """Extract text from a PDF file, one section per page.

    Text-based PDFs are read with ``pypdf`` (extra ``teff[rag-pdf]``).
    Scanned / image-only pages yield no text — feed those pages to
    :class:`~teff.rag.image_tool.ImageTool` instead.

    Args:
        config: Optional dict.  ``max_chars`` sets the default output
            limit (default 50000).  Kept for config parity with other
            tools in a workflow ``tools:`` block.
    """

    name = "pdf"
    description = "Extract text from a PDF file, one section per page"

    def __init__(self, config: dict | None = None):
        self.max_chars: int = 50000
        if isinstance(config, dict):
            self.max_chars = int(config.get("max_chars", 50000))

    def run(self, path: str, max_chars: int | None = None) -> str:  # type: ignore[override]
        """Return the PDF text as ``--- page N ---`` sections."""
        if not path:
            raise ValueError("path is required")
        docs = load_documents_pdf(path)
        if not docs:
            return "no text found in pdf"
        parts = [f"--- page {meta['page']} ---\n{text}" for text, meta in docs]
        result = "\n".join(parts)
        limit = max_chars if max_chars is not None else self.max_chars
        if limit and limit > 0:
            return result[:limit]
        return result

run 

run(path, max_chars=None)

Return the PDF text as --- page N --- sections.

Source code in teff/rag/pdf_tool.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
def run(self, path: str, max_chars: int | None = None) -> str:  # type: ignore[override]
    """Return the PDF text as ``--- page N ---`` sections."""
    if not path:
        raise ValueError("path is required")
    docs = load_documents_pdf(path)
    if not docs:
        return "no text found in pdf"
    parts = [f"--- page {meta['page']} ---\n{text}" for text, meta in docs]
    result = "\n".join(parts)
    limit = max_chars if max_chars is not None else self.max_chars
    if limit and limit > 0:
        return result[:limit]
    return result

RAGTool

Bases: Tool

Tool that searches a vector store and returns ranked results.

Usage::

store = InMemoryVectorStore(dim=768)
embedder = Embedder(provider="openai")
tool = RAGTool(store, embedder)
await tool.add_document("some long text")
result = await tool.arun(query="find this")

Can also be built from a config dict (e.g. a tools: entry in a workflow YAML)::

{
  "name": "rag_docs",  # optional; overrides the default "rag"
  "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
  "store": {"type": "in_memory", "dim": 768},
  "documents": [
    {"type": "csv", "path": "docs.csv"},
    {"type": "txt", "path": "corpus/*.txt"},
    {"type": "pdf", "path": "manual.pdf"},
    {"type": "excel", "path": "table.xlsx", "text_column": "content"},
  ],
  "filter": {"topic": "news"},        # metadata filter (DSL below)
  "similarity_threshold": 0.5,         # drop low-score hits
  "max_tokens": 1024,                  # context token budget
  "hybrid": true,                      # keyword + semantic blend
  "parent_chunks": true,               # keep full parent text per chunk
  "parent_retrieval": true,            # return whole parent documents
}

Supported document types (loaders): csv, txt (glob), pdf (teff[rag-pdf]), excel (teff[rag-excel]). Supported store types: in_memory (default), sqlite (stdlib file persistence), faiss, lance, chroma, qdrant, milvus, weaviate, pgvector, pinecone (via teff[embedding]). documents may also be a bare path (CSV shorthand) or a list of inline {"id": ..., "text": ...} dicts. Documents are embedded lazily on the first search.

Filter DSL: {"category": "news"} (equality), {"category": ["news", "tech"]} (membership), plus "$and" / "$or" keys combining sub-filters.

parent_chunks stores each chunk with a parent_id and the full parent_text; with parent_retrieval enabled, search returns whole parent documents (deduplicated) instead of individual chunks — the "small-to-big" pattern.

Methods:

Name Description
add_document

Chunk, embed, and store a document.
add_documents

Add multiple documents at once.
arun

Search documents and return formatted results.
Source code in teff/rag/tool.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
class RAGTool(Tool):
    """Tool that searches a vector store and returns ranked results.

    Usage::

        store = InMemoryVectorStore(dim=768)
        embedder = Embedder(provider="openai")
        tool = RAGTool(store, embedder)
        await tool.add_document("some long text")
        result = await tool.arun(query="find this")

    Can also be built from a config dict (e.g. a ``tools:`` entry in a
    workflow YAML)::

        {
          "name": "rag_docs",  # optional; overrides the default "rag"
          "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
          "store": {"type": "in_memory", "dim": 768},
          "documents": [
            {"type": "csv", "path": "docs.csv"},
            {"type": "txt", "path": "corpus/*.txt"},
            {"type": "pdf", "path": "manual.pdf"},
            {"type": "excel", "path": "table.xlsx", "text_column": "content"},
          ],
          "filter": {"topic": "news"},        # metadata filter (DSL below)
          "similarity_threshold": 0.5,         # drop low-score hits
          "max_tokens": 1024,                  # context token budget
          "hybrid": true,                      # keyword + semantic blend
          "parent_chunks": true,               # keep full parent text per chunk
          "parent_retrieval": true,            # return whole parent documents
        }

    Supported document types (loaders): ``csv``, ``txt`` (glob), ``pdf``
    (``teff[rag-pdf]``), ``excel`` (``teff[rag-excel]``). Supported store
    types: ``in_memory`` (default), ``sqlite`` (stdlib file persistence),
    ``faiss``, ``lance``, ``chroma``, ``qdrant``, ``milvus``, ``weaviate``,
    ``pgvector``, ``pinecone`` (via ``teff[embedding]``).
    ``documents`` may also be a bare path (CSV shorthand) or a list of
    inline ``{"id": ..., "text": ...}`` dicts. Documents are embedded
    lazily on the first search.

    Filter DSL: ``{"category": "news"}`` (equality), ``{"category":
    ["news", "tech"]}`` (membership), plus ``"$and"`` / ``"$or"`` keys
    combining sub-filters.

    ``parent_chunks`` stores each chunk with a ``parent_id`` and the full
    ``parent_text``; with ``parent_retrieval`` enabled, search returns
    whole parent documents (deduplicated) instead of individual chunks —
    the "small-to-big" pattern.
    """

    name = "rag"
    description = "Search documents using RAG"

    def __init__(
        self,
        config: dict | None = None,
        *,
        store: VectorStore | None = None,
        embedder: Embedder | None = None,
        chunker: Chunker | None = None,
        documents: list[tuple[str, dict]] | None = None,
        name: str | None = None,
        filter: dict | None = None,
        similarity_threshold: float | None = None,
        max_tokens: int | None = None,
        hybrid: bool = False,
        parent_chunks: bool = False,
        parent_retrieval: bool = False,
    ):
        self.store = store
        self.embedder = embedder
        self.chunker = chunker or Chunker()
        self._documents: list[tuple[str, dict]] = list(documents or [])
        self._seeded = False
        self._filters: dict | None = None
        self._threshold: float | None = None
        self._max_tokens: int | None = None
        self._hybrid = False
        self._parent_chunks = False
        self._parent_retrieval = False
        if isinstance(config, dict):
            self._apply_config(config)
        if filter is not None:
            self._filters = filter
        if similarity_threshold is not None:
            self._threshold = similarity_threshold
        if max_tokens is not None:
            self._max_tokens = max_tokens
        if hybrid:
            self._hybrid = True
        if parent_chunks:
            self._parent_chunks = True
        if parent_retrieval:
            self._parent_retrieval = True
        if name is not None:
            self.name = name

    def _apply_config(self, config: dict) -> None:
        if config.get("name"):
            self.name = config["name"]
        self.embedder = embedder_from_config(config)

        from teff.rag.stores.factory import store_from_config

        self.store = store_from_config(config.get("store") or {})

        if config.get("chunker"):
            self.chunker = Chunker(**config["chunker"])

        self._filters = config.get("filter") or config.get("filters")
        self._threshold = config.get("similarity_threshold")
        self._max_tokens = config.get("max_tokens")
        self._hybrid = bool(config.get("hybrid", False))
        self._parent_chunks = bool(config.get("parent_chunks", False))
        self._parent_retrieval = bool(config.get("parent_retrieval", False))

        documents = config.get("documents", [])
        if isinstance(documents, str):
            self._load_source({"path": documents})
        elif isinstance(documents, dict):
            self._load_source(documents)
        else:
            for doc in documents or []:
                if "text" in doc:
                    meta = {k: v for k, v in doc.items() if k != "text"}
                    self._documents.append((doc["text"], meta))
                else:
                    self._load_source(doc)

    def _load_source(self, cfg: dict) -> None:
        stype = cfg.get("type", "csv")
        loader = _DOCUMENT_LOADERS.get(stype)
        if loader is None:
            msg = f"unsupported document type: {stype}"
            raise ValueError(msg)
        kwargs = {k: v for k, v in cfg.items() if k != "type"}
        if "file" in kwargs and "path" not in kwargs:
            kwargs["path"] = kwargs.pop("file")
        self._documents.extend(loader(**kwargs))

    async def _ensure_seeded(self) -> None:
        if not self._seeded and self._documents:
            await self.add_documents(self._documents)
        self._seeded = True

    async def arun(  # type: ignore[override]
        self,
        query: str = "",
        k: int = 5,
        filter: dict | None = None,
        similarity_threshold: float | None = None,
        max_tokens: int | None = None,
        parent_retrieval: bool | None = None,
    ) -> str:
        """Search documents and return formatted results.

        Any optional argument overrides the value from the config for
        this call; ``None`` falls back to the configured default.
        """
        await self._ensure_seeded()
        assert self.embedder is not None
        assert self.store is not None

        eff_filter = filter if filter is not None else self._filters
        eff_threshold = (
            similarity_threshold
            if similarity_threshold is not None
            else self._threshold
        )
        eff_max_tokens = max_tokens if max_tokens is not None else self._max_tokens
        eff_parent = (
            parent_retrieval if parent_retrieval is not None else self._parent_retrieval
        )

        k_raw = max(k, k * 4) if eff_parent else k
        query_vec = await self.embedder.embed(query)
        results = await self.store.search(
            query_vec,
            k=k_raw,
            filter=eff_filter,
            hybrid=self._hybrid,
            query_text=query,
        )

        if eff_threshold is not None:
            results = [r for r in results if r[1] >= eff_threshold]

        if eff_parent:
            parents: dict[str, tuple[str, float]] = {}
            for doc_id, score, meta in results:
                pid = meta.get("parent_id")
                if pid is None:
                    continue
                parent_text = meta.get("parent_text") or meta.get("text", doc_id)
                if pid not in parents or score > parents[pid][1]:
                    parents[pid] = (parent_text, score)
            ranked = sorted(parents.items(), key=lambda kv: kv[1][1], reverse=True)
            results = [
                (pid, score, {"text": text}) for pid, (text, score) in ranked[:k]
            ]

        context_parts: list[str] = []
        total_tokens = 0
        for doc_id, score, meta in results:
            text = meta.get("text", doc_id)
            if eff_max_tokens is not None:
                tokens = _estimate_tokens(text)
                if total_tokens >= eff_max_tokens:
                    break
                if total_tokens + tokens > eff_max_tokens:
                    remaining = eff_max_tokens - total_tokens
                    text = _truncate_to_tokens(text, remaining)
                    total_tokens = eff_max_tokens
                else:
                    total_tokens += tokens
            context_parts.append(
                f"[{len(context_parts) + 1}] (score: {score:.3f}) {text}"
            )
        return "\n\n".join(context_parts)

    async def add_document(self, text: str, metadata: dict | None = None) -> None:
        """Chunk, embed, and store a document."""
        metadata = metadata or {}
        assert self.embedder is not None
        assert self.store is not None
        chunks = self.chunker.chunk(text)
        parent_id = metadata.get("id")
        if self._parent_chunks:
            parent_id = parent_id or f"doc_{uuid.uuid4().hex[:8]}"
            base_meta = {**metadata, "id": parent_id}
        else:
            base_meta = metadata
        vectors = []
        embeddings = await self.embedder.embed_many(chunks)
        for i, (chunk, vec) in enumerate(zip(chunks, embeddings)):
            if self._parent_chunks:
                doc_id = f"{parent_id}_{i}"
                meta = {
                    **base_meta,
                    "parent_id": parent_id,
                    "parent_text": text,
                    "text": chunk,
                    "chunk_index": i,
                }
            elif metadata.get("id"):
                doc_id = f"{metadata['id']}_{i}"
                meta = {**metadata, "text": chunk, "chunk_index": i}
            else:
                doc_id = f"chunk_{i}"
                meta = {**metadata, "text": chunk, "chunk_index": i}
            vectors.append((doc_id, vec, meta))
        await self.store.add(vectors)

    async def add_documents(self, docs: list[tuple[str, dict]]) -> None:
        """Add multiple documents at once."""
        for text, meta in docs:
            await self.add_document(text, meta)

add_document async 

add_document(text, metadata=None)

Chunk, embed, and store a document.

Source code in teff/rag/tool.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
async def add_document(self, text: str, metadata: dict | None = None) -> None:
    """Chunk, embed, and store a document."""
    metadata = metadata or {}
    assert self.embedder is not None
    assert self.store is not None
    chunks = self.chunker.chunk(text)
    parent_id = metadata.get("id")
    if self._parent_chunks:
        parent_id = parent_id or f"doc_{uuid.uuid4().hex[:8]}"
        base_meta = {**metadata, "id": parent_id}
    else:
        base_meta = metadata
    vectors = []
    embeddings = await self.embedder.embed_many(chunks)
    for i, (chunk, vec) in enumerate(zip(chunks, embeddings)):
        if self._parent_chunks:
            doc_id = f"{parent_id}_{i}"
            meta = {
                **base_meta,
                "parent_id": parent_id,
                "parent_text": text,
                "text": chunk,
                "chunk_index": i,
            }
        elif metadata.get("id"):
            doc_id = f"{metadata['id']}_{i}"
            meta = {**metadata, "text": chunk, "chunk_index": i}
        else:
            doc_id = f"chunk_{i}"
            meta = {**metadata, "text": chunk, "chunk_index": i}
        vectors.append((doc_id, vec, meta))
    await self.store.add(vectors)

add_documents async 

add_documents(docs)

Add multiple documents at once.

Source code in teff/rag/tool.py 
383
384
385
386
async def add_documents(self, docs: list[tuple[str, dict]]) -> None:
    """Add multiple documents at once."""
    for text, meta in docs:
        await self.add_document(text, meta)

arun async 

arun(
    query="",
    k=5,
    filter=None,
    similarity_threshold=None,
    max_tokens=None,
    parent_retrieval=None,
)

Search documents and return formatted results.

Any optional argument overrides the value from the config for this call; None falls back to the configured default.

Source code in teff/rag/tool.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
async def arun(  # type: ignore[override]
    self,
    query: str = "",
    k: int = 5,
    filter: dict | None = None,
    similarity_threshold: float | None = None,
    max_tokens: int | None = None,
    parent_retrieval: bool | None = None,
) -> str:
    """Search documents and return formatted results.

    Any optional argument overrides the value from the config for
    this call; ``None`` falls back to the configured default.
    """
    await self._ensure_seeded()
    assert self.embedder is not None
    assert self.store is not None

    eff_filter = filter if filter is not None else self._filters
    eff_threshold = (
        similarity_threshold
        if similarity_threshold is not None
        else self._threshold
    )
    eff_max_tokens = max_tokens if max_tokens is not None else self._max_tokens
    eff_parent = (
        parent_retrieval if parent_retrieval is not None else self._parent_retrieval
    )

    k_raw = max(k, k * 4) if eff_parent else k
    query_vec = await self.embedder.embed(query)
    results = await self.store.search(
        query_vec,
        k=k_raw,
        filter=eff_filter,
        hybrid=self._hybrid,
        query_text=query,
    )

    if eff_threshold is not None:
        results = [r for r in results if r[1] >= eff_threshold]

    if eff_parent:
        parents: dict[str, tuple[str, float]] = {}
        for doc_id, score, meta in results:
            pid = meta.get("parent_id")
            if pid is None:
                continue
            parent_text = meta.get("parent_text") or meta.get("text", doc_id)
            if pid not in parents or score > parents[pid][1]:
                parents[pid] = (parent_text, score)
        ranked = sorted(parents.items(), key=lambda kv: kv[1][1], reverse=True)
        results = [
            (pid, score, {"text": text}) for pid, (text, score) in ranked[:k]
        ]

    context_parts: list[str] = []
    total_tokens = 0
    for doc_id, score, meta in results:
        text = meta.get("text", doc_id)
        if eff_max_tokens is not None:
            tokens = _estimate_tokens(text)
            if total_tokens >= eff_max_tokens:
                break
            if total_tokens + tokens > eff_max_tokens:
                remaining = eff_max_tokens - total_tokens
                text = _truncate_to_tokens(text, remaining)
                total_tokens = eff_max_tokens
            else:
                total_tokens += tokens
        context_parts.append(
            f"[{len(context_parts) + 1}] (score: {score:.3f}) {text}"
        )
    return "\n\n".join(context_parts)

VectorStore

Bases: ABC

Abstract interface for vector storage and similarity search.

Implementations must provide add, search, and delete. The extended operations (count, list, get, update_metadata, clear) default to :class:NotImplementedError and are implemented by the built-in stores.

Methods:

Name Description
add

Store vectors with IDs and metadata.
clear

Remove all stored vectors.
count

Return the number of stored vectors.
delete

Remove vectors by ID.
entries

Return (id, metadata) pairs with pagination.
get

Return (id, metadata) pairs for existing IDs.
search

Search for the k nearest neighbours.
update_metadata

Merge metadata into the metadata of an existing ID.
Source code in teff/rag/base.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class VectorStore(ABC):
    """Abstract interface for vector storage and similarity search.

    Implementations must provide *add*, *search*, and *delete*.  The
    extended operations (*count*, *list*, *get*, *update_metadata*,
    *clear*) default to :class:`NotImplementedError` and are implemented
    by the built-in stores.
    """

    @abstractmethod
    async def add(self, vectors: list[tuple[str, list[float], dict]]) -> None:
        """Store vectors with IDs and metadata.

        Args:
            vectors: List of ``(id, embedding, metadata)`` tuples.
        """
        ...

    @abstractmethod
    async def search(
        self,
        query: list[float],
        k: int = 10,
        filter: dict | None = None,
        hybrid: bool = False,
        query_text: str | None = None,
    ) -> list[tuple[str, float, dict]]:
        """Search for the *k* nearest neighbours.

        Args:
            query: Query embedding.
            k: Maximum number of results.
            filter: Optional metadata filter DSL (see :func:`match_filter`).
            hybrid: When ``True``, blend a lexical keyword score with the
                cosine score (stores that support it; others ignore it).
            query_text: Original query text, required for ``hybrid``.

        Returns:
            List of ``(id, score, metadata)`` tuples sorted by score
            descending.  Scores are similarity-like (higher = more similar).
        """
        ...

    @abstractmethod
    async def delete(self, ids: list[str]) -> None:
        """Remove vectors by ID."""
        ...

    async def count(self) -> int:
        """Return the number of stored vectors."""
        raise NotImplementedError(f"{type(self).__name__} does not implement count()")

    async def entries(
        self, limit: int = 100, offset: int = 0
    ) -> list[tuple[str, dict]]:
        """Return ``(id, metadata)`` pairs with pagination."""
        raise NotImplementedError(f"{type(self).__name__} does not implement list()")

    async def get(self, ids: list[str]) -> list[tuple[str, dict]]:
        """Return ``(id, metadata)`` pairs for existing IDs."""
        raise NotImplementedError(f"{type(self).__name__} does not implement get()")

    async def update_metadata(self, id: str, metadata: dict) -> None:
        """Merge *metadata* into the metadata of an existing ID."""
        raise NotImplementedError(
            f"{type(self).__name__} does not implement update_metadata()"
        )

    async def clear(self) -> None:
        """Remove all stored vectors."""
        raise NotImplementedError(f"{type(self).__name__} does not implement clear()")

add abstractmethod async 

add(vectors)

Store vectors with IDs and metadata.

Parameters:

Name Type Description Default
vectors list[tuple[str, list[float], dict]]

List of (id, embedding, metadata) tuples.

 required
Source code in teff/rag/base.py 
92
93
94
95
96
97
98
99
@abstractmethod
async def add(self, vectors: list[tuple[str, list[float], dict]]) -> None:
    """Store vectors with IDs and metadata.

    Args:
        vectors: List of ``(id, embedding, metadata)`` tuples.
    """
    ...

clear async 

clear()

Remove all stored vectors.

Source code in teff/rag/base.py 
151
152
153
async def clear(self) -> None:
    """Remove all stored vectors."""
    raise NotImplementedError(f"{type(self).__name__} does not implement clear()")

count async 

count()

Return the number of stored vectors.

Source code in teff/rag/base.py 
131
132
133
async def count(self) -> int:
    """Return the number of stored vectors."""
    raise NotImplementedError(f"{type(self).__name__} does not implement count()")

delete abstractmethod async 

delete(ids)

Remove vectors by ID.

Source code in teff/rag/base.py 
126
127
128
129
@abstractmethod
async def delete(self, ids: list[str]) -> None:
    """Remove vectors by ID."""
    ...

entries async 

entries(limit=100, offset=0)

Return (id, metadata) pairs with pagination.

Source code in teff/rag/base.py 
135
136
137
138
139
async def entries(
    self, limit: int = 100, offset: int = 0
) -> list[tuple[str, dict]]:
    """Return ``(id, metadata)`` pairs with pagination."""
    raise NotImplementedError(f"{type(self).__name__} does not implement list()")

get async 

get(ids)

Return (id, metadata) pairs for existing IDs.

Source code in teff/rag/base.py 
141
142
143
async def get(self, ids: list[str]) -> list[tuple[str, dict]]:
    """Return ``(id, metadata)`` pairs for existing IDs."""
    raise NotImplementedError(f"{type(self).__name__} does not implement get()")

search abstractmethod async 

search(query, k=10, filter=None, hybrid=False, query_text=None)

Search for the k nearest neighbours.

Parameters:

Name Type Description Default
query list[float]

Query embedding.

 required
k int

Maximum number of results.

 10
filter dict | None

Optional metadata filter DSL (see :func:match_filter).

 None
hybrid bool

When True, blend a lexical keyword score with the cosine score (stores that support it; others ignore it).

 False
query_text str | None

Original query text, required for hybrid.

 None

Returns:

Type Description
list[tuple[str, float, dict]]

List of (id, score, metadata) tuples sorted by score
list[tuple[str, float, dict]]

descending. Scores are similarity-like (higher = more similar).
Source code in teff/rag/base.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@abstractmethod
async def search(
    self,
    query: list[float],
    k: int = 10,
    filter: dict | None = None,
    hybrid: bool = False,
    query_text: str | None = None,
) -> list[tuple[str, float, dict]]:
    """Search for the *k* nearest neighbours.

    Args:
        query: Query embedding.
        k: Maximum number of results.
        filter: Optional metadata filter DSL (see :func:`match_filter`).
        hybrid: When ``True``, blend a lexical keyword score with the
            cosine score (stores that support it; others ignore it).
        query_text: Original query text, required for ``hybrid``.

    Returns:
        List of ``(id, score, metadata)`` tuples sorted by score
        descending.  Scores are similarity-like (higher = more similar).
    """
    ...

update_metadata async 

update_metadata(id, metadata)

Merge metadata into the metadata of an existing ID.

Source code in teff/rag/base.py 
145
146
147
148
149
async def update_metadata(self, id: str, metadata: dict) -> None:
    """Merge *metadata* into the metadata of an existing ID."""
    raise NotImplementedError(
        f"{type(self).__name__} does not implement update_metadata()"
    )