weaviate 学习笔记

老羊肖恩

2026-03-19 (Updated: 2026-05-06)

笔记

阅读量次

最近在学习 Agent 相关的东西，需要用到向量数据库，这里记录一下 weaviate 的学习笔记

Weaviate

官网：https://weaviate.org.cn

向量数据库和传统关系型数据库的区别：可以进行语义检索和关键词检索

// 比如问：推荐几个笔记本
{
    "A4笔记本",     // 关键字检索
    "联想ThinkPad", //语义检索
    "惠普战99",     // 语义检索
    "....."
}

要实现语义检索，通过向量模型将数据向量化，再进行检索，所以向量模型决定了语义检索的效果

weaviate 内置了多种向量模型，如 text2vec-transformers，img2vec-neural 等，也支持用户使用自定义向量模型

关于向量模型如何选择，需要根据具体的业务需求来选择，可以在这里看到不同向量模型的性能对比：https://huggingface.co/spaces/mteb/leaderboard_dev

索引

关键字检索：

BM25 索引: 基于词频和逆文档频率的经典文本索引方法，适用于关键词检索

语义检索：

Flat 索引: 将所有向量存储在一个列表中，并搜索所有向量以找到最近的邻居
HNSW 索引: 图索引，分层图导航。构建多层图结构，从上往下快速定位最近邻
Dynamic 索引: 自适应切换。数据量小时用Flat，超过阈值（默认1万条）后自动转为HNSW

更详细的可以看官方文档：https://docs.weaviate.org.cn/weaviate/concepts/indexing

基础CURD

import weaviate
from weaviate.classes.config import Property, DataType
from weaviate.classes.config import Configure
import json
import random
import numpy as np


# 模拟一个简单的向量生成函数
def generate_mock_vector(text, dimension=384):
    """生成模拟向量（用于学习演示）"""
    random.seed(hash(text) % 2 ** 32)
    vector = [random.uniform(-1, 1) for _ in range(dimension)]
    # 归一化
    norm = np.linalg.norm(vector)
    vector = [v / norm for v in vector]
    return vector


with weaviate.connect_to_local() as client:
    if client.is_ready():
        print("成功连接到本地Weaviate！")

    # 删除已存在的 Collection
    if client.collections.exists("Article"):
        client.collections.delete("Article")

    try:
        # 修复：使用 vector_config 而不是 vectorizer_config
        articles = client.collections.create(
            name="Article",
            properties=[
                Property(name="title", data_type=DataType.TEXT),
                Property(name="content", data_type=DataType.TEXT),
            ],
            # 使用新的 vector_config 参数
            vector_config=Configure.VectorConfig.self_provided()
        )
        print(f"成功创建Collection: {articles.name}")
        print("注意：此Collection使用自带向量模式，不会自动向量化文本")
    except Exception as e:
        print(f"创建Collection时出错: {e}")

    # 准备示例数据
    data = [
        {
            "title": "成都世运会烟花秀",
            "content": "成都世运会烟花秀于2025年8月7日晚上8点在秦皇湖举行."
        },
        {
            "title": "成都世运会官方APP",
            "content": "作为2025年第12届世界运动会的官方移动应用，'世运通'（Chengdu2025）于近日全新上线，这座城市的热情与世运赛事的精彩，都将浓缩在这一方小小的屏幕里，为全球用户铺就一条通往成都世运会的便捷之路。作为赛会官方移动服务应用平台，世运通以运动员为中心、以赛事为核心，为赛会的组织和运行提供一体化移动整合服务，集成了门票预订、新闻资讯、世运指南等20余项功能。"
        },
        {
            "title": "Weaviate 介绍",
            "content": "Weaviate是一个使用Go语言从头构建的开源向量数据库"
        },
    ]

    articles_collection = client.collections.get("Article")

    # 插入带向量的数据
    for item in data:
        combined_text = item["title"] + " " + item["content"]
        vector = generate_mock_vector(combined_text, dimension=384)

        articles_collection.data.insert(
            properties=item,
            vector=vector
        )

    print(f"{len(data)}篇文章已成功索引（使用自带向量模式）！")

    print("\n" + "=" * 50)
    print("自带向量模式下的搜索示例：")
    print("=" * 50)

    # --- 案例1: 向量搜索 ---
    print("\n--- 向量搜索 ---")
    query_text = "世运会 APP"
    query_vector = generate_mock_vector(query_text, dimension=384)

    try:
        vector_result = articles_collection.query.near_vector(
            near_vector=query_vector,
            limit=2,
            return_metadata=["distance"]
        )

        print(f"查询: '{query_text}'")
        for item in vector_result.objects:
            print(json.dumps({
                "title": item.properties["title"],
                "content": item.properties["content"][:100] + "...",
                "距离": round(item.metadata.distance, 4)
            }, indent=2, ensure_ascii=False))
    except Exception as e:
        print(f"向量搜索出错: {e}")

    # --- 案例2: 关键词搜索 ---
    print("\n--- 关键词搜索 ---")
    try:
        keyword_result = articles_collection.query.bm25(
            query="向量数据库",
            limit=2
        )

        print(f"关键词搜索: '向量数据库'")
        if keyword_result.objects:
            for item in keyword_result.objects:
                print(json.dumps({
                    "title": item.properties["title"],
                    "content": item.properties["content"][:100] + "...",
                    "得分": getattr(item.metadata, 'score', 'N/A')
                }, indent=2, ensure_ascii=False))
        else:
            print("未找到匹配的结果")
    except Exception as e:
        print(f"关键词搜索出错: {e}")

    # --- 案例3: 正确的混合搜索方式 ---
    print("\n--- 混合搜索（使用 Weaviate 的 hybrid 方法）---")
    try:
        # 注意：hybrid 在自带向量模式下需要手动提供查询向量
        # 但 weaviate 的 hybrid 方法需要 vectorizer 支持
        # 这里我们直接使用 near_vector 做纯向量搜索
        query_vec = generate_mock_vector("向量数据库", dimension=384)

        vector_result = articles_collection.query.near_vector(
            near_vector=query_vec,
            limit=2,
            return_metadata=["distance"]
        )

        print(f"向量搜索（相当于混合搜索的向量部分）: '向量数据库'")
        for item in vector_result.objects:
            print(json.dumps({
                "title": item.properties["title"],
                "content": item.properties["content"][:100] + "...",
                "距离": round(item.metadata.distance, 4)
            }, indent=2, ensure_ascii=False))

    except Exception as e:
        print(f"搜索出错: {e}")

    # --- 案例4: 手动计算相似度（完整示例）---
    print("\n--- 手动计算相似度（模拟混合搜索）---")
    try:
        query_vec = generate_mock_vector("向量数据库", dimension=384)

        # 获取所有对象（注意：仅用于演示，生产环境数据量大时不建议）
        all_objects = []
        for obj in articles_collection.iterator():
            all_objects.append(obj)

        results = []
        for obj in all_objects:
            # 获取对象的向量 - 修复 'default' 问题
            if hasattr(obj, 'vector') and obj.vector:
                # 向量可能以字典形式存储，获取第一个向量
                obj_vector = obj.vector
                if isinstance(obj_vector, dict):
                    # 如果向量是字典，取第一个
                    obj_vector = next(iter(obj_vector.values()))

                # 计算余弦相似度
                similarity = np.dot(obj_vector, query_vec) / (np.linalg.norm(obj_vector) * np.linalg.norm(query_vec))
                results.append({
                    "title": obj.properties["title"],
                    "content": obj.properties["content"],
                    "similarity": round(float(similarity), 4)
                })

        # 按相似度排序
        results.sort(key=lambda x: x["similarity"], reverse=True)

        print(f"手动相似度计算: '向量数据库'")
        for r in results[:2]:
            print(json.dumps(r, indent=2, ensure_ascii=False))

    except Exception as e:
        print(f"手动计算出错: {e}")
        import traceback

        traceback.print_exc()

可视化工具

使用 vs code 插件：Weaviate Studio

alt text