vent_deep_agent/tools/chat_history_tools.py

# -*- coding: utf-8 -*-
"""
聊天记录查询工具模块

提供 Agent 直接查询 SQLite 聊天记录的能力，解决 Agent 用文件系统工具
找不到聊天记录的问题（聊天记录存在 SQLite，不在文件系统中）。

工具列表：
- search_chat_history: 按标题模糊搜索当前用户的会话
- get_session_chat:   获取指定会话的完整消息历史
"""

import json

from tools.vent_tools import _current_user, _current_session


async def search_chat_history(query: str, max_results: int = 5) -> str:
    """搜索当前用户的聊天历史记录（按会话标题匹配）。
    
    当用户询问"之前聊过什么""搜索历史记录""找一下关于xxx的对话"时调用。
    返回匹配的会话列表，包含标题、消息数量、会话 ID。
    
    如果需要查看某个会话的具体对话内容，再调用 get_session_chat。
    
    Args:
        query: 搜索关键词，匹配会话标题
        max_results: 最大返回条数，默认 5
    
    Returns:
        JSON 格式，包含 matching_sessions 列表。
        每条含 session_id、title、message_count、updated_at。
    """
    from db.chat_store import search_sessions as _db_search

    if not query or not query.strip():
        return json.dumps({"error": "搜索关键词不能为空"}, ensure_ascii=False)

    user_name = _current_user.get()
    try:
        sessions = _db_search(user_name, query.strip(), limit=max_results)
        results = [
            {
                "session_id": s["session_id"],
                "title": s.get("title", ""),
                "message_count": s.get("message_count", 0),
                "updated_at": s.get("updated_at", ""),
            }
            for s in sessions
        ]
        return json.dumps(
            {"query": query.strip(), "matching_sessions": results},
            ensure_ascii=False,
        )
    except Exception as e:
        return json.dumps({"error": f"搜索失败: {str(e)}"}, ensure_ascii=False)


async def get_session_chat(session_id: str, limit: int = 50) -> str:
    """获取指定会话的完整消息历史。
    
    当用户要求"查看那个会话的内容""把之前讨论的内容找出来""总结一下上次的对话"，
    且已知 session_id 时调用。通常先通过 search_chat_history 获取 session_id，
    再调用本工具读取具体内容。
    
    Args:
        session_id: 会话 ID（UUID 格式）
        limit: 最大返回消息条数，默认 50
    
    Returns:
        JSON 格式，包含 session_info（标题、创建时间等）和 messages 列表。
        每条消息含 role（user/assistant）、content、created_at。
    """
    from db.chat_store import get_messages as _db_msgs
    from db.chat_store import get_session_info as _db_info

    if not session_id or not session_id.strip():
        return json.dumps({"error": "会话 ID 不能为空"}, ensure_ascii=False)

    sid = session_id.strip()
    try:
        info = _db_info(sid)
        if not info:
            return json.dumps(
                {"error": f"会话不存在: {sid}", "session_id": sid},
                ensure_ascii=False,
            )

        msgs = _db_msgs(sid, limit=limit)
        messages = [
            {
                "role": m["role"],
                "content": m.get("content", ""),
                "created_at": m.get("created_at", ""),
            }
            for m in msgs
        ]

        return json.dumps(
            {
                "session_id": sid,
                "session_info": {
                    "title": info.get("title", ""),
                    "created_at": info.get("created_at", ""),
                    "updated_at": info.get("updated_at", ""),
                },
                "message_count": len(messages),
                "messages": messages,
            },
            ensure_ascii=False,
        )
    except Exception as e:
        return json.dumps(
            {"error": f"读取失败: {str(e)}", "session_id": sid},
            ensure_ascii=False,
        )


async def get_current_session_id() -> str:
    """获取当前本轮对话的数据库会话ID（session_id）。

    当需要知道当前正在进行的会话的唯一标识符时调用此工具。
    该 ID 可用于配合 get_session_chat 工具查看本轮会话的完整消息历史，
    或配合 search_chat_history 跨会话检索。

    无需任何参数，自动从请求上下文中获取。

    Returns:
        JSON 格式，包含当前会话的 session_id 和对应的 user_name。
    """
    user_name = _current_user.get()
    session_id = _current_session.get()
    if not session_id:
        return json.dumps({
            "error": "无法获取当前会话ID，可能不在对话上下文中",
            "session_id": "",
            "user_name": user_name,
        }, ensure_ascii=False)
    return json.dumps({
        "session_id": session_id,
        "user_name": user_name,
    }, ensure_ascii=False)