从指定记忆库中检索相关的事件记忆信息,可依据用户提问进行语义相似度检索,并根据查询条件划定查询范围。
collection.search_event_memory(query=None, filter=None, time_decay_config=None, limit=None, headers=None, timeout=None)
参数 | 类型 | 是否必须 | 描述 |
|---|---|---|---|
query | String | 否 | 用户的检索查询语句(最大长度 4000 字符)
|
filter | Object | 否 | 检索过滤条件和返回设置。 |
| String or Array of String | 否 | 用户 ID,支持单个 ID 或 ID 列表。 |
| String or Array of String | 否 | 助手 ID,支持单个 ID 或 ID 列表。 |
| Integer | 否 | 检索记忆的起始时间,毫秒级时间戳。 |
| Integer | 否 | 检索记忆的终止时间,毫秒级时间戳。 |
| String or Array of String | 是 | 要检索的事件类型名称或它们的列表。 |
| String or Array of String | 否 | 用于过滤的群组ID。可以是单个ID或ID列表。 |
| String or Array of String | 否 | 用于过滤的session ID。可以是单个ID或ID列表 |
limit | Integer | 否 | 返回的检索结果条数,默认为10,取值范围[1, 5000]。 |
time_decay_config | Object | 否 | 时间衰减配置。仅针对事件检索,对画像不生效。 |
| Float | 是 | 用于调节时间衰减影响力的权重,取值范围[0,1)。 |
| Integer | 否 | 无衰减期。即多长时间内的记忆权重相同,不衰减。配置单位为天,范围[0,10000]。默认值为 0。 |
custom_weight | Float | 否 | 用于调节业务字段的重要程度,取值范围[0,1)。 |
说明:
字段 | 类型 | 描述 |
|---|---|---|
code | Integer | 状态码,0 表示成功,其他表示错误。 |
message | String | 返回信息。 |
data | Object | 返回的详细检索结果。 |
| String | 被检索的记忆库名称。 |
| Integer | 返回的检索结果数量。 |
| Array of Object | 检索结果列表。 |
| String | 事件实例的唯一ID。 |
| Float | 与查询的相关性得分。 |
| Float | 原始的向量检索分数。 |
| Float | 时间衰减分数。 |
| Float | 自定义权重分数 |
| Array of String | 记忆类型名称。 |
| Array of String | 关联的用户ID列表。 |
| Array of String | 关联的助手ID列表。 |
| String | 关联的会话ID。 |
| String | 关联的群组ID。 |
| Integer | 记忆发生或最后更新的时间戳(毫秒)。 |
| String | 记忆条目的状态(预留字段)。 |
| String | 记忆条目的标签(预留字段)。 |
| Object | 额外的详细信息。其结构取决于memory_type。
|
| Integer | 本次检索消耗的token数量。 |
request_id | String | 标识每个请求的唯一ID。 |
import json import os import time from vikingdb import APIKey from vikingdb.memory import VikingMem def build_client() -> VikingMem: api_key = os.getenv("MEMORY_API_KEY") if not api_key: raise RuntimeError("Missing credentials: set MEMORY_API_KEY") auth = APIKey(api_key=api_key) return VikingMem( host="api-knowledgebase.mlp.cn-beijing.volces.com", region="cn-beijing", auth=auth, scheme="http", ) def main() -> None: client = build_client() collection = client.get_collection(collection_name="your_collection", project_name="default") now_ms = int(time.time() * 1000) user_id = f"sdk_example_user_{now_ms}" assistant_id = "sdk_example_assistant" try: collection.add_event( event_type="event_v1", user_id=user_id, assistant_id=assistant_id, memory_info={"summary": "用户说周末想去一家新的咖啡店,想试试手冲咖啡。"}, ) collection.add_event( event_type="event_v1", user_id=user_id, assistant_id=assistant_id, memory_info={"summary": "用户说最近在学习意式浓缩,并对萃取时间很感兴趣。"}, ) time.sleep(5) result = collection.search_event_memory( query="咖啡店 手冲", filter={ "user_id": user_id, "assistant_id": assistant_id, "memory_type": ["event_v1"], }, time_decay_config={"weight": 0.2, "no_decay_period": 7}, limit=10, ) print("search_event_memory response:") print(json.dumps(result, indent=2, ensure_ascii=False)) print("user_id:", user_id) print("assistant_id:", assistant_id) finally: collection.batch_delete_event( filter={"user_id": user_id, "event_type": "event_v1"}, delete_type="full", ) if __name__ == "__main__": main()