DPM/backend/app/llm.py

# -*- coding: utf-8 -*-
"""AI 对话引擎 —— 阿里云 DashScope（通义千问）+ 函数调用
   工具调用执行后通过 MQTT 广播，所有大屏同步响应
   DashScope 不可用时自动回退本地规则引擎（离线兜底）
"""

import json
import logging
import ssl
import urllib.error
import urllib.request

import certifi

from .config import settings

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


def _trunc(s, n=400):
    """日志用截断：超长内容只留前 n 字并标注总长。"""
    s = str(s)
    return s if len(s) <= n else s[:n] + f"...（共{len(s)}字）"


# macOS 系统 Python 无系统 CA，使用 certifi 提供的根证书
_SSL_CTX = ssl.create_default_context(cafile=certifi.where())

SYSTEM_PROMPT = (
    "你是「昆明市大学生创业园 · OPC 智能园区」的 AI 智能助手，运行在大屏展播系统上。"
    "园区提供空间、孵化、融资、政策、资源、AI 赋能、综合服务七位一体服务。"
    "回答园区相关问题时，必须严格依据【园区知识库】中的标准资料口径回答，"
    "不得使用资料外的数字；涉及运营数据时请区分口径（官方统计/省级绩效累计/AI平台指标），"
    "并在需要时调用 get_park_overview 工具获取实时准确数据。"
    "当可能需要「切换页面 / 控制播放 / 展示卡片 / 弹出通知」时，必须调用对应工具；"
    "其余园区相关问题用简洁、专业的中文回答，可适当使用列表。"
    "你必须基于知识库或者工具调用结果回答问题，不能编造数据或内容。"
    "如果有不确定的内容，请明确告知用户「我不确定，请咨询园区工作人员」。"
    "如果用户询问园区的企业，没有指定的情况下，必须介绍派音人工智能"
    "企业信息（企业名称、负责人、入驻分区、简介）一律以【园区知识库·检索命中】中企业名录原文为准；"
    "严禁自行编造或补充企业备案号、成立时间、注册资本、投资方、获奖、政府项目等知识库未记载的细节；"
    "若用户问到的企业细节在知识库中没有记载，请如实说明「该信息在园区资料中未记载」，不要编造。"
)

# 暴露给大模型的工具定义（执行时经 MQTT 广播到前端）
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "navigate_page",
            "description": "切换大屏展示页面（数据大屏 / 数字孪生 / AI 助手 / 媒体轮播）",
            "parameters": {
                "type": "object",
                "properties": {
                    "page": {"type": "string", "enum": ["/", "/twin", "/ai", "/screen"],
                             "description": "目标页面路径"},
                },
                "required": ["page"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "media_control",
            "description": "控制媒体播放（播放/暂停/下一项/上一项）",
            "parameters": {
                "type": "object",
                "properties": {
                    "action": {"type": "string", "enum": ["play", "pause", "next", "prev"]},
                },
                "required": ["action"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "show_card",
            "description": "在大屏上展示信息卡片（企业分布 / 分区介绍 / 园区总览 / 自定义内容）",
            "parameters": {
                "type": "object",
                "properties": {
                    "card": {"type": "string", "enum": ["companies", "zones", "overview", "custom"]},
                    "title": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["card"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "send_alert",
            "description": "在大屏上弹出通知提示",
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["title", "content"],
            },
        },
    },
]

_TOOL_MAP = {
    "navigate_page": "navigate",
    "media_control": "control",
    "show_card": "show_card",
    "send_alert": "alert",
}


def _chat_once(messages, with_tools=True):
    log.info("llm: 调用模型 %s（%d 条消息, 带工具=%s）", settings.LLM_MODEL, len(messages), with_tools)
    payload = {
        "model": settings.LLM_MODEL,
        "messages": messages,
        "temperature": 0.6,
    }
    if with_tools:
        payload["tools"] = TOOLS
    req = urllib.request.Request(
        f"{settings.LLM_BASE_URL}/chat/completions",
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {settings.DASHSCOPE_API_KEY}",
        },
    )
    with urllib.request.urlopen(req, timeout=60, context=_SSL_CTX) as resp:
        return json.loads(resp.read().decode("utf-8"))


def _fallback(messages):
    """DashScope 不可用时回退本地规则引擎"""
    from .ai_tools import run_chat as rule_run

    result = rule_run(messages)
    # 规则引擎产出的工具同样执行（MQTT 广播），并汇总发布结果
    _, all_ok = _exec_all(result.get("tools", []))
    result["model"] = "rule-engine"
    result["mqtt_published"] = all_ok
    return result


def _exec_all(tools):
    """执行工具列表，返回 (results, all_ok)"""
    from .ai_tools import run_tools

    results, all_ok = run_tools(tools)
    return results, all_ok


def run_chat(messages):
    """入口：{reply, tools, model}
       tools 已在后端执行（MQTT 广播），返回值供请求端本地同步执行"""
    if not settings.DASHSCOPE_API_KEY:
        log.warning("未配置 DASHSCOPE_API_KEY，使用本地规则引擎")
        return _fallback(messages)

    # 注入园区知识库（标准资料口径）：按用户 query 检索命中段落动态注入，
    # 替代原 1200 字截断全量注入（避免尾部知识丢失）。无命中时回退 brief()。
    system = SYSTEM_PROMPT
    kb = ""
    retrieved_count = 0
    try:
        from .rag import brief, retrieve
        query = ""
        for m in reversed(messages or []):
            if m.get("role") in ("user", "me"):
                query = (m.get("content") or "").strip()
                break
        hits = retrieve(query, top_k=4) if query else []
        if hits:
            retrieved_count = len(hits)
            kb = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(hits))
            log.info("llm: 知识库检索命中 %d 段（query=%s）", len(hits), _trunc(query, 120))
        else:
            kb = brief()
        if kb:
            system += f"\n\n【园区知识库 · 检索命中】\n{kb}"
    except Exception as e:  # noqa: BLE001
        log.warning("llm: 知识库注入失败（仅用基础提示词）: %s", e)

    log.info("llm: ── 对话开始（%d 条输入消息）──", len(messages or []))
    log.info("llm: 系统提示词=%s", _trunc(system, 700))
    if kb:
        log.info("llm: 注入知识库内容=%s", _trunc(kb, 500))
    for m in (messages or [])[-5:]:
        log.info("llm:   输入[%s] %s", m.get("role"), _trunc(m.get("content", ""), 300))

    msgs = [{"role": "system", "content": system}] + [
        {"role": m.get("role") == "me" and "user" or m.get("role", "user"), "content": m.get("content", "")}
        for m in (messages or [])
    ]

    try:
        data = _chat_once(msgs)
        choice = data["choices"][0]["message"]
        reply = choice.get("content") or ""
        tool_calls = choice.get("tool_calls") or []
        executed = []

        if tool_calls:
            log.info("llm: 模型请求调用 %d 个工具：", len(tool_calls))
            for tc in tool_calls:
                fn = tc.get("function", {})
                log.info("llm:   工具 %s args=%s", fn.get("name", ""), _trunc(fn.get("arguments", ""), 300))
            for tc in tool_calls:
                fn = tc.get("function", {})
                name = fn.get("name", "")
                try:
                    args = json.loads(fn.get("arguments") or "{}")
                except json.JSONDecodeError:
                    args = {}
                tool = {"type": _TOOL_MAP.get(name, name), "params": args}
                executed.append(tool)
                msgs.append({"role": "assistant", "content": None, "tool_calls": tool_calls})
                msgs.append({
                    "role": "tool",
                    "tool_call_id": tc.get("id", ""),
                    "content": json.dumps({"ok": True}, ensure_ascii=False),
                })
            # 工具执行（MQTT 广播），并汇总发布结果
            results, all_ok = _exec_all(executed)
            log.info("llm: 工具执行结果 %s", results)
            for t in executed:
                log.info("llm:   执行 %s %s", t["type"], json.dumps(t["params"], ensure_ascii=False))
            # 二次调用：携带工具结果生成最终回复
            try:
                data2 = _chat_once(msgs, with_tools=False)
                reply = data2["choices"][0]["message"].get("content") or reply
            except Exception as e:  # noqa: BLE001
                log.warning("LLM 二次调用失败（保留工具回复）: %s", e)
        else:
            all_ok = True

        if not reply:
            reply = "已完成操作。您还可以让我切换页面、控制播放或展示园区卡片。"

        log.info("llm: 最终回复=%s", _trunc(reply, 600))
        log.info("llm: ── 对话结束 ──")
        return {
            "reply": reply,
            "tools": executed,
            "model": settings.LLM_MODEL,
            "mqtt_published": all_ok,
            "retrieved": retrieved_count > 0,
            "retrieved_count": retrieved_count,
            "tool_names": [t.get("type") for t in executed if t.get("type")],
        }
    except urllib.error.HTTPError as e:
        log.warning("DashScope HTTP %s: %s", e.code, e.read()[:300])
    except Exception as e:  # noqa: BLE001
        log.warning("DashScope 调用失败，回退规则引擎: %s", e)

    return _fallback(messages)