Merge pull request #268 from Tencent-Hunyuan/feat/sdk-intention-dedup-switch

feat(sdk): configurable intention dedup (on by default, unchanged behaviour)
This commit is contained in:
feitianxiang authored and GitHub committed 2026-11-02 14:03:04 +08:00
commit bde0220373
7 files changed
+437 -11

No files matched your search

+9
View File
@@ -4,6 +4,15 @@ Hy Memory and MemHub are versioned and released independently.
## Unreleased
- SDK: Intention dedup is now configurable per client
(`MemoryConfig.recall.intention_dedup_enabled`, default `true`;
`recall.intention_dedup_threshold`, default `MEMORY_DEDUP_THRESHOLD` or
`0.95`). Env defaults: `MEMORY_INTENTION_DEDUP_ENABLED`,
`MEMORY_INTENTION_DEDUP_THRESHOLD`. When on, behaviour is unchanged
(duplicates found at read time are still deleted); when off, no intention
dedup runs, at read time or within a write batch, and nothing is hidden,
dropped or deleted. The threshold applies to read-time dedup only. Refs #241.
## MemHub 1.0.6 — 2026-11-02
Build 37, on `hy-memory==2.0.11`. Linux x64/arm64 Runtime and macOS arm64
+20
View File
@@ -182,9 +182,29 @@ SDK 不设置 Rerank Search 并发上限。开启后模型在 `HyMemoryClient`
| `MEMORY_RECALL_STRENGTH_TAU` | `180` | 衰减时间常数,天 |
| `MEMORY_RECALL_ACCESS_TRACKING` | `true` | Strength 开启时是否回写访问信息 |
| `MEMORY_ENTITY_STORE_ENABLED` | `false` | 是否维护独立 entity collection;当前 Reader 不直接读取 |
| `MEMORY_INTENTION_DEDUP_ENABLED` | `true` | 是否对 L7 Intention 判重(读侧和写入侧批内);见下文 |
| `MEMORY_INTENTION_DEDUP_THRESHOLD` | `MEMORY_DEDUP_THRESHOLD`,再缺省为 `0.95` | Intention 读侧判重的 cosine 阈值 |
旧数据可调用当前公开方法 `client.build_entity_store(user_id)` 补建 Entity Store。
Intention 判重:开启时(默认,与以往行为一致),Search 召回的 L7 Intention 中,
同 `user_id` + 同 `agent_id` + `memory_at` 同一天 + `valid_until` 相同、且 cosine ≥ 阈值的
条目视为复本,只返回最早写入的一条,其余在后台从向量库**删除**。判重不区分内容中的项目、
对象等语义差异,内容高度相似但对象不同的意图可能被误删;写入时同一批抽取结果也会做一次
批内判重。关闭后这两处都不判重,不隐藏、不丢弃、也不删除意图。两项对应
`MemoryConfig.recall.intention_dedup_enabled` /
`recall.intention_dedup_threshold`,按 client 生效,环境变量只提供默认值:
```python
client = HyMemoryClient.from_config({
"recall": {"intention_dedup_enabled": False},
})
```
`MEMORY_DEDUP_THRESHOLD` 同时被写入侧批内意图判重和 dedup 内核使用;
`MEMORY_INTENTION_DEDUP_THRESHOLD` 只影响读侧 Intention 判重,写入侧批内判重的阈值
仍取 `MEMORY_DEDUP_THRESHOLD`。
## 日志、Trace 与历史
| 变量 | 默认值 | 说明 |
+19
View File
@@ -432,6 +432,14 @@ class RecallConfig:
retrieval_clauses_require_keyword: bool = None # env: MEMORY_RETRIEVAL_CLAUSES_REQUIRE_KEYWORD
retrieval_clauses_min_keyword_score: float = None # normalized BM25 floor
supplemental_propositions_enabled: bool = None # env: MEMORY_SUPPLEMENTAL_PROPOSITIONS_ENABLED
# ── Intention(L7)判重 ──
# 开启(默认):召回时同 user+agent+同天 memory_at+同 valid_until 且
# cosine≥阈值 的意图视为复本,只返回最早一条,其余异步从 VDB 删除(历史行为)。
# 关闭:读侧与写入侧批内都不判重——不隐藏、不丢弃、不删除任何意图。
intention_dedup_enabled: bool = None # env: MEMORY_INTENTION_DEDUP_ENABLED
# 读侧意图判重 cosine 阈值。env: MEMORY_INTENTION_DEDUP_THRESHOLD,
# 未设置时回退到共享的 MEMORY_DEDUP_THRESHOLD(兼容旧部署),默认 0.95。
intention_dedup_threshold: float = None
# 最小分数阈值
min_score_threshold: float = 0.0
@@ -750,6 +758,15 @@ class RecallConfig:
self.supplemental_propositions_enabled = _get_env_bool(
"MEMORY_SUPPLEMENTAL_PROPOSITIONS_ENABLED", False
)
if self.intention_dedup_enabled is None:
self.intention_dedup_enabled = _get_env_bool(
"MEMORY_INTENTION_DEDUP_ENABLED", True
)
if self.intention_dedup_threshold is None:
self.intention_dedup_threshold = _get_env_float(
"MEMORY_INTENTION_DEDUP_THRESHOLD",
_get_env_float("MEMORY_DEDUP_THRESHOLD", 0.95),
)
@dataclass
@@ -1384,6 +1401,8 @@ class MemoryConfig:
"supplemental_propositions_enabled": (
self.recall.supplemental_propositions_enabled
),
"intention_dedup_enabled": self.recall.intention_dedup_enabled,
"intention_dedup_threshold": self.recall.intention_dedup_threshold,
},
"pipeline": {
"default_version": self.pipeline.default_version,
+31 -10
View File
@@ -15,6 +15,9 @@ HY Memory - Intention (L7) 召回 + 惰性过期转换 + 复本判定
→ 视为同一个意图的复本(extractor 复读机 / 多 session 重复表达)。
保留组里 `gmt_created` 最早的(老的先来先得,稳定 id);
- **异步删库**:其余复本 fire-and-forget 从 VDB 删除,不影响本次调用 RT。
- **开关**:判重受 `MemoryConfig.recall.intention_dedup_enabled`(默认开)与
`recall.intention_dedup_threshold` 控制,由 reader 按 client 配置传入。
关闭时完全跳过判重:不隐藏、不删除任何意图(写入侧批内判重同样跳过)。
Q: 为什么要看 memory_at 和 valid_until?
A: 意图是 point-in-time 信号。同一句"我周五要开会",如果一条 memory_at 在
@@ -38,8 +41,14 @@ from ...models.memory import MemoryLayer, MemoryStatus
logger = logging.getLogger(__name__)
def _get_dedup_threshold() -> float:
"""去重 cosine 阈值:沿用 dedup 内核的 MEMORY_DEDUP_THRESHOLD,默认 0.95。"""
def _get_dedup_threshold(configured: Optional[float] = None) -> float:
"""去重 cosine 阈值:优先用 client 配置;否则(或配置非法)沿用 dedup 内核的
MEMORY_DEDUP_THRESHOLD,默认 0.95。"""
try:
if configured is not None:
return float(configured)
except (TypeError, ValueError):
logger.warning(f"[intention] invalid dedup threshold: {configured!r}")
try:
return float(os.environ.get("MEMORY_DEDUP_THRESHOLD", 0.95))
except (TypeError, ValueError):
@@ -244,6 +253,8 @@ async def recall_intentions(
stats_out: Optional[Dict[str, Any]] = None,
include_sources: Optional[List[str]] = None,
exclude_sources: Optional[List[str]] = None,
dedup_enabled: bool = True,
dedup_threshold: Optional[float] = None,
) -> List[Dict[str, Any]]:
"""
召回未过期的 L7_INTENTION 节点(VDB),惰性过期转 L2_FACT,判重并异步删复本。
@@ -260,6 +271,11 @@ async def recall_intentions(
now: 当前时间(默认 datetime.now(),测试可注入)
stats_out: 可选 dict,本函数把分段耗时与数量**原地回填**进去,供调用方落
READ_INTENTION pipeline log。传 None 则完全不统计(零开销)。
dedup_enabled: 是否执行读侧判重(及复本异步删库)。False 时跳过判重,
所有未过期意图原样返回,不删除任何节点。默认 True(历史行为)。
dedup_threshold: 判重 cosine 阈值。None 时回退到 env
MEMORY_DEDUP_THRESHOLD(默认 0.95),兼容直接调用本函数的旧代码;
reader 总是传入 `recall.intention_dedup_threshold`。
Returns:
[{"node_id", "score", "node", "layer": "l7_intention", "source": "vdb_intention"}, ...]
@@ -272,7 +288,7 @@ async def recall_intentions(
埋点字段(回填进 stats_out):
耗时(ms): vdb_search_ms / expire_ms / backfill_embed_ms / dedup_ms / total_ms
数量: fetched / expired / survivors / dedup_dropped / returned
配置: fetch_limit / out_limit / dedup_threshold
配置: fetch_limit / out_limit / dedup_enabled / dedup_threshold
说明: expire_ms、backfill_embed_ms 只有真的发生了对应操作才非 0,
借此可判断慢在 VDB 检索、逐条过期转换、补拉向量还是判重计算。
"""
@@ -358,7 +374,7 @@ async def recall_intentions(
_t = time.perf_counter()
_backfilled = 0
_missing_cnt = 0
if survivors:
if survivors and dedup_enabled: # 补拉只为判重;关闭时无需这次 VDB 读
missing_ids = [
h.get("node_id") for h in survivors
if h.get("node_id") and not getattr(h.get("node"), "embedding", None)
@@ -391,18 +407,23 @@ async def recall_intentions(
)
# 2. 判重:同 user + 同天 memory_at + valid_until 同日 + cosine≥threshold
# dedup_enabled=False → 完全跳过:不隐藏、不删除。
_t = time.perf_counter()
threshold = _get_dedup_threshold()
try:
kept, delete_ids = _partition_duplicates(survivors, threshold)
except Exception as e:
# 任何 dedup 异常 → 降级为不去重(返回全部 survivors 截断 limit)
logger.warning(f"[intention] dedup partition failed, degrade to raw: {e}")
threshold = _get_dedup_threshold(dedup_threshold)
if dedup_enabled:
try:
kept, delete_ids = _partition_duplicates(survivors, threshold)
except Exception as e:
# 任何 dedup 异常 → 降级为不去重(返回全部 survivors 截断 limit)
logger.warning(f"[intention] dedup partition failed, degrade to raw: {e}")
kept, delete_ids = survivors, []
else:
kept, delete_ids = survivors, []
# 纯 CPU(cosine 两两比较);survivors 多时可能显著
_stat(
dedup_ms=round((time.perf_counter() - _t) * 1000, 2),
dedup_dropped=len(delete_ids),
dedup_enabled=bool(dedup_enabled),
dedup_threshold=threshold,
)
+6
View File
@@ -1856,6 +1856,12 @@ class BaseReadPipeline(ReadPipeline):
stats_out=_intent_stats,
include_sources=request.include_sources,
exclude_sources=request.exclude_sources,
dedup_enabled=bool(getattr(
self.config.recall, "intention_dedup_enabled", True
)),
dedup_threshold=getattr(
self.config.recall, "intention_dedup_threshold", None
),
)
await trace_log.log_intention(
stats=_intent_stats,
+6 -1
View File
@@ -635,6 +635,7 @@ class MemoryWriter(WritePipeline):
不查 VDB、不做"新写入 vs 库里已有"的对比,避免给 VDB 增加读压力。
库内存量污染由 recall 侧异步删除逐步清理。best-effort:任何异常都
回退到"不判重、全写入",绝不影响 upsert 主流程。
`recall.intention_dedup_enabled=False` 时跳过批内判重,全部写入。
"""
if not intentions:
return []
@@ -660,7 +661,11 @@ class MemoryWriter(WritePipeline):
# ============ 批内 dedup(不查 VDB,只对本批 intentions 互斥判重) ============
keep_mask: List[bool] = [True] * len(intentions)
try:
dropped = self._intention_batch_dedup(intentions, embeddings)
dropped = (
self._intention_batch_dedup(intentions, embeddings)
if getattr(self.config.recall, "intention_dedup_enabled", True)
else []
)
for i in dropped:
if 0 <= i < len(keep_mask):
keep_mask[i] = False
+346
View File
@@ -0,0 +1,346 @@
# -*- coding: utf-8 -*-
"""
Intention(L7)读侧判重开关(issue #241)。
覆盖:
1. 默认开启:行为不变,复本只返回最早一条,其余异步删库;
2. 关闭:不隐藏、不删除,两条意图都返回;
3. 阈值生效:cosine 低于阈值不判重,高于阈值判重;
4. reader 按 client 的 MemoryConfig.recall 读取开关/阈值,而非调用时的 env;
5. MemoryConfig 默认值与 env 回退;
6. 写入侧批内判重同样受开关控制;阈值不影响写入侧;
7. 关闭时不为判重补拉 embedding;配置阈值非法时不影响召回。
"""
from __future__ import annotations
import uuid
from datetime import datetime
import pytest
from hy_memory.config import MemoryConfig
from hy_memory.pipelines.base import ReadRequest
from hy_memory.pipelines._retrieval.intention import recall_intentions
from tests.test_intention_dedup import (
_FakeVectorStore,
_make_intention_node,
_run,
)
_MA = datetime(2025, 8, 6)
_VU = datetime(2025, 8, 31, 23, 59)
_NOW = datetime(2025, 8, 6, 12, 0)
def _unit(vec):
n = sum(x * x for x in vec) ** 0.5 or 1.0
return [x / n for x in vec]
def _seed_project_pair(vs, *, emb_a, emb_b):
"""A / B 两个项目的近似意图:同 user/agent/同天/同截止日。A 更早写入。"""
a = _make_intention_node(
"A 项目每周一触发发布前校验流程", memory_at=_MA, valid_until=_VU,
gmt_created=datetime(2025, 8, 1), embedding=emb_a,
)
b = _make_intention_node(
"B 项目每周一触发发布前校验流程", memory_at=_MA, valid_until=_VU,
gmt_created=datetime(2025, 8, 2), embedding=emb_b,
)
_run(vs.upsert(a))
_run(vs.upsert(b))
return a, b
# ============================================================================
# recall_intentions 参数
# ============================================================================
def test_default_on_keeps_earliest_and_deletes_duplicate():
vs = _FakeVectorStore()
emb = _unit([1.0, 0.0])
a, b = _seed_project_pair(vs, emb_a=emb, emb_b=list(emb))
stats = {}
result = _run(recall_intentions(
vs, emb, user_ids=["u1"], limit=10, now=_NOW, stats_out=stats,
))
assert [h["node_id"] for h in result] == [a.node_id]
assert vs.delete_calls == [b.node_id]
assert stats["dedup_enabled"] is True
assert stats["dedup_dropped"] == 1
def test_off_returns_both_and_deletes_nothing():
vs = _FakeVectorStore()
emb = _unit([1.0, 0.0])
a, b = _seed_project_pair(vs, emb_a=emb, emb_b=list(emb))
stats = {}
result = _run(recall_intentions(
vs, emb, user_ids=["u1"], limit=10, now=_NOW, stats_out=stats,
dedup_enabled=False,
))
assert {h["node_id"] for h in result} == {a.node_id, b.node_id}
assert vs.delete_calls == []
assert stats["dedup_enabled"] is False
assert stats["dedup_dropped"] == 0
def test_threshold_respected():
# cosine(a, b) = 0.9
emb_a = _unit([1.0, 0.0])
emb_b = _unit([0.9, (1 - 0.81) ** 0.5])
# 默认 0.95 → 0.9 不算复本
vs = _FakeVectorStore()
_seed_project_pair(vs, emb_a=emb_a, emb_b=emb_b)
result = _run(recall_intentions(
vs, emb_a, user_ids=["u1"], limit=10, now=_NOW, dedup_threshold=0.95,
))
assert len(result) == 2
assert vs.delete_calls == []
# 调低到 0.85 → 判重,删较晚的 B
vs = _FakeVectorStore()
a, b = _seed_project_pair(vs, emb_a=emb_a, emb_b=emb_b)
stats = {}
result = _run(recall_intentions(
vs, emb_a, user_ids=["u1"], limit=10, now=_NOW,
dedup_threshold=0.85, stats_out=stats,
))
assert [h["node_id"] for h in result] == [a.node_id]
assert vs.delete_calls == [b.node_id]
assert stats["dedup_threshold"] == 0.85
def test_explicit_threshold_wins_over_env(monkeypatch):
monkeypatch.setenv("MEMORY_DEDUP_THRESHOLD", "0.5")
emb_a = _unit([1.0, 0.0])
emb_b = _unit([0.9, (1 - 0.81) ** 0.5])
vs = _FakeVectorStore()
_seed_project_pair(vs, emb_a=emb_a, emb_b=emb_b)
result = _run(recall_intentions(
vs, emb_a, user_ids=["u1"], limit=10, now=_NOW, dedup_threshold=0.95,
))
assert len(result) == 2
assert vs.delete_calls == []
def test_off_skips_embedding_backfill():
vs = _FakeVectorStore()
emb = _unit([1.0, 0.0])
_seed_project_pair(vs, emb_a=emb, emb_b=list(emb))
backfill_calls = []
async def _get_embeddings(ids):
backfill_calls.append(ids)
return {}
vs.get_embeddings = _get_embeddings
for n in vs._nodes.values():
n.embedding = None
result = _run(recall_intentions(
vs, emb, user_ids=["u1"], limit=10, now=_NOW, dedup_enabled=False,
))
assert len(result) == 2
assert backfill_calls == []
def test_invalid_threshold_degrades_to_env_default(monkeypatch):
monkeypatch.delenv("MEMORY_DEDUP_THRESHOLD", raising=False)
vs = _FakeVectorStore()
emb = _unit([1.0, 0.0])
a, b = _seed_project_pair(vs, emb_a=emb, emb_b=list(emb))
stats = {}
result = _run(recall_intentions(
vs, emb, user_ids=["u1"], limit=10, now=_NOW, stats_out=stats,
dedup_threshold="not-a-number",
))
assert [h["node_id"] for h in result] == [a.node_id]
assert stats["dedup_threshold"] == 0.95
# ============================================================================
# 写入侧批内判重
# ============================================================================
class _SameVectorEmbed:
"""所有文本同一向量:cosine=1,开启判重时必判为复本。"""
async def embed_batch(self, texts):
return [_unit([1.0, 0.0]) for _ in texts]
async def embed_queued(self, text):
return _unit([1.0, 0.0])
def _store_batch(config_recall):
from hy_memory.pipelines.base import WriteRequest
from hy_memory.pipelines.writer import MemoryWriter
cfg = MemoryConfig.from_dict({"recall": config_recall})
vs = _FakeVectorStore()
writer = MemoryWriter(cfg, embed_service=_SameVectorEmbed(), vector_store=vs)
req = WriteRequest(
content="", messages=[], user_id="u1", agent_id="a1",
session_id="s1", request_id=uuid.uuid4().hex,
)
intentions = [
{"content": "A 项目每周一触发发布前校验流程", "tags": [], "valid_until": None},
{"content": "B 项目每周一触发发布前校验流程", "tags": [], "valid_until": None},
]
return _run(writer._store_intentions(intentions, req, vs, req.request_id)), vs
def test_write_batch_dedup_default_on_unchanged():
ids, vs = _store_batch({})
assert len(ids) == 1
assert [n.content for n in vs._nodes.values()] == [
"A 项目每周一触发发布前校验流程",
]
def test_write_batch_dedup_off_stores_both():
ids, vs = _store_batch({"intention_dedup_enabled": False})
assert len(ids) == 2
assert len(vs._nodes) == 2
def test_read_threshold_does_not_change_write_batch_dedup(monkeypatch):
"""intention_dedup_threshold 只管读侧;写入侧仍取 MEMORY_DEDUP_THRESHOLD。"""
monkeypatch.delenv("MEMORY_DEDUP_THRESHOLD", raising=False)
ids, _ = _store_batch({"intention_dedup_threshold": 1.5})
assert len(ids) == 1
# ============================================================================
# MemoryConfig
# ============================================================================
def test_config_defaults(monkeypatch):
for k in (
"MEMORY_INTENTION_DEDUP_ENABLED",
"MEMORY_INTENTION_DEDUP_THRESHOLD",
"MEMORY_DEDUP_THRESHOLD",
):
monkeypatch.delenv(k, raising=False)
cfg = MemoryConfig()
assert cfg.recall.intention_dedup_enabled is True
assert cfg.recall.intention_dedup_threshold == 0.95
def test_config_env_fallbacks(monkeypatch):
monkeypatch.setenv("MEMORY_INTENTION_DEDUP_ENABLED", "false")
monkeypatch.setenv("MEMORY_DEDUP_THRESHOLD", "0.9")
monkeypatch.delenv("MEMORY_INTENTION_DEDUP_THRESHOLD", raising=False)
cfg = MemoryConfig()
assert cfg.recall.intention_dedup_enabled is False
# 未设专用阈值 → 回退共享 MEMORY_DEDUP_THRESHOLD(兼容旧部署)
assert cfg.recall.intention_dedup_threshold == 0.9
monkeypatch.setenv("MEMORY_INTENTION_DEDUP_THRESHOLD", "0.8")
assert MemoryConfig().recall.intention_dedup_threshold == 0.8
def test_config_from_dict_overrides_env(monkeypatch):
monkeypatch.setenv("MEMORY_INTENTION_DEDUP_ENABLED", "true")
cfg = MemoryConfig.from_dict({"recall": {
"intention_dedup_enabled": False,
"intention_dedup_threshold": 0.99,
}})
assert cfg.recall.intention_dedup_enabled is False
assert cfg.recall.intention_dedup_threshold == 0.99
assert cfg.to_dict()["recall"]["intention_dedup_enabled"] is False
# ============================================================================
# reader:按 client 配置,而不是调用时的 env
# ============================================================================
def _reader_for(enabled: bool, vs):
from hy_memory.pipelines.reader_hybrid import HybridReadPipeline
cfg = MemoryConfig.from_dict({
"vector_store": {
"provider": "chroma",
"collection_name": f"t_intent_sw_{uuid.uuid4().hex[:8]}",
"embedding_dims": 2,
},
"recall": {"intention_dedup_enabled": enabled},
})
reader = HybridReadPipeline(cfg, vector_store=vs)
reader._vector_store = vs
reader._vector_store_initialized = True
return reader
async def _no_keyword_hits(**kw):
return []
def _read_project_pair(enabled: bool):
"""经 reader 完整读路径召回 A/B 两条近似意图;返回 (返回的意图 id, 删库调用, a, b)。"""
vs = _FakeVectorStore()
vs.keyword_search = _no_keyword_hits
emb = _unit([1.0, 0.0])
a, b = _seed_project_pair(vs, emb_a=emb, emb_b=list(emb))
# 读路径用真实 now:让两条都未过期且同一天
today = datetime.now()
for n in (a, b):
n.memory_at = today
n.valid_until = None
reader = _reader_for(enabled, vs)
_run(reader.initialize())
resp = _run(reader.read(ReadRequest(
query="发布前校验", query_embedding=emb,
user_ids=["u1"], agent_ids=["a1"], limit=10, intention_limit=5,
)))
ids = {
m["memory_id"] for m in resp.memories
if m.get("layer") == "l7_intention"
}
return ids, list(vs.delete_calls), a, b
@pytest.mark.parametrize("env_value", ["true", "false"])
def test_reader_uses_per_client_config_not_env(monkeypatch, env_value):
"""调用时的 env 与 client 配置无关:两个 client 一开一关,各自生效。"""
monkeypatch.setenv("MEMORY_INTENTION_DEDUP_ENABLED", env_value)
monkeypatch.setenv("MEMORY_DEDUP_THRESHOLD", "0.0")
ids_on, deleted_on, a, b = _read_project_pair(True)
assert ids_on == {a.node_id}
assert deleted_on == [b.node_id]
ids_off, deleted_off, a, b = _read_project_pair(False)
assert ids_off == {a.node_id, b.node_id}
assert deleted_off == []
def test_reader_passes_client_threshold(monkeypatch):
monkeypatch.setenv("MEMORY_DEDUP_THRESHOLD", "0.1")
seen = []
async def _spy(*args, **kwargs):
seen.append(kwargs)
return []
monkeypatch.setattr("hy_memory.pipelines.reader_base.recall_intentions", _spy)
vs = _FakeVectorStore()
vs.keyword_search = _no_keyword_hits
reader = _reader_for(True, vs)
reader.config.recall.intention_dedup_threshold = 0.97
_run(reader.initialize())
_run(reader.read(ReadRequest(
query="发布前校验", query_embedding=_unit([1.0, 0.0]),
user_ids=["u1"], agent_ids=["a1"], limit=10, intention_limit=5,
)))
assert seen and seen[0]["dedup_enabled"] is True
assert seen[0]["dedup_threshold"] == 0.97