server-core/app/park/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 _trunc(s, n=800):
    """日志用截断：超长内容只留前 n 字并标注总长。"""
    s = str(s)
    return s if len(s) <= n else s[:n] + f"...（共{len(s)}字）"


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

def _get_park_overview(args):
    """园区实时概览（官方统计口径 + 省级绩效累计口径 + AI 平台运行指标）"""
    from .sim_engine import sim_engine
    d = sim_engine.snapshot()
    return json.dumps({
        "数据口径": "官方统计（《2026年7月运行情况统计表》）；累计口径（附件1省级材料，截至2026年5月）；AI平台指标：TOKEN/工具调用",
        "实际入驻企业数(家)": d["projects"]["inPark"],
        "园区可容纳企业数(个)": d["projects"]["capacity"],
        "开园以来累计投入运营资金(万元)": d["projects"]["invested"],
        "带动就业人数(人,当年累计)": d["jobs"]["total"],
        "生产经营总额(万元,当年累计)": d["revenue"]["total"],
        "上缴税利总额(万元,当年累计)": d["revenue"]["tax"],
        "累计孵化企业(家,省级绩效累计)": d["cumulative"]["incubated"],
        "当前实有在孵实体(家,省级绩效累计)": d["cumulative"]["inIncubation"],
        "累计成功孵化出园企业(家,开园以来累计)": d["cumulative"]["graduated"],
        "入驻团队发明专利(项,累计)": d["cumulative"]["patents"],
        "累计带动就业(人,省级绩效累计)": d["cumulative"]["jobs"],
        "入驻团队累计经营收入(万元)": d["cumulative"]["revenue"],
        "入驻团队累计税收(万元)": d["cumulative"]["tax"],
        "创业指导专家团队(人)": d["cumulative"]["mentors"],
        "近3年入孵实体孵化成功率(%)": d["cumulative"]["successRate"],
        "园区建筑面积(㎡)": d["park"]["area"],
        "TOKEN月均消耗(亿)": 200,
        "TOKEN实时速率(t/s)": d["token"]["rate"],
        "今日TOKEN消耗(万)": d["token"]["today"],
    }, ensure_ascii=False)


def _query_companies(args):
    """园区入驻企业名录查询（关键词过滤，取自 park_config 主数据源）"""
    from . import park_config
    q = (args.get("keyword") or "").strip()
    all_comps = park_config.list_companies()
    names = [c["name"] for c in all_comps]
    names = [n for n in names if q in n] if q else names
    return json.dumps({
        "total": len(names),
        "keyword": q,
        "companies": names[:20],
    }, ensure_ascii=False)


def _control_display(args, screen_id=""):
    """大屏显示控制：切页 / 弹通知 / 媒体播放暂停（经 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, screen_id=screen_id).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, screen_id: str = ""):
    """执行工具 → 返回给 LLM 的字符串结果（JSON）。
       控制类工具（control_display）按发起屏幕 screen_id 只下发到该屏。"""
    t = TOOLS.get(name)
    if not t:
        log.warning("tools: 未知工具 %s", name)
        return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False)
    try:
        if name == "control_display":
            result = _control_display(args or {}, screen_id)
        else:
            result = t["fn"](args or {})
        log.info("tools: 执行工具 %s args=%s screen=%s", name, json.dumps(args, ensure_ascii=False), screen_id or "all")
        log.info("tools:   结果 -> %s", _trunc(result, 1000))
        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)