
Byted Viking Search Knowledgebase
- 2 installs
- 411 repo stars
- Updated August 4, 2026
- bytedance/agentkit-samples
viking-search-knowledgebase is a Claude skill that semantically retrieves ranked chunks from Volcengine Viking knowledge bases via an APIG gateway.
About
viking-search-knowledgebase calls the Volcengine Viking knowledge base API through an APIG gateway to semantically retrieve chunks relevant to a query. It exposes an info action for knowledge-base metadata and a search and auto action for single-base or multi-base parallel retrieval, returning ranked chunks with relevance and rerank scores plus document info. The skill enforces query construction rules that split multi-intent tasks into keyword queries run in parallel for better recall.
- Semantic retrieval of ranked chunks from Viking knowledge bases
- Single-base search and multi-base parallel auto retrieval
- Query-splitting and keyword rules to improve recall
Byted Viking Search Knowledgebase by the numbers
- 2 all-time installs (skills.sh)
- Ranked #13,956 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
byted-viking-search-knowledgebase capabilities & compatibility
Requires APIG gateway access and Viking knowledge base resource ids.
- Capabilities
- semantic retrieval · knowledge base search · query decomposition
- Use cases
- research · web search · memory
- Pricing
- Bring your own API key
What byted-viking-search-knowledgebase says it does
python scripts/viking_search.py --action auto --query "网络故障 原因 分类"
语义检索,根据 query 从知识库中获取相关切片,返回切片列表、相关度分数、文档信息等。
npx skills add https://github.com/bytedance/agentkit-samples --skill byted-viking-search-knowledgebaseAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 2 |
|---|---|
| repo stars | ★ 411 |
| Last updated | August 4, 2026 |
| Repository | bytedance/agentkit-samples ↗ |
What it does
Semantically retrieve relevant chunks from Viking knowledge bases to answer a question.
Who is it for?
Grounding agent answers by retrieving relevant chunks from Viking knowledge bases.
Skip if: Feishu wiki names, which are a separate namespace and not valid collection names.
When should I use this skill?
You need to search data from a knowledge base to answer a user question.
What you get
Ranked knowledge chunks with relevance and rerank scores for each sub-intent query.
By the numbers
- Three actions (info, search, auto)
- Recommended 2-4 parallel queries per multi-intent task
Files
Byted Viking Search Knowledgebase
该 Skill 用于通过 APIG 网关调用火山引擎 Viking 知识库的 API:
/api/knowledge/collection/info:查看知识库详情,获取collection_name和description。仅在路由/连接检查场景使用,不是知识问答的默认入口。/api/knowledge/collection/search_knowledge:语义检索,根据 query 从知识库中获取相关切片,返回切片列表、相关度分数、文档信息等。
---
输入前提(必读)
你接收到的不是用户原话,而是上级 Agent 分配给你的任务描述。这类描述通常具有以下特征,必须在拆分阶段处理掉,不能整段塞进 `--query`:
- 篇幅长(动辄数十到上百字)
- 分点("1、…2、…3、…" 或 "首先…其次…最后…")
- 多子意图叠加(同一段话里夹了"原因 + 步骤 + 流程 + 案例"等多个独立检索目标)
- 含大量过渡词、修饰语、上下文铺垫
直接把任务原文当 query 必然召回失配(向量被多个语义稀释,分数全部偏低)。正确做法是:先拆,再并行检索。
---
默认策略(最重要)
对于"基于知识库回答问题"类需求,默认走 `auto` 多库并行检索,且 query 必须经过【拆分 + 关键词化】处理。 不要在没有充分理由的情况下走 info → 推理 → search 的两步路由。
只有在以下少数场景才偏离默认策略:
| 场景 | 选择 |
|---|---|
| 知识问答(绝大多数) | auto,按"Query 构造规则"拆分多个独立 query 并行检索 |
任务已指定具体 resource_id 或 name | search |
| 配置中知识库数 ≤ 2 且任务是路由决策 | 可选 info 辅助 |
| 任务意图是"连接检查"/"看看这个库通了没"/"调用下这个知识库"/"列一下我有哪些库" | info |
| 任务描述完全无主题关键词(如"帮我查点东西") | 先 info 列表,再回报上级 Agent 请求澄清 |
经验法则:当你不确定走哪个动作时,默认选 `auto`。它本身就是为"未知目标 + 有具体语义"设计的。
---
Query 构造规则(拆分 + 关键词化 + 并行)
search / auto 的 --query 是语义检索向量入口。面对上级 Agent 的长任务描述,必须执行三步处理:
第 1 步:拆分子意图
逐句通读任务描述,识别其中独立的检索目标。每个分点、每个"和/与/以及/同时"连接的并列项,通常都是一个独立子意图。
示例:上级任务 = "排查网络连接失败的问题,需要:1、常见故障原因分类;2、对应的排查解决步骤;3、从易到难的标准化处理流程" → 识别出 3 个子意图:① 故障原因分类 ② 排查解决步骤 ③ 标准化处理流程
第 2 步:每个子意图压缩为关键词 query
对每个子意图,提炼成由 2~5 个核心关键词 构成的短 query,剔除连接词、修饰语、铺垫语。
形态要求:
- 长度:每个 query 控制在 5~15 个汉字 / 10~30 个英文词以内
- 由名词性关键词为主,允许少量动词,避免完整句式和疑问语气
- 保留专有名词、术语、产品名、错误码原文
示例(接上文):
- query₁ = "网络连接失败 故障原因 分类"
- query₂ = "网络故障 排查步骤 解决方法"
- query₃ = "网络故障 标准化处理流程"
第 3 步:并行检索(关键约束)
将拆出的多个 query 分别独立调用 auto,禁止拼接成一个长 query。多次调用应在同一轮内并行发起。
多个 query 之间必须满足:
| 约束 | 说明 | 反例 |
|---|---|---|
| 互相独立 | 每个 query 表达一个完整可检索的子意图 | "故障原因"(太空泛,必须带主题词"网络故障 原因") |
| 无重叠 | 关键词集合之间交集尽量小,不要让多个 query 检索同一片切片 | query₁="网络故障 原因 分类"、query₂="网络故障 原因 类型" ← 重叠过高 |
| 高区分度 | 每个 query 应能命中知识库的不同切片群 | 三个 query 都包含"网络故障 步骤" ← 区分度低 |
| 数量适中 | 通常 2~4 个 query;超过 5 个说明子意图拆得太碎,需合并 | — |
硬禁止清单
- ❌ 把任务原文(含"1、2、3、"分点或"首先…其次…")整段塞进
--query - ❌ 把多个子意图用顿号/逗号拼成一个长 query(如"原因分类、排查步骤、处理流程"——这是 R2 失败的根本原因)
- ❌ 用完整问句作为 query(如"网络连接失败时应该如何排查和处理?"——疑问句式会引入大量无关向量噪声)
- ❌ 串行检索(一个查完再查下一个);多个独立 query 应在同一轮并行发起
---
召回不足时的正确处置(避免无效重试)
如果一次检索的 top 切片明显与意图无关,不要简单放大 `--limit` 重跑同一个 query——top10 已经是相关度排序的前 10 名,把 limit 提到 20/30 只会拿到更不相关的切片,不会让答案变好。
正确做法按优先级:
1. 重写关键词组合:替换同义关键词、调整词序、增删一个核心名词,重试一次。 2. 进一步拆分:如果某个 query 仍承载了过多语义,按"Query 构造规则"再拆出 2 个更窄的子 query 并行检索。 3. 合并过窄 query:如果多个 query 都返回空,可能是拆得过细,尝试合并相邻子意图(仍保持关键词形态)。 4. 换知识库:如果命中库与主题不符,换 resource_id 重新 search,或回到 auto 让多库竞争。 5. 如实回报上级:上述都失败后,向上级 Agent 回报"当前知识库内未检索到与 {子意图} 相关的内容",不要继续盲目放大 limit 或编造来源。
只有当一次召回明显被截断(top-N 都高度相关、分数都很高)时,才考虑加大 --limit。
---
info 接口使用规约(防幻觉)
info 用于查看知识库元数据,不是知识问答的入口。调用前必须满足:
- ✅ 有可靠的
resource_id,且该 ID 在DATABASE_VIKING_COLLECTION列表内(来自配置或上级已明确指定且属于该列表),或 - ✅ 有可靠的
name(来自配置 / 用户原话,且确认为知识库 collection name,不是飞书 Wiki 名)
严格禁止:
- ❌
--name ""传空值 - ❌ 凭直觉/上下文猜一个中文名(如"网络知识库"、"网络")作为
--name。Viking 知识库的collection_name一般是英文/拼音/ID,不支持中文;用中文名几乎必然失败。 - ❌ 把飞书 Wiki 目录名(如
[网络知识库])当作 Viking collection name。飞书 Wiki 是被同步到 Viking 的数据源,与 Viking collection 是两套命名体系,不要混淆。 - ❌ 在任务只是想"基于知识库回答问题"时,先去查
info兜圈子。直接拆分 +auto即可。
何时该用 info:
| 用户意图 | 是否调用 info |
|---|---|
| 调用下这个知识库" / "看看 XXX 知识库连上了没" | ✅ 是,做连接检查 |
| "我有哪些知识库" / "列一下知识库" | ✅ 是,列出元数据 |
| 任务描述完全无主题(如"帮我查点东西") | ✅ 是,列出后回报上级请求澄清 |
| 基于知识库回答具体问题(绝大多数) | ❌ 否,直接拆分 + auto |
| 配置中知识库数量 ≥ 3 | ❌ 否,auto 的并发筛选比 info 推理更可靠 |
---
使用方式
脚本:scripts/viking_search.py
info - 查看知识库详情
获取指定知识库的 collection_name 和 description,用于连接检查或路由决策。
# 方式一:通过 resource_id 查询(推荐,唯一标识,不会歧义)
python scripts/viking_search.py --action info --resource-id <collection_resource_id>
# 方式二:通过 name + project 查询(name 必须来自配置或上级 Agent 指定,不要猜)
python scripts/viking_search.py --action info --name "XXX" --project "default"search - 单库检索
对已确定的知识库执行语义检索。query 必须是经过"Query 构造规则"处理的关键词组合。
# 方式一:通过 resource_id(推荐)
python scripts/viking_search.py --action search --resource-id <resource_id> --query "关键词1 关键词2 关键词3" --limit 10
# 方式二:通过 name + project
python scripts/viking_search.py --action search --name "XXX" --project "default" --query "关键词1 关键词2 关键词3"auto - 多库并行检索(知识问答的默认入口)
对所有有权限的知识库并发执行轻量级检索,这是知识问答的首选。 对于含多个子意图的任务,应分多次并行调用 `auto`,每次传一个独立子意图的关键词 query。
export DATABASE_VIKING_COLLECTION="rid1,rid2,rid3"
# 单一子意图
python scripts/viking_search.py --action auto --query "网络故障 原因 分类"
# 多子意图(在同一轮内并行发起,不要拼接进同一个 query)
python scripts/viking_search.py --action auto --query "网络故障 原因 分类"
python scripts/viking_search.py --action auto --query "网络故障 排查步骤 解决方法"
python scripts/viking_search.py --action auto --query "网络故障 标准化处理流程"再次强调:每个 --query 是关键词组合,不是任务原文;多子意图必须拆分 + 并行,禁止拼接。---
返回说明
info 返回示例
{
"code": 0,
"message": "success",
"data": {
"resource_id": "rid_xxx",
"collection_name": "xxx",
"description": "包含商品信息、订单数据、用户评价等电商相关文档。",
"project": "default"
}
}search 返回示例
{
"code": 0,
"message": "success",
"data": {
"result_list": [
{
"score": 0.892,
"rerank_score": 0.912,
"content": "Mac 配置 Python 开发环境的步骤:首先安装 Homebrew,然后通过 brew install pyenv 来管理 Python 版本...",
"chunk_title": "Python 环境配置",
"chunk_id": "chunk_101",
"doc_info": {
"doc_id": "doc_001",
"doc_name": "Mac 开发环境配置大全.md",
"doc_type": "markdown"
}
}
]
}
}auto 返回示例
{
"mode": "multi",
"query": "Mac 上怎么配 Python 环境?",
"collections": [
{
"resource_id": "rid1_mac_guide",
"search": {
"code": 0,
"data": {
"result_list": [
{
"score": 0.892,
"rerank_score": 0.912,
"content": "Mac 配置 Python 开发环境的步骤:首先安装 Homebrew,然后通过 brew install pyenv...",
"chunk_title": "Python 环境配置",
"chunk_id": "chunk_101",
"doc_info": { "doc_id": "doc_001", "doc_name": "Mac 开发环境配置大全.md", "doc_type": "markdown" }
}
]
}
},
"top_chunks": [
{
"score": 0.892,
"rerank_score": 0.912,
"content": "Mac 配置 Python 开发环境的步骤:首先安装 Homebrew,然后通过 brew install pyenv...",
"chunk_title": "Python 环境配置",
"chunk_id": "chunk_101",
"doc_id": "doc_001",
"doc_name": "Mac 开发环境配置大全.md",
"doc_type": "markdown"
}
]
},
{
"resource_id": "rid2_hr_policy",
"search": { "code": 0, "data": { "result_list": [] } },
"top_chunks": []
}
]
}---
Configuration
环境要求
- Python 3.7+
- requests 库(用于 HTTP 请求)
环境配置说明
本 Skill 执行所需的 API 地址及鉴权 Key 已在执行环境中预先配置。脚本会自动从环境变量读取必要凭证,无需用户干预,不应直接向用户暴露任何敏感配置。
可选环境变量
- DATABASE_VIKING_PROJECT:知识库所属项目名称,默认
default。用于按名称查询/检索时辅助定位。 - DATABASE_VIKING_COLLECTION:逗号分隔的 knowledge collection
resource_id列表,作为auto/info/search的可访问范围;同时也是合法resource_id/name的唯一可信来源——不要猜不在此列表里的 ID 或名称。info/search若指定了列表外的库,脚本会报错:没有权限访问当前知识库数据源。
---
注意事项
- 异常处理:如果脚本返回"检测到环境配置缺失"相关错误,应向上级 Agent 回报:当前知识库查询服务尚未完全配置,建议提示用户联系管理员补充必要的环境参数。
- 权限管控:若
info/search返回没有权限访问当前知识库数据源,说明请求的resource_id或name不在DATABASE_VIKING_COLLECTION内。不要猜测其他 ID/名称重试,应仅使用允许列表中的库,或改走auto。 - 安全:妥善保管 API 凭证及鉴权信息;严禁在输出中泄露任何敏感环境变量或 Key。
- 并发:
auto模式默认并发 8,可通过--max-workers调整。多子意图并行调用时,多次auto应在同一轮内发起。 - 性能:
auto模式的轻量检索默认limit=5;如果 top-N 均高度相关但被截断,再考虑调大--limit,不要把放大 limit 当作召回不准的兜底手段。 - 飞书 Wiki ≠ Viking collection:客户可能在 Viking 中同步了飞书 Wiki,但 Wiki 目录名(中文)≠ Viking collection name。看到飞书 Wiki 目录名时不要把它当作 Viking 知识库的
--name去查。
---
Resources
scripts/
viking_search.py- Viking 知识库检索脚本(支持 info / search / auto 三种动作)
references/
search_knowledge_api.md- 火山引擎 Viking 知识库 搜索 API 文档(原始接口说明)collection_info_api.md- 火山引擎 Viking 知识库 查看知识库详情 API 文档(原始接口说明)
---
本节将说明如何查看一个已创建知识库的信息。 <span id="818f2260"></span>
概述
/api/knowledge/collection/info 接口用于查看知识库详情,根据知识库名称返回知识库的描述,以及知识库配置的实验版本详细信息。 <span id="b2088fbc"></span>
前提条件
完成“签名鉴权方式“页面的注册账号、实名认证、AK/SK 密钥获取和签名获取后,可调用 API 接口实现知识库信息查看的功能。 <span id="5ca54010"></span>
请求接口
| | | | \
| URI | /api/knowledge/collection/info | 统一资源标识符 |
|---|---|---|
| 请求方法 | POST | 客户端对向量数据库服务器请求的操作类型 |
| 请求头 | Content-Type: application/json | 请求消息类型 |
| ^^ | ||
| Authorization: HMAC-SHA256 *** | 鉴权 |
<span id="4b10073f"></span>
请求参数(旗舰版、标准版通用)
| | | | | | \
| 参数 | 类型 | 是否必选 | 默认值 | 参数说明 |
|---|---|---|---|---|
| name | string | 否 | -- | 知识库名称 |
| project | string | 否 | default | 知识库所属项目,获取方式参见文档API 接入与技术支持 |
| :::warning | ||||
| 若需要操作指定项目下的知识库,需正确配置该字段。 | ||||
| ::: | ||||
| resource_id | string | 否 | -- | 知识库唯一 id |
| 可选择直接传 resource_id ,或同时传 name 和 project 作为知识库的唯一标识 |
<span id="85fbb85b"></span> # <span id="d47c3787"></span>
响应消息
| | | \
| 参数 | 参数说明 |
|---|---|
| code | 状态码 |
| message | 返回信息 |
| request_id | 标识每个请求的唯一标识符 |
| data | 检索返回内容 |
data 返回值
| | | | | \
| 字段 | 子字段 | 字段类型 | 说明 |
|---|---|---|---|
| collection_name | -- | string | 知识库名称 |
| version | -- | int | 2:标准版 |
| 4:旗舰版 | |||
| description | -- | string | 知识库描述 |
| doc_num | -- | int | 知识库内文档数 |
| create_time | -- | int | 知识库创建的时刻 |
| update_time | -- | int | 知识库更新的时刻 |
| creator | -- | string | 知识库创建用户 |
| pipeline_list | list | 知识库下实验版本 (pipeline) 列表 | |
| ^^ | |||
| pipeline_stat | json | pipeline 下文档导入状态 | |
| ```JSON | |||
| { | |||
| "doc_num": 1, // 导入文档数 | |||
| "finish_doc_num": 1, // 完成导入文档数 | |||
| "point_num": 1, // 切片数 | |||
| "success_doc_num": 1 // 成功导入文档数(已完成解析切片) | |||
| }, | |||
| ``` | |||
| ^^ | |||
| index_list | list | 知识库索引详情 | |
| ^^ | |||
| preprocessing_list | list | 知识预处理配置 | |
| ```JSON | |||
| { | |||
| "chunking_strategy": "custom_balance",// 切片策略 | |||
| "chunking_identifier": null, // 自定义分隔符 | |||
| "chunk_length": 2000, // 切片最大长度 | |||
| "merge_small_chunks": true // 是否合并短文本片 | |||
| "vlm_prompt": "xxx", // 视频切片规则 | |||
| ... | |||
| } | |||
| ``` | |||
| 完整参数说明参考 请求参数(旗舰版) | |||
| ^^ | |||
| table_config_list | list | 结构化知识库表结构 | |
| ```JSON | |||
| { | |||
| "table_type": "row","col", | |||
| // row表示从行开始解析,col表示从列开始解析, | |||
| "table_pos": "int", | |||
| // 字段位于第几行或第几列, | |||
| "start_pos": "int", | |||
| // 起始数据在第几行, | |||
| "table_fields": [ | |||
| { | |||
| "field_name": "xxx", //字段名称 | |||
| "field_type": "int64", //字段类型, 支持string, int64, float32, bool,list<string> | |||
| "if_embedding": true, //是否参与索引 | |||
| "default_value":"xxx", //默认值 | |||
| "if_filter": false //是否为标签过滤字段 | |||
| }, | |||
| ..... | |||
| ] | |||
| } | |||
| ``` | |||
| ^^ | |||
| data_type | string | 知识库内的数据类型 | |
| resource_id | -- | string | 知识库唯一标识id |
| project | -- | sring | 知识库所属项目 |
| type | list | 知识库类型信息 |
index_list 返回值
| | | | | \
| 字段 | 子字段 | 字段类型 | 说明 |
|---|---|---|---|
| index_type | -- | string | 索引算法 |
| index_config | list | 索引配置详情 | |
| ^^ | |||
| vector_field | json | 稠密向量字段 | |
| ```JSON | |||
| { | |||
| "field_name": "_sys_auto_content_vector", // 字段名 | |||
| "field_type": "vector", // 字段类型 | |||
| "dim": 2048 // 向量维度 | |||
| } | |||
| ``` | |||
| ^^ | |||
| sparse_vector_field | 稀疏向量字段 | ||
| ```JSON | |||
| { | |||
| "field_name": "_sys_auto_content_vector", // 字段名 | |||
| "field_type": "vector", // 字段类型 | |||
| } | |||
| ``` | |||
| ^^ | |||
| cpu_quota | int | CPU 配额 | |
| ^^ | |||
| distance | string | 距离类型 | |
| ^^ | |||
| quant | string | 量化方式 | |
| ^^ | |||
| embedding_model | string | 向量化模型 | |
| ^^ | |||
| embedding_dimension | int | 向量维度 | |
| ^^ | |||
| need_instruction | bool | 是否拼接 instruction 进行检索 | |
| ^^ | |||
| fields | list[object] | 数据集字段详情 | |
| ```JSON | |||
| [ | |||
| { | |||
| "field_name": "_sys_auto_id", // 字段名 | |||
| "field_type": "string", // 字段类型 | |||
| }, | |||
| ...... | |||
| ] | |||
| ``` | |||
| ^^ | |||
| field_enumerated_list | string | 标签列表 | |
| primary_key | -- | sring | 主键 |
| status | -- | int | 索引状态 |
| ```Python | |||
| Status: | |||
| -1: 待构建 | |||
| 0: 构建中 | |||
| 1: 构建完成 | |||
| 2: 构建失败 | |||
| 3: 变更中 | |||
| ``` | |||
<span id="af723eb1"></span>
状态码说明
| | | | | \
| 状态码 | http 状态码 | 返回信息 | 状态码说明 |
|---|---|---|---|
| 0 | 200 | success | 成功 |
| 1000001 | 401 | unauthorized | 鉴权失败 |
| 1000002 | 403 | no permission | 权限不足 |
| 1000003 | 400 | invalid request:%s | 非法参数 |
| 1000005 | 400 | collection not exist | collection不存在 |
<span id="f61abe0e"></span>
完整示例
<span id="7f13d566"></span>
请求消息
curl -i -X POST \
-H 'Content-Type: application/json' \
-H 'Authorization: HMAC-SHA256 ***' \
https://api-knowledgebase.mlp.cn-beijing.volces.com
/api/knowledge/collection/info \
-d '{
"name": "test_collection_name",
"project": ""
}'<span id="416e041f"></span>
响应消息
执行成功返回:
{
"code": 0,
"data": {
"collection_name": "apiexample",
"description": "test",
"create_time": 1724747158,
"update_time": 1724747158,
"creator": "xxx",
"pipeline_list": [
{
"pipeline_type": "user_define",
"pipeline_stat": {
"doc_num": 0,
"finish_doc_num": 0,
"point_num": 0,
"success_doc_num": 0
},
"index_list": [
{
"index_type": "hnsw_hybrid",
"index_config": {
"vector_field": {
"field_name": "_sys_auto_content_vector",
"field_type": "vector",
"dim": 2048
},
"sparse_vector_field": {
"field_name": "_sys_auto_content_sparse_vector",
"field_type": "sparse_vector"
},
"cpu_quota": 1,
"distance": "ip",
"quant": "int8",
"embedding_model": "doubao-embedding-and-m3",
"embedding_dimension": 2048,
"need_instruction": true,
"fields": [
{
"field_name": "_sys_auto_id",
"field_type": "string"
},
{
"field_name": "_sys_auto_doc_id",
"field_type": "string"
},
{
"field_name": "_sys_auto_chunk_id",
"field_type": "int64"
},
{
"field_name": "_sys_auto_doc_type",
"field_type": "string"
},
{
"field_name": "_sys_auto_add_type",
"field_type": "string"
}
]
},
"primary_key": "",
"status": -1
}
],
"preprocessing_list": [
{
"chunking_strategy": "default",
"chunking_identifier": null,
"chunk_length": 2000
}
],
"table_config_list": [
{
"table_type": "row",
"table_pos": 1,
"start_pos": 2,
"table_fields": [
{
"field_name": "讲解模块",
"field_type": "string",
"if_embedding": true,
"if_filter": false
},
{
"field_name": "子模块",
"field_type": "string",
"if_embedding": true,
"if_filter": false
},
{
"field_name": "问题示例",
"field_type": "string",
"if_embedding": true,
"if_filter": false
},
{
"field_name": "记忆化 ————讲解要点",
"field_type": "string",
"if_embedding": true,
"if_filter": false
}
]
}
],
"data_type": "structured_data"
}
],
"resource_id": "kb-be6833502748aaef",
"project": "default"
},
"message": "success",
"request_id": "02172474937697900000000000000000000ffff0a00501d3f1d67"
}执行失败返回:
HTTP/1.1 400 OK
Content-Length: 43
Content-Type: application/json
{"code":1000003, "message":"invalid request:%s", "request_id": "021695029757920fd001de6666600000000000000000002569b8f"}---
本节将说明如何基于一个已创建的知识库或某个实验版本进行语义检索
支持通过指定 pipeline_name 参数,来实现仅检索某个实验版本下的切片
<span id="170e6d2a"></span>
概述
/api/knowledge/collection/search_knowledge 接口用于对知识库进行检索和前后处理,当前会默认对原始文本加工后的知识内容进行检索
<span id="b42b015b"></span>
前提条件
- 知识库创建完成、文档导入且处理完成后,即代表可以进行在线检索
- 调用接口前请先完成 签名鉴权与调用示例 页面的注册账号、实名认证、AK/SK 密钥获取和签名获取
- search 和 search_knowledge 接口的区别:search_knowledge 接口是知识库在线链路升级后的最新接口,在原本 search 接口的基础上支持了多轮改写、文档聚合排序等新功能,与 chat_completions 接口联动,可以完成标准的检索生成链路
<span id="00f93c31"></span>
请求接口
| | | | \
| URI | /api/knowledge/collection/search_knowledge | 统一资源标识符 |
|---|---|---|
| 请求方法 | POST | 客户端对向量数据库服务器请求的操作类型 |
| 请求头 | Content-Type: application/json | 请求消息类型 |
| ^^ | ||
| Authorization: HMAC-SHA256 *** | 鉴权 |
<span id="2a020d4e"></span>
请求参数
| | | | | | | \
| 参数 | 子参数 | 类型 | 是否必选 | 默认值 | 参数说明 |
|---|---|---|---|---|---|
| name | -- | string | 否 | -- | 知识库名称 |
| project | -- | string | 否 | default | 知识库所属项目,获取方式参见文档API 接入与技术支持 |
| :::warning | |||||
| 若不指定该字段,则在default项目下创建。 | |||||
| 若需要操作指定项目下的知识库,需正确配置该字段。 | |||||
| ::: | |||||
| resource_id | -- | string | 否 | -- | 知识库唯一 id |
| 可选择直接传 resource_id,或同时传 name 和 project 作为知识库的唯一标识 | |||||
| query | \ | ||||
| -- | \ | ||||
| string | \ | ||||
| 是 | -- | 检索文本 | |||
| * 最大可输入长度为 8000,query 长度 > 8000 时,接口报错 | |||||
| * 所选 embedding 模型输入最大长度 < query 长度 < 8000 时,query 按所选模型自动截断 | |||||
| * query 长度 < 所选 embedding 模型输入最大长度时,正常检索返回目标切片 | |||||
| image_query | -- | string | 否 | -- | 检索图片 |
| 支持图片 URL 或 Base64 编码,详细要求见图片像素说明和图片文件格式 | |||||
| * 图片 URL 传入:适用于图片文件已存在公网可访问 URL 的场景,单张图片小于 10 MB | |||||
| Base64 编码传入:适用于图片文件较小的场景,支持 JPEG、PNG、WebP、BMP* 四种格式的 Base64 编码,单张图片小于 3 MB,请求体不能超过 4 MB | |||||
| limit | -- | int | 否 | 10 | 检索结果数量 |
| * 数量要求:[1, 1000] | |||||
| query_param | \ | ||||
| json | 否 | 检索的过滤和返回设置 | |||
| ^^ | |||||
| doc_filter | \ | ||||
| map | \ | ||||
| 否 | -- | 检索过滤条件 | |||
| * 支持对 doc 的 meta 信息过滤 | |||||
| * 详细使用方式和支持字段见filter表达式,可支持对 doc_id 做筛选 | |||||
| * 此处用过过滤的字段,需要在 collection/create 时添加到 index_config 的 fields 上 | |||||
| 例如: | |||||
| 单层 filter: | |||||
| ```JSON | |||||
| doc_filter = { | |||||
| "op": "must", // 查询算子 must/must_not/range/range_out | |||||
| "field": "doc_id", | |||||
| "conds": ["tos_doc_id_123", "tos_doc_id_456"] | |||||
| } | |||||
| query_param = { | |||||
| "doc_filter": doc_filter | |||||
| } | |||||
| ``` | |||||
| 多层 filter: | |||||
| ```JSON | |||||
| doc_filter = { | |||||
| "op": "and", // 逻辑算子 and/or | |||||
| "conds": [ // 条件列表,支持嵌套逻辑算子和查询算子 | |||||
| { | |||||
| "op": "must", | |||||
| "field": "type", | |||||
| "conds": [1] | |||||
| }, | |||||
| { | |||||
| ... // 支持>=1的任意数量的条件进行组合 | |||||
| } | |||||
| ] | |||||
| } | |||||
| query_param = { | |||||
| "doc_filter": doc_filter | |||||
| } | |||||
| ``` | |||||
| dense_weight | \ | ||||
| -- | \ | ||||
| float | 否 | 0.5 | 混合检索中稠密向量的权重 | ||
| * 1 表示纯稠密检索 ,0 表示纯字面检索,范围 [0.2, 1] | |||||
| * 只有在请求的知识库使用的是混合检索时有效,即索引算法为 hnsw_hybrid | |||||
| pre_processing | \ | ||||
| \ | |||||
| \ | |||||
| \ | |||||
| json | 检索预处理 | ||||
| ^^ | |||||
| need_instruction | bool | 否 | False | 是否拼接 instruction 进行检索 | |
| ^^ | |||||
| return_token_usage | bool | 否 | False | 是否返回 search 流程中各阶段的 token 使用量 | |
| ^^ | |||||
| rewrite | bool | 否 | False | 是否对 query 进行改写 | |
| 根据 messages 字段传入的历史对话信息进行改写,最多 3 轮 | |||||
| 注:只有在messages字段长度大于2且不为空时,设置参数值为True,才能返回有效的rewrite_query; | |||||
| ```JSON | |||||
| "messages":[ | |||||
| {"role": "user", "content": "prompt 1"}, | |||||
| {"role": "assistant", "content": "prompt2"}, | |||||
| {"role": "user", "content": "prompt 3"}, | |||||
| ] | |||||
| ``` | |||||
| ^^ | |||||
| messages | json | 是 | -- | 多轮对话信息 | |
| 仅开启改写时需要上传,可根据历史对话内容进行问题改写,注意上传对话轮数需 >= 1 | |||||
| 发出消息的对话参与者角色,可选值包括: | |||||
| * user:User Message 用户消息 | |||||
| * assistant:Assistant Message 对话助手消息 | |||||
| ```JSON | |||||
| [ | |||||
| {"role": "user", "content": "知识库支持哪些文档格式?"}, | |||||
| {"role": "assistant", "content": "知识库支持结构化和非结构化文档,其中结构化文档支持 excel、csv、jsonl 等常见格式,非结构化文档支持 pdf、docx、ppt 等常见格式。"}, | |||||
| {"role": "user", "content": "那大小呢?"}, | |||||
| ] | |||||
| ``` | |||||
| post_processing | \ | ||||
| \ | |||||
| \ | |||||
| \ | |||||
| \ | |||||
| \ | |||||
| \ | |||||
| \ | |||||
| json | 检索后处理 | ||||
| ^^ | |||||
| rerank_switch | bool | 否 | False | 自动对结果做 rerank | |
| 打开后,会自动请求 rerank 模型排序 | |||||
| ^^ | |||||
| retrieve_count | \ | ||||
| int | 否 | \ | |||
| 25 | \ | ||||
| 进入重排的切片数量,默认为 25 | |||||
| 只有在 rerank_switch 为 True 时生效。retrieve_count 需要大于等于 limit,否则会抛出错误 | |||||
| ^^ | |||||
| chunk_diffusion_count | \ | ||||
| int | 否 | 0 | 检索阶段返回命中切片的上下几片邻近切片 | ||
| 默认为 0,表示不进行 chunk diffusion。范围 [0, 5] | |||||
| ^^ | |||||
| chunk_group | \ | ||||
| bool | 否 | False | 文本聚合 | ||
| 默认不聚合,对于非结构化文件,考虑到原始文档内容语序对大模型的理解,可开启文本聚合。开启后,会根据文档及文档顺序,对切片进行重新聚合排序返回 | |||||
| ^^ | |||||
| rerank_model | string | 否 | "base-multilingual-rerank" | rerank 模型选择 | |
| 仅在 "rerank_switch" == True 的时候生效 | |||||
| 可选模型: | |||||
| * "doubao-seed-rerank"(即 doubao-seed-1.6-rerank):字节自研多模态重排模型、支持文本 / 图片 / 视频混合重排、精细语义匹配、可选阈值过滤与指令设置 | |||||
| * "base-multilingual-rerank":速度快、长文本、支持70+种语言 | |||||
| * "m3-v2-rerank":常规文本、支持100+种语言 | |||||
| ^^ | |||||
| rerank_threshold | float | 否 | -- | 阈值过滤 | |
| 仅当 rerank_model=="doubao-seed-rerank" 时生效,用于设置重排分数的过滤阈值,低于阈值的结果将不会被返回,取值范围为 0 到 1 | |||||
| ^^ | |||||
| rerank_instruction | string | 否 | -- | rerank 指令 | |
| 仅在 "rerank_switch" == True 且 "rerank_model" == "doubao-seed-rerank" 时生效,用于提供给模型一个明确的排序指令,提升重排效果。字符串长度不超过 1024 | |||||
| 如,Whether the document answers the query or matches the content retrieval intent | |||||
| ^^ | |||||
| rerank_only_chunk | \ | ||||
| bool | 否 | False | \ | ||
| 是否仅根据 chunk 内容计算重排分数 | |||||
| 可选值: | |||||
| * True: 只根据 chunk 内容计算分 | |||||
| * False:根据 chunk title + 内容 一起计算排序分 | |||||
| ^^ | |||||
| get_attachment_link | bool | 否 | False | 是否获取切片中图片的临时下载链接 | |
| pipeline_name | -- | string | 否 | -- | 实验版本名称 |
| * 指定当前参数可查询具体实验版本下的切片列表 | |||||
| * 不指定默认查询知识库主版本下的切片列表 |
<span id="5c1b5386"></span>
响应消息
| | | \
| 参数 | 参数说明 |
|---|---|
| code | 状态码 |
| message | 返回信息 |
| request_id | 标识每个请求的唯一标识符 |
| data | 检索召回切片信息 |
data 返回值
| | | | | \
| 字段 | 子字段 | 字段类型 | 说明 |
|---|---|---|---|
| collection_name | -- | string | 检索知识库名字 |
| count | -- | int | 检索返回的切片数量 |
| rewrite_query | -- | string | 改写的 query |
| token_usage | list | token 用量信息 | |
| ^^ | |||
| embedding_token_usage | 检索向量化阶段的 token 用量 | ||
| ```JSON | |||
| { | |||
| "prompt_tokens": 16, | |||
| "completion_tokens": 0, | |||
| "total_tokens": 16 | |||
| } | |||
| ``` | |||
| ^^ | |||
| rerank_token_usage | int | 在重排阶段的 token 用量 | |
| ^^ | |||
| rewrite_token_usage | int | query 改写的 token 用量 | |
| result_list | list | 返回切片信息 | |
| ^^ | |||
| id | string | 索引的主键 | |
| ^^ | |||
| content | string | 切片内容 | |
| 1、非结构化文件:content 返回切片内容 | |||
| 2、faq 文件:content 返回答案 | |||
| 3、结构化文件:content 返回参与索引的字段和取值,以 K:V 对拼接,使用 \n 区隔 | |||
| 4、音视频文件:content 返回音视频切片说话人,起始和结束时间单位 ms,通过 asr 转换的文本内容,如: | |||
| ```Plain Text | |||
| 说话人1[620,4160]:xxxxxx\n 说话人2[7660,19830]:xxxxxx\n | |||
| ``` | |||
| ^^ | |||
| md_content | string | markdown 格式的解析结果 | |
| 对于非结构化文档中的表格切片,可以额外返回 markdown 格式解析结果,保留更多表格原始信息 | |||
| 表格切片可以通过 chunk_type == table 判断 | |||
| ^^ | |||
| html_content | string | html 格式的解析结果 | |
| 对于非结构化文档中的表格切片,可以额外返回 html 格式解析结果,保留更多表格原始信息 | |||
| 表格切片可以通过 chunk_type == table 判断 | |||
| ^^ | |||
| description | string | 若创建知识库时开启切片内容分析,即enable_slice_analysis=true,此字段则返回切片内容分析结果 | |
| ^^ | |||
| table_chunk_fields | list<object> | 结构化数据检索返回单行全量数据 | |
| ^^ | |||
| original_question | string | faq 数据检索召回答案对应的原始问题 | |
| ^^ | |||
| score | float | 向量化语义检索得分 | |
| ^^ | |||
| point_id | string | 切片 id | |
| ^^ | |||
| chunk_title | string | 切片标题 | |
| ^^ | |||
| chunk_id | int | 切片位次 id | |
| 代表在原始文档中的位次顺序 | |||
| ^^ | |||
| process_time | int | 检索耗时(s) | |
| ^^ | |||
| rerank_score | float | 重排得分 | |
| ^^ | |||
| doc_info | list | 切片所属文档信息 | |
| ```JSON | |||
| { | |||
| "doc_id": "_sys_auto_gen_doc_id-134144883689", //文档 id | |||
| "doc_name": "2404.08817v2.pdf", //文档名字 | |||
| "create_time": 1727333117, //文档的创建时间 | |||
| "doc_type": "pdf", //知识所属原始文档的类型 | |||
| "doc_meta": "[{\"field_name\":\"doc_id\",\"field_type\":\"string\",\"field_value\":\"_sys_auto_gen_doc_id-13411829101044883689\"}]", //文档相关元信息(此处是一个包含文档 id 信息的列表形式的字符串) | |||
| "source": "url", //知识来源类型:tos lark tos_fe | |||
| "title": "Revisiting Code Similarity Evaluation with Abstract Syntax Tree Edit Distance", //知识所属文档的标题 | |||
| "url": "***" //原始文档的公开下载链接或飞书文档链接 | |||
| } | |||
| ``` | |||
| ^^ | |||
| recall_position | int | 向量化语义检索召回位次 | |
| ^^ | |||
| rerank_position | int | 重排位次 | |
| ^^ | |||
| chunk_type | string | 切片类型 | |
| 部分返回值类型有:doc-image、image、video、audio、table、mixed-table、text、structured、faq等 | |||
| ^^ | |||
| chunk_attachment | list<object> | 检索召回附件的临时下载链接,有效时间 10 分钟 | |
| chunk_type 为 image/doc_image 且 get_attachment_link 为 True 时,返回原始图片下载链接 | |||
| chunk_type 为 video 且 get_attachment_link 为 True 时,该字段会返回视频切片所抽取的关键帧图片列表,图片按照时间从小到大排序 | |||
| chunk_type 为 table/mixed-table 且 get_attachment_link 为 True 时,该字段会返回表格的图片或表格中所包含的图片列表 | |||
| ^^ | |||
| original_coordinate | 切片在原文中的位置坐标 | ||
| 目前仅支持 pdf 和 ppt 文档,返回示例如下: | |||
| ```Plain Text | |||
| { | |||
| "page_no": [0, 2], | |||
| "bbox": [ | |||
| [0.43075400818407467, 0.05201688247011952, 0.8814365734218823, 0.06268848178713717], | |||
| [0.1144293768909557, 0.08121951665355412, 0.8790083028120703, 0.7626920876297637] | |||
| ] | |||
| } | |||
| ``` | |||
| audio_start_time | long | 切片开始毫秒 | |
| audio_end_time | long | 切片结束毫秒 |
table_chunk_fields 返回值
| | | | \
| 字段 | 字段类型 | 说明 |
|---|---|---|
| field_name | \ | |
| string | 结构化数据的表字段名称 | |
| field_value | -- | 结构化数据的表字段取值 |
| 字段类型以创建知识库时表字段定义为准 |
chunk_attachment 返回值
| | | | \
| 字段 | 字段类型 | 说明 |
|---|---|---|
| uuid | \ | |
| string | 附件的唯一标识 | |
| caption | string | 图片所属标题,若未识别到标题则值为"\n" |
| type | string | image 等 |
| link | string | "get_attachment_link“ == True 时返回图片或视频抽帧图列或表格包含图片的临时下载链接,有效期 10 分钟 |
<span id="7086e767"></span>
状态码说明
| | | | | \
| 状态码 | http状态码 | 返回信息 | 状态码说明 |
|---|---|---|---|
| 0 | 200 | success | 成功 |
| 1000001 | 401 | unauthorized | 缺乏鉴权信息 |
| 1000002 | 403 | no permission | 权限不足 |
| 1000003 | 400 | invalid request:%s | 非法参数 |
| 1000005 | 400 | collection not exist | collection 不存在 |
<span id="a39afcd5"></span>
完整示例
<span id="367f68f4"></span>
请求消息
curl -i -X POST \
-H 'Content-Type: application/json' \
-H 'Authorization: HMAC-SHA256 ***' \
https://api-knowledgebase.mlp.cn-beijing.volces.com/api/knowledge/collection/search_knowledge \
-d '{
"name": "your_collection",
"query": "test",
"limit": 2,
"query_param": {
"doc_filter": {
"op": "must",
"field": "doc_id",
"conds": ["tos_doc_id_123", "tos_doc_id_456"]
}
},
"dense_weight": 0.5,
"pre_processing": {
"need_instruction": True,
"rewrite": True,
"messages": [
{
"role": "system",
"content": "prompt template"
},
{
"role": "user",
"content": "history content"
}
{
"role": "assistant",
"content": "history content"
},
{
"role": "user",
"content": "history content"
},
...
{
"role": "assistant",
"content": "history content"
}
],
"return_token_usage": True
},
"post_processing": {
"rerank_switch": False,
"rerank_model": "base-multilingual-rerank",
"rerank_only_chunk": False,
"retrieve_count": 25,
"endpoint_id": "ep",
"chunk_group": False,
"get_attachment_link": False
}
}
}
}'<span id="9ff8b1ec"></span>
响应消息
执行成功返回:
HTTP/1.1 200 OK
Content-Length: 209
Content-Type: application/json
{
"code": 0,
"data": {
"collection_name": "example",
"count": 2,
"rewrite_query": "xxx",
"token_usage": {
"embedding_token_usage": {
"prompt_tokens": 16,
"completion_tokens": 0,
"total_tokens": 16
},
"rerank_token_usage": 0
},
"result_list": [
{
"id": "_sys_auto_gen_doc_id-13411829101044883689-15",
"content": "content",
"score": 0.2639991044998169,
"point_id": "_sys_auto_gen_doc_id-13411829101044883689-15",
"chunk_title": "title",
"chunk_id": 15,
"process_time": 1727333127,
"doc_info": {
"doc_id": "_sys_auto_gen_doc_id-13411829101044883689",
"doc_name": "2404.08817v2.pdf",
"create_time": 1727333117,
"doc_type": "pdf",
"doc_meta": "[{\"field_name\":\"doc_id\",\"field_type\":\"string\",\"field_value\":\"_sys_auto_gen_doc_id-13411829101044883689\"}]",
"source": "tos_fe",
"title": "title"
},
"recall_position": 1,
"chunk_type": "text"
},
{
"id": "_sys_auto_gen_doc_id-13411829101044883689-7",
"content": "content",
"score": 0.2583845257759094,
"point_id": "_sys_auto_gen_doc_id-13411829101044883689-7",
"chunk_title": "title",
"chunk_id": 7,
"process_time": 1727333127,
"doc_info": {
"doc_id": "_sys_auto_gen_doc_id-13411829101044883689",
"doc_name": "2404.08817v2.pdf",
"create_time": 1727333117,
"doc_type": "pdf",
"doc_meta": "[{\"field_name\":\"doc_id\",\"field_type\":\"string\",\"field_value\":\"_sys_auto_gen_doc_id-13411829101044883689\"}]",
"source": "tos_fe",
"title": "title"
},
"recall_position": 2,
"chunk_type": "text"
}
]
},
"message": "success",
"request_id": "02172740884343900000000000000000000ffff0a00406f8a8861"
}执行失败返回:
HTTP/1.1 400 OK
Content-Length: 43
Content-Type: application/json
{"code":1000003, "message":"invalid request:%s", "request_id": "021695029757920fd001de6666600000000000000000002569b8f"}#!/usr/bin/env python3
"""Viking 知识库检索
支持三种用法:
1) 意图明确 + 已知目标知识库:
- 模型 在路由时明确知道目标知识库,通过 --resource-id 或 --name 指定。
- 直接对指定库执行 search,返回精准检索结果。
2) 意图明确 + 未知目标知识库(需推理路由):
- 用户有明确问题,但不知道该查哪个库。
- 模型需先调用 info 动作获取所有有权限库的 collection_name 和 description,
结合用户问题推理出目标库后,再调用 search 精准检索。
3) 意图不明确:多库并行检索(auto 模式):
- 用户的问题无法判断具体目标,或推理后仍不确定。
- 调用 auto 模式,脚本对所有有权限库并发执行轻量级 search,
直接返回各库的检索结果列表,供模型做最终决策。
鉴权方式:API Gateway 代理网关(Bearer apikey)
- DATABASE_VIKING_APIG_URL: 代理网关 URL(形如 http(s)://xxxx.apigateway-...volceapi.com)
- DATABASE_VIKING_APIG_KEY: 代理网关 API Key
有权限知识库列表:
- DATABASE_VIKING_COLLECTION: 逗号分隔的 knowledge collection resource_id 列表
权限管控:
- info / search 仅允许访问 DATABASE_VIKING_COLLECTION 中的知识库;
- 通过 --name 定位时,仅在允许列表内解析 collection_name,不对外部库发起 API 查询。
依赖:Python 3.7+、requests
"""
import argparse
import json
import os
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
SEARCH_KNOWLEDGE_PATH = "/api/knowledge/collection/search_knowledge"
COLLECTION_INFO_PATH = "/api/knowledge/collection/info"
ALLOWED_COLLECTION_IDS_ENV = "DATABASE_VIKING_COLLECTION"
PERMISSION_DENIED_MSG = "没有权限访问当前知识库数据源"
def _parse_csv_list(raw: str):
if not raw:
return []
return [part.strip() for part in raw.split(",") if part.strip()]
def _normalize_text(text: str):
return re.sub(r"\s+", "", (text or "")).lower()
def _truncate(text: str, max_len: int):
if text is None:
return ""
if len(text) <= max_len:
return text
return text[: max_len - 1] + "…"
class ApigKnowledgeClient:
"""通过 APIG 代理网关访问 Viking 知识库 API。"""
def __init__(self, *, apig_url: str, apig_key: str):
if not apig_url:
raise ValueError("apig_url is required")
if not apig_key:
raise ValueError("apig_key is required")
self.apig_url = apig_url.rstrip("/")
self.apig_key = apig_key
def _post(self, path: str, payload_dict: dict):
url = f"{self.apig_url}{path}"
payload = json.dumps(payload_dict, ensure_ascii=False)
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.apig_key}",
}
response = requests.post(url, headers=headers, data=payload.encode("utf-8"))
response.raise_for_status()
return response.json()
def collection_info(
self,
*,
resource_id: str = "",
name: str = "",
project: str = "",
):
payload = {}
if resource_id:
payload["resource_id"] = resource_id
if name:
payload["name"] = name
if project:
payload["project"] = project
return self._post(COLLECTION_INFO_PATH, payload)
def search_knowledge(
self,
*,
query: str,
name: str = "",
project: str = "",
resource_id: str = "",
limit: int = 10,
image_query: str = "",
query_param=None,
dense_weight: float = 0.5,
pre_processing=None,
post_processing=None,
pipeline_name: str = "",
):
payload = {
"query": query,
"limit": limit,
"dense_weight": dense_weight,
}
if name:
payload["name"] = name
if project:
payload["project"] = project
if resource_id:
payload["resource_id"] = resource_id
if image_query:
payload["image_query"] = image_query
if query_param:
payload["query_param"] = query_param
if pre_processing:
payload["pre_processing"] = pre_processing
if post_processing:
payload["post_processing"] = post_processing
if pipeline_name:
payload["pipeline_name"] = pipeline_name
return self._post(SEARCH_KNOWLEDGE_PATH, payload)
def _extract_top_chunks(search_resp: dict, top_k: int):
result_list = (
(search_resp or {}).get("data", {}) or {}
).get("result_list") or []
chunks = []
for item in result_list[:top_k]:
doc_info = item.get("doc_info") or {}
chunks.append(
{
"score": item.get("score"),
"rerank_score": item.get("rerank_score"),
"content": item.get("content"),
"chunk_title": item.get("chunk_title"),
"chunk_id": item.get("chunk_id"),
"doc_id": doc_info.get("doc_id"),
"doc_name": doc_info.get("doc_name"),
"doc_type": doc_info.get("doc_type"),
"url": doc_info.get("url"),
}
)
return chunks
def _load_allowed_collection_ids(args):
raw = args.allowed_collection_ids or os.getenv(ALLOWED_COLLECTION_IDS_ENV, "")
return _parse_csv_list(raw)
def _require_allowed_collection_ids(allowed_ids: list):
if not allowed_ids:
raise SystemExit(PERMISSION_DENIED_MSG)
def _assert_resource_id_allowed(resource_id: str, allowed_set: set):
if not resource_id or resource_id not in allowed_set:
raise SystemExit(PERMISSION_DENIED_MSG)
def _extract_resource_id_from_response(resp: dict):
data = (resp or {}).get("data") or {}
return (data.get("resource_id") or "").strip()
def _assert_response_collection_allowed(resp: dict, allowed_set: set):
resource_id = _extract_resource_id_from_response(resp)
if resource_id and resource_id not in allowed_set:
raise SystemExit(PERMISSION_DENIED_MSG)
def _find_allowed_collection_by_name(
client: ApigKnowledgeClient,
allowed_ids: list,
*,
name: str,
project: str,
max_workers: int,
):
"""仅在 DATABASE_VIKING_COLLECTION 范围内按 name+project 解析,避免对外部库发起查询。"""
target_name = _normalize_text(name)
target_project = (project or "default").strip()
for item in _fetch_collections_info(client, allowed_ids, max_workers=max_workers):
if item.get("error"):
continue
if _normalize_text(item.get("collection_name")) != target_name:
continue
item_project = (item.get("project") or "default").strip()
if item_project != target_project:
continue
return item
raise SystemExit(PERMISSION_DENIED_MSG)
def _resolve_authorized_target(
*,
client: ApigKnowledgeClient,
resource_id: str,
name: str,
project: str,
allowed_ids: list,
allowed_set: set,
max_workers: int,
):
_require_allowed_collection_ids(allowed_ids)
if resource_id:
_assert_resource_id_allowed(resource_id, allowed_set)
return resource_id, None
if name:
matched = _find_allowed_collection_by_name(
client,
allowed_ids,
name=name,
project=project,
max_workers=max_workers,
)
resolved_id = (matched.get("resource_id") or "").strip()
_assert_resource_id_allowed(resolved_id, allowed_set)
return resolved_id, matched.get("raw")
raise SystemExit(PERMISSION_DENIED_MSG)
def _fetch_collections_info(client: ApigKnowledgeClient, resource_ids: list, *, max_workers: int = 8):
results = []
def fetch(rid: str):
resp = client.collection_info(resource_id=rid)
if (resp or {}).get("code") != 0:
raise RuntimeError(f"collection_info failed: {resp}")
data = resp.get("data") or {}
return {
"resource_id": data.get("resource_id") or rid,
"collection_name": data.get("collection_name") or "",
"description": data.get("description") or "",
"project": data.get("project") or "",
"raw": resp,
}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {executor.submit(fetch, rid): rid for rid in resource_ids}
for future in as_completed(future_map):
rid = future_map[future]
try:
results.append(future.result())
except Exception as exc:
results.append(
{
"resource_id": rid,
"collection_name": "",
"description": "",
"project": "",
"error": str(exc),
}
)
results.sort(key=lambda x: x.get("collection_name") or x.get("resource_id"))
return results
def _search_one_collection(client: ApigKnowledgeClient, *, resource_id: str, query: str, limit: int):
resp = client.search_knowledge(query=query, resource_id=resource_id, limit=limit)
if (resp or {}).get("code") != 0:
raise RuntimeError(f"search_knowledge failed: {resp}")
return resp
def _run_multi_search(client: ApigKnowledgeClient, *, collections: list, query: str, limit: int, top_k: int, max_workers: int):
results = []
def search(item: dict):
rid = item.get("resource_id")
resp = _search_one_collection(client, resource_id=rid, query=query, limit=limit)
return {
**item,
"search": resp,
"top_chunks": _extract_top_chunks(resp, top_k),
}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {executor.submit(search, item): item for item in collections if item.get("resource_id")}
for future in as_completed(future_map):
item = future_map[future]
try:
results.append(future.result())
except Exception as exc:
results.append({**item, "error": str(exc)})
results.sort(key=lambda x: x.get("collection_name") or x.get("resource_id"))
return results
def main():
parser = argparse.ArgumentParser(description="Viking 知识库检索(代理网关)")
parser.add_argument(
"--action",
choices=["search", "info", "auto"],
default="search",
help="search: 单库检索;info: 查看知识库详情;auto: 自动路由/多库并行检索",
)
parser.add_argument("--query", default="", help="搜索查询文本(search/auto 必填)")
parser.add_argument("--name", default="", help="知识库名称(可选)")
parser.add_argument("--project", default=os.getenv("DATABASE_VIKING_PROJECT"), help="项目名称(默认读取 DATABASE_VIKING_PROJECT,使用 --name 检索时必填)")
parser.add_argument("--resource-id", default="", help="知识库 resource_id(推荐使用,auto模式下作为明确路由的标志)")
parser.add_argument("--limit", type=int, default=10, help="search_knowledge 返回结果数量")
parser.add_argument("--top-k", type=int, default=3, help="summary 提取的 top chunk 数")
parser.add_argument(
"--allowed-collection-ids",
default="",
help=f"有权限的知识库 resource_id 列表(逗号分隔)。默认读取 env {ALLOWED_COLLECTION_IDS_ENV}",
)
parser.add_argument(
"--max-workers",
type=int,
default=8,
help="并发 worker 数(auto 多库检索/拉取 info 时使用)",
)
parser.add_argument(
"--apig-url",
default=os.getenv("DATABASE_VIKING_APIG_URL"),
help="代理网关 URL(默认读取环境变量 DATABASE_VIKING_APIG_URL)",
)
parser.add_argument(
"--apig-key",
default=os.getenv("DATABASE_VIKING_APIG_KEY"),
help="代理网关 API Key(默认读取环境变量 DATABASE_VIKING_APIG_KEY)",
)
args = parser.parse_args()
if not args.apig_url or not args.apig_key:
raise SystemExit(
"检测到环境配置缺失:请联系管理员在执行环境中补充 DATABASE_VIKING_APIG_URL 和 DATABASE_VIKING_APIG_KEY 等环境变量。"
)
client = ApigKnowledgeClient(apig_url=args.apig_url, apig_key=args.apig_key)
allowed_ids = _load_allowed_collection_ids(args)
allowed_set = set(allowed_ids)
if args.action == "info":
if not args.resource_id and not args.name:
raise SystemExit("info action requires --resource-id or --name")
if not args.resource_id and args.name and not args.project:
raise SystemExit("when using --name, --project (or env DATABASE_VIKING_PROJECT) is required")
target_rid, cached_resp = _resolve_authorized_target(
client=client,
resource_id=args.resource_id,
name=args.name,
project=args.project,
allowed_ids=allowed_ids,
allowed_set=allowed_set,
max_workers=args.max_workers,
)
resp = cached_resp or client.collection_info(resource_id=target_rid)
_assert_response_collection_allowed(resp, allowed_set)
print(json.dumps(resp, ensure_ascii=False, indent=2))
return
if args.action == "search":
if not args.query:
raise SystemExit("search action requires --query")
if not args.resource_id and not args.name:
raise SystemExit("search action requires --resource-id or --name")
if not args.resource_id and args.name and not args.project:
raise SystemExit("when using --name, --project (or env DATABASE_VIKING_PROJECT) is required")
target_rid, _ = _resolve_authorized_target(
client=client,
resource_id=args.resource_id,
name=args.name,
project=args.project,
allowed_ids=allowed_ids,
allowed_set=allowed_set,
max_workers=args.max_workers,
)
resp = client.search_knowledge(
query=args.query,
resource_id=target_rid,
limit=args.limit,
)
_assert_response_collection_allowed(resp, allowed_set)
print(json.dumps(resp, ensure_ascii=False, indent=2))
return
# auto
if not args.query:
raise SystemExit("auto action requires --query")
if not allowed_ids:
raise SystemExit(
f"auto action requires allowed collection ids: set env {ALLOWED_COLLECTION_IDS_ENV} "
"or pass --allowed-collection-ids"
)
collections = [{"resource_id": rid} for rid in allowed_ids]
searched = _run_multi_search(
client,
collections=collections,
query=args.query,
limit=min(args.limit, 5),
top_k=args.top_k,
max_workers=args.max_workers,
)
output = {
"mode": "multi",
"query": args.query,
"collections": searched,
}
print(json.dumps(output, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()Related skills
FAQ
What is the default retrieval action?
For knowledge-QA tasks the default is auto, multi-base parallel retrieval, with the query split and keyword-ized.
Should I pass the full task as one query?
No, the skill forbids passing multi-intent task text as one query and requires splitting into 2-4 independent keyword queries run in parallel.