diff --git a/CHANGELOG.md b/CHANGELOG.md index 812e70a..0240c6a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,33 @@ Hy Memory and MemHub are versioned and released independently. ## Unreleased +- SDK: `search(..., rerank=False)` skips every reranking stage for that one + call (the local cross-encoder, the memory-graph rerank, and the BM25 rerank of + working/coding facts), so results come back in retrieval-score order. The + flag travels on the read request, so concurrent searches do not affect each + other. The default (`True`) keeps current behavior; `/api/v1/search` accepts + it too. +- SDK: MMR diversity selection now runs only when the requested `limit` is + below 20 (`MMR_MAX_TOP_K`) and is skipped for larger limits, where its + quadratic token-overlap pass is slow and gains little. Previously it ran for + every limit. `MEMORY_HYBRID_V2_MMR_LAMBDA` still sets the weight. +- SDK: `search(scene="all")` searches chat, working and coding at the same time + and merges them with Reciprocal Rank Fusion (k=60) over each scene's own + ranking, so no scene is crowded out by the other scenes' score scales. Each + result says which scene and kind it came from. A scene that fails or is turned + off is reported in `errors` and does not fail the search. Source filters apply + to all three. +- SDK: `list_memories(scene="all")` lists chat, working and coding memories + together, newest update first, with `limit`/`offset` paging over the merged + order and `total` summing all three. `scene="coding"` and `"working"` now + honor `include_sources` / `exclude_sources`, and `total` counts only the + filtered memories. +- SDK: `update()` and `delete()` now work on working/coding experiences and + facts, not just chat memories. They take the same optional `kind` / `variant` + as `get()`, and an ID found in more than one place raises + `AmbiguousMemoryIdError` instead of guessing. Updating an experience replaces + its solution and keeps its task, unless a new `task=` is also passed; updating + a fact replaces its statement. Deleting removes the stored row and its vectors. - SDK: `list_memories(scene="chat" | "coding" | "working")` lists one kind of memory. `coding` and `working` list that scene's whole store, newest update first (`workspace_id` still narrows it), and `total` is now the real count diff --git a/hy_memory/client.py b/hy_memory/client.py index 2b09091..dd37ed3 100644 --- a/hy_memory/client.py +++ b/hy_memory/client.py @@ -88,6 +88,47 @@ _WORKING_EXPERIENCE_MIN_SCORE_DELTA = 0.10 # （默认 999），这是防病态长链的安全阀，不是返回条数控制。 _CODING_GET_MANY_MAX_IDS = 500 +# search(scene="all") 的 Reciprocal Rank Fusion 常数：item 在本路第 rank 名 +# （从 1 起）贡献 1 / (_SCENE_RRF_K + rank)。60 是 RRF 原论文（Cormack 2009）的 +# 惯用值：只看名次不看分数，chat 与 working/coding 的分数量纲不同也不会互相压制。 +_SCENE_RRF_K = 60 + +# scene="all" 的三条路，也是同分时的先后顺序。 +_SCENE_LANES = ("chat", "working", "coding") + + +def _epoch_seconds(value: Any, *, naive_is_utc: bool = False) -> Optional[int]: + """datetime / ISO 字符串 / epoch 数字 → 秒级 epoch；取不到返回 None。""" + stamp = _epoch_float(value, naive_is_utc=naive_is_utc) + return int(stamp) if stamp is not None else None + + +def _epoch_float(value: Any, *, naive_is_utc: bool = False) -> Optional[float]: + """同 _epoch_seconds，但保留亚秒精度（排序用，避免同一秒内的先后被抹平）。 + + naive_is_utc：生产力库的时间戳用 utcnow() 写、不带时区，必须按 UTC 解读， + 否则与 chat 侧（本地时间 naive，经 .timestamp() 已是正确 epoch）比较先后时 + 会差出一个时区偏移。 + """ + from datetime import timezone + + if value is None or value == "": + return None + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + try: + value = datetime.fromisoformat(value.rstrip("Z")) + except ValueError: + return None + if not isinstance(value, datetime): + return None + if value.tzinfo is None and naive_is_utc: + value = value.replace(tzinfo=timezone.utc) + return value.timestamp() + _WRITE_DIAGNOSTIC_KEYS = ( "agent_status", @@ -1309,6 +1350,8 @@ class HyMemoryClient: variant: str = "working", include_sources: Optional[List[str]] = None, exclude_sources: Optional[List[str]] = None, + rerank: bool = True, + query_embedding: Optional[List[float]] = None, ) -> Dict[str, Any]: """ Coding 召回路径（事实主导的双层召回）： @@ -1345,6 +1388,9 @@ class HyMemoryClient: 层。这是可接受的——事实是"这个 repo 里发生过什么"，跨 workspace 拿它去带 经验没有意义——但结果里看不出来，所以退化时打 warning。 + rerank=False 时事实层跳过 BM25 词面融合重排（不传 query_text 给 + store.search_facts），按稠密分排序。 + 失败时 fail-safe：返回 success=True + 两层都空（不抛错）。 """ fact_min_score = (_WORKING_FACT_DEFAULT_MIN_SCORE @@ -1417,7 +1463,13 @@ class HyMemoryClient: try: _t_embed = time.perf_counter() - embedding, embed_meta = await self._embed_service.embed_with_meta(target_query) + if query_embedding: + # 调用方已算好（scene="all" 三路共用一次 embed） + embedding, embed_meta = list(query_embedding), {} + else: + embedding, embed_meta = await self._embed_service.embed_with_meta( + target_query + ) if _tracer is not None: await _tracer.log_coding_embed( query=target_query, dims=len(embedding or []), @@ -1437,7 +1489,7 @@ class HyMemoryClient: top=max(limit, 5), # 词面通道 + 把下面那道闸门下移到重排之前：先砍掉过不了闸的候选再重排， # 返回条数就恒等于纯稠密，词面只改选谁（理由见 store.search_facts）。 - query_text=target_query, + query_text=(target_query if rerank else None), min_score=fact_min_score, include_sources=include_sources, exclude_sources=exclude_sources, @@ -1783,6 +1835,7 @@ class HyMemoryClient: layers: Optional[List[Any]] = None, embed_override: Optional[Dict[str, Any]] = None, query_embedding: Optional[List[float]] = None, + rerank: bool = True, ) -> Dict[str, Any]: """ 搜索记忆。 @@ -1812,6 +1865,18 @@ class HyMemoryClient: - "working"：强制走 coding（生产力）召回 - "coding"：同上，额外返回版本演进链块 （"chain_block" / "chain_used" 两个字段，见下方 Returns） + - "all"：chat / working / coding 三路并发召回后用 RRF + 融合成一个列表（见下方 Returns）。chat 路用 + limit/min_score 等 chat 参数；working/coding 路用各自的 + working_fact_min_score / working_experience_min_score。 + 融合只看每路**自己的名次**：score = 1/(60 + rank)， + rank 从 1 起、k=60 为常数（RRF 惯用值），所以不同量纲 + 的分数不会互相压制；同分时按 chat → working → coding + 排。融合后截断到 limit。某一路失败或未启用（coding 关、 + lite 模式）只是不贡献结果，记进 "errors"，不会让整次 + search 失败；没写过的生产力场景直接跳过（不建空库）。 + include_sources / exclude_sources 对三路都生效； + created_after / layers 只作用于 chat 路。 （旧别名 "normal"/"productivity" 仍兼容） search_mode: 搜索执行模式： - "default"（默认）：query="..." 是 light search，直接 @@ -1825,7 +1890,13 @@ class HyMemoryClient: session_ids: Session ID 列表（空列表 = 搜索所有 session） include_sources: 调用来源白名单；空列表 = 不限。 exclude_sources: 调用来源黑名单；与白名单冲突时黑名单优先。 - limit: Normal 路的 topk 数量，默认 10 + limit: Normal 路的 topk 数量，默认 10。这也是 MMR 判定用的 + top-k：limit < 20（MMR_MAX_TOP_K，定义在 + pipelines/_retrieval/diversity.py）时 chat 路自动做 MMR + 多样性选择（权重仍取 MEMORY_HYBRID_V2_MMR_LAMBDA）； + limit >= 20 时跳过 MMR（O(k²·pool) 的 Jaccard 贪心， + 大 k 下代价高、收益小）。重排池是 limit 的若干倍，但判定 + 只看调用方给的 limit。 min_score: **chat 路专用**的最低分数阈值，默认 0.4。working 路不看 这个值，两层各有自己的阈值（见下两项）。 working_fact_min_score: @@ -1854,6 +1925,11 @@ class HyMemoryClient: layers: 可选。pro/ultra 下只召回指定层级（字符串或 MemoryLayer），在召回阶段生效。lite 下忽略该参数并 强制只召回 L1_RAW。 + rerank: 默认 True（原行为）。False 时本次调用跳过全部重排阶段： + chat 路的本地 cross-encoder 重排（recall.local_rerank_enabled） + 与记忆关系图重排（graph_rerank / episode 扩展），working/ + coding 路事实层的 BM25 词面融合重排；结果按召回分排序。 + 按次生效（随请求对象下传），并发 search 互不影响。 Returns: chat 路径：{"request_id": "...", "memories": [...], "elapsed_ms": ...} @@ -1865,6 +1941,20 @@ class HyMemoryClient: "chain_used" —— 已挂进 chain_block 的 facts 下标列表。 facts **没有**被去重过（不悄悄少给）；需要 "块 + 剩余 fact" 的拼法就按这个列表自行过滤。 + scene="all"：{"request_id", "scene": "all", + "memories": [统一条目...], + "lanes": {"chat": n, "working": n, "coding": n}, # 各路参与融合的候选数 + "errors": {lane: "原因"}, # 失败/未启用的路，全成功时为 {} + "elapsed_ms"} + 统一条目在各路原字段之上加/覆盖： + "scene" "chat" | "working" | "coding" + "kind" "chat" | "experience" | "fact" + "memory_id" chat 的 memory_id / experience_id / fact_id + "content" chat 正文；经验为 "task\nsolution"；事实为 statement + "score" RRF 融合分；"lane_score" 为该路原始分；"lane_rank" 为该路名次 + "source", "created_at", "updated_at"（秒级 epoch，取不到为 None） + chat 路内部名次：normal → schema → proactive 依次排；working/coding 路 + 内部名次：事实与经验各按原顺序交替（事实在前），两层谁也不压谁。 """ return self._search_loop_thread.run(self.async_search( query, queries=queries, scene=scene, search_mode=search_mode, @@ -1880,6 +1970,7 @@ class HyMemoryClient: request_id=request_id, layers=layers, embed_override=embed_override, query_embedding=query_embedding, + rerank=rerank, )) def get( @@ -1912,30 +2003,54 @@ class HyMemoryClient: self, memory_id: str, content: str, + *, + kind: Optional[MemoryKind] = None, + variant: Optional[MemoryVariant] = None, + task: Optional[str] = None, ) -> Dict[str, Any]: """ - 更新记忆内容。 + 更新记忆内容。chat / working / coding 都支持，ID 解析与 get() 相同。 Args: - memory_id: 记忆 ID - content: 新内容 + memory_id: 记忆 ID（chat memory_id / experience_id / fact_id） + content: 新内容。chat → 正文；experience → solution（task 不变）； + fact → statement。 + kind: 可选 namespace（"chat" / "experience" / "fact"），消歧用。 + variant: 可选生产力变体（"working" / "coding"），消歧用。 + task: 可选，仅经验：同时替换 task（会重嵌检索 key）。 Returns: {"success": True, "memory_id": "...", "request_id": "...", "elapsed_ms": ...} + 生产力条目额外带 "kind" / "variant"。ID 同时命中多个 namespace 时抛 + AmbiguousMemoryIdError。 """ - return self._loop_thread.run(self.async_update(memory_id, content)) + return self._loop_thread.run(self.async_update( + memory_id, content, kind=kind, variant=variant, task=task, + )) - def delete(self, memory_id: str) -> Dict[str, Any]: + def delete( + self, + memory_id: str, + *, + kind: Optional[MemoryKind] = None, + variant: Optional[MemoryVariant] = None, + ) -> Dict[str, Any]: """ - 删除单条记忆。 + 删除单条记忆。chat / working / coding 都支持，ID 解析与 get() 相同。 Args: - memory_id: 记忆 ID + memory_id: 记忆 ID（chat memory_id / experience_id / fact_id） + kind: 可选 namespace（"chat" / "experience" / "fact"），消歧用。 + variant: 可选生产力变体（"working" / "coding"），消歧用。 Returns: {"success": True, "deleted_count": 1, "request_id": "...", "elapsed_ms": ...} + 生产力条目额外带 "kind" / "variant"。ID 同时命中多个 namespace 时抛 + AmbiguousMemoryIdError。 """ - return self._loop_thread.run(self.async_delete(memory_id)) + return self._loop_thread.run( + self.async_delete(memory_id, kind=kind, variant=variant) + ) def delete_all( self, @@ -2009,13 +2124,27 @@ class HyMemoryClient: 保持兼容，只返回 ACTIVE。可显式读取 SUPERSEDED / SHADOW / NEGATED / CONFLICTED / ARCHIVED；物理删除的节点不会返回。 with_evolution: 可选，为当前页条目附加 evolution_chain（默认关闭）。 - scene: 可选，列哪一类记忆："chat" | "coding" | "working"。 - "coding"/"working" 列该类全部经验（workspace_id 可再收窄）， - 按 updated_at 倒序，total 为真实总数。不传时保持原行为。 + scene: 可选，列哪一类记忆："chat" | "coding" | "working" | "all"。 + "coding"/"working" 列该类全部经验（workspace_id 可再收窄， + include_sources/exclude_sources 按 source 列过滤）， + 按 updated_at 倒序，total 为过滤后的真实总数。 + "all" 把 chat（向量库节点，与 memory_stats 的 chat 同口径） + 与 working/coding 经验合并，按更新时间倒序（chat 取 + gmt_modified，缺省用 gmt_created；经验取 updated_at）分页， + total 为三路之和。include_sources/exclude_sources/statuses + 作用于三路（经验只有 ACTIVE，statuses 不含 ACTIVE 时两路为空）； + agent/session/layers 只作用于 chat，workspace_id/branch 只作用 + 于经验。只支持 order="desc"。不传 scene 时保持原行为。 Returns: chat 路径: {"vdb": {...}, "graph": {...} | 省略, "elapsed_ms": ...} coding 路径: {"scene": "productivity", "experiences": [...], ...} + scene="all": {"scene": "all", "memories": [...], "total", + "totals": {"chat", "working", "coding"}, "errors": {...}, + "limit", "offset", "request_id", "elapsed_ms"} + 每条带 "scene"、"kind"（chat | experience）、"memory_id"、"content" + （经验为 "task\nsolution"）、"source"、"status"、"created_at"、 + "updated_at"（秒级 epoch），以及该路原有字段。 """ return self._loop_thread.run( self.async_list_memories( @@ -3409,6 +3538,226 @@ class HyMemoryClient: # 无事件循环（极少见）：同步跳过，不影响主流程 logger.debug("[search] no running loop for dedup; skipped") + @staticmethod + def _unify_productivity_item( + item: Dict[str, Any], *, kind: str, scene: str, + ) -> Dict[str, Any]: + """生产力经验/事实 → scene="all" 统一条目（保留原字段）。""" + if kind == "experience": + task = str(item.get("task") or "").strip() + solution = str(item.get("solution") or "").strip() + content = "\n".join(part for part in (task, solution) if part) + memory_id = item.get("experience_id") + created = _epoch_seconds(item.get("created_at"), naive_is_utc=True) + updated = _epoch_seconds(item.get("updated_at"), naive_is_utc=True) + else: + content = str(item.get("statement") or "") + memory_id = item.get("fact_id") + created = _epoch_seconds(item.get("created_at"), naive_is_utc=True) + # 事实没有修改时间列：写入后只会被 SUPERSEDE，不会原地改 + updated = created + return { + **item, + "scene": scene, + "kind": kind, + "memory_id": memory_id, + "content": content, + "source": item.get("source") or "default", + "created_at": created, + "updated_at": updated if updated is not None else created, + } + + @staticmethod + def _search_lane_items(lane: str, resp: Dict[str, Any]) -> List[Dict[str, Any]]: + """某一路的 search 结果 → 按本路名次排好的统一条目列表。 + + chat：normal → schema → proactive（与 reader 的组装顺序一致）。 + working/coding：事实与经验各按原顺序交替、事实在前。两层本就分开计数 + （各自 ≤ limit），交替让融合后两层都有代表，谁也不压谁。 + """ + out: List[Dict[str, Any]] = [] + if lane == "chat": + memories = resp.get("memories") or {} + if isinstance(memories, dict): + ordered = ( + list(memories.get("normal") or []) + + list(memories.get("schema") or []) + + list(memories.get("proactive") or []) + ) + else: + ordered = list(memories) + for mem in ordered: + created = _epoch_seconds(mem.get("gmt_created")) + out.append({ + **mem, + "scene": "chat", + "kind": "chat", + "lane_score": mem.get("score"), + "source": mem.get("source") or "default", + "created_at": created, + "updated_at": _epoch_seconds(mem.get("gmt_modified")) or created, + }) + return out + + facts = [ + HyMemoryClient._unify_productivity_item(f, kind="fact", scene=lane) + for f in (resp.get("facts") or []) + ] + exps = [ + HyMemoryClient._unify_productivity_item(e, kind="experience", scene=lane) + for e in (resp.get("experiences") or []) + ] + for index in range(max(len(facts), len(exps))): + if index < len(facts): + out.append(facts[index]) + if index < len(exps): + out.append(exps[index]) + for item in out: + item["lane_score"] = item.get("score") + return out + + @staticmethod + def _rrf_fuse_lanes( + lane_items: Dict[str, List[Dict[str, Any]]], + *, + limit: int, + k: int = _SCENE_RRF_K, + ) -> List[Dict[str, Any]]: + """按各路自己的名次做 RRF：score = 1 / (k + rank)，rank 从 1 起。 + + 条目来自不同 namespace，不跨路合并同 id；同分按 _SCENE_LANES 顺序、再按 + 本路名次排，结果稳定。截断到 limit。 + """ + fused: List[Tuple[float, int, int, Dict[str, Any]]] = [] + for lane_order, lane in enumerate(_SCENE_LANES): + for rank, item in enumerate(lane_items.get(lane) or [], start=1): + score = 1.0 / (k + rank) + fused.append((score, lane_order, rank, { + **item, + "score": round(score, 6), + "lane_rank": rank, + })) + fused.sort(key=lambda entry: (-entry[0], entry[1], entry[2])) + return [entry[3] for entry in fused[:max(0, int(limit))]] + + async def _search_all_scenes( + self, + query: Optional[str], + *, + queries: Optional[List[str]], + search_mode: str, + user_ids: List[str], + agent_ids: List[str], + session_ids: List[str], + include_sources: List[str], + exclude_sources: List[str], + limit: int, + min_score: float, + working_fact_min_score: Optional[float], + working_experience_min_score: Optional[float], + schema_min_score: float, + schema_limit: int, + intention_limit: int, + created_after: Optional[float], + reader: str, + workspace_id: Optional[str], + branch: Optional[str], + request_id: str, + layers: Optional[List[Any]], + query_embedding: Optional[List[float]], + rerank: bool, + target_query: str, + lite_mode: bool, + t0: float, + ) -> Dict[str, Any]: + """search(scene="all")：三路并发召回 + RRF 融合。字段见 search() docstring。 + + scene 是显式路由，不做 LLM 判类/改写：三路都用 queries[-1]（或 query）。 + query 向量只算一次、三路共用。任一路异常只记进 errors，不让整次失败。 + """ + errors: Dict[str, str] = {} + coding_cfg = getattr(getattr(self, "_config", None), "coding", None) + coding_enabled = bool(coding_cfg is not None and coding_cfg.enable) + + embedding = query_embedding + if not embedding: + try: + embedding = await self._embed_service.embed(target_query) + except Exception as e: + # 每一路自己再 embed 一次（也可能各自失败，届时分别记错） + logger.warning(f"[search-all] shared embed failed: {e}") + embedding = None + + async def _chat_lane() -> Dict[str, Any]: + return await self._async_search_impl( + query, queries=queries, scene="chat", search_mode=search_mode, + user_ids=user_ids, agent_ids=agent_ids, session_ids=session_ids, + include_sources=include_sources, exclude_sources=exclude_sources, + limit=limit, min_score=min_score, + schema_min_score=schema_min_score, schema_limit=schema_limit, + intention_limit=intention_limit, created_after=created_after, + reader=reader, request_id=request_id, layers=layers, + query_embedding=embedding, rerank=rerank, + ) + + async def _productivity_lane(variant: str) -> Optional[Dict[str, Any]]: + # 没写过的场景直接跳过：不为一次检索建空库 + store = await self._coding_store_for_exact_get(variant) + if store is None: + return None + return await self._search_coding( + target_query, + user_ids=user_ids, agent_ids=agent_ids, session_ids=session_ids, + workspace_id=workspace_id, branch=branch, limit=limit, + working_fact_min_score=working_fact_min_score, + working_experience_min_score=working_experience_min_score, + request_id=request_id, t0=t0, variant=variant, + include_sources=include_sources, exclude_sources=exclude_sources, + rerank=rerank, query_embedding=embedding, + ) + + lanes_to_run: Dict[str, Any] = {"chat": _chat_lane()} + for variant in ("working", "coding"): + if lite_mode: + errors[variant] = "disabled: lite mode is chat raw-only" + elif not coding_enabled: + errors[variant] = "disabled: productivity link disabled in config" + else: + lanes_to_run[variant] = _productivity_lane(variant) + + names = list(lanes_to_run) + results = await asyncio.gather( + *lanes_to_run.values(), return_exceptions=True, + ) + lane_items: Dict[str, List[Dict[str, Any]]] = {} + for name, result in zip(names, results): + if isinstance(result, BaseException): + if not isinstance(result, Exception): + raise result + logger.warning(f"[search-all] lane {name} failed: {result}") + errors[name] = f"{type(result).__name__}: {result}" + continue + if result is None: + continue + lane_items[name] = self._search_lane_items(name, result) + + memories = self._rrf_fuse_lanes(lane_items, limit=limit) + lanes = {lane: len(lane_items.get(lane) or []) for lane in _SCENE_LANES} + elapsed = round((time.perf_counter() - t0) * 1000, 2) + logger.info( + f"[search-all] lanes={lanes} returned={len(memories)} " + f"errors={sorted(errors)} rerank={rerank} elapsed_ms={elapsed}" + ) + out: Dict[str, Any] = { + "request_id": request_id, + "scene": "all", + "memories": memories, + "lanes": lanes, + "errors": errors, + "elapsed_ms": elapsed, + } + return out + @_ensure_search_loop async def async_search( self, @@ -3437,6 +3786,7 @@ class HyMemoryClient: layers: Optional[List[Any]] = None, embed_override: Optional[Dict[str, Any]] = None, query_embedding: Optional[List[float]] = None, + rerank: bool = True, ) -> Dict[str, Any]: """搜索记忆（异步）。详见 search() docstring。 @@ -3465,6 +3815,7 @@ class HyMemoryClient: reader=reader, workspace_id=workspace_id, branch=branch, request_id=request_id, layers=layers, query_embedding=query_embedding, + rerank=rerank, ) return await self._async_search_impl( query, queries=queries, scene=scene, search_mode=search_mode, @@ -3479,6 +3830,7 @@ class HyMemoryClient: reader=reader, workspace_id=workspace_id, branch=branch, request_id=request_id, layers=layers, query_embedding=query_embedding, + rerank=rerank, ) async def _async_search_impl( @@ -3507,6 +3859,7 @@ class HyMemoryClient: request_id: Optional[str] = None, layers: Optional[List[Any]] = None, query_embedding: Optional[List[float]] = None, + rerank: bool = True, ) -> Dict[str, Any]: """搜索记忆（异步）。详见 search() docstring。 @@ -3542,7 +3895,8 @@ class HyMemoryClient: # "chat" → "normal"（走 chat 召回） # "productivity" / "normal" 作为旧别名保留，向后兼容。 scene_norm: Optional[str] = None - if scene is not None: + scene_all = scene is not None and str(scene).strip().lower() == "all" + if scene is not None and not scene_all: sn = str(scene).strip().lower() _scene_map = { "working": "productivity", @@ -3553,7 +3907,7 @@ class HyMemoryClient: } if sn not in _scene_map: raise ValueError( - f"scene must be 'chat', 'working' or 'coding', got {scene!r}" + f"scene must be 'chat', 'working', 'coding' or 'all', got {scene!r}" ) scene_norm = _scene_map[sn] @@ -3589,6 +3943,25 @@ class HyMemoryClient: "Use single agent_id with multiple session_ids, or multiple agent_ids without session_ids." ) + if scene_all: + return await self._search_all_scenes( + query, queries=queries, search_mode=search_mode, + user_ids=effective_user_ids, + agent_ids=effective_agent_ids, + session_ids=effective_session_ids, + include_sources=effective_include_sources, + exclude_sources=effective_exclude_sources, + limit=limit, min_score=min_score, + working_fact_min_score=working_fact_min_score, + working_experience_min_score=working_experience_min_score, + schema_min_score=schema_min_score, schema_limit=schema_limit, + intention_limit=intention_limit, created_after=created_after, + reader=reader, workspace_id=workspace_id, branch=branch, + request_id=request_id, layers=layers, + query_embedding=query_embedding, rerank=rerank, + target_query=target_query, lite_mode=lite_mode, t0=t0, + ) + # ──────────────────────────────────────────────────────────── # Search mode + scene 路由 # @@ -3857,6 +4230,7 @@ class HyMemoryClient: variant=decided_variant, include_sources=effective_include_sources, exclude_sources=effective_exclude_sources, + rerank=rerank, ) # ──────────────────────────────────────────────────────────── @@ -3924,6 +4298,7 @@ class HyMemoryClient: if effective_search_layers else None ), query_embedding=query_embedding, + rerank=bool(rerank), ) reader_impl = await self._registry.get_reader(reader_name=reader) @@ -4121,6 +4496,92 @@ class HyMemoryClient: "elapsed_ms": elapsed_ms, } + @staticmethod + def _validate_memory_namespace( + kind: Optional[str], variant: Optional[str], + ) -> None: + if kind not in (None, "chat", "experience", "fact"): + raise ValueError( + "kind must be one of: chat, experience, fact" + ) + if variant not in (None, "working", "coding"): + raise ValueError("variant must be one of: working, coding") + if kind == "chat" and variant is not None: + raise ValueError("variant is only valid for experience/fact lookups") + + async def _resolve_memory_namespace( + self, + memory_id: str, + *, + kind: Optional[str], + variant: Optional[str], + chat_lookups: Dict[str, Any], + ) -> Tuple[List[Tuple[str, Any]], Dict[str, Any], Dict[str, Any]]: + """按 kind/variant 在 chat 与 working/coding 的经验/事实里精确查 ID。 + + get / update / delete 共用这一套解析： + - kind/variant 都不给：查 chat + 两个变体的经验和事实 + - variant 给了：只查该变体（不查 chat） + - kind 给了：只查该类 + 没写过的变体不建空库（_coding_store_for_exact_get）。命中多于一个 + namespace 时 fail closed，抛 AmbiguousMemoryIdError。 + + chat_lookups：{"chat:": 零参协程函数}，按顺序取第一个非 None 作为 + chat 命中（get 传 vdb + graph，update 只传 vdb）。 + + Returns: + (matches, by_name, stores) —— matches 是 [(namespace, item)]， + namespace 为 "chat" 或 ":"；by_name 是每次查询的原始 + 结果；stores 是参与查询的 {variant: CodingMemoryStore}。 + """ + query_chat = kind in (None, "chat") and variant is None + coding_kinds = ( + ("experience", "fact") if kind is None else (kind,) + ) + coding_variants = (variant,) if variant is not None else ("working", "coding") + + stores: Dict[str, Any] = {} + if kind != "chat": + for candidate_variant in coding_variants: + store = await self._coding_store_for_exact_get(candidate_variant) + if store is not None: + stores[candidate_variant] = store + + tasks = [] + task_names: List[str] = [] + if query_chat: + for name, lookup in chat_lookups.items(): + tasks.append(lookup()) + task_names.append(name) + for candidate_variant, store in stores.items(): + if "experience" in coding_kinds: + tasks.append(store.get_by_id(memory_id)) + task_names.append(f"{candidate_variant}:experience") + if "fact" in coding_kinds: + tasks.append(store.get_fact(memory_id)) + task_names.append(f"{candidate_variant}:fact") + + values = await asyncio.gather(*tasks) if tasks else [] + by_name = dict(zip(task_names, values)) + + matches: List[Tuple[str, Any]] = [] + chat_node = next( + (by_name[name] for name in chat_lookups + if by_name.get(name) is not None), + None, + ) + if chat_node is not None: + matches.append(("chat", chat_node)) + for name, value in by_name.items(): + if not name.startswith("chat:") and value is not None: + matches.append((name, value)) + + if len(matches) > 1: + raise AmbiguousMemoryIdError( + memory_id, [namespace for namespace, _ in matches] + ) + return matches, by_name, stores + @_ensure_internal_loop async def async_get( self, @@ -4144,14 +4605,7 @@ class HyMemoryClient: Returns: 统一 memory dict，或 None。 """ - if kind not in (None, "chat", "experience", "fact"): - raise ValueError( - "kind must be one of: chat, experience, fact" - ) - if variant not in (None, "working", "coding"): - raise ValueError("variant must be one of: working, coding") - if kind == "chat" and variant is not None: - raise ValueError("variant is only valid for experience/fact lookups") + self._validate_memory_namespace(kind, variant) request_id = str(uuid.uuid4()) set_request_id(request_id) @@ -4174,50 +4628,13 @@ class HyMemoryClient: logger.debug(f"[get] graph_store query failed: {e}") return None - query_chat = kind in (None, "chat") and variant is None - coding_kinds = ( - ("experience", "fact") if kind is None else (kind,) + matches, by_name, _stores = await self._resolve_memory_namespace( + memory_id, kind=kind, variant=variant, + chat_lookups={"chat:vdb": _get_vdb, "chat:graph": _get_graph}, ) - coding_variants = (variant,) if variant is not None else ("working", "coding") - - stores = {} - if kind != "chat": - for candidate_variant in coding_variants: - store = await self._coding_store_for_exact_get(candidate_variant) - if store is not None: - stores[candidate_variant] = store - - tasks = [] - task_names = [] - if query_chat: - tasks.extend((_get_vdb(), _get_graph())) - task_names.extend(("chat:vdb", "chat:graph")) - for candidate_variant, store in stores.items(): - if "experience" in coding_kinds: - tasks.append(store.get_by_id(memory_id)) - task_names.append(f"{candidate_variant}:experience") - if "fact" in coding_kinds: - tasks.append(store.get_fact(memory_id)) - task_names.append(f"{candidate_variant}:fact") - - values = await asyncio.gather(*tasks) if tasks else [] - by_name = dict(zip(task_names, values)) vdb_node = by_name.get("chat:vdb") graph_node = by_name.get("chat:graph") - matches = [] - chat_node = vdb_node if vdb_node is not None else graph_node - if chat_node is not None: - matches.append(("chat", chat_node)) - for name, value in by_name.items(): - if not name.startswith("chat:") and value is not None: - matches.append((name, value)) - - if len(matches) > 1: - raise AmbiguousMemoryIdError( - memory_id, [namespace for namespace, _ in matches] - ) - elapsed = round((time.perf_counter() - t0) * 1000, 2) _hit_namespace, _hit_node = matches[0] if matches else ("", None) _source = ( @@ -4321,14 +4738,139 @@ class HyMemoryClient: return None @_ensure_internal_loop - async def async_update(self, memory_id: str, content: str) -> Dict[str, Any]: + async def async_update( + self, + memory_id: str, + content: str, + *, + kind: Optional[MemoryKind] = None, + variant: Optional[MemoryVariant] = None, + task: Optional[str] = None, + ) -> Dict[str, Any]: """ 更新记忆内容（异步）。 - 流程: get_by_id → 404 校验 → embed → 更新节点字段 → upsert → 记录历史 + namespace 解析与 get() 相同（_resolve_memory_namespace）：kind/variant 都不 + 给时查 chat + working/coding 的经验与事实，命中多个 namespace 抛 + AmbiguousMemoryIdError，用 kind/variant 消歧。chat 只认向量库里的节点 + （与改动前一致）。 + + content 的映射： + - chat：替换正文并重嵌（原行为） + - experience：替换 solution（与 get() 返回的 content 同一字段）；task + 不变，除非另传 task=。solution 不进向量，只改它时不重嵌；改了 task + 则删旧 key 向量行、重嵌 task + search_keys。updated_at 刷新。 + - fact：替换 statement，active 事实重嵌其向量行。 + task= 只对经验有效，给 chat / fact 传会 ValueError。 Returns: - {"success", "memory_id", "request_id", "elapsed_ms"} + {"success", "memory_id", "request_id", "elapsed_ms"}；生产力条目额外带 + "kind" / "variant"。不存在时 success=False + error_message。 + """ + self._validate_memory_namespace(kind, variant) + if task is not None and kind in ("chat", "fact"): + raise ValueError("task is only valid when updating an experience") + matches, _, stores = await self._resolve_memory_namespace( + memory_id, kind=kind, variant=variant, + chat_lookups={ + "chat:vdb": lambda: self._vector_store.get_by_id(memory_id), + }, + ) + namespace = matches[0][0] if matches else "" + if namespace and namespace != "chat": + hit_variant, hit_kind = namespace.split(":", 1) + if task is not None and hit_kind != "experience": + raise ValueError("task is only valid when updating an experience") + return await self._update_productivity( + stores[hit_variant], memory_id, content, + kind=hit_kind, variant=hit_variant, task=task, + ) + if task is not None: + raise ValueError("task is only valid when updating an experience") + if not namespace and (kind not in (None, "chat") or variant is not None): + return self._productivity_not_found( + memory_id, kind=kind, variant=variant, op="update", + ) + return await self._update_chat(memory_id, content) + + @staticmethod + def _productivity_not_found( + memory_id: str, *, kind: Optional[str], variant: Optional[str], op: str, + ) -> Dict[str, Any]: + request_id = str(uuid.uuid4()) + logger.warning( + f"[{op}] not found: memory_id={memory_id} kind={kind} variant={variant}" + ) + out: Dict[str, Any] = { + "success": False, + "memory_id": memory_id, + "kind": kind, + "variant": variant, + "request_id": request_id, + "elapsed_ms": 0.0, + "error_message": f"memory_id '{memory_id}' not found", + } + if op == "delete": + out.update({"deleted_count": 0, "partial": False, "error_code": None}) + return out + + async def _update_productivity( + self, + store, + memory_id: str, + content: str, + *, + kind: str, + variant: str, + task: Optional[str], + ) -> Dict[str, Any]: + request_id = str(uuid.uuid4()) + set_request_id(request_id) + t0 = time.perf_counter() + logger.info( + f"[update] memory_id={memory_id} namespace={variant}:{kind} " + f"new_content({len(content)} chars)" + ) + if kind == "experience": + updated = await store.edit_experience( + memory_id, solution=content, task=task, + ) + else: + updated = await store.edit_fact(memory_id, statement=content) + elapsed = round((time.perf_counter() - t0) * 1000, 2) + if updated is None: + # 解析之后、写入之前被并发删掉 + return { + "success": False, + "memory_id": memory_id, + "kind": kind, + "variant": variant, + "request_id": request_id, + "elapsed_ms": elapsed, + "error_message": f"memory_id '{memory_id}' not found", + } + await self._maybe_store_op_log( + step="UPDATE", + request_id=request_id, + user_id=getattr(updated, "user_id", "") or "", + agent_id=getattr(updated, "agent_id", "") or "", + parsed={"memory_id": memory_id, "namespace": f"{variant}:{kind}"}, + memory_ids=[memory_id], + elapsed_ms=elapsed, + ) + return { + "success": True, + "memory_id": memory_id, + "kind": kind, + "variant": variant, + "request_id": request_id, + "elapsed_ms": elapsed, + } + + async def _update_chat(self, memory_id: str, content: str) -> Dict[str, Any]: + """chat 记忆的更新（原 async_update 逻辑）。 + + 流程: get_by_id → 404 校验 → embed → 更新节点字段 → upsert → 记录历史 """ request_id = str(uuid.uuid4()) set_request_id(request_id) @@ -4471,10 +5013,104 @@ class HyMemoryClient: logger.debug(f"[{step.lower()}] store op trace failed (non-fatal): {e}") @_ensure_internal_loop - async def async_delete(self, memory_id: str) -> Dict[str, Any]: + async def async_delete( + self, + memory_id: str, + *, + kind: Optional[MemoryKind] = None, + variant: Optional[MemoryVariant] = None, + ) -> Dict[str, Any]: """ 删除单条记忆（异步）。 + namespace 解析与 get() 相同：kind/variant 都不给时查 chat（向量库 + 图库） + 与 working/coding 的经验和事实，命中多个 namespace 抛 + AmbiguousMemoryIdError。 + + - chat：原逻辑（见 _delete_chat） + - experience：删 SQLite 行 + 经验向量集合里它的全部 key 行 + - fact：删 SQLite 行 + 事实向量集合里那一行；别处指向它的引用不清， + 召回侧本就按存在性过滤死链 + + Returns: + {"success", "deleted_count", "request_id", "elapsed_ms", ...}； + 生产力条目额外带 "kind" / "variant"。不存在时 success=False、 + deleted_count=0。 + """ + self._validate_memory_namespace(kind, variant) + + async def _get_graph(): + if getattr(self, "_graph_store", None) is None: + return None + try: + return await self._graph_store.get_node(memory_id) + except Exception: + # 图库查不通交给 _delete_chat 报 delete_incomplete + return None + + matches, _, stores = await self._resolve_memory_namespace( + memory_id, kind=kind, variant=variant, + chat_lookups={ + "chat:vdb": lambda: self._vector_store.get_by_id(memory_id), + "chat:graph": _get_graph, + }, + ) + namespace = matches[0][0] if matches else "" + if namespace and namespace != "chat": + hit_variant, hit_kind = namespace.split(":", 1) + return await self._delete_productivity( + stores[hit_variant], memory_id, matches[0][1], + kind=hit_kind, variant=hit_variant, + ) + if not namespace and (kind not in (None, "chat") or variant is not None): + return self._productivity_not_found( + memory_id, kind=kind, variant=variant, op="delete", + ) + return await self._delete_chat(memory_id) + + async def _delete_productivity( + self, store, memory_id: str, item, *, kind: str, variant: str, + ) -> Dict[str, Any]: + request_id = str(uuid.uuid4()) + set_request_id(request_id) + t0 = time.perf_counter() + logger.info(f"[delete] memory_id={memory_id} namespace={variant}:{kind}") + user_id = getattr(item, "user_id", "") or "" + if kind == "experience": + deleted = await store.delete(memory_id, user_id=user_id) + else: + deleted = await store.delete_fact(memory_id, user_id=user_id or None) + elapsed = round((time.perf_counter() - t0) * 1000, 2) + await self._maybe_store_op_log( + step="DELETE", + request_id=request_id, + user_id=user_id, + agent_id=getattr(item, "agent_id", "") or "", + parsed={ + "memory_id": memory_id, + "namespace": f"{variant}:{kind}", + "found": True, + "deleted_count": 1 if deleted else 0, + }, + memory_ids=[memory_id], + elapsed_ms=elapsed, + ) + return { + "success": bool(deleted), + "deleted_count": 1 if deleted else 0, + "memory_id": memory_id, + "kind": kind, + "variant": variant, + "partial": False, + "error_code": None, + "request_id": request_id, + "elapsed_ms": elapsed, + } + + async def _delete_chat(self, memory_id: str) -> Dict[str, Any]: + """ + 删除单条 chat 记忆（原 async_delete 逻辑）。 + 同时删除向量库和图数据库（如有）中的对应节点。 演化链处理： @@ -5210,6 +5846,8 @@ class HyMemoryClient: t0: float, statuses: Optional[List] = None, variant: Optional[str] = None, + include_sources: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, ) -> Dict[str, Any]: """按 workspace_id（+可选 branch）纯过滤列出经验层完整详情。 @@ -5218,7 +5856,14 @@ class HyMemoryClient: variant 给定时读该变体自己的库（没写过的变体不建空库，直接返回空）， workspace_id 为空表示该 user 的全部经验。未给定时沿用"最近一次用的那份"。 + + include_sources / exclude_sources 按 source 列在 SQL 里过滤（排除优先）， + 分页与 total 都在过滤之后计算。 """ + from .utils.source_filter import normalize_source_filters + include_sources, exclude_sources = normalize_source_filters( + include_sources, exclude_sources, + ) empty = { "request_id": request_id, "scene": "productivity", @@ -5249,9 +5894,13 @@ class HyMemoryClient: mems = await store.list_by_workspace( user_id=user_id, workspace_id=workspace_id or "", branch=branch, limit=limit, offset=offset, + include_sources=include_sources or None, + exclude_sources=exclude_sources or None, ) total = (await store.stats( user_id, workspace_id=workspace_id or None, branch=branch, + include_sources=include_sources or None, + exclude_sources=exclude_sources or None, ))["count"] except Exception as e: logger.warning(f"[list-coding] list_by_workspace failed: {e}") @@ -5298,14 +5947,165 @@ class HyMemoryClient: @staticmethod def _list_scene(scene: Optional[str]) -> Optional[str]: - """list/stats 的 scene：None（原有路由）| chat | coding | working。""" + """list/stats 的 scene：None（原有路由）| chat | coding | working | all。""" if scene is None: return None value = str(scene).strip().lower() - if value not in ("chat", "coding", "working"): - raise ValueError(f"scene must be chat, coding or working, got {scene!r}") + if value not in ("chat", "coding", "working", "all"): + raise ValueError( + f"scene must be chat, coding, working or all, got {scene!r}" + ) return value + @staticmethod + def _chat_list_item_unified(node) -> Dict[str, Any]: + """chat 节点 → list(scene="all") 统一条目（在原 list 条目上加字段）。""" + item = HyMemoryClient._memory_node_to_list_item(node) + created = item.get("gmt_created") + updated = _epoch_seconds(getattr(node, "gmt_modified", None)) + item.update({ + "scene": "chat", + "kind": "chat", + "source": item.get("source") or "default", + "created_at": created, + "updated_at": updated if updated is not None else created, + }) + return item + + async def _list_all_scenes( + self, + *, + user_id: str, + agent_id: Optional[str], + user_ids: Optional[List[str]], + agent_ids: Optional[List[str]], + session_ids: Optional[List[str]], + include_sources: Optional[List[str]], + exclude_sources: Optional[List[str]], + layers: Optional[List], + statuses: Optional[List], + workspace_id: Optional[str], + branch: Optional[str], + limit: int, + offset: int, + order: str, + request_id: str, + t0: float, + ) -> Dict[str, Any]: + """list_memories(scene="all")：chat + working + coding 按更新时间合并分页。 + + 分页做法（k 路归并的前缀）：合并序的前 offset+limit 条，必然落在每一路 + 各自的前 offset+limit 条里。所以生产力两路只向 SQLite 要 + ``ORDER BY updated_at DESC LIMIT offset+limit``（有 updated_at 索引）， + total 用 COUNT 聚合；chat 路本来就是整表读回再排序（VDB 没有按时间的 + 有序分页），与 scene=None 的 chat 列表开销相同。三路合并后按 + (updated_at desc, scene, memory_id) 排序再切 [offset, offset+limit)。 + 翻页越深，生产力两路取的前缀越长（O(offset+limit)），但从不整表扫描。 + """ + from .models.memory import MemoryStatus + from .utils.source_filter import normalize_source_filters + + if str(order or "desc").strip().lower() != "desc": + raise ValueError("list_memories(scene='all') only supports order='desc'") + include, exclude = normalize_source_filters(include_sources, exclude_sources) + window = max(0, int(offset)) + max(0, int(limit)) + + nodes = await self._list_vdb_nodes( + user_id=user_id, agent_id=agent_id, + user_ids=user_ids, agent_ids=agent_ids, + session_ids=session_ids, + include_sources=include or None, + exclude_sources=exclude or None, + layers=layers, statuses=statuses, + ) + # 排序键 (-更新时间(亚秒精度), 路序, 本路名次)：本路名次就是该路自己的 + # 排序（生产力路 = SQL 的 updated_at DESC, experience_id ASC），这样合并序 + # 在每一路内部与那一路的取数顺序一致，"只取每路前 offset+limit 条" 才严格成立。 + lane_order = {lane: index for index, lane in enumerate(_SCENE_LANES)} + keyed: List[Tuple[Tuple[float, int, int], Dict[str, Any]]] = [] + + def _chat_ts(node) -> float: + for value in ( + getattr(node, "gmt_modified", None), + getattr(node, "gmt_created", None), + getattr(node, "valid_from", None), + ): + stamp = _epoch_float(value) + if stamp is not None: + return stamp + return float("-inf") + + chat_sorted = sorted( + nodes, key=lambda n: (-_chat_ts(n), str(getattr(n, "node_id", ""))) + ) + for position, node in enumerate(chat_sorted): + keyed.append(( + (-_chat_ts(node), lane_order["chat"], position), + self._chat_list_item_unified(node), + )) + totals: Dict[str, int] = {"chat": len(nodes), "working": 0, "coding": 0} + errors: Dict[str, str] = {} + + include_productivity = not statuses or MemoryStatus.ACTIVE in statuses + productivity_user = user_id or (user_ids[0] if user_ids else "") + for variant in ("working", "coding"): + if not include_productivity: + continue + try: + store = await self._coding_store_for_exact_get(variant) + if store is None: + continue + mems = await store.list_by_workspace( + user_id=productivity_user, + workspace_id=workspace_id or "", branch=branch, + limit=window, offset=0, + include_sources=include or None, + exclude_sources=exclude or None, + ) + totals[variant] = int((await store.stats( + productivity_user, + workspace_id=workspace_id or None, branch=branch, + include_sources=include or None, + exclude_sources=exclude or None, + ))["count"]) + except Exception as e: + logger.warning(f"[list-all] {variant} lane failed: {e}") + errors[variant] = f"{type(e).__name__}: {e}" + continue + for position, d in enumerate(mems): + payload = d.to_dict() + payload["status"] = MemoryStatus.ACTIVE.value + stamp = _epoch_float( + d.updated_at or d.created_at, naive_is_utc=True, + ) + keyed.append(( + (-(stamp if stamp is not None else float("-inf")), + lane_order[variant], position), + self._unify_productivity_item( + payload, kind="experience", scene=variant, + ), + )) + + keyed.sort(key=lambda entry: entry[0]) + page = [item for _, item in keyed[offset: offset + limit]] + total = sum(totals.values()) + elapsed = round((time.perf_counter() - t0) * 1000, 2) + logger.info( + f"[list-all] user={user_ids or user_id} totals={totals} " + f"returned={len(page)} offset={offset} limit={limit} elapsed={elapsed}ms" + ) + return { + "request_id": request_id, + "scene": "all", + "memories": page, + "total": total, + "totals": totals, + "errors": errors, + "limit": limit, + "offset": offset, + "elapsed_ms": elapsed, + } + @_ensure_internal_loop async def async_memory_stats( self, @@ -5356,8 +6156,8 @@ class HyMemoryClient: raw = await store.stats(user_id) stats = { "count": raw["count"], - "latest_created_at": _epoch(raw["latest_created_at"]), - "latest_updated_at": _epoch(raw["latest_updated_at"]), + "latest_created_at": _epoch_seconds(raw["latest_created_at"], naive_is_utc=True), + "latest_updated_at": _epoch_seconds(raw["latest_updated_at"], naive_is_utc=True), "facts": raw["facts"], } scenes[variant] = stats @@ -5463,6 +6263,7 @@ class HyMemoryClient: user_id=user_id, workspace_id=workspace_id, branch=branch, limit=limit, offset=offset, request_id=request_id, t0=t0, statuses=effective_statuses, variant=list_scene, + include_sources=include_sources, exclude_sources=exclude_sources, ) # workspace_id 非空且未指定 scene → 走 coding memory 列表路径（原有行为） if workspace_id and list_scene is None: @@ -5470,9 +6271,21 @@ class HyMemoryClient: user_id=user_id, workspace_id=workspace_id, branch=branch, limit=limit, offset=offset, request_id=request_id, t0=t0, statuses=effective_statuses, + include_sources=include_sources, exclude_sources=exclude_sources, ) effective_layers = self._normalize_layers(layers) + if list_scene == "all": + return await self._list_all_scenes( + user_id=user_id, agent_id=agent_id, + user_ids=user_ids, agent_ids=agent_ids, + session_ids=session_ids, + include_sources=include_sources, exclude_sources=exclude_sources, + layers=effective_layers, statuses=effective_statuses, + workspace_id=workspace_id, branch=branch, + limit=limit, offset=offset, order=order, + request_id=request_id, t0=t0, + ) vdb = await self._list_vdb_bucket( user_id=user_id, agent_id=agent_id, diff --git a/hy_memory/coding/store.py b/hy_memory/coding/store.py index 0f5225c..5503934 100644 --- a/hy_memory/coding/store.py +++ b/hy_memory/coding/store.py @@ -217,6 +217,21 @@ def _in_clause(column: str, values: Optional[List[str]]) -> Tuple[str, List[Any] return f"{column} IN ({placeholders})", list(values) +def _source_where( + include_sources: Optional[List[str]], exclude_sources: Optional[List[str]], +) -> Tuple[str, List[Any]]: + """Caller-source filter as SQL (``1`` when unfiltered); exclusion wins.""" + clauses: List[str] = [] + params: List[Any] = [] + if include_sources: + clauses.append(f"source IN ({','.join('?' for _ in include_sources)})") + params.extend(include_sources) + if exclude_sources: + clauses.append(f"source NOT IN ({','.join('?' for _ in exclude_sources)})") + params.extend(exclude_sources) + return (" AND ".join(clauses) or "1"), params + + def _scope_matches( row_workspace_id: Optional[str], row_branch: Optional[str], @@ -603,6 +618,86 @@ class CodingMemoryStore: logger.info(f"[coding-store] deleted experience_id={experience_id}") return True + async def edit_experience( + self, + experience_id: str, + *, + user_id: Optional[str] = None, + task: Optional[str] = None, + solution: Optional[str] = None, + search_keys: Optional[List[str]] = None, + ) -> Optional[CodingMemory]: + """调用方手动改一条经验（client.update 用）。返回改后的经验；不存在返回 None。 + + 与 update() 的区别：update 是写入链路的覆盖式重写，带写授权校验（同 + agent/session/workspace 才能改）；这里是用户对已有条目的直接编辑，只动 + 给定的字段，其余原样保留，因此不做授权校验。 + + 向量行只嵌 task + search_keys（solution 不进向量），所以只改 solution 时 + 不重嵌、不碰向量库；改了 task / search_keys 才删旧 key 行重写。 + updated_at 总是刷新。 + """ + prev = await self.get_by_id(experience_id, user_id=user_id) + if prev is None: + return None + updated = deepcopy(prev) + if task is not None: + updated.task = task + if solution is not None: + updated.solution = solution + if search_keys is not None: + updated.search_keys = list(search_keys) + updated.updated_at = datetime.utcnow() + await _run_in_pool(self._upsert_sql_sync, updated) + if updated.task != prev.task or updated.search_keys != prev.search_keys: + await self._delete_vec_keys_by_count(experience_id, 1 + len(prev.search_keys)) + await self._upsert_vec_keys(updated) + logger.info(f"[coding-store] edited experience_id={experience_id}") + return updated + + async def edit_fact( + self, + fact_id: str, + *, + statement: str, + user_id: Optional[str] = None, + ) -> Optional[WorkingFact]: + """调用方手动改一条事实的 statement。返回改后的事实；不存在返回 None。 + + statement 是事实层的检索文本，所以 active 事实会重嵌并覆盖它那一行向量 + （node_id 就是 fact_id，upsert 即覆盖）。superseded 事实的向量行早已撤掉， + 这里不把它加回去，否则会让过期结论重新占召回名额。 + """ + prev = await self.get_fact(fact_id, user_id=user_id) + if prev is None: + return None + updated = deepcopy(prev) + updated.statement = statement + await _run_in_pool(self._upsert_fact_sql_sync, updated) + if updated.status == "active": + await self._upsert_fact_vec(updated) + logger.info(f"[coding-store] edited fact_id={fact_id}") + return updated + + async def delete_fact(self, fact_id: str, user_id: Optional[str] = None) -> bool: + """物理删除一条事实（SQLite 行 + 事实向量行），返回是否真有删到。 + + 别处留下的引用（经验的 source_fact_ids、后继事实的 fact_link）不清： + 召回侧本来就按存在性过滤死链（get_many / get_facts 查不到自然掉队）。 + """ + prev = await self.get_fact(fact_id, user_id=user_id) + if prev is None: + logger.info(f"[coding-store] delete fact skipped (not found): {fact_id}") + return False + if self._fact_vec is not None: + try: + await self._fact_vec.delete(fact_id) + except Exception as e: + logger.debug(f"[coding-store] delete fact vec {fact_id} ignored: {e}") + await _run_in_pool(self._delete_fact_sql_sync, fact_id, prev.user_id) + logger.info(f"[coding-store] deleted fact_id={fact_id}") + return True + # ------------------------------------------------------------ # 读取 # ------------------------------------------------------------ @@ -687,26 +782,35 @@ class CodingMemoryStore: branch: Optional[str] = None, limit: int = 100, offset: int = 0, + include_sources: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, ) -> List["CodingMemory"]: """按 workspace_id（+可选 branch）纯过滤列出该 user 的完整 coding memory。 作用域规则见 _scope_where（与 search 路径共用同一套）。无语义排序，按 - updated_at 倒序（最近修改优先）。 + updated_at 倒序（最近修改优先）。include_sources / exclude_sources 按 + source 列在 SQL 里过滤（排除优先），所以 LIMIT/OFFSET 在过滤之后生效。 """ return await _run_in_pool( - self._list_by_workspace_sync, user_id, workspace_id, branch, limit, offset + self._list_by_workspace_sync, user_id, workspace_id, branch, limit, offset, + include_sources, exclude_sources, ) def _list_by_workspace_sync( self, user_id: str, workspace_id: str, branch: Optional[str], limit: int, offset: int, + include_sources: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, ) -> List["CodingMemory"]: scope_sql, scope_params = _scope_where(workspace_id, branch) + source_sql, source_params = _source_where(include_sources, exclude_sources) with self._lock: assert self._conn is not None sql = (f"SELECT * FROM coding_memory_meta WHERE user_id = ? AND {scope_sql} " - f"ORDER BY updated_at DESC LIMIT ? OFFSET ?") - params: List[Any] = [user_id] + scope_params + [limit, offset] + f"AND {source_sql} " + f"ORDER BY updated_at DESC, experience_id ASC LIMIT ? OFFSET ?") + params: List[Any] = ([user_id] + scope_params + source_params + + [limit, offset]) cur = self._conn.execute(sql, params) return [_row_to_memory(r) for r in cur.fetchall()] @@ -715,29 +819,39 @@ class CodingMemoryStore: user_id: str, workspace_id: Optional[str] = None, branch: Optional[str] = None, + include_sources: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, ) -> Dict[str, Any]: """经验数、最新创建/修改时间、active 事实数。纯聚合，不读正文。 - 作用域与 list_by_workspace 相同：workspace_id 为空表示该 user 全部。 + 作用域与 list_by_workspace 相同：workspace_id 为空表示该 user 全部； + 来源过滤同样与 list_by_workspace 一致，count 因此恒等于分页可翻到的总数。 """ - return await _run_in_pool(self._stats_sync, user_id, workspace_id, branch) + return await _run_in_pool( + self._stats_sync, user_id, workspace_id, branch, + include_sources, exclude_sources, + ) def _stats_sync( self, user_id: str, workspace_id: Optional[str], branch: Optional[str], + include_sources: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, ) -> Dict[str, Any]: scope_sql, scope_params = _scope_where(workspace_id, branch) + source_sql, source_params = _source_where(include_sources, exclude_sources) with self._lock: assert self._conn is not None row = self._conn.execute( f"SELECT COUNT(*) AS n, MAX(created_at) AS created, " f"MAX(updated_at) AS updated FROM coding_memory_meta " - f"WHERE user_id = ? AND {scope_sql}", - [user_id] + scope_params, + f"WHERE user_id = ? AND {scope_sql} AND {source_sql}", + [user_id] + scope_params + source_params, ).fetchone() facts = self._conn.execute( f"SELECT COUNT(*) AS n FROM working_fact " - f"WHERE user_id = ? AND status = 'active' AND {scope_sql}", - [user_id] + scope_params, + f"WHERE user_id = ? AND status = 'active' AND {scope_sql} " + f"AND {source_sql}", + [user_id] + scope_params + source_params, ).fetchone() return { "count": int(row["n"] or 0), @@ -1124,6 +1238,14 @@ class CodingMemoryStore: (experience_id, user_id), ) + def _delete_fact_sql_sync(self, fact_id: str, user_id: str) -> None: + with self._lock: + assert self._conn is not None + self._conn.execute( + "DELETE FROM working_fact WHERE fact_id = ? AND user_id = ?", + (fact_id, user_id), + ) + def _get_by_id_sync( self, experience_id: str, user_id: Optional[str] ) -> Optional[CodingMemory]: diff --git a/hy_memory/pipelines/_retrieval/diversity.py b/hy_memory/pipelines/_retrieval/diversity.py index 34b99af..b81ede6 100644 --- a/hy_memory/pipelines/_retrieval/diversity.py +++ b/hy_memory/pipelines/_retrieval/diversity.py @@ -160,6 +160,17 @@ def near_dup_demote( return out +# MMR 只在调用方请求的 top-k（search 的 limit）小于这个值时自动启用。 +# mmr_select 的贪心是 O(k^2 * pool) 次 Jaccard，k 大时代价陡增，而大 k 本身已 +# 足够容纳多样性，收益很小。权重仍由 MEMORY_HYBRID_V2_MMR_LAMBDA 控制。 +MMR_MAX_TOP_K = 20 + + +def mmr_applies(top_k: int, lambda_: float) -> bool: + """本次检索是否走 MMR：λ 在有效区间内，且 top_k < MMR_MAX_TOP_K。""" + return 0.0 < float(lambda_) < 0.999 and int(top_k) < MMR_MAX_TOP_K + + def mmr_select( hits: Sequence[Dict[str, Any]], limit: int, diff --git a/hy_memory/pipelines/base.py b/hy_memory/pipelines/base.py index e3c57b3..427f3a3 100644 --- a/hy_memory/pipelines/base.py +++ b/hy_memory/pipelines/base.py @@ -401,6 +401,11 @@ class ReadRequest: intention_limit: int = 0 # Intention 路的 topk，默认 0 关闭 + # 重排开关（按次生效）。False 时跳过全部重排阶段：本地 cross-encoder + # (recall.local_rerank_enabled) 与记忆关系图重排 (_apply_graph_rerank)， + # 结果按召回分排序。走请求对象而非环境变量，并发请求互不影响。 + rerank: bool = True + # 时间过滤 created_after: Optional[float] = None # Unix timestamp (float)，只返回 gmt_created >= 此值的记忆 diff --git a/hy_memory/pipelines/reader_base.py b/hy_memory/pipelines/reader_base.py index 5a5180b..afeab3c 100644 --- a/hy_memory/pipelines/reader_base.py +++ b/hy_memory/pipelines/reader_base.py @@ -58,6 +58,7 @@ from ._retrieval.ner import ( ) from ._retrieval.diversity import ( near_dup_demote, + mmr_applies, mmr_select, social_boilerplate_demote, pool_df_demote, @@ -1279,7 +1280,10 @@ class BaseReadPipeline(ReadPipeline): SAME_SERIES_TIMELINE_ON = bool( getattr(self.config.recall, "same_series_timeline_enabled", False) ) - LOCAL_RERANK_ON = bool( + # 按次的重排开关（search(rerank=False)）：关掉本地 cross-encoder 与图重排， + # 结果按召回分排序。只读请求对象，不碰共享状态，并发请求互不影响。 + RERANK_ON = bool(getattr(request, "rerank", True)) + LOCAL_RERANK_ON = RERANK_ON and bool( getattr(self.config.recall, "local_rerank_enabled", False) ) LOCAL_RERANK_POOL_FACTOR = max(1, int( @@ -1431,6 +1435,9 @@ class BaseReadPipeline(ReadPipeline): MIN_SCORE = _resolve_normal_min_score(request.min_score) final_limit = request.limit if request.limit > 0 else 10 + # MMR 只在调用方的 top-k（final_limit，即 search 的 limit）< MMR_MAX_TOP_K + # 时启用；池子（LOCAL_RERANK 下是 final_limit 的若干倍）不参与判定。 + MMR_ON = mmr_applies(final_limit, MMR_LAMBDA) vdb_sem_limit = max(final_limit * 3, 30) vdb_kw_limit = max(final_limit * 5, 80) graph_limit = max( @@ -2462,7 +2469,7 @@ class BaseReadPipeline(ReadPipeline): # 只把已选出的 top-N 换个顺序收益很小。 relation_paths: List[Dict[str, Any]] = [] _graph_stats: Dict[str, Any] = {} - if self._expansion_on(): + if RERANK_ON and self._expansion_on(): _t_graph = datetime.now() _pool_before = len(normal_results) normal_results, relation_paths, _graph_stats = await self._apply_graph_rerank( @@ -2538,7 +2545,7 @@ class BaseReadPipeline(ReadPipeline): )) if LOCAL_RERANK_ON else final_limit ) - if 0.0 < MMR_LAMBDA < 0.999 and len(normal_results) > _pre_rerank_limit: + if MMR_ON and len(normal_results) > _pre_rerank_limit: normal_results = mmr_select( normal_results, _pre_rerank_limit, @@ -2715,7 +2722,7 @@ class BaseReadPipeline(ReadPipeline): reverse=True, ) if ( - 0.0 < MMR_LAMBDA < 0.999 + MMR_ON and len(multi_seed_pool) > _pre_rerank_limit ): multi_seed_pool = mmr_select( @@ -3544,6 +3551,8 @@ class BaseReadPipeline(ReadPipeline): "episode_capsule_render_mode": EPISODE_CAPSULE_RENDER_MODE, "same_series_timeline_enabled": SAME_SERIES_TIMELINE_ON, "local_rerank_enabled": LOCAL_RERANK_ON, + "rerank": RERANK_ON, + "mmr_applied": MMR_ON, "local_rerank_model": getattr( self.config.recall, "local_rerank_model", diff --git a/hy_memory/server.py b/hy_memory/server.py index 4d51487..e18ccab 100644 --- a/hy_memory/server.py +++ b/hy_memory/server.py @@ -33,9 +33,28 @@ import threading from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Dict, Optional +from urllib.parse import parse_qs, urlparse + +from .retrieval import AmbiguousMemoryIdError logger = logging.getLogger("hy_memory.server") + +def _namespace_args(values: Dict[str, Any]) -> Dict[str, Any]: + """kind / variant from a query string (lists) or a JSON body (scalars). + + They disambiguate an id that exists in more than one namespace (chat, + experience, fact; working or coding). Absent means "resolve it". + """ + args: Dict[str, Any] = {} + for key in ("kind", "variant"): + value = values.get(key) + if isinstance(value, list): + value = value[0] if value else None + if value: + args[key] = str(value) + return args + # Lazy-init global client _client = None _client_lock = None @@ -148,7 +167,6 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): if path == "/api/v1/metrics": minutes = 5 if "?" in self.path: - from urllib.parse import parse_qs, urlparse qs = parse_qs(urlparse(self.path).query) minutes = int(qs.get("minutes", ["5"])[0]) client = _get_client() @@ -161,7 +179,8 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): if m: memory_id = m.group(1) client = _get_client() - result = client.get(memory_id) + query = parse_qs(urlparse(self.path).query) + result = client.get(memory_id, **_namespace_args(query)) if result is None: _json_response(self, 404, {"error": "not_found", "memory_id": memory_id}) else: @@ -170,6 +189,10 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): _json_response(self, 404, {"error": "not_found", "path": path}) + except AmbiguousMemoryIdError as e: + _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)}) + except ValueError as e: + _json_response(self, 400, {"error": str(e)}) except Exception as e: logger.error(f"[server] GET {path} error: {e}", exc_info=True) _json_response(self, 500, {"error": str(e)}) @@ -213,7 +236,7 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): _json_response(self, 500, {"error": str(e)}) def do_PUT(self): - path = self.path.rstrip("/") + path = self.path.split("?")[0].rstrip("/") body = _read_json_body(self) if body is None: _json_response(self, 400, {"error": "invalid JSON body"}) @@ -228,29 +251,43 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): _json_response(self, 400, {"error": "content is required"}) return client = _get_client() - result = client.update(memory_id, content) + result = client.update( + memory_id, + content, + **_namespace_args(body), + **({"task": body["task"]} if body.get("task") else {}), + ) _json_response(self, 200, result) return _json_response(self, 404, {"error": "not_found", "path": path}) + except AmbiguousMemoryIdError as e: + _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)}) + except ValueError as e: + _json_response(self, 400, {"error": str(e)}) except Exception as e: logger.error(f"[server] PUT {path} error: {e}", exc_info=True) _json_response(self, 500, {"error": str(e)}) def do_DELETE(self): - path = self.path.rstrip("/") + path = self.path.split("?")[0].rstrip("/") try: m = re.match(r"^/api/v1/memories/([^/]+)$", path) if m: memory_id = m.group(1) client = _get_client() - result = client.delete(memory_id) + query = parse_qs(urlparse(self.path).query) + result = client.delete(memory_id, **_namespace_args(query)) _json_response(self, 200, result) return _json_response(self, 404, {"error": "not_found", "path": path}) + except AmbiguousMemoryIdError as e: + _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)}) + except ValueError as e: + _json_response(self, 400, {"error": str(e)}) except Exception as e: logger.error(f"[server] DELETE {path} error: {e}", exc_info=True) _json_response(self, 500, {"error": str(e)}) @@ -455,7 +492,8 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler): if key in body: kwargs[key] = body[key] - for key in ("limit", "min_score", "schema_min_score", "schema_limit", "reader"): + for key in ("limit", "min_score", "schema_min_score", "schema_limit", "reader", + "rerank"): if key in body: kwargs[key] = body[key] diff --git a/tests/coding/test_all_scenes.py b/tests/coding/test_all_scenes.py new file mode 100644 index 0000000..e3328e9 --- /dev/null +++ b/tests/coding/test_all_scenes.py @@ -0,0 +1,665 @@ +# -*- coding: utf-8 -*- +"""scene="all"（search / list）与生产力条目的 update / delete。 + +客户端用 object.__new__(HyMemoryClient) 搭最小骨架，公开 async_* 方法通过 +__wrapped__ 直接调（不需要 client 自己的 loop 线程），与 +test_list_and_stats_by_scene.py 同一套写法。 +""" + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from hy_memory.client import _SCENE_RRF_K, HyMemoryClient +from hy_memory.coding.store import CodingMemoryStore +from hy_memory.coding.types import CodingMemory, WorkingFact +from hy_memory.models.memory import MemoryLayer, MemoryStatus +from hy_memory.retrieval import AmbiguousMemoryIdError + +BASE = datetime(2026, 10, 20, 8, 0, 0) + + +def _unwrapped(name): + return getattr(HyMemoryClient, name).__wrapped__ + + +async def _store(coding_config, mock_embed, variant): + store = CodingMemoryStore(coding_config, embed_service=mock_embed, variant=variant) + await store.initialize() + return store + + +async def _noop(*_args, **_kwargs): + return None + + +def _client(coding_config, mock_embed, stores, *, nodes=()): + client = object.__new__(HyMemoryClient) + client._config = coding_config + client._coding_stores = dict(stores) + client._embed_service = mock_embed + client._cache = None + client._graph_store = None + client._history_store = None + client._vector_store = _FakeVectorStore(list(nodes)) + client._get_coding_writer = _noop + client._maybe_store_op_log = _noop + client._decorate_with_chain = lambda *a, **k: ("", []) + return client + + +class _FakeVectorStore: + def __init__(self, nodes): + self.nodes = {n.node_id: n for n in nodes} + + async def list_by_user(self, *, layers=None, **_): + if layers == [MemoryLayer.L0_BASIC_INFO]: + return [] + return list(self.nodes.values()) + + async def get_by_id(self, memory_id): + return self.nodes.get(memory_id) + + +def _chat_node(node_id, *, created, modified=None, source="cursor"): + return SimpleNamespace( + node_id=node_id, layer=MemoryLayer.L2_FACT, status=MemoryStatus.ACTIVE, + content=f"chat {node_id}", memory_at=None, + gmt_created=created, gmt_modified=modified, valid_from=None, + user_id="u", agent_id="a", session_id="s", source=source, + speculate=None, tags=[], custom={}, + ) + + +# ================================================================ +# search(scene="all") +# ================================================================ + +def _chat_resp(*ids): + return { + "request_id": "r", + "memories": { + "normal": [ + {"memory_id": i, "content": f"chat {i}", "score": 0.9 - n * 0.1, + "gmt_created": 1_700_000_000 + n, "source": "cursor"} + for n, i in enumerate(ids) + ], + "schema": [], + "proactive": [], + }, + } + + +def _prod_resp(facts=(), exps=()): + return { + "request_id": "r", + "scene": "productivity", + "facts": [ + {"fact_id": f, "statement": f"stmt {f}", "score": 0.5, + "created_at": "2026-10-20T08:00:00", "source": "claude"} + for f in facts + ], + "experiences": [ + {"experience_id": e, "task": f"task {e}", "solution": f"sol {e}", + "score": 0.7, "updated_at": "2026-10-20T09:00:00", "source": "claude"} + for e in exps + ], + } + + +def _fake_all_client(*, chat, working=None, coding=None, fail=(), enable=True, + seen=None): + """_search_all_scenes 的依赖全部换成桩：只测融合与容错。""" + client = object.__new__(HyMemoryClient) + client._config = SimpleNamespace(coding=SimpleNamespace(enable=enable)) + + class _Embed: + calls = 0 + + async def embed(self, text): + _Embed.calls += 1 + return [0.1, 0.2] + + client._embed_service = _Embed() + seen = seen if seen is not None else {} + + async def _impl(query, **kw): + seen["chat"] = kw + if "chat" in fail: + raise RuntimeError("chat backend down") + return chat + + async def _coding_store(variant): + return object() if {"working": working, "coding": coding}[variant] is not None else None + + async def _search_coding(target_query, **kw): + seen[kw["variant"]] = kw + if kw["variant"] in fail: + raise RuntimeError(f"{kw['variant']} store broken") + return {"working": working, "coding": coding}[kw["variant"]] + + client._async_search_impl = _impl + client._coding_store_for_exact_get = _coding_store + client._search_coding = _search_coding + return client, seen + + +async def _search_all(client, **kw): + params = dict( + queries=None, search_mode="default", user_ids=["u"], agent_ids=[], + session_ids=[], include_sources=["claude"], exclude_sources=["mixed"], + limit=10, min_score=0.4, working_fact_min_score=None, + working_experience_min_score=None, schema_min_score=0.4, schema_limit=10, + intention_limit=0, created_after=None, reader="", workspace_id=None, + branch=None, request_id="req", layers=None, query_embedding=None, + rerank=True, target_query="q", lite_mode=False, t0=0.0, + ) + params.update(kw) + return await HyMemoryClient._search_all_scenes(client, "q", **params) + + +async def test_search_all_fuses_by_lane_rank_with_rrf(): + client, seen = _fake_all_client( + chat=_chat_resp("c1", "c2", "c3"), + working=_prod_resp(facts=["wf1"], exps=["we1", "we2"]), + coding=_prod_resp(exps=["ce1"]), + ) + + out = await _search_all(client, rerank=False) + + assert out["scene"] == "all" + assert out["request_id"] == "req" + assert out["lanes"] == {"chat": 3, "working": 3, "coding": 1} + assert out["errors"] == {} + order = [(m["scene"], m["memory_id"]) for m in out["memories"]] + # 名次 1：chat → working → coding（同分按路序），名次 2：chat / working ... + assert order == [ + ("chat", "c1"), ("working", "wf1"), ("coding", "ce1"), + ("chat", "c2"), ("working", "we1"), + ("chat", "c3"), ("working", "we2"), + ] + first = out["memories"][0] + assert first["score"] == round(1.0 / (_SCENE_RRF_K + 1), 6) + assert first["lane_score"] == pytest.approx(0.9) + assert first["lane_rank"] == 1 and first["kind"] == "chat" + fact = out["memories"][1] + assert fact["kind"] == "fact" and fact["content"] == "stmt wf1" + assert fact["lane_score"] == 0.5 + exp = out["memories"][4] + assert exp["kind"] == "experience" + assert exp["memory_id"] == "we1" + assert exp["content"] == "task we1\nsol we1" + assert exp["score"] == round(1.0 / (_SCENE_RRF_K + 2), 6) + # 生产力时间戳是 naive UTC，按 UTC 换算成 epoch + assert exp["updated_at"] == int( + datetime(2026, 10, 20, 9, tzinfo=timezone.utc).timestamp() + ) + # rerank / 来源过滤 / 阈值传到每一路，query 向量只算一次 + for lane in ("chat", "working", "coding"): + assert seen[lane]["rerank"] is False + assert seen[lane]["include_sources"] == ["claude"] + assert seen[lane]["query_embedding"] == [0.1, 0.2] + assert seen["chat"]["scene"] == "chat" + assert client._embed_service.calls == 1 + + +async def test_search_all_truncates_after_fusion(): + client, _ = _fake_all_client( + chat=_chat_resp("c1", "c2", "c3"), + working=_prod_resp(exps=["we1", "we2"]), + coding=None, + ) + out = await _search_all(client, limit=3) + assert [m["memory_id"] for m in out["memories"]] == ["c1", "we1", "c2"] + # coding 没写过：不贡献、不算错误 + assert out["lanes"]["coding"] == 0 + assert "coding" not in out["errors"] + + +async def test_search_all_isolates_a_failing_lane(): + client, _ = _fake_all_client( + chat=_chat_resp("c1"), + working=_prod_resp(exps=["we1"]), + coding=_prod_resp(exps=["ce1"]), + fail=("working",), + ) + out = await _search_all(client) + assert [m["memory_id"] for m in out["memories"]] == ["c1", "ce1"] + assert out["lanes"] == {"chat": 1, "working": 0, "coding": 1} + assert "store broken" in out["errors"]["working"] + + client, _ = _fake_all_client( + chat=_chat_resp("c1"), working=_prod_resp(exps=["we1"]), fail=("chat",), + ) + out = await _search_all(client) + assert [m["memory_id"] for m in out["memories"]] == ["we1"] + assert "chat backend down" in out["errors"]["chat"] + + +async def test_search_all_reports_disabled_productivity(): + client, seen = _fake_all_client( + chat=_chat_resp("c1"), working=_prod_resp(exps=["we1"]), enable=False, + ) + out = await _search_all(client) + assert [m["memory_id"] for m in out["memories"]] == ["c1"] + assert out["lanes"] == {"chat": 1, "working": 0, "coding": 0} + assert set(out["errors"]) == {"working", "coding"} + assert "working" not in seen + + +async def test_async_search_routes_scene_all(coding_config): + seen = {} + + async def _spy(query, **kw): + seen.update(kw) + return {"scene": "all"} + + client = object.__new__(HyMemoryClient) + client._config = coding_config + client._search_all_scenes = _spy + out = await _unwrapped("async_search")( + client, "q", scene=" ALL ", user_ids=["u"], include_sources=["Claude"], + rerank=False, limit=7, + ) + assert out == {"scene": "all"} + assert seen["rerank"] is False and seen["limit"] == 7 + assert seen["include_sources"] == ["claude"] + assert seen["exclude_sources"] == ["mixed"] + + with pytest.raises(ValueError, match="all"): + await _unwrapped("async_search")(client, "q", scene="every", user_ids=["u"]) + + +async def test_search_all_productivity_lanes_respect_sources(coding_config, mock_embed): + """真 store：两路生产力检索按 source 过滤，经验与事实都过。""" + coding_config.coding.enable = True + working = await _store(coding_config, mock_embed, "working") + coding = await _store(coding_config, mock_embed, "coding") + try: + await working.insert(CodingMemory( + experience_id="w-claude", user_id="u", task="deploy service", + solution="kubectl apply", source="claude", + )) + await working.insert(CodingMemory( + experience_id="w-cursor", user_id="u", task="deploy service", + solution="helm upgrade", source="cursor", + )) + await working.insert_fact(WorkingFact( + fact_id="f-cursor", user_id="u", statement="deploy service", + source="cursor", + )) + await coding.insert(CodingMemory( + experience_id="c-claude", user_id="u", task="deploy service", + solution="make deploy", source="claude", + )) + client = _client(coding_config, mock_embed, + {"working": working, "coding": coding}) + + async def _chat(query, **kw): + return _chat_resp() + + client._async_search_impl = _chat + out = await _search_all( + client, target_query="deploy service", + include_sources=["claude"], exclude_sources=["mixed"], + ) + ids = {(m["scene"], m["memory_id"]) for m in out["memories"]} + assert ids == {("working", "w-claude"), ("coding", "c-claude")} + assert all(m["source"] == "claude" for m in out["memories"]) + finally: + await working.close() + await coding.close() + + +async def test_productivity_search_rerank_false_skips_fact_lexical_rerank( + coding_config, mock_embed, monkeypatch, +): + from hy_memory.coding import lexical + + calls = [] + real = lexical.rerank + + def _spy(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + monkeypatch.setattr(lexical, "rerank", _spy) + working = await _store(coding_config, mock_embed, "working") + try: + for index in range(8): + await working.insert_fact(WorkingFact( + fact_id=f"f{index}", user_id="u", statement="deploy service", + )) + client = _client(coding_config, mock_embed, {"working": working}) + + async def _run(rerank): + return await HyMemoryClient._search_coding( + client, "deploy service", user_ids=["u"], workspace_id=None, + branch=None, limit=2, request_id="r", t0=0.0, + variant="working", rerank=rerank, + ) + + await _run(False) + assert calls == [] + await _run(True) + assert len(calls) == 1 + finally: + await working.close() + + +# ================================================================ +# list_memories(scene="all") 与生产力 list 的来源过滤 +# ================================================================ + +async def test_list_productivity_scene_filters_sources_with_true_total( + coding_config, mock_embed, +): + coding = await _store(coding_config, mock_embed, "coding") + try: + for index, source in enumerate(["claude", "cursor", "claude", "Claude"]): + await coding.insert(CodingMemory( + experience_id=f"c{index}", user_id="u", task="T", solution="S", + source=source, + )) + client = _client(coding_config, mock_embed, {"coding": coding}) + list_memories = _unwrapped("async_list_memories") + + page = await list_memories( + client, user_id="u", scene="coding", include_sources=["CLAUDE"], limit=2, + ) + assert page["total"] == 3 + assert len(page["experiences"]) == 2 + assert all(e["source"] == "claude" for e in page["experiences"]) + + page = await list_memories( + client, user_id="u", scene="coding", exclude_sources=["claude"], + ) + assert page["total"] == 1 + assert [e["experience_id"] for e in page["experiences"]] == ["c1"] + + unfiltered = await list_memories(client, user_id="u", scene="coding") + assert unfiltered["total"] == 4 + finally: + await coding.close() + + +async def _seed_all(coding_config, mock_embed): + coding = await _store(coding_config, mock_embed, "coding") + working = await _store(coding_config, mock_embed, "working") + # 生产力时间戳按 UTC 写；这里直接给 updated_at 再改回去（insert 会刷新） + for eid, hours, source in (("c-old", 1, "claude"), ("c-new", 5, "cursor")): + await coding.insert(CodingMemory( + experience_id=eid, user_id="u", task=f"task {eid}", solution="S", + source=source, + )) + await _set_updated(coding, eid, BASE + timedelta(hours=hours)) + await working.insert(CodingMemory( + experience_id="w-mid", user_id="u", task="task w", solution="W", + source="claude", + )) + await _set_updated(working, "w-mid", BASE + timedelta(hours=3)) + # chat 节点是本地时间 naive；与 UTC 的经验比较时要换算到同一 epoch + utc_offset = datetime.now() - datetime.utcnow() + nodes = [ + _chat_node("m-newest", created=BASE + utc_offset + timedelta(hours=6), + source="claude"), + _chat_node("m-edited", created=BASE + utc_offset, + modified=BASE + utc_offset + timedelta(hours=4), source="cursor"), + _chat_node("m-oldest", created=BASE + utc_offset, source="claude"), + ] + return coding, working, nodes + + +async def _set_updated(store, experience_id, when): + def _sync(): + with store._lock: + store._conn.execute( + "UPDATE coding_memory_meta SET updated_at = ? WHERE experience_id = ?", + (when.isoformat(), experience_id), + ) + + _sync() + + +async def test_list_all_merges_newest_first_and_paginates(coding_config, mock_embed): + coding, working, nodes = await _seed_all(coding_config, mock_embed) + try: + client = _client(coding_config, mock_embed, + {"coding": coding, "working": working}, nodes=nodes) + list_memories = _unwrapped("async_list_memories") + + out = await list_memories(client, user_id="u", scene="all", limit=10) + assert out["scene"] == "all" + assert out["total"] == 6 + assert out["totals"] == {"chat": 3, "working": 1, "coding": 2} + assert [(m["scene"], m["memory_id"]) for m in out["memories"]] == [ + ("chat", "m-newest"), + ("coding", "c-new"), + ("chat", "m-edited"), + ("working", "w-mid"), + ("coding", "c-old"), + ("chat", "m-oldest"), + ] + exp = out["memories"][1] + assert exp["kind"] == "experience" and exp["content"] == "task c-new\nS" + assert exp["status"] == "active" + assert exp["updated_at"] > out["memories"][2]["updated_at"] + assert out["memories"][0]["kind"] == "chat" + + pages = [] + for offset in (0, 2, 4): + page = await list_memories( + client, user_id="u", scene="all", limit=2, offset=offset, + ) + assert page["total"] == 6 + pages.extend(m["memory_id"] for m in page["memories"]) + assert pages == [m["memory_id"] for m in out["memories"]] + + with pytest.raises(ValueError, match="desc"): + await list_memories(client, user_id="u", scene="all", order="asc") + finally: + await coding.close() + await working.close() + + +async def test_list_all_applies_source_and_status_filters(coding_config, mock_embed): + coding, working, nodes = await _seed_all(coding_config, mock_embed) + try: + client = _client(coding_config, mock_embed, + {"coding": coding, "working": working}, nodes=nodes) + list_memories = _unwrapped("async_list_memories") + + out = await list_memories( + client, user_id="u", scene="all", include_sources=["claude"], + ) + assert [m["memory_id"] for m in out["memories"]] == [ + "m-newest", "w-mid", "c-old", "m-oldest", + ] + assert out["totals"] == {"chat": 2, "working": 1, "coding": 1} + assert out["total"] == 4 + + out = await list_memories( + client, user_id="u", scene="all", statuses=["archived"], + ) + # 经验只有 ACTIVE：两路生产力不贡献；chat 假库不按状态过滤，原样返回 + assert out["totals"]["working"] == 0 and out["totals"]["coding"] == 0 + assert all(m["scene"] == "chat" for m in out["memories"]) + finally: + await coding.close() + await working.close() + + +async def test_list_all_does_not_create_unwritten_scene_databases(coding_config, mock_embed): + client = _client(coding_config, mock_embed, {}, nodes=[ + _chat_node("m1", created=BASE), + ]) + out = await _unwrapped("async_list_memories")(client, user_id="u", scene="all") + assert out["total"] == 1 + assert out["totals"] == {"chat": 1, "working": 0, "coding": 0} + for variant in ("working", "coding"): + assert not CodingMemoryStore.database_exists( + coding_config, db_path=coding_config.coding.db_path, variant=variant, + ) + + +# ================================================================ +# update / delete 生产力条目 +# ================================================================ + +async def test_update_experience_replaces_solution_and_keeps_task(coding_config, mock_embed): + working = await _store(coding_config, mock_embed, "working") + try: + await working.insert(CodingMemory( + experience_id="e1", user_id="u", task="deploy service", + search_keys=["rollout"], solution="old way", + )) + await _set_updated(working, "e1", BASE) + client = _client(coding_config, mock_embed, {"working": working}) + + out = await _unwrapped("async_update")(client, "e1", "new way") + assert out["success"] is True + assert out["memory_id"] == "e1" + assert out["kind"] == "experience" and out["variant"] == "working" + assert {"request_id", "elapsed_ms"} <= set(out) + + stored = await working.get_by_id("e1") + assert stored.task == "deploy service" + assert stored.solution == "new way" + assert stored.updated_at > BASE + # solution 不进向量：仍能按原 task 召回 + hits = await working.search_by_query_embedding( + await mock_embed.embed("deploy service"), user_id="u", + ) + assert hits and hits[0]["experience_id"] == "e1" + + out = await _unwrapped("async_update")( + client, "e1", "newer way", task="ship service", + ) + assert out["success"] is True + stored = await working.get_by_id("e1") + assert stored.task == "ship service" + hits = await working.search_by_query_embedding( + await mock_embed.embed("ship service"), user_id="u", + ) + assert hits[0]["experience_id"] == "e1" + assert hits[0]["score"] == pytest.approx(1.0, abs=1e-4) + finally: + await working.close() + + +async def test_update_and_delete_fact(coding_config, mock_embed): + coding = await _store(coding_config, mock_embed, "coding") + try: + await coding.insert_fact(WorkingFact( + fact_id="f1", user_id="u", statement="build uses make", + )) + client = _client(coding_config, mock_embed, {"coding": coding}) + + out = await _unwrapped("async_update")( + client, "f1", "build uses bazel", kind="fact", + ) + assert out["success"] is True and out["variant"] == "coding" + assert (await coding.get_fact("f1")).statement == "build uses bazel" + hits = await coding.search_facts( + await mock_embed.embed("build uses bazel"), user_id="u", + ) + assert [f.fact_id for f, _ in hits] == ["f1"] + + with pytest.raises(ValueError, match="task"): + await _unwrapped("async_update")(client, "f1", "x", task="t") + + out = await _unwrapped("async_delete")(client, "f1") + assert out["success"] is True and out["deleted_count"] == 1 + assert out["kind"] == "fact" + assert await coding.get_fact("f1") is None + assert await coding.search_facts( + await mock_embed.embed("build uses bazel"), user_id="u", + ) == [] + finally: + await coding.close() + + +async def test_delete_experience_removes_row_and_vectors(coding_config, mock_embed): + working = await _store(coding_config, mock_embed, "working") + try: + await working.insert(CodingMemory( + experience_id="e1", user_id="u", task="deploy service", + search_keys=["rollout"], solution="S", + )) + client = _client(coding_config, mock_embed, {"working": working}) + + out = await _unwrapped("async_delete")( + client, "e1", kind="experience", variant="working", + ) + assert out["success"] is True + assert out["deleted_count"] == 1 + assert out["memory_id"] == "e1" + assert await working.get_by_id("e1") is None + assert await working._vec.get_by_id("e1:0") is None + assert await working._vec.get_by_id("e1:1") is None + finally: + await working.close() + + +async def test_productivity_not_found_and_ambiguity(coding_config, mock_embed): + working = await _store(coding_config, mock_embed, "working") + coding = await _store(coding_config, mock_embed, "coding") + try: + for store in (working, coding): + await store.insert(CodingMemory( + experience_id="same", user_id="u", task="T", solution="S", + )) + client = _client(coding_config, mock_embed, + {"working": working, "coding": coding}) + + with pytest.raises(AmbiguousMemoryIdError): + await _unwrapped("async_update")(client, "same", "x") + with pytest.raises(AmbiguousMemoryIdError): + await _unwrapped("async_delete")(client, "same") + + out = await _unwrapped("async_delete")(client, "same", variant="coding") + assert out["success"] is True and out["variant"] == "coding" + assert await working.get_by_id("same") is not None + + missing = await _unwrapped("async_update")(client, "nope", "x", kind="experience") + assert missing["success"] is False and "not found" in missing["error_message"] + missing = await _unwrapped("async_delete")(client, "nope", kind="fact") + assert missing["success"] is False and missing["deleted_count"] == 0 + # variant 给了就不回落到 chat + missing = await _unwrapped("async_delete")(client, "nope", variant="working") + assert missing["success"] is False and missing["deleted_count"] == 0 + + with pytest.raises(ValueError): + await _unwrapped("async_delete")(client, "same", kind="chat", variant="coding") + finally: + await working.close() + await coding.close() + + +async def test_update_and_delete_without_kind_still_handle_chat(coding_config, mock_embed): + """默认行为不变：chat ID 走原 chat 路径（store 都没建时也不建库）。""" + client = _client(coding_config, mock_embed, {}) + seen = [] + + async def _chat_update(memory_id, content): + seen.append(("update", memory_id, content)) + return {"success": True, "memory_id": memory_id} + + async def _chat_delete(memory_id): + seen.append(("delete", memory_id)) + return {"success": True, "deleted_count": 1} + + client._update_chat = _chat_update + client._delete_chat = _chat_delete + client._vector_store = _FakeVectorStore([_chat_node("m1", created=BASE)]) + + assert (await _unwrapped("async_update")(client, "m1", "new"))["success"] + assert (await _unwrapped("async_delete")(client, "m1"))["success"] + # 哪都没有：仍交给 chat 路径给出原来的 not-found 结果 + await _unwrapped("async_delete")(client, "ghost") + assert seen == [("update", "m1", "new"), ("delete", "m1"), ("delete", "ghost")] + for variant in ("working", "coding"): + assert not CodingMemoryStore.database_exists( + coding_config, db_path=coding_config.coding.db_path, variant=variant, + ) diff --git a/tests/coding/test_list_and_stats_by_scene.py b/tests/coding/test_list_and_stats_by_scene.py index e05d4d9..5511ab7 100644 --- a/tests/coding/test_list_and_stats_by_scene.py +++ b/tests/coding/test_list_and_stats_by_scene.py @@ -6,7 +6,7 @@ working 两类经验既列不出也数不到：一个 45 条记忆的库在界 """ import asyncio -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from types import SimpleNamespace import pytest @@ -97,13 +97,14 @@ async def test_stats_counts_every_scene_with_its_latest_times(coding_config, moc "latest_updated_at": int((BASE + timedelta(days=1)).timestamp()), } assert stats["scenes"]["coding"]["count"] == 2 + # Productivity stores write naive UTC (utcnow), so naive means UTC. assert stats["scenes"]["coding"]["latest_created_at"] == int( - (BASE + timedelta(hours=2)).timestamp() + (BASE + timedelta(hours=2)).replace(tzinfo=timezone.utc).timestamp() ) # The store stamps updated_at on write; the stat is the newest it holds. stored = [await coding.get_by_id(eid, user_id="u") for eid in ("c1", "c2")] assert stats["scenes"]["coding"]["latest_updated_at"] == max( - int(item.updated_at.timestamp()) for item in stored + int(item.updated_at.replace(tzinfo=timezone.utc).timestamp()) for item in stored ) assert stats["scenes"]["working"]["count"] == 1 assert stats["scenes"]["working"]["facts"] == 1 diff --git a/tests/test_search_rerank_mmr_gates.py b/tests/test_search_rerank_mmr_gates.py new file mode 100644 index 0000000..aca395e --- /dev/null +++ b/tests/test_search_rerank_mmr_gates.py @@ -0,0 +1,258 @@ +# -*- coding: utf-8 -*- +"""search(rerank=...) 与 MMR 的按次开关。 + +- rerank=False：本地 cross-encoder 与图重排都不跑（看 spy 是否被调到）。 +- MMR：调用方的 limit < MMR_MAX_TOP_K 时启用，>= 时跳过。 +- 默认值（rerank=True、limit=10）保持原行为。 +""" + +import asyncio +import math +import uuid + +import pytest + +from hy_memory.config import MemoryConfig +from hy_memory.models.memory import MemoryLayer, MemoryNode, MemoryStatus +from hy_memory.pipelines import reader_base +from hy_memory.pipelines._retrieval.diversity import MMR_MAX_TOP_K, mmr_applies +from hy_memory.pipelines.base import ReadRequest + +DIMS = 32 + + +def _unit(vec): + n = sum(x * x for x in vec) ** 0.5 or 1.0 + return [x / n for x in vec] + + +def _query_vec(): + return _unit([1.0] + [0.0] * (DIMS - 1)) + + +def _vec(cos): + return _unit([cos, math.sqrt(1 - cos * cos)] + [0.0] * (DIMS - 2)) + + +@pytest.fixture +def config(tmp_path): + return MemoryConfig.from_dict({"vector_store": { + "provider": "chroma", + "collection_name": f"t_gates_{uuid.uuid4().hex[:8]}", + "persist_directory": str(tmp_path / "chroma"), + "embedding_dims": DIMS, + }}) + + +async def _reader_with(config, count): + from hy_memory.data.vector_store_chroma import ChromaVectorStore + from hy_memory.pipelines.reader_legacy import LegacyReadPipeline + + store = ChromaVectorStore(config) + await store.initialize() + for index in range(count): + node = MemoryNode( + node_id=f"m{index:02d}", user_id="u1", agent_id="a1", session_id="s1", + content=f"memory number {index} about topic{index} and detail{index}", + layer=MemoryLayer.L2_FACT, status=MemoryStatus.ACTIVE, is_latest=True, + ) + node.embedding = _vec(0.95 - index * 0.01) + await store.upsert(node) + reader = LegacyReadPipeline(config, vector_store=store) + await reader.initialize() + return reader, store + + +def _request(limit, **kw): + return ReadRequest( + query="memory topic", query_embedding=_query_vec(), + user_ids=["u1"], agent_ids=["a1"], + limit=limit, min_score=0.2, schema_min_score=0.2, **kw, + ) + + +@pytest.fixture +def mmr_spy(monkeypatch): + calls = [] + real = reader_base.mmr_select + + def _spy(hits, limit, **kw): + calls.append(limit) + return real(hits, limit, **kw) + + monkeypatch.setattr(reader_base, "mmr_select", _spy) + return calls + + +def test_mmr_threshold_is_defined_once(): + assert MMR_MAX_TOP_K == 20 + assert mmr_applies(19, 0.55) + assert not mmr_applies(20, 0.55) + # 权重仍由 MMR_LAMBDA 控制：关掉 λ 时任何 top-k 都不走 MMR + assert not mmr_applies(5, 0.0) + assert not mmr_applies(5, 1.0) + + +def test_mmr_runs_when_limit_below_threshold(config, mmr_spy): + async def _run(): + reader, store = await _reader_with(config, 12) + try: + resp = await reader.read(_request(5)) + assert resp.success + assert resp.extra["config"]["mmr_applied"] is True + assert len(resp.memories) == 5 + finally: + await store.close() + + asyncio.run(_run()) + assert mmr_spy == [5] + + +def test_mmr_skipped_when_limit_reaches_threshold(config, mmr_spy): + async def _run(): + reader, store = await _reader_with(config, MMR_MAX_TOP_K + 6) + try: + resp = await reader.read(_request(MMR_MAX_TOP_K)) + assert resp.success + assert resp.extra["config"]["mmr_applied"] is False + ids = [m["memory_id"] for m in resp.memories] + assert len(ids) == MMR_MAX_TOP_K + # 无 MMR：按召回分（构造时单调递减）原序截断 + assert ids == [f"m{index:02d}" for index in range(MMR_MAX_TOP_K)] + finally: + await store.close() + + asyncio.run(_run()) + assert mmr_spy == [] + + +class _FakeReranker: + model_name = "fake-cross-encoder" + device = "cpu" + load_error = "" + + def __init__(self): + self.calls = 0 + + def score(self, query, documents): + self.calls += 1 + # 反转召回序：分数与原名次相反 + return [float(i) for i in range(len(documents))] + + +@pytest.fixture +def rerank_spies(config, monkeypatch): + config.recall.local_rerank_enabled = True + reranker = _FakeReranker() + factory_calls = [] + + def _get_local_reranker(*args, **kwargs): + factory_calls.append(kwargs) + return reranker + + monkeypatch.setattr(reader_base, "get_local_reranker", _get_local_reranker) + graph_calls = [] + + async def _graph_rerank(self, *, hits, **kwargs): + graph_calls.append(len(hits)) + return hits, [], {} + + monkeypatch.setattr( + reader_base.BaseReadPipeline, "_apply_graph_rerank", _graph_rerank, + ) + monkeypatch.setattr( + reader_base.BaseReadPipeline, "_expansion_on", lambda self: True, + ) + return {"reranker": reranker, "factory": factory_calls, "graph": graph_calls} + + +def test_rerank_false_skips_local_and_graph_rerank(config, rerank_spies): + async def _run(): + reader, store = await _reader_with(config, 8) + try: + resp = await reader.read(_request(5, rerank=False)) + assert resp.success + assert resp.extra["config"]["rerank"] is False + assert resp.extra["config"]["local_rerank_enabled"] is False + assert resp.extra["local_rerank"]["applied"] is False + return [m["memory_id"] for m in resp.memories] + finally: + await store.close() + + ids = asyncio.run(_run()) + assert rerank_spies["factory"] == [] + assert rerank_spies["reranker"].calls == 0 + assert rerank_spies["graph"] == [] + assert len(ids) == 5 + + +def test_rerank_default_true_keeps_rerank_stages(config, rerank_spies): + async def _run(): + reader, store = await _reader_with(config, 8) + try: + req = _request(5) + assert req.rerank is True + resp = await reader.read(req) + assert resp.success + assert resp.extra["config"]["rerank"] is True + finally: + await store.close() + + asyncio.run(_run()) + assert len(rerank_spies["factory"]) == 1 + assert rerank_spies["reranker"].calls >= 1 + assert len(rerank_spies["graph"]) == 1 + + +def test_rerank_is_per_request_under_concurrency(config, rerank_spies): + """并发的两次 read 各按自己的 rerank 走，不互相影响。""" + async def _run(): + reader, store = await _reader_with(config, 8) + try: + off, on = await asyncio.gather( + reader.read(_request(5, rerank=False)), + reader.read(_request(5, rerank=True)), + ) + assert off.extra["config"]["rerank"] is False + assert on.extra["config"]["rerank"] is True + assert off.extra["local_rerank"]["applied"] is False + finally: + await store.close() + + asyncio.run(_run()) + assert len(rerank_spies["factory"]) == 1 + assert len(rerank_spies["graph"]) == 1 + + +@pytest.mark.parametrize("kwargs, expected", [({}, True), ({"rerank": False}, False)]) +def test_client_search_threads_rerank_into_read_request(kwargs, expected): + """client.async_search → ReadRequest.rerank；默认 True（原行为）。""" + from hy_memory.client import HyMemoryClient + from hy_memory.pipelines.base import ReadResponse + + seen = {} + + class _Reader: + VERSION = "fake" + + async def read(self, request): + seen["rerank"] = request.rerank + seen["limit"] = request.limit + return ReadResponse() + + class _Registry: + async def get_reader(self, reader_name=""): + return _Reader() + + client = object.__new__(HyMemoryClient) + client._config = MemoryConfig.from_env() + client._cache = None + client._history_store = None + client._registry = _Registry() + client._mode = "pro" + + out = asyncio.run(HyMemoryClient.async_search.__wrapped__( + client, "q", user_ids=["u"], limit=7, **kwargs, + )) + assert "memories" in out + assert seen == {"rerank": expected, "limit": 7} diff --git a/tests/test_server_memory_namespace.py b/tests/test_server_memory_namespace.py new file mode 100644 index 0000000..4dd2ce5 --- /dev/null +++ b/tests/test_server_memory_namespace.py @@ -0,0 +1,103 @@ +"""The memory routes report an ambiguous id as 409 and pass kind/variant on. + +An id can exist in more than one namespace (chat, experience, fact). The +client fails closed with AmbiguousMemoryIdError; over HTTP that was a 500. +""" + +import json +import threading +from http.server import ThreadingHTTPServer +from urllib.request import Request, urlopen +from urllib.error import HTTPError + +import pytest + +from hy_memory import server +from hy_memory.retrieval import AmbiguousMemoryIdError + + +class _Client: + def __init__(self): + self.calls = [] + + def get(self, memory_id, **kwargs): + self.calls.append(("get", memory_id, kwargs)) + if not kwargs: + raise AmbiguousMemoryIdError(memory_id, ("chat", "working/experience")) + return {"memory_id": memory_id} + + def update(self, memory_id, content, **kwargs): + self.calls.append(("update", memory_id, kwargs)) + return {"success": True, "memory_id": memory_id} + + def delete(self, memory_id, **kwargs): + self.calls.append(("delete", memory_id, kwargs)) + if kwargs.get("kind") == "bogus": + raise ValueError("kind must be one of: chat, experience, fact") + return {"success": True, "deleted_count": 1} + + +@pytest.fixture +def running(monkeypatch): + client = _Client() + monkeypatch.setattr(server, "_get_client", lambda: client) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.MemoryHTTPHandler) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + try: + yield client, f"http://127.0.0.1:{httpd.server_address[1]}" + finally: + httpd.shutdown() + httpd.server_close() + + +def _call(method, url, body=None): + data = json.dumps(body).encode() if body is not None else None + request = Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) + try: + with urlopen(request) as response: + return response.status, json.loads(response.read()) + except HTTPError as error: + return error.code, json.loads(error.read()) + + +def test_an_ambiguous_id_is_409_not_500(running): + _, base = running + + status, body = _call("GET", f"{base}/api/v1/memories/m1") + + assert status == 409 + assert body["error"] == "ambiguous_memory_id" + + +def test_kind_and_variant_disambiguate_get_update_and_delete(running): + client, base = running + + assert _call("GET", f"{base}/api/v1/memories/m1?kind=experience&variant=coding")[0] == 200 + assert _call( + "PUT", f"{base}/api/v1/memories/m1", + {"content": "new", "kind": "experience", "variant": "working", "task": "T"}, + )[0] == 200 + assert _call("DELETE", f"{base}/api/v1/memories/m1?kind=fact&variant=coding")[0] == 200 + + assert client.calls == [ + ("get", "m1", {"kind": "experience", "variant": "coding"}), + ("update", "m1", {"kind": "experience", "variant": "working", "task": "T"}), + ("delete", "m1", {"kind": "fact", "variant": "coding"}), + ] + + +def test_a_bad_namespace_is_400(running): + _, base = running + + status, _ = _call("DELETE", f"{base}/api/v1/memories/m1?kind=bogus") + + assert status == 400 + + +def test_chat_calls_without_a_namespace_are_unchanged(running): + client, base = running + + _call("PUT", f"{base}/api/v1/memories/m1", {"content": "new"}) + + assert client.calls == [("update", "m1", {})]