API reference

class ClickHouseStore:
    """A table of versioned passages using an application-owned ClickHouse client."""

    def __init__(
        self,
        client: Client,
        *,
        table: str,
        embedding_model: str,
        dimensions: int,
        database: str = "default",
    ) -> None:
        integer(dimensions, "dimensions", _MAX_DIMENSIONS)
        nonblank(embedding_model, "embedding_model")
        self._client = client
        self._table = f"{identifier(database)}.{identifier(table)}"
        self._location = {"database": database, "table": table}
        self._dimensions = dimensions
        self._embedding_model = embedding_model
        self._manifest = "clickhouserag:" + json.dumps(
            {"schema": 2, "model": embedding_model, "dimensions": dimensions},
            sort_keys=True,
            separators=(",", ":"),
        )
        self._validated = False

    _passage_source = _TABLE_SOURCE

    @property
    def dimensions(self) -> int:
        """Expected number of components in each vector."""
        return self._dimensions

    @property
    def embedding_model(self) -> str:
        """Identifier of the embedding model associated with this table."""
        return self._embedding_model

    def initialize(self) -> None:
        """Create an empty table if absent, then validate its schema identity."""
        # The qualified name uses the validated identifier grammar. The comment
        # and dimension are passed through driver binding.
        self._client.command(
            f"""
            CREATE TABLE IF NOT EXISTS {self._table} (
                namespace String,
                chunk_id String,
                document_id String,
                text String,
                source String,
                revision UInt64,
                metadata Map(String, String),
                embedding Array(Float32),
                lexical_terms Array(String),
                INDEX lexical_idx(lexical_terms) TYPE text(tokenizer = 'array'),
                CONSTRAINT embedding_dimensions CHECK length(embedding) = %(dimensions)s
            )
            ENGINE = ReplacingMergeTree(revision)
            ORDER BY (namespace, chunk_id)
            COMMENT %(manifest)s
            """,
            parameters={"dimensions": self._dimensions, "manifest": self._manifest},
        )
        self.validate_schema()

    def validate_schema(self) -> None:
        """Verify that the existing table belongs to this schema and embedding model."""
        self._validated = False
        result = self._client.query(
            """
            SELECT comment
            FROM system.tables
            WHERE database = {database:String} AND name = {table:String}
            """,
            parameters=self._location,
        )
        if result.result_rows != [(self._manifest,)]:
            raise SchemaError(
                "ClickHouse table is missing or does not match the store schema and embedding model"
            )
        self._validated = True

    def insert(self, chunks: Iterable[EmbeddedChunk], *, namespace: str = "default") -> int:
        """Insert one validated batch of up to 10,000 embedded passages.

        Repeat the same payload at the same revision to retry a failed write. Use
        a strictly greater revision for replacements of a given namespace/chunk ID.
        """
        return self._insert_rows(self._prepare_rows(chunks, namespace))

    def _prepare_rows(self, chunks: Iterable[EmbeddedChunk], namespace: str) -> list[_PassageRow]:
        nonblank(namespace, "namespace")
        rows = []
        ids = set()
        for item in chunks:
            if len(rows) >= _MAX_BATCH:
                raise ValueError(f"insert accepts at most {_MAX_BATCH} chunks per batch")
            chunk = item.chunk
            metadata = chunk_fields(chunk)
            if chunk.chunk_id in ids:
                raise ValueError("chunk IDs must be unique within a batch")
            ids.add(chunk.chunk_id)
            rows.append(
                (
                    namespace,
                    chunk.chunk_id,
                    chunk.document_id,
                    chunk.text,
                    chunk.source,
                    chunk.revision,
                    metadata,
                    unit_vector(item.embedding, self.dimensions),
                    terms(chunk.text),
                )
            )
        return rows

    def _insert_rows(
        self,
        rows: list[_PassageRow],
    ) -> int:
        if not rows:
            return 0
        self._ensure_schema()
        self._client.insert(
            self._table,
            rows,
            column_names=_INSERT_COLUMNS,
            settings={"async_insert": 0},
        )
        return len(rows)

    def search_vector(
        self,
        embedding: Sequence[float],
        *,
        namespace: str = "default",
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
    ) -> list[SearchHit]:
        """Find the nearest passages by exact cosine similarity (higher is closer)."""
        parameters = _scope(namespace, metadata, limit)
        parameters.update(self._location)
        parameters["embedding"] = unit_vector(embedding, self.dimensions)
        return self._search(
            """
            SELECT chunk_id, document_id, text, source, revision, metadata,
                1 - cosineDistance(embedding, {embedding:Array(Float32)}) AS score
            FROM @passages@
            WHERE namespace = {namespace:String}
                AND arrayAll(pair -> mapContains(metadata, pair.1)
                    AND metadata[pair.1] = pair.2, {filters:Array(Tuple(String, String))})
            ORDER BY score DESC, chunk_id ASC
            LIMIT {limit:UInt32}
            """,
            parameters,
        )

    def search_lexical(
        self,
        query: str,
        *,
        namespace: str = "default",
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
    ) -> list[SearchHit]:
        """Rank matching passages by distinct query token coverage."""
        parameters = _scope(namespace, metadata, limit)
        query_terms = _query_terms(query)
        if not query_terms:
            return []
        parameters.update(self._location, terms=query_terms)
        return self._search(
            """
            SELECT chunk_id, document_id, text, source, revision, metadata,
                length(arrayIntersect(lexical_terms, {terms:Array(String)})) AS score
            FROM @passages@
            WHERE namespace = {namespace:String}
                AND arrayAll(pair -> mapContains(metadata, pair.1)
                    AND metadata[pair.1] = pair.2, {filters:Array(Tuple(String, String))})
                AND hasAnyTokens(lexical_terms, {terms:Array(String)})
            ORDER BY score DESC, chunk_id ASC
            LIMIT {limit:UInt32}
            """,
            parameters,
        )

    def search_hybrid(
        self,
        query: str,
        embedding: Sequence[float],
        *,
        namespace: str = "default",
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        candidates: int | None = None,
    ) -> list[SearchHit]:
        """Fuse independently ranked vector and lexical candidates."""
        integer(limit, "limit", _MAX_LIMIT)
        if candidates is None:
            candidates = min(limit * 4, _MAX_LIMIT)
        integer(candidates, "candidates", _MAX_LIMIT)
        if candidates < limit:
            raise ValueError("candidates must be at least limit")
        parameters = _scope(namespace, metadata, candidates)
        query_terms = _query_terms(query)
        if not query_terms:
            dense = self.search_vector(
                embedding, namespace=namespace, metadata=metadata, limit=limit
            )
            return reciprocal_rank_fusion(dense, [], limit)
        parameters.update(
            self._location, terms=query_terms, embedding=unit_vector(embedding, self.dimensions)
        )
        self._ensure_schema()
        result = self._client.query(
            _HYBRID_SQL.replace("@passages@", self._passage_source),
            parameters=parameters,
            settings=_SELECT_SETTINGS,
        )
        lanes: tuple[list[SearchHit], list[SearchHit]] = ([], [])
        for *row, lane in result.result_rows:
            lanes[lane].append(_hit(row))
        for hits in lanes:
            hits.sort(key=lambda hit: (-hit.score, hit.chunk.chunk_id))
        return reciprocal_rank_fusion(*lanes, limit)

    def _ensure_schema(self) -> None:
        if not self._validated:
            self.validate_schema()

    def _search(self, query: str, parameters: dict[str, Any]) -> list[SearchHit]:
        self._ensure_schema()
        rows = self._client.query(
            query.replace("@passages@", self._passage_source),
            parameters=parameters,
            settings=_SELECT_SETTINGS,
        )
        return [_hit(row) for row in rows.result_rows]

@dataclass(frozen=True, slots=True)
class Chunk:
    """A passage with a stable identity, provenance and a write revision."""

    chunk_id: str
    document_id: str
    text: str
    source: str
    revision: int = 1
    metadata: Mapping[str, str] = field(default_factory=dict)

@dataclass(frozen=True, slots=True)
class EmbeddedChunk:
    """A passage paired with its document embedding."""

    chunk: Chunk
    embedding: Sequence[float]

@dataclass(frozen=True, slots=True)
class SearchHit:
    """A passage and its score, ordered by the selected search strategy."""

    chunk: Chunk
    score: float

class SchemaError(RuntimeError):
    """The configured ClickHouse table is missing or incompatible."""

class EmbeddingProvider(Protocol):
    """Application-supplied encoder with separate document and query methods."""

    @property
    def model_id(self) -> str:
        """Identity of the model and encoding configuration."""
        ...

    def embed_documents(self, texts: Sequence[str]) -> Sequence[Sequence[float]]:
        """Encode the passages in input order."""
        ...

    def embed_query(self, text: str) -> Sequence[float]:
        """Encode a search query in the same vector space."""
        ...

class Retriever:
    """Encode and retrieve passages using a specific embedding provider."""

    def __init__(self, store: ClickHouseStore, provider: EmbeddingProvider) -> None:
        if store.embedding_model != provider.model_id:
            raise ValueError("provider model_id must match the store embedding_model")
        self._store = store
        self._provider = provider

    def index(self, chunks: Sequence[Chunk], *, namespace: str = "default") -> int:
        """Embed and insert a batch of passages."""
        nonblank(namespace, "namespace")
        return self._store.insert(self._embed(chunks), namespace=namespace)

    def _embed(self, chunks: Sequence[Chunk]) -> list[EmbeddedChunk]:
        if not chunks:
            return []
        if len(chunks) > _MAX_BATCH:
            raise ValueError(f"index accepts at most {_MAX_BATCH} chunks per batch")
        for chunk in chunks:
            chunk_fields(chunk)
        if len({chunk.chunk_id for chunk in chunks}) != len(chunks):
            raise ValueError("chunk IDs must be unique within a batch")
        vectors = self._provider.embed_documents([chunk.text for chunk in chunks])
        if len(vectors) != len(chunks):
            raise ValueError("embedding provider returned the wrong number of document vectors")
        return [EmbeddedChunk(chunk, vector) for chunk, vector in zip(chunks, vectors, strict=True)]

    def search(
        self,
        query: str,
        *,
        namespace: str = "default",
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        hybrid: bool = True,
    ) -> list[SearchHit]:
        """Embed a query and perform hybrid or exact vector retrieval."""
        nonblank(query, "query")
        embedding = self._provider.embed_query(query)
        if hybrid:
            return self._store.search_hybrid(
                query, embedding, namespace=namespace, metadata=metadata, limit=limit
            )
        return self._store.search_vector(
            embedding, namespace=namespace, metadata=metadata, limit=limit
        )

class RAG:
    """Index text documents, retrieve cited evidence and optionally ask a model."""

    def __init__(
        self,
        client: Client,
        provider: EmbeddingProvider,
        *,
        dimensions: int,
        table: str = "rag_passages",
        database: str = "default",
        namespace: str = "default",
        extractor: RelationExtractor | None = None,
        generator: AnswerProvider | None = None,
        graph: bool = False,
        schema: tuple[Predicate, ...] | None = None,
        planner: KnowledgePlanner | None = None,
        search_planner: SearchPlanner | None = None,
        reranker: Reranker | None = None,
        rerank_candidates: int = 32,
        chunker: Chunker | None = None,
        index_revision: str | None = None,
        chunk_size: int = 1600,
        chunk_overlap: int = 160,
    ) -> None:
        nonblank(namespace, "namespace")
        integer(chunk_size, "chunk_size", 100_000)
        integer(rerank_candidates, "rerank_candidates", _MAX_LIMIT)
        if type(chunk_overlap) is not int or not 0 <= chunk_overlap < chunk_size:
            raise ValueError("chunk_overlap must be a nonnegative integer less than chunk_size")
        if chunker is not None and index_revision is None:
            raise ValueError("a custom chunker requires an index_revision")
        if index_revision is not None:
            nonblank(index_revision, "index_revision")
        self.passages = DocumentStore(
            client,
            table=table,
            database=database,
            embedding_model=provider.model_id,
            dimensions=dimensions,
        )
        self.graph = (
            GraphStore(self.passages)
            if graph or extractor is not None or schema is not None
            else None
        )
        self._knowledge = (
            Knowledge(self.graph, schema) if self.graph is not None and schema is not None else None
        )
        self._retriever = Retriever(self.passages, provider)
        self._namespace = namespace
        if planner is not None and schema is None:
            raise ValueError("a knowledge planner requires a schema")
        self._schema = schema or ()
        self._planner = planner
        self._search_planner = search_planner
        self._extractor = extractor
        self._generator = generator
        self._reranker = reranker
        self._rerank_candidates = rerank_candidates
        self._chunker = chunker
        self._index_revision = index_revision
        self._chunk_size = chunk_size
        self._chunk_overlap = chunk_overlap

    def initialize(self) -> None:
        """Provision the managed passage and fact table."""
        self.passages.initialize()

    def index(self, document: Document) -> int:
        """Publish this document's full contents; return the number of staged passages."""
        nonblank(document.document_id, "document_id")
        nonblank(document.source, "source")
        nonblank(document.text, "text")
        metadata = string_map(document.metadata)
        text = document.text.strip()
        config = [
            self._chunk_size,
            self._chunk_overlap,
            self._extractor is not None,
        ]
        if self._index_revision is not None:
            config.extend((self._index_revision, self._chunker is not None))
        identity = hashlib.sha256(
            json.dumps(
                [
                    text,
                    document.source,
                    metadata,
                    *config,
                ],
                sort_keys=True,
                separators=(",", ":"),
            ).encode()
        ).hexdigest()
        revision = self.passages.next_revision(
            self._namespace, document.document_id, identity, document.revision
        )
        if revision is None:
            return 0
        chunks = self._chunks(document, text, metadata, revision)
        rows = self.passages._prepare_rows(self._retriever._embed(chunks), self._namespace)
        facts = [self._facts(chunk) for chunk in chunks]
        self.passages.publish(
            self._namespace, document.document_id, revision, identity, rows, facts
        )
        return len(rows)

    def delete(self, document_id: str, *, revision: int | None = None) -> bool:
        """Hide a document and its facts by publishing a tombstone."""
        nonblank(document_id, "document_id")
        chosen = self.passages.next_revision(self._namespace, document_id, "", revision)
        if chosen is None:
            return False
        self.passages.publish(self._namespace, document_id, chosen, "", [], [])
        return True

    def _facts(self, chunk: Chunk) -> list[_Fact]:
        if self._extractor is None:
            return []
        extracted = self._extractor.extract(chunk)
        if len(extracted) > _MAX_BATCH:
            raise ValueError(f"a passage may contain at most {_MAX_BATCH} relations")
        return sorted(
            _document_fact(
                Relation(
                    item.subject,
                    item.predicate,
                    item.object,
                    chunk.chunk_id,
                    chunk.revision,
                    item.subject_type,
                    item.object_type,
                )
            )
            for item in extracted
        )

    def retrieve(
        self,
        query: str,
        *,
        mode: _Mode = "hybrid",
        metadata: Mapping[str, str] | None = None,
        seeds: Sequence[str] | None = None,
        limit: int = 10,
        depth: int = 2,
    ) -> Context:
        """Retrieve passages and whole cited paths within the passage limit."""
        nonblank(query, "query")
        integer(limit, "limit", _MAX_LIMIT)
        if mode not in (*_PASSAGE_MODES, "global"):
            raise ValueError("unknown retrieval mode")
        if seeds is not None and mode not in ("graph", "mixed"):
            raise ValueError("explicit seeds require graph or mixed mode")
        if mode == "global":
            return self.overview(query, metadata=metadata, limit=limit)
        if mode == "graph" and self.graph is None:
            raise ValueError("graph mode requires graph configuration")
        filters = string_map(metadata) if metadata is not None else None
        scope: _Scope = {"namespace": self._namespace, "metadata": filters}
        candidate_limit = (
            max(limit, self._rerank_candidates) if self._reranker is not None else limit
        )
        hits: list[SearchHit] = []
        if mode == "lexical":
            hits = self.passages.search_lexical(query, limit=candidate_limit, **scope)
        elif mode != "graph":
            hits = self._retriever.search(
                query, limit=candidate_limit, hybrid=mode != "dense", **scope
            )
        paths = []
        if self.graph is not None and mode in ("graph", "mixed"):
            selected = self.graph.seed_entities(query, **scope) if seeds is None else seeds
            paths = self.graph.expand(selected, depth=depth, limit=min(limit * 4, 1000), **scope)
            hits, paths = self._visible_evidence(hits, paths)
        if self._reranker is not None and len(hits) > 1:
            order = tuple(self._reranker.rank(query, tuple(hit.chunk for hit in hits)))
            if (
                len(order) != len(hits)
                or any(type(index) is not int for index in order)
                or set(order) != set(range(len(hits)))
            ):
                raise ValueError("reranker must return a permutation of the candidate indices")
            hits, paths = self._visible_evidence([hits[index] for index in order], paths)
        return _context(hits, paths, limit)

    def retrieve_many(
        self,
        queries: Sequence[SearchQuery],
        *,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        candidates_per_query: int = 16,
    ) -> Context:
        """Fuse scoped searches and retain complete graph paths under one passage budget."""
        integer(len(queries), "number of searches", _MAX_QUERY_LANES)
        integer(limit, "limit", _MAX_LIMIT)
        integer(candidates_per_query, "candidates_per_query", _MAX_LIMIT)
        filters = string_map(metadata) if metadata is not None else None
        searches = tuple(dict.fromkeys(queries))
        if any(query.mode not in _PASSAGE_MODES for query in searches):
            raise ValueError("search plans require passage retrieval modes")
        passages: dict[str, Chunk] = {}
        ranks: list[list[str]] = []
        paths: dict[tuple[str, tuple[tuple[Relation, str], ...]], GraphPath] = {}
        for query in searches:
            result = self.retrieve(
                query.text,
                mode=query.mode,
                metadata=filters,
                seeds=query.seeds,
                limit=candidates_per_query,
                depth=query.depth,
            )
            ranks.append([chunk.chunk_id for chunk in result.passages])
            passages.update((chunk.chunk_id, chunk) for chunk in result.passages)
            for path in result.paths:
                key = (
                    path.seed,
                    tuple((step.relation, step.chunk.chunk_id) for step in path.steps),
                )
                paths.setdefault(key, path)
        hits = [SearchHit(passages[key], score) for key, score in fused_ranks(ranks)]
        hits, visible_paths = self._visible_evidence(hits, tuple(paths.values()))
        return _context(hits, visible_paths, limit)

    def ask_many(
        self,
        question: str,
        queries: Sequence[SearchQuery],
        *,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        candidates_per_query: int = 16,
    ) -> Answer:
        """Answer the original question using evidence fused from several searches."""
        nonblank(question, "question")
        if self._generator is None:
            raise ValueError("ask_many requires an answer provider")
        context = self.retrieve_many(
            queries, metadata=metadata, limit=limit, candidates_per_query=candidates_per_query
        )
        return self._answer(question, context)

    def ask_planned(
        self,
        question: str,
        *,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        candidates_per_query: int = 16,
    ) -> Answer:
        """Plan searches first and answer from their shared scoped evidence."""
        nonblank(question, "question")
        integer(limit, "limit", _MAX_LIMIT)
        integer(candidates_per_query, "candidates_per_query", _MAX_LIMIT)
        filters = string_map(metadata) if metadata is not None else None
        if self._search_planner is None or self._generator is None:
            raise ValueError("ask_planned requires a search planner and an answer provider")
        modes = _PASSAGE_MODES if self.graph is not None else _PASSAGE_MODES[:3]
        queries = self._search_planner.plan_search(question, modes)
        if any(query.mode not in modes for query in queries):
            raise ValueError("search planner returned an unavailable mode")
        return self.ask_many(
            question,
            queries or (SearchQuery(question),),
            metadata=filters,
            limit=limit,
            candidates_per_query=candidates_per_query,
        )

    def overview(
        self,
        query: str,
        *,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
        max_facts: int = _DEFAULT_EDGE_BUDGET,
    ) -> Context:
        """Return cited representatives of connected groups across the scoped corpus."""
        nonblank(query, "query")
        integer(limit, "limit", _MAX_LIMIT)
        integer(max_facts, "max_facts", _MAX_EDGE_BUDGET)
        if self.graph is None:
            raise ValueError("global mode requires graph configuration")
        facts = self.graph._corpus_facts(self._namespace, metadata, max_facts + 1)
        communities = select_communities(query, facts[:max_facts], limit=limit)
        passages = {
            chunk.chunk_id: chunk
            for row in zip_longest(*(item.passages for item in communities))
            for chunk in row
            if chunk is not None
        }
        return Context(
            tuple(passages.values()), communities=communities, truncated=len(facts) > max_facts
        )

    def reason(
        self,
        query: KnowledgeQuery | KnowledgeIntersection,
        *,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
    ) -> KnowledgeResult:
        """Resolve a typed knowledge plan in the same collection as text retrieval."""
        if self._knowledge is None:
            raise ValueError("reason requires a knowledge schema")
        result = self._knowledge.solve(
            query, namespace=self._namespace, metadata=metadata, limit=limit
        )
        chunks = [step.chunk for path in result.evidence for step in path.steps]
        visible = self.passages.visible_chunks(self._namespace, chunks)
        evidence = [
            (leg, path)
            for leg, path in zip(result.evidence_legs, result.evidence, strict=True)
            if all((step.chunk.chunk_id, step.chunk.revision) in visible for step in path.steps)
        ]
        proof: defaultdict[str, set[int]] = defaultdict(set)
        for leg, path in evidence:
            proof[path.target].add(leg)
        targets = tuple(
            target for target in result.entities if len(proof[target]) == result.required_legs
        )
        target_set = set(targets)
        return KnowledgeResult(
            targets,
            tuple(path for _, path in evidence if path.target in target_set),
            tuple(leg for leg, path in evidence if path.target in target_set),
            result.required_legs,
        )

    def ask_knowledge(
        self,
        question: str,
        *,
        plan: KnowledgeQuery | KnowledgeIntersection | None = None,
        metadata: Mapping[str, str] | None = None,
        limit: int = 10,
    ) -> Answer:
        """Execute typed reasoning and answer with cited proof, optionally using a supplied plan."""
        nonblank(question, "question")
        integer(limit, "limit", _MAX_LIMIT)
        if self._generator is None:
            raise ValueError("ask_knowledge requires an answer provider")
        filters = string_map(metadata) if metadata is not None else None
        if plan is None:
            if self._planner is None:
                raise ValueError("ask_knowledge requires a knowledge planner or an explicit plan")
            proposed = self._planner.plan(question, self._schema)
            try:
                result = self.reason(proposed, metadata=filters, limit=limit) if proposed else None
            except KnowledgePlanError:
                result = None
            if result is None:
                return Answer(None, (), Context(()))
        else:
            result = self.reason(plan, metadata=filters, limit=limit)
        context = _knowledge_context(result, limit)
        answer = self._answer(question, context)
        if answer.text is None:
            return answer
        cited = {chunk.chunk_id for chunk in answer.citations}
        covered: defaultdict[str, set[int]] = defaultdict(set)
        for leg, path in zip(result.evidence_legs, result.evidence, strict=True):
            if all(step.chunk.chunk_id in cited for step in path.steps):
                covered[path.target].add(leg)
        if any(len(proof) == result.required_legs for proof in covered.values()):
            return answer
        return Answer(None, (), context)

    def ask(
        self,
        question: str,
        *,
        mode: _Mode = "hybrid",
        metadata: Mapping[str, str] | None = None,
        seeds: Sequence[str] | None = None,
        limit: int = 10,
        depth: int = 2,
    ) -> Answer:
        """Generate with evidence and reject citation IDs outside the context."""
        if self._generator is None:
            raise ValueError("ask requires an answer provider")
        context = self.retrieve(
            question, mode=mode, metadata=metadata, seeds=seeds, limit=limit, depth=depth
        )
        return self._answer(question, context)

    def _answer(self, question: str, context: Context) -> Answer:
        if self._generator is None:
            raise ValueError("asking a question requires an answer provider")
        if not context.passages:
            return Answer(None, (), context)
        result = self._generator.generate(question, context)
        if result.text is None:
            if result.citation_ids:
                raise ValueError("an abstention must not have citations")
            return Answer(None, (), context)
        nonblank(result.text, "answer")
        known = {chunk.chunk_id: chunk for chunk in context.passages}
        if (
            isinstance(result.citation_ids, str)
            or not result.citation_ids
            or any(key not in known for key in result.citation_ids)
        ):
            raise ValueError("answer provider must cite IDs from the supplied context")
        return Answer(
            result.text, tuple(known[key] for key in dict.fromkeys(result.citation_ids)), context
        )

    def _visible_evidence(
        self, hits: list[SearchHit], paths: Sequence[GraphPath]
    ) -> tuple[list[SearchHit], list[GraphPath]]:
        chunks = [hit.chunk for hit in hits]
        chunks.extend(step.chunk for path in paths for step in path.steps)
        visible = self.passages.visible_chunks(self._namespace, chunks)
        return (
            [hit for hit in hits if (hit.chunk.chunk_id, hit.chunk.revision) in visible],
            [
                path
                for path in paths
                if all((step.chunk.chunk_id, step.chunk.revision) in visible for step in path.steps)
            ],
        )

    def _chunks(
        self, document: Document, text: str, metadata: Mapping[str, str], revision: int
    ) -> list[Chunk]:
        if self._chunker is None:
            located = self._character_chunks(text)
        else:
            parts = self._chunker.split(text)
            if isinstance(parts, (str, bytes)):
                raise ValueError("chunker must return an iterable of passages, not a string")
            located = enumerate(parts)
        chunks = []
        for offset, part in located:
            nonblank(part, "chunk text")
            if len(chunks) == _MAX_BATCH:
                raise ValueError(f"document produces more than {_MAX_BATCH} chunks")
            key = hashlib.sha256(
                f"{document.document_id}\0{revision}\0{offset}".encode()
            ).hexdigest()
            chunks.append(
                Chunk(key, document.document_id, part, document.source, revision, metadata)
            )
        if not chunks:
            raise ValueError("chunker must produce at least one passage")
        return chunks

    def _character_chunks(self, text: str) -> Iterable[tuple[int, str]]:
        for start in range(0, len(text), self._chunk_size - self._chunk_overlap):
            part = text[start : start + self._chunk_size].strip()
            if part:
                yield start, part
            if start + self._chunk_size >= len(text):
                break