vent_agent_3/service/history_manager.py

import json
import os
from typing import List, Dict

from config.settings import HISTORY_DIR, TITLE_GENERATOR_PROMPT
from core.agent_builder import model

os.makedirs(HISTORY_DIR, exist_ok=True)


def _session_file_path(session_id: str) -> str:
    return os.path.join(HISTORY_DIR, f"{session_id}.txt")


def _generate_title(user_query: str) -> str:
    """Ask the LLM to produce a short title for a conversation."""
    prompt = f"{TITLE_GENERATOR_PROMPT}{user_query}"
    return model.invoke(prompt).content.strip()


def get_or_create_title(session_id: str, user_message: str) -> str:
    """Return the existing title for *session_id*, or create one from *user_message*.

    New sessions get a title generated by the LLM and persisted as the first
    line of the history file.  Existing sessions re-read that first line.
    """
    file_path = _session_file_path(session_id)
    if os.path.exists(file_path):
        with open(file_path, "r", encoding="utf-8") as file_handle:
            return file_handle.readline().strip()

    title = _generate_title(user_message)
    with open(file_path, "w", encoding="utf-8") as file_handle:
        file_handle.write(title + "\n")
    return title


def append_line(session_id: str, event_type: str, content: str) -> None:
    """Append a single JSON log entry to the session history file."""
    file_path = _session_file_path(session_id)
    entry = json.dumps({"type": event_type, "content": content}, ensure_ascii=False) + "\n"
    # Touch the file first to ensure the title line exists
    with open(file_path, "a", encoding="utf-8"):
        pass
    with open(file_path, "a", encoding="utf-8") as file_handle:
        file_handle.write(entry)


def fetch_all_sessions() -> List[Dict[str, str]]:
    """Return every session (id + title), sorted by last-modified time descending."""
    session_entries: List[Dict] = []
    for filename in os.listdir(HISTORY_DIR):
        if not filename.endswith(".txt"):
            continue
        session_id = filename[:-4]
        file_path = os.path.join(HISTORY_DIR, filename)

        try:
            modified_time = os.path.getmtime(file_path)
        except OSError:
            continue

        try:
            with open(file_path, "r", encoding="utf-8") as file_handle:
                title = file_handle.readline().strip()
        except Exception:
            title = "未知会话"

        session_entries.append({
            "session_id": session_id,
            "intent": title,
            "mtime": modified_time,
        })

    session_entries.sort(key=lambda entry: entry["mtime"], reverse=True)
    return [
        {"session_id": entry["session_id"], "intent": entry["intent"]}
        for entry in session_entries
    ]


def fetch_session_detail(session_id: str) -> List[Dict]:
    """Return all log entries for *session_id*, excluding the title line."""
    file_path = _session_file_path(session_id)
    if not os.path.exists(file_path):
        return []
    entries: List[Dict] = []
    with open(file_path, "r", encoding="utf-8") as file_handle:
        lines = file_handle.readlines()
    for line in lines[1:]:
        stripped = line.strip()
        if not stripped:
            continue
        try:
            entries.append(json.loads(stripped))
        except json.JSONDecodeError:
            continue
    return entries


def remove_session_file(session_id: str) -> bool:
    """Delete the history file for *session_id*. Returns True on success."""
    file_path = _session_file_path(session_id)
    if not os.path.exists(file_path):
        return False
    try:
        os.remove(file_path)
        return True
    except OSError as exc:
        print(f"删除会话失败: {exc}")
        return False