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，投影如实降级而不是失败。