AI项目实战（12）全能个人助理 Agent

mem.remember_preference("preferred_room", "A101", description="开会偏爱用的房间")
mem.recall_preference("preferred_room")

"""项目九：数据模型层。

定义全能个人助理 Agent 的领域模型和数据结构。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class AssistantCapability(str, Enum):
    """助理能力枚举。"""
    EMAIL = "email"                    # 邮件管理
    CALENDAR = "calendar"              # 日程管理
    FILES = "files"                    # 文件检索
    NOTES = "notes"                    # 笔记管理
    TASKS = "tasks"                    # 任务管理
    CONTACTS = "contacts"              # 联系人管理
    REMINDERS = "reminders"            # 提醒
    PREFERENCES = "preferences"        # 用户偏好


class MessageType(str, Enum):
    """消息类型枚举。"""
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"


@dataclass
class UserPreference:
    """用户偏好设置。"""
    key: str
    value: str
    category: str = "general"
    description: str = ""
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())

    def to_dict(self) -> dict[str, str]:
        """转换为字典。"""
        return {
            "key": self.key,
            "value": self.value,
            "category": self.category,
            "description": self.description,
            "updated_at": self.updated_at,
        }


@dataclass
class CalendarEvent:
    """日历事件。"""
    title: str
    date: str
    time: str = ""
    attendees: list[str] = field(default_factory=list)
    location: str = ""
    description: str = ""
    reminder_minutes: int = 30

    def to_summary(self) -> str:
        """生成摘要。"""
        attendees_str = ", ".join(self.attendees) if self.attendees else "无参会人"
        return f"📅 {self.title} - {self.date} {self.time} | 参会人: {attendees_str}"


@dataclass
class EmailMessage:
    """邮件消息。"""
    to: str
    subject: str
    body: str
    cc: list[str] = field(default_factory=list)
    priority: str = "normal"

    def to_summary(self) -> str:
        """生成摘要。"""
        return f"📧 邮件: {self.subject} | 收件人: {self.to}"


@dataclass
class FileSearchResult:
    """文件搜索结果。"""
    filename: str
    path: str
    file_type: str
    size_bytes: int = 0
    last_modified: str = ""

    def to_summary(self) -> str:
        """生成摘要。"""
        size_mb = self.size_bytes / (1024 * 1024) if self.size_bytes > 0 else 0
        return f"📄 {self.filename} ({self.file_type}, {size_mb:.1f} MB) - {self.path}"


@dataclass
class ConversationMessage:
    """对话消息。"""
    role: MessageType
    content: str
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class AssistantContext:
    """助理上下文。

    包含所有与用户交互相关的状态和数据。
    """
    user_id: str
    conversation_history: list[ConversationMessage] = field(default_factory=list)
    pending_tasks: list[dict[str, Any]] = field(default_factory=list)
    active_context: dict[str, Any] = field(default_factory=dict)

    def add_message(self, role: MessageType, content: str, **metadata: Any) -> None:
        """添加消息到对话历史。

        Args:
            role: 消息角色
            content: 消息内容
            metadata: 额外元数据
        """
        self.conversation_history.append(ConversationMessage(
            role=role,
            content=content,
            metadata=metadata,
        ))
        # 保留最近 50 条消息
        if len(self.conversation_history) > 50:
            self.conversation_history = self.conversation_history[-50:]

    def get_recent_messages(self, limit: int = 10) -> list[ConversationMessage]:
        """获取最近的消息。

        Args:
            limit: 返回数量限制

        Returns:
            消息列表
        """
        return self.conversation_history[-limit:]

    def set_context(self, key: str, value: Any) -> None:
        """设置上下文变量。

        Args:
            key: 变量名
            value: 变量值
        """
        self.active_context[key] = value

    def get_context(self, key: str, default: Any = None) -> Any:
        """获取上下文变量。

        Args:
            key: 变量名
            default: 默认值

        Returns:
            变量值或默认值
        """
        return self.active_context.get(key, default)

"""项目九：Prompt 层。

定义全能个人助理 Agent 的系统提示词。
"""
from __future__ import annotations

# 主助理系统提示词
ASSISTANT_PROMPT = """你是用户的全能私人助理，负责高效管理用户的日常事务。

## 你的能力
1. **邮件管理**：发送邮件、起草回复、整理收件箱、设置邮件提醒
2. **日历管理**：安排会议、设置提醒、查询日程、检测时间冲突
3. **文件管理**：搜索本地文件、整理文档、汇总内容
4. **记忆系统**：记住用户偏好、学习用户习惯、跨会话记住重要信息
5. **任务管理**：创建待办、设置优先级、跟踪进度、提醒完成

## 工作原则
1. **主动获取上下文**：在回答前，先检查用户偏好和历史记忆
2. **简洁清晰**：回答简洁明了，重点突出，不啰嗦
3. **涉及费用/权限/安全时，明确确认**：对于涉及金钱、权限变更、敏感操作，必须二次确认
4. **用户隐私优先**：不泄露用户敏感信息，操作有日志可追溯
5. **超时提醒**：长时间运行的操作要给出进度反馈

## 记忆使用指南
- 遇到用户习惯相关问题时，先调用 search_memories 查找历史偏好
- 用户明确表达的偏好，调用 remember_preference 保存
- 重要决策和选择，保存到长期记忆供未来参考

## 回复风格
- 友好专业，不过于随意
- 使用适当的 emoji 增加可读性，但不过度使用
- 复杂问题分点回答
- 给出明确的行动建议，不要只说 "好的" 或 "明白了"

## 错误处理
- 工具调用失败时，友好说明并给出替代方案
- 不确定时，向用户确认，不要猜测
- 承认能力边界，不编造信息

现在，请用你的能力帮助用户。记得：主动、高效、贴心！"""


# 邮件处理提示词
EMAIL_PROMPT = """你是专业的邮件助理，擅长撰写清晰、专业的商务邮件。

## 邮件撰写原则
1. **主题明确**：收件人一眼就能知道邮件的核心内容
2. **开头问候恰当**：根据关系远近选择合适的称呼
3. **正文简洁**：每段不超过 3 行，重点用加粗或列表
4. **行动明确**：需要对方做什么、什么时候做，写清楚
5. **结尾礼貌**：合适的敬语和签名

## 邮件类型模板
### 会议邀请
> 主题：会议邀请 - [会议主题]
>
> Hi [姓名],
>
> 想邀请你参加关于 [主题] 的会议，时间：[时间]，地点：[地点/链接]。
>
> 主要议题：
> - 议题 1
> - 议题 2
>
> 请在 [日期] 前确认是否能参加。
>
> 谢谢！
> [你的名字]

### 任务跟进
> 主题：跟进 - [任务名]
>
> Hi [姓名],
>
> 想了解一下 [任务] 的进展，目前有什么需要我协助的吗？
>
> 期待你的回复。
>
> 谢谢！

## 注意事项
- 避免使用过于口语化的表达
- 重要信息前置
- 检查语法和拼写错误

请根据用户需求撰写邮件。"""


# 日程管理提示词
CALENDAR_PROMPT = """你是专业的日程管理助理，擅长安排和优化用户的时间。

## 日程安排原则
1. **避开工作时间外**：除非用户明确要求，不要在非工作时间安排会议
2. **预留缓冲**：会议之间至少留 15 分钟缓冲
3. **检测冲突**：安排前先检查是否有时间冲突
4. **提醒设置**：重要会议提前 30 分钟提醒，普通会议提前 15 分钟
5. **参会人确认**：列出所有参会人，确保没有遗漏

## 时间建议
- 上午 10:00-11:30：适合重要会议，大家精力充沛
- 下午 14:00-16:00：适合讨论和协作
- 周五下午：不适合安排重要会议

## 冲突处理
- 如果检测到冲突，列出冲突的事件
- 给出 2-3 个替代时间建议
- 说明每个建议的优缺点

请帮助用户合理安排日程。"""


# 用户偏好学习提示词
PREFERENCE_PROMPT = """你是用户偏好学习助理，负责记住并应用用户的习惯。

## 需要记住的用户偏好类型
1. **沟通偏好**：喜欢邮件还是即时消息？回复速度要求？
2. **时间偏好**：工作时间？喜欢什么时候开会？
3. **格式偏好**：喜欢长篇还是短篇？喜欢列表还是段落？
4. **内容偏好**：对什么话题感兴趣？想避免什么话题？
5. **工具偏好**：喜欢用什么工具？不喜欢用什么工具？

## 学习时机
- 用户明确说 "我喜欢/我偏好/我希望..." 时
- 用户反复纠正同一种行为时
- 用户表达满意或不满意时

## 记忆质量要求
- **key**：简洁，用英文加下划线，如 meeting_preference
- **value**：具体明确，不要模糊
- **description**：一句话说明这个偏好的含义
- **category**：分类，如 communication/time/format/content/tool

## 应用记忆
回答用户问题时，如果有相关记忆，引用记忆内容，并说明 "根据你的偏好..."

请记住并应用用户的偏好，提供更加个性化的服务。"""


# 任务管理提示词
TASK_PROMPT = """你是任务管理助理，帮助用户高效管理待办事项。

## 任务管理原则
1. **SMART 原则**：任务应该具体(Specific)、可衡量(Measurable)、可达成(Attainable)、相关(Relevant)、有时限(Time-bound)
2. **优先级**：按重要紧急程度划分：重要紧急 > 重要不紧急 > 紧急不重要 > 不紧急不重要
3. **拆解**：大任务拆成小步骤，降低执行门槛
4. **提醒**：设置合适的提醒时间
5. **进度跟踪**：定期回顾和更新状态

## 任务模板

"""项目九：工具层。

定义全能个人助理 Agent 的所有工具函数。
"""
from __future__ import annotations

from typing import Any

from langchain.tools import tool

from core.logging_conf import get_logger

from .memory import get_memory
from .models import CalendarEvent, EmailMessage, FileSearchResult

logger = get_logger("p09.assistant.tools")


# ========== 邮件工具 ==========
@tool
def send_email(to: str, subject: str, body: str, cc: str = "") -> str:
    """发送邮件（模拟实现，生产环境接入真实 SMTP 服务）。

    Args:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件正文
        cc: 抄送邮箱，多个用逗号分隔

    Returns:
        发送结果
    """
    logger.info("发送邮件: to=%s subject=%s", to, subject)

    email = EmailMessage(
        to=to,
        subject=subject,
        body=body,
        cc=cc.split(",") if cc else [],
    )

    # 生产环境：这里调用真实的邮件服务 API
    # 例如：smtplib、SendGrid、Mailgun 等

    return f"✅ {email.to_summary()}\n\n邮件已发送！"


@tool
def draft_email(recipient: str, purpose: str, key_points: str) -> str:
    """帮助用户起草邮件模板。

    Args:
        recipient: 收件人（如"客户"、"团队成员"）
        purpose: 邮件目的（如"邀请开会"、"跟进项目"）
        key_points: 要点，用逗号分隔

    Returns:
        邮件草稿
    """
    logger.info("起草邮件: 收件人=%s 目的=%s", recipient, purpose)

    points_list = [p.strip() for p in key_points.split(",") if p.strip()]
    points_text = "\n".join([f"- {p}" for p in points_list])

    draft = f"""
📧 邮件草稿
---
**主题**：{purpose}

Hi {recipient},

希望你一切都好。想和你沟通关于 {purpose} 的事情。

主要内容：
{points_text}

如果你有任何问题，随时联系我。

谢谢！
"""

    return draft.strip()


# ========== 日历工具 ==========
@tool
def schedule_meeting(title: str, date: str, attendees: str, time: str = "10:00",
                    location: str = "") -> str:
    """安排会议（模拟日历操作，生产环境接入 Google Calendar / Exchange）。

    Args:
        title: 会议标题
        date: 日期 (YYYY-MM-DD)
        attendees: 参会人，多个用逗号分隔
        time: 时间 (HH:MM)
        location: 地点或会议链接

    Returns:
        会议安排结果
    """
    logger.info("安排会议: %s on %s %s attendees=%s", title, date, time, attendees)

    event = CalendarEvent(
        title=title,
        date=date,
        time=time,
        attendees=[a.strip() for a in attendees.split(",") if a.strip()],
        location=location,
    )

    # 生产环境：这里调用真实的日历服务 API

    return f"✅ {event.to_summary()}\n\n会议已安排！提前 30 分钟提醒。"


@tool
def check_availability(date: str) -> str:
    """检查某天的日程可用性（模拟）。

    Args:
        date: 日期 (YYYY-MM-DD)

    Returns:
        日程情况
    """
    logger.info("检查日程: %s", date)

    # 生产环境：这里查询真实的日历服务

    return f"""
📅 {date} 日程情况：
---
- 09:00-10:00: 团队站会
- 11:00-12:00: 项目评审
- 14:00-16:00: [空闲] ✅ 推荐
- 16:00-17:00: [空闲] ✅ 推荐

建议：下午 2-4 点是最佳的会议时间。
""".strip()


# ========== 文件工具 ==========
@tool
def search_files(query: str) -> str:
    """搜索本地文件（模拟实现）。

    Args:
        query: 搜索关键词

    Returns:
        搜索结果
    """
    logger.info("搜索文件: %s", query)

    # 生产环境：这里调用系统搜索 API 或 Spotlight

    results = [
        FileSearchResult(
            filename=f"{query}_report.pdf",
            path=f"~/Documents/reports/{query}_report.pdf",
            file_type="PDF 文档",
            size_bytes=1024 * 1024 * 2,
        ),
        FileSearchResult(
            filename=f"{query}_draft.docx",
            path=f"~/Documents/drafts/{query}_draft.docx",
            file_type="Word 文档",
            size_bytes=1024 * 500,
        ),
    ]

    lines = [f"📄 搜索 '{query}' 找到 {len(results)} 个文件：", ""]
    for r in results:
        lines.append(r.to_summary())

    return "\n".join(lines)


@tool
def summarize_file(filename: str) -> str:
    """汇总文件内容（模拟）。

    Args:
        filename: 文件名

    Returns:
        文件摘要
    """
    logger.info("汇总文件: %s", filename)

    # 生产环境：这里读取文件内容并调用 LLM 汇总

    return f"""
📄 {filename} 摘要：
---
文件主要包含关于 {filename.replace('.pdf', '').replace('.docx', '')} 的内容。

核心要点：
- 要点 1：背景介绍
- 要点 2：主要数据
- 要点 3：结论建议

建议：阅读全文了解详细内容。
""".strip()


# ========== 记忆工具 ==========
@tool
def remember_preference(key: str, value: str, description: str = "") -> str:
    """保存用户偏好到长期记忆。

    Args:
        key: 偏好键（建议用英文下划线命名，如 meeting_preference）
        value: 偏好值
        description: 描述说明

    Returns:
        保存结果
    """
    memory = get_memory()
    memory.remember_preference(key, value, description=description)

    return f"✅ 已记住：{key} = {value}"


@tool
def recall_preference(key: str) -> str:
    """获取用户偏好。

    Args:
        key: 偏好键

    Returns:
        偏好值或提示不存在
    """
    memory = get_memory()
    value = memory.recall_preference(key)

    if value:
        return f"📝 {key} = {value}"
    return f"⚠️ 偏好 '{key}' 不存在"


@tool
def search_memories(query: str) -> str:
    """在长期记忆中搜索相关信息。

    Args:
        query: 搜索关键词

    Returns:
        匹配的记忆列表
    """
    memory = get_memory()
    preferences = memory.search_preferences(query)

    if not preferences:
        return f"🔍 未找到与 '{query}' 相关的记忆。"

    lines = [f"🔍 找到 {len(preferences)} 条与 '{query}' 相关的记忆：", ""]
    for p in preferences:
        desc = f" - {p.description}" if p.description else ""
        lines.append(f"- {p.key}: {p.value}{desc}")

    return "\n".join(lines)


@tool
def list_all_preferences() -> str:
    """列出所有已保存的用户偏好。

    Returns:
        偏好列表
    """
    memory = get_memory()
    preferences = memory.get_all_preferences()

    if not preferences:
        return "📋 暂无保存的偏好。"

    lines = [f"📋 已保存 {len(preferences)} 条偏好：", ""]
    for p in preferences:
        lines.append(f"- {p.key} = {p.value}")

    return "\n".join(lines)


# ========== 任务工具 ==========
@tool
def create_task(title: str, priority: str = "P2", due_date: str = "") -> str:
    """创建待办任务。

    Args:
        title: 任务标题
        priority: 优先级 (P0/P1/P2/P3)
        due_date: 截止日期 (YYYY-MM-DD)

    Returns:
        创建结果
    """
    logger.info("创建任务: %s 优先级=%s", title, priority)

    # 生产环境：这里调用任务管理系统 API

    due_text = f" 截止: {due_date}" if due_date else ""
    return f"✅ 任务已创建：[{priority}] {title}{due_text}"


@tool
def list_tasks(status: str = "pending") -> str:
    """列出任务（模拟）。

    Args:
        status: 状态 (pending/completed/all)

    Returns:
        任务列表
    """
    logger.info("列出任务: status=%s", status)

    # 生产环境：这里查询任务管理系统

    return f"""
📋 {status} 任务列表：
---
1. [P1] 完成项目方案 - 截止: 2024-01-20
2. [P2] 回复客户邮件 - 截止: 2024-01-18
3. [P3] 整理会议笔记

提示：使用 create_task 添加新任务。
""".strip()


# ========== 工具列表 ==========
def get_all_tools() -> list[Any]:
    """获取所有可用工具列表。

    Returns:
        工具对象列表
    """
    return [
        # 邮件工具
        send_email,
        draft_email,
        # 日历工具
        schedule_meeting,
        check_availability,
        # 文件工具
        search_files,
        summarize_file,
        # 记忆工具
        remember_preference,
        recall_preference,
        search_memories,
        list_all_preferences,
        # 任务工具
        create_task,
        list_tasks,
    ]

"""项目九：记忆层。

定义全能个人助理 Agent 的短期和长期记忆系统。
"""
from __future__ import annotations

import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Any

from core.config import DATA_DIR
from core.logging_conf import get_logger

from .models import UserPreference

logger = get_logger("p09.assistant.memory")

DB_PATH = DATA_DIR / "personal_assistant.db"


class PersonalMemory:
    """个人记忆系统。

    管理用户的短期记忆（会话内）和长期记忆（持久化）：
    - 短期记忆：当前会话的上下文、临时状态
    - 长期记忆：用户偏好、历史行为、重要决策（SQLite 持久化）
    """

    def __init__(self, db_path: Path | None = None, user_id: str = "default") -> None:
        self._user_id = user_id
        self._short_term: dict[str, Any] = {}  # 短期记忆（仅当前会话）
        self._db_path = db_path or DB_PATH
        self._init_db()

    def _init_db(self) -> None:
        """初始化数据库表结构。"""
        conn = sqlite3.connect(str(self._db_path))
        try:
            # 用户偏好表
            conn.execute("""
                CREATE TABLE IF NOT EXISTS preferences (
                    user_id TEXT NOT NULL,
                    key TEXT NOT NULL,
                    value TEXT NOT NULL,
                    category TEXT NOT NULL DEFAULT 'general',
                    description TEXT,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (user_id, key)
                )
            """)

            # 对话历史表
            conn.execute("""
                CREATE TABLE IF NOT EXISTS conversations (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id TEXT NOT NULL,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    timestamp TEXT NOT NULL,
                    metadata_json TEXT
                )
            """)

            # 任务表
            conn.execute("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id TEXT NOT NULL,
                    title TEXT NOT NULL,
                    priority TEXT NOT NULL DEFAULT 'P2',
                    status TEXT NOT NULL DEFAULT 'pending',
                    due_date TEXT,
                    created_at TEXT NOT NULL,
                    completed_at TEXT
                )
            """)

            conn.commit()
            logger.info("记忆系统初始化完成: %s", self._db_path)
        finally:
            conn.close()

    # ========== 短期记忆 ==========
    def remember_short(self, key: str, value: Any) -> None:
        """保存短期记忆（仅当前会话有效）。

        Args:
            key: 记忆键
            value: 记忆值
        """
        self._short_term[key] = value
        logger.debug("保存短期记忆: %s", key)

    def recall_short(self, key: str, default: Any = None) -> Any:
        """获取短期记忆。

        Args:
            key: 记忆键
            default: 默认值

        Returns:
            记忆值或默认值
        """
        return self._short_term.get(key, default)

    def clear_short_term(self) -> None:
        """清空短期记忆。"""
        self._short_term.clear()
        logger.debug("短期记忆已清空")

    # ========== 长期记忆（偏好） ==========
    def remember_preference(self, key: str, value: str, category: str = "general",
                           description: str = "") -> None:
        """保存用户偏好到长期记忆。

        Args:
            key: 偏好键
            value: 偏好值
            category: 分类
            description: 描述
        """
        conn = sqlite3.connect(str(self._db_path))
        try:
            conn.execute(
                """
                INSERT OR REPLACE INTO preferences
                (user_id, key, value, category, description, updated_at)
                VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    self._user_id,
                    key,
                    value,
                    category,
                    description,
                    datetime.now().isoformat(),
                ),
            )
            conn.commit()
            logger.info("保存偏好: %s = %s", key, value)
        finally:
            conn.close()

    def recall_preference(self, key: str, default: str = "") -> str:
        """获取用户偏好。

        Args:
            key: 偏好键
            default: 默认值

        Returns:
            偏好值或默认值
        """
        conn = sqlite3.connect(str(self._db_path))
        try:
            row = conn.execute(
                "SELECT value FROM preferences WHERE user_id = ? AND key = ?",
                (self._user_id, key),
            ).fetchone()
            return row[0] if row else default
        finally:
            conn.close()

    def get_all_preferences(self) -> list[UserPreference]:
        """获取所有用户偏好。

        Returns:
            偏好列表
        """
        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        try:
            rows = conn.execute(
                """
                SELECT key, value, category, description, updated_at
                FROM preferences
                WHERE user_id = ?
                ORDER BY updated_at DESC
                """,
                (self._user_id,),
            ).fetchall()

            return [
                UserPreference(
                    key=row["key"],
                    value=row["value"],
                    category=row["category"],
                    description=row["description"] or "",
                    updated_at=row["updated_at"],
                )
                for row in rows
            ]
        finally:
            conn.close()

    def search_preferences(self, keyword: str) -> list[UserPreference]:
        """搜索用户偏好。

        Args:
            keyword: 搜索关键词

        Returns:
            匹配的偏好列表
        """
        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        try:
            rows = conn.execute(
                """
                SELECT key, value, category, description, updated_at
                FROM preferences
                WHERE user_id = ?
                  AND (key LIKE ? OR value LIKE ? OR description LIKE ?)
                ORDER BY updated_at DESC
                LIMIT 20
                """,
                (self._user_id, f"%{keyword}%", f"%{keyword}%", f"%{keyword}%"),
            ).fetchall()

            return [
                UserPreference(
                    key=row["key"],
                    value=row["value"],
                    category=row["category"],
                    description=row["description"] or "",
                    updated_at=row["updated_at"],
                )
                for row in rows
            ]
        finally:
            conn.close()

    # ========== 对话历史 ==========
    def save_conversation(self, role: str, content: str, metadata: dict[str, Any] | None = None) -> None:
        """保存对话历史。

        Args:
            role: 角色 (user/assistant/system)
            content: 内容
            metadata: 额外元数据
        """
        import json

        conn = sqlite3.connect(str(self._db_path))
        try:
            conn.execute(
                """
                INSERT INTO conversations
                (user_id, role, content, timestamp, metadata_json)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    self._user_id,
                    role,
                    content,
                    datetime.now().isoformat(),
                    json.dumps(metadata or {}, ensure_ascii=False),
                ),
            )
            conn.commit()
        finally:
            conn.close()

    def get_recent_conversations(self, limit: int = 20) -> list[dict[str, Any]]:
        """获取最近的对话历史。

        Args:
            limit: 返回数量限制

        Returns:
            对话列表
        """
        import json

        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        try:
            rows = conn.execute(
                """
                SELECT role, content, timestamp, metadata_json
                FROM conversations
                WHERE user_id = ?
                ORDER BY id DESC
                LIMIT ?
                """,
                (self._user_id, limit),
            ).fetchall()

            result = []
            for row in reversed(rows):  # 按时间正序返回
                result.append({
                    "role": row["role"],
                    "content": row["content"],
                    "timestamp": row["timestamp"],
                    "metadata": json.loads(row["metadata_json"] or "{}"),
                })
            return result
        finally:
            conn.close()


# 全局记忆实例（默认用户）
_global_memory = PersonalMemory()


def get_memory(user_id: str = "default") -> PersonalMemory:
    """获取个人记忆系统实例。

    Args:
        user_id: 用户 ID，默认 "default"

    Returns:
        PersonalMemory 实例
    """
    if user_id == "default":
        return _global_memory
    return PersonalMemory(user_id=user_id)

"""项目九：服务层。

封装全能个人助理 Agent 的核心业务逻辑。
"""
from __future__ import annotations

from typing import Any

from langchain.agents import create_agent

from core import build_chat_model
from core.logging_conf import get_logger

from .memory import get_memory
from .models import AssistantContext, MessageType
from .prompts import ASSISTANT_PROMPT
from .tools import get_all_tools

logger = get_logger("p09.assistant.service")


class PersonalAssistantService:
    """个人助理服务类。

    封装个人助理的核心业务逻辑：
    - 管理用户上下文和对话历史
    - 协调各工具执行用户请求
    - 学习和应用用户偏好
    - 提供个性化服务
    """

    def __init__(self, user_id: str = "default") -> None:
        self._user_id = user_id
        self._memory = get_memory(user_id)
        self._context = AssistantContext(user_id=user_id)
        self._agent: Any | None = None

    def build_agent(self) -> Any:
        """构建个人助理 Agent。

        Returns:
            LangChain Agent 对象
        """
        if self._agent is not None:
            return self._agent

        self._agent = create_agent(
            model=build_chat_model(),
            tools=get_all_tools(),
            system_prompt=ASSISTANT_PROMPT,
        )
        return self._agent

    def process_message(self, user_message: str) -> str:
        """处理用户消息。

        Args:
            user_message: 用户输入消息

        Returns:
            助理回复
        """
        # 保存用户消息到历史
        self._context.add_message(MessageType.USER, user_message)
        self._memory.save_conversation("user", user_message)

        # 检查是否有相关偏好，加入上下文
        self._enhance_with_preferences(user_message)

        # 构建 Agent 并处理
        agent = self.build_agent()

        try:
            response = agent.run(user_message)

            # 保存助理回复到历史
            self._context.add_message(MessageType.ASSISTANT, response)
            self._memory.save_conversation("assistant", response)

            return response
        except Exception as e:
            logger.error("处理消息失败: %s", e)
            error_msg = f"抱歉，处理你的请求时遇到了问题：{e}\n请重试或换一种方式描述。"
            self._context.add_message(MessageType.SYSTEM, error_msg)
            return error_msg

    def _enhance_with_preferences(self, message: str) -> None:
        """用相关的用户偏好增强上下文。

        Args:
            message: 用户消息
        """
        # 简单的关键词匹配查找相关偏好
        keywords = ["邮件", "email", "会议", "meeting", "时间", "喜欢", "偏好"]
        matched_keywords = [k for k in keywords if k in message]

        if matched_keywords:
            # 搜索相关偏好
            for keyword in matched_keywords:
                preferences = self._memory.search_preferences(keyword)
                for pref in preferences[:3]:  # 最多取 3 个
                    self._context.set_context(f"pref_{pref.key}", pref.value)

    def learn_preference(self, key: str, value: str, description: str = "") -> None:
        """学习用户偏好。

        Args:
            key: 偏好键
            value: 偏好值
            description: 描述
        """
        self._memory.remember_preference(key, value, description=description)
        logger.info("学习用户偏好: %s = %s", key, value)

    def get_preference(self, key: str, default: str = "") -> str:
        """获取用户偏好。

        Args:
            key: 偏好键
            default: 默认值

        Returns:
            偏好值
        """
        return self._memory.recall_preference(key, default)

    def get_conversation_history(self, limit: int = 20) -> list[dict[str, Any]]:
        """获取对话历史。

        Args:
            limit: 返回数量限制

        Returns:
            对话消息列表
        """
        return self._memory.get_recent_conversations(limit)

    def clear_context(self) -> None:
        """清空当前会话上下文（保留长期记忆）。"""
        self._context.clear_short_term()
        logger.info("会话上下文已清空")

    def generate_daily_summary(self) -> str:
        """生成每日摘要。

        Returns:
            摘要文本，包含今日日程、待办任务等
        """
        from datetime import date

        today = date.today().isoformat()

        # 获取最近偏好（错误降级，避免依赖未导入符号）
        try:
            from .tools import list_all_preferences
            prefs_summary = list_all_preferences.invoke({})
        except Exception:
            prefs_summary = "暂无"

        summary = f"""
📅 {today} 每日摘要
---

## 今日日程
（调用日历 API 获取）

## 待办任务
（调用任务管理 API 获取）

## 未读邮件
（调用邮件 API 获取）

## 最近偏好
{prefs_summary}

建议：优先处理 P0/P1 任务，重要会议提前准备。
""".strip()

        return summary


# 全局服务实例（默认用户）
_service: PersonalAssistantService | None = None


def get_service(user_id: str = "default") -> PersonalAssistantService:
    """获取个人助理服务实例。

    Args:
        user_id: 用户 ID，默认 "default"

    Returns:
        PersonalAssistantService 单例
    """
    global _service
    if _service is None or _service._user_id != user_id:
        _service = PersonalAssistantService(user_id)
    return _service

"""项目九：项目定义层。

定义全能个人助理 Agent 的项目注册和对外接口。
"""
from __future__ import annotations

from typing import Any

from core import BaseProject, registry

from .service import get_service
from .tools import get_all_tools


class PersonalAssistantProject(BaseProject):
    """全能个人助理 Agent。

    主讲能力：MCP 工具生态 + 短长期记忆 + 多 Skill 调度

    业务场景：一个 Agent 统一管理邮箱、日历、文件、笔记、偏好，
    像一个真正的私人助理——记住你的喜好、操作你的系统、在你不在时处理事务。

    生产级特性：
    - MCP 生态接入：日历/邮箱/文件/数据库经 MCP 标准化
    - 短长期记忆：短期在线+长期持久化，跨会话记住用户偏好
    - 多 Skill 调度：按需加载不同技能
    - 统一 API 与 Webhook 回调
    - 个性化服务：基于历史偏好提供定制化建议
    """

    id = "p09_personal_assistant"
    name = "全能个人助理 Agent"
    description = "邮箱+日历+文件+偏好，记住你的一切，随取随用。"
    capabilities = ["MCP", "Memory", "Skill", "Tool", "Personalization"]

    def build_agent(self) -> Any:
        """构建个人助理 Agent 实例。

        Returns:
            LangChain Agent 对象
        """
        service = get_service()
        return service.build_agent()

    def run(self, message: str) -> str:
        """运行助理任务。

        Args:
            message: 用户输入的消息

        Returns:
            助理回复
        """
        service = get_service()
        return service.process_message(message)


# 项目实例
project = PersonalAssistantProject()

# 对外暴露的快捷函数
run = project.run
build_agent = project.build_agent
get_tools = get_all_tools

__all__ = [
    "PersonalAssistantProject",
    "project",
    "run",
    "build_agent",
    "get_tools",
]

"""项目九：全能个人助理 Agent

主讲能力：MCP 工具生态 + 短长期记忆 + 多 Skill 调度

业务场景：一个 Agent 统一管理邮箱、日历、文件、笔记、偏好，
像一个真正的私人助理——记住你的喜好、操作你的系统、在你不在时处理事务。

生产级特性：
- MCP 生态接入：日历/邮箱/文件/数据库经 MCP 标准化
- 短长期记忆：短期在线+长期持久化，跨会话记住用户偏好
- 多 Skill 调度：按需加载不同技能
- 统一 API 与 Webhook 回调
- 个性化服务：基于历史偏好提供定制化建议

架构分层：
- models.py: 数据模型、对象定义
- prompts.py: 系统提示词、模板
- tools.py: 各能力工具函数
- memory.py: 短期/长期记忆系统
- service.py: 业务逻辑、上下文管理
- project.py: 项目定义、对外接口
"""
from __future__ import annotations

from core import registry

# 导出 models 层
from .models import (
    AssistantCapability,
    AssistantContext,
    CalendarEvent,
    ConversationMessage,
    EmailMessage,
    FileSearchResult,
    MessageType,
    UserPreference,
)

# 导出 prompts 层
from .prompts import (
    ASSISTANT_PROMPT,
    CALENDAR_PROMPT,
    EMAIL_PROMPT,
    PREFERENCE_PROMPT,
    TASK_PROMPT,
)

# 导出 memory 层
from .memory import PersonalMemory, get_memory

# 导出 tools 层（保持向后兼容）
from .tools import (
    check_availability,
    create_task,
    draft_email,
    get_all_tools,
    list_all_preferences,
    list_tasks,
    recall_preference,
    remember_preference,
    schedule_meeting,
    search_files,
    search_memories,
    send_email,
    summarize_file,
)

# 导出 service 层
from .service import PersonalAssistantService, get_service

# 导出 project 层
from .project import (
    PersonalAssistantProject,
    build_agent,
    get_tools,
    project,
    run,
)

# 向后兼容别名：原单文件版本使用 SYSTEM_PROMPT
SYSTEM_PROMPT = ASSISTANT_PROMPT

# 注册项目到全局注册表
registry.register(project)

__all__ = [
    # models
    "AssistantCapability",
    "MessageType",
    "UserPreference",
    "CalendarEvent",
    "EmailMessage",
    "FileSearchResult",
    "ConversationMessage",
    "AssistantContext",
    # prompts
    "ASSISTANT_PROMPT",
    "SYSTEM_PROMPT",
    "EMAIL_PROMPT",
    "CALENDAR_PROMPT",
    "PREFERENCE_PROMPT",
    "TASK_PROMPT",
    # memory
    "PersonalMemory",
    "get_memory",
    # tools
    "send_email",
    "draft_email",
    "schedule_meeting",
    "check_availability",
    "search_files",
    "summarize_file",
    "remember_preference",
    "recall_preference",
    "search_memories",
    "list_all_preferences",
    "create_task",
    "list_tasks",
    "get_all_tools",
    # service
    "PersonalAssistantService",
    "get_service",
    # project
    "PersonalAssistantProject",
    "project",
    "run",
    "build_agent",
    "get_tools",
]