This reverts commit 0244ebe38c.
Co-authored-by: alvinfei <alvinfei@tencent.com>
This commit is contained in:
feitianxiangandalvinfei authored and 
GitHub committed
1 parent
0244ebe38c
commit
c0737acb44
4 files changed
+33
-379
No files matched your search
+1
-177
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user