From 962a0a116f89cea462bbda375f0fb3fcb7ed491c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 21:36:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AE=B0=E5=BF=86=E2=86=92=E7=94=BB?= =?UTF-8?q?=E5=83=8F=E6=89=93=E9=80=9A=EF=BC=88=E4=BA=8B=E5=AE=9E=E6=8F=90?= =?UTF-8?q?=E5=8D=87=20+=20=E7=94=BB=E5=83=8F=E7=BB=84=E8=A3=85=20+=20?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E5=BF=AB=E7=85=A7=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补齐"记忆系统为画像服务"的断链，按 docs/23 的分层设计实现后三层。 1. 新增 app/model/profile.py：user_facts 与 profile_snapshots 的 ORM 映射。此前这两张表 只有结构、没有 Model，实际没有任何代码在用。两处表结构特例在 docstring 里显式标注， 避免后续有人按直觉写入踩坑： · user_facts.id 无 auto_increment，主键必须由应用提供（本实现用微秒时间戳，单调递增）； · profile_snapshots.current_customer_id 是生成列（IF(is_current=1, customer_id, NULL)）， 故意不映射——映射了反而会在写入时与之冲突。 2. 新增 app/service/profile_assembly_service.py，三段职责： · 事实提升（中期→长期）：evidence_count ≥ 2 或 confidence ≥ 0.90 才从 memory_unit 提炼进 user_facts —— 这条门槛就是"客户随口一说不能变成画像结论"的落地方式； · 画像组装（长期→画像）：按白名单映射进 fin_customer_profile，未列入白名单的事实 （如 profile:family）只进 user_facts，保证画像的信噪比； · 版本留痕：每次重建写一条 profile_snapshots，generation_basis 逐字段记录来源， 用于回答"当时凭什么这么判断"。 3. 新增 tools/rebuild_profile.py：手工触发入口（单客户或 --all）。画像暂无自动触发， 这是目前唯一的重建方式，也便于排查"画像为什么没更新"。 红线由代码保证而非约定：investor_type 只从 fin_risk_assessment 最新一条读取，实现中 不存在任何记忆路径能写它。实测——客户 9001 问卷为 C2、对话自述"稳健型"，重建后 investor_type 仍为 C2，自述信息进入 risk_tags 并标注"自述:"前缀。三方不一致保持可见， 但等级判定只认问卷，客户无法靠对话改变自己的可购范围。 另一处由实测修正的设计：fin_customer_profile 的 trade_account/real_name/total_asset/ behavior_score 均为 NOT NULL，说明画像行由开户流程创建（也印证了"注册时填问卷"是开户 前置条件）。原先"首次重建时创建画像行"的做法是错的——会写出一条假的开户记录，而画像 恰恰是风控要读的数据。已改为只更新已存在的画像，未开户时返回 reason=profile_row_not_opened 并如实报告，而不是静默成功。 同时新增 docs/23-记忆分层与画像设计.md：短期/中期/长期/画像四层各自存在哪里、谁写、 提升门槛、是否进画像，以及三条路径（问卷/行为/对话）在画像层汇合的设计。 验证：ruff 通过、mypy 109 文件无错；tools/rebuild_profile.py 对客户 9001 连续两次重建 产生 version=1/2 两条快照且 is_current 正确轮转（旧版本置 0）。 --- app/model/profile.py | 62 ++++++ app/service/profile_assembly_service.py | 259 ++++++++++++++++++++++++ docs/23-记忆分层与画像设计.md | 114 +++++++++++ tools/rebuild_profile.py | 74 +++++++ 4 files changed, 509 insertions(+) create mode 100644 app/model/profile.py create mode 100644 app/service/profile_assembly_service.py create mode 100644 docs/23-记忆分层与画像设计.md create mode 100644 tools/rebuild_profile.py diff --git a/app/model/profile.py b/app/model/profile.py new file mode 100644 index 0000000..6e8bf24 --- /dev/null +++ b/app/model/profile.py @@ -0,0 +1,62 @@ +"""画像与用户事实的 ORM 映射。 + +只映射既有表结构，**不改变任何字段**（`AGENTS.md` 第 3/4 条）。两处与常见表不同的设计在 +这里显式标注，避免后续有人按直觉写入而踩坑： + +1. `user_facts.id` 在库里**没有 auto_increment**，插入时必须由应用显式提供主键； +2. `profile_snapshots.current_customer_id` 是**生成列**（`IF(is_current=1, customer_id, NULL)`）， + 与唯一键 `uk_profile_snapshot_current` 共同保证「每个客户最多一条当前快照」。生成列由数据库 + 维护，因此这里**故意不映射**——映射了反而会在写入时与之冲突。 +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import CHAR, JSON, BigInteger, Boolean, DateTime, Float, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.model.base import Base + + +class UserFact(Base): + """用户事实（user_facts）：由中期记忆提升而来的稳定事实，供画像组装读取。 + + 表上**没有** (customer_id, fact_key) 唯一键（只有两个普通索引），因此"同一事实只保留一条" + 必须由服务层用"先查后写"保证，不能依赖数据库约束。 + """ + + __tablename__ = "user_facts" + + # 注意：库中该列无 auto_increment，主键由服务层显式赋值（单调递增） + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + fact_key: Mapped[str] = mapped_column(String(128), nullable=False) + fact_value: Mapped[Any] = mapped_column(JSON, nullable=False) + source_portal: Mapped[str] = mapped_column(String(16), nullable=False) + source_episode_id: Mapped[int | None] = mapped_column(BigInteger) + confidence: Mapped[float] = mapped_column(Float, nullable=False, default=1.0) + is_critical: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class ProfileSnapshot(Base): + """画像快照（profile_snapshots）：画像的版本化留痕。 + + 存在意义是回答"当时依据的是什么"——风控复盘与合规检查都需要它，所以画像变更 + **只新增版本、不原地覆盖**。 + """ + + __tablename__ = "profile_snapshots" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + profile_uuid: Mapped[str | None] = mapped_column(CHAR(36), unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + version: Mapped[int] = mapped_column(BigInteger, nullable=False) + snapshot: Mapped[Any] = mapped_column(JSON, nullable=False) + generation_basis: Mapped[Any | None] = mapped_column(JSON) + snapshot_hash: Mapped[str | None] = mapped_column(CHAR(64)) + is_current: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + generated_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + # current_customer_id 为生成列，由数据库维护，故意不映射（见模块 docstring） diff --git a/app/service/profile_assembly_service.py b/app/service/profile_assembly_service.py new file mode 100644 index 0000000..b0f6c55 --- /dev/null +++ b/app/service/profile_assembly_service.py @@ -0,0 +1,259 @@ +"""画像组装：中期记忆 → 长期事实 → 画像 + 版本快照。 + +这是"记忆系统为画像服务"的落地环节。三段职责： + +1. **事实提升（中期 → 长期）**：把 `memory_unit` 里证据足够的记忆提炼进 `user_facts`。 + 门槛是 `evidence_count >= 2` **或** `confidence >= 0.90` —— 这条门槛就是 + "客户随口一说不能变成画像结论"的落地方式。 +2. **画像组装（长期 → 画像）**：把 `user_facts` 按**白名单**映射进 `fin_customer_profile`。 + 未列入白名单的事实只进 `user_facts`，不进画像，避免画像被噪声撑大。 +3. **版本留痕**：每次重建写一条 `profile_snapshots`，并用 `generation_basis` 记录 + **每个字段分别来自哪里**——风控与合规复盘时要能回答"当时凭什么这么判断"。 + +## 两条必须由代码保证的红线 + +- **`investor_type` 只来自问卷测评**（`fin_risk_assessment` 最新一条）。下面的实现里 + 它只从问卷查询取数，任何记忆路径都碰不到它。客户在对话里说"我是激进型"不会改变它 + ——这是合规底线，不能只靠约定。 +- **按字段所有权写入**：交易侧的客观字段（`total_asset`/`trading_frequency`/`behavior_score`） + 本服务**不写**，留给交易模块，避免两个模块抢写同一列。 +""" + +import json as _json +from datetime import UTC, datetime +from hashlib import sha256 +from typing import Any +from uuid import uuid4 + +from sqlalchemy import or_, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.model.fund import FundCustomerProfile +from app.model.memory import MemoryUnit +from app.model.profile import ProfileSnapshot, UserFact + +# 提升门槛 +MIN_EVIDENCE = 2 +HIGH_CONFIDENCE = 0.90 + +# 事实键 → 画像字段的**白名单**映射。没列在这里的事实（如 profile:family）只进 user_facts， +# 不进画像字段——画像要保持"能直接支撑决策"的信噪比。 +FACT_TO_PROFILE_FIELD: dict[str, str] = { + "preference:asset_class": "preferred_asset_class", + "preference:horizon": "investment_horizon", +} + +# 自述类事实（客户自己说的偏好）统一进 risk_tags，并标注来源为"自述"。 +# 保留它们的价值在于：当出现「问卷 C4 / 自述稳健 / 行为买 R4」三方不一致时， +# 这种矛盾本身就是风控信号——但绝不能与问卷等级混进同一个字段。 +SELF_REPORTED_PREFIXES = ("preference:risk_level", "preference:", "profile:") + +# 关键事实：参与决策，标记出来便于下游优先读取 +CRITICAL_FACTS = frozenset({ + "preference:risk_level", "preference:horizon", "preference:asset_class", +}) + +# 画像中允许本服务写入的字段（其余字段归交易/注册侧所有） +PROFILE_OWNED_FIELDS = ("investor_type", "preferred_asset_class", "investment_horizon", "risk_tags") + + +def _now() -> datetime: + return datetime.now(UTC).replace(tzinfo=None) + + +def _fact_id() -> int: + """`user_facts.id` 没有 auto_increment，主键由应用生成。 + + 用微秒时间戳：单调递增、无需额外序列、同客户同微秒重复在单进程写入下不可能发生。 + """ + return int(datetime.now(UTC).timestamp() * 1_000_000) + + +class ProfileAssemblyService: + def __init__(self, session: AsyncSession) -> None: + self.session = session + + # ---------- 中期 → 长期 ---------- + + async def promote_facts(self, customer_id: int) -> list[str]: + """把证据足够的记忆提炼为长期事实；返回本次提升的事实键。""" + now = _now() + rows = list(await self.session.scalars( + select(MemoryUnit).where( + MemoryUnit.customer_id == customer_id, + MemoryUnit.status == "active", + or_( + MemoryUnit.evidence_count >= MIN_EVIDENCE, + MemoryUnit.confidence >= HIGH_CONFIDENCE, + ), + or_(MemoryUnit.valid_until.is_(None), MemoryUnit.valid_until > now), + ) + )) + promoted: list[str] = [] + for memory in rows: + key = str(memory.memory_key) + value = self._fact_value(memory) + existing = await self.session.scalar( + select(UserFact).where( + UserFact.customer_id == customer_id, UserFact.fact_key == key + ) + ) + if existing is None: + self.session.add(UserFact( + # 主键显式赋值：该表无 auto_increment + id=_fact_id(), + customer_id=customer_id, + fact_key=key, + fact_value=value, + source_portal=str(memory.source_type or "conversation"), + source_episode_id=None, + confidence=float(memory.confidence or 0.0), + is_critical=key in CRITICAL_FACTS, + created_at=now, + )) + else: + existing.fact_value = value + existing.confidence = float(memory.confidence or 0.0) + existing.is_critical = key in CRITICAL_FACTS + promoted.append(key) + await self.session.flush() + return promoted + + @staticmethod + def _fact_value(memory: MemoryUnit) -> Any: + """事实值优先取结构化值，回退到正文；始终以 JSON 可存的形式返回。""" + structured = memory.structured_value + if isinstance(structured, dict) and "value" in structured: + return structured["value"] + if structured is not None: + return structured + return memory.content or "" + + # ---------- 长期 → 画像 ---------- + + async def rebuild_profile(self, customer_id: int) -> dict[str, Any]: + """用长期事实 + 问卷重建画像，并写一条版本快照。""" + now = _now() + facts = list(await self.session.scalars( + select(UserFact).where(UserFact.customer_id == customer_id) + )) + assessment = (await self.session.execute(text( + """ + SELECT investor_type, questionnaire_version, assessed_at, valid_until + FROM fin_risk_assessment + WHERE customer_id = :customer_id + ORDER BY assessed_at DESC, id DESC + LIMIT 1 + """ + ), {"customer_id": customer_id})).first() + + values: dict[str, Any] = {} + basis: dict[str, Any] = {} + + # 红线：风险等级只从问卷取；记忆里哪怕有 preference:risk_level 也不写这个字段 + if assessment is not None and assessment[0]: + values["investor_type"] = str(assessment[0]) + basis["investor_type"] = { + "source": "fin_risk_assessment", + "questionnaire_version": assessment[1], + "assessed_at": str(assessment[2]), + "valid_until": str(assessment[3]), + } + + tags: list[str] = [] + for fact in facts: + key = str(fact.fact_key) + field = FACT_TO_PROFILE_FIELD.get(key) + if field is not None: + values[field] = self._as_text(fact.fact_value) + basis[field] = { + "source": "user_facts", "fact_key": key, + "confidence": float(fact.confidence or 0.0), + } + elif key.startswith(SELF_REPORTED_PREFIXES): + # 自述信息进标签，并显式标注"自述"，与问卷等级区分开 + tags.append(f"自述:{key}={self._as_text(fact.fact_value)}") + basis.setdefault("risk_tags", {"source": "user_facts", "items": []}) + basis["risk_tags"]["items"].append(key) + if tags: + values["risk_tags"] = "；".join(tags) + + profile = await self.session.get(FundCustomerProfile, customer_id) + if profile is None: + # 画像行由开户流程创建：`trade_account` 等身份字段在库里是 NOT NULL，属注册／账户侧 + # 所有。本服务不代替开户去造这些数据——否则会写出一条**假的**开户记录， + # 而画像恰恰是风控要读的东西，假数据比没有数据更危险。未开户时如实报告。 + return { + "profile": None, + "reason": "profile_row_not_opened", + "generation_basis": basis, + "promoted": len(facts), + } + for field in PROFILE_OWNED_FIELDS: + if field in values: + setattr(profile, field, values[field]) + profile.updated_at = now + await self.session.flush() + + snapshot = { + **{field: getattr(profile, field, None) for field in PROFILE_OWNED_FIELDS}, + "generated_at": now.isoformat(), + } + await self._write_snapshot(customer_id, snapshot, basis, now) + return {"profile": snapshot, "generation_basis": basis, "promoted": len(facts)} + + @staticmethod + def _as_text(value: Any) -> str: + """把 JSON 列里取出的值渲染成可读字符串。 + + 字符串类型的值可能带着 JSON 序列化时的外层引号（取决于驱动如何回读 JSON 列）， + 这里去掉它们——`risk_tags` 是给风控与投顾看的，多一对引号会让人以为值本身包含引号。 + """ + if isinstance(value, str): + return value.strip().strip('"') + return _json.dumps(value, ensure_ascii=False) + + async def _write_snapshot( + self, customer_id: int, snapshot: dict[str, Any], basis: dict[str, Any], now: datetime + ) -> None: + """写入新版本快照并把旧版本置为非当前。 + + 唯一键 `uk_profile_snapshot_current` 建立在生成列 `current_customer_id` 上， + 保证「每个客户最多一条 current」；因此必须先清旧再写新，顺序不能反。 + """ + previous = list(await self.session.scalars( + select(ProfileSnapshot).where( + ProfileSnapshot.customer_id == customer_id, ProfileSnapshot.is_current.is_(True) + ) + )) + for row in previous: + row.is_current = False + row.updated_at = now + await self.session.flush() + + latest = await self.session.scalar(text( + "SELECT COALESCE(MAX(version), 0) FROM profile_snapshots WHERE customer_id = :cid" + ), {"cid": customer_id}) + version = int(latest or 0) + 1 + payload = _json.dumps(snapshot, ensure_ascii=False, sort_keys=True) + self.session.add(ProfileSnapshot( + profile_uuid=str(uuid4()), + customer_id=customer_id, + version=version, + snapshot=snapshot, + generation_basis=basis, + snapshot_hash=sha256(payload.encode("utf-8")).hexdigest(), + is_current=True, + generated_at=now, + created_at=now, + updated_at=now, + )) + await self.session.flush() + + # ---------- 完整链路 ---------- + + async def rebuild(self, customer_id: int) -> dict[str, Any]: + promoted = await self.promote_facts(customer_id) + outcome = await self.rebuild_profile(customer_id) + outcome["promoted_keys"] = promoted + return outcome diff --git a/docs/23-记忆分层与画像设计.md b/docs/23-记忆分层与画像设计.md new file mode 100644 index 0000000..045603e --- /dev/null +++ b/docs/23-记忆分层与画像设计.md @@ -0,0 +1,114 @@ +# 记忆分层与画像设计 + +> 目的：讲清短期／中期／长期／画像四层**各自存在哪里、谁写、什么条件下向上提升、是否进画像**。 +> 全部基于底座已有的真实表，不是另起一套；每节末尾标注实现现状。 + +## 0. 一句话 + +``` +短期（会话上下文）→ 中期（候选事实）→ 长期（稳定事实）→ 画像（决策依据） +``` + +**逐层提高门槛**，是为了让"随口一说"永远变不成画像结论； +**三条路径（问卷／行为／对话）在画像层汇合**，但按字段划分所有权，不互相覆盖。 + +## 1. 总览 + +| 层 | 存在哪里 | 装什么 | 谁写 | 提升门槛 | 进画像 | +| --- | --- | --- | --- | --- | --- | +| **短期** | Redis 会话列表（设计）；`conversation_message` 表（现状只有存档） | 当前会话最近的对话轮次 | 每次 run 追加 | — | ❌ 不进 | +| **中期** | `memory_unit` + `memory_evidence` | 从对话抽出的候选事实，带证据链 | Worker 记忆抽取 | `evidence_count ≥ 2` 或 `confidence ≥ 0.90` | ✅ 过门槛后提升 | +| **长期** | `user_facts`（`is_critical` 标关键事实） | 已确认的稳定事实 | 由中期提升 | 由 `user_facts` 判定 | ✅ 唯一用途就是喂画像 | +| **画像** | `fin_customer_profile`（当前）+ `profile_snapshots`（版本化，带 `generation_basis`） | 整合三路来源的客户视图 | 画像组装服务 | — | 它就是画像 | + +## 2. 三条路径在画像汇合 + +这是整套设计的核心结构。**只有对话路径需要累积证据**，另外两条是权威／客观数据，直接写： + +``` +对话消息 ──→ [短期] 会话上下文 ──→ Agent 用它理解指代（"它""那个"） + │ + └─(run 完成，发 memory.extraction_requested 事件) + ↓ + [中期] memory_unit + memory_evidence + │ 证据累积过门槛 + ↓ + [长期] user_facts + │ + │ ┌── 问卷测评（权威）────┐ + └──组装──────┤ ├──→ [画像] + └── 行为／流水（客观）──┘ + │ + 投顾 / 风控 / 客服 +``` + +**为什么问卷和行为不需要走前三层**：它们是**权威**（问卷由客户答题、系统评分）与**客观**（交易记录无法自称）数据，不需要"多次出现才可信"这层保护；对话自述才有"记错、修饰、被引导"的风险。 + +## 3. 各层细节 + +### 3.1 短期：会话上下文 + +- **装什么**：当前会话最近若干轮对话，用于解析指代与保持连贯。 +- **怎么工作**：run 开始时按 `session_id` 读出历史 → 与当前消息一起交给意图分类与回答生成；run 结束时把本轮追加进去。 +- **生命周期**：会话结束或长时间无活动即失效（方案 §2.2 给的是 TTL 30min、最长 24h、超 4096 token 截断旧消息）。 +- **为什么不进画像**：会话上下文是"他刚才说了什么"，不是"他是什么样的人"。把临时对话当画像会造成画像抖动。 +- **实现现状**：❌ **未实现**。`conversation_message` 表存了全部消息、`svc_conversation_session.message_count` 只在计数，但**没有任何"加载最近 N 条"的代码**，run 时只拿到当前这一条消息。后果是多轮指代无法解析（问"那它风险高吗"无法知道"它"指谁）。 + +### 3.2 中期：候选事实 + +- **装什么**：从对话里抽出的事实（风险偏好自述、投资期限、家庭情况、资产类别偏好等），每条都带证据。 +- **怎么工作**： + - 写入：run 完成 → 发布 `memory.extraction_requested` → **Worker 消费事件** → 调模型抽取 → 写 `memory_unit`，同时写一条 `memory_evidence`（含原文摘录，可回溯"这句话从哪来"）。 + - 读取：run 开始时 `recall_memory` 自动召回，`recall_count` 累加。 + - 演进：同一事实再次出现 → `evidence_count` 增加；出现相反证据 → `conflict_count` 增加（**不直接覆盖**，留冲突待判）。 +- **生命周期**：`valid_from` / `valid_until`；过期或长期无新证据即降级／失效（`memory_lifecycle_service` 负责级联失效）。 +- **进画像的条件**：`evidence_count ≥ 2` **或** `confidence ≥ 0.90` —— 这条门槛就是"一次性说法不该成为画像结论"的落地方式。 +- **实现现状**：✅ **已实现并可验证**。实测：客户说"我的风险偏好是稳健型，平时只买债券基金" → 抽出 `preference:risk_level = 稳健型`（置信 0.95）+ 1 条证据。 +- **⚠️ 运维前提**：抽取依赖**常驻 Worker** 消费 Outbox 事件。只起 API 服务不起 Worker，事件会一直堆在 `pending`，记忆永远不产生。 + +### 3.3 长期：稳定事实 + +- **装什么**：经过证据累积确认的稳定事实，`is_critical` 标记其中对决策关键的那些（如风险偏好、投资期限）。 +- **怎么工作**：只由中期提升而来（不直接从对话写）；`source_portal` 记录事实来自哪个入口，`source_episode_id` 记录来自哪个会话片段。 +- **为什么不直接从对话写**：长期层是画像的输入，必须保证"这条结论经过验证"，否则画像会被一次性说法污染。 +- **进画像**：✅ 它的存在意义就是喂画像。 +- **实现现状**：❌ **表在（9 列），无生成代码**，当前 0 行。 + +### 3.4 画像：决策依据 + +- **装什么**：整合三路来源后的客户视图，是投顾、风控、客服共同读取的**唯一**客户数据入口。 +- **怎么工作**：画像组装服务按字段所有权写入 → `fin_customer_profile` 保存当前值 → 每次重建写一条 `profile_snapshots`（带 `version`、`generation_basis`、`snapshot_hash`、`is_current`），从而**每次变化都可追溯**。 +- **字段所有权（关键：按字段切开，避免互相覆盖）**： + + | 字段 | 唯一写入方 | 性质 | + | --- | --- | --- | + | `investor_type`（C1–C5） | **只有问卷测评** | 权威、合规硬约束 | + | `total_asset`、`trading_frequency`、`behavior_score` | 交易／流水侧 | 客观行为 | + | `preferred_asset_class`、`investment_horizon`、`risk_tags` | 记忆系统（经长期层） | 软信息、需累积 | + | `real_name`、`birth_date`、`occupation`、`mobile_masked`、`opened_at` | 注册／账户流程 | 身份 | + +- **客户自述风险偏好放哪**：放 `risk_tags` 并标注"客户自述"，**不得写入 `investor_type`**。 + 保留它的价值在于：当出现「问卷 C4 / 自述稳健 / 行为买 R4」三方不一致时，**这种矛盾本身就是风控信号**。 +- **实现现状**：❌ **两张表都在，无组装代码**，当前 0 行。 + +## 4. 红线（必须由代码保证，不能只靠约定） + +1. **适当性判定只认问卷**。`investor_type` 只能由问卷写入，记忆与对话在任何情况下都不得修改它。客户在对话里说"我是激进型"不能让他买到 R5。 +2. **风控与投顾只读画像**，不直接读 `memory_unit`。否则同一条记忆会被多处按各自口径解释。 +3. **每条画像字段都要能回答"凭什么"**。写入时通过 `generation_basis` 记录依据来源。 +4. **画像变更留版本**，不原地覆盖 —— 风控复盘时需要"当时看到的是什么"。 +5. **内部资料不进入面向客户的知识库**（已按此处理：反洗钱手册 `visibility=internal`）。 + +## 5. 现状与缺口一览 + +| 环节 | 状态 | +| --- | --- | +| 对话 → 中期记忆（抽取 + 证据 + 可回溯） | ✅ 已实现（依赖常驻 Worker） | +| 中期 → 长期（证据累积与门槛） | ❌ 待实现 | +| 长期 → 画像（组装 + 版本快照） | ❌ 待实现 | +| 问卷 → 画像 | ❌ 待实现（注册问卷流程未做） | +| 行为 → 画像 | ❌ 待实现（交易模块未做） | +| 短期会话上下文 | ❌ 待实现（影响多轮指代） | +| 画像 → 投顾 / 风控 | ❌ 待实现 | + +**建议顺序**：先做「中期 → 长期 → 画像」这一段（现在就有记忆数据可验证），再接问卷与行为两路，最后接下游消费者。 diff --git a/tools/rebuild_profile.py b/tools/rebuild_profile.py new file mode 100644 index 0000000..a0127e4 --- /dev/null +++ b/tools/rebuild_profile.py @@ -0,0 +1,74 @@ +"""手工重建客户画像：中期记忆 → 长期事实 → 画像 + 版本快照。 + +用法： + +```powershell +# 重建一个客户 +python tools/rebuild_profile.py 9001 + +# 重建全部有记忆的客户 +python tools/rebuild_profile.py --all +``` + +为什么需要这个入口：画像组装目前没有自动触发（记忆写入后不会立刻重建画像）， +在把它接上 Worker 事件之前，这是唯一的触发方式，也便于运维排查"画像为什么没更新"。 + +关于输出里的 `reason=profile_row_not_opened`：客户尚未开户时不会创建画像行 +（`trade_account` 等身份字段是 NOT NULL，属注册／开户流程所有），这是**正确行为**而非失败。 +""" + +import argparse +import asyncio +import sys + +from sqlalchemy import select + +from app.infrastructure.db import SessionFactory +from app.model.memory import MemoryUnit +from app.service.profile_assembly_service import ProfileAssemblyService + +sys.stdout.reconfigure(errors="replace") + + +async def rebuild_one(customer_id: int) -> None: + async with SessionFactory() as session: + async with session.begin(): + outcome = await ProfileAssemblyService(session).rebuild(customer_id) + print(f"\n== 客户 {customer_id} ==") + if outcome.get("profile") is None: + print(f" 未重建画像：{outcome.get('reason')}" + f"（尚未开户；已提升事实 {len(outcome.get('promoted_keys') or [])} 条）") + else: + print(f" 提升事实：{outcome.get('promoted_keys')}") + print(f" 画像内容：{outcome.get('profile')}") + print(f" 生成依据：{outcome.get('generation_basis')}") + + +async def main() -> int: + parser = argparse.ArgumentParser(description="重建客户画像") + parser.add_argument("customer_id", nargs="?", type=int, help="客户 id") + parser.add_argument("--all", action="store_true", help="重建全部有记忆的客户") + args = parser.parse_args() + + if args.all: + async with SessionFactory() as session: + rows = list(await session.scalars( + select(MemoryUnit.customer_id).distinct() + )) + targets = [int(row) for row in rows] + if not targets: + print("没有任何客户有记忆数据，无需重建") + return 0 + print(f"将重建 {len(targets)} 个客户：{targets}") + for customer_id in targets: + await rebuild_one(customer_id) + return 0 + + if args.customer_id is None: + parser.print_help() + return 2 + await rebuild_one(args.customer_id) + return 0 + + +sys.exit(asyncio.run(main()))