From fed7cd45e44cd9d6007fa43182271326095989c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AC=A7=E9=98=B3=E6=B4=8B?= <2443479321@qq.com> Date: Mon, 14 Sep 2026 11:08:52 +0800 Subject: [PATCH] =?UTF-8?q?fix:=E7=94=A8=E6=88=B7=E7=94=BB=E5=83=8F?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/advisor_agent/memory.py | 16 ++++- api/router.py | 5 +- service/memory/facade.py | 108 +++++++++++++++++++++------------- 3 files changed, 85 insertions(+), 44 deletions(-) diff --git a/agent/advisor_agent/memory.py b/agent/advisor_agent/memory.py index 9740749..ef07a78 100644 --- a/agent/advisor_agent/memory.py +++ b/agent/advisor_agent/memory.py @@ -17,11 +17,22 @@ class EmptyMemoryProvider: class MemoryServiceProvider: - """将现有 MemoryService 的统一上下文转换为投顾侧记忆列表。""" + """将现有 MemoryService 的统一上下文转换为投顾侧记忆列表。 - def __init__(self, memory_service, *, session_prefix: str = "advisor-agent"): + 投顾推荐场景**需要**长期记忆（偏好标签、投资目标），因此显式开启 + ``include_long_term=True``。客服 Agent 走默认值 False，不重复召回。 + """ + + def __init__( + self, + memory_service, + *, + session_prefix: str = "advisor-agent", + include_long_term: bool = True, + ): self.memory_service = memory_service self.session_prefix = session_prefix + self.include_long_term = include_long_term self.last_warnings: list[str] = [] async def recall(self, *, customer_id: int, query: str) -> list[dict]: @@ -31,6 +42,7 @@ class MemoryServiceProvider: customer_id=customer_id, session_id=f"{self.session_prefix}:{customer_id}", query=query, + include_long_term=self.include_long_term, ) except Exception as exc: self.last_warnings = [f"advisor_memory_recall_failed:{type(exc).__name__}"] diff --git a/api/router.py b/api/router.py index f99603a..d398bd8 100644 --- a/api/router.py +++ b/api/router.py @@ -5,7 +5,7 @@ from fastapi import APIRouter from api.chat import client_agent, customer_agent, knowledge from api.routers import product, questionnaire -from api.routers import account, auth, holdings, purchase, redeem, risk, work_order +from api.routers import account, auth, holdings, profile, purchase, redeem, risk, work_order from api.routers import advisor_agent, health, nl2sql, nl2sql_admin from api.advisor import audit, customers, dashboard, data_query, diagnosis, drafts, report, todos, visits @@ -17,6 +17,7 @@ api_router.include_router(purchase.router, prefix="/api", tags=["交易"]) api_router.include_router(redeem.router, prefix="/api", tags=["交易"]) api_router.include_router(risk.router, prefix="/api", tags=["风控"]) api_router.include_router(work_order.router, prefix="/api", tags=["工单"]) +api_router.include_router(profile.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=["知识库"]) @@ -37,4 +38,4 @@ for workbench_router in ( api_router.include_router(workbench_router, prefix="/api/advisor") api_router.include_router(health.router, prefix="/api", tags=["健康检查"]) api_router.include_router(nl2sql.router, prefix="/api", tags=["NL2SQL"]) -api_router.include_router(nl2sql_admin.router, prefix="/api", tags=["NL2SQL管理"]) +api_router.include_router(nl2sql_admin.router, prefix="/api", tags=["NL2SQL管理"]) \ No newline at end of file diff --git a/service/memory/facade.py b/service/memory/facade.py index 88a5dc9..0e680ea 100644 --- a/service/memory/facade.py +++ b/service/memory/facade.py @@ -48,13 +48,15 @@ class MemoryService: self.relations = relations or CustomerRelationMemory() self.products = products or CustomerProductMemory() self.holdings = holdings or CustomerHoldingsMemory() + self.long_term = long_term or LongTermMemoryService() + # composed 依赖上面三个已初始化的组件：必须先赋值 self.long_term， + # 否则默认分支会把 None 传进去（历史 bug）。 self.composed = composed or ComposedProfileService( profile=self.profile, long_term=self.long_term, holdings=self.holdings, redis=self.short_term.redis, ) - self.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) @@ -72,8 +74,24 @@ class MemoryService: session_id: str, query: str | None = None, limit: int = 10, + include_long_term: bool = False, + include_composed: bool = False, ) -> CustomerMemoryContext: - """召回并组装客服 Agent 当前请求的全部可用记忆。""" + """召回并组装客服 Agent 当前请求的全部可用记忆。 + + 默认**不含**长期记忆与最终画像：客服 Agent 的职责是知识问答/开户引导/ + 公司信息，合规上不推荐产品，长期记忆（偏好标签、投资目标）对其没有 + 合法用途，却要付出 memory_unit 查询 + Embedding + Milvus 检索的代价。 + + 长期记忆与最终画像的两个合法消费方各自显式开启： + + - 投顾 Agent：``include_long_term=True``（推荐场景需要偏好与目标）； + - 最终画像接口（``/profile/composed``，仅员工）：直接走 + ``ComposedProfileService.compose_response``，不经本方法。 + + 写入路径（``save_memory`` / ``record_interest_signal``）不受影响， + 客服会话仍持续沉淀长期记忆，供投顾与画像使用。 + """ if limit < 0: raise ValueError("limit 必须是非负整数") warnings: list[str] = [] @@ -81,14 +99,15 @@ class MemoryService: 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__}") + if include_long_term: + try: + refresh_confidence = getattr(self.long_term, "refresh_confidence", None) + if refresh_confidence is not None: + await refresh_confidence( + db, customer_id, limit=max(limit, 10) + ) + except Exception as exc: + warnings.append(f"confidence_refresh_failed:{type(exc).__name__}") profile, profile_warnings = await self.profile.get(db, customer_id) warnings.extend(profile_warnings) try: @@ -106,14 +125,45 @@ class MemoryService: 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__}") + + memories: list[MemoryUnitDTO] = [] + if include_long_term: + try: + memories, memory_warnings = await self.long_term.recall( + db, customer_id, limit=max(limit, 10), query=query + ) + warnings.extend(memory_warnings) + except Exception as exc: + memories = [] + warnings.append(f"long_term_recall_failed:{type(exc).__name__}") + + holdings_summary = None + composed_payload: dict[str, Any] | None = None + if include_composed: + try: + holdings_summary, holdings_warnings = await self.holdings.summary( + db, customer_id + ) + warnings.extend(holdings_warnings) + except Exception as exc: + warnings.append(f"holdings_recall_failed:{type(exc).__name__}") + + try: + # ComposedProfileService 期望 list[dict]（与 _recall_memories + # 的返回一致），这里传已序列化的记忆，避免在块外再触碰会话。 + composed_payload, composed_warnings = await self.composed.compose( + db, + customer_id=customer_id, + query=query, + profile=profile, + memories=[ + memory.model_dump(mode="json") for memory in memories + ], + holdings_summary=holdings_summary, + ) + warnings.extend(composed_warnings) + except Exception as exc: + warnings.append(f"composed_profile_failed:{type(exc).__name__}") ranked = self.rank_tool.rank( [memory.model_dump(mode="json") for memory in memories], @@ -121,28 +171,6 @@ class MemoryService: ) 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,