437 lines
18 KiB
Python
437 lines
18 KiB
Python
"""最终用户画像整合服务：fin_customer_profile + memory_unit + 持仓 三源加权。
|
||
|
||
- 权重语义是 LLM 整合时的注意力权重（配置于 sys_config：profile.weight.*），
|
||
不做算术加权——三源是异构数据。
|
||
- 只做派生视图：结果缓存在 Redis（成功 30 分钟 / 降级 5 分钟），
|
||
不落库、不回写 fin_customer_profile，适当性校验仍以问卷风评原值为准。
|
||
- LLM 失败时降级为规则拼接摘要，沿项目 warnings 机制上报，绝不阻塞调用方。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from repositories.sys_config import SysConfigRepo
|
||
from tool.confidence_rank import FinalConfidenceRankTool
|
||
|
||
from .holdings import CustomerHoldingsMemory
|
||
from .long_term import LongTermMemoryService
|
||
from .profile import CustomerProfileMemory
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ComposedProfileService:
|
||
"""拉取三路数据，调用统一 LLM 整合为最终用户画像。"""
|
||
|
||
CACHE_TTL = 20 * 60
|
||
DEGRADED_CACHE_TTL = 20 * 60
|
||
RESPONSE_CACHE_TTL = 20 * 60
|
||
RESPONSE_CACHE_PREFIX = "composed_profile_response"
|
||
DEFAULT_WEIGHTS = {"profile": 0.5, "memory": 0.2, "holdings": 0.3}
|
||
WEIGHT_KEYS = {
|
||
"profile": "profile.weight.profile",
|
||
"memory": "profile.weight.memory",
|
||
"holdings": "profile.weight.holdings",
|
||
}
|
||
ENABLED_KEY = "profile.composed.enabled"
|
||
MEMORY_TOP_K_KEY = "profile.composed.memory_top_k"
|
||
DEFAULT_MEMORY_TOP_K = 10
|
||
|
||
SYSTEM_PROMPT = """
|
||
你是用户画像整合器。根据三路客户数据生成最终用户画像 JSON，只输出 JSON，不要 Markdown：
|
||
{{"summary": "一段话画像结论", "risk_tendency": "风险偏好结论",
|
||
"preference_tags": ["标签"], "behavior_traits": ["行为特征"], "advice_focus": "服务建议"}}
|
||
权重指引：结构化画像信息可信度最高（权重 {w_profile}）；持仓反映真实风险行为（权重 {w_holdings}）；
|
||
对话记忆是口头表达补充（权重 {w_memory}）。三源冲突时按权重取舍，且必须在 summary 中显式说明冲突。
|
||
status 为 candidate 的记忆只能作参考，不能作为确定性结论。
|
||
只能基于输入事实归纳，禁止编造；数据缺失的字段留空数组或空字符串，不要推测。
|
||
禁止在结果中出现内部字段名、ID、SQL 或证据引用。
|
||
""".strip()
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
profile: CustomerProfileMemory | None = None,
|
||
long_term: LongTermMemoryService | None = None,
|
||
holdings: CustomerHoldingsMemory | None = None,
|
||
redis=None,
|
||
llm_client=None,
|
||
rank_tool: FinalConfidenceRankTool | None = None,
|
||
config_repo_cls=SysConfigRepo,
|
||
):
|
||
from config.database.redis import client as redis_client
|
||
from tool.llm import llm as default_llm
|
||
|
||
self.profile = profile or CustomerProfileMemory()
|
||
self.long_term = long_term or LongTermMemoryService()
|
||
self.holdings = holdings or CustomerHoldingsMemory()
|
||
self.redis = redis or redis_client()
|
||
self.llm_client = llm_client or default_llm
|
||
self.rank_tool = rank_tool or FinalConfidenceRankTool()
|
||
self.config_repo_cls = config_repo_cls
|
||
|
||
@staticmethod
|
||
def cache_key(customer_id: int) -> str:
|
||
"""生成画像负载缓存 Key。"""
|
||
return f"composed_profile:{customer_id}"
|
||
|
||
@staticmethod
|
||
def response_cache_key(customer_id: int) -> str:
|
||
"""生成接口完整响应缓存 Key。"""
|
||
return f"{ComposedProfileService.RESPONSE_CACHE_PREFIX}:{customer_id}"
|
||
|
||
async def compose(
|
||
self,
|
||
db: AsyncSession,
|
||
*,
|
||
customer_id: int,
|
||
query: str | None = None,
|
||
profile: dict[str, Any] | None = None,
|
||
memories: list[dict[str, Any]] | None = None,
|
||
holdings_summary: dict[str, Any] | None = None,
|
||
) -> tuple[dict[str, Any] | None, list[str]]:
|
||
"""整合最终画像；可传入 facade 已召回的数据避免重复取数。"""
|
||
warnings: list[str] = []
|
||
enabled = await self._enabled(db, warnings)
|
||
if not enabled:
|
||
return None, warnings
|
||
|
||
cached = await self._cache_get(customer_id)
|
||
if cached is not None:
|
||
return cached, warnings
|
||
|
||
weights, weight_warnings = await self._load_weights(db)
|
||
warnings.extend(weight_warnings)
|
||
top_k = await self._memory_top_k(db, warnings)
|
||
|
||
if profile is None:
|
||
profile, profile_warnings = await self.profile.get(db, customer_id)
|
||
warnings.extend(profile_warnings)
|
||
if memories is None:
|
||
memories = await self._recall_memories(db, customer_id, query, top_k, warnings)
|
||
if holdings_summary is None:
|
||
holdings_summary, holdings_warnings = await self.holdings.summary(db, customer_id)
|
||
warnings.extend(holdings_warnings)
|
||
|
||
slim_memories = self._slim_memories(memories, top_k)
|
||
if profile is None and not slim_memories and holdings_summary is None:
|
||
return None, warnings
|
||
|
||
try:
|
||
result = await self._integrate(weights, profile, slim_memories, holdings_summary)
|
||
cache_ttl = self.CACHE_TTL
|
||
except Exception as exc: # noqa: BLE001 降级不阻塞
|
||
warnings.append(f"composed_profile_llm_failed:{type(exc).__name__}")
|
||
result = self._degraded(profile, slim_memories, holdings_summary)
|
||
cache_ttl = self.DEGRADED_CACHE_TTL
|
||
|
||
payload = {
|
||
"profile": result,
|
||
"weights": weights,
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
}
|
||
await self._cache_set(customer_id, payload, cache_ttl)
|
||
return payload, warnings
|
||
|
||
async def invalidate(self, customer_id: int) -> list[str]:
|
||
"""删除画像负载与接口响应两级缓存，供画像/持仓变更后调用。"""
|
||
warnings: list[str] = []
|
||
for key in (self.cache_key(customer_id), self.response_cache_key(customer_id)):
|
||
try:
|
||
await self.redis.delete(key)
|
||
except Exception as exc: # noqa: BLE001
|
||
warnings.append(f"composed_profile_cache_invalidate_failed:{type(exc).__name__}")
|
||
return warnings
|
||
|
||
async def compose_response(
|
||
self, db: AsyncSession, *, customer_id: int
|
||
) -> tuple[dict[str, Any] | None, list[str]]:
|
||
"""接口层入口：完整响应体先查 Redis（20 分钟过期），未命中才整合。
|
||
|
||
返回结构即接口 data：
|
||
{"customer_id": ..., "profile": <画像负载或 None>, "warnings": [...]}
|
||
profile 为 None（功能关闭或无任何输入）时不缓存，便于开关立即生效。
|
||
"""
|
||
key = self.response_cache_key(customer_id)
|
||
try:
|
||
raw = await self.redis.get(key)
|
||
if raw:
|
||
return json.loads(raw), []
|
||
except Exception: # noqa: BLE001 缓存读取失败视为未命中
|
||
pass
|
||
|
||
payload, warnings = await self.compose(db, customer_id=customer_id)
|
||
data = {
|
||
"customer_id": customer_id,
|
||
"profile": payload,
|
||
"warnings": warnings,
|
||
}
|
||
if payload is not None:
|
||
try:
|
||
await self.redis.set(
|
||
key,
|
||
json.dumps(data, ensure_ascii=False, default=str),
|
||
ex=self.RESPONSE_CACHE_TTL,
|
||
)
|
||
except Exception: # noqa: BLE001 缓存写入失败不影响返回
|
||
pass
|
||
return data, warnings
|
||
|
||
# ---- 配置 -----------------------------------------------------------
|
||
|
||
async def _enabled(self, db: AsyncSession, warnings: list[str]) -> bool:
|
||
"""读取功能开关；配置读取失败时默认开启，保证功能可用。"""
|
||
try:
|
||
value = await self.config_repo_cls(db).get_value(self.ENABLED_KEY, "true")
|
||
except Exception as exc: # noqa: BLE001
|
||
warnings.append(f"composed_config_failed:{type(exc).__name__}")
|
||
return True
|
||
return str(value).strip().lower() not in {"false", "0", "off"}
|
||
|
||
async def _load_weights(self, db: AsyncSession) -> tuple[dict[str, float], list[str]]:
|
||
"""读取三源权重；缺失用默认值，非法或权重和不为 1 时整体回退默认。"""
|
||
warnings: list[str] = []
|
||
try:
|
||
repo = self.config_repo_cls(db)
|
||
raw = {
|
||
name: await repo.get_value(key, str(default))
|
||
for name, key, default in (
|
||
(name, key, self.DEFAULT_WEIGHTS[name])
|
||
for name, key in self.WEIGHT_KEYS.items()
|
||
)
|
||
}
|
||
except Exception as exc: # noqa: BLE001
|
||
return dict(self.DEFAULT_WEIGHTS), [f"composed_config_failed:{type(exc).__name__}"]
|
||
|
||
weights: dict[str, float] = {}
|
||
valid = True
|
||
for name, value in raw.items():
|
||
try:
|
||
parsed = float(value)
|
||
except (TypeError, ValueError):
|
||
valid = False
|
||
break
|
||
if not 0.0 <= parsed <= 1.0:
|
||
valid = False
|
||
break
|
||
weights[name] = parsed
|
||
if not valid or abs(sum(weights.values()) - 1.0) > 0.01:
|
||
warnings.append("composed_weight_invalid")
|
||
return dict(self.DEFAULT_WEIGHTS), warnings
|
||
return weights, warnings
|
||
|
||
async def _memory_top_k(self, db: AsyncSession, warnings: list[str]) -> int:
|
||
"""读取进入整合的记忆条数上限，非法回退默认。"""
|
||
try:
|
||
raw = await self.config_repo_cls(db).get_value(
|
||
self.MEMORY_TOP_K_KEY, str(self.DEFAULT_MEMORY_TOP_K)
|
||
)
|
||
top_k = int(str(raw))
|
||
except Exception: # noqa: BLE001
|
||
return self.DEFAULT_MEMORY_TOP_K
|
||
if top_k <= 0:
|
||
warnings.append("composed_top_k_invalid")
|
||
return self.DEFAULT_MEMORY_TOP_K
|
||
return top_k
|
||
|
||
# ---- 数据与整合 ------------------------------------------------------
|
||
|
||
async def _recall_memories(
|
||
self,
|
||
db: AsyncSession,
|
||
customer_id: int,
|
||
query: str | None,
|
||
top_k: int,
|
||
warnings: list[str],
|
||
) -> list[dict[str, Any]]:
|
||
"""召回长期记忆并按综合置信分重排；失败降级为空列表。"""
|
||
try:
|
||
dtos, memory_warnings = await self.long_term.recall(
|
||
db, customer_id, limit=max(top_k, 10), query=query
|
||
)
|
||
warnings.extend(memory_warnings)
|
||
ranked = self.rank_tool.rank(
|
||
[dto.model_dump(mode="json") for dto in dtos], top_k=top_k
|
||
)
|
||
return ranked
|
||
except Exception as exc: # noqa: BLE001
|
||
warnings.append(f"composed_memory_recall_failed:{type(exc).__name__}")
|
||
return []
|
||
|
||
@staticmethod
|
||
def _slim_memories(
|
||
memories: list[dict[str, Any]] | None, top_k: int
|
||
) -> list[dict[str, Any]]:
|
||
"""裁剪记忆字段：只保留整合所需的最小集合，不带证据引用。"""
|
||
slim = []
|
||
for item in (memories or [])[:top_k]:
|
||
slim.append(
|
||
{
|
||
"tag": item.get("tag"),
|
||
"memory_type": item.get("memory_type"),
|
||
"content": item.get("content"),
|
||
"status": item.get("status"),
|
||
"confidence": item.get("confidence"),
|
||
}
|
||
)
|
||
return slim
|
||
|
||
async def _integrate(
|
||
self,
|
||
weights: dict[str, float],
|
||
profile: dict[str, Any] | None,
|
||
memories: list[dict[str, Any]],
|
||
holdings: dict[str, Any] | None,
|
||
) -> dict[str, Any]:
|
||
"""调用统一 LLM 整合三源数据，解析失败时重试一次。"""
|
||
prompt = [
|
||
{
|
||
"role": "system",
|
||
"content": self.SYSTEM_PROMPT.format(
|
||
w_profile=weights["profile"],
|
||
w_memory=weights["memory"],
|
||
w_holdings=weights["holdings"],
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": json.dumps(
|
||
{
|
||
"weights": weights,
|
||
"profile": profile,
|
||
"memories": memories,
|
||
"holdings": holdings,
|
||
},
|
||
ensure_ascii=False,
|
||
default=str,
|
||
),
|
||
},
|
||
]
|
||
last_error: Exception | None = None
|
||
for _ in range(2):
|
||
try:
|
||
# max_tokens 传 None：沿用全局 LLM_MAX_TOKENS。思考型模型（如
|
||
# qwen3）会先输出推理段，限太小会截断 JSON 导致解析失败。
|
||
text = await self.llm_client.chat(
|
||
prompt, temperature=0, max_tokens=None
|
||
)
|
||
result = self._parse_json(text)
|
||
except Exception as exc: # noqa: BLE001 网络或解析失败都重试一次
|
||
last_error = exc
|
||
logger.warning(
|
||
"composed profile integrate attempt failed: %s: %s; raw=%r",
|
||
type(exc).__name__,
|
||
exc,
|
||
(text if "text" in locals() else "")[:300],
|
||
)
|
||
continue
|
||
if isinstance(result.get("summary"), str) and result["summary"].strip():
|
||
return result
|
||
last_error = ValueError("整合结果缺少 summary 字段")
|
||
raise last_error or ValueError("LLM 整合结果非法")
|
||
|
||
_THINK_PATTERN = re.compile(r"<think>.*?</think>", re.S | re.I)
|
||
|
||
@classmethod
|
||
def _parse_json(cls, text: str) -> dict[str, Any]:
|
||
"""解析 LLM JSON 输出：剥离思考标签和围栏，容忍正文夹杂说明文字。"""
|
||
payload = text.strip()
|
||
payload = cls._THINK_PATTERN.sub("", payload).strip()
|
||
fenced = re.search(r"```(?:json)?\s*(.*?)\s*```", payload, re.S | re.I)
|
||
if fenced:
|
||
payload = fenced.group(1).strip()
|
||
try:
|
||
data = json.loads(payload)
|
||
except json.JSONDecodeError:
|
||
# 模型可能在 JSON 前后加了说明文字：提取首个 { 到最后一个 } 的片段重试
|
||
start = payload.find("{")
|
||
end = payload.rfind("}")
|
||
if start < 0 or end <= start:
|
||
raise
|
||
data = json.loads(payload[start : end + 1])
|
||
if not isinstance(data, dict):
|
||
raise ValueError("整合结果必须是 JSON 对象")
|
||
return data
|
||
|
||
@staticmethod
|
||
def _degraded(
|
||
profile: dict[str, Any] | None,
|
||
memories: list[dict[str, Any]],
|
||
holdings: dict[str, Any] | None,
|
||
) -> dict[str, Any]:
|
||
"""LLM 失败时的规则拼接摘要，保证接口始终有可用输出。"""
|
||
parts: list[str] = []
|
||
if profile:
|
||
parts.append(
|
||
f"画像：风险等级 {profile.get('risk_level') or '未知'}"
|
||
f"（评分 {profile.get('risk_score') or '未知'}）"
|
||
)
|
||
if memories:
|
||
confirmed = [m["content"] for m in memories if m.get("status") == "confirmed"]
|
||
chosen = confirmed or [m["content"] for m in memories]
|
||
parts.append("记忆：" + "；".join(chosen[:3]))
|
||
if holdings:
|
||
parts.append(
|
||
f"持仓：{holdings.get('holding_count', 0)} 只在持"
|
||
f"，总市值 {holdings.get('total_market_value', 0)}"
|
||
)
|
||
return {
|
||
"summary": "。".join(parts) if parts else "暂无可用客户信息",
|
||
"risk_tendency": (profile or {}).get("risk_level") or "",
|
||
"preference_tags": [m.get("tag") for m in memories if m.get("tag")][:5],
|
||
"behavior_traits": [],
|
||
"advice_focus": "",
|
||
"degraded": True,
|
||
"sources": {
|
||
"profile": profile is not None,
|
||
"memory": bool(memories),
|
||
"holdings": holdings is not None,
|
||
},
|
||
}
|
||
|
||
# ---- 缓存 -------------------------------------------------------------
|
||
|
||
async def _cache_get(self, customer_id: int) -> dict[str, Any] | None:
|
||
"""读取缓存；读取或解析失败视为未命中。"""
|
||
try:
|
||
raw = await self.redis.get(self.cache_key(customer_id))
|
||
if raw:
|
||
return json.loads(raw)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
return None
|
||
|
||
async def _cache_set(
|
||
self, customer_id: int, payload: dict[str, Any], ttl: int
|
||
) -> None:
|
||
"""写入缓存；失败静默（缓存只是加速手段）。"""
|
||
try:
|
||
await self.redis.set(
|
||
self.cache_key(customer_id),
|
||
json.dumps(payload, ensure_ascii=False, default=str),
|
||
ex=ttl,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
return
|
||
|
||
|
||
async def invalidate_composed_profile(customer_id: int) -> None:
|
||
"""轻量失效入口：画像回写、申购/赎回落账后调用，静默失败不阻塞业务。"""
|
||
from config.database.redis import client as redis_client
|
||
|
||
try:
|
||
redis = redis_client()
|
||
await redis.delete(ComposedProfileService.cache_key(customer_id))
|
||
await redis.delete(ComposedProfileService.response_cache_key(customer_id))
|
||
except Exception: # noqa: BLE001 缓存失效失败只影响新鲜度，不阻塞业务事务
|
||
return
|
||
|
||
|
||
__all__ = ["ComposedProfileService", "invalidate_composed_profile"]
|