graph LR
    A[Embeddings模块] -->|向量化| B[VectorStores模块]
    C[Documents模块] -->|数据结构| B
    B -->|适配为| D[Retrievers模块]
    B -->|支撑| E[RAG Chain]
    B -->|集成| F[Agents工具]
    G[向量数据库SDK] -.->|底层存储| B

flowchart TB
    subgraph Application["应用层 - 上层使用方"]
        RAG[RAG Chain<br/>检索增强生成]
        AGENT[Agent Tools<br/>智能体工具]
        QA[QA System<br/>问答系统]
    end

    subgraph Interface["接口层 - 适配器"]
        RETRIEVER[VectorStoreRetriever<br/>检索器适配器]
        RUNNABLE[Runnable接口<br/>LCEL链式调用]
    end

    subgraph Core["核心层 - VectorStore抽象"]
        VECTORSTORE[VectorStore ABC<br/>抽象基类]

        subgraph CoreOps["核心操作接口"]
            ADD[add_documents/add_texts<br/>文档索引]
            SEARCH[similarity_search<br/>相似度检索]
            SEARCHSCORE[similarity_search_with_score<br/>带分数检索]
            MMR[max_marginal_relevance_search<br/>MMR多样性检索]
            DELETE[delete/get_by_ids<br/>CRUD操作]
        end

        subgraph Factory["工厂方法"]
            FROM_DOCS[from_documents<br/>文档工厂]
            FROM_TEXTS[from_texts<br/>文本工厂]
            AS_RETRIEVER[as_retriever<br/>转换为Retriever]
        end
    end

    subgraph Impl["实现层 - 具体数据库"]
        INMEM[InMemoryVectorStore<br/>内存实现]
        CHROMA[Chroma<br/>ChromaDB]
        QDRANT[Qdrant<br/>Qdrant]
        PINECONE[Pinecone<br/>Pinecone]
        FAISS[FAISS<br/>Facebook AI Similarity Search]
        WEAVIATE[Weaviate<br/>混合搜索]
    end

    subgraph Infra["基础设施层"]
        EMB[Embeddings<br/>向量化接口]
        DOC[Document<br/>文档结构]
        UTILS[utils<br/>相似度计算/MMR算法]
        DB[(向量数据库<br/>Chroma/Qdrant/Pinecone)]
    end

    RAG --> RETRIEVER
    AGENT --> RETRIEVER
    QA --> RETRIEVER

    RETRIEVER --> VECTORSTORE
    RUNNABLE -.-> RETRIEVER

    VECTORSTORE --> CoreOps
    VECTORSTORE --> Factory

    CoreOps --> Impl
    Factory --> Impl

    VECTORSTORE <|.. INMEM
    VECTORSTORE <|.. CHROMA
    VECTORSTORE <|.. QDRANT
    VECTORSTORE <|.. PINECONE
    VECTORSTORE <|.. FAISS
    VECTORSTORE <|.. WEAVIATE

    Impl --> EMB
    Impl --> DOC
    Impl --> UTILS
    Impl --> DB

    style VECTORSTORE fill:#e1f5ff,stroke:#0066cc,stroke-width:3px
    style RETRIEVER fill:#e8f5e9,stroke:#00cc66,stroke-width:2px
    style CoreOps fill:#fff4e1,stroke:#ff9900,stroke-width:2px
    style Impl fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px

class VectorStoreRetriever(BaseRetriever):
    vectorstore: VectorStore  # 持有向量存储实例
    search_type: str = "similarity"  # 检索策略
    search_kwargs: dict = {}  # 检索参数

    def _get_relevant_documents(self, query: str) -> list[Document]:
        if self.search_type == "similarity":
            return self.vectorstore.similarity_search(query, **self.search_kwargs)
        elif self.search_type == "mmr":
            return self.vectorstore.max_marginal_relevance_search(query, **self.search_kwargs)
        # ...

vectorstore.add_documents([doc1, doc2, doc3])

docs = vectorstore.similarity_search("查询文本", k=4)

docs = vectorstore.max_marginal_relevance_search("查询", k=4, fetch_k=20)

classDiagram
    class VectorStore {
        <<abstract>>
        +embeddings: Embeddings
        +add_documents(documents) list[str]
        +add_texts(texts, metadatas) list[str]
        +similarity_search(query, k) list[Document]
        +similarity_search_with_score(query, k) list[tuple]
        +max_marginal_relevance_search(query, k, fetch_k) list[Document]
        +delete(ids) bool
        +as_retriever() VectorStoreRetriever
    }

    class Document {
        +page_content: str
        +metadata: dict
        +id: str|None
    }

    class VectorStoreRetriever {
        +vectorstore: VectorStore
        +search_type: str
        +search_kwargs: dict
        +invoke(input) list[Document]
    }

    VectorStore --> Document : returns
    VectorStore --> VectorStoreRetriever : creates

documents: list[Document]  # 文档列表
embedding: Embeddings  # 嵌入模型
**kwargs  # 特定数据库的配置

@classmethod
def from_documents(
    cls,
    documents: list[Document],
    embedding: Embeddings,
    **kwargs,
) -> VectorStore:
    """从文档创建向量存储

    Args:
        documents: 文档列表
        embedding: 嵌入模型
        **kwargs: 数据库特定配置

    Returns:
        VectorStore实例
    """
    # 1) 提取文本和元数据
    texts = [doc.page_content for doc in documents]
    metadatas = [doc.metadata for doc in documents]

    # 2) 调用from_texts
    return cls.from_texts(texts, embedding, metadatas=metadatas, **kwargs)

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# 准备文档
documents = [
    Document(page_content="LangChain是一个LLM应用框架", metadata={"source": "doc1"}),
    Document(page_content="向量数据库用于存储嵌入向量", metadata={"source": "doc2"}),
    Document(page_content="RAG结合检索和生成", metadata={"source": "doc3"}),
]

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

query: str  # 查询文本
k: int  # 返回结果数量,默认4
**kwargs  # 额外搜索参数(如filter)

list[Document]  # 文档列表,按相似度降序排列

def similarity_search(
    self,
    query: str,
    k: int = 4,
    **kwargs,
) -> list[Document]:
    """相似度搜索

    Args:
        query: 查询文本
        k: 返回文档数量
        **kwargs: 额外参数

    Returns:
        最相似的k个文档
    """
    # 1) 将查询文本转换为嵌入向量
    query_embedding = self.embeddings.embed_query(query)

    # 2) 在向量空间中搜索
    docs_and_scores = self._similarity_search_with_relevance_scores(
        query_embedding,
        k=k,
        **kwargs
    )

    # 3) 提取文档(不带分数)
    return [doc for doc, score in docs_and_scores]

def _similarity_search_with_relevance_scores(
    self,
    query_embedding: list[float],
    k: int,
    **kwargs,
) -> list[tuple[Document, float]]:
    """底层向量搜索(子类实现)

    Returns:
        (Document, similarity_score)列表
    """
    # 子类实现,调用具体数据库API
    pass

# 基本搜索
docs = vectorstore.similarity_search("什么是RAG?", k=3)
for doc in docs:
    print(doc.page_content)
# "RAG结合检索和生成"
# "向量数据库用于存储嵌入向量"
# ...

# 带过滤器搜索
docs = vectorstore.similarity_search(
    "LangChain",
    k=2,
    filter={"source": "doc1"}  # 仅搜索source=doc1的文档
)

# 带分数搜索
docs_with_scores = vectorstore.similarity_search_with_score("RAG", k=3)
for doc, score in docs_with_scores:
    print(f"[{score:.2f}] {doc.page_content}")
# [0.95] RAG结合检索和生成
# [0.72] 向量数据库...

query: str  # 查询文本
k: int  # 最终返回文档数量
fetch_k: int  # 候选文档数量
lambda_mult: float  # 相关性权重(0-1),1表示纯相关性,0表示纯多样性

def max_marginal_relevance_search(
    self,
    query: str,
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    **kwargs,
) -> list[Document]:
    """MMR搜索

    Args:
        query: 查询文本
        k: 返回文档数量
        fetch_k: 候选文档数量(>= k)
        lambda_mult: 相关性权重,0.5表示相关性和多样性各占50%

    Returns:
        k个文档,平衡相关性和多样性
    """
    # 1) 计算查询嵌入
    query_embedding = self.embeddings.embed_query(query)

    # 2) 获取fetch_k个候选文档
    candidate_docs = self._similarity_search_with_relevance_scores(
        query_embedding,
        k=fetch_k,
        **kwargs
    )

    # 3) 计算候选文档的嵌入
    candidate_embeddings = [
        self.embeddings.embed_query(doc.page_content)
        for doc, _ in candidate_docs
    ]

    # 4) MMR算法选择k个文档
    selected_indices = maximal_marginal_relevance(
        query_embedding,
        candidate_embeddings,
        lambda_mult=lambda_mult,
        k=k
    )

    # 5) 返回选中的文档
    return [candidate_docs[i][0] for i in selected_indices]

def maximal_marginal_relevance(
    query_embedding: list[float],
    embedding_list: list[list[float]],
    lambda_mult: float,
    k: int
) -> list[int]:
    """MMR算法实现

    选择文档时平衡:
    - 与查询的相关性
    - 与已选文档的差异性
    """
    selected = []
    remaining = list(range(len(embedding_list)))

    # 计算所有文档与查询的相似度
    similarities_to_query = cosine_similarity([query_embedding], embedding_list)[0]

    for _ in range(k):
        if not remaining:
            break

        mmr_scores = []
        for idx in remaining:
            # 相关性分数
            relevance = similarities_to_query[idx]

            # 多样性分数(与已选文档的最大相似度)
            if selected:
                max_sim_to_selected = max(
                    cosine_similarity([embedding_list[idx]], [embedding_list[s]])[0][0]
                    for s in selected
                )
            else:
                max_sim_to_selected = 0

            # MMR分数
            mmr = lambda_mult * relevance - (1 - lambda_mult) * max_sim_to_selected
            mmr_scores.append(mmr)

        # 选择MMR分数最高的文档
        best_idx = remaining[mmr_scores.index(max(mmr_scores))]
        selected.append(best_idx)
        remaining.remove(best_idx)

    return selected

# MMR搜索
docs = vectorstore.max_marginal_relevance_search(
    query="LangChain框架",
    k=4,
    fetch_k=20,
    lambda_mult=0.7  # 70%相关性,30%多样性
)

# 对比:
# similarity_search会返回4个非常相似的文档
# MMR会返回4个相关但内容多样的文档

# 极端情况:
# lambda_mult=1.0: 等同于similarity_search
docs_pure_similarity = vectorstore.max_marginal_relevance_search(
    "query", k=4, lambda_mult=1.0
)

# lambda_mult=0.0: 纯多样性(不推荐)
docs_pure_diversity = vectorstore.max_marginal_relevance_search(
    "query", k=4, lambda_mult=0.0
)

search_type: str  # "similarity"|"mmr"|"similarity_score_threshold"
search_kwargs: dict  # 搜索参数

def as_retriever(self, **kwargs) -> VectorStoreRetriever:
    """转换为Retriever

    Args:
        search_type: 搜索类型
        search_kwargs: 搜索参数

    Returns:
        VectorStoreRetriever实例
    """
    return VectorStoreRetriever(vectorstore=self, **kwargs)

class VectorStoreRetriever(BaseRetriever):
    """向量存储检索器"""

    vectorstore: VectorStore
    search_type: str = "similarity"
    search_kwargs: dict = Field(default_factory=dict)

    def _get_relevant_documents(self, query: str) -> list[Document]:
        """检索相关文档"""
        if self.search_type == "similarity":
            return self.vectorstore.similarity_search(query, **self.search_kwargs)
        elif self.search_type == "mmr":
            return self.vectorstore.max_marginal_relevance_search(query, **self.search_kwargs)
        elif self.search_type == "similarity_score_threshold":
            docs_and_scores = self.vectorstore.similarity_search_with_relevance_scores(
                query, **self.search_kwargs
            )
            # 过滤低于阈值的文档
            score_threshold = self.search_kwargs.get("score_threshold", 0.0)
            return [doc for doc, score in docs_and_scores if score >= score_threshold]

# 创建Retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20}
)

# 在链中使用
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

answer = chain.invoke("什么是RAG?")

# 相似度阈值检索
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8, "k": 5}
)
# 仅返回相似度>=0.8的文档

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1) 准备文档
documents = [
    Document(page_content="LangChain是一个用于构建LLM应用的框架..."),
    Document(page_content="RAG将检索和生成结合..."),
    # ... 更多文档
]

# 2) 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents, embeddings)

# 3) 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 4) 构建RAG链
template = """基于以下上下文回答问题:

上下文: {context}

问题: {question}

答案:"""

prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 5) 使用
answer = chain.invoke("什么是RAG?")
print(answer)

# 添加新文档
new_docs = [
    Document(page_content="新功能介绍...", metadata={"version": "2.0", "doc_id": "doc_123"})
]
doc_ids = vectorstore.add_documents(new_docs)

# 删除旧文档
vectorstore.delete(doc_ids)

# 批量更新
# 1) 删除旧版本
old_ids = vectorstore.get_ids(filter={"version": "1.0"})
vectorstore.delete(old_ids)

# 2) 添加新版本
new_docs = load_updated_documents()
vectorstore.add_documents(new_docs)

# 场景: 不同类型文档使用不同的向量存储
technical_docs = Chroma.from_documents(technical_documents, embeddings, collection_name="technical")
marketing_docs = Chroma.from_documents(marketing_documents, embeddings, collection_name="marketing")

# 创建多检索器
tech_retriever = technical_docs.as_retriever(search_kwargs={"k": 2})
marketing_retriever = marketing_docs.as_retriever(search_kwargs={"k": 2})

# 组合检索
from langchain.retrievers import MergerRetriever

combined_retriever = MergerRetriever(retrievers=[tech_retriever, marketing_retriever])

# 使用
docs = combined_retriever.invoke("产品特性")
# 返回技术文档2篇 + 营销文档2篇

# 原型开发: Chroma(本地,零配置)
from langchain_chroma import Chroma
vectorstore = Chroma.from_documents(docs, embeddings)

# 生产环境(中小规模): Qdrant
from langchain_qdrant import Qdrant
vectorstore = Qdrant.from_documents(
    docs,
    embeddings,
    url="http://localhost:6333",
    collection_name="my_collection"
)

# 生产环境(大规模): Pinecone
from langchain_pinecone import Pinecone
vectorstore = Pinecone.from_documents(
    docs,
    embeddings,
    index_name="my-index"
)

# 推荐: 使用适当的k值
# k太小: 可能遗漏相关信息
# k太大: 增加成本和噪音
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})  # 平衡点

# 推荐: MMR用于需要多样性的场景
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.7}
)

# 推荐: 使用相似度阈值过滤低质量结果
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.75}
)

# 推荐: 使用元数据缩小搜索范围
docs = vectorstore.similarity_search(
    "查询",
    k=4,
    filter={"source": "official_docs", "version": "v2"}
)

# 在retriever中使用
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {"department": "engineering"}
    }
)