gangyan/langchain-chat/server/chat/tools_v2.py

"""
LangGraph 版工具集：闭包工厂注入 uuid，统一异常包装。

为什么要重写：
1. 旧版 tools 用 `query` 字符串里塞 JSON + uuid 的 hack 传 metadata
2. 旧版 LLM 工具调度靠多次 LLM 路由，慢且容易循环
3. 这里给每个工具暴露结构化 args_schema，交给 LangGraph ReAct 直接 function-calling
"""
import json
import logging
from typing import List, Optional
from langchain_core.tools import tool

# 旧版工具函数仍然复用——只改包装层
from server.agent.tools.search_tool import search_tool as _legacy_kb_search
from server.agent.tools.knowledgebase_kgo_search import knowledgebase_kgo_search as _legacy_kgo_search
from server.agent.tools.draw_plot import create_and_save_plot as _legacy_draw_plot
from server.agent.tools.math import math_count as _legacy_math, code_count as _legacy_code
from server.agent.tools.weather_check import weathercheck as _legacy_weather
from server.agent.tools.search_picture import search_pic as _legacy_search_pic
from server.agent.tools.get_statistical_data import mysql_statistic as _legacy_mysql

logger = logging.getLogger(__name__)


def _safe_call(name: str, fn, *args, **kwargs) -> str:
    """统一异常包装：把 raise 转成给模型的字符串提示，让 ReAct 可恢复。"""
    try:
        result = fn(*args, **kwargs)
        return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
    except Exception as e:
        logger.exception(f"工具 {name} 调用异常")
        return f"[工具 {name} 调用异常] {type(e).__name__}: {str(e)[:200]}。请使用其他工具或基于已有信息回答。"


def make_tools(uuid: str) -> list:
    """根据本次请求的 uuid 生成一组闭包工具。

    每个工具内部用闭包捕获 uuid，调用旧版 func 时按旧 hack 拼装入参字符串。
    模型看到的工具入参是结构化的，看不到 uuid。
    """

    @tool("知识库联想")
    def kb_search(
        query: str,
        knowledge_name: List[str],
        keywords: Optional[List[str]] = None,
    ) -> str:
        """从指定知识库检索资料。

        knowledge_name 必须从如下列表中选择（可多选）：
        【中国钢铁行业动态库、政策库、期刊论文库、冶金新闻库、冶金中文期刊库、
         冶金外文期刊库、冶金OA期刊库、冶金行业新闻库、冶金专业知识库、
         冶金行业报告库、报告库、美术专业知识库】。
        涉及国家政策时优先选政策库；钢铁行业问题优先选中国钢铁行业动态库。
        keywords 是相关关键词，2-4 个为宜。
        """
        payload = json.dumps({
            "query": query,
            "knowledge_name": knowledge_name,
            "keywords": keywords or [],
        }, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("知识库联想", _legacy_kb_search, legacy_input)

    @tool("联网思索")
    def web_search(query: str) -> str:
        """联网搜索（智谱 search）。query 必须是用户原文，禁止改写。"""
        payload = json.dumps({"query": query}, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("联网思索", _legacy_kgo_search, legacy_input)

    @tool("图表绘制")
    def draw_plot(
        data: dict,
        title: str,
        xlabel: str,
        ylabel: str,
        plot_type: str,
    ) -> str:
        """绘制图表。

        data 形如 {"分类A": 23, "分类B": 17}，xlabel/ylabel 描述坐标轴含义。
        plot_type 必须是 bar / pie / line 之一。
        本工具一次只能画一张图；输出图片链接后必须按工具说明输出 markdown 引用。
        """
        payload = json.dumps({
            "data": data,
            "title": title,
            "xlabel": xlabel,
            "ylabel": ylabel,
            "plot_type": plot_type,
        }, ensure_ascii=False)
        # 旧版 draw_plot 接受 <param>...</param> 包裹的 JSON
        wrapped = f"<param>{payload}</param>{json.dumps({'uuid': uuid})}"
        return _safe_call("图表绘制", _legacy_draw_plot, wrapped)

    @tool("数学运算")
    def math_solve(query: str) -> str:
        """数学问题求解。query 描述要求解的数学问题。"""
        payload = json.dumps({"query": query}, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("数学运算", _legacy_math, legacy_input)

    @tool("代码专家")
    def code_solve(query: str) -> str:
        """代码相关问题，包括写代码、解释代码、调试。"""
        payload = json.dumps({"query": query}, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("代码专家", _legacy_code, legacy_input)

    @tool("天气工具")
    def weather(location: str) -> str:
        """查询某城市三天内天气。location 是中文城市名，如"北京"。"""
        payload = json.dumps({"location": location}, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("天气工具", _legacy_weather, legacy_input)

    @tool("美术作品获取")
    def art_search(query: str) -> str:
        """查询美术作品图片。query 是作品类型描述（如"山水画"、"草原"），不要传"美术作品"等通用词。"""
        payload = json.dumps({"query": query}, ensure_ascii=False)
        legacy_input = payload + json.dumps({"uuid": uuid}, ensure_ascii=False)
        return _safe_call("美术作品获取", _legacy_search_pic, legacy_input)

    @tool("统计数据查询")
    def stat_query(query: str) -> str:
        """统计数据库查询。仅有 199x-2023 数据。query 是详细的查询问题描述。"""
        payload = json.dumps({"query": query}, ensure_ascii=False)
        wrapped = f"<param>{payload}</param>{json.dumps({'uuid': uuid})}"
        return _safe_call("统计数据查询", _legacy_mysql, wrapped)

    return [
        kb_search,
        web_search,
        draw_plot,
        math_solve,
        code_solve,
        weather,
        art_search,
        stat_query,
    ]