跨会话实体画像演进：从单轮对话到全生命周期用户认知-平芜编程栈

跨会话实体画像演进：从单轮对话到全生命周期用户认知

在构建面向 C 端陪伴型或 B 端高净值客户专属的智能体（Agent）系统时，用户与系统的交互通常分散在**跨越数月甚至数年的成百上千次独立会话（Cross-Session Dialogues）**之中：

3 个月前，用户在会话 A 中提到“我刚刚跳槽到了上海一家新能源车企”；
1 个月前，用户在会话 B 中提到“我的大儿子今年刚上小学一年级”；
昨天，用户在会话 C 中提到“我打算给家里换一辆 7 座 SUV”。

如果智能体缺乏跨会话的实体画像动态演进与聚合中枢（Cross-Session Entity Profile Evolution），每次用户新开一个会话窗口，智能体就会像得了健忘症一样从零开始，机械地重复提问“请问您在哪个城市？预算多少？”，破坏用户的信任感。

而在实现跨会话记忆时，简单的“把所有历史对话全部堆叠”会引发上下文撑爆与实体冲突。

如何构建一套**“会话结束异步实体抽取（Post-Session Entity Extraction） + 增量合并冲突消解（Incremental Conflict Resolution） + 实体生命周期置信度演变（Entity Confidence Decay）”**的全生命周期用户认知画像体系？

一、跨会话实体画像演进全景流水线

[ 会话 1 结束 (Session 1 Closed) ] ──► [ 异步提取实体: User.City="上海", User.Job="车企" ] │ ▼ (原子合并入长期图谱) ┌────────────────────────────────────────────────────────┐ │ 用户全局实体画像库 (Global Entity Profile) │ │ { │ │ "city": {"val": "上海", "confidence": 0.95}, │ │ "career": {"val": "新能源车企", "confidence": 0.90}, │ │ "family": {"children": "1个(小学生)"} │ │ } │ └──────────────────────────┬─────────────────────────────┘ │ (当 3 个月后开启会话 100 时) ▼ [ 智能体新会话冷启动 ──► 0 延迟秒级注入精炼实体画像 ──► "张先生您好，您关注的 7 座 SUV 有新降价!"]

二、生产级 Python 跨会话实体抽取与演化合并器实现实操

import json import time from typing import Dict, Any, Optional from pydantic import BaseModel, Field class ExtractedEntityAttribute(BaseModel): attribute_name: str # 如 "home_city", "programming_language", "annual_budget" attribute_value: str confidence_score: float # 0.0 ~ 1.0 evidence_quote: str class UserLifecycleProfileManager: def __init__(self, extraction_llm, persistent_db): self.llm = extraction_llm self.db = persistent_db # 持久化数据库 (PostgreSQL JSONB / Redis) def on_session_closed_async(self, user_id: str, session_chat_history: str): """当一个会话结束时，在后台异步执行实体提纯与画像演化""" print(f"🔄 【异步会话记忆提纯 🧠】正在从用户 [{user_id}] 的最新会话中提纯实体事实...") # 1. 调用大模型提取当前会话中暴露的关键用户事实与实体 prompt = f""" 你是一名资深用户画像实体抽取专家。请从以下对话记录中，提取关于用户【长期稳定特征、家庭、职业、核心偏好】的结构化实体。 绝对忽略临时琐事（如'今天中午吃了拉面'）。 【对话历史】: {session_chat_history} 请以 JSON 格式输出: {{ "entities": [ {{"attribute_name": "...", "attribute_value": "...", "confidence_score": 0.9, "evidence_quote": "..."}} ] }} """ raw_json = self.llm.generate(prompt) parsed = json.loads(raw_json.strip("`").replace("json", "")) extracted_entities = [ExtractedEntityAttribute(**item) for item in parsed.get("entities", [])] # 2. 增量合并入用户的长期全局画像 current_profile = self.db.get_user_profile(user_id) or {} for ent in extracted_entities: attr = ent.attribute_name new_val = ent.attribute_value if attr in current_profile: old_entry = current_profile[attr] if old_entry["value"] == new_val: # 事实再次验证 -> 强化置信度 old_entry["confidence"] = min(1.0, old_entry["confidence"] + 0.1) old_entry["last_confirmed_at"] = time.time() else: # 事实发生变更 -> 覆盖并重置置信度 print(f"🔄 [画像演进] 用户属性 [{attr}] 由 [{old_entry['value']}] 变更为 [{new_val}]") current_profile[attr] = { "value": new_val, "confidence": 0.7, "last_confirmed_at": time.time() } else: # 新增属性 current_profile[attr] = { "value": new_val, "confidence": ent.confidence_score, "last_confirmed_at": time.time() } # 3. 持久化落盘 self.db.save_user_profile(user_id, current_profile) print(f"🎉 【用户画像演进完毕 ✅】用户 [{user_id}] 长期画像已更新，当前包含 {len(current_profile)} 个核心特征维度。")