⚡

SurrealDBPython数据库操作技能 surrealdb-python

本技能提供使用Python操作SurrealDB多模型数据库的完整指南，涵盖CRUD操作、图关系建模、向量相似性搜索和实时查询订阅。适用于构建RAG应用、知识图谱、语义搜索系统和实时数据应用。关键词：SurrealDB Python、多模型数据库、图数据库、向量搜索、RAG应用、实时查询、知识图谱、语义搜索、数据库操作、Python SDK。

RAG应用 0 次安装 0 次浏览 更新于 3/1/2026

名称: surrealdb-python 描述: 使用Python掌握SurrealDB 2.3.x，进行多模型数据库操作，包括CRUD、图关系、向量搜索和实时查询。适用于处理SurrealDB数据库、实现图遍历、基于嵌入的语义搜索或构建RAG应用程序的场景。

SurrealDB Python

概述

SurrealDB是一个多模型数据库，在单一系统中集成了文档、图和向量搜索能力。本技能提供了使用Python SDK处理SurrealDB 2.3.x的全面指导，涵盖了标准数据库操作、图关系、向量相似性搜索和实时数据订阅。

何时使用此技能

在以下场景应用此技能：

  • 建立SurrealDB连接并执行CRUD操作
  • 使用RELATE语句和遍历查询实现图数据库
  • 利用向量嵌入构建语义搜索或RAG（检索增强生成）应用程序
  • 通过实时查询创建实时应用程序
  • 设计结合文档、图和向量的多模型架构

核心能力

1. 连接与认证

使用Python SDK建立与SurrealDB的连接并进行身份验证：

from surrealdb import AsyncSurreal

async with AsyncSurreal("ws://localhost:8000/rpc") as db:
    await db.signin({"user": "root", "pass": "root"})
    await db.use("namespace", "database")
    # 执行操作

2. CRUD操作

使用直观的Python方法执行标准数据库操作：

  • create() - 使用特定ID或自动生成的ID插入新记录
  • select() - 检索单个记录或整个表
  • update() - 替换整个记录内容
  • merge() - 通过合并新字段来部分更新记录
  • patch() - 应用JSON Patch操作进行精确修改
  • upsert() - 不存在则创建，存在则更新
  • delete() - 删除记录或整个表
  • query() - 执行原始SurrealQL进行复杂操作

示例工作流：

# 创建
user = await db.create("user", {
    "name": "Alice",
    "email": "alice@example.com"
})

# 更新特定字段
await db.merge("user:alice", {"age": 30})

# 使用参数查询
results = await db.query(
    "SELECT * FROM user WHERE age > $min_age",
    {"min_age": 25}
)

3. 图数据库操作

利用SurrealDB的原生图能力，无需JOIN即可建模和遍历关系：

创建关系：

# 创建实体
await db.create("person:alice", {"name": "Alice"})
await db.create("person:bob", {"name": "Bob"})

# 创建带元数据的关系
await db.query("""
    RELATE person:alice->knows->person:bob
    SET since = "2024-01-01", strength = "close"
""")

遍历图：

# 查找朋友的朋友
result = await db.query("""
    SELECT ->knows->person->knows->person AS friends_of_friends
    FROM person:alice
""")

# 双向遍历
result = await db.query("""
    SELECT <->connected_to<->city AS connected_cities
    FROM city:nyc
""")

# 递归查询（可变深度）
result = await db.query("""
    SELECT @.{1,5}->manages->person AS management_chain
    FROM person:ceo
""")

关键图特性：

  • RELATE语句用于在节点之间创建边
  • 箭头语法（->、<->）用于直观遍历
  • 使用@.{depth}表示法的递归模式
  • 用于在遍历期间过滤的图子句
  • 存储关系元数据的边表

有关全面的图模式、架构定义和最佳实践，请参阅references/graph_operations.md。

4. 向量搜索与嵌入

使用向量嵌入实现语义搜索和基于相似性的检索：

存储向量：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

# 生成并存储嵌入
text = "SurrealDB是一个多模型数据库"
embedding = model.encode(text).tolist()

await db.create("documents", {
    "content": text,
    "embedding": embedding,
    "metadata": {"source": "docs"}
})

使用KNN进行语义搜索：

# 生成查询嵌入
query_text = "数据库特性"
query_embedding = model.encode(query_text).tolist()

# 查找5个最相似的文档
result = await db.query("""
    SELECT content,
           vector::similarity::cosine(embedding, $query_vector) AS similarity
    FROM documents
    WHERE embedding <|5|> $query_vector
    ORDER BY similarity DESC
""", {"query_vector": query_embedding})

关键向量特性：

  • KNN运算符<|k|>用于k近邻搜索
  • 多种距离度量（余弦、欧几里得、曼哈顿）
  • 结合向量和全文搜索的混合搜索
  • 与流行嵌入模型（OpenAI、HuggingFace、Sentence Transformers）集成

有关完整的RAG实现、嵌入模型比较和优化技术，请参考references/vector_search.md。

5. 实时查询

订阅实时数据更改以用于实时应用程序：

# 启动实时查询
live_id = await db.live("user")

# 订阅更改
async for notification in db.subscribe_live(live_id):
    action = notification['action']  # 'CREATE', 'UPDATE', 'DELETE'
    data = notification['result']
    print(f"检测到更改：{action} - {data}")

# 完成后停止实时查询
await db.kill(live_id)

工作流模式

构建RAG应用程序

  1. 定义架构：

    await db.query("""
    DEFINE TABLE documents SCHEMAFULL;
    DEFINE FIELD content ON TABLE documents TYPE string;
    DEFINE FIELD embedding ON TABLE documents TYPE array;
    DEFINE FIELD metadata ON TABLE documents TYPE object;
""")

  2. 索引文档：

    for doc in documents:
    embedding = model.encode(doc["content"]).tolist()
    await db.create("documents", {
        "content": doc["content"],
        "embedding": embedding,
        "metadata": doc.get("metadata", {})
    })

  3. 语义搜索：

    query_embedding = model.encode("用户查询").tolist()
results = await db.query("""
    SELECT content, metadata,
           vector::similarity::cosine(embedding, $query_vector) AS score
    FROM documents
    WHERE embedding <|5|> $query_vector
    ORDER BY score DESC
""", {"query_vector": query_embedding})

  4. 传递给LLM： 使用检索到的上下文与语言模型进行生成

实现知识图谱

  1. 创建实体：

    await db.create("concept:ai", {"name": "人工智能"})
await db.create("concept:ml", {"name": "机器学习"})

  2. 定义关系：

    await db.query("""
    RELATE concept:ml->is_subset_of->concept:ai
    SET confidence = 0.95
""")

  3. 遍历和查询：

    # 递归查找所有父概念
result = await db.query("""
    SELECT @.{1,}->is_subset_of->concept AS parents
    FROM concept:ml
""")

结合图和向量搜索

利用图关系和语义相似性：

# 通过图关系查找语义相似的文档
result = await db.query("""
    SELECT *,
           vector::similarity::cosine(embedding, $query_vector) AS vec_score
    FROM documents
    WHERE embedding <|10|> $query_vector
      AND ->cited_by->document<-authored_by<-person = $author_id
    ORDER BY vec_score DESC
""", {
    "query_vector": query_embedding,
    "author_id": "person:researcher1"
})

最佳实践

连接管理

  • 使用上下文管理器（async with）进行自动清理
  • 处理SurrealException以进行适当的错误处理
  • 为长时间运行的查询设置适当的超时

架构设计

  • 使用SCHEMAFULL定义架构以确保数据完整性
  • 在频繁查询的字段上创建索引
  • 使用断言进行验证和引用完整性

图操作

  • 在边表上定义唯一索引以防止重复关系
  • 为递归查询添加超时（TIMEOUT 5s）
  • 使用FETCH优化需要相关数据的查询
  • 当关系具有属性时，将元数据存储在边表中

向量搜索

  • 选择适合您领域和延迟要求的嵌入模型
  • 使用余弦相似度时对嵌入进行归一化
  • 为生产工作负载在向量字段上创建索引
  • 对长文档（500-1000个标记）进行分块并重叠以获得更好的检索效果
  • 结合向量搜索和元数据过滤以提高精度

查询优化

  • 参数化查询以防止注入攻击
  • 插入多条记录时进行批量操作
  • 为KNN使用适当的k值（精度用3-5，召回用10-20）
  • 利用混合搜索（向量+全文）以获得最佳结果

常见用例

语义搜索与RAG： 存储文档嵌入并执行相似性搜索，以为语言模型检索相关上下文。

知识图谱： 使用带类型边和元数据的实体建模复杂关系，实现复杂的图遍历查询。

社交网络： 表示用户及其连接，遍历朋友关系，并查找共同连接或推荐。

推荐系统： 结合协同过滤（图关系）和基于内容的过滤（向量相似性）进行混合推荐。

实时应用程序： 订阅数据更改以用于实时仪表板、聊天应用程序或通知系统。

参考文档

本技能包含全面的参考文档：

  • references/graph_operations.md - 图数据库操作、RELATE语法、遍历模式和架构设计的深入指南
  • references/vector_search.md - 向量搜索实现细节、嵌入模型比较、RAG模式和LangChain集成

在实现特定功能或排除故障时加载这些参考文档。

其他资源