258 lines
11 KiB
Python
258 lines
11 KiB
Python
"""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"]
|