DPM/backend/app/tools.py

# -*- coding: utf-8 -*-
"""智能体工具注册表 —— LLM function calling 的后端执行器

- TOOLS：工具名 → 描述 + JSON Schema 参数（给 LLM 声明用）
- exec_tool(name, args)：执行工具，返回给 LLM 的字符串结果（JSON）
- 供 POST /api/tools/exec 调用；前端 s2s 客户端收到 toolcall 事件后转发到此执行
"""
import json
import logging
from datetime import datetime

log = logging.getLogger("dpm.tools")


# ── 工具实现 ────────────────────────────────────────────────────────────

def _get_park_overview(args):
    """园区实时概览（在园项目/就业/营收/在园人数等）"""
    from .sim_engine import sim_engine
    d = sim_engine.snapshot()
    return json.dumps({
        "在园项目": d["projects"]["inPark"],
        "累计孵化企业": d["projects"]["cum"],
        "今日新增项目": d["projects"]["todayNew"],
        "带动就业(人)": d["jobs"]["total"],
        "今日新增就业": d["jobs"]["todayNew"],
        "累计营收(万元)": d["revenue"]["total"],
        "今日营收(万元)": d["revenue"]["today"],
        "营收同比增速(%)": d["revenue"]["growth"],
        "设备在线率(%)": d["park"]["devices"],
        "今日能耗(kWh)": d["park"]["energy"],
        "当前在园人数": d["park"]["people"],
        "会议预约": d["park"]["meeting"],
        "工位使用": d["park"]["desk"],
    }, ensure_ascii=False)


def _query_companies(args):
    """园区入驻企业名录查询（关键词过滤）"""
    from .sim_engine import COMPANY_NAMES
    q = (args.get("keyword") or "").strip()
    names = [n for n in COMPANY_NAMES if q in n] if q else list(COMPANY_NAMES)
    return json.dumps({
        "total": len(names),
        "keyword": q,
        "companies": names[:20],
    }, ensure_ascii=False)


def _control_display(args):
    """大屏显示控制：切页 / 弹通知 / 媒体播放暂停（经 MQTT 广播到大屏）"""
    from .mqtt import hub
    action = (args.get("action") or "").strip()
    target = (args.get("target") or "").strip()
    if action not in ("switch_page", "alert", "media_play", "media_pause"):
        return json.dumps({"error": f"不支持的 action: {action}"}, ensure_ascii=False)
    params = {"target": target} if target else {}
    ok = hub.publish_command(action, params).get("published", False)
    return json.dumps({"ok": ok, "action": action, "target": target}, ensure_ascii=False)


def _get_time(args):
    """当前日期时间"""
    return json.dumps({"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, ensure_ascii=False)


# ── 注册表 ──────────────────────────────────────────────────────────────

TOOLS = {
    "get_park_overview": {
        "fn": _get_park_overview,
        "description": "获取园区实时运营概览：在园项目数、累计孵化企业、带动就业、营收（今日/累计）、设备在线率、在园人数、能耗等。回答园区数据类问题时使用。",
        "parameters": {"type": "object", "properties": {}},
    },
    "query_companies": {
        "fn": _query_companies,
        "description": "查询园区入驻企业名录，支持按企业名关键词过滤。回答'有哪些企业/某企业是否入驻'时使用。",
        "parameters": {
            "type": "object",
            "properties": {"keyword": {"type": "string", "description": "企业名关键词，可为空字符串"}},
        },
    },
    "control_display": {
        "fn": _control_display,
        "description": "控制大屏显示：switch_page 切换页面、alert 弹通知、media_play/media_pause 控制媒体播放。用户要求控制大屏时使用。",
        "parameters": {
            "type": "object",
            "properties": {
                "action": {"type": "string", "enum": ["switch_page", "alert", "media_play", "media_pause"]},
                "target": {"type": "string", "description": "目标：页面路径（如 /、/twin、/ai、/voice）或通知文本"},
            },
        },
    },
    "get_time": {
        "fn": _get_time,
        "description": "获取当前日期时间。用户问'现在几点/今天几号'时使用。",
        "parameters": {"type": "object", "properties": {}},
    },
}


def tool_schemas():
    """给 LLM 的 tools 定义（OpenAI function calling 格式）"""
    return [
        {
            "type": "function",
            "name": name,
            "description": t["description"],
            "parameters": t["parameters"],
        }
        for name, t in TOOLS.items()
    ]


def exec_tool(name: str, args: dict):
    """执行工具 → 返回给 LLM 的字符串结果（JSON）"""
    t = TOOLS.get(name)
    if not t:
        log.warning("tools: 未知工具 %s", name)
        return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False)
    try:
        result = t["fn"](args or {})
        log.info("tools: %s args=%s -> %s", name, json.dumps(args, ensure_ascii=False)[:120], str(result)[:120])
        return result
    except Exception as e:  # noqa: BLE001
        log.error("tools: %s 执行失败 %s", name, e, exc_info=True)
        return json.dumps({"error": f"{name} 执行失败: {e}"}, ensure_ascii=False)