AgentCoreService/pqai_agent_server/api_server.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

import logging
import werkzeug.exceptions
from flask import Flask, request, jsonify
from argparse import ArgumentParser

from pqai_agent import configs

from pqai_agent import logging_service, chat_service, prompt_templates
from pqai_agent.history_dialogue_service import HistoryDialogueService
from pqai_agent.user_manager import MySQLUserManager, MySQLUserRelationManager
from pqai_agent_server.const import AgentApiConst
from pqai_agent_server.utils import wrap_response, quit_human_intervention_status
from pqai_agent_server.utils import (
    run_extractor_prompt,
    run_chat_prompt,
    run_response_type_prompt,
)

app = Flask("agent_api_server")
logger = logging_service.logger
const = AgentApiConst()


@app.route("/api/listStaffs", methods=["GET"])
def list_staffs():
    staff_data = app.user_relation_manager.list_staffs()
    return wrap_response(200, data=staff_data)


@app.route("/api/getStaffProfile", methods=["GET"])
def get_staff_profile():
    staff_id = request.args["staff_id"]
    profile = app.user_manager.get_staff_profile(staff_id)
    if not profile:
        return wrap_response(404, msg="staff not found")
    else:
        return wrap_response(200, data=profile)


@app.route("/api/getUserProfile", methods=["GET"])
def get_user_profile():
    user_id = request.args["user_id"]
    profile = app.user_manager.get_user_profile(user_id)
    if not profile:
        resp = {"code": 404, "msg": "user not found"}
    else:
        resp = {"code": 200, "msg": "success", "data": profile}
    return jsonify(resp)


@app.route("/api/listUsers", methods=["GET"])
def list_users():
    user_name = request.args.get("user_name", None)
    user_union_id = request.args.get("user_union_id", None)
    if not user_name and not user_union_id:
        resp = {"code": 400, "msg": "user_name or user_union_id is required"}
        return jsonify(resp)
    data = app.user_manager.list_users(user_name=user_name, user_union_id=user_union_id)
    return jsonify({"code": 200, "data": data})


@app.route("/api/getDialogueHistory", methods=["GET"])
def get_dialogue_history():
    staff_id = request.args["staff_id"]
    user_id = request.args["user_id"]
    recent_minutes = int(request.args.get("recent_minutes", 1440))
    dialogue_history = app.history_dialogue_service.get_dialogue_history(
        staff_id, user_id, recent_minutes
    )
    return jsonify({"code": 200, "data": dialogue_history})


@app.route("/api/listModels", methods=["GET"])
def list_models():
    models = [
        {
            "model_type": "openai_compatible",
            "model_name": chat_service.VOLCENGINE_MODEL_DEEPSEEK_V3,
            "display_name": "DeepSeek V3 on 火山",
        },
        {
            "model_type": "openai_compatible",
            "model_name": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_32K,
            "display_name": "豆包Pro 32K",
        },
        {
            "model_type": "openai_compatible",
            "model_name": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_1_5,
            "display_name": "豆包Pro 1.5",
        },
        {
            "model_type": "openai_compatible",
            "model_name": chat_service.VOLCENGINE_BOT_DEEPSEEK_V3_SEARCH,
            "display_name": "DeepSeek V3联网 on 火山",
        },
        {
            "model_type": "openai_compatible",
            "model_name": chat_service.VOLCENGINE_MODEL_DOUBAO_1_5_VISION_PRO,
            "display_name": "豆包1.5视觉理解Pro",
        },
    ]
    return wrap_response(200, data=models)


@app.route("/api/listScenes", methods=["GET"])
def list_scenes():
    scenes = [
        {"scene": "greeting", "display_name": "问候"},
        {"scene": "chitchat", "display_name": "闲聊"},
        {"scene": "profile_extractor", "display_name": "画像提取"},
        {"scene": "response_type_detector", "display_name": "回复模态判断"},
        {"scene": "custom_debugging", "display_name": "自定义调试场景"},
    ]
    return wrap_response(200, data=scenes)


@app.route("/api/getBasePrompt", methods=["GET"])
def get_base_prompt():
    scene = request.args["scene"]
    prompt_map = {
        "greeting": prompt_templates.GENERAL_GREETING_PROMPT,
        "chitchat": prompt_templates.CHITCHAT_PROMPT_COZE,
        "profile_extractor": prompt_templates.USER_PROFILE_EXTRACT_PROMPT,
        "response_type_detector": prompt_templates.RESPONSE_TYPE_DETECT_PROMPT,
        "custom_debugging": "",
    }
    model_map = {
        "greeting": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_32K,
        "chitchat": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_32K,
        "profile_extractor": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_1_5,
        "response_type_detector": chat_service.VOLCENGINE_MODEL_DOUBAO_PRO_1_5,
        "custom_debugging": chat_service.VOLCENGINE_BOT_DEEPSEEK_V3_SEARCH,
    }
    if scene not in prompt_map:
        return wrap_response(404, msg="scene not found")
    data = {"model_name": model_map[scene], "content": prompt_map[scene]}
    return wrap_response(200, data=data)


@app.route("/api/runPrompt", methods=["POST"])
def run_prompt():
    try:
        req_data = request.json
        logger.debug(req_data)
        scene = req_data["scene"]
        if scene == "profile_extractor":
            response = run_extractor_prompt(req_data)
            return wrap_response(200, data=response)
        elif scene == "response_type_detector":
            response = run_response_type_prompt(req_data)
            return wrap_response(200, data=response.choices[0].message.content)
        else:
            response = run_chat_prompt(req_data)
            return wrap_response(200, data=response.choices[0].message.content)
    except Exception as e:
        logger.error(e)
        return wrap_response(500, msg="Error: {}".format(e))


@app.route("/api/healthCheck", methods=["GET"])
def health_check():
    return wrap_response(200, msg="OK")


@app.route("/api/getStaffSessionSummary", methods=["GET"])
def get_staff_session_summary():
    staff_id = request.args.get("staff_id")
    status = request.args.get("status", const.DEFAULT_STAFF_STATUS)
    page_id = request.args.get("page_id", const.DEFAULT_PAGE_ID)
    page_size = request.args.get("page_size", const.DEFAULT_PAGE_SIZE)

    # check params
    try:
        page_id = int(page_id)
        page_size = int(page_size)
        status = int(status)
    except Exception as e:
        return wrap_response(404, msg="Invalid parameter: {}".format(e))

    staff_session_summary = app.user_manager.get_staff_sessions_summary_v1(
        staff_id, page_id, page_size, status
    )

    if not staff_session_summary:
        return wrap_response(404, msg="staff not found")
    else:
        return wrap_response(200, data=staff_session_summary)


@app.route("/api/getStaffSessionList", methods=["GET"])
def get_staff_session_list():
    staff_id = request.args.get("staff_id")
    if not staff_id:
        return wrap_response(404, msg="staff_id is required")

    page_size = request.args.get("page_size", const.DEFAULT_PAGE_SIZE)
    page_id = request.args.get("page_id", const.DEFAULT_PAGE_ID)
    staff_session_list = app.user_manager.get_staff_session_list_v1(
        staff_id, page_id, page_size
    )
    if not staff_session_list:
        return wrap_response(404, msg="staff not found")

    return wrap_response(200, data=staff_session_list)


@app.route("/api/getStaffList", methods=["GET"])
def get_staff_list():
    page_size = request.args.get("page_size", const.DEFAULT_PAGE_SIZE)
    page_id = request.args.get("page_id", const.DEFAULT_PAGE_ID)
    staff_list = app.user_manager.get_staff_list(page_id, page_size)
    if not staff_list:
        return wrap_response(404, msg="staff not found")
    return wrap_response(200, data=staff_list)


@app.route("/api/getConversationList", methods=["GET"])
def get_conversation_list():
    """
    获取staff && customer的 私聊对话列表
    :return:
    """
    staff_id = request.args.get("staff_id")
    customer_id = request.args.get("customer_id")
    if not staff_id or not customer_id:
        return wrap_response(404, msg="staff_id and customer_id are required")

    page = request.args.get("page")
    response = app.user_manager.get_conversation_list_v1(staff_id, customer_id, page, const.DEFAULT_CONVERSATION_SIZE)
    return wrap_response(200, data=response)


@app.route("/api/quitHumanInterventionStatus", methods=["GET"])
def quit_human_interventions_status():
    """
    退出人工介入状态
    :return:
    """
    staff_id = request.args.get("staff_id")
    customer_id = request.args.get("customer_id")
    # 测试环境： staff_id 强制等于1688854492669990
    staff_id = 1688854492669990
    if not customer_id or not staff_id:
        return wrap_response(404, msg="user_id and staff_id are required")
    response = quit_human_intervention_status(customer_id, staff_id)

    return wrap_response(200, data=response)


@app.route("/api/sendMessage", methods=["POST"])
def send_message():
    return wrap_response(200, msg="暂不实现功能")


@app.errorhandler(werkzeug.exceptions.BadRequest)
def handle_bad_request(e):
    logger.error(e)
    return wrap_response(400, msg="Bad Request: {}".format(e.description))


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("--prod", action="store_true")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8083)
    parser.add_argument("--log-level", default="INFO")
    args = parser.parse_args()

    config = configs.get()
    logging_level = logging.getLevelName(args.log_level)
    logging_service.setup_root_logger(
        level=logging_level, logfile_name="agent_api_server.log"
    )

    user_db_config = config["storage"]["user"]
    staff_db_config = config["storage"]["staff"]
    user_manager = MySQLUserManager(
        user_db_config["mysql"], user_db_config["table"], staff_db_config["table"]
    )
    app.user_manager = user_manager

    wecom_db_config = config["storage"]["user_relation"]
    user_relation_manager = MySQLUserRelationManager(
        user_db_config["mysql"],
        wecom_db_config["mysql"],
        config["storage"]["staff"]["table"],
        user_db_config["table"],
        wecom_db_config["table"]["staff"],
        wecom_db_config["table"]["relation"],
        wecom_db_config["table"]["user"],
    )
    app.user_relation_manager = user_relation_manager
    app.history_dialogue_service = HistoryDialogueService(
        config["storage"]["history_dialogue"]["api_base_url"]
    )

    app.run(debug=not args.prod, host=args.host, port=args.port)