diff --git a/api/services/kb/chunker.py b/api/services/kb/chunker.py index 1e2e486..398c8a8 100644 --- a/api/services/kb/chunker.py +++ b/api/services/kb/chunker.py @@ -13,6 +13,10 @@ from typing import Literal # Approx tokens per char for Hebrew/mixed text (voyage-multilingual-2 tokenizer). _TOKENS_PER_CHAR = 0.35 +# Hard safety ceiling per chunk. Voyage accepts up to 32K tokens per item, +# but a single batch is capped at 120K tokens. Keep chunks well below both. +_MAX_TOKENS_PER_CHUNK = 2000 + # A "section" in Israeli statutes: "סעיף 12", "תקנה 5", "5א.", "12(ב)" _SECTION_RE = re.compile( r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?\.?)\s*(?:\(([א-ת\d]+)\))?\s*[\.\:]", @@ -50,6 +54,33 @@ def _as_dicts(chunks: list[Chunk]) -> list[dict]: ] +def _split_oversized(chunks: list[Chunk]) -> list[Chunk]: + """Split any chunk whose token_count exceeds the ceiling into fixed-size pieces. + + Preserves heading_path/section_ref on each piece and re-indexes. + """ + max_chars = int(_MAX_TOKENS_PER_CHUNK / _TOKENS_PER_CHAR) + result: list[Chunk] = [] + for c in chunks: + if c.token_count <= _MAX_TOKENS_PER_CHUNK: + c.chunk_index = len(result) + result.append(c) + continue + text = c.content + for start in range(0, len(text), max_chars): + piece = text[start : start + max_chars] + result.append( + Chunk( + chunk_index=len(result), + heading_path=c.heading_path, + section_ref=c.section_ref, + content=piece, + token_count=_count_tokens(piece), + ) + ) + return result + + def chunk_statute(text: str) -> list[dict]: """Law / regulation: one chunk per section. Preserves section numbering.""" text = text.strip() @@ -160,5 +191,18 @@ def chunk_circular( def chunk(text: str, kind: Literal["law", "regulation", "circular"]) -> list[dict]: if kind in ("law", "regulation"): - return chunk_statute(text) - return chunk_circular(text) + raw = chunk_statute(text) + else: + raw = chunk_circular(text) + # Reconstruct Chunk objects to apply the safety ceiling. + chunks = [ + Chunk( + chunk_index=r["chunk_index"], + heading_path=r["heading_path"] or "", + section_ref=r["section_ref"] or "", + content=r["content"], + token_count=r["token_count"], + ) + for r in raw + ] + return _as_dicts(_split_oversized(chunks)) diff --git a/api/services/kb/ingest.py b/api/services/kb/ingest.py index 9f3211c..25695e3 100644 --- a/api/services/kb/ingest.py +++ b/api/services/kb/ingest.py @@ -86,14 +86,28 @@ async def ingest_source( pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): - existing = await conn.fetchrow( - """ - SELECT id, checksum FROM kb_source - WHERE kind = $1 AND COALESCE(identifier,'') = COALESCE($2,'') - ORDER BY created_at DESC LIMIT 1 - """, - kind, identifier, - ) + # Match a prior source by (kind + identifier) when identifier is provided, + # otherwise fall back to (kind + title). Without this, two different + # sources with NULL identifier would both compare equal and the second + # ingest would incorrectly supersede the first. + if identifier: + existing = await conn.fetchrow( + """ + SELECT id, checksum FROM kb_source + WHERE kind = $1 AND identifier = $2 AND superseded_by IS NULL + ORDER BY created_at DESC LIMIT 1 + """, + kind, identifier, + ) + else: + existing = await conn.fetchrow( + """ + SELECT id, checksum FROM kb_source + WHERE kind = $1 AND title = $2 AND identifier IS NULL AND superseded_by IS NULL + ORDER BY created_at DESC LIMIT 1 + """, + kind, title, + ) was_updated = False if existing and existing["checksum"] == checksum: