向量数据库

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 示例
vec_a = np.array([1.0, 2.0, 3.0])
vec_b = np.array([2.0, 4.0, 6.0])
print(cosine_similarity(vec_a, vec_b))  # 输出: 1.0（方向完全相同）

def euclidean_distance(a, b):
    return np.linalg.norm(a - b)

vec_a = np.array([1.0, 2.0, 3.0])
vec_b = np.array([4.0, 5.0, 6.0])
print(euclidean_distance(vec_a, vec_b))  # 输出: 5.196

def dot_product(a, b):
    return np.dot(a, b)

vec_a = np.array([1.0, 2.0, 3.0])
vec_b = np.array([2.0, 4.0, 6.0])
print(dot_product(vec_a, vec_b))  # 输出: 28.0

pip install chromadb

import chromadb

# 创建 Chroma 客户端（内存模式）
client = chromadb.Client()

# 创建一个集合（Collection），类似于数据库中的表
collection = client.create_collection(
    name="my_knowledge_base",
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# 添加文档（Chroma 内置了默认的嵌入模型）
collection.add(
    documents=[
        "Docker 是一个开源的容器化平台，用于构建、发布和运行应用",
        "Kubernetes 是一个容器编排系统，用于自动化部署和管理容器化应用",
        "Nginx 是一个高性能的 HTTP 服务器和反向代理服务器",
        "Redis 是一个开源的内存数据结构存储系统，可用作数据库和缓存",
        "MySQL 是最流行的开源关系型数据库管理系统",
    ],
    ids=["doc1", "doc2", "doc3", "doc4", "doc5"],
    metadatas=[
        {"category": "container"},
        {"category": "container"},
        {"category": "web"},
        {"category": "database"},
        {"category": "database"},
    ]
)

# 查询相似文档
results = collection.query(
    query_texts=["如何部署容器应用？"],
    n_results=3  # 返回最相似的 3 条
)

print("查询结果：")
for i, doc in enumerate(results["documents"][0]):
    distance = results["distances"][0][i]
    print(f"  [{i+1}] (距离: {distance:.4f}) {doc}")

import chromadb
from chromadb.utils import embedding_functions

# 使用 OpenAI 嵌入模型
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-api-key",
    model_name="text-embedding-3-small"
)

# 创建集合时指定嵌入函数
collection = client.create_collection(
    name="openai_collection",
    embedding_function=openai_ef
)

# 使用持久化客户端，数据保存到磁盘
client = chromadb.PersistentClient(path="./chroma_db")

# 之后的操作与内存模式相同
collection = client.get_or_create_collection("my_collection")

# 只在 category 为 "database" 的文档中检索
results = collection.query(
    query_texts=["高性能缓存方案"],
    n_results=2,
    where={"category": "database"}  # 元数据过滤
)