使用 OpenViking 存储

openviking 长期记忆后端把会话消息写入 OpenViking 的用户记忆，并按用户身份检索实体、事件和偏好。该后端从 VeADK 1.0.3 开始提供。

Runner.user_id 是终端用户标识，VeADK 将其作为 OpenViking 的 peer_id 用于记忆隔离。openviking_user_id 是 OpenViking 中记忆所属的 owner/context，用于在同一 OpenViking 服务内隔离不同应用或租户的记忆。两者是不同的概念。

使用示例

基础安装已包含 openviking-sdk。运行前准备支持用户记忆的 OpenViking 服务，设置 DATABASE_OPENVIKING_URL、DATABASE_OPENVIKING_API_KEY 和 DATABASE_OPENVIKING_USER_ID。自动或手动保存均会把会话内容发送到该服务；先确认访问权限和数据保留策略

import os

from veadk import Agent
from veadk.memory.long_term_memory import LongTermMemory

memory = LongTermMemory(
    backend="openviking",
    app_name="support_app",
    backend_config={
        "index": "support_app",
        "url": "https://openviking.example.com",
        "api_key": os.environ["DATABASE_OPENVIKING_API_KEY"],
        # 可选；不传时默认 default
        "openviking_user_id": "support_app",
        # 可选；显式指定记忆策略，不传时由 OpenViking 服务应用其官方默认策略
        "memory_policy": {
            "self": {"enabled": False},
            "peer": {"enabled": True},
            "memory_types": ["entities", "events", "preferences"],
        },
    },
)

agent = Agent(
    long_term_memory=memory,
    auto_save_session=True,
)

参数

`LongTermMemory` 参数

参数	类型	默认值	说明
`backend`	`str`	`opensearch`	使用 OpenViking 时设为 `openviking`
`backend_config`	`dict`	`{}`	OpenViking 后端的显式配置
`top_k`	`int`	`5`	每次检索返回的记忆数量
`index`	`str`	`""`	应用名称的回退值；检索隔离依赖 openviking_user_id 与 peer_id，不能只靠 index
`app_name`	`str`	`""`	应用名称，也是 `index` 的回退值
`user_id`	`str`	`""`	已废弃，仅为兼容旧代码保留。运行时用户以会话或 `Runner` 的 `user_id` 为准

OpenViking 后端参数

参数	类型	默认值	说明
`index`	`str`	由 `LongTermMemory` 传入	应用或记忆索引名称，不能为空
`openviking_config`	`OpenVikingConfig`	从 `DATABASE_OPENVIKING_*` 读取	OpenViking 服务配置
`url`	`str`	`DATABASE_OPENVIKING_URL`	OpenViking 服务地址，必填
`api_key`	`str`	`DATABASE_OPENVIKING_API_KEY`	OpenViking API Key，必填
`openviking_user_id`	`str`	`DATABASE_OPENVIKING_USER_ID` 或 `default`	OpenViking owner/context 标识，用于构建记忆路径。只允许字母、数字、`.`、`_`、`@`、`-`，不能为 `.` 或 `..`
`memory_policy`	`dict \| None`	`OpenVikingConfig.memory_policy` / `None`	OpenViking 记忆提取策略。未配置时不向 OpenViking 发送策略，由服务端应用其官方默认策略
`peer_id_resolver`	`Callable \| None`	使用 `user_id`	把 `app_name` 与 `user_id` 转换为 OpenViking peer ID
`timeout`	`float`	`30`	OpenViking 请求超时秒数

默认使用 user_id 作为 peer ID，只允许字母、数字、点、下划线、@ 和连字符。需要改变用户映射时，可传入 peer_id_resolver。 openviking_user_id 未配置时按 DATABASE_OPENVIKING_USER_ID 顺序查找，都未配置时使用 default 并输出告警日志。建议为每个应用或租户显式设置，避免不同应用的记忆混在同一个 default 路径下。未配置 memory_policy 时，VeADK 不向 OpenViking 发送策略，由 OpenViking 服务应用其官方默认策略。如需显式控制记忆的抽取范围与隔离方式，传入合法的 memory_policy，其结构以 OpenViking 会话接口为准。例如：

{"self": {"enabled": false}, "peer": {"enabled": true}, "memory_types": ["entities", "events", "preferences"]}

该示例关闭自身记忆提取，仅从对话对象提取实体、事件与偏好；实际取值与可用字段以 OpenViking 服务端支持的策略为准。

auto_save_session=True 会把会话内容写入外部 OpenViking 服务。使用前应确认数据处理、访问控制和保留策略符合业务要求。

环境变量

环境变量	默认值	说明
`DATABASE_OPENVIKING_URL`	无	OpenViking 服务地址
`DATABASE_OPENVIKING_API_KEY`	无	OpenViking service owner API Key
`DATABASE_OPENVIKING_USER_ID`	`default`	OpenViking owner/context 标识
`DATABASE_OPENVIKING_MEMORY_POLICY`	不发送策略	长期记忆 `memory_policy` 的 JSON 字符串；未配置时不向 OpenViking 发送策略，由服务端应用官方默认策略

自定义用户映射

多租户应用可以提供 peer_id_resolver，把应用和用户组合成单个隔离标识：

import os

from veadk.memory.long_term_memory import LongTermMemory


def resolve_peer_id(app_name: str, user_id: str) -> str:
    return f"{app_name}-{user_id}"


memory = LongTermMemory(
    backend="openviking",
    backend_config={
        "index": "support_app",
        "url": "https://openviking.example.com",
        "api_key": os.environ["DATABASE_OPENVIKING_API_KEY"],
        "peer_id_resolver": resolve_peer_id,
    },
)

返回值不能为空，只能包含字母、数字、点、下划线、@ 和连字符，且不能为 . 或 ..。上线后修改映射规则会使既有记忆保留在旧 peer 路径下，新的检索可能无法命中这些数据。

验证写入和检索

配置本页依赖与凭证后运行此独立示例。它直接保存用户文本，再检索同一用户的记忆，不需要调用对话模型

import asyncio

from google.adk.events import Event
from google.adk.sessions import Session
from google.genai import types
from veadk.memory.long_term_memory import LongTermMemory

async def main():
    memory = LongTermMemory(backend="openviking", index="ltm_demo")
    session = Session(
        id="memory_check", app_name="ltm_demo", user_id="user_42",
        events=[Event(author="user", content=types.Content(
            role="user", parts=[types.Part(text="My preferred language is Chinese")]
        ))],
    )
    await memory.add_session_to_memory(session)
    result = await memory.search_memory(
        app_name="ltm_demo", user_id="user_42", query="preferred language"
    )
    for entry in result.memories:
        print(entry.content)

asyncio.run(main())

结果应包含所保存的语言偏好。托管服务可能异步完成记忆提取，写入返回不保证立刻可检索；空结果也可能来自权限、网络或服务失败，应结合错误日志与服务端记录判断。此方法不返回保存成功的布尔值

概览

快速开始

核心组件

可观测

扩展

前端

安全

部署

使用示例

参数

`LongTermMemory` 参数

OpenViking 后端参数

环境变量

自定义用户映射

验证写入和检索

​使用示例

​参数

​LongTermMemory 参数

​OpenViking 后端参数

​环境变量

​自定义用户映射

​验证写入和检索

使用示例

参数

`LongTermMemory` 参数

OpenViking 后端参数

环境变量

自定义用户映射

验证写入和检索