VectorStore 与 Retriever：让向量索引可以过滤、更新和追踪

python -m pip install "langchain-core>=1,<2"

import hashlib
import math
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import InMemoryVectorStore


class HashEmbeddings(Embeddings):
    def _one(self, text: str) -> list[float]:
        vector = [0.0] * 64
        for char in text:
            index = int(hashlib.sha256(char.encode()).hexdigest()[:8], 16) % 64
            vector[index] += 1
        norm = math.sqrt(sum(x * x for x in vector)) or 1
        return [x / norm for x in vector]

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return [self._one(text) for text in texts]

    def embed_query(self, text: str) -> list[float]:
        return self._one(text)


store = InMemoryVectorStore(HashEmbeddings())
docs = [
    Document(page_content="关键告警必须立即通知负责人", metadata={
        "source": "duty.md", "version": "v1", "scope": "team-a"
    }),
    Document(page_content="恢复后观察十分钟并记录处理人", metadata={
        "source": "recovery.md", "version": "v1", "scope": "team-a"
    }),
]
ids = ["duty:v1:000", "recovery:v1:000"]
store.add_documents(docs, ids=ids)

retriever = store.as_retriever(search_kwargs={"k": 2})
for doc in retriever.invoke("关键告警通知谁"):
    print(doc.metadata, doc.page_content)

store.delete(ids=["duty:v1:000"])

{"index_version":"kb-2026-08-27","embedding_model":"...","dimension":1024,
 "splitter":"recursive-v2","document_version":"v3"}