部署 Qdrant 向量数据库应用模板

cat /home/lighthouse/credentials.txt  

from qdrant_client import models
from qdrant_client import QdrantClient

# Qdrant 服务器的地址和端口
# 如果您是按照标准 Docker 部署的，那么这个地址通常是正确的
QDRANT_HOST = "<http://您的实例公网 IP:6333>"
QDRANT_PORT = 6333
TEST_COLLECTION_NAME = "my_test_collection"

def run_qdrant_test():
    """
    执行一个完整的测试流程来验证 Qdrant 服务是否正常。
    """

    try:
        # 1. 连接到 Qdrant 客户端
        # 如果 Qdrant 服务没有启动，这一步就会抛出异常
        # 使用完整的 URL 来明确指定 HTTP 协议
        # 注意：会有警告提示在不安全连接中使用 API Key，这在测试环境中是可以接受的
        client = QdrantClient(
            url=f"{QDRANT_HOST}:{QDRANT_PORT}",
            api_key="您的 API Key"
        )
        print("✅ 步骤 1/5: 成功连接到 Qdrant 客户端。")

        # 2. 创建一个新的集合
        if client.collection_exists(collection_name=TEST_COLLECTION_NAME):
            client.delete_collection(collection_name=TEST_COLLECTION_NAME)
            print(f"🗑️  删除已存在的集合 '{TEST_COLLECTION_NAME}'。")

        client.create_collection(
            collection_name=TEST_COLLECTION_NAME,
            vectors_config=models.VectorParams(size=4, distance=models.Distance.DOT),
        )
        print(f"✅ 步骤 2/5: 成功创建测试集合 '{TEST_COLLECTION_NAME}'。")

        # 3. 插入一个测试向量
        client.upsert(
            collection_name=TEST_COLLECTION_NAME,
            points=[
                models.PointStruct(id=1, vector=[0.9, 0.1, 0.1, 0.7], payload={"color": "red"})
            ],
            wait=True, # 等待操作完成
        )
        print("✅ 步骤 3/5: 成功插入一个测试向量。")

        # 4. 检索刚才插入的向量，进行验证
        retrieved_points = client.retrieve(
            collection_name=TEST_COLLECTION_NAME,
            ids=[1],
            with_payload=True,
        )
        assert len(retrieved_points) == 1, "检索到的点数量不为1"
        assert retrieved_points[0].payload['color'] == 'red', "检索到的payload不匹配"
        print("✅ 步骤 4/5: 成功检索并验证了插入的向量。")

        # 5. 删除测试集合，进行清理
        client.delete_collection(collection_name=TEST_COLLECTION_NAME)
        print(f"✅ 步骤 5/5: 成功删除测试集合 '{TEST_COLLECTION_NAME}'。")

        print("\\n🎉🎉🎉 恭喜！Qdrant 服务一切正常！🎉🎉🎉")

    except Exception as e:
        print("\\n❌❌❌ 测试失败！❌❌❌")
        print(f"错误详情: {e}")
        print("\\n请检查：")
        print("  - Qdrant Docker 容器是否正在运行？ (使用 `docker ps` 命令查看)")
        print(f"  - 连接地址和端口 ({QDRANT_HOST}:{QDRANT_PORT}) 是否正确？")
        print("  - 防火墙是否阻止了该端口的连接？")
        print("  - API Key是否正确？")
        return False

    return True


if __name__ == "__main__":
    run_qdrant_test()