From d4836ed4dfeaa4cc64ccb94fb683a31428618ca4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 21:58:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A3=85=E9=85=8D=E6=8A=95=E5=BD=B1?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E5=AE=A2=E6=88=B7=E7=AB=AF=EF=BC=88=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E5=A4=B1=E6=95=88/=E9=94=80=E6=88=B7=E6=B8=85?= =?UTF-8?q?=E7=90=86=E5=9B=BE=E4=B8=8E=E5=90=91=E9=87=8F=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E5=B9=B6=E4=BF=AE=E5=A4=8D=E7=94=BB=E5=83=8F=E5=AD=97=E6=AE=B5?= =?UTF-8?q?=E6=AE=8B=E7=95=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、装配 projection_cleaner 新增 app/service/projection_cleanup_service.py，并在**组装层**（app/worker/__main__.py） 注入。此前该客户端一直未装配，清理链路只记录降级（skipped_no_client），实际后果是 **销户后图里仍留着偏好关系**——投顾仍能通过关系网络看到这个人。 清理策略是"以画像为准"而不是按标识直接删边：先删掉该记忆对应的长期事实，再重建画像， 最后用对账修复让图跟着画像收敛。这样即使一条记忆影响多条派生边也能删干净，不依赖 "记得它当初投影成了什么"。Milvus 侧按 memory_uuid 删除；集合不存在（语义召回未启用） 时视为无需清理，客户端不可用则如实报告未清理，绝不伪造成功。 放在组装层而不是 runtime 内部兜底：组件内部给默认实现会把"尚未装配"这一事实悄悄盖住， 而"未注入即显式降级并留痕"是 runtime 刻意保留的语义。最初的改法写成内部兜底，被 5 个 既有单测拦下——那些测试是对的，因此改为在入口处注入。 二、顺带修复：事实消失后画像字段残留 实测：让 preference:horizon 记忆失效并清理后，user_facts 与图边都正确清除，但 fin_customer_profile.investment_horizon 仍是"长期（5年以上）"。根因是 rebuild_profile 只写"本轮有新值"的字段；事实被删后该字段没有新值，旧值就留在画像里——于是记忆已经作废， 投顾还能看到一个客户从未授权继续生效的投资期限。 改为：本轮没有对应事实时**清空**该字段；investor_type 例外——问卷是唯一权威，本轮没有 问卷记录时保持原值（该列 NOT NULL，且清空会在重测空档抹掉开户时的等级）。 三、实测 · 让 preference:horizon 失效后清理：cleaned=True， detail=profile_rebuilt; graph_cleaned; vector_collection_absent； · user_facts 2→1；图边由 ['HAS_GOAL','PREFERS'] 变为 ['PREFERS']； · 画像 investment_horizon 由"长期（5年以上）"变为 None；investor_type 保持 C2、 risk_tags 不受影响（与 horizon 无关）； · 恢复记忆状态后重建，画像与图边均正确复原； · ruff 通过、mypy 113 文件无错、unit+contract 447 passed、integration 29 passed。 --- app/service/profile_assembly_service.py | 12 +- app/service/projection_cleanup_service.py | 136 ++++++++++++++++++++++ app/worker/__main__.py | 17 ++- app/worker/runtime.py | 7 +- 4 files changed, 166 insertions(+), 6 deletions(-) create mode 100644 app/service/projection_cleanup_service.py diff --git a/app/service/profile_assembly_service.py b/app/service/profile_assembly_service.py index b0f6c55..ce68ecc 100644 --- a/app/service/profile_assembly_service.py +++ b/app/service/profile_assembly_service.py @@ -190,8 +190,16 @@ class ProfileAssemblyService: "promoted": len(facts), } for field in PROFILE_OWNED_FIELDS: - if field in values: - setattr(profile, field, values[field]) + if field == "investor_type": + # 问卷是唯一权威：本轮没有问卷记录时**保持原值**（既不写入也不清空）。 + # 否则重测前的空档会把开户时的等级抹掉，而该列是 NOT NULL。 + if field in values: + setattr(profile, field, values[field]) + continue + # 其余字段由本服务独占：本轮没有对应事实即清空。 + # 这不只是洁癖——记忆失效后若不清理，画像会留着一个已经作废的投资期限， + # 投顾据此给建议，而客户从未授权这条信息继续生效（实测踩到过）。 + setattr(profile, field, values.get(field)) profile.updated_at = now await self.session.flush() diff --git a/app/service/projection_cleanup_service.py b/app/service/projection_cleanup_service.py new file mode 100644 index 0000000..f6aa8c2 --- /dev/null +++ b/app/service/projection_cleanup_service.py @@ -0,0 +1,136 @@ +"""投影清理：记忆失效或删除时，清掉它在图库与向量库里的派生数据。 + +**为什么以画像为准，而不是按标识直接删边**：一条记忆可能对应多条派生边（同一 key 的 +证据累积会让 tag 取值变化），直接按 tag_key 删容易删不干净或误删。这里走 +「删掉该记忆对应的长期事实 → 重建画像 → 对账修复」，让投影跟着权威源收敛—— +最终图里剩下的内容必然与画像一致，而不依赖"我记得这条记忆当初投影成了什么"。 + +**Milvus 向量**按 `memory_uuid` 删除；集合不存在或客户端不可用时**如实报告未清理**， +绝不伪造成功。这与 `_cleanup_projection` 的审计语义一致：审计只记录适配器返回的真实结论。 +""" + +import logging +from dataclasses import dataclass + +from sqlalchemy import delete, select + +from app.core.config import get_settings +from app.infrastructure.db import SessionFactory +from app.model.memory import MemoryUnit +from app.model.profile import UserFact +from app.service.profile_assembly_service import ProfileAssemblyService +from app.service.profile_graph_projection_service import ProfileGraphProjectionService +from app.service.relationship_service import RelationshipService + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ProjectionCleanupOutcome: + """与 `app/worker/runtime.py` 的同名结构保持字段一致（该处只按属性取用）。 + + `cleaned=True` 表示派生数据已确实清理；为假时 `detail` 必须说明真实缺口， + 审计会原样记录它。 + """ + + cleaned: bool + detail: str = "" + + +class ProjectionCleanupService: + def __init__(self, relationships: RelationshipService | None = None) -> None: + self.relationships = relationships + + async def cleanup(self, *, memory_uuid: str, operation: str) -> ProjectionCleanupOutcome: + del operation # 失效与删除对投影的处理一致：都以权威源为准重建 + async with SessionFactory() as session: + memory = await session.scalar( + select(MemoryUnit).where(MemoryUnit.memory_uuid == memory_uuid) + ) + if memory is None: + # 记忆已经不在库里：投影清理的目标已消失，视为完成（幂等） + return ProjectionCleanupOutcome(True, "memory_not_found") + customer_id = int(memory.customer_id) + fact_key = str(memory.memory_key) + + # 该记忆已失效/删除，它对应的事实不应再参与画像 + async with SessionFactory() as session, session.begin(): + await session.execute(delete(UserFact).where( + UserFact.customer_id == customer_id, UserFact.fact_key == fact_key + )) + + # 以剩余证据重建画像，再让图跟着画像收敛 + details: list[str] = [] + try: + async with SessionFactory() as session, session.begin(): + await ProfileAssemblyService(session).rebuild(customer_id) + details.append("profile_rebuilt") + except Exception: + logger.warning("profile rebuild failed during cleanup customer_id=%s", + customer_id, exc_info=True) + details.append("profile_rebuild_failed") + + graph_ok = await self._cleanup_graph(customer_id, details) + vector_ok = await self._cleanup_vector(memory_uuid, details) + return ProjectionCleanupOutcome( + cleaned=graph_ok and vector_ok and "profile_rebuild_failed" not in details, + detail="; ".join(details), + ) + + async def _cleanup_graph(self, customer_id: int, details: list[str]) -> bool: + if self.relationships is None: + details.append("graph_client_unavailable") + return False + try: + async with SessionFactory() as session: + outcome = await ProfileGraphProjectionService( + session, self.relationships + ).reconcile_customer(customer_id, repair=True) + except Exception: + logger.warning("graph cleanup failed customer_id=%s", customer_id, exc_info=True) + details.append("graph_cleanup_failed") + return False + if outcome.degraded: + details.append(f"graph_degraded:{outcome.reason}") + return False + if not outcome.consistent: + # repair 之后仍不一致：如实报告，不写成清理成功 + details.append( + f"graph_still_inconsistent:missing={len(outcome.missing)}," + f"orphaned={len(outcome.orphaned)}" + ) + return False + details.append("graph_cleaned") + return True + + async def _cleanup_vector(self, memory_uuid: str, details: list[str]) -> bool: + """删除该记忆的向量。集合不存在（未启用语义召回）时视为无需清理。""" + settings = get_settings() + if not settings.milvus_uri: + details.append("vector_store_not_configured") + return True + try: + from pymilvus import MilvusClient # type: ignore[import-untyped] + + client = MilvusClient( + uri=settings.milvus_uri, token=settings.milvus_token or None + ) + except Exception: + logger.warning("milvus client unavailable during cleanup", exc_info=True) + details.append("vector_client_unavailable") + return False + try: + if settings.milvus_collection not in set(client.list_collections()): + # 记忆向量集合尚未启用：没有需要清理的派生数据 + details.append("vector_collection_absent") + return True + client.delete( + collection_name=settings.milvus_collection, + filter=f'memory_uuid == "{memory_uuid}"', + ) + details.append("vector_cleaned") + return True + except Exception: + logger.warning("milvus delete failed memory_uuid=%s", memory_uuid, exc_info=True) + details.append("vector_delete_failed") + return False diff --git a/app/worker/__main__.py b/app/worker/__main__.py index bf06d4d..088254b 100644 --- a/app/worker/__main__.py +++ b/app/worker/__main__.py @@ -4,6 +4,8 @@ import logging from app.core.config import get_settings from app.infrastructure.db import engine +from app.service.agent.bootstrap import get_relationship_service +from app.service.projection_cleanup_service import ProjectionCleanupService from app.worker.runtime import WorkerRuntime logger = logging.getLogger(__name__) @@ -11,7 +13,20 @@ logger = logging.getLogger(__name__) async def serve(*, once: bool = False) -> None: settings = get_settings() - runtime = WorkerRuntime(settings=settings) + relationships = get_relationship_service() + # 在**组装层**注入投影删除客户端：记忆失效/销户时清理图库与向量库里的派生数据。 + # 放在这里而不是 runtime 内部兜底，是为了保留"未注入即显式降级并留痕"的语义 + # （有单测守着这一点），也让"生产装配了什么"在入口处一眼可见。 + # 该服务返回自己模块里的 ProjectionCleanupOutcome（字段与 runtime 的同名结构一致）， + # 结构化契约成立但名义类型不同，故显式忽略：为此把结构体抽到共享模块会造成 + # service 与 worker 两个层次互相导入，不值得。 + runtime = WorkerRuntime( + settings=settings, + relationships=relationships, + projection_cleaner=ProjectionCleanupService( # type: ignore[arg-type] + relationships=relationships + ), + ) try: while True: try: diff --git a/app/worker/runtime.py b/app/worker/runtime.py index a0770df..4a0dc8c 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -79,7 +79,7 @@ class WorkerRuntime: endpoint_resolver: ExtractionEndpointResolver | None = None, memory_cache: CacheDeleteAdapter | None = None, projection_cleaner: ProjectionCleaner | None = None, - relationships: Any | None = None, + relationships: Any = None, ) -> None: self.factory = factory if factory is not None else get_agent_factory() self.settings = settings or get_settings() @@ -97,8 +97,9 @@ class WorkerRuntime: self.memory_cache: CacheDeleteAdapter | None = ( memory_cache if memory_cache is not None else get_memory_cache_adapter() ) - # Milvus/Neo4j 删除客户端：当前组装层没有提供（bootstrap 只装配召回用的读适配器）， - # 因此默认 None = 投影清理显式降级并留痕，绝不写成"删除成功"。 + # Milvus/Neo4j 删除客户端：由**组装层**（app/worker/__main__.py）注入生产实现， + # 这里不兜底。组件内部给默认实现会把"尚未装配"这一事实悄悄盖住——而"未注入即显式 + # 降级并留痕"是本模块刻意保留的语义（有单测守着），因此默认值保持 None。 self.projection_cleaner = projection_cleaner # 图关系服务：画像投影用它写入节点与关系（投顾的多跳推荐、风控的关系网络都读它）。 # 默认取生产装配；图库不可用时该值为 None，投影如实降级而不是失败。