依赖服务使用指南

更新时间：2026-05-13

依赖服务使用指南

在高代码编辑器中编写检索逻辑时，你可以使用平台提供的 Python SDK 来调用以下三种依赖服务：

sdk 源码： sdk.tar

服务	说明	SDK 入口函数
Search（文档检索）	基于 DSL 查询语法的文档搜索服务，支持全文检索、向量语义检索等	`create_ontology_service()`
Embedding（向量化）	将文本转换为向量表示，用于语义检索	`create_embedding_service()`
Rerank（重排序）	对候选文档按照与查询的相关性进行重排序，提升检索精度	`create_rerank_service()`

关于服务配置：平台已通过环境变量自动注入依赖服务的 endpoint、apiKey 等连接信息，你无需手动配置，直接调用即可。

1. Search（文档检索）

Search 服务用于根据 DSL 查询语法从数据表中检索文档。底层将 DSL 转换为检索 SQL 语句执行。

1.1 创建服务

                Python
                
                from databuilder_logic_sdk.core import create_ontology_service

# 无需传参，配置由平台自动注入
service = create_ontology_service()

1.2 执行查询

调用 service.search(dsl=...) 方法，传入 DSL 查询字典。DSL 必须包含 ontology（本体名）和 apiName（表名）两个字段。

                Python
                
            

                dsl = {
    "ontology": "my_ontology",       # 本体名
    "apiName": "my_object_type",     # 表名
    "limit": 10,                     # 返回条数上限
    "select": [                      # 选择返回的字段
        {"field": "content"},
        {"field": "title"},
    ],
    "where": {                       # 过滤条件
        "type": "and",
        "filter": [
            {"type": "eq", "field": "status", "value": "active"},
        ]
    }
}

response = service.search(dsl=dsl)

# 响应结构
print(response.code)      # "SUCCESS" 表示成功
print(response.message)   # 响应消息
print(response.result)    # 结果数据字典，通常通过 response.result["data"] 获取数据列表
            

1.3 DSL 语法参考

select 字段

select 指定返回哪些字段。每个元素是一个字典，包含 field（字段名），可选 orderBy（排序方向）。

                Python
                
                "select": [
    {"field": "content"},
    {"field": "__SCORE", "orderBy": "desc"},    # 按相关性分数降序
]

where 过滤条件

where 支持嵌套的逻辑组合和多种过滤类型：

逻辑组合：

                Python
                
                # and 组合
{"type": "and", "filter": [ ... ]}

# or 组合
{"type": "or", "filter": [ ... ]}

过滤类型：

type	说明	示例
`eq`	等值匹配	`{"type": "eq", "field": "status", "value": "active"}`
`in`	多值匹配	`{"type": "in", "field": "id", "value": ["id1", "id2"]}`
`match_any`	全文检索（BM25）	`{"type": "match_any", "field": "content", "value": "搜索关键词"}`
`contains`	向量语义检索（ANN）	`{"type": "contains", "field": "embedding", "value": [0.1, 0.2, ...]}`

1.4 完整示例

                Python
                
            

                from databuilder_logic_sdk.core import create_ontology_service

service = create_ontology_service()

# 全文检索示例
dsl = {
    "ontology": "whb_demo01",
    "apiName": "datasearch0122",
    "limit": 5,
    "select": [
        {"field": "content"},
        {"field": "chunkId"},
        {"field": "__SCORE", "orderBy": "desc"},
    ],
    "where": {
        "type": "and",
        "filter": [
            {"type": "match_any", "field": "content", "value": "人工智能"},
            {"type": "eq", "field": "type", "value": "sentence"},
        ]
    }
}

response = service.search(dsl=dsl)
if response.code == "SUCCESS" and response.result:
    for row in response.result.get("data", []):
        print(f"chunkId: {row.get('chunkId')}, content: {row.get('content')}")
            

1.5 响应模型

字段	类型	说明
`code`	`str`	响应码，`"SUCCESS"` 表示成功
`message`	`str`	响应消息
`result`	`dict`	结果数据，`result["data"]` 为数据列表

2. Embedding（向量化）

Embedding 服务将文本转换为向量（embedding vector），用于语义检索场景。

2.1 创建服务

                Python
                
                from databuilder_logic_sdk.core import create_embedding_service

# 传入模型名称即可，endpoint 和 apiKey 由平台自动注入
service = create_embedding_service("text-embed_7b_bf16")

支持的模型：

模型名称	说明
`text-embed_7b_bf16`	7B 参数文本嵌入模型

2.2 获取向量

调用 service.embed(texts) 方法，传入文本列表，返回对应的向量列表。

                Python
                
                response = service.embed(["你好世界", "人工智能技术"])

for result in response.results:
    print(f"文本索引: {result.index}")
    print(f"向量维度: {len(result.embedding)}")
    print(f"向量前5维: {result.embedding[:5]}")

2.3 完整示例

                Python
                
                from databuilder_logic_sdk.core import create_embedding_service

service = create_embedding_service("bge-large-zh")

# 将查询文本和文档文本一起向量化
texts = ["什么是机器学习？", "机器学习是人工智能的一个分支", "今天天气不错"]
response = service.embed(texts)

query_vector = response.results[0].embedding
doc_vectors = [r.embedding for r in response.results[1:]]

print(f"查询向量维度: {len(query_vector)}")
print(f"文档向量数量: {len(doc_vectors)}")

2.4 响应模型

EmbeddingResponse：

字段	类型	说明
`results`	`list[EmbeddingResult]`	嵌入结果列表

EmbeddingResult：

字段	类型	说明
`embedding`	`list[float]`	嵌入向量
`index`	`int`	对应输入文本的索引位置

3. Rerank（重排序）

Rerank 服务对候选文档按照与查询的语义相关性进行重排序，通常在初次检索后使用以提升结果质量。

3.1 创建服务

                Python
                
                from databuilder_logic_sdk.core import create_rerank_service

# 传入模型名称即可，endpoint 和 apiKey 由平台自动注入
service = create_rerank_service("bce-reranker-base")

支持的模型：

模型名称	说明
`bce-reranker-base`	BCE 基础重排序模型

3.2 执行重排序

调用 service.rerank(query, documents, top_k) 方法：

                Python
                
            

                response = service.rerank(
    query="北京人工智能发展",
    documents=[
        "致力于创建能够执行通常需要人类智能的任务的系统。",
        "机器学习、自然语言处理、计算机视觉等领域。",
        "北京天气预报显示明天晴朗。",
    ],
    top_k=2    # 返回相关性最高的前2个文档
)

for result in response.results:
    print(f"文档: {result.document}")
    print(f"相关性得分: {result.relevance_score}")
    print(f"原始索引: {result.index}")
    print("---")
            

3.4 完整示例

                Python
                
            

                from databuilder_logic_sdk.core import create_rerank_service

service = create_rerank_service("bce-reranker-base")

query = "人工智能在医疗领域的应用"
documents = [
    "深度学习技术被广泛应用于医学影像诊断。",
    "自然语言处理可以帮助医生快速检索病历。",
    "区块链技术在供应链管理中的应用。",
    "机器人辅助手术系统提高了手术精确度。",
]

response = service.rerank(query=query, documents=documents, top_k=3)

print(f"查询: {query}")
print(f"返回 Top {len(response.results)} 结果:")
for r in response.results:
    print(f"  [{r.index}] 得分={r.relevance_score:.4f} | {r.document}")
            

3.5 响应模型

RerankResponse：

字段	类型	说明
`results`	`list[RerankResult]`	重排序结果列表（按相关性降序排列）

RerankResult：

字段	类型	说明
`document`	`str`	文档内容
`relevance_score`	`float`	相关性分数
`index`	`int`	该文档在原始输入列表中的索引位置

4. 组合使用示例

以下示例展示如何将三个服务组合起来实现一个完整的「语义检索 + 重排序」流程：

                Python
                
            

                from databuilder_logic_sdk.core import (
    create_embedding_service,
    create_ontology_service,
    create_rerank_service,
)

# 1. 创建服务
ontology_service = create_ontology_service()
embedding_service = create_embedding_service("bge-large-zh")
rerank_service = create_rerank_service("bce-reranker-base")

# 2. 将查询文本向量化
query = "人工智能在教育领域的应用"
query_embedding = embedding_service.embed([query]).results[0].embedding

# 3. 使用向量进行语义检索
dsl = {
    "ontology": "my_ontology",
    "apiName": "my_object_type",
    "limit": 20,
    "select": [
        {"field": "content"},
        {"field": "chunkId"},
    ],
    "where": {
        "type": "and",
        "filter": [
            {"type": "contains", "field": "embedding", "value": query_embedding},
            {"type": "eq", "field": "type", "value": "chunk"},
        ]
    }
}

response = ontology_service.search(dsl=dsl)
rows = response.result.get("data", [])

# 4. 提取候选文档内容
candidates = [row.get("content", "") for row in rows]
chunk_ids = [row.get("chunkId", "") for row in rows]

# 5. 使用 Rerank 对候选文档进行重排序
if candidates:
    rerank_response = rerank_service.rerank(
        query=query,
        documents=candidates,
        top_k=5
    )

    # 6. 输出最终结果
    sorted_rows = rerank_service.sort(rows, rerank_response.results)
    for i, row in enumerate(sorted_rows):
        score = rerank_response.results[i].relevance_score
        print(f"[{i+1}] 得分={score:.4f} | chunkId={row.get('chunkId')} | {row.get('content', '')[:80]}")
            

5. 异常处理

SDK 定义了以下异常类型，均继承自 DataBuilderLogicSDKError：

异常类	说明
`EnvironmentError`	环境配置解析异常（如环境变量缺失或格式错误）
`AuthenticationError`	认证失败
`ValidationError`	输入参数校验失败
`ConnectionError`	服务连接失败
`RequestError`	请求执行失败

                Python
                
            

                from databuilder_logic_sdk import DataBuilderLogicSDKError, RequestError

try:
    response = service.embed(["测试文本"])
except RequestError as e:
    print(f"请求失败: {e}")
except DataBuilderLogicSDKError as e:
    print(f"SDK 异常: {e}")
except Exception as e:
    print(f"未知异常: {e}")
            

6. API 速查表

create_ontology_service()

                Python
                
                create_ontology_service() -> OntologyService

方法	参数	返回值	说明
`search(dsl)`	`dsl: dict` — 完整的 DSL 查询字典，必须包含 `ontology` 和 `apiName`	`OntologyResponse`	执行文档检索查询

create_embedding_service(model_name)

                Python
                
                create_embedding_service(model_name: str, dimension: int = 0) -> EmbeddingService

参数	类型	说明
`model_name`	`str`	模型名称，如 `"bge-large-zh"`、`"bge-large-en"`、`"text-embed_7b_bf16"`
`dimension`	`int`	向量维度，0 表示使用模型默认维度

方法	参数	返回值	说明
`embed(texts)`	`texts: list[str]` — 待向量化的文本列表	`EmbeddingResponse`	获取文本嵌入向量

create_rerank_service(model_name)

                Python
                
                create_rerank_service(model_name: str) -> RerankService

参数	类型	说明
`model_name`	`str`	模型名称，如 `"bce-reranker-base"`

方法	参数	返回值	说明
`rerank(query, documents, top_k)`	`query: str` — 查询文本`documents: list[str]` — 候选文档列表`top_k: int` — 返回前 K 个结果	`RerankResponse`	对文档进行相关性重排序
`sort(original_obj, rerank_result)`	`original_obj: list[Any]` — 原始对象列表`rerank_result: list[RerankResult]` — 重排序结果	`list[Any]`	根据 rerank 结果对原始对象重排序

评价此篇文章

有帮助没帮助

预置Data Search实现

逻辑建模

百度智能云

百度胜算

百度胜算

依赖服务使用指南

依赖服务使用指南

1. Search（文档检索）

1.1 创建服务

1.2 执行查询

1.3 DSL 语法参考

select 字段

where 过滤条件

1.4 完整示例

1.5 响应模型

2. Embedding（向量化）

2.1 创建服务

2.2 获取向量

2.3 完整示例

2.4 响应模型

3. Rerank（重排序）

3.1 创建服务

3.2 执行重排序

3.4 完整示例

3.5 响应模型

4. 组合使用示例

5. 异常处理

6. API 速查表

create_ontology_service()

create_embedding_service(model_name)

create_rerank_service(model_name)