文档中心

向量数据库VikingDB

用户管理（Users）

RegisterOpenVikingUser -注册库用户（企业版支持）

概述

RegisterOpenVikingUser 接口用于在指定库下注册新用户（仅企业版支持）。

前置条件

完成 "数据库管理使用说明" 页面的 AK/SK 获取后，可调用本接口。当前库版本需为企业版（enterprise），调用者需具备管理员权限。

请求接口

Action	RegisterOpenVikingUser
请求方法	POST
Content-Type	application/json

请求参数

参数	类型	是否必填	说明
`ResourceID`	String	是	OpenViking 库 ID。
`AccountID`	String	否	User 所属的数据空间 ID，缺省为 default（默认数据空间）。
`UserID`	String	是	User ID，需满足 [a-zA-Z0-9_-]{1,64}。
`MemoryPolicy`	Object	否	User 记忆策略。不传时使用默认全量策略。
`MemoryPolicy`.`MemoryTypes`	Array of String	`MemoryPolicy` 传入时必填	允许抽取的记忆类型。支持传空数组。

MemoryPolicy

User 记忆策略使用以下结构：

{
  "MemoryTypes": [
    "profile",
    "preferences",
    "events",
    "entities",
    "cases",
    "trajectories",
    "experiences",
    "soul",
    "identity"
  ]
}

当前支持的记忆类型如下：

取值	说明
profile	用户画像与长期稳定信息。
preferences	用户偏好、习惯和倾向。
events	用户相关事件、决定和阶段性信息。
entities	用户提及的重要人物、组织、地点、产品等实体。
cases	Agent 任务案例。
trajectories	Agent 执行轨迹。
experiences	从任务执行中沉淀的 Agent 经验。
soul	用户表达的对 agent 人格、行为准则的预期。
identity	用户表达的对 agent 身份和个性的预期。

注意事项：

MemoryTypes 按集合处理，返回顺序不表示优先级。
重复的记忆类型会被去重。
MemoryTypes: [] 表示该 User 不启用任何可配置的长期记忆类型；该设置不影响 Session 消息保存。
未传 MemoryPolicy 创建 User 时，默认启用当前全部 7 种记忆类型。
User 记忆策略变更只影响后续 Session Commit，不会重新处理已经生成的记忆文件。
cases、trajectories 和 experiences 作为一组 Agent 自进化记忆生效。配置 experiences 时，系统会同时启用 cases 和 trajectories；未配置 experiences 时，这三种类型均不生效，不能单独开启其中一种。
Agent 自进化记忆除了需要在 User 策略中启用，还要求对应数据空间已开启 Agent 自进化。

响应消息

参数	类型	说明
`Success`	Boolean	是否创建成功。
`UserID`	String	已创建的 User ID。
`MemoryPolicy`	Object	创建后实际生效的 User 记忆策略。
`MemoryPolicy.MemoryTypes`	Array of String	实际启用的记忆类型。

常见错误码

HTTP 状态码	错误码	说明
400	InvalidParameter	请求参数非法、User 已存在，或者不支持显式设置 User MemoryPolicy。
403	AccessDenied	当前账号无权操作该 OpenViking 库。
404	ResourceNotExist	指定的库或数据空间不存在。
409	ResourceExist	指定资源已经存在。
429	QuotaExceeded	User 数量或创建频率超过配额。
500	InternalError	服务内部错误。

完整示例

Python 请求示例

import os
import datetime
import hashlib
import hmac
from urllib.parse import quote

import requests, json

def norm_query(params):
    query = ""
    for key in sorted(params.keys()):
        if type(params[key]) == list:
            for k in params[key]:
                query = (
                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
                )
        else:
            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
    query = query[:-1]
    return query.replace("+", "%20")

def hmac_sha256(key: bytes, content: str):
    return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()

def hash_sha256(content: str):
    return hashlib.sha256(content.encode("utf-8")).hexdigest()

class ClientForConsoleApi:
    def __init__(self, ak, sk, host, region):
        self.ak = ak
        self.sk = sk
        self.host = host
        self.region = region
        self.service_code = "vikingdb"
        self.version = "2025-06-09"

    def request(self, method, action, body, query=None, header=None):
        if query is None:
            query = {}
        if header is None:
            header = {}
        credential = {
            "access_key_id": self.ak,
            "secret_access_key": self.sk,
            "service": self.service_code,
            "region": self.region,
        }
        request_param = {
            "body": json.dumps(body),
            "host": self.host,
            "path": "/",
            "method": method,
            "content_type": 'application/json',
            "date": datetime.datetime.utcnow(),
            "query": {"Action": action, "Version": self.version, **query},
        }
        if body is None:
            request_param["body"] = ""
        x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
        short_x_date = x_date[:8]
        x_content_sha256 = hash_sha256(request_param["body"])
        sign_result = {
            "Host": request_param["host"],
            "X-Content-Sha256": x_content_sha256,
            "X-Date": x_date,
            "Content-Type": request_param["content_type"],
        }
        signed_headers_str = ";".join(
            ["content-type", "host", "x-content-sha256", "x-date"]
        )
        canonical_request_str = "\n".join(
            [request_param["method"].upper(),
             request_param["path"],
             norm_query(request_param["query"]),
             "\n".join(
                 [
                     "content-type:" + request_param["content_type"],
                     "host:" + request_param["host"],
                     "x-content-sha256:" + x_content_sha256,
                     "x-date:" + x_date,
                 ]
             ),
             "",
             signed_headers_str,
             x_content_sha256,
             ]
        )

        hashed_canonical_request = hash_sha256(canonical_request_str)
        credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
        string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])

        k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
        k_region = hmac_sha256(k_date, credential["region"])
        k_service = hmac_sha256(k_region, credential["service"])
        k_signing = hmac_sha256(k_service, "request")
        signature = hmac_sha256(k_signing, string_to_sign).hex()

        sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
            credential["access_key_id"] + "/" + credential_scope,
            signed_headers_str,
            signature,
        )
        header = {**header, **sign_result}
        r = requests.request(method=method,
                             url="https://{}{}".format(request_param["host"], request_param["path"]),
                             headers=header,
                             params=request_param["query"],
                             data=request_param["body"],
                             )
        return r.status_code, r.headers, r.json()

if __name__ == '__main__':
    client = ClientForConsoleApi(
        ak = "*",#替换为您的 AK
        sk ="*" ,#替换为您的 SK
        host="vikingdb.cn-beijing.volcengineapi.com", #替换为您所在地区的域名
        region="cn-beijing"
    )
    http_code, _, result_json = client.request(
        method="POST",
        action="RegisterOpenVikingUser",#action替换为当前操作
        body={
          "ResourceID": "ov-xxxxxxxxxxxxxxxx",
          "AccountID": "default",
          "UserID": "user_001",
          "MemoryPolicy": {
            "MemoryTypes": [
              "profile",
              "preferences",
              "events",
              "entities",
              "cases",
              "trajectories",
              "experiences",
              "soul",
              "identity"
            ]
          }
        }
    )
    print("req http status code: ", http_code)
    print("req result: \n", result_json)

响应示例

{
  "ResponseMetadata": {
    "RequestId": "20260904120000XXXXXXXXXXXXXXXX",
    "Action": "RegisterOpenVikingUser",
    "Version": "2025-06-09",
    "Service": "vikingdb",
    "Region": "cn-beijing"
  },
  "Result": {
    "Success": true,
    "UserID": "user_001",
    "MemoryPolicy": {
      "MemoryTypes": [
        "cases",
        "entities",
        "events",
        "experiences",
        "preferences",
        "profile",
        "trajectories"
      ]
    }
  }
}

最近更新时间：2026.09.11 16:21:09

这个页面对您有帮助吗？

有用

无用

向量数据库VikingDB

MemoryPolicy #

Python 请求示例 #

响应示例 #

MemoryPolicy

Python 请求示例

响应示例