41518bda11
Embedding 端口增加 is_available()，聊天和回忆录日志用统一方式表示向量是否真能调用。 记忆整理（compaction）支持 Beat 定期扫用户； 事实抽取提示与 subject 归一化，减少同一人多种称呼；
86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
"""Memory evidence 组装与检索契约（纯函数 / 无 DB）。"""
|
||
|
||
import pytest
|
||
|
||
from app.features.memory import evidence as evidence_mod
|
||
from app.features.memory.evidence import (
|
||
EMPTY_EVIDENCE_BUNDLE,
|
||
_facts_to_dicts,
|
||
_stories_to_dicts,
|
||
_timeline_to_dicts,
|
||
retrieve_evidence_bundle_sync,
|
||
)
|
||
from app.features.memory.schemas import EvidenceBundle
|
||
|
||
|
||
class _FakeEmbedding:
|
||
def is_available(self) -> bool:
|
||
return True
|
||
|
||
def embed_text_sync(self, text: str) -> list[float]:
|
||
return [0.25, 0.5, 0.75]
|
||
|
||
|
||
def test_retrieve_evidence_bundle_sync_uses_vector_search(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
searched: list[tuple] = []
|
||
|
||
def fake_search(session, user_id, emb, top_k):
|
||
searched.append((user_id, emb, top_k))
|
||
return [
|
||
{
|
||
"id": "c1",
|
||
"content": "chunk body",
|
||
"chunk_index": 0,
|
||
"distance": 0.1,
|
||
}
|
||
]
|
||
|
||
def fake_meta(session, user_id, q, top_k):
|
||
return {
|
||
"relevant_facts": [],
|
||
"timeline_hints": [],
|
||
"relevant_summaries": [],
|
||
"relevant_stories": [],
|
||
}
|
||
|
||
monkeypatch.setattr(evidence_mod, "search_chunks_vector_sync", fake_search)
|
||
monkeypatch.setattr(evidence_mod, "fetch_evidence_metadata_sync", fake_meta)
|
||
|
||
out = retrieve_evidence_bundle_sync(
|
||
session=object(),
|
||
user_id="u1",
|
||
query=" hello ",
|
||
top_k=7,
|
||
embedding_provider=_FakeEmbedding(),
|
||
)
|
||
assert len(searched) == 1
|
||
assert searched[0][0] == "u1"
|
||
assert searched[0][1] == [0.25, 0.5, 0.75]
|
||
assert searched[0][2] == 7
|
||
assert out["relevant_chunks"] == [
|
||
{"id": "c1", "content": "chunk body", "chunk_index": 0},
|
||
]
|
||
|
||
|
||
def test_empty_evidence_bundle_keys() -> None:
|
||
assert set(EMPTY_EVIDENCE_BUNDLE.keys()) == {
|
||
"relevant_chunks",
|
||
"relevant_summaries",
|
||
"relevant_facts",
|
||
"timeline_hints",
|
||
"relevant_stories",
|
||
}
|
||
|
||
|
||
def test_evidence_bundle_model_accepts_dict() -> None:
|
||
b = EvidenceBundle.model_validate(EMPTY_EVIDENCE_BUNDLE)
|
||
assert b.relevant_chunks == []
|
||
|
||
|
||
def test_format_helpers_empty() -> None:
|
||
assert _facts_to_dicts([]) == []
|
||
assert _timeline_to_dicts([]) == []
|
||
assert _stories_to_dicts([]) == []
|