Cheatsheets

Appearance

场景 3: 个性化对话助手 ​

模块：Store（长期记忆）优先级：🟡 P2（中高）业务价值：跨会话记忆，提升用户粘性和体验

一、业务背景 ​

1.1 当前痛点 ​

当前项目每次会话都是独立的，无法记住用户的偏好：

具体痛点：

  • 用户每次都要重新设置偏好（语言、风格等）
  • AI 无法记住用户的专业背景、兴趣爱好
  • 跨会话上下文丢失，体验不连贯

1.2 期望效果 ​

二、记忆设计 ​

2.1 记忆类型 ​

2.2 数据结构 ​

Store Namespace 结构:
├── (memories, user_123, preferences)  # 用户偏好
│   ├── language: {"value": "中文", "confidence": 0.9}
│   ├── style: {"value": "简洁", "confidence": 0.8}
│   └── tone: {"value": "专业", "confidence": 0.7}
│
├── (memories, user_123, facts)  # 用户事实
│   ├── fact_001: {"text": "用户是一名程序员", "source": "对话中提到"}
│   ├── fact_002: {"text": "用户喜欢玩游戏", "source": "明确说明"}
│   └── fact_003: {"text": "用户住在上海", "source": "对话推断"}
│
└── (memories, user_123, sessions)  # 会话摘要
    ├── session_001: {"summary": "讨论了 Python 异步编程", "date": "2024-01-15"}
    └── session_002: {"summary": "帮助调试了数据库问题", "date": "2024-01-20"}

三、代码实现 ​

3.1 记忆管理服务 ​

创建文件: services/memory_store.py

python
"""用户记忆管理服务

基于 LangGraph Store 实现跨会话长期记忆。
"""
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
import logging
import uuid
import os

logger = logging.getLogger(__name__)


@dataclass
class MemoryItem:
    """记忆项"""
    key: str
    text: str
    source: str  # explicit, inferred, conversation
    confidence: float
    created_at: str


class MemoryService:
    """用户记忆管理服务"""

    def __init__(self, store=None):
        """
        初始化记忆服务

        Args:
            store: LangGraph Store 实例（可选，用于测试）
        """
        self._store = store
        self._memory_cache: Dict[str, Dict] = {}  # 简单缓存

    @property
    def store(self):
        """获取 Store 实例"""
        if self._store is None:
            # 使用内存 Store（开发用）
            from langgraph.store.memory import InMemoryStore
            self._store = InMemoryStore()
        return self._store

    def _get_namespace(self, user_id: str, category: str) -> tuple:
        """获取命名空间"""
        return ("memories", user_id, category)

    # ============ 偏好管理 ============

    def set_preference(
        self,
        user_id: str,
        preference_type: str,
        value: str,
        confidence: float = 1.0
    ) -> None:
        """
        设置用户偏好

        Args:
            user_id: 用户 ID
            preference_type: 偏好类型（language, style, tone）
            value: 偏好值
            confidence: 置信度（0-1）
        """
        namespace = self._get_namespace(user_id, "preferences")
        self.store.put(
            namespace,
            preference_type,
            {"value": value, "confidence": confidence}
        )
        # 清除缓存
        self._memory_cache.pop(user_id, None)
        logger.info(f"已存储用户偏好: {user_id} -> {preference_type} = {value}")

    def get_preference(self, user_id: str, preference_type: str) -> Optional[str]:
        """获取用户偏好"""
        namespace = self._get_namespace(user_id, "preferences")
        try:
            item = self.store.get(namespace, preference_type)
            return item.value.get("value") if item else None
        except Exception:
            return None

    def get_all_preferences(self, user_id: str) -> Dict[str, str]:
        """获取用户所有偏好"""
        namespace = self._get_namespace(user_id, "preferences")
        preferences = {}
        try:
            items = self.store.search(namespace, query="", limit=10)
            for item in items:
                preferences[item.key] = item.value.get("value")
        except Exception as e:
            logger.warning(f"获取偏好失败: {e}")
        return preferences

    # ============ 事实管理 ============

    def store_fact(
        self,
        user_id: str,
        text: str,
        source: str = "explicit",
        confidence: float = 0.8
    ) -> str:
        """
        存储用户事实

        Args:
            user_id: 用户 ID
            text: 事实内容
            source: 来源（explicit=用户明确说明, inferred=推断, conversation=对话中提到）
            confidence: 置信度

        Returns:
            事实 ID
        """
        namespace = self._get_namespace(user_id, "facts")
        fact_id = str(uuid.uuid4())[:8]

        self.store.put(
            namespace,
            fact_id,
            {
                "text": text,
                "source": source,
                "confidence": confidence,
                "created_at": datetime.datetime.now().isoformat()
            }
        )

        # 清除缓存
        self._memory_cache.pop(user_id, None)
        logger.info(f"已存储用户事实: {user_id} -> {text[:50]}...")

        return fact_id

    def search_memories(
        self,
        user_id: str,
        query: str,
        limit: int = 5
    ) -> List[MemoryItem]:
        """
        搜索相关记忆

        Args:
            user_id: 用户 ID
            query: 搜索查询
            limit: 返回数量

        Returns:
            相关记忆列表
        """
        namespace = self._get_namespace(user_id, "facts")
        memories = []

        try:
            items = self.store.search(namespace, query=query, limit=limit)
            for item in items:
                memories.append(MemoryItem(
                    key=item.key,
                    text=item.value.get("text", ""),
                    source=item.value.get("source", "unknown"),
                    confidence=item.value.get("confidence", 0.5),
                    created_at=item.value.get("created_at", "")
                ))
        except Exception as e:
            logger.warning(f"搜索记忆失败: {e}")

        return memories

    # ============ 记忆上下文构建 ============

    def build_memory_context(
        self,
        user_id: str,
        current_query: str
    ) -> str:
        """
        构建记忆上下文

        用于在系统提示中添加用户相关信息。

        Args:
            user_id: 用户 ID
            current_query: 当前用户查询（用于搜索相关记忆）

        Returns:
            格式化的记忆上下文字符串
        """
        # 检查缓存
        cache_key = f"{user_id}:{hash(current_query)}"
        if cache_key in self._memory_cache:
            return self._memory_cache[cache_key]

        context_parts = []

        # 获取偏好
        preferences = self.get_all_preferences(user_id)
        if preferences:
            pref_str = ", ".join(f"{k}: {v}" for k, v in preferences.items())
            context_parts.append(f"用户偏好: {pref_str}")

        # 搜索相关事实
        relevant_facts = self.search_memories(user_id, current_query, limit=3)
        if relevant_facts:
            facts_str = "\n".join(f"- {f.text}" for f in relevant_facts)
            context_parts.append(f"关于用户的信息:\n{facts_str}")

        result = "\n\n".join(context_parts)

        # 缓存结果
        self._memory_cache[cache_key] = result

        return result

    # ============ 智能提取 ============

    def extract_and_store_from_message(
        self,
        user_id: str,
        user_message: str
    ) -> None:
        """
        从用户消息中智能提取并存储信息

        自动检测用户声明偏好和事实。
        """
        message_lower = user_message.lower()

        # 检测偏好声明
        preference_patterns = {
            "language": [
                ("我喜欢用中文", "中文"),
                ("请用中文", "中文"),
                ("用中文回复", "中文"),
                ("Speak English", "英文"),
                ("用英文", "英文"),
            ],
            "style": [
                ("简洁一点", "简洁"),
                ("简单说", "简洁"),
                ("简短回答", "简洁"),
                ("详细一点", "详细"),
                ("详细解释", "详细"),
                ("展开说", "详细"),
            ],
            "tone": [
                ("专业一点", "专业"),
                ("正式一点", "正式"),
                ("轻松一点", "轻松"),
                ("幽默一点", "幽默"),
            ]
        }

        for pref_type, patterns in preference_patterns.items():
            for pattern, value in patterns:
                if pattern.lower() in message_lower:
                    self.set_preference(user_id, pref_type, value)
                    logger.info(f"自动检测到偏好: {pref_type} = {value}")
                    return  # 一次只处理一个

        # 检测事实声明
        fact_keywords = [
            "我是", "我叫", "我住在", "我在", "我喜欢",
            "我的工作是", "我是做", "我从事", "我的专业是"
        ]

        for keyword in fact_keywords:
            if keyword in user_message:
                # 提取包含关键词的句子
                import re
                sentence = re.search(
                    rf"[^.。]*{re.escape(keyword)}[^.。]*",
                    user_message
                )
                if sentence:
                    self.store_fact(
                        user_id,
                        sentence.group(),
                        source="explicit"
                    )
                return


# 全局实例
_memory_service: Optional[MemoryService] = None


def get_memory_service() -> MemoryService:
    """获取记忆服务单例"""
    global _memory_service
    if _memory_service is None:
        _memory_service = MemoryService()
    return _memory_service

3.2 集成到聊天服务 ​

修改 services/langgraph_chat.py:

python
# 在 call_model 节点中添加记忆支持
def call_model(state: MessagesState, user_id: Optional[str] = None):
    # 构建记忆上下文
    memory_context = ""
    if user_id:
        memory_service = get_memory_service()
        query = str(state["messages"][-1].content) if state["messages"] else ""
        memory_context = memory_service.build_memory_context(user_id, query)

        # 尝试提取新记忆
        last_user_msg = next(
            (m for m in reversed(state["messages"]) if isinstance(m, HumanMessage)),
            None
        )
        if last_user_msg:
            memory_service.extract_and_store_from_message(user_id, last_user_msg.content)

    # 构建系统提示
    if memory_context:
        system_prompt = f"""{base_system_prompt}

## 关于用户的信息
{memory_context}

请根据用户的偏好和背景提供个性化的回答。"""
    else:
        system_prompt = base_system_prompt

    # ... 继续原有逻辑

四、前端集成 ​

4.1 偏好设置页面 ​

javascript
// static/js/preferences.js

class UserPreferences {
    constructor() {
        this.preferences = {};
        this.facts = [];
        this.init();
    }

    async init() {
        await this.loadPreferences();
        await this.loadFacts();
        this.render();
    }

    async loadPreferences() {
        try {
            const response = await fetch('/api/memory/preferences');
            const data = await response.json();
            if (data.success) {
                this.preferences = data.preferences;
            }
        } catch (error) {
            console.error('加载偏好失败:', error);
        }
    }

    async loadFacts() {
        try {
            const response = await fetch('/api/memory/facts?limit=10');
            const data = await response.json();
            if (data.success) {
                this.facts = data.facts;
            }
        } catch (error) {
            console.error('加载事实失败:', error);
        }
    }

    async setPreference(type, value) {
        try {
            const response = await fetch(`/api/memory/preferences/${type}`, {
                method: 'PUT',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({value})
            });

            const data = await response.json();
            if (data.success) {
                this.preferences[type] = value;
                this.showToast(`已保存偏好: ${this.getPreferenceLabel(type, value)}`);
            }
        } catch (error) {
            console.error('保存偏好失败:', error);
        }
    }

    async addFact(text) {
        try {
            const response = await fetch('/api/memory/facts', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({text, source: 'manual'})
            });

            const data = await response.json();
            if (data.success) {
                this.facts.unshift(data.fact);
                this.renderFacts();
                this.showToast('已添加记忆');
            }
        } catch (error) {
            console.error('添加事实失败:', error);
        }
    }

    render() {
        const container = document.getElementById('preferences-container');
        if (!container) return;

        container.innerHTML = `
            <div class="preferences-section">
                <h3>🎨 对话偏好</h3>

                <div class="preference-item">
                    <label>回复语言</label>
                    <select id="pref-language" onchange="userPrefs.setPreference('language', this.value)">
                        <option value="">自动检测</option>
                        <option value="中文" ${this.preferences.language === '中文' ? 'selected' : ''}>中文</option>
                        <option value="英文" ${this.preferences.language === '英文' ? 'selected' : ''}>英文</option>
                        <option value="日文" ${this.preferences.language === '日文' ? 'selected' : ''}>日文</option>
                    </select>
                </div>

                <div class="preference-item">
                    <label>回复风格</label>
                    <select id="pref-style" onchange="userPrefs.setPreference('style', this.value)">
                        <option value="">默认</option>
                        <option value="简洁" ${this.preferences.style === '简洁' ? 'selected' : ''}>简洁</option>
                        <option value="详细" ${this.preferences.style === '详细' ? 'selected' : ''}>详细</option>
                    </select>
                </div>

                <div class="preference-item">
                    <label>语气风格</label>
                    <select id="pref-tone" onchange="userPrefs.setPreference('tone', this.value)">
                        <option value="">默认</option>
                        <option value="专业" ${this.preferences.tone === '专业' ? 'selected' : ''}>专业</option>
                        <option value="轻松" ${this.preferences.tone === '轻松' ? 'selected' : ''}>轻松</option>
                        <option value="幽默" ${this.preferences.tone === '幽默' ? 'selected' : ''}>幽默</option>
                    </select>
                </div>
            </div>

            <div class="facts-section">
                <h3>🧠 关于我</h3>
                <p class="hint">AI 会记住这些信息，提供个性化回答</p>

                <div class="add-fact-form">
                    <input type="text" id="new-fact-input" placeholder="添加一条关于你的信息...">
                    <button onclick="userPrefs.addFact(document.getElementById('new-fact-input').value)">添加</button>
                </div>

                <div id="facts-list"></div>
            </div>
        `;

        this.renderFacts();
    }

    renderFacts() {
        const container = document.getElementById('facts-list');
        if (!container) return;

        container.innerHTML = this.facts.map(fact => `
            <div class="fact-item" data-id="${fact.key}">
                <span class="fact-text">${fact.text}</span>
                <span class="fact-source">${this.getSourceLabel(fact.source)}</span>
                <button class="delete-fact" onclick="userPrefs.deleteFact('${fact.key}')">×</button>
            </div>
        `).join('');
    }

    getSourceLabel(source) {
        const labels = {
            'explicit': '用户说明',
            'inferred': 'AI 推断',
            'manual': '手动添加'
        };
        return labels[source] || source;
    }

    getPreferenceLabel(type, value) {
        const labels = {
            'language': `语言: ${value}`,
            'style': `风格: ${value}`,
            'tone': `语气: ${value}`
        };
        return labels[type] || `${type}: ${value}`;
    }

    showToast(message) {
        const toast = document.createElement('div');
        toast.className = 'toast success';
        toast.textContent = message;
        document.body.appendChild(toast);
        setTimeout(() => toast.remove(), 3000);
    }
}

const userPrefs = new UserPreferences();

五、用户交互流程 ​

六、预期收益 ​

6.1 用户体验提升 ​

场景改进前改进后
语言偏好每次都要设置自动应用
回复风格需要反复强调自动适配
专业背景每次对话需要说明自动识别
个性化推荐通用推荐基于兴趣推荐

6.2 业务价值 ​

七、实施计划 ​

步骤任务预估时间
1创建 services/memory_store.py3h
2集成到 services/langgraph_chat.py2h
3添加 API 端点1h
4前端偏好设置页面3h
5测试和优化1h
总计10h (1.5天)