SKILL：Bohrium 知识库管理

概述

在 Bohrium 平台上管理知识库（Literature Sage）。知识库支持文献管理、文件夹组织、标签分类、笔记记录、文献内容搜索与召回，以及权限管理。

无 CLI 支持 —— 与 bohrium-job/node/image 不同，知识库不提供 bohr CLI 命令。所有操作均通过 HTTP API 完成。

认证

ACCESS_KEY 从 OpenClaw 配置文件 ~/.openclaw/openclaw.json 中读取：

"bohrium-knowledge-base": {
  "enabled": true,
  "apiKey": "YOUR_ACCESS_KEY",
  "env": {
    "ACCESS_KEY": "YOUR_ACCESS_KEY"
  }
}

OpenClaw 会自动将 env.ACCESS_KEY 注入运行时环境。

路由映射

外部调用：GET/POST https://open.bohrium.com/openapi/v1/knowledge/{path}
               请求头：accessKey: YOUR_ACCESS_KEY

网关转发：→ literature-sage.bohrium.com/api/v1/{path}
                  请求头：X-User-Id, X-Org-Id（由 accessKey 转换而来）

通用代码模板

import os, requests

AK = os.environ.get("ACCESS_KEY", "")
BASE = "https://open.bohrium.com/openapi/v1/knowledge"
HEADERS = {"accessKey": AK}
HEADERS_JSON = {**HEADERS, "Content-Type": "application/json"}

知识库管理

创建知识库

r = requests.post(f"{BASE}/knowledge_base/create", headers=HEADERS_JSON, json={
    "knowledgeBaseName": "我的知识库",
    "cover": "",
    "introduction": "关于分子动力学的论文合集",
    "privilege": 1  # 1=私有，2=公开
})
print(r.json())
# {"code": 0, "data": {"id": 123, "msg": "知识库创建成功。"}}

参数说明：

列出我的知识库

r = requests.get(f"{BASE}/knowledge_base/list", headers=HEADERS,
    params={"keyword": "", "pageSize": 10, "pageNum": 1})
data = r.json()["data"]
for kb in data["list"]:
    print(f"[{kb['id']}] {kb['name']} (nodeId={kb['nodeId']}, privilege={kb['privilege']})")

查询参数：

响应字段：

更新知识库

r = requests.post(f"{BASE}/knowledge_base/update", headers=HEADERS_JSON, json={
    "knowledgeBaseName": "更新后的名称",
    "cover": "",
    "introduction": "更新后的介绍",
    "NodesId": 456,       # 知识库节点 ID
    "privilege": 2         # 变更为公开
})
# {"code": 0, "data": {"id": 123, "msg": "知识库更新成功。"}}

获取知识库详情

node_id = 456
r = requests.get(f"{BASE}/knowledge_base/{node_id}", headers=HEADERS)
print(r.json())

发现公开知识库

r = requests.get(f"{BASE}/knowledge_base/discover", headers=HEADERS,
    params={"pageSize": 10, "pageNum": 1})

推荐知识库

r = requests.get(f"{BASE}/knowledge_base/recommendation", headers=HEADERS)
# 返回数据结构：{list, total, config_reason}

收藏 / 取消收藏

requests.post(f"{BASE}/knowledge_base/favorite",   headers=HEADERS_JSON, json={"nodesId": 11743137})
requests.post(f"{BASE}/knowledge_base/unfavorite", headers=HEADERS_JSON, json={"nodesId": 11743137})

# 查询我的收藏知识库
r = requests.get(f"{BASE}/knowledge_base/favorite", headers=HEADERS,
    params={"pageNum": 1, "pageSize": 10})

浏览历史 / 搜索历史

# 记录一次浏览行为
requests.post(f"{BASE}/knowledge_base/browse/add", headers=HEADERS_JSON, json={
    "fileId": 12345,
    "fileType": 1,
    "parentId": 11743137,
    "rootFolderId": 11743137,
    "nodesId": 11743137
})

# 查询浏览或搜索历史
requests.post(f"{BASE}/knowledge_base/browse/query",  headers=HEADERS_JSON, json={"pageNum":1,"pageSize":10})
requests.post(f"{BASE}/knowledge_base/history/query", headers=HEADERS_JSON, json={"pageNum":1,"pageSize":10})

搜索知识库

r = requests.post(f"{BASE}/knowledge_base/search/name", headers=HEADERS_JSON, json={
    "nodesId": 456,              # 知识库节点 ID；0 表示在可访问的所有知识库中搜索
    "searchText": "分子动力学",
    "searchType": 0,             # 0=全部，1=仅文件夹，2=仅文件
    "pageNum": 1,
    "pageSize": 10
})
data = r.json()["data"]
print(f"总计：{data['total']}")
for f in data["folders"]: print("[DIR]", f["name"])
for f in data["files"]:   print("[FILE]", f["fileName"])

删除知识库

没有独立的 knowledge_base/delete 接口。删除知识库 = 删除其根节点 nodesId（知识库本质上是一个根文件夹）：

r = requests.post(f"{BASE}/folder/delete", headers=HEADERS_JSON, json={
    "nodesId": 11743137   # 知识库的 nodesId
})
# 仅拥有者角色可执行删除操作。

文件夹管理

列出根目录（知识库）

r = requests.get(f"{BASE}/folder/root", headers=HEADERS)
data = r.json()["data"]
print(f"总数: {data['total']} 个知识库，{data['docCount']} 个文档")
for item in data["list"]:
    print(f"  [{item['id']}] {item['name']} ({item['docCount']} 个文档，"
          f"关系={'所有者' if item['relationship'] == 1 else '成员'})")

响应字段：

列出子级（一级）

r = requests.get(f"{BASE}/folder/children", headers=HEADERS,
    params={"folderId": 456, "pageNum": 1, "pageSize": 20})
data = r.json()["data"]

# 导航路径
for p in data["path"]:
    print(f"  {'>' * p['depth']} {p['name']} (id={p['nodesId']})")

# 子文件夹
for f in data["folders"]:
    print(f"  [DIR] {f['name']} ({f['docCount']} 个文档，id={f['id']})")

# 文件
for f in data["files"]:
    print(f"  [FILE] {f['name']} | {f['fileName']} | {f['date']}")

响应字段：

目录树

r = requests.get(f"{BASE}/folder/directory", headers=HEADERS,
    params={"folderId": 456})
# 递归树结构：{result: [{nodesId, name, subFolders: [...]}]}

# 完整文件树（包含文献）：
r = requests.get(f"{BASE}/folder/file_tree", headers=HEADERS,
    params={"folderId": 456})

创建文件夹

r = requests.post(f"{BASE}/folder/create", headers=HEADERS_JSON, json={
    "parentId": 456,             # 父文件夹 ID
    "folderName": "我的文件夹"
})
# {"code": 0, "data": {"message": "..."}}

重命名文件夹

r = requests.post(f"{BASE}/folder/update", headers=HEADERS_JSON, json={
    "nodesId": 789,              # 文件夹节点 ID
    "folderName": "新名称"
})

移动文件夹

r = requests.post(f"{BASE}/folder/move", headers=HEADERS_JSON, json={
    "sourceFolderId": 789,       # 源文件夹 ID
    "targetFolderId": 456        # 目标文件夹 ID
})

删除文件夹

r = requests.post(f"{BASE}/folder/delete", headers=HEADERS_JSON, json={
    "nodesId": 789
})

文献管理

列出文献

r = requests.get(f"{BASE}/file", headers=HEADERS,
    params={
        "parentId": 456,        # 文件夹 ID
        "pageNum": 1,
        "pageSize": 20,
        "order": 1,             # 1=升序，2=降序（整数，非字符串）
        "orderBy": 1,           # 1=创建时间，2=更新时间，等
        "noTag": False          # True = 仅未标记项目
    })

文献详情

r = requests.get(f"{BASE}/file/detail", headers=HEADERS,
    params={"resourceId": "12345"})
detail = r.json()["data"]
print(f"标题: {detail['enName']}")
print(f"作者: {', '.join(a['name'] for a in detail.get('authorDetails', []))}")
print(f"DOI: {detail.get('doi', '')}")
print(f"摘要: {detail.get('enAbstract', '')}")

响应字段（精选）：

下载文献链接

r = requests.post(f"{BASE}/file/read", headers=HEADERS_JSON, json={
    "userResourceId": [12345]    # 数组；参数名为 userResourceId
})
# 返回下载链接列表，每个资源对应一个链接。

上传文件

支持将本地文件（PDF、Markdown 等）上传至知识库。

上传流程（三步）

1) GET  /file/multipart   → 获取上传凭证（host、path、token）
2) POST {host}/api/upload/binary  → 上传文件二进制内容
3) POST /file/submit       → 注册文件至知识库（使其可见）

获取上传凭证

# 计算文件 MD5 和大小
import hashlib
def md5_hex(path):
    h = hashlib.md5()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()

# 获取上传凭证
file_path = "paper.pdf"
file_name = "paper.pdf"
file_size = os.path.getsize(file_path)
file_md5 = md5_hex(file_path)
parent_id = 456  # 知识库或文件夹的 nodeId

r = requests.get(f"{BASE}/file/multipart", headers=HEADERS,
    params={
        "fileName": file_name,
        "md5": file_md5,
        "parentId": parent_id,
        "size": file_size
    })

multipart_data = r.json()["data"]
if multipart_data.get("fileExist"):
    print("文件已存在，无需重新上传")
else:
    host = multipart_data["host"]
    path = multipart_data["path"]
    token = multipart_data["token"]
    print(f"成功获取上传凭证: {host}, {path}")

执行二进制上传

import base64
import json
import urllib.request

def make_storage_param(remote_path: str, encoded_file_name: str, content_type: str) -> str:
    payload = {
        "path": remote_path,
        "option": {
            "contentDisposition": (
                f'inline; filename="{encoded_file_name}"; '
                f"filename*=UTF-8''{encoded_file_name}"
            ),
            "contentType": content_type,
        },
    }
    b = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    return base64.b64encode(b).decode("utf-8")

# 上传文件到存储服务
file_content = open(file_path, "rb").read()
content_type = "application/pdf"  # 或其他合适的 MIME 类型
encoded_file_name = urllib.parse.quote(file_name, safe="-_.!~*'()")

storage_param = make_storage_param(path, encoded_file_name, content_type)
upload_url = host.rstrip("/") + "/api/upload/binary"

req = urllib.request.Request(upload_url, method="POST", data=file_content)
req.add_header("Authorization", f"Bearer {token}")
req.add_header("X-Storage-Param", storage_param)
req.add_header("Content-Type", "application/octet-stream")

with urllib.request.urlopen(req, timeout=300) as resp:
    upload_result = json.loads(resp.read().decode("utf-8"))

print("二进制文件上传完成")

将文件注册到知识库

# 将已上传的文件注册到知识库，使其在知识库中可见
final_path = (upload_result.get("data") or {}).get("path") or path

r = requests.post(f"{BASE}/file/submit", headers=HEADERS_JSON, json={
    "parentId": parent_id,
    "fileName": file_name,
    "md5": file_md5,
    "size": file_size,
    "url": final_path  # 上传完成后返回的路径
})

result = r.json()
if result.get("code") == 0:
    print("文件成功注册到知识库")
elif result.get("code") == 230117:  # 文件已存在
    print("文件已在知识库中存在")
else:
    print(f"注册失败: {result}")

完整上传函数示例

import hashlib
import base64
import json
import urllib.request
import mimetypes
import os

def guess_content_type(path):
    suffix = os.path.splitext(path)[1].lower()
    if suffix in {".md", ".markdown"}:
        return "text/markdown; charset=utf-8"
    if suffix in {".txt"}:
        return "text/plain; charset=utf-8"

    ctype, _ = mimetypes.guess_type(path)
    if ctype is None:
        return "application/octet-stream"

    if ctype.startswith("text/"):
        return f"{ctype}; charset=utf-8"
    return ctype

def upload_file_to_knowledge_base(file_path, parent_id, custom_file_name=None):
    # 准备文件信息
    file_name = custom_file_name or os.path.basename(file_path)
    file_size = os.path.getsize(file_path)

    # 计算 MD5 值
    def md5_hex(path):
        h = hashlib.md5()
        with open(path, "rb") as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b""):
                h.update(chunk)
        return h.hexdigest()

    file_md5 = md5_hex(file_path)

    # 步骤 1：获取上传凭证
    r = requests.get(f"{BASE}/file/multipart", headers=HEADERS,
        params={
            "fileName": file_name,
            "md5": file_md5,
            "parentId": parent_id,
            "size": file_size
        })

    multipart_data = r.json()["data"]

    if multipart_data.get("fileExist"):
        print("文件已存在，跳过上传但继续注册到知识库...")

        # 即使文件已存在，仍需注册到知识库
        r_submit = requests.post(f"{BASE}/file/submit", headers=HEADERS_JSON, json={
            "parentId": parent_id,
            "fileName": file_name,
            "md5": file_md5,
            "size": file_size,
            "url": multipart_data.get("path", "")  # 使用已有路径
        })
        return r_submit.json()

    # 获取上传凭证
    host = multipart_data["host"]
    path = multipart_data["path"]
    token = multipart_data["token"]

    # 准备上传参数
    content_type = guess_content_type(file_path)
    encoded_file_name = urllib.parse.quote(file_name, safe="-_.!~*'()")

    storage_param = base64.b64encode(json.dumps({
        "path": path,
        "option": {
            "contentDisposition": (
                f'inline; filename="{encoded_file_name}"; '
                f"filename*=UTF-8''{encoded_file_name}"
            ),
            "contentType": content_type,
        },
    }, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).decode("utf-8")

    # 步骤 2：二进制上传
    file_content = open(file_path, "rb").read()
    upload_url = host.rstrip("/") + "/api/upload/binary"

    req = urllib.request.Request(upload_url, method="POST", data=file_content)
    req.add_header("Authorization", f"Bearer {token}")
    req.add_header("X-Storage-Param", storage_param)
    req.add_header("Content-Type", "application/octet-stream")

    with urllib.request.urlopen(req, timeout=300) as resp:
        upload_result = json.loads(resp.read().decode("utf-8"))

    # 步骤 3：将文件注册到知识库
    final_path = (upload_result.get("data") or {}).get("path") or path

    r_submit = requests.post(f"{BASE}/file/submit", headers=HEADERS_JSON, json={
        "parentId": parent_id,
        "fileName": file_name,
        "md5": file_md5,
        "size": file_size,
        "url": final_path
    })

    return r_submit.json()

# 示例用法
result = upload_file_to_knowledge_base("./paper.pdf", 456)
print(result)

幂等性说明

当文件已存在（fileExist=true）时，系统会跳过二进制上传步骤，但仍调用 submit 接口以确保文件在知识库中可见。这使得多次上传同一文件成为安全操作。

脚本化上传

项目还提供了一个 scripts/bohrium-kb-upload.py 脚本，支持通过命令行进行文件上传：

Bohrium 知识库管理技能

上传单个文件

python3 scripts/bohrium-kb-upload.py ./paper.pdf --parent-id 456

批量上传目录中所有 PDF 文件

for f in pdfs/*.pdf; do
  python3 scripts/bohrium-kb-upload.py "$f" --parent-id 456
done

编辑文献元数据

r = requests.post(f"{BASE}/file/edit", headers=HEADERS_JSON, json={
    "id": 12345,
    "name": '{"cn":"中文标题","en":"英文标题"}',
    "doi": "10.1234/example",
    "authors": '["作者A","作者B"]',
    "date": "2024-01-15",
    "journal": "Nature",
    "abstract": '{"cn":"中文摘要","en":"英文摘要"}',
    "importance": 1,
    "recallable": True,
    "unableRecallEnMsg": "",
    "unableRecallZhMsg": ""
})

参数说明：

删除文献

r = requests.post(f"{BASE}/file/delete_literature", headers=HEADERS_JSON, json={
    "userResourceId": 12345    # 参数名为 userResourceId
})

重命名文献

r = requests.post(f"{BASE}/file/update_literature", headers=HEADERS_JSON, json={
    "userResourceId": 12345,   # 参数名为 userResourceId
    "fileName": "新文献名称"   # 参数名为 fileName
})

移动文献

r = requests.post(f"{BASE}/file/move", headers=HEADERS_JSON, json={
    "fileNodesIdList": [12345, 12346],   # 文件节点 ID 列表
    "folderNodesId": 789                 # 目标文件夹节点 ID
})

搜索文献内容

r = requests.post(f"{BASE}/file/search", headers=HEADERS_JSON, json={
    "queryContent": "分子动力学模拟",
    "nodesId": 456,
    "knowledgeBaseId": 123
})
data = r.json()["data"]
print(f"找到 {data['total']} 个结果")
for f in data["Files"]:
    print(f"  [{f['userResourceId']}] {f['fileName']}: {f['content'][:100]}...")

返回字段说明：

文件元数据 / 辅助接口

# 根据 userResourceId 获取元数据（md5 / paperId / publicationId 等）
requests.post(f"{BASE}/file/fileinfo", headers=HEADERS_JSON, json={
    "userResourceId": 12345,
    # 或使用 "enclosureId": 67890
})

# 获取某个文件夹下的所有文件 ID（无需分页）
requests.post(f"{BASE}/file/ids", headers=HEADERS_JSON, json={"parentId": 456})

# 获取单篇文献的标签信息
requests.get(f"{BASE}/file/tagInfo", headers=HEADERS, params={"resourceId": 12345})

# 上传记录 / 存储容量查询
requests.get(f"{BASE}/file/upload/record", headers=HEADERS, params={"pageNum":1,"pageSize":10})
requests.get(f"{BASE}/file/capacity",      headers=HEADERS)   # 返回 {remainingCapacity, totalCapacity, usedCapacity}

标签管理

列出标签

r = requests.get(f"{BASE}/tag", headers=HEADERS,
    params={"keyword": ""})  # 可选关键词过滤
data = r.json()["data"]
for tag in data["list"]:
    print(f"  [{tag['id']}] {tag['name']} ({tag['count']} 篇文档)")

创建标签

r = requests.post(f"{BASE}/tag", headers=HEADERS_JSON, json={
    "name": "机器学习"
})
tag = r.json()["data"]
print(f"创建标签: {tag['id']} - {tag['name']}")

编辑标签

r = requests.put(f"{BASE}/tag", headers=HEADERS_JSON, json={
    "tagId": 101,
    "name": "深度学习"
})

删除标签

r = requests.delete(f"{BASE}/tag", headers=HEADERS_JSON, json={
    "tagId": 101
})

为文献打标签

r = requests.post(f"{BASE}/file/tag", headers=HEADERS_JSON, json={
    "tagId": 101,
    "resourceId": 12345
})

取消文献标签

r = requests.post(f"{BASE}/file/untag", headers=HEADERS_JSON, json={
    "tagId": 101,
    "resourceId": 12345
})

文献标签统计

r = requests.get(f"{BASE}/file/tag", headers=HEADERS,
    params={
        "parentId": 456,       # 文件夹节点 ID（可选）
        "rootFolderId": 123,   # 知识库根文件夹 ID（可选）
        "query": 2,            # 1=按作者搜索，2=按关键词搜索（可选）
        "keyword": "ML"        # 搜索关键词（可选）
    })
data = r.json()["data"]
print(f"总文档数: {data['allDocCount']}, 未打标签: {data['noTagCount']}")
for tag in data["tags"]:
    print(f"  [{tag['id']}] {tag['name']}: {tag['count']} 篇文档")

笔记功能

获取笔记

r = requests.get(f"{BASE}/note", headers=HEADERS,
    params={"resourceId": 12345})
note = r.json()["data"]
print(f"笔记内容: {note['note']}")

创建或更新笔记

r = requests.post(f"{BASE}/note", headers=HEADERS_JSON, json={
    "resourceId": 12345,
    "note": "本文提出了一种新颖的方法..."
})
# 返回 {"code": 0}

文献召回与搜索

查看文献切片

查看特定论文的解析切片：

r = requests.post(f"{BASE}/box/search_by_md5_paper_id", headers=HEADERS_JSON, json={
    "md5": "abc123...",
    "paper_id": "paper_001",
    "page_num": 1,
    "page_size": 10
})

从指定论文中召回内容

技能：Bohrium 知识库管理

版本：1.0.0

分块：5/5

在指定论文中执行语义召回。**注意：上游字段名为 papers/text/k，而非 query/paperIds/topK。**

r = requests.post(f"{BASE}/recall/papers", headers=HEADERS_JSON, json={
    "papers": [
        {"paperId": "paper_001", "md5": ""},
        {"paperId": "", "md5": "abc123..."}
    ],
    "text": "分子动力学力场",
    "k": 5
})

混合召回（知识库级别）

在整套知识库中执行混合语义搜索。**注意：使用 snake_case 字段名（如 knowledge_base_id）；keywords 字段为必填项，且不能为空。**

r = requests.post(f"{BASE}/recall/hybrid", headers=HEADERS_JSON, json={
    "knowledge_base_id": 456,                # KB 节点 ID
    "text": "深度势能面",
    "k": 10,
    "keywords": {"深度势能": 1.0, "能面": 0.5}   # 必填，非空
})

权限管理

重要提示： 所有 /account/* 接口均使用 nodesId（即知识库的节点 ID），而非 knowledgeBaseId。

列出权限

r = requests.get(f"{BASE}/account/acl", headers=HEADERS,
    params={"nodesId": 11743137})
# 返回数据：{privilege, shareMode, userList: [{id, role, isCreator, userName, ...}]}

设置共享状态

r = requests.post(f"{BASE}/account/share_status", headers=HEADERS_JSON, json={
    "nodesId": 11743137,
    "privilege": 1,    # 1=公开，2=私有
    "shareMode": 1     # 1=不共享，2=链接共享
})

更新用户角色

r = requests.post(f"{BASE}/account/user_role", headers=HEADERS_JSON, json={
    "nodesId": 11743137,
    "userId": 456,
    "role": 67801    # 参见下方角色说明
})

移除用户权限

r = requests.delete(f"{BASE}/account/user_role", headers=HEADERS_JSON, json={
    "nodesId": 11743137,
    "userId": 456
})

批量添加读者

r = requests.post(f"{BASE}/account/batch_add_readers", headers=HEADERS_JSON, json={
    "nodesId": 11743137,
    "userList": [
        {"id": 456, "role": 67801},
        {"id": 789, "role": 67801}
    ]
    # 或使用 phones / emails / userNos
    # "phones": ["13800000000"]
})

申请加入知识库

r = requests.post(f"{BASE}/account/join_request", headers=HEADERS_JSON, json={
    "nodesId": 11743137
})

查询用户角色

r = requests.get(f"{BASE}/account/user_knowledge_base_role", headers=HEADERS,
    params={"nodesId": 11743137})
# 返回数据：{roles: [int, ...]}

其他权限接口

权限角色参考

知识库可见性（privilege）：

故障排查

网关限制

• 所有 /openapi/v1/knowledge/<path> 请求将被代理至 literature-sage 的 /api/v1/<path>。
• literature-sage 的 /api/v2/* 路由无法通过此网关访问。但本文档所有示例均使用 v1 路径，可正常运行。
• 无专用知识库删除接口；请使用 POST /folder/delete {nodesId: <KB_nodesID>}。