"""用户画像服务 - 从对话中提取用户偏好并持久化到本地md文件 内存缓存策略（参考 Claude Code 记忆模式）： - 首次读取后缓存到 _profile_cache，避免重复磁盘 I/O - 通过文件 mtime 检测外部修改，自动刷新缓存 - 写入时同时更新缓存和文件，保证读写一致性 - 每条用户画像使用 frontmatter 记录元数据（更新时间、来源会话） 画像提取逻辑委托给 ProfileExtractionAgent 子代理执行， 而非直接调用 LLM，保持多智能体架构一致性。 """ import time from pathlib import Path from typing import Optional from ..agents.profile_extraction_agent import ProfileExtractionAgent from ..services.llm_service import get_llm from ..database import get_db # 用户画像存储目录 PROFILES_DIR = Path(__file__).parent.parent.parent / "user_profiles" # 内存缓存：user_id -> (profile_text, mtime, cached_at) # mtime：文件最后修改时间（用于检测外部修改） # cached_at：缓存写入时间（用于 TTL 过期） _profile_cache: dict[int, tuple[str, float, float]] = {} # 会话级快照缓存：session_id -> context_text # 同一场对话内首条消息固话，后续消息复用，保证 LLM prompt cache 命中 _session_snapshot_cache: dict[int, str] = {} # 缓存 TTL：300 秒（5 分钟内认为缓存新鲜，无需 stat 文件） _CACHE_TTL = 300 # 画像提取 Agent 全局实例（惰性初始化） _profile_extraction_agent: Optional[ProfileExtractionAgent] = None def _ensure_profiles_dir(): """确保画像目录存在""" PROFILES_DIR.mkdir(parents=True, exist_ok=True) def _profile_path(user_id: int) -> Path: """获取用户画像文件路径""" return PROFILES_DIR / f"user_{user_id}.md" def _read_file_with_frontmatter(path: Path) -> tuple[str, str]: """ 读取 md 文件，分离 frontmatter 和正文 返回: (frontmatter_yaml, body) 无 frontmatter 时 frontmatter 返回空字符串 """ if not path.exists(): return "", "" content = path.read_text(encoding="utf-8").strip() if content.startswith("---"): parts = content.split("---", 2) if len(parts) >= 3: frontmatter = parts[1].strip() body = parts[2].strip() return frontmatter, body return "", content def _build_frontmatter(user_id: int) -> str: """构建 YAML frontmatter""" now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) return ( f"---\n" f"user_id: {user_id}\n" f"updated_at: '{now}'\n" f"---" ) def get_profile_extraction_agent() -> ProfileExtractionAgent: """获取画像提取 Agent 实例（单例，惰性初始化）""" global _profile_extraction_agent if _profile_extraction_agent is None: llm = get_llm() _profile_extraction_agent = ProfileExtractionAgent(llm) print(f" ✅ 用户画像提取 Agent 初始化成功") return _profile_extraction_agent def _invalidate_cache(user_id: int): """清除指定用户的缓存""" _profile_cache.pop(user_id, None) def _refresh_from_disk(user_id: int) -> str: """从磁盘加载用户画像正文（跳过 frontmatter），更新缓存""" path = _profile_path(user_id) if not path.exists(): _profile_cache[user_id] = ("", 0.0, time.time()) return "" mtime = path.stat().st_mtime _, body = _read_file_with_frontmatter(path) _profile_cache[user_id] = (body, mtime, time.time()) return body def load_profile_text(user_id: int) -> str: """ 加载用户画像文本（带内存缓存） 缓存策略： 1. 缓存命中且未超过 TTL → 直接返回 2. 缓存命中但超过 TTL → stat 检查文件 mtime，未变则续期缓存 3. 缓存未命中或文件已变 → 重新从磁盘读取 Returns: 用户画像正文（仅 "- " 开头的条目行），不存在则返回空字符串 """ path = _profile_path(user_id) cached = _profile_cache.get(user_id) now = time.time() if cached is not None: body, mtime, cached_at = cached # TTL 内：直接返回缓存 if now - cached_at < _CACHE_TTL: return body # TTL 已过：检查文件 mtime if path.exists(): current_mtime = path.stat().st_mtime if current_mtime == mtime: # 文件未变，续期缓存 _profile_cache[user_id] = (body, mtime, now) return body # 缓存失效或文件变更，从磁盘重新加载 return _refresh_from_disk(user_id) def save_profile(user_id: int, profile_text: str): """ 保存用户画像到 md 文件（frontmatter + 正文） 格式： --- user_id: 1 updated_at: '2026-06-04 12:00:00' --- # 用户旅行画像 - 条目1 - 条目2 """ _ensure_profiles_dir() frontmatter = _build_frontmatter(user_id) content = f"{frontmatter}\n\n# 用户旅行画像\n\n{profile_text}\n" path = _profile_path(user_id) # 先写磁盘，再更新缓存（保证缓存与磁盘一致） path.write_text(content, encoding="utf-8") mtime = path.stat().st_mtime # 只缓存有效条目行作为正文 lines = [l for l in profile_text.split("\n") if l.strip().startswith("- ")] body = "\n".join(lines) _profile_cache[user_id] = (body, mtime, time.time()) def extract_and_update_profile( user_id: int, user_message: str, history: Optional[list] = None, cross_session_context: str = "", ): """ 从用户消息中提取偏好，与现有画像对比合并（冲突时以最新为准），然后更新保存 与旧版的关键区别： 1. 支持传入跨会话上下文（cross_session_context），让 LLM 能理解 用户在其他会话中表达过的偏好，避免将长期偏好误判为一次性信息 2. 内存缓存：每次提取后自动更新缓存，后续 load 直接命中 Args: user_id: 用户ID user_message: 用户发送的消息 history: 当前会话的最近消息列表（用于理解上下文） cross_session_context: 跨会话上下文文本（来自其他会话的消息摘要） """ user_msg = user_message.strip() # 太短或纯语气词，跳过 if len(user_msg) < 3: return # 加载现有画像（走缓存） existing = load_profile_text(user_id) # 构建对话上下文 context_parts = [] # 优先注入跨会话上下文 if cross_session_context: context_parts.append("=== 历史会话摘要 ===\n" + cross_session_context) # 当前会话的最近消息作为细粒度上下文 if history: recent = history[-6:] # 最近 3 轮对话（最多 6 条） context_parts.append("=== 当前会话 ===") for msg in recent: role = msg.get("role", "") content = msg.get("content", "") if role == "user": context_parts.append(f"用户：{content[:150]}") elif role == "assistant": context_parts.append(f"助手：{content[:150]}") conversation_context = "\n".join(context_parts) if context_parts else "（无）" try: agent = get_profile_extraction_agent() new_profile = agent.extract( existing_profile=existing, conversation_context=conversation_context, user_message=user_msg[:300], ) if not new_profile: return # 解析有效条目 lines = [] for line in new_profile.split("\n"): line = line.strip() if line.startswith("- ") and len(line) > 3: lines.append(line) if lines: save_profile(user_id, "\n".join(lines)) print(f" ✅ 用户 {user_id} 画像更新成功 ({len(lines)} 条)") except Exception as e: print(f" ⚠️ 用户画像提取失败: {e}") def get_profile_context(user_id: int, session_id: int = None) -> str: """ 获取用户画像上下文文本（用于注入到系统提示词） 支持会话级快照：传入 session_id 后，同一场对话内首条消息固话画像字符串， 后续消息无论画像如何更新都复用该字符串，保证 LLM prompt cache 不变。 缓存层级（从快到慢）： session snapshot → memory cache → disk Args: user_id: 用户ID session_id: 可选，会话ID。传入后启用会话级快照。 Returns: 格式化的画像上下文，如果不存在则返回空字符串 """ # 1. 会话级快照命中 → 直接返回（零开销） if session_id is not None and session_id in _session_snapshot_cache: return _session_snapshot_cache[session_id] # 2. 加载画像（走内存缓存 → disk） profile = load_profile_text(user_id) if not profile: return "" context = ( f"\n## 关于用户\n" f"根据过往对话，我了解到该用户的一些偏好：\n{profile}\n" f"**注意：用户当前的问题/要求始终优先于历史偏好。" f"如果用户现在的说法与历史偏好矛盾，以用户现在说的为准。**\n" ) # 3. 固话到会话级快照（后续同一 session 不再变动） if session_id is not None: _session_snapshot_cache[session_id] = context return context def get_cross_session_context(user_id: int, max_sessions: int = 5, max_messages: int = 6) -> str: """ 获取用户跨会话的近期消息摘要（用于提取画像时的跨会话上下文） 查询该用户最近 N 个会话的前几条消息，拼接为纯文本返回。 这些文本不用于注入系统提示词，仅作为提取画像时的参考上下文。 Args: user_id: 用户ID max_sessions: 最多取多少个会话 max_messages: 每个会话最多取多少条消息 Returns: 格式化的跨会话上下文文本 """ conn = get_db() try: # 获取用户最近的会话 sessions = conn.execute( """SELECT id, title, created_at FROM chat_sessions WHERE user_id = ? ORDER BY updated_at DESC LIMIT ?""", (user_id, max_sessions) ).fetchall() if not sessions: return "" parts = [] for sess in sessions: sess_id = sess["id"] # 每个会话取前几条消息 messages = conn.execute( """SELECT role, content FROM chat_messages WHERE session_id = ? ORDER BY id ASC LIMIT ?""", (sess_id, max_messages) ).fetchall() if messages: msg_text = [] for msg in messages: role_label = "用户" if msg["role"] == "user" else "助手" content = msg["content"][:100] msg_text.append(f" {role_label}：{content}") parts.append( f"【会话 {sess_id}】\n" + "\n".join(msg_text) ) return "\n\n".join(parts) if parts else "" finally: conn.close()