Implement retrieval/query.py with the retrieve function and get_lineage function that enable semantic search against the LanceDB vector index and full provenance lookup from the document store.
Requirements
- Implement
retrieve(query: str, embedder: Embedder, lance_db_path: Path, document_store: DocumentStore, top_k: int = 10) -> list[dict] function in retrieval/query.py:
- Embed the query:
query_vector = embedder.embed_query(query)
- Open LanceDB at
lance_db_path and access the chunk vectors table
- Perform nearest-neighbor search:
results = table.search(query_vector).limit(top_k).to_list()
- Map each result to a dict:
{"chunk_text": ..., "chunk_hash": ..., "source_id": ..., "score": ...}
- Return results ordered by descending relevance score
- Implement
get_lineage(chunk_hash: str, document_store: DocumentStore) -> LineageRecord | None function:
- Calls
document_store.get_lineage(chunk_hash) and returns the result
- The
retrieve function must accept top_k and return exactly top_k results (or fewer if fewer chunks exist)
- Results must include:
chunk_text (the chunk content text), chunk_hash, source_id, score (relevance score from LanceDB)
- Create
tests/retrieval/test_query.py with integration tests: index chunks, retrieve with query, verify result fields, verify ordering by score, verify lineage lookup
Design Guidance
retrieval/query.py already exists as a stub — implement the functions there
- LanceDB open pattern:
import lancedb; db = lancedb.connect(lance_db_path); table = db.open_table("chunk_vectors")
- LanceDB search pattern:
table.search(query_vector).limit(top_k).to_pandas() or .to_list() — the result includes a _distance column (lower = closer). Convert distance to score: score = 1.0 - result["_distance"] or use as-is and document the ordering semantics.
ChunkVector LanceDB model (from storage/vector_store.py) has fields: chunk_hash, content, source_id, embedding — map content → chunk_text in the result dict
- The
retrieve function is a module-level function, not a class method — keep it simple
get_lineage is also a module-level function
- For integration tests: use a temp LanceDB directory, write known chunks via the pipeline or directly via
VectorStore, then retrieve and verify
- The retrieval layer has no business logic beyond embedding, searching, and mapping — keep it thin
- Import:
from context_library.storage.models import LineageRecord, from context_library.core.embedder import Embedder, from context_library.storage.document_store import DocumentStore
Acceptance Criteria
Dependencies
Phase 6: Embedder, Phase 7: Ingestion Pipeline Orchestrator
Parent Issue
Part of #40
Discussion
This work is detailed in discussion 41
Implement
retrieval/query.pywith theretrievefunction andget_lineagefunction that enable semantic search against the LanceDB vector index and full provenance lookup from the document store.Requirements
retrieve(query: str, embedder: Embedder, lance_db_path: Path, document_store: DocumentStore, top_k: int = 10) -> list[dict]function inretrieval/query.py:query_vector = embedder.embed_query(query)lance_db_pathand access the chunk vectors tableresults = table.search(query_vector).limit(top_k).to_list(){"chunk_text": ..., "chunk_hash": ..., "source_id": ..., "score": ...}get_lineage(chunk_hash: str, document_store: DocumentStore) -> LineageRecord | Nonefunction:document_store.get_lineage(chunk_hash)and returns the resultretrievefunction must accepttop_kand return exactlytop_kresults (or fewer if fewer chunks exist)chunk_text(the chunk content text),chunk_hash,source_id,score(relevance score from LanceDB)tests/retrieval/test_query.pywith integration tests: index chunks, retrieve with query, verify result fields, verify ordering by score, verify lineage lookupDesign Guidance
retrieval/query.pyalready exists as a stub — implement the functions thereimport lancedb; db = lancedb.connect(lance_db_path); table = db.open_table("chunk_vectors")table.search(query_vector).limit(top_k).to_pandas()or.to_list()— the result includes a_distancecolumn (lower = closer). Convert distance to score:score = 1.0 - result["_distance"]or use as-is and document the ordering semantics.ChunkVectorLanceDB model (fromstorage/vector_store.py) has fields:chunk_hash,content,source_id,embedding— mapcontent→chunk_textin the result dictretrievefunction is a module-level function, not a class method — keep it simpleget_lineageis also a module-level functionVectorStore, then retrieve and verifyfrom context_library.storage.models import LineageRecord,from context_library.core.embedder import Embedder,from context_library.storage.document_store import DocumentStoreAcceptance Criteria
retrieve()returns a list of dicts with keys:chunk_text,chunk_hash,source_id,scoreretrieve()returns at mosttop_kresults; fewer if fewer chunks exist in the indexget_lineage()returns the correctLineageRecordfor a givenchunk_hashafter ingestget_lineage()returnsNonefor an unknownchunk_hashsource_idtests/retrieval/test_query.pypassesDependencies
Phase 6: Embedder, Phase 7: Ingestion Pipeline Orchestrator
Parent Issue
Part of #40
Discussion
This work is detailed in discussion 41