Add a real LangChain in-memory vector retriever to the versioned corpus, then replay the saved development questions. The program searches actual stored vectors and makes no new embedding or answer-model request.
When you need this. Use it when you want to inspect vector retrieval behavior on a fixed corpus and recorded queries before changing the application or making another model run.
Prerequisites. Use the versioned corpus from the preceding module. The vector course supplies the embedding and vector-search prerequisites. Install the lab dependencies from requirements.lock in a virtual environment.
Bind the corpus to the preserved vectors
The lab reuses the saved Qwen document and question vectors from Record a RAG Baseline. The accompanying manifest identifies their model, query instruction and file hash.
SavedEmbeddings implements LangChain’s embeddings interface by lookup. Document text can only select a recorded document vector. A question can only select its recorded query vector. The adapter validates dimensions, finite values and nonzero finite norms, and returns copies so the caller cannot change its saved inputs.
Changed recording. A different raw vector-file hash raises ValueError before the store is built. The copied versioning validator also checks snapshot consistency. Editing chunk text cannot silently reuse the old document vector.
Only the supplied development questions are exposed by the adapter. Historic held-out vectors remain in the preserved raw file, but the lab does not query them. This is a replay boundary, not secrecy or an access-control system.
Call the actual LangChain store
The program inserts the snapshot chunks into InMemoryVectorStore, keeping chunk IDs and source/version metadata beside each text. LangChain calls embed_documents; the adapter returns the preserved vectors. At query time, LangChain calls embed_query; it receives the original vector for that exact recorded question.
No prompt prefix is added during replay. The original embedding run already used the instruction recorded in its provenance. A different question, instruction or corpus needs new embeddings; this adapter cannot generate a replacement vector.
Unrecorded question. An unknown string raises ValueError with unrecorded_development_query. It does not call a provider, invent a vector or return another question’s embedding.
The store performs real cosine search over the saved vectors. The adapter asks for all eligible scores, sorts by descending score with chunk ID as a stable tie-breaker, then takes the requested result count. This favors inspectable behavior for the small fixture; it is not an approximate-nearest-neighbor index or a large-corpus benchmark.
Read candidate passages before accepting an answer
The driver replays the development cases under their requested Python version. A version filter removes an ineligible documentation version; it cannot change what the remaining passages say. The UTF-8 writing question still retrieves a reading passage ahead of the writing passage in the recorded eligible results.
The unsupported backup question also returns nearest passages. Its supplied question class describes the dataset’s expected handling, not a new model judgment. A similarity score orders stored passages even when none supplies the requested operational policy.
Compare the candidate text with the question and relevant dataset labels. No answer is generated here, and no new retrieval-quality or performance improvement is claimed. Review the vector course for the earlier embedding and similarity mechanisms instead of rebuilding them in this module.
Execute the local retrieval contract
Download the complete lab. It includes the README, fixtures, implementation, tests, dependency information and actual run record.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.lock"""Real LangChain store backed only by preserved model vectors; no inference."""
import hashlib
import json
import math
from pathlib import Path
import re
from version_documents import validate_snapshot
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import InMemoryVectorStore
VECTORS_SHA256 = 'bbadb40c26e0241de36f09f6068354f7589b9ce1a38bb49ba6998be4ca3b25e6'
DIMENSION = 1024
def checked_vector(vector):
if not isinstance(vector,list) or len(vector)!=DIMENSION:
raise ValueError('invalid_vector_dimension')
if any(type(v) not in (int,float) or not math.isfinite(v) for v in vector):
raise ValueError('nonfinite_or_invalid_vector')
norm=math.sqrt(math.fsum(v*v for v in vector))
if not math.isfinite(norm) or norm<=0:
raise ValueError('invalid_vector_norm')
return list(vector)
class SavedEmbeddings(Embeddings):
def __init__(self,recording,allowed_questions):
self.documents={text:checked_vector(vector) for text,vector in recording['documents'].items()}
self.questions={text:checked_vector(recording['questions'][text]) for text in allowed_questions}
def embed_documents(self,texts):
if any(text not in self.documents for text in texts):raise ValueError('unrecorded_document')
return [list(self.documents[text]) for text in texts]
def embed_query(self,text):
if text not in self.questions:raise ValueError('unrecorded_development_query')
return list(self.questions[text])
def load_inputs(root=None):
root=root or Path(__file__).parent/'fixtures'
raw=(root/'vectors.json').read_bytes()
if hashlib.sha256(raw).hexdigest()!=VECTORS_SHA256:raise ValueError('vector_recording_hash_mismatch')
snapshot=json.loads((root/'snapshot.json').read_text())
cases=json.loads((root/'development-cases.json').read_text())
if len(cases)!=4 or any(c['split']!='development' for c in cases):raise ValueError('development_cases_required')
return snapshot,SavedEmbeddings(json.loads(raw),[c['input'] for c in cases]),cases
def version_filter(policy):
if policy is None or policy=={}:return None
if not isinstance(policy,dict) or set(policy)!={'version'}:raise ValueError('unsupported_filter')
operator=policy['version']
if not isinstance(operator,dict) or set(operator)!={'$eq'}:raise ValueError('unsupported_filter_operator')
version=operator['$eq']
if not isinstance(version,str) or re.fullmatch(r'[0-9]+\.[0-9]+',version) is None:raise ValueError('invalid_filter_version')
return version
def rows_from_snapshot(snapshot):
validate_snapshot(snapshot)
rows=[];seen=set()
for source in snapshot['documents']:
if not isinstance(source.get('version'),str) or re.fullmatch(r'[0-9]+\.[0-9]+',source['version']) is None:raise ValueError('invalid_source_version')
for chunk in source['chunks']:
if chunk['chunk_id'] in seen:raise ValueError('duplicate_chunk_id')
seen.add(chunk['chunk_id'])
rows.append({'chunk_id':chunk['chunk_id'],'source_id':source['source_id'],'version':source['version'],'url':source['url'],'text':chunk['text']})
return rows
class VectorRetriever:
def __init__(self,snapshot,embeddings):
self.rows=rows_from_snapshot(snapshot)
self.store=InMemoryVectorStore(embeddings)
self.store.add_documents([Document(id=r['chunk_id'],page_content=r['text'],metadata={k:v for k,v in r.items() if k!='text'}) for r in self.rows],ids=[r['chunk_id'] for r in self.rows])
def search(self,query,k=2,policy=None):
if type(k) is not int or k<1:raise ValueError('invalid_k')
version=version_filter(policy)
selected=None if version is None else lambda doc:doc.metadata['version']==version
matches=self.store.similarity_search_with_score(query,k=len(self.rows),filter=selected)
matches=sorted(matches,key=lambda item:(-item[1],item[0].metadata["chunk_id"]))[:k]
return [{**doc.metadata,'text':doc.page_content,'score':float(score)} for doc,score in matches]python run.py
python -m unittest -v test_vector_retriever.py[
{
"case_id": "utf8-write",
"results": [
{
"source_id": "path-read",
"score": 0.637539995393079
},
{
"source_id": "path-write",
"score": 0.5762678845903342
}
]
},
{
"case_id": "replace-file",
"results": [
{
"source_id": "path-write",
"score": 0.6357498076122466
},
{
"source_id": "path-read",
"score": 0.5547892797395033
}
]
},
{
"case_id": "json-result-type",
"results": [
{
"source_id": "json-text",
"score": 0.652333570145617
},
{
"source_id": "path-read",
"score": 0.42563890525675496
}
]
},
{
"case_id": "backup-schedule",
"results": [
{
"source_id": "path-read",
"score": 0.3202785226491292
},
{
"source_id": "path-write",
"score": 0.26563513588901855
}
]
}
]Inspect the complete saved result
{
"execution_mode": "offline-recorded-vector-replay",
"model_called": false,
"store": "langchain_core.vectorstores.InMemoryVectorStore",
"snapshot_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"development_cases": [
{
"case_id": "utf8-write",
"case_class": "answerable",
"query": "How do I explicitly select UTF-8 when saving text with pathlib?",
"requested_version": "3.12",
"results": [
{
"chunk_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.read_text",
"text": "Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file.",
"score": 0.637539995393079
},
{
"chunk_id": "c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"source_id": "path-write",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text",
"text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
"score": 0.5762678845903342
}
]
},
{
"case_id": "replace-file",
"case_class": "answerable",
"query": "Will write_text replace the contents of an existing file?",
"requested_version": "3.12",
"results": [
{
"chunk_id": "c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"source_id": "path-write",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text",
"text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
"score": 0.6357498076122466
},
{
"chunk_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.read_text",
"text": "Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file.",
"score": 0.5547892797395033
}
]
},
{
"case_id": "json-result-type",
"case_class": "answerable",
"query": "Does json.dumps produce text or a bytes object?",
"requested_version": "3.12",
"results": [
{
"chunk_id": "e5455737b11664449f653280d89dfc2faaa9a14de16e9c3eb24296ca2ceea51c",
"source_id": "json-text",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/json.html#json.dumps",
"text": "json.dumps(obj, ensure_ascii=True) returns a JSON string. Setting ensure_ascii=False leaves non-ASCII characters in that string. It does not write a file or select the encoding used by a file writer.",
"score": 0.652333570145617
},
{
"chunk_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.read_text",
"text": "Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file.",
"score": 0.42563890525675496
}
]
},
{
"case_id": "backup-schedule",
"case_class": "unsupported",
"query": "How frequently does our cloud provider back up these files?",
"requested_version": "3.12",
"results": [
{
"chunk_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.read_text",
"text": "Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file.",
"score": 0.3202785226491292
},
{
"chunk_id": "c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"source_id": "path-write",
"version": "3.12",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text",
"text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
"score": 0.26563513588901855
}
]
}
],
"unknown_query_refusal": "unrecorded_development_query",
"interpretation": "Similarity scores order stored passages; returned passages do not establish that a question is answerable."
}Compare the recorded behavior
Run with the already prepared pinned environment. Compare the candidate text for the UTF-8 writing question with its expected source, then inspect the unsupported backup question. Change a question string in a scratch run: expect an explicit replay refusal. Do not relax the adapter into a fallback that hides an unrecorded input.
Read the explanation
Keep each adapter's input, filtering and score contract explicit; a retrieved passage is not an answer-validation result.
What’s next
Continue with Apply Filters Across Retrievers.
References
The official LangChain InMemoryVectorStore reference and installed LangChain Core 1.7.0 source were checked on 23 October 2026. The pinned implementation applies its filter callback before cosine scoring and selection. Original Qwen model details and query instruction remain in the copied baseline provenance; this lesson executes local replay only.