Mutual_Fund/service/memory/facade.py

"""MemoryService Facade：客服 Agent 的唯一记忆调用入口。"""

from __future__ import annotations

from contextlib import asynccontextmanager
from typing import Any

from config.database.mysql import get_session_factory
from tool.confidence_rank import FinalConfidenceRankTool

from .archive import ConversationArchiver
from .composed_profile import ComposedProfileService
from .customer_relation import CustomerRelationMemory
from .customer_product import CustomerProductMemory
from .context_builder import build_customer_memory_context
from .holdings import CustomerHoldingsMemory
from .long_term import LongTermMemoryService
from .interest_topic import InterestTopicTracker
from .profile import CustomerProfileMemory
from .schemas import CustomerMemoryContext, MemoryUnitDTO, ShortTermMessage
from .short_term import ShortTermMemory
from .work_order import WorkOrderMemory


class MemoryService:
    """协调短期、中期和长期记忆，屏蔽底层数据库细节。"""

    def __init__(
        self,
        *,
        session_factory=None,
        short_term: ShortTermMemory | None = None,
        profile: CustomerProfileMemory | None = None,
        work_orders: WorkOrderMemory | None = None,
        relations: CustomerRelationMemory | None = None,
        products: CustomerProductMemory | None = None,
        holdings: CustomerHoldingsMemory | None = None,
        composed: ComposedProfileService | None = None,
        long_term: LongTermMemoryService | None = None,
        archiver: ConversationArchiver | None = None,
        rank_tool: FinalConfidenceRankTool | None = None,
        interest_tracker: InterestTopicTracker | None = None,
    ):
        self.session_factory = session_factory or get_session_factory()
        self.short_term = short_term or ShortTermMemory()
        self.profile = profile or CustomerProfileMemory(redis=self.short_term.redis)
        self.work_orders = work_orders or WorkOrderMemory()
        self.relations = relations or CustomerRelationMemory()
        self.products = products or CustomerProductMemory()
        self.holdings = holdings or CustomerHoldingsMemory()
        self.long_term = long_term or LongTermMemoryService()
        # composed 依赖上面三个已初始化的组件：必须先赋值 self.long_term，
        # 否则默认分支会把 None 传进去（历史 bug）。
        self.composed = composed or ComposedProfileService(
            profile=self.profile,
            long_term=self.long_term,
            holdings=self.holdings,
            redis=self.short_term.redis,
        )
        self.archiver = archiver or ConversationArchiver(short_term=self.short_term)
        self.rank_tool = rank_tool or FinalConfidenceRankTool()
        self.interest_tracker = interest_tracker or InterestTopicTracker(self.short_term.redis)

    @asynccontextmanager
    async def _db(self):
        """按请求获取并释放 MySQL 会话。"""
        async with self.session_factory() as session:
            yield session

    async def recall(
        self,
        *,
        customer_id: int,
        session_id: str,
        query: str | None = None,
        limit: int = 10,
        include_long_term: bool = False,
        include_composed: bool = False,
    ) -> CustomerMemoryContext:
        """召回并组装客服 Agent 当前请求的全部可用记忆。

        默认**不含**长期记忆与最终画像：客服 Agent 的职责是知识问答/开户引导/
        公司信息，合规上不推荐产品，长期记忆（偏好标签、投资目标）对其没有
        合法用途，却要付出 memory_unit 查询 + Embedding + Milvus 检索的代价。

        长期记忆与最终画像的两个合法消费方各自显式开启：

        - 投顾 Agent：``include_long_term=True``（推荐场景需要偏好与目标）；
        - 最终画像接口（``/profile/composed``，仅员工）：直接走
          ``ComposedProfileService.compose_response``，不经本方法。

        写入路径（``save_memory`` / ``record_interest_signal``）不受影响，
        客服会话仍持续沉淀长期记忆，供投顾与画像使用。
        """
        if limit < 0:
            raise ValueError("limit 必须是非负整数")
        warnings: list[str] = []
        short_term_messages = await self.short_term.load_messages(session_id)
        warnings.extend(self.short_term.last_warnings)

        async with self._db() as db:
            if include_long_term:
                try:
                    refresh_confidence = getattr(self.long_term, "refresh_confidence", None)
                    if refresh_confidence is not None:
                        await refresh_confidence(
                            db, customer_id, limit=max(limit, 10)
                        )
                except Exception as exc:
                    warnings.append(f"confidence_refresh_failed:{type(exc).__name__}")
            profile, profile_warnings = await self.profile.get(db, customer_id)
            warnings.extend(profile_warnings)
            try:
                work_orders = await self.work_orders.list(db, customer_id)
            except Exception as exc:
                work_orders = []
                warnings.append(f"work_order_recall_failed:{type(exc).__name__}")
            try:
                customer_relations = await self.relations.list(db, customer_id)
            except Exception as exc:
                customer_relations = []
                warnings.append(f"customer_relation_recall_failed:{type(exc).__name__}")
            try:
                customer_products = await self.products.list(db, customer_id)
            except Exception as exc:
                customer_products = []
                warnings.append(f"customer_product_recall_failed:{type(exc).__name__}")

            memories: list[MemoryUnitDTO] = []
            if include_long_term:
                try:
                    memories, memory_warnings = await self.long_term.recall(
                        db, customer_id, limit=max(limit, 10), query=query
                    )
                    warnings.extend(memory_warnings)
                except Exception as exc:
                    memories = []
                    warnings.append(f"long_term_recall_failed:{type(exc).__name__}")

            holdings_summary = None
            composed_payload: dict[str, Any] | None = None
            if include_composed:
                try:
                    holdings_summary, holdings_warnings = await self.holdings.summary(
                        db, customer_id
                    )
                    warnings.extend(holdings_warnings)
                except Exception as exc:
                    warnings.append(f"holdings_recall_failed:{type(exc).__name__}")

                try:
                    # ComposedProfileService 期望 list[dict]（与 _recall_memories
                    # 的返回一致），这里传已序列化的记忆，避免在块外再触碰会话。
                    composed_payload, composed_warnings = await self.composed.compose(
                        db,
                        customer_id=customer_id,
                        query=query,
                        profile=profile,
                        memories=[
                            memory.model_dump(mode="json") for memory in memories
                        ],
                        holdings_summary=holdings_summary,
                    )
                    warnings.extend(composed_warnings)
                except Exception as exc:
                    warnings.append(f"composed_profile_failed:{type(exc).__name__}")

        ranked = self.rank_tool.rank(
            [memory.model_dump(mode="json") for memory in memories],
            top_k=limit,
        )
        ranked_memories = [MemoryUnitDTO.model_validate(item) for item in ranked]

        return build_customer_memory_context(
            customer_id=customer_id,
            session_id=session_id,
            short_term_messages=short_term_messages,
            customer_profile=profile,
            composed_profile=composed_payload,
            work_orders=work_orders,
            customer_relations=customer_relations,
            customer_products=customer_products,
            long_term_memories=ranked_memories,
            warnings=warnings,
        )

    async def append_message(
        self,
        *,
        customer_id: int,
        session_id: str,
        message: ShortTermMessage,
    ) -> list[str]:
        """写入短期消息，并统一返回降级 warnings。"""
        await self.short_term.append_message(
            session_id,
            message.role,
            message.content,
            message_id=message.message_id,
            agent_run_id=message.agent_run_id,
            tool_calls=message.tool_calls,
        )
        return list(self.short_term.last_warnings)

    async def save_memory(
        self,
        *,
        customer_id: int,
        memory: MemoryUnitDTO,
    ) -> tuple[MemoryUnitDTO, list[str]]:
        """保存长期记忆，并返回记忆结果和底层 warnings。"""
        if memory.customer_id != customer_id:
            raise ValueError("memory.customer_id 与当前客户不一致")
        async with self._db() as db:
            return await self.long_term.save(db, memory)

    async def record_interest_signal(self, *, customer_id: int, tag: str) -> tuple[int, bool]:
        """记录一次重复兴趣主题信号，达到阈值后允许写入长期记忆。"""
        return await self.interest_tracker.record(customer_id, tag)

    async def invalidate_composed(self, customer_id: int) -> list[str]:
        """失效最终画像缓存，供画像回写/持仓变更后调用。"""
        return await self.composed.invalidate(customer_id)

    async def close_session(
        self,
        *,
        customer_id: int,
        session_id: str,
        agent_run_id: str | None = None,
    ) -> list[str]:
        """归档并关闭客户会话；归档失败时保留 Redis 消息。"""
        try:
            async with self._db() as db:
                await self.archiver.archive_session(
                    db,
                    session_id=session_id,
                    user_id=customer_id,
                    customer_id=customer_id,
                    agent_type="customer",
                    agent_run_id=agent_run_id,
                )
            return []
        except Exception as exc:
            return [f"session_close_failed:{type(exc).__name__}"]


def normalize_warnings(value: Any) -> list[str]:
    """将底层异常或组件返回的提示统一为字符串列表。"""
    if value is None:
        return []
    if isinstance(value, str):
        return [value]
    return [str(item) for item in value]


__all__ = ["MemoryService", "normalize_warnings"]