LongjoeAgent/core/agent.py

import os
import dotenv
import datetime
from pathlib import Path

from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, BaseMessage
from tools.tool_factory import get_all_tools
from langchain_core.runnables import RunnableConfig
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any, List, Sequence
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
import sqlite3
from config.settings import settings

dotenv.load_dotenv()


def create_system_prompt(
    backend_url: str = "", token: str = "", username: str = "default"
) -> str:
    """
    创建动态的system_prompt，支持参数化配置

    Args:
        backend_url: 后端API地址
        token: 访问后端的认证令牌
        username: 用户名
    """
    # 判断token状态
    if token:
        token_status = "已配置有效的认证令牌，可以调用后端API获取用户数据"
    else:
        token_status = "未提供认证令牌，后端API调用可能受限"

    # 判断backend_url状态
    if backend_url:
        backend_status = f"已配置后端地址: {backend_url}"
    else:
        backend_status = "未配置后端地址，只能访问知识库"

    system_prompt = f"""你是属于龙嘉软件公司的AI助手，名字叫小龙。

# 当前会话信息
- 当前用户: {username}
- 后端服务状态: {backend_status}
- 认证状态: {token_status}
现在时间是{datetime.datetime.now().isoformat()}

# 核心能力
你可为客户提供ERP问题的解决方案，也可回答与龙嘉软件相关的问题。软件面向全球客户，你需按用户提问的语言回答。

# 工作流程
1. 分析用户问题的意图，提取关键词
2. 根据意图及关键词调用相应工具
3. 可以访问知识库工具，也可以调用后端API获取数据

# 后端API使用指南
{"- 当用户需要查询个人数据、订单信息、账户状态时，可使用后端API" if backend_url and token else "- 由于缺少认证信息，暂时无法调用后端API"}
{"- 后端地址: " + backend_url if backend_url else ""}
{"- API调用会自动包含用户的认证令牌: " + token if token else ""}

# 知识库搜索规则
- 判断问题所属模块（销售、采购、生产、财务、仓储、权限等）并纳入关键词
- 文章匹配要精准，例如"销售订单新建权限"，拆分为："销售订单"、"新建"、"权限"
- 避免使用"的"、"地"、"得"、"了"、"在"等无意义词汇
- 关键词可以多个，要判断问题属于哪个模块并将其纳入关键字
- 如果匹配文章太少（少于3篇），尝试以下方法：
  a) 变更关键字（同义词、近义词）
  b) 把关键字拆得更细
  c) 扩大搜索范围（减少关键词数量）
  d) 重新搜索
- 获取到文章列表后，用工具获取文章内容然后回答用户问题

# 回答策略
- 优先使用知识库中的准确信息
- 如果知识库中有相关文章，结合文章内容进行回答
- 如果需要实时数据且认证有效，可调用后端API
- 如果找不到对应知识库文章，向客户说明："我正在学习这个问题的解决方案，很快就能正式为您服务"
- 如果用户的问题需要后端数据但认证无效，提示："查看个人数据需要登录验证，请确保已提供正确的访问令牌"

# 注意事项
- 保护用户隐私，不在回复中暴露敏感信息
- 如果API调用失败，提供友好的错误信息
- 保持回答的专业性和准确性
- 对于不确定的问题，可以建议用户联系客服或技术支持
"""
    # print(system_prompt)
    return system_prompt


def get_day_number(date=None):
    """获取日期编号 (YYYYMMDD 格式)"""
    if date is None:
        date = datetime.datetime.now()
    return date.strftime("%Y%m%d")  # 格式: 20251229


def get_sqlite_checkpointer():
    """创建按天分割的SQLite检查点保存器"""
    try:
        from langgraph.checkpoint.sqlite import SqliteSaver

        # 获取当前日期编号
        current_day = get_day_number()

        # 数据库文件存放目录
        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        base_dir = os.path.join(project_root, "data", "checkpoints")
        os.makedirs(base_dir, exist_ok=True)

        # 数据库文件名格式: checkpoints_20251229.db
        db_filename = f"checkpoints_{current_day}.db"
        db_path = os.path.join(base_dir, db_filename)
        # checkpointer = SqliteSaver.from_conn_string(db_path)

        conn = sqlite3.connect(db_path, check_same_thread=False)
        conn.execute("PRAGMA wal_autocheckpoint=500")  # 2MB 就提交
        conn.execute("PRAGMA journal_size_limit=52428800")  # 最大 50MB
        checkpointer = SqliteSaver(conn)

        return checkpointer

    except Exception as e:
        print(f"❌❌ 创建 SQLite 检查器失败: {e}")
        import traceback

        traceback.print_exc()

        # 回退到内存保存器
        from langgraph.checkpoint.memory import InMemorySaver

        print("⚠️ 使用 InMemorySaver 作为回退")
        return InMemorySaver()


def cleanup_old_checkpoints(max_days=7):
    """清理超过指定天数的旧检查点文件（可选功能）"""
    try:
        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        base_dir = os.path.join(project_root, "data", "checkpoints")
        if not os.path.exists(base_dir):
            return

        # 获取当前日期
        current_date = datetime.datetime.now()

        # 遍历目录中的所有.db文件
        for filename in os.listdir(base_dir):
            if filename.startswith("checkpoints_") and filename.endswith(".db"):
                try:
                    print(f"检查旧检查点文件: {filename}")
                    # 提取日期 (checkpoints_day_20251229.db -> 20251229)
                    date_str = filename.replace("checkpoints_day_", "").replace(
                        ".db", ""
                    )
                    file_date = datetime.datetime.strptime(date_str, "%Y%m%d")

                    # 计算天数差
                    days_diff = (current_date - file_date).days

                    # 删除超过 max_days 天的旧数据
                    if days_diff > max_days:
                        file_path = os.path.join(base_dir, filename)
                        os.remove(file_path)
                        print(f"🧹🧹 清理旧检查点文件: {filename} (超过 {max_days} 天)")

                except (ValueError, IndexError):
                    # 文件名不符合预期，跳过
                    continue

    except Exception as e:
        print(f"⚠️ 清理旧检查点失败: {e}")


# 创建agent
def create_langchain_agent(
    backend_url: str = "",
    token: str = "",
    username: str = "default",
    thread_id: str = "default",
):
    llm = ChatOpenAI(
        model=settings.LLM_MODEL,
        temperature=settings.LLM_TEMPERATURE,
        api_key=settings.DEEPSEEK_API_KEY,
        base_url=settings.DEEPSEEK_BASE_URL,
        max_tokens=settings.LLM_MAX_TOKENS,
    )

    tools = get_all_tools()
    # 添加调试信息
    print(f"🔧🔧🔧🔧 Agent 创建调试信息:")
    print(f"  - 用户: {username}")
    print(f"  - Thread ID: {thread_id}")
    print(f"  - 后端地址: {backend_url}")
    print(f"  - Token: {'已提供' if token else '未提供'}")
    print(f"  - 工具数量: {len(tools)}")

    for i, tool in enumerate(tools):
        print(f"  - 工具 {i+1}: {tool.name}")

    # 获取动态的system_prompt
    system_prompt = create_system_prompt(backend_url, token, username)

    def simple_turn_based_trim(
        messages: Sequence[BaseMessage],
        keep_turns: int = 3,
        system_message: BaseMessage = None,
    ) -> List[BaseMessage]:
        """
        修正版：按完整对话轮次修剪消息
        每轮对话从Human开始，到下一个Human之前结束
        """
        if not messages:
            return []
        # 分离系统消息（始终保留）
        system_messages = []
        other_messages = []
        for msg in messages:
            if (
                isinstance(msg, SystemMessage)
                or getattr(msg, "type", None) == "system"
                or getattr(msg, "role", None) == "system"
                or msg.__class__.__name__ == "SystemMessage"
            ):
                system_messages.append(msg)
            else:
                other_messages.append(msg)

        if len(other_messages) <= 1:
            return system_messages + other_messages

        # 找出所有Human消息的位置
        human_indices = []
        for i, msg in enumerate(other_messages):
            if (
                isinstance(msg, HumanMessage)
                or getattr(msg, "type", None) == "human"
                or getattr(msg, "role", None) == "user"
            ):
                human_indices.append(i)

        # 如果Human消息不足keep_turns轮，返回所有
        if not human_indices or len(human_indices) <= keep_turns:
            return system_messages + other_messages

        # 计算起始索引
        start_idx = human_indices[-keep_turns]

        # 获取要保留的消息
        preserved_messages = other_messages[start_idx:]

        # 4. 返回从该索引开始的所有消息
        result = system_messages + preserved_messages
        # print(f"修剪后消息数: {len(result)}")

        return result

    @before_model
    def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        """Keep only the last few messages to fit context window."""
        messages = state["messages"]

        if len(messages) <= 3:
            return None  # No changes needed

        # 保留最后4轮对话
        trimmed_messages = simple_turn_based_trim(messages, keep_turns=4)

        return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)] + trimmed_messages}

    # 使用SQLiteSaver（按天分割）
    checkpointer = get_sqlite_checkpointer()
    # print(f"打印检查点保存器: {checkpointer}")

    # 可选：清理旧检查点（可配置为定期执行）
    if os.getenv("AUTO_CLEANUP", "false").lower() == "true":
        cleanup_old_checkpoints(max_days=7)  # 保留最近7天数据

    agent = create_agent(
        llm,
        tools,
        checkpointer=checkpointer,
        system_prompt=system_prompt,
        middleware=[trim_messages],
    )

    return agent