7c7b649775
Writers now enforce type and created_at:
- encoding.py: ValueError raised at write_embeddings_batch if row dict lacks
'type'. created_at remains SQL-supplied (NOW() server-side). ON CONFLICT
DO UPDATE now also rewrites type=EXCLUDED.type and preserves the original
created_at via COALESCE(embeddings.created_at, EXCLUDED.created_at) — a
re-ingest re-classifies type but does not overwrite a backfilled mtime.
- ingest_conversations.py: same assertion. ON CONFLICT intentionally keeps
EXCLUDED.created_at semantics (Aaron-AI conversation created_at tracks
convo.updated_at; re-runs should refresh).
- Column-level NOT NULL is not added; application-layer raise gives a
faster, more debuggable failure than a Postgres constraint error.
Retrieval propagates type into chunks:
- retrieve() SELECT now includes type; chunk dicts carry "type": etype.
- WHERE clause built dynamically from excluded_sources and the new
--type-filter CLI arg (experimental, default None, pgvector retrieval
only — Graphiti chunks have no embeddings.type to filter on).
- retrieve_graphiti unchanged; its chunks lack the type field.
Manifests carry type_distribution per stage:
- dream_pipeline writes stage_data[<stage>]["type_distribution"] for nrem,
early_rem, late_rem — a Counter over chunk types, filtering None so
Graphiti chunks (when DREAMER_SUBSTRATE=graphiti) don't pollute the
distribution. Pgvector chunks always carry type post-backfill; if None
appears, the backfill or writer enforcement has regressed.
Verification:
B1 force re-ingest of "Finite and infinite games -- James Carse.pdf":
all 84 chunks preserved created_at=2026-04-27T06:11:55Z
B2 missing-type assertion raises ValueError, no row leaked to embeddings
B3 ast.parse(*) clean; EXPLAIN renders for {no excl/no filter,
type_filter only, excl 2 elems, excl 1 elem edge case, both};
all five plans use HNSW index scan with correct Filter clauses
C1 retrieve("nrem") returns 8 chunks each carrying "type" key
C2 type_distribution = {'document': 5, 'chatgpt_conversation': 3} —
2 distinct types, 62.5/37.5 split (looser bar: >=2 types,
no single type >=90%)
The type and created_at fields are now load-bearing: every dream manifest
emits type_distribution per stage. Reverting the backfill makes the
distribution show NULLs at every dream run.
136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
"""
|
|
Aaron AI Stage 1 encoding helpers — single canonical implementation of:
|
|
- extract_text(filepath) — four-extension text extraction
|
|
- chunk_text(text, chunk_size, overlap) — word-based chunking
|
|
- chunk_and_embed(text, source, embedder, filepath, folder) — produce ready-to-write rows
|
|
- write_embeddings_batch(conn, batch) — server-side NOW() canonical INSERT
|
|
|
|
Used by watcher.py, ingest.py, corpus_integrity.py, and api.py /api/corpus/retry.
|
|
Replaces four separate extract reimplementations and two extract-chunk-embed paths.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from docx import Document as DocxDocument
|
|
from pypdf import PdfReader
|
|
from pptx import Presentation
|
|
|
|
log = logging.getLogger("encoding")
|
|
|
|
SUPPORTED = {".docx", ".pdf", ".pptx", ".txt", ".md"}
|
|
DEFAULT_CHUNK_SIZE = 500
|
|
DEFAULT_CHUNK_OVERLAP = 50
|
|
|
|
|
|
def extract_text(filepath: Path) -> str:
|
|
"""Return the text of a supported file. Returns "" on any failure or
|
|
unsupported extension. Does not write to ingest_failures — caller decides."""
|
|
suffix = filepath.suffix.lower()
|
|
try:
|
|
if suffix == ".docx":
|
|
doc = DocxDocument(filepath)
|
|
return "\n".join(p.text for p in doc.paragraphs if p.text.strip())
|
|
elif suffix == ".pdf":
|
|
reader = PdfReader(filepath)
|
|
return "".join(
|
|
page.extract_text() + "\n"
|
|
for page in reader.pages if page.extract_text()
|
|
)
|
|
elif suffix == ".pptx":
|
|
prs = Presentation(filepath)
|
|
return "\n".join(
|
|
shape.text for slide in prs.slides
|
|
for shape in slide.shapes
|
|
if hasattr(shape, "text") and shape.text.strip()
|
|
)
|
|
elif suffix in {".txt", ".md"}:
|
|
return filepath.read_text(encoding="utf-8", errors="ignore")
|
|
except Exception as e:
|
|
log.warning(f"Text extraction failed for {filepath.name}: {e}")
|
|
return ""
|
|
|
|
|
|
def chunk_text(text: str,
|
|
chunk_size: int = DEFAULT_CHUNK_SIZE,
|
|
overlap: int = DEFAULT_CHUNK_OVERLAP) -> list[str]:
|
|
"""Word-based chunking. Empty chunks filtered."""
|
|
words = text.split()
|
|
chunks = []
|
|
start = 0
|
|
while start < len(words):
|
|
chunk = " ".join(words[start:start + chunk_size])
|
|
if chunk.strip():
|
|
chunks.append(chunk)
|
|
start += chunk_size - overlap
|
|
return chunks
|
|
|
|
|
|
def _chunk_id(filepath, source: str, index: int) -> str:
|
|
basis = str(filepath) if filepath else source
|
|
return f"{hashlib.md5(basis.encode()).hexdigest()[:8]}_{index}"
|
|
|
|
|
|
def chunk_and_embed(text: str,
|
|
source: str,
|
|
embedder,
|
|
filepath=None,
|
|
folder=None) -> list[dict]:
|
|
"""Chunk text, embed each chunk, return rows ready for write_embeddings_batch."""
|
|
chunks = chunk_text(text)
|
|
if not chunks:
|
|
return []
|
|
embeddings = embedder.encode(chunks).tolist()
|
|
rows = []
|
|
for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
|
|
rows.append({
|
|
"id": _chunk_id(filepath, source, i),
|
|
"document": chunk,
|
|
"embedding": emb,
|
|
"source": source,
|
|
"type": "document",
|
|
"metadata": {
|
|
"source": source,
|
|
"filepath": str(filepath) if filepath else source,
|
|
"folder": folder,
|
|
},
|
|
})
|
|
return rows
|
|
|
|
|
|
def write_embeddings_batch(conn, batch: list[dict]) -> int:
|
|
"""Single canonical INSERT. Sets created_at = NOW() server-side. Commits.
|
|
|
|
Every row dict must supply 'type'. created_at is SQL-supplied (NOW()), so
|
|
callers do not need to provide it. The application-layer assertion is the
|
|
primary enforcement point for type — the column lacks NOT NULL because
|
|
historical NULLs were resolved by the Improvement #2 backfill, and a
|
|
Python-level raise gives a faster, more debuggable failure than a
|
|
Postgres constraint error.
|
|
"""
|
|
if not batch:
|
|
return 0
|
|
cur = conn.cursor()
|
|
for row in batch:
|
|
if not row.get("type"):
|
|
raise ValueError(
|
|
f"row {row.get('id')!r} missing 'type'; writers must supply it "
|
|
f"(see Improvement #2 in docs/birdai-component-inventory)"
|
|
)
|
|
cur.execute("""
|
|
INSERT INTO embeddings (id, document, embedding, source, type, created_at, metadata)
|
|
VALUES (%s, %s, %s::vector, %s, %s, NOW(), %s)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
document = EXCLUDED.document,
|
|
embedding = EXCLUDED.embedding,
|
|
source = EXCLUDED.source,
|
|
type = EXCLUDED.type,
|
|
created_at = COALESCE(embeddings.created_at, EXCLUDED.created_at),
|
|
metadata = EXCLUDED.metadata
|
|
""", (row["id"], row["document"], row["embedding"],
|
|
row["source"], row["type"], json.dumps(row["metadata"])))
|
|
conn.commit()
|
|
return len(batch)
|