35 lines
1.0 KiB
Python
35 lines
1.0 KiB
Python
"""将 memory 检索 bundle 压成可入库的轻量 trace（仅稳定 id，限长）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
|
||
def _capped_ids(items: list[Any] | None, *, cap: int = 100) -> list[str]:
|
||
out: list[str] = []
|
||
for x in items or []:
|
||
if len(out) >= cap:
|
||
break
|
||
if isinstance(x, dict) and x.get("id"):
|
||
out.append(str(x["id"]))
|
||
return out
|
||
|
||
|
||
def chat_memory_retrieval_trace_from_bundle(
|
||
bundle: dict,
|
||
*,
|
||
top_k: int,
|
||
query_len: int,
|
||
) -> dict:
|
||
"""供 conversation_messages.memory_retrieval_trace_json 使用。"""
|
||
return {
|
||
"schema_version": 1,
|
||
"source": "chat_memory_retrieval",
|
||
"top_k": top_k,
|
||
"query_len": query_len,
|
||
"chunk_ids": _capped_ids(bundle.get("relevant_chunks")),
|
||
"fact_ids": _capped_ids(bundle.get("relevant_facts")),
|
||
"summary_ids": _capped_ids(bundle.get("relevant_summaries")),
|
||
"story_ids": _capped_ids(bundle.get("relevant_stories")),
|
||
}
|