diff --git a/hy_memory/client.py b/hy_memory/client.py index d110a26..e2ca32d 100644 --- a/hy_memory/client.py +++ b/hy_memory/client.py @@ -2631,7 +2631,6 @@ class HyMemoryClient: branch: Optional[str] = None, request_id: Optional[str] = None, meta: Optional[Dict[str, Any]] = None, - _existing_raw_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: """写入记忆（异步） @@ -2800,7 +2799,6 @@ class HyMemoryClient: # 落库 memory_ops / WRITE_SUMMARY / pipeline_logs 的单一真相源。 request_id=request_id, client_meta=meta, - existing_raw_ids=list(_existing_raw_ids or []), ) # mode → writer: @@ -2849,10 +2847,7 @@ class HyMemoryClient: logger.info(f"[add] done: memory_id={result.memory_id} elapsed={result.elapsed_ms:.0f}ms") # 记录历史（非阻塞，失败仅 warning） - if ( - self._history_store is not None - and not _existing_raw_ids - ): + if self._history_store is not None: try: isolation_key = self._build_isolation_key(uid, agent_id) # 从 content 或 messages 中提取用于记录的文本 @@ -2984,35 +2979,6 @@ class HyMemoryClient: include_runtime_diag=True, ) - def process_raw( - self, - raw_ids: List[str], - *, - user_id: str, - agent_id: str = "default_agent", - session_id: str = "default_session", - source: str = "default", - scene: Optional[str] = None, - workspace_id: Optional[str] = None, - branch: Optional[str] = None, - request_id: Optional[str] = None, - ) -> Dict[str, Any]: - """Extract/reconcile existing L1 raw records without inserting raw again.""" - return self._loop_thread.run( - self.async_process_raw( - raw_ids, - user_id=user_id, - agent_id=agent_id, - session_id=session_id, - source=source, - scene=scene, - workspace_id=workspace_id, - branch=branch, - request_id=request_id, - ), - include_runtime_diag=True, - ) - def get_raw(self, raw_id: str) -> Optional[Dict[str, Any]]: """Read one SDK raw record through the explicit raw API.""" return self._loop_thread.run(self.async_get_raw(raw_id)) @@ -3300,148 +3266,6 @@ class HyMemoryClient: raise RuntimeError("raw import changed the requested identity") return {"success": True, "raw_id": node_id, "request_id": request_id} - @_ensure_internal_loop - async def async_process_raw( - self, - raw_ids: List[str], - *, - user_id: str, - agent_id: str = "default_agent", - session_id: str = "default_session", - source: str = "default", - scene: Optional[str] = None, - workspace_id: Optional[str] = None, - branch: Optional[str] = None, - request_id: Optional[str] = None, - ) -> Dict[str, Any]: - """Process exact raw records through extraction/reconciliation in caller order.""" - from .models.memory import MemoryLayer - from .utils.source_filter import normalize_source - - if ( - not isinstance(raw_ids, list) - or not raw_ids - or len(raw_ids) > 100 - or any(not isinstance(raw_id, str) or not raw_id for raw_id in raw_ids) - or len(set(raw_ids)) != len(raw_ids) - ): - raise ValueError("raw_ids must contain 1-100 unique non-empty ids") - if not user_id: - raise ValueError("user_id is required") - normalized_source = normalize_source(source) - request_id = request_id or str(uuid.uuid4()) - replay = await self._claim_write_outcome( - request_id=request_id, - user_id=user_id, - agent_id=agent_id, - session_id=session_id, - operation="process_raw", - ) - if replay is not None: - return replay - pipeline_invoked = False - try: - nodes = [] - for raw_id in raw_ids: - node = await self._vector_store.get_by_id(raw_id) - if node is None or node.layer != MemoryLayer.L1_RAW: - raise ValueError(f"raw record is unavailable: {raw_id}") - if ( - node.user_id != user_id - or node.agent_id != agent_id - or node.session_id != session_id - or node.source != normalized_source - ): - raise ValueError("raw record is outside the requested write scope") - nodes.append(node) - source_orders = [(node.custom or {}).get("source_order") for node in nodes] - numeric_orders = [value for value in source_orders if type(value) is int] - if numeric_orders and ( - len(numeric_orders) != len(nodes) - or numeric_orders != sorted(numeric_orders) - or len(set(numeric_orders)) != len(numeric_orders) - ): - raise ValueError("raw_ids are not in stable source order") - roles = [str((node.custom or {}).get("role") or "user") for node in nodes] - if any(role not in {"user", "assistant", "system", "tool"} for role in roles): - raise ValueError("raw record has an unsupported message role") - messages = [ - { - "role": role, - "content": node.content, - **( - {"name": str((node.custom or {})["name"])} - if (node.custom or {}).get("name") - else {} - ), - **( - {"tool_call_id": str((node.custom or {})["tool_call_id"])} - if (node.custom or {}).get("tool_call_id") - else {} - ), - **( - {"tool_name": str((node.custom or {})["tool_name"])} - if (node.custom or {}).get("tool_name") - else {} - ), - } - for node, role in zip(nodes, roles) - ] - pipeline_invoked = True - result = await self._async_add_impl( - messages, - user_id=user_id, - agent_id=agent_id, - session_id=session_id, - source=normalized_source, - _existing_raw_ids=list(raw_ids), - memory_at=max( - (node.memory_at for node in nodes if node.memory_at is not None), - default=None, - ), - scene=scene, - workspace_id=workspace_id, - branch=branch, - request_id=request_id, - ) - memory_ids = list( - dict.fromkeys( - result.get("memory_ids") - or [ - *(result.get("experience_ids") or []), - *(result.get("fact_ids") or []), - ] - ) - ) - result = { - **result, - "raw_ids": list(raw_ids), - "memory_ids": memory_ids, - "request_id": request_id, - } - except BaseException: - if pipeline_invoked: - await self._mark_write_outcome_unknown(request_id) - else: - await self._finish_write_outcome( - request_id, - "failed", - { - "success": False, - "request_id": request_id, - "raw_ids": list(raw_ids), - "memory_ids": [], - "error_code": "process_raw_invalid", - }, - ) - raise - await self._finish_write_outcome( - request_id, - "succeeded" if result.get("success") is True else "failed", - result, - ) - return result - @_ensure_internal_loop async def async_import_memory_exact( self, diff --git a/hy_memory/pipelines/base.py b/hy_memory/pipelines/base.py index 28811f6..e3c57b3 100644 --- a/hy_memory/pipelines/base.py +++ b/hy_memory/pipelines/base.py @@ -231,10 +231,6 @@ class WriteRequest: # 已有记忆 (V1 合并检测用) existing_memories: Optional[List[Dict[str, Any]]] = None - # Public client process_raw() only: reuse these exact L1 nodes as input. - # Ordinary add() never populates this field. - existing_raw_ids: List[str] = field(default_factory=list) - # 记忆时间戳（不传则用当前时间 datetime.now()） memory_at: Optional[datetime] = None diff --git a/hy_memory/pipelines/writer.py b/hy_memory/pipelines/writer.py index 6350fcd..23d5a52 100644 --- a/hy_memory/pipelines/writer.py +++ b/hy_memory/pipelines/writer.py @@ -280,27 +280,6 @@ class MemoryWriter(WritePipeline): custom["source_raw_memory_ids"] = source_ids return custom - async def _attach_existing_raw_ids( - self, - vector_store, - memory_ids: List[str], - raw_ids: List[str], - ) -> None: - """Attach every reused raw record to canonical outputs.""" - for memory_id in memory_ids: - node = await vector_store.get_by_id(memory_id) - if node is None: - continue - custom = dict(node.custom or {}) - source_ids = list(custom.get("source_raw_memory_ids") or []) - legacy_source = str(getattr(node, "source_raw_memory_id", "") or "") - for raw_id in (legacy_source, *raw_ids): - if raw_id and raw_id not in source_ids: - source_ids.append(raw_id) - if source_ids: - custom["source_raw_memory_ids"] = source_ids - await vector_store.update_payload(memory_id, {"custom": custom}) - @staticmethod def _attach_client_meta(custom: Dict[str, Any], request: Optional[WriteRequest]) -> None: """给新建节点挂上调用方 meta（单份 list）。 @@ -2617,23 +2596,13 @@ class MemoryWriter(WritePipeline): from ..metrics import MetricsCollector MetricsCollector.get().sys1_start() - existing_raw_ids = request.existing_raw_ids - if ( - not isinstance(existing_raw_ids, list) - or any(not isinstance(raw_id, str) or not raw_id for raw_id in existing_raw_ids) - ): - raise ValueError("_existing_raw_ids must be a list of non-empty ids") - - # 2. 向量化（处理已有 raw 时复用已存节点，不再次 embed） + # 2. 向量化 + 持久化（L1_RAW） _t_embed = datetime.now() - if existing_raw_ids: - embedding, embed_meta = [], {"reused_raw": True} - else: - with tracer.span("embed") as s: - embedding, embed_meta = await self.embed_service.embed_queued_with_meta( - request.content - ) - s.set_output({"dims": len(embedding), "content_len": len(request.content)}) + with tracer.span("embed") as s: + embedding, embed_meta = await self.embed_service.embed_queued_with_meta( + request.content + ) + s.set_output({"dims": len(embedding), "content_len": len(request.content)}) _embed_ms = (datetime.now() - _t_embed).total_seconds() * 1000 response.extra["embedding"] = embedding await self._emit_pipeline_step( @@ -2656,43 +2625,27 @@ class MemoryWriter(WritePipeline): with tracer.span("qdrant_upsert") as s: try: vector_store = await self._get_vector_store() - if existing_raw_ids: - memory_id = existing_raw_ids[-1] - mem_node = await vector_store.get_by_id(memory_id) - if ( - mem_node is None - or mem_node.layer != MemoryLayer.L1_RAW - or mem_node.user_id != request.user_id - or mem_node.agent_id != (request.agent_id or "default_agent") - or mem_node.session_id != (request.session_id or "default_session") - or mem_node.source != request.source - ): - raise ValueError("existing raw record is outside the write scope") - else: - layer_enum = MemoryLayer.from_string(suggested_layer) - _raw_custom: Dict[str, Any] = {} - self._attach_client_meta(_raw_custom, request) - mem_node = MemoryNode( - user_id=request.user_id, - agent_id=request.agent_id or "default_agent", - session_id=request.session_id or "default_session", - source=request.source, - layer=layer_enum, - content=request.content, - source_type=SourceType.EXPLICIT, - status=MemoryStatus.ACTIVE, - is_latest=True, - embedding=embedding, - memory_at=request.memory_at, - custom=_raw_custom, - ) - memory_id = await vector_store.upsert(mem_node) + layer_enum = MemoryLayer.from_string(suggested_layer) + _raw_custom: Dict[str, Any] = {} + self._attach_client_meta(_raw_custom, request) + mem_node = MemoryNode( + user_id=request.user_id, + agent_id=request.agent_id or "default_agent", + session_id=request.session_id or "default_session", + source=request.source, + layer=layer_enum, + content=request.content, + source_type=SourceType.EXPLICIT, + status=MemoryStatus.ACTIVE, + is_latest=True, + embedding=embedding, + memory_at=request.memory_at, + custom=_raw_custom, + ) + memory_id = await vector_store.upsert(mem_node) response.memory_id = memory_id s.set_output({"memory_id": memory_id, "layer": suggested_layer}) - logger.debug( - f"[write] {'reused' if existing_raw_ids else 'persisted'}: " - f"id={memory_id} layer={suggested_layer}" - ) + logger.debug(f"[write] persisted: id={memory_id} layer={suggested_layer}") except Exception as persist_err: _l1_error = str(persist_err) s.set_error(_l1_error) @@ -2756,12 +2709,6 @@ class MemoryWriter(WritePipeline): recent_history_count=_recent_history_count, ) total_tokens += response.extra.get("_agent_tokens", 0) - if existing_raw_ids and response.extra.get("agent_stored_ids"): - await self._attach_existing_raw_ids( - vector_store, - list(response.extra["agent_stored_ids"]), - existing_raw_ids, - ) if response.extra.get("agent_status") == "failed": await self._emit_pipeline_step( request, @@ -2771,13 +2718,12 @@ class MemoryWriter(WritePipeline): "error": response.extra.get("agent_error", ""), }, ) - if not existing_raw_ids: - await self._rollback_failed_l1( - request=request, - response=response, - vector_store=vector_store, - memory_id=memory_id, - ) + await self._rollback_failed_l1( + request=request, + response=response, + vector_store=vector_store, + memory_id=memory_id, + ) response.tokens_used = total_tokens _failed_workflow_ms = ( datetime.now() - _t_workflow diff --git a/tests/test_raw_public_api.py b/tests/test_raw_public_api.py index 6d470bb..cf2e12b 100644 --- a/tests/test_raw_public_api.py +++ b/tests/test_raw_public_api.py @@ -5,15 +5,7 @@ from unittest.mock import AsyncMock import pytest from hy_memory.client import HyMemoryClient -from hy_memory.config import MemoryConfig -from hy_memory.models.memory import ( - MemoryLayer, - MemoryNode, - MemoryStatus, - SourceType, -) -from hy_memory.pipelines.base import WriteRequest -from hy_memory.pipelines.writer import MemoryWriter +from hy_memory.models.memory import MemoryLayer, MemoryStatus class OutcomeCache: @@ -48,24 +40,15 @@ class RawStore: def __init__(self): self.nodes = {} self.upserts = 0 - self.supports_fulltext = False async def get_by_id(self, node_id): return self.nodes.get(node_id) - async def initialize(self): - return None - async def upsert(self, node): self.upserts += 1 self.nodes[node.node_id] = node return node.node_id - async def update_payload(self, node_id, payload): - node = self.nodes[node_id] - for key, value in payload.items(): - setattr(node, key, value) - async def list_by_user( self, *, @@ -160,101 +143,6 @@ async def test_import_raw_request_replay_does_not_write_twice(): assert client._vector_store.upserts == 1 -@pytest.mark.asyncio -async def test_process_raw_is_idempotent_and_does_not_insert_raw_again(): - client = _client() - for index, role in enumerate(("user", "assistant"), start=1): - await HyMemoryClient.async_import_raw.__wrapped__( - client, - f"turn-{index}", - user_id="u1", - agent_id="a1", - session_id="s1", - source="cursor", - raw_id=f"raw-{index}", - request_id=f"import-{index}", - metadata={"role": role, "source_order": index}, - ) - client._async_add_impl = AsyncMock( - return_value={"success": True, "memory_ids": ["memory-1"]} - ) - - first = await HyMemoryClient.async_process_raw.__wrapped__( - client, - ["raw-1", "raw-2"], - user_id="u1", - agent_id="a1", - session_id="s1", - source="cursor", - request_id="process-1", - ) - second = await HyMemoryClient.async_process_raw.__wrapped__( - client, - ["raw-1", "raw-2"], - user_id="u1", - agent_id="a1", - session_id="s1", - source="cursor", - request_id="process-1", - ) - - assert first["raw_ids"] == ["raw-1", "raw-2"] - assert first["memory_ids"] == ["memory-1"] - assert second == {**first, "replayed": True} - assert client._vector_store.upserts == 2 - client._async_add_impl.assert_awaited_once() - assert client._async_add_impl.await_args.kwargs["_existing_raw_ids"] == [ - "raw-1", - "raw-2", - ] - - -@pytest.mark.asyncio -async def test_writer_reuses_existing_raw_without_embedding_or_upsert(): - store = RawStore() - raw = MemoryNode( - node_id="raw-1", - user_id="u1", - agent_id="a1", - session_id="s1", - source="cursor", - layer=MemoryLayer.L1_RAW, - content="verbatim", - source_type=SourceType.EXPLICIT, - status=MemoryStatus.ACTIVE, - is_latest=True, - embedding=[0.1, 0.2], - ) - store.nodes[raw.node_id] = raw - embed = AsyncMock() - writer = MemoryWriter( - MemoryConfig(), - embed_service=embed, - vector_store=store, - cache=None, - ) - - response = await writer.write( - WriteRequest( - content="[user]: verbatim", - user_id="u1", - agent_id="a1", - session_id="s1", - source="cursor", - request_id="process-1", - existing_raw_ids=["raw-1"], - extra={ - "agent_mode": "disabled", - }, - ) - ) - - assert response.success is True - assert response.memory_id == "raw-1" - assert store.upserts == 0 - embed.embed_queued_with_meta.assert_not_awaited() - - @pytest.mark.asyncio async def test_existing_raw_id_must_match_exact_content_and_owner(): client = _client()