230 lines
9.0 KiB
Python
230 lines
9.0 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.composed = composed or ComposedProfileService(
|
|
profile=self.profile,
|
|
long_term=self.long_term,
|
|
holdings=self.holdings,
|
|
redis=self.short_term.redis,
|
|
)
|
|
self.long_term = long_term or LongTermMemoryService()
|
|
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,
|
|
) -> CustomerMemoryContext:
|
|
"""召回并组装客服 Agent 当前请求的全部可用记忆。"""
|
|
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:
|
|
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__}")
|
|
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__}")
|
|
|
|
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]
|
|
|
|
holdings_summary = None
|
|
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__}")
|
|
composed_payload: dict[str, Any] | None = None
|
|
try:
|
|
composed_payload, composed_warnings = await self.composed.compose(
|
|
db,
|
|
customer_id=customer_id,
|
|
query=query,
|
|
profile=profile,
|
|
memories=ranked,
|
|
holdings_summary=holdings_summary,
|
|
)
|
|
warnings.extend(composed_warnings)
|
|
except Exception as exc:
|
|
warnings.append(f"composed_profile_failed:{type(exc).__name__}")
|
|
|
|
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"]
|