diff --git a/.gitignore b/.gitignore index 0ba3157..810c1cb 100644 --- a/.gitignore +++ b/.gitignore @@ -75,4 +75,9 @@ data/output/ # 本地临时需求文档 DK2_客服Agent模块完整开发计划（v1.1）.md DK4_客服Agent需求文档（修复完善版v1.4）.md -客服Agent模块 · TODO List（最终修复版v1.2）.md \ No newline at end of file +客服Agent模块 · TODO List（最终修复版v1.2）.md +三层记忆模块需求规格说明书.md +客服Agent记忆模块TODO清单.md +客服Agent记忆模块开发计划.md +开发计划_用户画像置信度实时更新.md +客服Agent记忆模块接入说明.md \ No newline at end of file diff --git a/agent/client_agent/__init__.py b/agent/client_agent/__init__.py new file mode 100644 index 0000000..db39a6b --- /dev/null +++ b/agent/client_agent/__init__.py @@ -0,0 +1,3 @@ +"""Logged-in client Agent domain layer.""" + +package_name = "client_agent" diff --git a/agent/client_agent/context.py b/agent/client_agent/context.py new file mode 100644 index 0000000..3c3d322 --- /dev/null +++ b/agent/client_agent/context.py @@ -0,0 +1,7 @@ +"""Client Agent context aliases during the architecture bootstrap phase.""" + +from agent.customer_agent.context import RedisConversationContext + +ClientConversationContext = RedisConversationContext + +__all__ = ["ClientConversationContext"] diff --git a/agent/client_agent/session.py b/agent/client_agent/session.py new file mode 100644 index 0000000..fabca83 --- /dev/null +++ b/agent/client_agent/session.py @@ -0,0 +1,66 @@ +"""Redis sessions owned by authenticated client Agent customers.""" + +from __future__ import annotations + +import time +import uuid +from inspect import isawaitable + +from agent.customer_agent.session import SessionOwnershipError + + +async def _config(config_getter, key: str, default): + value = config_getter(key, str(default)) + if isawaitable(value): + value = await value + return type(default)(value) + + +class ClientSessionService: + """Create and verify Redis sessions bound to a customer ID.""" + + def __init__(self, redis, *, config_getter, clock=time.time): + self.redis = redis + self.config_getter = config_getter + self.clock = clock + + async def create_session(self, customer_id: int) -> str: + """Create a session whose Redis owner value is the current customer.""" + session_id = uuid.uuid4().hex + ttl = await _config(self.config_getter, "agent.customer.session.ttl", 1800) + await self.redis.set(f"session:{session_id}", str(customer_id), ex=ttl) + await self.redis.rpush(f"session:{session_id}:messages", "") + await self.redis.expire(f"session:{session_id}:messages", ttl) + return session_id + + async def verify_session_ownership(self, session_id: str, *, customer_id: int) -> None: + """Reject missing sessions and sessions owned by another customer.""" + owner = await self.redis.get(f"session:{session_id}") + if owner is None: + raise SessionOwnershipError(404, "会话不存在或已过期") + if isinstance(owner, bytes): + owner = owner.decode() + if str(owner) != str(customer_id): + raise SessionOwnershipError(403, "无权访问该会话") + + async def consume_chat_quota(self, session_id: str, *, customer_id: int) -> int | None: + """Apply the existing per-session rate limit with a customer-scoped key.""" + window = await _config( + self.config_getter, "agent.customer.rate_limit.window_sec", 60 + ) + maximum = await _config( + self.config_getter, "agent.customer.rate_limit.max_requests", 20 + ) + key = f"rate:limit:client:{customer_id}:{session_id}:chat" + now = self.clock() + await self.redis.zremrangebyscore(key, 0, (now - window) * 1000) + count = await self.redis.zcard(key) + if count >= maximum: + entries = getattr(self.redis, "sorted_sets", {}).get(key, []) + oldest = min((score for score, _ in entries), default=now * 1000) + return max(1, int((oldest / 1000 + window) - now)) + await self.redis.zadd(key, {uuid.uuid4().hex: now * 1000}) + await self.redis.expire(key, window) + return None + +__all__ = ["ClientSessionService", "SessionOwnershipError"] diff --git a/api/chat/client_agent.py b/api/chat/client_agent.py new file mode 100644 index 0000000..d8a4062 --- /dev/null +++ b/api/chat/client_agent.py @@ -0,0 +1,123 @@ +"""Logged-in client Agent HTTP endpoints. + +Authentication and customer-owned session enforcement are added in C2. +""" + +from __future__ import annotations + +import json + +from fastapi import APIRouter, Depends, Request +from fastapi.responses import JSONResponse, StreamingResponse + +from agent.client_agent.session import SessionOwnershipError +from api.deps import require_customer +from model.sys_user import SysUser +from service.customer_agent.chat import QueryTooLongError +from utils.request_id import get_request_id, new_request_id +from utils.response import fail, success + + +router = APIRouter() + + +def _runtime(request: Request): + """Return the client Agent runtime installed by the application lifespan.""" + runtime = getattr(request.app.state, "client_agent_runtime", None) + if runtime is None: + raise RuntimeError("client agent runtime is not configured") + return runtime + + +@router.post("/session/create") +async def create_session( + request: Request, + user: SysUser = Depends(require_customer), +): + """Create a client Agent session during the architecture bootstrap phase.""" + runtime = _runtime(request) + session_id = await runtime.session_service.create_session(user.id) + return success({"session_id": session_id, "customer_id": user.id}) + + +@router.post("/chat") +async def chat( + request: Request, + body: dict, + user: SysUser = Depends(require_customer), +): + """Reuse the existing customer Agent chat orchestration.""" + session_id = body.get("session_id", "") + query = body.get("query", "") + runtime = _runtime(request) + trace_id = request.headers.get("X-Trace-Id") or get_request_id() or new_request_id() + try: + await runtime.session_service.verify_session_ownership( + session_id, customer_id=user.id + ) + retry_after = await runtime.session_service.consume_chat_quota( + session_id, customer_id=user.id + ) + if retry_after is not None: + response = fail(429, "请求过于频繁", {"retry_after": retry_after}) + return JSONResponse( + status_code=429, + headers={"Retry-After": str(retry_after)}, + content=response.model_dump(), + ) + result = await runtime.agent.handle( + session_id, + query, + trace_id=trace_id, + customer_id=user.id, + ) + except SessionOwnershipError as exc: + return JSONResponse( + status_code=exc.code, + content=fail(exc.code, exc.message).model_dump(), + ) + except QueryTooLongError as exc: + return JSONResponse( + status_code=400, + content=fail(400, str(exc)).model_dump(), + ) + + async def events(): + yield f"data: {json.dumps(result, ensure_ascii=False)}\n\n" + + return StreamingResponse( + events(), media_type="text/event-stream", headers={"X-Trace-Id": trace_id} + ) + + +@router.post("/session/end") +async def end_session( + request: Request, + body: dict, + user: SysUser = Depends(require_customer), +): + """Close the client Agent session during the architecture bootstrap phase.""" + runtime = _runtime(request) + session_id = body.get("session_id", "") + try: + await runtime.session_service.verify_session_ownership( + session_id, customer_id=user.id + ) + except SessionOwnershipError as exc: + return JSONResponse( + status_code=exc.code, + content=fail(exc.code, exc.message).model_dump(), + ) + warnings = await runtime.memory_service.close_session( + customer_id=user.id, + session_id=session_id, + ) + if warnings: + return success({"session_id": session_id, "archived": False, "warnings": warnings}) + if hasattr(runtime.redis, "delete"): + await runtime.redis.delete( + f"session:{session_id}", + f"session:{session_id}:messages", + f"rate:limit:client:{user.id}:{session_id}:chat", + ) + return success({"session_id": session_id, "archived": True, "warnings": []}) diff --git a/api/deps.py b/api/deps.py index 94deffe..71c0db7 100644 --- a/api/deps.py +++ b/api/deps.py @@ -37,3 +37,10 @@ async def require_knowledge_operator(user: SysUser = Depends(get_current_user)) if user.user_type != "EMPLOYEE" or user.employee_role not in KNOWLEDGE_OPERATOR_ROLES: raise ForbiddenError("仅运营人员可以管理知识库") return user + + +async def require_customer(user: SysUser = Depends(get_current_user)) -> SysUser: + """Allow only authenticated customer accounts to use client Agent APIs.""" + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户用户可以访问 client_agent") + return user diff --git a/api/router.py b/api/router.py index e623950..bd25ed8 100644 --- a/api/router.py +++ b/api/router.py @@ -3,7 +3,7 @@ """ from fastapi import APIRouter -from api.chat import customer_agent, knowledge +from api.chat import client_agent, customer_agent, knowledge from api.routers import product, questionnaire from api.routers import account, auth, holdings, purchase, redeem @@ -14,6 +14,7 @@ api_router.include_router(holdings.router, prefix="/api", tags=["持仓"]) api_router.include_router(purchase.router, prefix="/api", tags=["交易"]) api_router.include_router(redeem.router, prefix="/api", tags=["交易"]) api_router.include_router(customer_agent.router, prefix="/api/agent/customer", tags=["客服Agent"]) +api_router.include_router(client_agent.router, prefix="/api/agent/client", tags=["ClientAgent"]) api_router.include_router(knowledge.router, prefix="/api/knowledge", tags=["知识库"]) api_router.include_router(product.router, prefix="/api", tags=["产品"]) api_router.include_router(questionnaire.router, prefix="/api", tags=["问卷"]) diff --git a/main.py b/main.py index 83d527c..dab8c25 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ from service.customer_agent.bootstrap import ( build_default_knowledge_upload_service, build_default_runtime, ) +from service.client_agent.bootstrap import build_default_runtime as build_client_runtime from utils.exceptions import register_exception_handlers from utils.logger import setup_logging from utils.request_id import RequestIdMiddleware @@ -21,6 +22,7 @@ async def lifespan(app: FastAPI): setup_logging() await ensure_collections() app.state.customer_agent_runtime = build_default_runtime() + app.state.client_agent_runtime = build_client_runtime() app.state.knowledge_upload_service = build_default_knowledge_upload_service() yield await database.dispose() @@ -35,4 +37,4 @@ app.include_router(api_router) @app.get("/") async def root(): - return {"message": "智能公募基金系统 API", "docs": "/docs"} \ No newline at end of file + return {"message": "智能公募基金系统 API", "docs": "/docs"} diff --git a/model/biz_work_order.py b/model/biz_work_order.py new file mode 100644 index 0000000..10a50d8 --- /dev/null +++ b/model/biz_work_order.py @@ -0,0 +1,37 @@ +"""biz_work_order 工单表 ORM 模型。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from sqlalchemy import BigInteger, DateTime, JSON, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class BizWorkOrder(Base): + """客户工单状态记录。""" + + __tablename__ = "biz_work_order" + __table_args__ = {"comment": "通用业务工单表"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + work_order_no: Mapped[str] = mapped_column(String(32), nullable=False) + order_type: Mapped[str] = mapped_column(String(32), nullable=False) + sub_type: Mapped[str | None] = mapped_column(String(32), nullable=True) + customer_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + submitter_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + handler_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + current_node: Mapped[str] = mapped_column(String(32), nullable=False) + priority: Mapped[str] = mapped_column(String(8), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + biz_content: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + create_time: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + update_time: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now(), nullable=False + ) + diff --git a/model/conversation_archive.py b/model/conversation_archive.py new file mode 100644 index 0000000..66d6c5b --- /dev/null +++ b/model/conversation_archive.py @@ -0,0 +1,40 @@ +"""conversation_archive 会话归档表 ORM 模型。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from sqlalchemy import BigInteger, DateTime, Index, JSON, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class ConversationArchive(Base): + """客服会话消息归档记录。""" + + __tablename__ = "conversation_archive" + __table_args__ = ( + UniqueConstraint("session_id", "message_id", name="uk_session_message"), + Index("idx_session", "session_id"), + Index("idx_user_time", "user_id", "create_time"), + Index("idx_agent", "agent_type"), + Index("idx_agent_run", "agent_run_id"), + {"comment": "会话归档表（审计回溯 + Agent 持续学习素材）"}, + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + session_id: Mapped[str] = mapped_column(String(64), nullable=False) + customer_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + agent_type: Mapped[str] = mapped_column(String(32), nullable=False) + role: Mapped[str] = mapped_column(String(16), nullable=False) + content: Mapped[str | None] = mapped_column(Text, nullable=True) + tool_calls: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON, nullable=True) + message_id: Mapped[str] = mapped_column(String(64), nullable=False) + agent_run_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + trace_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + create_time: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) diff --git a/model/customer_relation.py b/model/customer_relation.py new file mode 100644 index 0000000..7ab50cd --- /dev/null +++ b/model/customer_relation.py @@ -0,0 +1,27 @@ +"""customer_relation 客户-投顾关系表 ORM 模型。""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, String +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class CustomerRelation(Base): + """客户与投顾的当前或历史关系。""" + + __tablename__ = "customer_relation" + __table_args__ = {"comment": "客户-投顾关系表"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + advisor_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + assign_time: Mapped[datetime] = mapped_column(DateTime, nullable=False) + signed_time: Mapped[datetime | None] = mapped_column(DateTime) + end_time: Mapped[datetime | None] = mapped_column(DateTime) + status: Mapped[str] = mapped_column(String(16), nullable=False) + reason: Mapped[str | None] = mapped_column(String(128)) + diff --git a/model/memory_unit.py b/model/memory_unit.py new file mode 100644 index 0000000..50b88a6 --- /dev/null +++ b/model/memory_unit.py @@ -0,0 +1,56 @@ +"""memory_unit 客户长期记忆主体表 ORM 模型。""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import BigInteger, DateTime, Integer, JSON, Numeric, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class MemoryUnit(Base): + """客户长期记忆主体，向量和关系索引保存在外部库。""" + + __tablename__ = "memory_unit" + __table_args__ = {"comment": "客户长期记忆主体表"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + session_id: Mapped[str | None] = mapped_column(String(64)) + agent_run_id: Mapped[str | None] = mapped_column(String(64)) + memory_type: Mapped[str] = mapped_column(String(32), nullable=False) + tag: Mapped[str] = mapped_column(String(64), nullable=False) + content: Mapped[str] = mapped_column(String(512), nullable=False) + info_type: Mapped[str] = mapped_column(String(8), nullable=False) + source: Mapped[str] = mapped_column(String(32), nullable=False) + evidence_ref: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) + source_confidence: Mapped[Decimal] = mapped_column(Numeric(5, 2), server_default="0.20") + confidence: Mapped[Decimal] = mapped_column(Numeric(5, 2), server_default="0.20") + historical_accuracy: Mapped[Decimal] = mapped_column(Numeric(5, 2), server_default="0.50") + confidence_version: Mapped[str | None] = mapped_column(String(32)) + confidence_reason: Mapped[str | None] = mapped_column(String(255)) + confidence_update_time: Mapped[datetime | None] = mapped_column(DateTime) + evidence_count: Mapped[int] = mapped_column(Integer, server_default="0") + conflict_count: Mapped[int] = mapped_column(Integer, server_default="0") + recall_count: Mapped[int] = mapped_column(Integer, server_default="0") + memory_version: Mapped[int] = mapped_column(Integer, server_default="1") + update_time: Mapped[datetime | None] = mapped_column(DateTime) + last_recall_time: Mapped[datetime | None] = mapped_column(DateTime) + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + status: Mapped[str] = mapped_column(String(16), server_default="candidate") + valid_from: Mapped[datetime | None] = mapped_column(DateTime) + valid_until: Mapped[datetime | None] = mapped_column(DateTime) + last_verified_at: Mapped[datetime | None] = mapped_column(DateTime) + milvus_id: Mapped[str | None] = mapped_column(String(128)) + graph_node_id: Mapped[str | None] = mapped_column(String(128)) + milvus_sync_status: Mapped[str] = mapped_column(String(16), server_default="pending") + neo4j_sync_status: Mapped[str] = mapped_column(String(16), server_default="pending") + sync_retry_count: Mapped[int] = mapped_column(Integer, server_default="0") + last_sync_error: Mapped[str | None] = mapped_column(String(500)) + next_retry_at: Mapped[datetime | None] = mapped_column(DateTime) + last_synced_at: Mapped[datetime | None] = mapped_column(DateTime) + diff --git a/repositories/conversation_archive.py b/repositories/conversation_archive.py new file mode 100644 index 0000000..1a5f929 --- /dev/null +++ b/repositories/conversation_archive.py @@ -0,0 +1,68 @@ +"""conversation_archive 会话归档仓储。""" + +from __future__ import annotations + +from collections.abc import Iterable + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError + +from model.conversation_archive import ConversationArchive +from repositories.base import BaseRepository + + +class ConversationArchiveRepo(BaseRepository): + """提供按会话读取和幂等批量归档能力。""" + + model = ConversationArchive + + async def list_by_session(self, session_id: str) -> list[ConversationArchive]: + """按消息创建顺序读取完整归档会话。""" + statement = ( + select(ConversationArchive) + .where(ConversationArchive.session_id == session_id) + .order_by(ConversationArchive.create_time, ConversationArchive.id) + ) + return list((await self.db.scalars(statement)).all()) + + async def archive_batch(self, rows: Iterable[dict]) -> int: + """批量写入归档记录，重复的 session_id + message_id 自动跳过。""" + rows = list(rows) + if not rows: + return 0 + + session_id = rows[0]["session_id"] + message_ids = [row["message_id"] for row in rows] + existing = await self.db.scalars( + select(ConversationArchive.message_id).where( + ConversationArchive.session_id == session_id, + ConversationArchive.message_id.in_(message_ids), + ) + ) + existing_ids = set(existing.all()) + pending = [row for row in rows if row["message_id"] not in existing_ids] + if not pending: + return 0 + + self.db.add_all( + [ConversationArchive(**row) for row in pending] + ) + try: + await self.db.commit() + except IntegrityError: + await self.db.rollback() + # 并发归档时，唯一键冲突代表其他调用已完成该消息归档。 + remaining = await self.db.scalars( + select(ConversationArchive.message_id).where( + ConversationArchive.session_id == session_id, + ConversationArchive.message_id.in_(message_ids), + ) + ) + if set(remaining.all()) >= set(message_ids): + return 0 + raise + return len(pending) + + +__all__ = ["ConversationArchiveRepo"] + diff --git a/repositories/customer_profile_change_log.py b/repositories/customer_profile_change_log.py new file mode 100644 index 0000000..c0b9a6f --- /dev/null +++ b/repositories/customer_profile_change_log.py @@ -0,0 +1,57 @@ +"""客户画像变更日志仓储。""" + +from __future__ import annotations + +import json + +from sqlalchemy import text + + +class CustomerProfileChangeLogRepo: + """记录画像更新原因，供缓存失效和后续审计复用。""" + + def __init__(self, db): + self.db = db + + async def record( + self, + *, + customer_id: int, + tag: str | None, + old_value, + new_value, + source: str | None, + confidence: float | None, + reason: str | None, + operator_id: int | None = None, + ) -> None: + """写入一次画像变更日志。""" + await self.db.execute( + text( + """ + INSERT INTO customer_profile_change_log + (customer_id, tag, old_value, new_value, source, + confidence, reason, operator_id) + VALUES + (:customer_id, :tag, :old_value, :new_value, :source, + :confidence, :reason, :operator_id) + """ + ), + { + "customer_id": customer_id, + "tag": tag, + "old_value": json.dumps(old_value, ensure_ascii=False) + if isinstance(old_value, (dict, list)) else old_value, + "new_value": json.dumps(new_value, ensure_ascii=False) + if isinstance(new_value, (dict, list)) else new_value, + "source": source, + "confidence": confidence, + "reason": reason, + "operator_id": operator_id, + }, + ) + await self.db.commit() + + +__all__ = ["CustomerProfileChangeLogRepo"] + diff --git a/repositories/customer_relation.py b/repositories/customer_relation.py new file mode 100644 index 0000000..17ef156 --- /dev/null +++ b/repositories/customer_relation.py @@ -0,0 +1,28 @@ +"""客户-投顾关系查询仓储。""" + +from sqlalchemy import select + +from model.customer_relation import CustomerRelation +from repositories.base import BaseRepository + + +class CustomerRelationRepo(BaseRepository): + """按客户 ID 隔离读取有效关系。""" + + model = CustomerRelation + + async def list_by_customer(self, customer_id: int) -> list[CustomerRelation]: + """返回指定客户尚未结束的关系。""" + statement = ( + select(CustomerRelation) + .where( + CustomerRelation.customer_id == customer_id, + CustomerRelation.status != "已结束", + ) + .order_by(CustomerRelation.assign_time.desc(), CustomerRelation.id.desc()) + ) + return list((await self.db.scalars(statement)).all()) + + +__all__ = ["CustomerRelationRepo"] + diff --git a/repositories/fin_holdings.py b/repositories/fin_holdings.py index 253a2a9..260f5fa 100644 --- a/repositories/fin_holdings.py +++ b/repositories/fin_holdings.py @@ -7,6 +7,7 @@ from sqlalchemy import case, select, update from sqlalchemy.dialects.mysql import insert as mysql_insert from model.fin_holdings import FinHoldings +from model.fin_product import FinProduct from repositories.base import BaseRepository @@ -23,6 +24,20 @@ class FinHoldingsRepo(BaseRepository): stmt = stmt.order_by(FinHoldings.id) return list((await self.db.scalars(stmt)).all()) + async def list_with_products( + self, customer_id: int, *, include_closed: bool = True + ) -> list[tuple[FinHoldings, FinProduct | None]]: + """按客户关联查询持仓和基金产品，产品缺失时保留持仓记录。""" + stmt = ( + select(FinHoldings, FinProduct) + .outerjoin(FinProduct, FinProduct.id == FinHoldings.product_id) + .where(FinHoldings.customer_id == customer_id) + .order_by(FinHoldings.update_time.desc(), FinHoldings.id.desc()) + ) + if not include_closed: + stmt = stmt.where(FinHoldings.status == "持有中") + return list((await self.db.execute(stmt)).all()) + async def get_by_customer_product( self, customer_id: int, product_id: int ) -> FinHoldings | None: diff --git a/repositories/memory_unit.py b/repositories/memory_unit.py new file mode 100644 index 0000000..08128ca --- /dev/null +++ b/repositories/memory_unit.py @@ -0,0 +1,107 @@ +"""memory_unit 长期记忆仓储。""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import select + +from model.memory_unit import MemoryUnit +from repositories.base import BaseRepository + + +ACTIVE_STATUSES = ("candidate", "confirmed") +INACTIVE_STATUSES = ("expired", "rejected", "archived") + + +class MemoryUnitRepo(BaseRepository): + """负责 MySQL 主体记忆的隔离、去重和同步状态管理。""" + + model = MemoryUnit + + async def add_memory(self, values: dict) -> MemoryUnit: + """先写入 MySQL 主体并返回数据库 ID。""" + obj = MemoryUnit(**values) + self.db.add(obj) + await self.db.commit() + await self.db.refresh(obj) + return obj + + async def find_exact( + self, customer_id: int, memory_type: str, tag: str, content: str + ) -> MemoryUnit | None: + """按客户、类型、标签和内容精确查找去重对象。""" + return await self.db.scalar( + select(MemoryUnit).where( + MemoryUnit.customer_id == customer_id, + MemoryUnit.memory_type == memory_type, + MemoryUnit.tag == tag, + MemoryUnit.content == content, + MemoryUnit.status.not_in(INACTIVE_STATUSES), + ) + ) + + async def list_for_customer( + self, + customer_id: int, + *, + memory_type: str | None = None, + tag: str | None = None, + now: datetime | None = None, + limit: int = 100, + ) -> list[MemoryUnit]: + """查询默认可召回的客户记忆，过滤拒绝、归档和过期记录。""" + now = now or datetime.now() + conditions = [ + MemoryUnit.customer_id == customer_id, + MemoryUnit.status.in_(ACTIVE_STATUSES), + (MemoryUnit.valid_until.is_(None) | (MemoryUnit.valid_until > now)), + ] + if memory_type: + conditions.append(MemoryUnit.memory_type == memory_type) + if tag: + conditions.append(MemoryUnit.tag == tag) + statement = ( + select(MemoryUnit) + .where(*conditions) + .order_by(MemoryUnit.update_time.desc(), MemoryUnit.id.desc()) + .limit(limit) + ) + return list((await self.db.scalars(statement)).all()) + + async def merge_evidence( + self, memory: MemoryUnit, *, evidence_count: int = 1, conflict_count: int = 0 + ) -> MemoryUnit: + """合并证据和冲突计数，不改变客户隔离范围。""" + memory.evidence_count = (memory.evidence_count or 0) + evidence_count + memory.conflict_count = (memory.conflict_count or 0) + conflict_count + memory.update_time = datetime.now() + await self.db.commit() + await self.db.refresh(memory) + return memory + + async def update_sync_status(self, memory_id: int, **values) -> None: + """更新向量/图谱索引 ID 和同步状态。""" + memory = await self.db.get(MemoryUnit, memory_id) + if memory is None: + return + for key, value in values.items(): + setattr(memory, key, value) + memory.update_time = datetime.now() + await self.db.commit() + + async def list_pending_sync(self, limit: int = 100) -> list[MemoryUnit]: + """返回存在失败或空索引的记忆，供重试入口使用。""" + statement = ( + select(MemoryUnit) + .where( + (MemoryUnit.milvus_id.is_(None) | (MemoryUnit.milvus_sync_status == "failed")) + | (MemoryUnit.graph_node_id.is_(None) | (MemoryUnit.neo4j_sync_status == "failed")) + ) + .order_by(MemoryUnit.id) + .limit(limit) + ) + return list((await self.db.scalars(statement)).all()) + + +__all__ = ["MemoryUnitRepo"] diff --git a/repositories/work_order.py b/repositories/work_order.py new file mode 100644 index 0000000..6adf0f5 --- /dev/null +++ b/repositories/work_order.py @@ -0,0 +1,34 @@ +"""客户工单查询仓储。""" + +from __future__ import annotations + +from sqlalchemy import select + +from model.biz_work_order import BizWorkOrder +from repositories.base import BaseRepository + + +class WorkOrderRepo(BaseRepository): + """按客户隔离查询工单状态。""" + + model = BizWorkOrder + COMPLETED_STATUSES = ("已完成", "已驳回") + + async def list_by_customer( + self, customer_id: int, *, active_only: bool = True, limit: int = 100 + ) -> list[BizWorkOrder]: + """按更新时间倒序返回指定客户的工单。""" + conditions = [BizWorkOrder.customer_id == customer_id] + if active_only: + conditions.append(BizWorkOrder.status.not_in(self.COMPLETED_STATUSES)) + statement = ( + select(BizWorkOrder) + .where(*conditions) + .order_by(BizWorkOrder.update_time.desc(), BizWorkOrder.id.desc()) + .limit(limit) + ) + return list((await self.db.scalars(statement)).all()) + + +__all__ = ["WorkOrderRepo"] + diff --git a/service/client_agent/__init__.py b/service/client_agent/__init__.py new file mode 100644 index 0000000..0b67223 --- /dev/null +++ b/service/client_agent/__init__.py @@ -0,0 +1,3 @@ +"""Logged-in client Agent service layer.""" + +package_name = "client_agent" diff --git a/service/client_agent/bootstrap.py b/service/client_agent/bootstrap.py new file mode 100644 index 0000000..eec3b6a --- /dev/null +++ b/service/client_agent/bootstrap.py @@ -0,0 +1,27 @@ +"""Application wiring for the client Agent.""" + +from __future__ import annotations + +from config.database.milvus import client as milvus_client +from config.database.mysql import get_session_factory +from config.database.redis import client as redis_client +from service.client_agent.runtime import build_client_runtime +from service.customer_agent.config import DatabaseConfigProvider +from tool.llm import llm as llm_client + + +def build_default_runtime(): + """Build the client Agent runtime using project-wide dependencies.""" + provider = DatabaseConfigProvider(session_factory=get_session_factory()) + return build_client_runtime( + redis=redis_client(), + milvus_client=milvus_client(), + llm_client=llm_client, + config_getter=provider.get, + audit_writer=provider.write_audit, + ) + + +def get_memory_service(runtime): + """返回 client_agent 使用的统一 MemoryService 入口。""" + return runtime.memory_service diff --git a/service/client_agent/memory_extractor.py b/service/client_agent/memory_extractor.py new file mode 100644 index 0000000..6b5c947 --- /dev/null +++ b/service/client_agent/memory_extractor.py @@ -0,0 +1,74 @@ +"""从客服对话中提取长期记忆候选。""" + +from __future__ import annotations + +import json +import re +from typing import Any + +from service.memory.schemas import MemorySource, MemoryType + + +class DialogueMemoryExtractor: + """使用项目统一 LLM 提取结构化客户记忆候选。""" + + SYSTEM_PROMPT = """ +你是客服记忆提取器，只提取用户明确表达或稳定陈述的客户信息。 +只返回 JSON 数组，不要输出 Markdown 或解释文字。 +每项必须包含：tag、content、memory_type、source。 +memory_type 只能是 PROFILE_FACT、PROFILE_CANDIDATE、CUSTOMER_PREFERENCE、INVESTMENT_GOAL、SERVICE_FACT。 +source 只能是 dialogue_confirmed、dialogue_stated、dialogue_inferred。 +客服知识问题、产品政策、寒暄、客服回复内容不要提取。 +不确定的信息使用 dialogue_inferred，无法形成客户画像的信息不要提取。 +""".strip() + + def __init__(self, llm_client): + self.llm_client = llm_client + + async def extract(self, query: str, *, context: dict[str, Any] | None = None) -> list[dict[str, str]]: + """提取并校验本轮用户消息中的客户记忆候选。""" + prompt = [{"role": "system", "content": self.SYSTEM_PROMPT}] + prompt.append( + { + "role": "user", + "content": json.dumps( + {"query": query, "existing_memory": context or {}}, + ensure_ascii=False, + ), + } + ) + response = await self.llm_client.chat(prompt, temperature=0, max_tokens=800) + return self._parse(response) + + @staticmethod + def _parse(response: str) -> list[dict[str, str]]: + """解析 LLM JSON，并过滤不符合记忆契约的内容。""" + text = response.strip() + fenced = re.search(r"```(?:json)?\s*(.*?)\s*```", text, re.S | re.I) + if fenced: + text = fenced.group(1) + data = json.loads(text) + if not isinstance(data, list): + raise ValueError("记忆提取结果必须是数组") + valid_types = {item.value for item in MemoryType} + valid_sources = {item.value for item in MemorySource} + result = [] + for item in data: + if not isinstance(item, dict): + continue + if not all(item.get(key) for key in ("tag", "content", "memory_type", "source")): + continue + if item["memory_type"] not in valid_types or item["source"] not in valid_sources: + continue + result.append( + { + "tag": str(item["tag"])[:64], + "content": str(item["content"])[:512], + "memory_type": item["memory_type"], + "source": item["source"], + } + ) + return result + + +__all__ = ["DialogueMemoryExtractor"] diff --git a/service/client_agent/runtime.py b/service/client_agent/runtime.py new file mode 100644 index 0000000..15f911e --- /dev/null +++ b/service/client_agent/runtime.py @@ -0,0 +1,200 @@ +"""Runtime assembly for the client Agent.""" + +from __future__ import annotations + +import contextvars +import uuid +from types import SimpleNamespace + +from agent.client_agent.session import ClientSessionService +from rag.embedding import embed_texts +from rag.generation import generate_answer +from rag.intent import intent_recognize +from rag.retrieve import rag_retrieve +from service.customer_agent.chat import AnonymousCustomerAgent +from service.client_agent.memory_extractor import DialogueMemoryExtractor +from service.memory.facade import MemoryService +from service.memory.schemas import CustomerMemoryContext, MemoryUnitDTO, ShortTermMessage + + +_active_customer = contextvars.ContextVar("client_agent_customer", default=None) +_active_messages = contextvars.ContextVar("client_agent_messages", default=None) +_active_warnings = contextvars.ContextVar("client_agent_memory_warnings", default=None) +_active_memory_context = contextvars.ContextVar("client_agent_memory_context", default=None) + + +class MemoryConversationContext: + """将现有客服 Agent 的上下文接口桥接到 MemoryService 短期记忆。""" + + def __init__(self, memory_service: MemoryService): + self.memory_service = memory_service + + async def append(self, session_id: str, role: str, content: str) -> None: + """通过统一记忆入口写入消息，并更新当前请求上下文。""" + customer_id = _active_customer.get() + if customer_id is None: + raise RuntimeError("client_agent customer context is missing") + message = ShortTermMessage( + message_id=uuid.uuid4().hex, + session_id=session_id, + role=role, + content=content, + ) + warnings = await self.memory_service.append_message( + customer_id=customer_id, + session_id=session_id, + message=message, + ) + _active_warnings.get().extend(warnings) + messages = _active_messages.get() + if messages is not None: + messages.append({"role": role, "content": content}) + + async def get(self, _session_id: str) -> list[dict]: + """返回本轮召回上下文加上本轮新增消息。""" + return list(_active_messages.get() or []) + + +class MemoryAwareClientAgent: + """在现有客服 Agent 外包裹记忆召回、候选保存和降级处理。""" + + def __init__(self, *, agent, memory_service, context, extractor=None): + self.agent = agent + self.memory_service = memory_service + self.context = context + self.extractor = extractor + + async def handle(self, session_id: str, query: str, *, trace_id: str, customer_id: int) -> dict: + """执行记忆召回、客服回答、消息写入和候选记忆保存。""" + warnings: list[str] = [] + try: + memory_context = await self.memory_service.recall( + customer_id=customer_id, + session_id=session_id, + query=query, + ) + warnings.extend(memory_context.warnings) + except Exception as exc: + memory_context = CustomerMemoryContext( + customer_id=customer_id, + session_id=session_id, + ) + warnings.append(f"memory_recall_failed:{type(exc).__name__}") + + message_token = _active_customer.set(customer_id) + messages_token = _active_messages.set( + [{"role": item.role, "content": item.content} for item in memory_context.short_term_messages] + ) + warnings_token = _active_warnings.set(warnings) + context_token = _active_memory_context.set(memory_context) + try: + result = await self.agent.handle(session_id, query, trace_id=trace_id) + if self.extractor is not None: + await self._save_candidates(customer_id, session_id, query, memory_context, warnings) + result["memory_warnings"] = list(warnings) + return result + finally: + _active_customer.reset(message_token) + _active_messages.reset(messages_token) + _active_warnings.reset(warnings_token) + _active_memory_context.reset(context_token) + + async def _save_candidates(self, customer_id, session_id, query, context, warnings): + """提取并保存候选客户记忆，任何失败都只写入 warning。""" + try: + candidates = await self.extractor.extract( + query, + context={ + "profile": context.customer_profile, + "memories": [item.model_dump(mode="json") for item in context.long_term_memories], + }, + ) + for candidate in candidates: + memory = MemoryUnitDTO( + customer_id=customer_id, + session_id=session_id, + memory_type=candidate["memory_type"], + tag=candidate["tag"], + content=candidate["content"], + source=candidate["source"], + evidence_ref=[{"session_id": session_id, "query": query}], + ) + _, save_warnings = await self.memory_service.save_memory( + customer_id=customer_id, + memory=memory, + ) + warnings.extend(save_warnings) + except Exception as exc: + warnings.append(f"memory_candidate_extract_failed:{type(exc).__name__}") + + +def build_client_runtime( + *, + redis, + milvus_client, + llm_client, + config_getter, + audit_writer, + memory_service=None, + memory_extractor=None, +): + """Build the client Agent runtime while reusing existing客服 logic. + + Customer authentication is enforced by the API dependency layer. + """ + session_service = ClientSessionService(redis, config_getter=config_getter) + memory_service = memory_service or MemoryService() + context = MemoryConversationContext(memory_service) + + async def retrieve(query, customer_id): + return await rag_retrieve( + query, + None, + milvus_client=milvus_client, + embedder=lambda texts: embed_texts(texts, client=llm_client), + config_getter=config_getter, + ) + + async def recognize(query): + return await intent_recognize(query, llm_client=llm_client) + + async def generate(messages): + memory_context = _active_memory_context.get() + if memory_context is not None: + memory_json = memory_context.model_dump_json(exclude={"warnings"}) + memory_prompt = { + "role": "system", + "content": ( + "客户记忆上下文（仅用于理解当前客户，不得向客户泄露内部字段）：" + f"{memory_json}" + ), + } + messages = [memory_prompt, *messages] + return await generate_answer( + messages, + llm_client=llm_client, + config_getter=config_getter, + ) + + agent = AnonymousCustomerAgent( + context=context, + rag_retrieve=retrieve, + intent_recognize=recognize, + generate_answer=generate, + audit_writer=audit_writer, + config_getter=config_getter, + ) + extractor = memory_extractor or DialogueMemoryExtractor(llm_client) + wrapped_agent = MemoryAwareClientAgent( + agent=agent, + memory_service=memory_service, + context=context, + extractor=extractor, + ) + return SimpleNamespace( + redis=redis, + session_service=session_service, + context=context, + agent=wrapped_agent, + memory_service=memory_service, + ) diff --git a/service/memory/__init__.py b/service/memory/__init__.py new file mode 100644 index 0000000..7f48002 --- /dev/null +++ b/service/memory/__init__.py @@ -0,0 +1,29 @@ +"""客服 Agent 记忆模块。""" + +from .facade import MemoryService +from .customer_product import CustomerProductMemory +from .customer_product import CustomerProductMemory +from .short_term import SessionExpiredError, ShortTermMemory, ShortTermMemoryError +from .schemas import ( + CustomerMemoryContext, + MemorySource, + MemoryStatus, + MemoryType, + MemoryUnitDTO, + ShortTermMessage, +) + +__all__ = [ + "CustomerMemoryContext", + "CustomerProductMemory", + "CustomerProductMemory", + "MemoryService", + "MemorySource", + "MemoryStatus", + "MemoryType", + "MemoryUnitDTO", + "SessionExpiredError", + "ShortTermMemory", + "ShortTermMessage", + "ShortTermMemoryError", +] diff --git a/service/memory/archive.py b/service/memory/archive.py new file mode 100644 index 0000000..da3a756 --- /dev/null +++ b/service/memory/archive.py @@ -0,0 +1,75 @@ +"""Redis 短期消息到 MySQL conversation_archive 的归档服务。""" + +from __future__ import annotations + +import re +from typing import Any + +from repositories.conversation_archive import ConversationArchiveRepo +from service.memory.short_term import ShortTermMemory + + +class ConversationArchiver: + """处理会话读取、脱敏、批量写入和成功后清理。""" + + def __init__( + self, + *, + short_term: ShortTermMemory, + repository_factory=ConversationArchiveRepo, + ): + self.short_term = short_term + self.repository_factory = repository_factory + + async def archive_session( + self, + db, + *, + session_id: str, + user_id: int, + customer_id: int | None, + agent_type: str = "customer", + trace_id: str | None = None, + agent_run_id: str | None = None, + ) -> int: + """归档完整会话；只有数据库成功后才清理 Redis。""" + messages = await self.short_term.load_messages(session_id) + rows = [ + { + "session_id": session_id, + "customer_id": customer_id, + "user_id": user_id, + "agent_type": agent_type, + "role": message.role, + "content": self.redact(message.content), + "tool_calls": self.redact(message.tool_calls), + "message_id": message.message_id, + "agent_run_id": message.agent_run_id or agent_run_id, + "trace_id": trace_id, + } + for message in messages + ] + archived_count = await self.repository_factory(db).archive_batch(rows) + await self.short_term.clear_session(session_id) + return archived_count + + @staticmethod + def redact(value: Any) -> Any: + """脱敏手机号、邮箱和常见身份证号，保留消息可读性。""" + if isinstance(value, str): + value = re.sub(r"(? CustomerMemoryContext: + """构造稳定的客服记忆上下文结构。""" + return CustomerMemoryContext( + customer_id=customer_id, + session_id=session_id, + short_term_messages=short_term_messages, + customer_profile=customer_profile, + work_orders=work_orders, + long_term_memories=long_term_memories, + customer_relations=customer_relations or [], + customer_products=customer_products or [], + warnings=warnings or [], + ) + + +__all__ = ["build_customer_memory_context"] diff --git a/service/memory/customer_product.py b/service/memory/customer_product.py new file mode 100644 index 0000000..4c5d2e0 --- /dev/null +++ b/service/memory/customer_product.py @@ -0,0 +1,52 @@ +"""客户-产品关系中期记忆读取。""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from repositories.fin_holdings import FinHoldingsRepo + + +class CustomerProductMemory: + """读取客户持仓及其关联基金产品，供客服上下文使用。""" + + def __init__(self, *, repository_factory=FinHoldingsRepo): + self.repository_factory = repository_factory + + async def list( + self, db, customer_id: int, *, include_closed: bool = True + ) -> list[dict[str, Any]]: + """按客户 ID 返回持仓与产品信息，避免跨客户读取。""" + holdings = await self.repository_factory(db).list_with_products( + customer_id, include_closed=include_closed + ) + return [self._to_dict(holding, product) for holding, product in holdings] + + @staticmethod + def _to_dict(holding, product) -> dict[str, Any]: + """将持仓和产品 ORM 对象转换为上下文可用的 JSON 友好结构。""" + def scalar(value): + return str(value) if isinstance(value, Decimal) else value + + return { + "holding_id": holding.id, + "customer_id": holding.customer_id, + "product_id": holding.product_id, + "shares": scalar(holding.shares), + "cost_amount": scalar(holding.cost_amount), + "current_value": scalar(holding.current_value), + "profit_loss": scalar(holding.profit_loss), + "profit_ratio": scalar(holding.profit_ratio), + "holding_status": holding.status, + "holding_create_time": holding.create_time, + "holding_update_time": holding.update_time, + "product_code": product.product_code if product else None, + "product_name": product.product_name if product else None, + "product_type": product.product_type if product else None, + "risk_level": product.risk_level if product else None, + "product_status": product.status if product else None, + } + + +__all__ = ["CustomerProductMemory"] diff --git a/service/memory/customer_relation.py b/service/memory/customer_relation.py new file mode 100644 index 0000000..3b25c8e --- /dev/null +++ b/service/memory/customer_relation.py @@ -0,0 +1,30 @@ +"""客户关系中期记忆读取。""" + +from repositories.customer_relation import CustomerRelationRepo + + +class CustomerRelationMemory: + """读取客服上下文需要的客户-投顾关系。""" + + def __init__(self, *, repository_factory=CustomerRelationRepo): + self.repository_factory = repository_factory + + async def list(self, db, customer_id: int) -> list[dict]: + """按客户 ID 返回有效关系。""" + relations = await self.repository_factory(db).list_by_customer(customer_id) + return [ + { + "id": relation.id, + "customer_id": relation.customer_id, + "advisor_id": relation.advisor_id, + "assign_time": relation.assign_time, + "signed_time": relation.signed_time, + "end_time": relation.end_time, + "status": relation.status, + "reason": relation.reason, + } + for relation in relations + ] + + +__all__ = ["CustomerRelationMemory"] diff --git a/service/memory/facade.py b/service/memory/facade.py new file mode 100644 index 0000000..0ada857 --- /dev/null +++ b/service/memory/facade.py @@ -0,0 +1,175 @@ +"""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 .customer_relation import CustomerRelationMemory +from .customer_product import CustomerProductMemory +from .context_builder import build_customer_memory_context +from .long_term import LongTermMemoryService +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, + long_term: LongTermMemoryService | None = None, + archiver: ConversationArchiver | None = None, + rank_tool: FinalConfidenceRankTool | 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.long_term = long_term or LongTermMemoryService() + self.archiver = archiver or ConversationArchiver(short_term=self.short_term) + self.rank_tool = rank_tool or FinalConfidenceRankTool() + + @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: + 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) + ) + 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] + return build_customer_memory_context( + customer_id=customer_id, + session_id=session_id, + short_term_messages=short_term_messages, + customer_profile=profile, + 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 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"] diff --git a/service/memory/long_term.py b/service/memory/long_term.py new file mode 100644 index 0000000..f49324f --- /dev/null +++ b/service/memory/long_term.py @@ -0,0 +1,193 @@ +"""MySQL + Milvus + Neo4j 客户长期记忆同步服务。""" + +from __future__ import annotations + +from datetime import datetime +from inspect import isawaitable +from typing import Any, Callable + +from tool.llm import llm +from tool.confidence import BaseConfidenceCalcTool + +from repositories.memory_unit import MemoryUnitRepo +from service.memory.milvus_memory import MilvusMemoryStore +from service.memory.neo4j_memory import Neo4jMemoryStore +from service.memory.schemas import MemoryUnitDTO + + +class LongTermMemoryService: + """以 MySQL 为主体事实源，向 Milvus 和 Neo4j 同步镜像。""" + + def __init__( + self, + *, + milvus_store: MilvusMemoryStore | None = None, + neo4j_store: Neo4jMemoryStore | None = None, + repository_factory=MemoryUnitRepo, + embedder: Callable[[str], Any] | None = None, + ): + self.milvus_store = milvus_store or MilvusMemoryStore() + self.neo4j_store = neo4j_store or Neo4jMemoryStore() + self.repository_factory = repository_factory + self.embedder = embedder or llm.embed_one + self.confidence_tool = BaseConfidenceCalcTool() + + async def save(self, db, memory: MemoryUnitDTO) -> tuple[MemoryUnitDTO, list[str]]: + """保存主体并尽力同步两个外部索引，返回记忆和 warnings。""" + repo = self.repository_factory(db) + existing = await repo.find_exact( + memory.customer_id, + memory.memory_type.value, + memory.tag, + memory.content, + ) + warnings: list[str] = [] + if existing is not None: + existing = await repo.merge_evidence(existing) + entity = existing + else: + values = memory.model_dump(mode="json", exclude={"id", "milvus_id", "graph_node_id"}) + values["memory_type"] = memory.memory_type.value + values["source"] = memory.source.value + confidence_result, confidence_warning = self._calculate_confidence(memory) + warnings.extend(confidence_warning) + values.update(confidence_result) + entity = await repo.add_memory(values) + + if existing is not None: + confidence_result, confidence_warning = self._calculate_confidence(entity) + warnings.extend(confidence_warning) + await repo.update_sync_status(entity.id, **confidence_result) + for key, value in confidence_result.items(): + setattr(entity, key, value) + + try: + vector = self.embedder(entity.content) + if isawaitable(vector): + vector = await vector + milvus_id = await self.milvus_store.upsert(entity, vector) + await repo.update_sync_status( + entity.id, milvus_id=milvus_id, milvus_sync_status="success" + ) + entity.milvus_id = milvus_id + entity.milvus_sync_status = "success" + except Exception as exc: + warnings.append(f"milvus_sync_failed:{type(exc).__name__}") + await repo.update_sync_status( + entity.id, + milvus_sync_status="failed", + sync_retry_count=(entity.sync_retry_count or 0) + 1, + last_sync_error=str(exc)[:500], + ) + + try: + graph_id = await self.neo4j_store.upsert(entity) + await repo.update_sync_status( + entity.id, graph_node_id=graph_id, neo4j_sync_status="success" + ) + entity.graph_node_id = graph_id + entity.neo4j_sync_status = "success" + except Exception as exc: + warnings.append(f"neo4j_sync_failed:{type(exc).__name__}") + await repo.update_sync_status( + entity.id, + neo4j_sync_status="failed", + sync_retry_count=(entity.sync_retry_count or 0) + 1, + last_sync_error=str(exc)[:500], + ) + return self._to_dto(entity), warnings + + def _calculate_confidence(self, memory) -> tuple[dict[str, Any], list[str]]: + """计算记忆置信度；异常时强制降级为候选记忆。""" + source = memory.source.value if hasattr(memory.source, "value") else memory.source + memory_type = ( + memory.memory_type.value + if hasattr(memory.memory_type, "value") + else memory.memory_type + ) + create_time = getattr(memory, "create_time", None) + age_days = max(0, (datetime.now() - create_time).days) if create_time else 0 + try: + result = self.confidence_tool.evaluate( + tag=memory.tag, + source=source, + evidence_count=memory.evidence_count or 0, + conflict_count=memory.conflict_count or 0, + age_days=age_days, + memory_type=memory_type, + ) + result.pop("age_days", None) + result.pop("threshold", None) + result["confidence_update_time"] = datetime.now() + return result, [] + except Exception as exc: + return { + "status": "candidate", + "confidence_reason": "置信度计算失败，降级保存为候选记忆", + "confidence_version": BaseConfidenceCalcTool.VERSION, + "confidence_update_time": datetime.now(), + }, [f"confidence_calculation_failed:{type(exc).__name__}"] + + async def recall( + self, + db, + customer_id: int, + *, + memory_type: str | None = None, + tag: str | None = None, + limit: int = 100, + ) -> tuple[list[MemoryUnitDTO], list[str]]: + """按客户、类型、标签和有效期召回主体记忆。""" + entities = await self.repository_factory(db).list_for_customer( + customer_id, memory_type=memory_type, tag=tag, limit=limit + ) + return [self._to_dto(entity) for entity in entities], [] + + async def retry_pending(self, db, *, limit: int = 100) -> dict[str, int]: + """重试 MySQL 中缺少外部索引或同步失败的记忆。""" + entities = await self.repository_factory(db).list_pending_sync(limit) + success = 0 + failed = 0 + for entity in entities: + dto = self._to_dto(entity) + _, warnings = await self.save(db, dto) + if warnings: + failed += 1 + else: + success += 1 + return {"success": success, "failed": failed} + + @staticmethod + def _to_dto(entity) -> MemoryUnitDTO: + """将 ORM 实体转换为跨层 DTO。""" + data = { + "id": entity.id, + "customer_id": entity.customer_id, + "session_id": entity.session_id, + "agent_run_id": entity.agent_run_id, + "memory_type": entity.memory_type, + "tag": entity.tag, + "content": entity.content, + "info_type": entity.info_type, + "source": entity.source, + "evidence_ref": entity.evidence_ref or [], + "source_confidence": float(entity.source_confidence or 0), + "confidence": float(entity.confidence or 0), + "historical_accuracy": float(entity.historical_accuracy or 0), + "confidence_version": getattr(entity, "confidence_version", None), + "confidence_reason": getattr(entity, "confidence_reason", None), + "confidence_update_time": getattr(entity, "confidence_update_time", None), + "evidence_count": entity.evidence_count or 0, + "conflict_count": entity.conflict_count or 0, + "recall_count": entity.recall_count or 0, + "status": entity.status, + "valid_from": entity.valid_from, + "valid_until": entity.valid_until, + "last_verified_at": entity.last_verified_at, + "milvus_id": entity.milvus_id, + "graph_node_id": entity.graph_node_id, + } + return MemoryUnitDTO.model_validate(data) + + +__all__ = ["LongTermMemoryService"] diff --git a/service/memory/milvus_memory.py b/service/memory/milvus_memory.py new file mode 100644 index 0000000..40a041a --- /dev/null +++ b/service/memory/milvus_memory.py @@ -0,0 +1,114 @@ +"""客户长期记忆的 Milvus 向量镜像。""" + +from __future__ import annotations + +from pymilvus import AsyncMilvusClient, DataType + +from config.database.milvus import client as configured_client +from rag.embedding import EMBEDDING_DIMENSION + + +CUSTOMER_MEMORY_COLLECTION = "customer_memory" + + +def build_memory_schema(): + """构造客户记忆向量集合结构，维度来自 LLM_EMBED_DIMENSIONS。""" + schema = AsyncMilvusClient.create_schema(auto_id=False, enable_dynamic_field=False) + schema.add_field("memory_id", DataType.VARCHAR, is_primary=True, max_length=128) + schema.add_field("customer_id", DataType.INT64) + schema.add_field("memory_type", DataType.VARCHAR, max_length=32) + schema.add_field("tag", DataType.VARCHAR, max_length=64) + schema.add_field("content", DataType.VARCHAR, max_length=2048) + schema.add_field("status", DataType.VARCHAR, max_length=16) + schema.add_field("vector", DataType.FLOAT_VECTOR, dim=EMBEDDING_DIMENSION) + return schema + + +def build_memory_index_params(): + """构造客户记忆向量索引。""" + params = AsyncMilvusClient.prepare_index_params() + params.add_index( + field_name="vector", + index_type="HNSW", + metric_type="COSINE", + params={"M": 16, "efConstruction": 200}, + ) + return params + + +class MilvusMemoryStore: + """封装客户记忆向量写入、查询和删除。""" + + def __init__(self, client=None, *, collection_name=CUSTOMER_MEMORY_COLLECTION): + self.client = client or configured_client() + self.collection_name = collection_name + + async def ensure_collection(self) -> None: + """创建集合或校验已有集合的向量维度。""" + if not await self.client.has_collection(self.collection_name): + await self.client.create_collection( + collection_name=self.collection_name, + schema=build_memory_schema(), + index_params=build_memory_index_params(), + ) + return + desc = await self.client.describe_collection(self.collection_name) + for field in desc.get("fields", []): + if field.get("name") == "vector": + dim = field.get("params", {}).get("dim") + if dim is not None and int(dim) != EMBEDDING_DIMENSION: + raise RuntimeError( + f"Milvus collection {self.collection_name!r} vector dim={dim}, " + f"expected {EMBEDDING_DIMENSION}" + ) + + async def upsert(self, memory, vector: list[float]) -> str: + """写入一条客户记忆向量并返回 Milvus 主键。""" + await self.ensure_collection() + memory_id = str(memory.id) + await self.client.delete( + collection_name=self.collection_name, + filter=f'memory_id == "{memory_id}"', + ) + await self.client.insert( + collection_name=self.collection_name, + data=[ + { + "memory_id": memory_id, + "customer_id": memory.customer_id, + "memory_type": memory.memory_type, + "tag": memory.tag, + "content": memory.content, + "status": memory.status, + "vector": vector, + } + ], + ) + return memory_id + + async def search(self, vector: list[float], customer_id: int, *, limit: int = 10) -> list[dict]: + """按客户 ID 过滤向量查询结果。""" + await self.ensure_collection() + return await self.client.search( + collection_name=self.collection_name, + data=[vector], + limit=limit, + filter=f"customer_id == {int(customer_id)}", + output_fields=["memory_id", "customer_id", "memory_type", "tag", "content", "status"], + ) + + async def delete(self, memory_id: int | str) -> None: + """删除一条客户记忆向量。""" + await self.client.delete( + collection_name=self.collection_name, + filter=f'memory_id == "{memory_id}"', + ) + + +__all__ = [ + "CUSTOMER_MEMORY_COLLECTION", + "MilvusMemoryStore", + "build_memory_index_params", + "build_memory_schema", +] + diff --git a/service/memory/neo4j_memory.py b/service/memory/neo4j_memory.py new file mode 100644 index 0000000..34d9184 --- /dev/null +++ b/service/memory/neo4j_memory.py @@ -0,0 +1,60 @@ +"""客户长期记忆的 Neo4j 关系镜像。""" + +from __future__ import annotations + +from config.database.neo4j import client as configured_client + + +class Neo4jMemoryStore: + """保存客户节点、记忆节点及其 HAS_MEMORY 关系。""" + + def __init__(self, driver=None): + self.driver = driver or configured_client() + + async def upsert(self, memory) -> str: + """写入客户和记忆节点，返回图谱记忆节点 ID。""" + memory_id = str(memory.id) + query = """ + MERGE (c:Customer {customer_id: $customer_id}) + MERGE (m:CustomerMemory {memory_id: $memory_id}) + SET m.memory_type = $memory_type, + m.tag = $tag, + m.content = $content, + m.status = $status + MERGE (c)-[:HAS_MEMORY]->(m) + RETURN m.memory_id AS memory_id + """ + async with self.driver.session() as session: + record = await session.run( + query, + customer_id=int(memory.customer_id), + memory_id=memory_id, + memory_type=memory.memory_type, + tag=memory.tag, + content=memory.content, + status=memory.status, + ) + row = await record.single() + return row["memory_id"] if row else memory_id + + async def list_by_customer(self, customer_id: int, *, limit: int = 100) -> list[dict]: + """按客户查询图谱记忆关系。""" + query = """ + MATCH (c:Customer {customer_id: $customer_id})-[:HAS_MEMORY]->(m:CustomerMemory) + RETURN m.memory_id AS memory_id, m.memory_type AS memory_type, + m.tag AS tag, m.content AS content, m.status AS status + LIMIT $limit + """ + async with self.driver.session() as session: + result = await session.run(query, customer_id=int(customer_id), limit=limit) + return [dict(record) async for record in result] + + async def delete(self, memory_id: int | str) -> None: + """删除记忆节点及其关系。""" + query = "MATCH (m:CustomerMemory {memory_id: $memory_id}) DETACH DELETE m" + async with self.driver.session() as session: + await session.run(query, memory_id=str(memory_id)) + + +__all__ = ["Neo4jMemoryStore"] + diff --git a/service/memory/profile.py b/service/memory/profile.py new file mode 100644 index 0000000..769e501 --- /dev/null +++ b/service/memory/profile.py @@ -0,0 +1,88 @@ +"""客户画像中期记忆：MySQL 事实源 + Redis Cache-Aside。""" + +from __future__ import annotations + +import json +from decimal import Decimal +from typing import Any + +from config.database.redis import client as redis_client +from repositories.fin_customer_profile import FinCustomerProfileRepo + + +class CustomerProfileMemory: + """提供客户画像读取、缓存失效和刷新能力。""" + + CACHE_TTL = 7 * 24 * 60 * 60 + + def __init__(self, *, redis=None, repository_factory=FinCustomerProfileRepo): + self.redis = redis or redis_client() + self.repository_factory = repository_factory + + @staticmethod + def cache_key(customer_id: int) -> str: + """生成客户画像缓存 Key。""" + return f"profile:{customer_id}" + + async def get(self, db, customer_id: int) -> tuple[dict[str, Any] | None, list[str]]: + """优先读取缓存，未命中后回源 MySQL，并返回 warnings。""" + warnings: list[str] = [] + key = self.cache_key(customer_id) + try: + cached = await self.redis.get(key) + if cached: + return json.loads(cached), warnings + except Exception as exc: + warnings.append(f"profile_cache_read_failed:{type(exc).__name__}") + + try: + profile = await self.repository_factory(db).get_by_customer_id(customer_id) + except Exception as exc: + warnings.append(f"profile_mysql_read_failed:{type(exc).__name__}") + return None, warnings + if profile is None: + return None, warnings + + payload = self._to_dict(profile) + try: + await self.redis.set(key, json.dumps(payload, ensure_ascii=False), ex=self.CACHE_TTL) + except Exception as exc: + warnings.append(f"profile_cache_write_failed:{type(exc).__name__}") + return payload, warnings + + async def invalidate(self, customer_id: int) -> list[str]: + """删除客户画像缓存，确保更新后不会长期读取旧值。""" + try: + await self.redis.delete(self.cache_key(customer_id)) + return [] + except Exception as exc: + return [f"profile_cache_invalidate_failed:{type(exc).__name__}"] + + async def refresh(self, db, customer_id: int) -> tuple[dict[str, Any] | None, list[str]]: + """先删除缓存，再从 MySQL 读取并重新缓存画像。""" + warnings = await self.invalidate(customer_id) + profile, read_warnings = await self.get(db, customer_id) + return profile, warnings + read_warnings + + @staticmethod + def _to_dict(profile) -> dict[str, Any]: + """将 ORM 画像转换为可安全写入 Redis 的字典。""" + data = { + "customer_id": profile.customer_id, + "risk_level": profile.risk_level, + "risk_score": profile.risk_score, + "investment_experience": profile.investment_experience, + "annual_income_range": profile.annual_income_range, + "total_assets": profile.total_assets, + "asset_allocation": profile.asset_allocation, + "product_preference": profile.product_preference, + "customer_level": profile.customer_level, + "confidence_score": profile.confidence_score, + "profile_version": profile.profile_version, + "update_time": profile.update_time.isoformat() if profile.update_time else None, + } + return json.loads(json.dumps(data, default=lambda value: str(value), ensure_ascii=False)) + + +__all__ = ["CustomerProfileMemory"] + diff --git a/service/memory/readiness.py b/service/memory/readiness.py new file mode 100644 index 0000000..036ebcd --- /dev/null +++ b/service/memory/readiness.py @@ -0,0 +1,8 @@ +"""记忆模块依赖健康检查。""" + +from config.database import check_ready_detail + + +async def check_memory_dependencies() -> dict[str, dict]: + """复用项目统一数据库健康检查，避免记忆模块重复管理连接。""" + return await check_ready_detail() diff --git a/service/memory/schemas.py b/service/memory/schemas.py new file mode 100644 index 0000000..57976f0 --- /dev/null +++ b/service/memory/schemas.py @@ -0,0 +1,106 @@ +"""记忆模块跨层共享的数据契约。 + +这些模型只描述记忆模块与客服 Agent 之间的输入输出，不绑定 Redis、MySQL、 +Milvus 或 Neo4j 的实现细节。 +""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +class MemoryType(StrEnum): + """客服记忆可以保存的业务类型。""" + + PROFILE_FACT = "PROFILE_FACT" + PROFILE_CANDIDATE = "PROFILE_CANDIDATE" + CUSTOMER_PREFERENCE = "CUSTOMER_PREFERENCE" + INVESTMENT_GOAL = "INVESTMENT_GOAL" + SERVICE_FACT = "SERVICE_FACT" + CUSTOMER_RELATION = "CUSTOMER_RELATION" + + +class MemoryStatus(StrEnum): + """记忆生命周期状态。""" + + CANDIDATE = "candidate" + CONFIRMED = "confirmed" + EXPIRED = "expired" + REJECTED = "rejected" + ARCHIVED = "archived" + + +class MemorySource(StrEnum): + """客服对话形成记忆的证据来源。""" + + DIALOGUE_CONFIRMED = "dialogue_confirmed" + DIALOGUE_STATED = "dialogue_stated" + DIALOGUE_INFERRED = "dialogue_inferred" + + +class ShortTermMessage(BaseModel): + """Redis 短期会话中的一条消息。""" + + model_config = ConfigDict(extra="forbid") + + message_id: str = Field(min_length=1, max_length=64) + session_id: str = Field(min_length=1, max_length=64) + role: str = Field(min_length=1, max_length=16) + content: str = Field(min_length=1) + token_count: int = Field(default=0, ge=0) + agent_run_id: str | None = Field(default=None, max_length=64) + tool_calls: list[dict[str, Any]] = Field(default_factory=list) + create_time: datetime | None = None + + +class MemoryUnitDTO(BaseModel): + """统一表示一条客户记忆，供写入、召回和重排使用。""" + + model_config = ConfigDict(extra="forbid") + + id: int | None = None + customer_id: int + session_id: str | None = Field(default=None, max_length=64) + agent_run_id: str | None = Field(default=None, max_length=64) + memory_type: MemoryType + tag: str = Field(min_length=1, max_length=64) + content: str = Field(min_length=1, max_length=512) + info_type: str = Field(default="FACT", min_length=1, max_length=8) + source: MemorySource + evidence_ref: list[dict[str, Any]] = Field(default_factory=list) + source_confidence: float = Field(default=0.2, ge=0.0, le=1.0) + confidence: float = Field(default=0.2, ge=0.0, le=1.0) + historical_accuracy: float = Field(default=0.5, ge=0.0, le=1.0) + confidence_version: str | None = Field(default=None, max_length=32) + confidence_reason: str | None = Field(default=None, max_length=255) + confidence_update_time: datetime | None = None + final_score: float | None = Field(default=None, ge=0.0, le=1.0) + evidence_count: int = Field(default=0, ge=0) + conflict_count: int = Field(default=0, ge=0) + recall_count: int = Field(default=0, ge=0) + status: MemoryStatus = MemoryStatus.CANDIDATE + valid_from: datetime | None = None + valid_until: datetime | None = None + last_verified_at: datetime | None = None + milvus_id: str | None = Field(default=None, max_length=128) + graph_node_id: str | None = Field(default=None, max_length=128) + + +class CustomerMemoryContext(BaseModel): + """客服 Agent 每轮请求可使用的统一记忆上下文。""" + + model_config = ConfigDict(extra="forbid") + + customer_id: int + session_id: str + short_term_messages: list[ShortTermMessage] = Field(default_factory=list) + customer_profile: dict[str, Any] | None = None + work_orders: list[dict[str, Any]] = Field(default_factory=list) + long_term_memories: list[MemoryUnitDTO] = Field(default_factory=list) + customer_relations: list[dict[str, Any]] = Field(default_factory=list) + customer_products: list[dict[str, Any]] = Field(default_factory=list) + warnings: list[str] = Field(default_factory=list) diff --git a/service/memory/short_term.py b/service/memory/short_term.py new file mode 100644 index 0000000..528705f --- /dev/null +++ b/service/memory/short_term.py @@ -0,0 +1,232 @@ +"""Redis 短期会话记忆。""" + +from __future__ import annotations + +import asyncio +import json +import time +import uuid +from datetime import datetime, timezone +from inspect import isawaitable +from typing import Any, Callable + +from config.database.redis import client as redis_client + +from .schemas import ShortTermMessage + + +class ShortTermMemoryError(RuntimeError): + """短期记忆操作失败。""" + + +class SessionExpiredError(ShortTermMemoryError): + """会话已超过最长生命周期。""" + + +async def _config(config_getter, key: str, default): + """读取项目配置，并将数据库配置值转换为默认值类型。""" + value = config_getter(key, str(default)) + if isawaitable(value): + value = await value + return type(default)(value) + + +class ShortTermMemory: + """管理当前客服会话的 Redis 消息、Token 预算和生命周期。""" + + MESSAGE_KEY = "session:{session_id}:messages" + META_KEY = "session:{session_id}:meta" + ALLOWED_ROLES = frozenset({"user", "assistant", "system"}) + + def __init__( + self, + redis=None, + *, + config_getter=None, + token_counter: Callable[[str], int] | None = None, + clock=time.time, + fail_soft: bool = True, + ): + """创建短期记忆服务，默认复用项目级 Redis 客户端。""" + self.redis = redis or redis_client() + self.config_getter = config_getter or (lambda _key, default: default) + self.token_counter = token_counter or (lambda text: max(1, len(text) // 4)) + self.clock = clock + self.fail_soft = fail_soft + self.last_warnings: list[str] = [] + self._locks: dict[str, asyncio.Lock] = {} + + @classmethod + def message_key(cls, session_id: str) -> str: + """生成会话消息列表 Key。""" + return cls.MESSAGE_KEY.format(session_id=session_id) + + @classmethod + def meta_key(cls, session_id: str) -> str: + """生成会话元数据 Hash Key。""" + return cls.META_KEY.format(session_id=session_id) + + def _lock(self, session_id: str) -> asyncio.Lock: + """获取进程内会话锁，避免并发截断互相覆盖。""" + return self._locks.setdefault(session_id, asyncio.Lock()) + + async def append_message( + self, + session_id: str, + role: str, + content: str, + *, + message_id: str | None = None, + agent_run_id: str | None = None, + tool_calls: list[dict[str, Any]] | None = None, + ) -> ShortTermMessage | None: + """追加消息、刷新 TTL，并按 Token 预算从旧到新截断。""" + self.last_warnings = [] + if role not in self.ALLOWED_ROLES: + raise ValueError(f"非法消息角色: {role}") + if not isinstance(content, str) or not content.strip(): + raise ValueError("消息内容不能为空") + + message = ShortTermMessage( + message_id=message_id or uuid.uuid4().hex, + session_id=session_id, + role=role, + content=content, + token_count=self.token_counter(content), + agent_run_id=agent_run_id, + tool_calls=tool_calls or [], + create_time=datetime.fromtimestamp(self.clock(), tz=timezone.utc), + ) + try: + async with self._lock(session_id): + await self._ensure_meta(session_id) + await self.redis.rpush( + self.message_key(session_id), self._serialize(message) + ) + await self._touch(session_id) + await self._trim(session_id) + return message + except SessionExpiredError: + raise + except Exception as exc: + return self._degrade("append_message", exc) + + async def load_messages(self, session_id: str) -> list[ShortTermMessage]: + """按写入顺序读取当前会话消息，并刷新空闲 TTL。""" + self.last_warnings = [] + try: + if not await self._session_is_active(session_id): + return [] + raw_messages = await self.redis.lrange(self.message_key(session_id), 0, -1) + await self._touch(session_id) + return [self._deserialize(raw) for raw in raw_messages if raw] + except Exception as exc: + return self._degrade("load_messages", exc) or [] + + async def get_message_count(self, session_id: str) -> int: + """返回当前会话消息数量。""" + return len(await self.load_messages(session_id)) + + async def get_token_count(self, session_id: str) -> int: + """返回当前会话消息的 Token 估算总数。""" + return sum(message.token_count for message in await self.load_messages(session_id)) + + async def clear_session(self, session_id: str) -> None: + """清理会话消息和元数据。""" + self.last_warnings = [] + try: + await self.redis.delete(self.message_key(session_id), self.meta_key(session_id)) + except Exception as exc: + self._degrade("clear_session", exc) + + async def _ensure_meta(self, session_id: str) -> None: + """初始化会话元数据，并固定 24 小时绝对过期时间。""" + meta_key = self.meta_key(session_id) + meta = await self.redis.hgetall(meta_key) + now = self.clock() + if meta: + absolute_expire_at = float(meta.get("absolute_expire_at", now)) + if absolute_expire_at <= now: + await self.clear_session(session_id) + raise SessionExpiredError("会话已超过最长生命周期") + return + max_lifetime = await _config( + self.config_getter, "agent.customer.session.max_lifetime", 86400 + ) + await self.redis.hset( + meta_key, + mapping={ + "created_at": str(now), + "absolute_expire_at": str(now + max_lifetime), + }, + ) + await self.redis.expire(meta_key, max_lifetime) + + async def _session_is_active(self, session_id: str) -> bool: + """检查会话元数据是否存在且未达到绝对过期时间。""" + meta = await self.redis.hgetall(self.meta_key(session_id)) + if not meta: + return False + if float(meta.get("absolute_expire_at", 0)) <= self.clock(): + await self.clear_session(session_id) + return False + return True + + async def _touch(self, session_id: str) -> None: + """刷新空闲 TTL，但不超过绝对过期时间。""" + meta = await self.redis.hgetall(self.meta_key(session_id)) + if not meta: + return + remaining = int(float(meta["absolute_expire_at"]) - self.clock()) + if remaining <= 0: + await self.clear_session(session_id) + raise SessionExpiredError("会话已超过最长生命周期") + idle_ttl = await _config( + self.config_getter, "agent.customer.session.ttl", 1800 + ) + ttl = min(idle_ttl, remaining) + await self.redis.expire(self.message_key(session_id), ttl) + await self.redis.expire(self.meta_key(session_id), remaining) + + async def _trim(self, session_id: str) -> None: + """保留最新消息，确保最新一条消息不会因超预算被删除。""" + limit = await _config( + self.config_getter, "agent.customer.session_max_token", 4096 + ) + key = self.message_key(session_id) + raw_messages = [raw for raw in await self.redis.lrange(key, 0, -1) if raw] + total = 0 + keep_from = len(raw_messages) + for index in range(len(raw_messages) - 1, -1, -1): + message = self._deserialize(raw_messages[index]) + total += message.token_count + keep_from = index + if total > limit: + keep_from = index + 1 + break + if raw_messages and keep_from == len(raw_messages): + keep_from = len(raw_messages) - 1 + await self.redis.ltrim(key, keep_from, -1) + + @staticmethod + def _serialize(message: ShortTermMessage) -> str: + """将消息转换为 Redis List 中的 JSON 字符串。""" + return json.dumps(message.model_dump(mode="json"), ensure_ascii=False) + + @staticmethod + def _deserialize(raw: str | bytes) -> ShortTermMessage: + """将 Redis JSON 字符串恢复为消息 DTO。""" + if isinstance(raw, bytes): + raw = raw.decode("utf-8") + return ShortTermMessage.model_validate(json.loads(raw)) + + def _degrade(self, operation: str, exc: Exception): + """记录降级原因；fail_soft 模式下不让 Redis 故障击穿客服请求。""" + warning = f"short_term_{operation}_degraded:{type(exc).__name__}" + self.last_warnings = [warning] + if not self.fail_soft: + raise ShortTermMemoryError(warning) from exc + return None + + +__all__ = ["SessionExpiredError", "ShortTermMemory", "ShortTermMemoryError"] diff --git a/service/memory/work_order.py b/service/memory/work_order.py new file mode 100644 index 0000000..ea066df --- /dev/null +++ b/service/memory/work_order.py @@ -0,0 +1,40 @@ +"""客户工单中期记忆。""" + +from __future__ import annotations + +from typing import Any + +from repositories.work_order import WorkOrderRepo + + +class WorkOrderMemory: + """以 MySQL 为事实来源读取客户工单状态。""" + + def __init__(self, *, repository_factory=WorkOrderRepo): + self.repository_factory = repository_factory + + async def list(self, db, customer_id: int, *, active_only: bool = True) -> list[dict[str, Any]]: + """按客户 ID 查询工单并转换为客服上下文结构。""" + orders = await self.repository_factory(db).list_by_customer( + customer_id, active_only=active_only + ) + return [ + { + "id": order.id, + "work_order_no": order.work_order_no, + "order_type": order.order_type, + "sub_type": order.sub_type, + "customer_id": order.customer_id, + "handler_id": order.handler_id, + "current_node": order.current_node, + "priority": order.priority, + "status": order.status, + "biz_content": order.biz_content, + "create_time": order.create_time, + "update_time": order.update_time, + } + for order in orders + ] + + +__all__ = ["WorkOrderMemory"] diff --git a/sql/schema.sql b/sql/schema.sql index cc44f66..ddd766a 100644 --- a/sql/schema.sql +++ b/sql/schema.sql @@ -1,10 +1,43 @@ -- ===================================================================== -- 智能公募基金系统 · MySQL 建表脚本（24 张，见开发计划 §3.1） --- 执行：mysql -u -p < sql/schema.sql +-- 执行：mysql --default-character-set=utf8mb4 -u -p < sql/schema.sql -- 约定：InnoDB / utf8mb4；采用"逻辑外键"（不建物理 FK，便于 Mock 数据灌入）， -- 注释即文档，NL2SQL 以 COMMENT 注入 Schema。 -- ===================================================================== +SET NAMES utf8mb4; + +-- ===================================================================== +-- 重建模式：执行本脚本前清理已有表 +-- 注意：以下语句会删除当前数据库中的业务数据，仅用于本地/测试环境重建。 +-- ===================================================================== +SET FOREIGN_KEY_CHECKS = 0; +DROP TABLE IF EXISTS portfolio_benchmark; +DROP TABLE IF EXISTS fund_performance; +DROP TABLE IF EXISTS memory_unit; +DROP TABLE IF EXISTS sys_config; +DROP TABLE IF EXISTS sys_message; +DROP TABLE IF EXISTS sys_announcement; +DROP TABLE IF EXISTS audit_log; +DROP TABLE IF EXISTS customer_profile_change_log; +DROP TABLE IF EXISTS customer_relation; +DROP TABLE IF EXISTS fin_knowledge_meta; +DROP TABLE IF EXISTS conversation_archive; +DROP TABLE IF EXISTS biz_work_order; +DROP TABLE IF EXISTS risk_rule; +DROP TABLE IF EXISTS fin_risk_alert; +DROP TABLE IF EXISTS ops_question; +DROP TABLE IF EXISTS ops_questionnaire; +DROP TABLE IF EXISTS fin_risk_assessment; +DROP TABLE IF EXISTS fin_holdings; +DROP TABLE IF EXISTS fin_transaction; +DROP TABLE IF EXISTS trade_order; +DROP TABLE IF EXISTS fund_nav_history; +DROP TABLE IF EXISTS fin_product; +DROP TABLE IF EXISTS fin_customer_profile; +DROP TABLE IF EXISTS sys_user; +SET FOREIGN_KEY_CHECKS = 1; + -- --------------------------------------------------------------------- -- 01 统一用户表（客户 + 员工共用） -- --------------------------------------------------------------------- @@ -260,20 +293,20 @@ CREATE TABLE IF NOT EXISTS biz_work_order ( CREATE TABLE IF NOT EXISTS conversation_archive ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, session_id VARCHAR(64) NOT NULL COMMENT '会话ID，同轮共用', + customer_id BIGINT UNSIGNED NULL COMMENT '关联客户ID', user_id BIGINT UNSIGNED NOT NULL COMMENT '对话用户', agent_type VARCHAR(32) NOT NULL COMMENT 'customer/advisor/risk/analyst', role VARCHAR(16) NOT NULL COMMENT 'user/assistant/system', content MEDIUMTEXT NULL COMMENT '对话内容', tool_calls JSON NULL COMMENT '工具调用记录 [{"tool":"nl2sql",...}]', + message_id VARCHAR(64) NOT NULL COMMENT '消息唯一ID，用于归档幂等', + agent_run_id VARCHAR(64) NULL COMMENT 'Agent运行ID，用于调用链路追踪', trace_id VARCHAR(64) NULL COMMENT '请求链路追踪ID', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - trace_id VARCHAR(64) NULL COMMENT '请求链路追踪ID', KEY idx_session (session_id), KEY idx_user_time (user_id, create_time), KEY idx_agent (agent_type), - message_id VARCHAR(64) NULL COMMENT '消息唯一ID，用于归档幂等', - agent_run_id VARCHAR(64) NULL COMMENT 'Agent运行ID，用于调用链路追踪', - UNIQUE KEY uk_session_message (session_id, message_id), + UNIQUE KEY uk_session_message (session_id, message_id), KEY idx_agent_run (agent_run_id) ) COMMENT='会话归档表（审计回溯 + Agent 持续学习素材，归档前脱敏）'; @@ -396,23 +429,47 @@ CREATE TABLE IF NOT EXISTS sys_config ( CREATE TABLE IF NOT EXISTS memory_unit ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, customer_id BIGINT UNSIGNED NOT NULL COMMENT '归属客户', + session_id VARCHAR(64) NULL COMMENT '产生该记忆的会话ID', + agent_run_id VARCHAR(64) NULL COMMENT '产生该记忆的Agent运行ID', + memory_type VARCHAR(32) NOT NULL COMMENT '记忆类型：PROFILE_FACT/PROFILE_CANDIDATE/CUSTOMER_PREFERENCE/INVESTMENT_GOAL/SERVICE_FACT/CUSTOMER_RELATION', tag VARCHAR(64) NOT NULL COMMENT '标签名，如 risk_preference', content VARCHAR(512) NOT NULL COMMENT '内容，如 偏好稳健型理财', info_type VARCHAR(8) NOT NULL COMMENT 'FACT 客观事实 / OPINION 主观观点', - source VARCHAR(32) NOT NULL COMMENT '风评问卷/行为推断/AI对话提取/用户自述/系统默认', + source VARCHAR(32) NOT NULL COMMENT 'user_confirmed/user_stated/ai_inferred/behavior_inference', + evidence_ref JSON NULL COMMENT '证据引用列表，如消息ID、交易ID或人工记录ID', source_confidence DECIMAL(5,2) NOT NULL DEFAULT 0.20 COMMENT '来源初始置信度', confidence DECIMAL(5,2) NOT NULL DEFAULT 0.20 COMMENT '当前置信度（动态）', + historical_accuracy DECIMAL(5,2) NOT NULL DEFAULT 0.50 COMMENT '历史准确性评分', + confidence_version VARCHAR(32) NULL COMMENT '置信度计算版本', + confidence_reason VARCHAR(255) NULL COMMENT '本次置信度评估原因', + confidence_update_time DATETIME NULL COMMENT '置信度最近计算时间', evidence_count INT NOT NULL DEFAULT 0 COMMENT '证据数', conflict_count INT NOT NULL DEFAULT 0 COMMENT '冲突数', recall_count INT NOT NULL DEFAULT 0 COMMENT '召回数', + memory_version INT NOT NULL DEFAULT 1 COMMENT '记忆版本号', update_time DATETIME NULL, last_recall_time DATETIME NULL, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT 'active/demoted/archived/deleted', - valid_until DATE NULL COMMENT '有效期', + status VARCHAR(16) NOT NULL DEFAULT 'candidate' COMMENT 'candidate/confirmed/expired/rejected/archived', + valid_from DATETIME NULL COMMENT '记忆生效时间', + valid_until DATETIME NULL COMMENT '有效期', + last_verified_at DATETIME NULL COMMENT '最近确认或验证时间', + milvus_id VARCHAR(128) NULL COMMENT 'Milvus 向量记录ID', + graph_node_id VARCHAR(128) NULL COMMENT 'Neo4j 节点或关系ID', + milvus_sync_status VARCHAR(16) NOT NULL DEFAULT 'pending' COMMENT 'Milvus同步状态：pending/success/failed', + neo4j_sync_status VARCHAR(16) NOT NULL DEFAULT 'pending' COMMENT 'Neo4j同步状态：pending/success/failed', + sync_retry_count INT NOT NULL DEFAULT 0 COMMENT '三库同步重试次数', + last_sync_error VARCHAR(500) NULL COMMENT '最近一次同步错误', + next_retry_at DATETIME NULL COMMENT '下次同步重试时间', + last_synced_at DATETIME NULL COMMENT '最近一次同步成功时间', KEY idx_customer_status (customer_id, status), - KEY idx_tag (tag) -) COMMENT='记忆单元表（三层记忆中期主体，向量镜像在 Milvus customer_memory）'; + KEY idx_customer_type_status (customer_id, memory_type, status), + KEY idx_customer_tag_status (customer_id, tag, status), + KEY idx_session (session_id), + KEY idx_customer_session (customer_id, session_id), + KEY idx_milvus (milvus_id), + KEY idx_graph (graph_node_id) +) COMMENT='客户长期记忆主体表，向量镜像在 Milvus customer_memory，关系镜像在 Neo4j'; -- --------------------------------------------------------------------- -- 23 基金业绩指标缓存（定时任务从净值计算） @@ -442,4 +499,4 @@ CREATE TABLE IF NOT EXISTS portfolio_benchmark ( create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_risk_level (risk_level) -) COMMENT='组合基准配置表（投顾Agent 再平衡参照，运营可调整）'; \ No newline at end of file +) COMMENT='组合基准配置表（投顾Agent 再平衡参照，运营可调整）'; diff --git a/tool/confidence.py b/tool/confidence.py new file mode 100644 index 0000000..276d12a --- /dev/null +++ b/tool/confidence.py @@ -0,0 +1,115 @@ +"""客服对话记忆的基础置信度计算工具。""" + +from __future__ import annotations + +from typing import Any + + +class BaseConfidenceCalcTool: + """根据来源、证据、冲突和时间计算单条记忆的长期置信度。""" + + SOURCE_INITIAL = { + "dialogue_confirmed": 0.75, + "dialogue_stated": 0.50, + "dialogue_inferred": 0.45, + } + MEMORY_THRESHOLDS = { + "PROFILE_FACT": 0.75, + "CUSTOMER_PREFERENCE": 0.65, + "INVESTMENT_GOAL": 0.70, + "SERVICE_FACT": 0.75, + } + DEFAULT_THRESHOLD = 0.80 + VERSION = "confidence-v1" + + def calc( + self, + tag: str, + source: str, + evidence_count: int, + conflict_count: int, + age_days: int, + ) -> float: + """计算基础置信度分数，返回范围为 0 到 1 的浮点数。""" + self._validate(tag, source, evidence_count, conflict_count, age_days) + base = self.SOURCE_INITIAL[source] + gain = min(evidence_count * 0.05, 0.30) + penalty = min(conflict_count * 0.10, 0.50) + decay = max(0.80, 1 - age_days / 365 * 0.20) + return max(0.0, min(1.0, (base + gain - penalty) * decay)) + + def evaluate( + self, + *, + tag: str, + source: str, + evidence_count: int, + conflict_count: int, + age_days: int, + memory_type: str | None = None, + threshold: float | None = None, + ) -> dict[str, Any]: + """返回可供记忆模块保存的完整置信度评估结果。""" + score = self.calc(tag, source, evidence_count, conflict_count, age_days) + if threshold is None: + threshold = self.MEMORY_THRESHOLDS.get( + memory_type or "", self.DEFAULT_THRESHOLD + ) + if not 0.0 <= threshold <= 1.0: + raise ValueError("threshold 必须在 [0.0, 1.0] 范围内") + base = self.SOURCE_INITIAL[source] + status = "confirmed" if score >= threshold else "candidate" + return { + "source_confidence": base, + "confidence": score, + "status": status, + "evidence_count": evidence_count, + "conflict_count": conflict_count, + "age_days": age_days, + "threshold": threshold, + "confidence_reason": self._reason( + source, evidence_count, conflict_count, age_days + ), + "confidence_version": self.VERSION, + } + + def batch_calc(self, tags: list[dict[str, Any]]) -> list[float]: + """批量计算基础分数。""" + return [self.calc(**tag) for tag in tags] + + def batch_evaluate(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]: + """批量生成完整评估结果。""" + return [self.evaluate(**item) for item in items] + + @classmethod + def _validate( + cls, + tag: str, + source: str, + evidence_count: int, + conflict_count: int, + age_days: int, + ) -> None: + """校验工具输入，避免非法计数污染记忆分数。""" + if not tag or not tag.strip(): + raise ValueError("tag 不能为空") + if source not in cls.SOURCE_INITIAL: + raise ValueError(f"不支持的客服对话来源: {source}") + for name, value in ( + ("evidence_count", evidence_count), + ("conflict_count", conflict_count), + ("age_days", age_days), + ): + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError(f"{name} 必须是非负整数") + + @staticmethod + def _reason(source: str, evidence_count: int, conflict_count: int, age_days: int) -> str: + """生成便于审计和排查的评分原因。""" + return ( + f"来源={source}; 支持证据={evidence_count}; 冲突证据={conflict_count}; " + f"存在天数={age_days}; 采用证据增益、冲突惩罚和时间衰减" + ) + + +__all__ = ["BaseConfidenceCalcTool"] diff --git a/tool/confidence_rank.py b/tool/confidence_rank.py new file mode 100644 index 0000000..61516d5 --- /dev/null +++ b/tool/confidence_rank.py @@ -0,0 +1,99 @@ +"""客服记忆候选的综合置信分重排工具。""" + +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime +from typing import Any + + +class FinalConfidenceRankTool: + """仅服务客服记忆召回的临时重排工具。""" + + WEIGHTS = { + "semantic": 0.30, + "timeliness": 0.20, + "accuracy": 0.20, + "base": 0.25, + "conflict": 0.05, + } + INVALID_STATUSES = frozenset({"expired", "rejected", "archived"}) + + def rank( + self, + memory_units: list[dict[str, Any]], + *, + top_k: int | None = None, + now: datetime | None = None, + ) -> list[dict[str, Any]]: + """过滤无效候选并按当前客服召回分数降序返回副本。""" + if top_k is not None and (not isinstance(top_k, int) or top_k < 0): + raise ValueError("top_k 必须是非负整数或 None") + now = now or datetime.now() + ranked = [] + for original in memory_units: + unit = self._as_dict(original) + if self._is_invalid(unit, now): + continue + semantic = self._bounded(unit.get("semantic_similarity", 0.5), 0.5) + timeliness = self._calc_timeliness(unit.get("age_days", 0)) + accuracy = self._bounded(unit.get("historical_accuracy", 0.5), 0.5) + base = self._bounded(unit.get("confidence", 0.5), 0.5) + conflict_penalty = min(self._non_negative_int(unit.get("conflict_count", 0)), 5) * 0.1 + final_score = ( + self.WEIGHTS["semantic"] * semantic + + self.WEIGHTS["timeliness"] * timeliness + + self.WEIGHTS["accuracy"] * accuracy + + self.WEIGHTS["base"] * base + - self.WEIGHTS["conflict"] * conflict_penalty + ) + unit["final_score"] = max(0.0, min(1.0, final_score)) + ranked.append(unit) + ranked.sort(key=lambda item: item["final_score"], reverse=True) + return ranked if top_k is None else ranked[:top_k] + + @staticmethod + def _as_dict(unit: dict[str, Any] | Any) -> dict[str, Any]: + """兼容字典和 Pydantic/ORM 风格候选对象。""" + if isinstance(unit, dict): + return deepcopy(unit) + if hasattr(unit, "model_dump"): + return deepcopy(unit.model_dump()) + return deepcopy(vars(unit)) + + @classmethod + def _is_invalid(cls, unit: dict[str, Any], now: datetime) -> bool: + """过滤拒绝、归档和已过有效期的记忆。""" + if unit.get("status") in cls.INVALID_STATUSES: + return True + valid_until = unit.get("valid_until") + return valid_until is not None and valid_until <= now + + @staticmethod + def _bounded(value: Any, default: float) -> float: + """将缺失或异常评分转换为保守默认值。""" + try: + value = float(value) + except (TypeError, ValueError): + return default + return max(0.0, min(1.0, value)) + + @staticmethod + def _non_negative_int(value: Any) -> int: + """将冲突次数转换为非负整数。""" + try: + return max(0, int(value)) + except (TypeError, ValueError): + return 0 + + @staticmethod + def _calc_timeliness(age_days: Any) -> float: + """按每年 20% 计算平滑时效分，最低保留 0.8。""" + try: + age_days = max(0, int(age_days)) + except (TypeError, ValueError): + age_days = 0 + return max(0.80, 1 - age_days / 365 * 0.20) + + +__all__ = ["FinalConfidenceRankTool"]