发表更新AI / LangChain22 分钟读完 (大约3242个字)

LangChain：集成向量数据库

在RAG（检索增强生成）中，向量存储库（通常指向量数据库）是连接大模型与外部知识源的核心“记忆中枢”，扮演着核心知识库的角色。它的主要作用是让大模型能够突破自身知识的局限，通过高效的语义搜索，实时、准确地利用私有的或最新的数据来生成更准确、更符合实际的答案。

为什么需要向量数据库

RAG（检索增强生成）的工作流程可以概括为“检索+生成”：当用户提出问题时，系统先从外部知识源中检索出最相关的信息片段，然后将这些信息连同问题一起提交给大模型，让其基于这些上下文生成答案。

这里的关键在于如何快速、准确地从海量知识中找到与问题语义匹配的内容。传统的基于关键词的搜索（如SQL的LIKE或Elasticsearch的关键词匹配）无法理解语义，容易漏掉用词不同但意思相近的信息。而向量数据库专门用于存储和检索由深度学习模型生成的“嵌入向量”（即文本的语义表示），能够实现真正的语义搜索，这正是RAG所需要的。

简单来说，RAG需要向量数据库是因为：

  • 实现“语义”检索，而非“关键词”匹配

    • 原理：向量存储库存储的是由“嵌入模型”将文本、图像等数据转换而来的高维向量（可以理解为数据的“语义指纹”）。当用户提问时，系统会将问题也转换成向量，然后在库中快速查找与之“距离”最近（即语义最相似）的向量及其对应的原始内容。

    • 优势：语义理解需求，需要根据问题的含义，而不是关键词，找到相关知识。这种方式能精准理解用户意图。例如，搜索“有什么健康的水果？”时，即使文档中只提到“苹果富含维生素”，也能被成功召回，因为它理解了两者在语义上的相关性。

  • 高效处理海量非结构化数据

    • 企业中的知识库（如PDF、内部Wiki、聊天记录）大多是非结构化的。向量存储库专为处理这些数据设计，能将它们统一转化为可供计算的向量，并进行高效管理和检索。

    • 高效检索需求，面对可能达到百万、亿级的文档片段，必须有一种能在毫秒级完成语义搜索的存储系统。

  • 作为大模型的“外挂大脑”，解决核心痛点

    • 解决知识滞后：大模型的知识截止于训练时，而向量存储库可以随时更新，无需重新训练模型，模型就能基于最新信息回答问题。

    • 缓解“幻觉”问题：通过向大模型提供检索到的、确切的上下文，能有效约束其生成范围，大大提升回答的准确性和可信度。

    • 利用私域数据：无需 costly 的模型微调，只需将企业内部文档存入向量存储库，即可让大模型化身为“业务专家”。

LangChain集成向量数据库

LangChain 集成向量数据库，LangChain > Vector store integrations。

概述

向量存储用于存储嵌入数据并执行相似性搜索。

flowchart LR

    subgraph "📥 索引阶段 (存储)"
        A[📄 Documents] --> B[🔢 Embedding model]
        B --> C[🔘 Embedding vectors]
        C --> D[(Vector store)]
    end

    subgraph "📤 查询阶段 (检索)"
        E[❓ Query text] --> F[🔢 Embedding model]
        F --> G[🔘 Query vector]
        G --> H[🔍 Similarity search]
        H --> D
        D --> I[📄 Top-k results]
    end

    classDef process fill:#DBEAFE,stroke:#2563EB,stroke-width:2px,color:#1E3A8A
    class A,B,C,D,E,F,G,H,I process

接口

LangChain 为向量存储提供了统一接口，允许您：

  • add_documents - 向存储中添加文档。
  • delete - 通过 ID 删除已存储的文档。
  • similarity_search - 查询语义相似的文档。

初始化

要初始化一个向量存储，需要为其提供一个嵌入模型：

1
2
from langchain_core.vectorstores import InMemoryVectorStore
vector_store = InMemoryVectorStore(embedding=SomeEmbeddingModel())

添加文档

可以像这样添加文档对象（包含 page_content 和可选的元数据）：

1
vector_store.add_documents(documents=[doc1, doc2], ids=["id1", "id2"])

删除文档

通过指定 ID 删除：

1
vector_store.delete(ids=["id1"])

相似性搜索

使用 similarity_search 发起语义查询，它将返回最接近的嵌入文档：

许多向量存储支持以下参数：

1
similar_docs = vector_store.similarity_search("your query here")
  • k — 返回的结果数量
  • filter — 基于元数据的条件过滤

相似性度量与索引

嵌入相似性可以通过以下方式计算：

  • 余弦相似度
  • 欧几里得距离
  • 点积

元数据过滤

通过元数据（例如，来源、日期）进行过滤可以优化搜索结果：

1
2
3
4
5
vector_store.similarity_search(
  "query",
  k=3,
  filter={"source": "tweets"}
)

热门集成

选择嵌入模型

OpenAI

1
pip install -qU langchain-openai
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

Azure

1
pip install -qU langchain-azure-ai
1
2
3
4
5
6
7
8
9
10
11
12
13
import getpass
import os

if not os.environ.get("AZURE_OPENAI_API_KEY"):
  os.environ["AZURE_OPENAI_API_KEY"] = getpass.getpass("Enter API key for Azure: ")

from langchain_openai import AzureOpenAIEmbeddings

embeddings = AzureOpenAIEmbeddings(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    openai_api_version=os.environ["AZURE_OPENAI_API_VERSION"],
)

Google Gemini

1
pip install -qU langchain-google-genai
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("GOOGLE_API_KEY"):
  os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter API key for Google Gemini: ")

from langchain_google_genai import GoogleGenerativeAIEmbeddings

embeddings = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")

Google Vertex

1
pip install -qU langchain-google-vertexai
1
2
3
from langchain_google_vertexai import VertexAIEmbeddings

embeddings = VertexAIEmbeddings(model="text-embedding-005")

AWS

1
pip install -qU langchain-aws
1
2
3
from langchain_aws import BedrockEmbeddings

embeddings = BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0")

HuggingFace

1
pip install -qU langchain-huggingface
1
2
3
from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")

Ollama

1
pip install -qU langchain-ollama
1
2
3
from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="llama3")

Cohere

1
pip install -qU langchain-cohere
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("COHERE_API_KEY"):
  os.environ["COHERE_API_KEY"] = getpass.getpass("Enter API key for Cohere: ")

from langchain_cohere import CohereEmbeddings

embeddings = CohereEmbeddings(model="embed-english-v3.0")

Mistral AI

1
pip install -qU langchain-mistralai
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("MISTRALAI_API_KEY"):
  os.environ["MISTRALAI_API_KEY"] = getpass.getpass("Enter API key for MistralAI: ")

from langchain_mistralai import MistralAIEmbeddings

embeddings = MistralAIEmbeddings(model="mistral-embed")

Nomic

1
pip install -qU langchain-nomic
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("NOMIC_API_KEY"):
  os.environ["NOMIC_API_KEY"] = getpass.getpass("Enter API key for Nomic: ")

from langchain_nomic import NomicEmbeddings

embeddings = NomicEmbeddings(model="nomic-embed-text-v1.5")

NVIDIA

1
pip install -qU langchain-nvidia-ai-endpoints
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("NVIDIA_API_KEY"):
  os.environ["NVIDIA_API_KEY"] = getpass.getpass("Enter API key for NVIDIA: ")

from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings

embeddings = NVIDIAEmbeddings(model="NV-Embed-QA")

Voyage AI

1
pip install -qU langchain-voyageai
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("VOYAGE_API_KEY"):
  os.environ["VOYAGE_API_KEY"] = getpass.getpass("Enter API key for Voyage AI: ")

from langchain_voyageai import VoyageAIEmbeddings

embeddings = VoyageAIEmbeddings(model="voyage-3")

IBM watsonx

1
pip install -qU langchain-ibm
1
2
3
4
5
6
7
8
9
10
11
12
13
import getpass
import os

if not os.environ.get("WATSONX_APIKEY"):
  os.environ["WATSONX_APIKEY"] = getpass.getpass("Enter API key for IBM watsonx: ")

from langchain_ibm import WatsonxEmbeddings

embeddings = WatsonxEmbeddings(
    model_id="ibm/slate-125m-english-rtrvr",
    url="https://us-south.ml.cloud.ibm.com",
    project_id="<WATSONX PROJECT_ID>",
)

Fake

1
pip install -qU langchain-core
1
2
3
from langchain_core.embeddings import DeterministicFakeEmbedding

embeddings = DeterministicFakeEmbedding(size=4096)

xAI

1
pip install -qU langchain-xai
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("XAI_API_KEY"):
  os.environ["XAI_API_KEY"] = getpass.getpass("Enter API key for xAI: ")

from langchain.chat_models import init_chat_model

model = init_chat_model("grok-2", model_provider="xai")

Perplexity

1
pip install -qU langchain-perplexity
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("PPLX_API_KEY"):
  os.environ["PPLX_API_KEY"] = getpass.getpass("Enter API key for Perplexity: ")

from langchain.chat_models import init_chat_model

model = init_chat_model("llama-3.1-sonar-small-128k-online", model_provider="perplexity")

DeepSeek

1
pip install -qU langchain-deepseek
1
2
3
4
5
6
7
8
9
import getpass
import os

if not os.environ.get("DEEPSEEK_API_KEY"):
  os.environ["DEEPSEEK_API_KEY"] = getpass.getpass("Enter API key for DeepSeek: ")

from langchain.chat_models import init_chat_model

model = init_chat_model("deepseek-chat", model_provider="deepseek")

选择向量数据库

In-memory

1
pip install -qU langchain-core
1
2
3
from langchain_core.vectorstores import InMemoryVectorStore

vector_store = InMemoryVectorStore(embeddings)

Amazon OpenSearch

1
pip install -qU boto3
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from opensearchpy import RequestsHttpConnection

service = "es"  # 必须将服务设置为 'es'
region = "us-east-2"
credentials = boto3.Session(
    aws_access_key_id="xxxxxx", aws_secret_access_key="xxxxx"
).get_credentials()
awsauth = AWS4Auth("xxxxx", "xxxxxx", region, service, session_token=credentials.token)

vector_store = OpenSearchVectorSearch.from_documents(
    docs,
    embeddings,
    opensearch_url="host url",
    http_auth=awsauth,
    timeout=300,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection,
    index_name="test-index",
)

Astra DB

1
pip install -qU langchain-astradb
1
2
3
4
5
6
7
8
9
from langchain_astradb import AstraDBVectorStore

vector_store = AstraDBVectorStore(
    embedding=embeddings,
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    collection_name="astra_vector_langchain",
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_NAMESPACE,
)

Azure Cosmos DB NoSQL

1
pip install -qU langchain-azure-ai azure-cosmos
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain_azure_ai.vectorstores.azure_cosmos_db_no_sql import (
    AzureCosmosDBNoSqlVectorSearch,
)
vector_search = AzureCosmosDBNoSqlVectorSearch.from_documents(
    documents=docs,
    embedding=openai_embeddings,
    cosmos_client=cosmos_client,
    database_name=database_name,
    container_name=container_name,
    vector_embedding_policy=vector_embedding_policy,
    full_text_policy=full_text_policy,
    indexing_policy=indexing_policy,
    cosmos_container_properties=cosmos_container_properties,
    cosmos_database_properties={},
    full_text_search_enabled=True,
)

Azure Cosmos DB Mongo vCore

1
pip install -qU langchain-azure-ai pymongo
1
2
3
4
5
6
7
8
9
10
from langchain_azure_ai.vectorstores.azure_cosmos_db_mongo_vcore import (
    AzureCosmosDBMongoVCoreVectorSearch,
)

vectorstore = AzureCosmosDBMongoVCoreVectorSearch.from_documents(
    docs,
    openai_embeddings,
    collection=collection,
    index_name=INDEX_NAME,
)

Chroma

1
pip install -qU langchain-chroma
1
2
3
4
5
6
7
from langchain_chroma import Chroma

vector_store = Chroma(
    collection_name="example_collection",
    embedding_function=embeddings,
    persist_directory="./chroma_langchain_db",  # 本地数据保存位置，如不需要可移除
)

CockroachDB

1
pip install -qU langchain-cockroachdb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain_cockroachdb import AsyncCockroachDBVectorStore, CockroachDBEngine

CONNECTION_STRING = "cockroachdb://user:pass@host:26257/db?sslmode=verify-full"

engine = CockroachDBEngine.from_connection_string(CONNECTION_STRING)
await engine.ainit_vectorstore_table(
    table_name="vectors",
    vector_dimension=1536,
)

vector_store = AsyncCockroachDBVectorStore(
    engine=engine,
    embeddings=embeddings,
    collection_name="vectors",
)

Elasticsearch

安装包并使用 start-local 脚本在本地启动 Elasticsearch：

这会创建一个

1
2
pip install -qU langchain-elasticsearch
curl -fsSL https://elastic.co/start-local | sh

elastic-start-local 文件夹。要启动 Elasticsearch：

1
2
cd elastic-start-local
./start.sh

Elasticsearch 将在 http://localhost:9200 可用。elastic 用户的密码和 API 密钥存储在 elastic-start-local 文件夹的 .env 文件中。

1
2
3
4
5
6
7
from langchain_elasticsearch import ElasticsearchStore

vector_store = ElasticsearchStore(
    index_name="langchain-demo",
    embedding=embeddings,
    es_url="http://localhost:9200",
)

FAISS

1
pip install -qU langchain-community
1
2
3
4
5
6
7
8
9
10
11
12
13
import faiss
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_community.vectorstores import FAISS

embedding_dim = len(embeddings.embed_query("hello world"))
index = faiss.IndexFlatL2(embedding_dim)

vector_store = FAISS(
    embedding_function=embeddings,
    index=index,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={},
)

Milvus

1
pip install -qU langchain-milvus
1
2
3
4
5
6
7
8
9
from langchain_milvus import Milvus

URI = "./milvus_example.db"

vector_store = Milvus(
    embedding_function=embeddings,
    connection_args={"uri": URI},
    index_params={"index_type": "FLAT", "metric_type": "L2"},
)

MongoDB

1
pip install -qU langchain-mongodb
1
2
3
4
5
6
7
8
from langchain_mongodb import MongoDBAtlasVectorSearch

vector_store = MongoDBAtlasVectorSearch(
    embedding=embeddings,
    collection=MONGODB_COLLECTION,
    index_name=ATLAS_VECTOR_SEARCH_INDEX_NAME,
    relevance_score_fn="cosine",
)

PGVector

1
pip install -qU langchain-postgres
1
2
3
4
5
6
7
from langchain_postgres import PGVector

vector_store = PGVector(
    embeddings=embeddings,
    collection_name="my_docs",
    connection="postgresql+psycopg://..."
)

PGVectorStore

1
pip install -qU langchain-postgres
1
2
3
4
5
6
7
8
9
10
11
from langchain_postgres import PGEngine, PGVectorStore

engine = PGEngine.from_connection_string(
    url="postgresql+psycopg://..."
)

vector_store = PGVectorStore.create_sync(
    engine=engine,
    table_name='test_table',
    embedding_service=embedding
)

Pinecone

1
pip install -qU langchain-pinecone
1
2
3
4
5
6
7
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

pc = Pinecone(api_key=...)
index = pc.Index(index_name)

vector_store = PineconeVectorStore(embedding=embeddings, index=index)

Qdrant

1
pip install -qU langchain-qdrant
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from qdrant_client.models import Distance, VectorParams
from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient

client = QdrantClient(":memory:")

vector_size = len(embeddings.embed_query("sample text"))

if not client.collection_exists("test"):
    client.create_collection(
        collection_name="test",
        vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE)
    )
vector_store = QdrantVectorStore(
    client=client,
    collection_name="test",
    embedding=embeddings,
)

Oracle AI Database

1
pip install -qU langchain-oracledb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import oracledb
from langchain_oracledb.vectorstores import OracleVS
from langchain_oracledb.vectorstores.oraclevs import create_index
from langchain_community.vectorstores.utils import DistanceStrategy

username = "<username>"
password = "<password>"
dsn = "<hostname>:<port>/<service_name>"

connection = oracledb.connect(user=username, password=password, dsn=dsn)

vector_store = OracleVS(
    client=connection,
    embedding_function=embedding_model,
    table_name="VECTOR_SEARCH_DEMO",
    distance_strategy=DistanceStrategy.EUCLIDEAN_DISTANCE
)

turbopuffer

1
pip install -qU langchain-turbopuffer
1
2
3
4
5
6
7
from langchain_turbopuffer import TurbopufferVectorStore
from turbopuffer import Turbopuffer

tpuf = Turbopuffer(region="gcp-us-central1")
ns = tpuf.namespace("langchain-test")

vector_store = TurbopufferVectorStore(embedding=embeddings, namespace=ns)

Valkey

1
pip install -qU "langchain-aws[valkey]"
1
2
3
4
5
6
7
from langchain_aws.vectorstores import ValkeyVectorStore

vector_store = ValkeyVectorStore(
    embedding=embeddings,
    valkey_url="valkey://localhost:6379",
    index_name="my_index"
)

向量数据库比较

向量存储 按ID删除 过滤 按向量搜索 带分数搜索 异步 通过标准测试 多租户 添加文档时支持ID
AstraDBVectorStore ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
AzureCosmosDBNoSqlVectorStore ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅
AzureCosmosDBMongoVCoreVectorStore ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅
Chroma ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
Clickhouse ✅ ✅ ❌ ✅ ❌ ❌ ❌ ✅
AsyncCockroachDBVectorStore ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
CouchbaseSearchVectorStore ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅
DatabricksVectorSearch ✅ ✅ ✅ ✅ ✅ ❌ ❌ ✅
ElasticsearchStore ✅ ✅ ✅ ✅ ✅ ❌ ❌ ✅
FAISS ✅ ✅ ✅ ✅ ✅ ❌ ❌ ✅
InMemoryVectorStore ✅ ✅ ❌ ✅ ✅ ❌ ❌ ✅
LambdaDB ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅
Milvus ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
Moorcheh ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
MongoDBAtlasVectorSearch ✅ ✅ ✅ ✅ ✅ ✅ ✅ ✅
openGauss ✅ ✅ ✅ ✅ ❌ ✅ ❌ ✅
PGVector ✅ ✅ ✅ ✅ ✅ ❌ ❌ ✅
PGVectorStore ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅
PineconeVectorStore ✅ ✅ ✅ ❌ ✅ ❌ ❌ ✅
QdrantVectorStore ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅
Weaviate ✅ ✅ ✅ ✅ ✅ ❌ ✅ ✅
SQLServer ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅
TurbopufferVectorStore ✅ ✅ ✅ ✅ ❌ ✅ ✅ ✅
ValkeyVectorStore ✅ ✅ ✅ ✅ ❌ ❌ ❌ ✅
ZeusDB ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅
Oracle AI Database ✅ ✅ ✅ ✅ ✅ ✅ ❌ ✅

LangChain：集成向量数据库

http://blog.gxitsky.com/2026/03/16/AI-LangChain-026-VectorStor/

作者

光星

发布于

2026-03-16

更新于

2026-03-16

许可协议

#

喜欢这篇文章？打赏一下作者吧

评论

关注我
×