Create app.py

app.py

@@ -0,0 +1,865 @@
+"""
+基于 smolagents 框架的 Hugging Face 数据集自然语言查询智能代理
+
+功能：
+1. 启动时自动获取数据集文件列表，作为系统提示词告知用户可用数据
+2. 接收用户自然语言输入，从中提取数据集查询所需的参数信息
+3. 使用提取的参数从 Hugging Face 数据集获取数据（支持 ZIP/CSV/Excel 等格式）
+4. 返回结构化的查询结果
+
+依赖安装：
+    pip install 'smolagents[gradio]' requests pandas huggingface_hub openpyxl
+
+使用前：
+    1. 将文件顶部 HF_TOKEN 常量的值替换为个人 Hugging Face Settings 下申请的 Read Token（用于数据集访问）
+    2. 将文件顶部 OPENAI_API_KEY 常量的值替换为你的 OpenAI API Key（用于 LLM 接入）
+    3. 如使用第三方兼容服务，修改 OPENAI_API_BASE 常量的地址
+"""
+
+import io
+import json
+import logging
+import re
+import zipfile
+from typing import Optional, List
+
+import requests
+import pandas as pd
+from huggingface_hub import HfApi, hf_hub_download
+from smolagents import Tool, ToolCallingAgent, OpenAIServerModel, GradioUI
+
+# ===========================================================================
+# Hugging Face 数据集访问 Token（用于从 HF 数据集获取数据）
+# ===========================================================================
+HF_TOKEN = 
+
+# ===========================================================================
+# 数据集仓库 ID
+# ===========================================================================
+HF_DATASET_REPO = "squid-lab/squid_dataset"  # Hugging Face 数据集仓库 ID
+
+# ===========================================================================
+# OpenAI 模型配置常量
+# ===========================================================================
+OPENAI_API_KEY =   # 请替换为你的 OpenAI API Key
+OPENAI_API_BASE = "https://api.deepseek.com/v1"  # OpenAI API 地址（注意需要 /v1 后缀）
+MODEL_ID = "deepseek-chat"  # DeepSeek 模型：deepseek-chat 支持工具调用，deepseek-reasoner 不支持
+
+# ===========================================================================
+# DeepSeek 思考模式配置
+# ===========================================================================
+THINKING_MODE_ENABLED = True  # 是否启用思考模式（True 启用，False 禁用）
+THINKING_EFFORT = "high"  # 思考强度：low/medium/high/max（实际 low/medium 会映射为 high）
+
+# ---------------------------------------------------------------------------
+# 日志配置
+# ---------------------------------------------------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+)
+logger = logging.getLogger("hf_data_agent")
+
+
+# ===========================================================================
+# 数据集文件列表获取函数
+# ===========================================================================
+def get_dataset_file_list() -> List[str]:
+    """
+    从 Hugging Face 数据集仓库获取一级文件和文件夹名称列表。
+
+    使用 HfApi.dataset_info() 获取数据集元信息，提取 siblings 中的文件路径，
+    只保留一级目录/文件（不包含深层嵌套的文件）。
+
+    Returns:
+        一级文件和文件夹名称列表，如 ['中西太平洋WCPFC/', 'README.md', 'data.csv']
+    """
+    logger.info("正在获取数据集文件列表: %s", HF_DATASET_REPO)
+
+    try:
+        api = HfApi(token=HF_TOKEN)
+        dataset_info = api.dataset_info(repo_id=HF_DATASET_REPO)
+
+        # 提取所有文件路径
+        all_files = [sibling.rfilename for sibling in dataset_info.siblings]
+
+        # 只保留一级目录/文件（路径中不包含 '/' 的项，或者一级文件夹）
+        first_level_items = set()
+        for file_path in all_files:
+            parts = file_path.split("/")
+            # 一级文件（无子目录）
+            if len(parts) == 1:
+                first_level_items.add(parts[0])
+            # 一级文件夹（取第一部分，添加 '/' 后缀标识为文件夹）
+            else:
+                first_level_items.add(parts[0] + "/")
+
+        file_list = sorted(list(first_level_items))
+        logger.info("获取到 %d 个一级文件/文件夹: %s", len(file_list), file_list)
+        return file_list
+
+    except Exception as e:
+        logger.error("获取数据集文件列表失败: %s", e, exc_info=True)
+        return []
+
+
+def build_system_prompt_with_file_list() -> str:
+    """
+    构建包含数据集文件列表的系统提示词。
+
+    在对话启动时告知用户当前可用的数据集文件，帮助用户了解可查询的内容。
+
+    Returns:
+        格式化的系统提示词字符串
+    """
+    file_list = get_dataset_file_list()
+
+    if not file_list:
+        return (
+            "你是一个 Hugging Face 数据集查询助手。"
+            "用户可以询问数据集内容，你将帮助用户查询和获取数据。"
+            "当前无法获取数据集文件列表，请检查 HF_TOKEN 配置是否正确。"
+        )
+
+    # 格式化文件列表为提示词
+    files_str = "\n".join([f"  - {item}" for item in file_list])
+    system_prompt = (
+        "你是一个 Hugging Face 数据集查询助手。\n\n"
+        "当前可查询的数据集仓库: " + HF_DATASET_REPO + "\n\n"
+        "数据集一级文件和文件夹列表:\n" + files_str + "\n\n"
+        "用户可以询问数据集中的具体内容，例如:\n"
+        "  - '查看中西太平洋的渔获数据'\n"
+        "  - '查询1967-2024年的月尺度数据'\n"
+        "  - '获取某个特定文件的内容'\n\n"
+        "请根据用户的需求，使用工具帮助用户查询和获取数据。"
+    )
+
+    logger.info("系统提示词已构建，包含 %d 个文件/文件夹", len(file_list))
+    return system_prompt
+
+
+# ===========================================================================
+# DeepSeek 思考模式自定义模型类
+# ===========================================================================
+class DeepSeekThinkingModel(OpenAIServerModel):
+    """
+    继承 OpenAIServerModel，为 DeepSeek 思考模式添加特殊处理。
+
+    DeepSeek 思考模式特殊要求：
+    1. 使用 extra_body={"thinking": {"type": "enabled"}} 开启思考模式
+    2. 设置 reasoning_effort 参数控制思考强度
+    3. 思考模式不支持 tool_choice 参数（会导致 400 错误）
+    4. 工具调用后必须回传 reasoning_content
+    """
+
+    def __init__(
+        self,
+        model_id: str = MODEL_ID,
+        api_key: str = OPENAI_API_KEY,
+        api_base: str = OPENAI_API_BASE,
+        thinking_enabled: bool = THINKING_MODE_ENABLED,
+        reasoning_effort: str = THINKING_EFFORT,
+        **kwargs,
+    ):
+        """
+        初始化 DeepSeek 思考模式模型。
+
+        Args:
+            model_id: 模型 ID
+            api_key: API Key
+            api_base: API 地址
+            thinking_enabled: 是否启用思考模式
+            reasoning_effort: 思考强度（low/medium/high/max）
+        """
+        # 调用父类初始化（移除不支持的工具调用参数）
+        super().__init__(
+            model_id=model_id,
+            api_key=api_key,
+            api_base=api_base,
+            **kwargs,
+        )
+
+        self.thinking_enabled = thinking_enabled
+        self.reasoning_effort = reasoning_effort
+        logger.info(
+            "DeepSeek 思考模式模型初始化: model_id=%s, thinking=%s, effort=%s",
+            model_id,
+            thinking_enabled,
+            reasoning_effort,
+        )
+
+    def generate(self, messages, tools=None, tool_choice=None, **kwargs):
+        """
+        重写 generate 方法，添加思考模式特殊处理。
+
+        关键处理：
+        1. 移除 tool_choice 参数（思考模式不支持）
+        2. 添加 extra_body 参数开启思考模式
+        3. 设置 reasoning_effort 参数
+        """
+        # --- 处理 tool_choice 参数 ---
+        # DeepSeek 思考模式不支持 tool_choice，必须移除或设置为 None
+        if self.thinking_enabled and tool_choice is not None:
+            logger.warning(
+                "DeepSeek 思考模式不支持 tool_choice 参数，已自动移除。"
+                "如果需要强制工具调用，请设置 THINKING_MODE_ENABLED = False"
+            )
+            tool_choice = None  # 强制设为 None，避免 400 错误
+
+        # --- 构建思考模式参数 ---
+        if self.thinking_enabled:
+            # 使用 extra_body 传递思考模式参数（OpenAI SDK 不直接支持）
+            extra_body = {
+                "thinking": {"type": "enabled"},
+            }
+            # 设置 reasoning_effort（可以通过 kwargs 或直接传入）
+            if "reasoning_effort" not in kwargs:
+                kwargs["reasoning_effort"] = self.reasoning_effort
+
+            # 将 extra_body 传入 kwargs（OpenAI SDK 会将其传递给 API）
+            if "extra_body" not in kwargs:
+                kwargs["extra_body"] = extra_body
+            else:
+                # 合并已有的 extra_body
+                kwargs["extra_body"].update(extra_body)
+
+        # --- 调用父类 generate ---
+        # 注意：移除不支持的温度等参数
+        unsupported_params = ["temperature", "top_p", "presence_penalty", "frequency_penalty"]
+        for param in unsupported_params:
+            if param in kwargs:
+                logger.warning("DeepSeek 思考模式不支持 %s 参数，已自动移除", param)
+                kwargs.pop(param)
+
+        return super().generate(
+            messages=messages,
+            tools=tools,
+            tool_choice=tool_choice,
+            **kwargs,
+        )
+
+
+# ===========================================================================
+# Tool 1: 参数提取 —— 从自然语言中解析出数据集查询参数
+# ===========================================================================
+class ParameterExtractionTool(Tool):
+    """从用户的自然语言描述中提取 Hugging Face 数据集查询所需的结构化参数。"""
+
+    name = "parameter_extractor"
+    description = (
+        "从用户的自然语言查询请求中提取数据集查询参数，包括："
+        "region（区域，如中西太平洋）、time_range（时间范围，如1967-2024）、"
+        "time_scale（时间尺度，如月尺度）、spatial_resolution（空间分辨率，如1x1）、"
+        "file_name（文件名）。返回 JSON 格式的参数字典。"
+    )
+    inputs = {
+        "user_query": {
+            "type": "string",
+            "description": (
+                "用户的自然语言查询请求，例如'查询中西太平洋1967-2024年月尺度1x1分辨率的渔获数据'"
+            ),
+        }
+    }
+    output_type = "string"
+
+    # --- 区域关键字映射表 ---
+    REGION_KEYWORDS = {
+        "中西太平洋": "中西太平洋WCPFC",
+        "wcpfc": "中西太平洋WCPFC",
+        "东太平洋": "东太平洋",
+        "太平洋": "中西太平洋WCPFC",
+        "印度洋": "印度洋",
+        "大西洋": "大西洋",
+        "南海": "南海",
+    }
+
+    # --- 时间尺度映射表 ---
+    TIME_SCALE_KEYWORDS = {
+        "月尺度": "月尺度",
+        "月度": "月尺度",
+        "月": "月尺度",
+        "年尺度": "年尺度",
+        "年度": "年尺度",
+        "年": "年尺度",
+        "日尺度": "日尺度",
+        "日度": "日尺度",
+        "日": "日尺度",
+    }
+
+    # --- 空间分辨率映射表 ---
+    SPATIAL_RESOLUTION_KEYWORDS = {
+        "1x1": "1x1",
+        "1度": "1x1",
+        "0.5x0.5": "0.5x0.5",
+        "0.5度": "0.5x0.5",
+        "0.1x0.1": "0.1x0.1",
+        "0.1度": "0.1x0.1",
+    }
+
+    def forward(self, user_query: str) -> str:
+        """
+        解析自然语言，提取数据集查询参数。
+
+        参数提取流程：
+        1. 识别目标区域（优先匹配关键字映射表）
+        2. 识别时间范围（起始年份-结束年份）
+        3. 识别时间尺度（月尺度/年尺度/日尺度）
+        4. 识别空间分辨率（1x1/0.5x0.5等）
+        5. 识别文件名（用户指定的具体文件名）
+        """
+        logger.info("开始提取参数，输入: %s", user_query)
+
+        try:
+            params = {
+                "region": self._extract_region(user_query),
+                "time_range": self._extract_time_range(user_query),
+                "time_scale": self._extract_time_scale(user_query),
+                "spatial_resolution": self._extract_spatial_resolution(user_query),
+                "file_name": self._extract_file_name(user_query),
+            }
+
+            # 校验必要参数：区域不能为空
+            if not params["region"]:
+                error_msg = (
+                    "无法从输入中识别目标区域，请在查询中明确指定区域。"
+                    f"支持的区域关键字: {list(self.REGION_KEYWORDS.keys())}"
+                )
+                logger.warning(error_msg)
+                return json.dumps({"error": error_msg}, ensure_ascii=False)
+
+            logger.info("参数提取成功: %s", params)
+            return json.dumps(params, ensure_ascii=False)
+
+        except Exception as e:
+            logger.error("参数提取失败: %s", e, exc_info=True)
+            return json.dumps(
+                {"error": f"参数提取过程发生错误: {str(e)}"},
+                ensure_ascii=False,
+            )
+
+    # ----- 以下为私有辅助方法，按提取维度拆分 -----
+
+    def _extract_region(self, query: str) -> Optional[str]:
+        """
+        从自然语言中识别目标区域。
+        优先级：关键字映射 → 直接匹配区域名称
+        """
+        # 优先通过关键字映射识别
+        for cn_keyword, region_name in self.REGION_KEYWORDS.items():
+            if cn_keyword.lower() in query.lower():
+                return region_name
+
+        return None
+
+    def _extract_time_range(self, query: str) -> Optional[str]:
+        """
+        从自然语言中识别时间范围。
+        匹配"XXXX-XXXX"格式或"XXXX年到XXXX年"格式。
+        """
+        # 匹配"1967-2024"格式
+        match = re.search(r"(\d{4})-(\d{4})", query)
+        if match:
+            return f"{match.group(1)}-{match.group(2)}"
+
+        # 匹配"1967年到2024年"或"1967至2024"格式
+        match = re.search(r"(\d{4})\s*(?:年到|至|-)\s*(\d{4})", query)
+        if match:
+            return f"{match.group(1)}-{match.group(2)}"
+
+        # 匹配单一年份"2024年"
+        match = re.search(r"(\d{4})\s*年", query)
+        if match:
+            year = match.group(1)
+            return f"{year}-{year}"
+
+        return None
+
+    def _extract_time_scale(self, query: str) -> Optional[str]:
+        """
+        从自然语言中识别时间尺度。
+        优先级：关键字映射 → 默认月尺度
+        """
+        for cn_keyword, scale_name in self.TIME_SCALE_KEYWORDS.items():
+            if cn_keyword in query:
+                return scale_name
+
+        return "月尺度"  # 默认月尺度
+
+    def _extract_spatial_resolution(self, query: str) -> Optional[str]:
+        """
+        从自然语言中识别空间分辨率。
+        匹配"XxX"格式或"X度"格式。
+        """
+        # 优先通过关键字映射识别
+        for cn_keyword, resolution in self.SPATIAL_RESOLUTION_KEYWORDS.items():
+            if cn_keyword in query:
+                return resolution
+
+        # 匹配"1x1"格式
+        match = re.search(r"(\d+(?:\.\d+)?)x(\d+(?:\.\d+)?)", query)
+        if match:
+            return f"{match.group(1)}x{match.group(2)}"
+
+        return "1x1"  # 默认1x1分辨率
+
+    def _extract_file_name(self, query: str) -> Optional[str]:
+        """
+        从自然语言中识别具体文件名。
+        匹配"xxx.zip"或"xxx数据"模式。
+        """
+        # 匹配".zip"结尾的文件名
+        match = re.search(r"(\S+)\.zip", query)
+        if match:
+            return f"{match.group(1)}.zip"
+
+        # 匹配"xxx文件"或"xxx数据"模式
+        match = re.search(r"(\w+)\s*(?:文件|数据)", query)
+        if match:
+            return f"{match.group(1)}.zip"
+
+        return None
+
+
+# ===========================================================================
+# Tool 2: Hugging Face 数据集获取 —— 从云端获取并解析数据
+# ===========================================================================
+class HFDataQueryTool(Tool):
+    """从 Hugging Face 数据集安全地获取数据并返回结构化结果。"""
+
+    name = "hf_data_query"
+    description = (
+        "根据提取的结构化参数从 Hugging Face 数据集获取数据。"
+        "参数需包含 file_path（文件路径，如 '文件夹名/文件名.csv'）。"
+        "支持多种格式：ZIP（自动解压）、CSV、Excel、Parquet、JSON。"
+        "使用 hf_hub_download 下载文件，自动缓存到本地。"
+    )
+    inputs = {
+        "query_params_json": {
+            "type": "string",
+            "description": (
+                "JSON 格式的查询参数，由 parameter_extractor 工具生成或用户直接提供文件路径。"
+                "格式示例: {\"file_path\": \"中西太平洋WCPFC 1967-2024 月尺度 1x1/渔获数据.csv\"}"
+            ),
+        }
+    }
+    output_type = "string"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._hf_token = HF_TOKEN
+        self._repo_id = HF_DATASET_REPO
+
+    def forward(self, query_params_json: str) -> str:
+        """
+        解析参数 JSON，使用 hf_hub_download 获取文件并解析数据。
+
+        处理流程：
+        1. 解析参数 JSON，获取文件路径
+        2. 使用 hf_hub_download 下载文件（支持任意格式）
+        3. 根据文件格式解析数据（ZIP/CSV/Excel/Parquet/JSON）
+        4. 返回结构化的查询结果
+        """
+        logger.info("收到数据获取请求，参数: %s", query_params_json)
+
+        # ----- 步骤1: 解析参数 JSON -----
+        try:
+            params = json.loads(query_params_json)
+        except json.JSONDecodeError as e:
+            logger.error("参数 JSON 解析失败: %s", e)
+            return json.dumps(
+                {"error": f"参数格式错误，无法解析为 JSON: {str(e)}"},
+                ensure_ascii=False,
+            )
+
+        # 检查参数提取阶段是否已产生错误
+        if "error" in params:
+            return query_params_json  # 直接返回上游错误信息
+
+        # ----- 步骤2: 获取文件路径 -----
+        # 优先使用用户直接指定的 file_path，否则尝试从其他参数构建
+        file_path = params.get("file_path")
+        if not file_path:
+            # 尝试从旧格式参数构建路径
+            file_path = self._build_file_path(params)
+
+        if not file_path:
+            return json.dumps(
+                {"error": "缺少必填参数: file_path（文件路径），请指定要获取的文件"},
+                ensure_ascii=False,
+            )
+
+        logger.info("准备获取文件: %s", file_path)
+
+        # ----- 步骤3: 使用 hf_hub_download 下载文件 -----
+        try:
+            local_file_path = hf_hub_download(
+                repo_id=self._repo_id,
+                filename=file_path,
+                repo_type="dataset",
+                token=self._hf_token,
+            )
+            logger.info("文件下载成功，本地路径: %s", local_file_path)
+        except Exception as e:
+            logger.error("文件下载失败: %s", e, exc_info=True)
+            return json.dumps(
+                {"error": f"文件下载失败: {str(e)}"},
+                ensure_ascii=False,
+            )
+
+        # ----- 步骤4: 根据文件格式解析数据 -----
+        try:
+            data_content = self._parse_file_by_format(local_file_path)
+            if not data_content:
+                return json.dumps(
+                    {"error": f"文件解析失败，未找到有效的数据内容"},
+                    ensure_ascii=False,
+                )
+            logger.info("数据解析成功，返回 %d 条记录", len(data_content))
+        except Exception as e:
+            logger.error("数据解析失败: %s", e, exc_info=True)
+            return json.dumps(
+                {"error": f"数据解析过程发生错误: {str(e)}"},
+                ensure_ascii=False,
+            )
+
+        # ----- 步骤5: 返回结构化结果 -----
+        display_limit = 100
+        result = {
+            "success": True,
+            "source": f"{self._repo_id}/{file_path}",
+            "local_path": local_file_path,
+            "total_records": len(data_content),
+            "display_records": min(len(data_content), display_limit),
+            "data_preview": data_content[:display_limit],
+            "columns": list(data_content[0].keys()) if data_content else [],
+        }
+
+        return json.dumps(result, ensure_ascii=False, default=str)
+
+    def _build_file_path(self, params: dict) -> Optional[str]:
+        """
+        从旧格式参数构建文件路径（兼容之前的参数提取方式）。
+
+        Args:
+            params: 包含 region、time_range、time_scale、spatial_resolution、file_name 的字典
+
+        Returns:
+            构建的文件路径字符串，如 '中西太平洋WCPFC 1967-2024 月尺度 1x1/渔获数据.zip'
+        """
+        region = params.get("region", "")
+        time_range = params.get("time_range", "")
+        time_scale = params.get("time_scale", "月尺度")
+        spatial_resolution = params.get("spatial_resolution", "1x1")
+        file_name = params.get("file_name")
+
+        if not region or not file_name:
+            return None
+
+        folder_path = f"{region} {time_range} {time_scale} {spatial_resolution}"
+        return f"{folder_path}/{file_name}"
+
+    def _parse_file_by_format(self, local_file_path: str) -> List[dict]:
+        """
+        根据文件扩展名选择解析方式。
+
+        支持的格式：
+        - ZIP：解压后解析内部数据文件
+        - CSV：直接读取
+        - Excel (.xlsx/.xls)：使用 pandas 读取
+        - Parquet：使用 pandas 读取
+        - JSON：直接解析
+
+        Args:
+            local_file_path: 本地文件路径
+
+        Returns:
+            解析后的数据列表，每个元素为字典
+        """
+        file_lower = local_file_path.lower()
+
+        # ZIP 文件：解压并解析
+        if file_lower.endswith(".zip"):
+            logger.info("检测到 ZIP 文件，开始解压并解析")
+            with open(local_file_path, "rb") as f:
+                zip_content = f.read()
+            return self._extract_and_parse_zip(zip_content)
+
+        # CSV 文件：直接读取
+        if file_lower.endswith(".csv"):
+            logger.info("检测到 CSV 文件，开始解析")
+            return self._parse_csv_file(local_file_path)
+
+        # Excel 文件：使用 pandas 读取
+        if file_lower.endswith(".xlsx") or file_lower.endswith(".xls"):
+            logger.info("检测到 Excel 文件，开始解析")
+            return self._parse_excel_file(local_file_path)
+
+        # Parquet 文件：使用 pandas 读取
+        if file_lower.endswith(".parquet"):
+            logger.info("检测到 Parquet 文件，开始解析")
+            return self._parse_parquet_file(local_file_path)
+
+        # JSON 文件：直接解析
+        if file_lower.endswith(".json"):
+            logger.info("检测到 JSON 文件，开始解析")
+            return self._parse_json_file(local_file_path)
+
+        logger.warning("未知文件格式: %s，尝试作为 CSV 解析", local_file_path)
+        # 尝试作为 CSV 解析
+        return self._parse_csv_file(local_file_path)
+
+    def _parse_csv_file(self, file_path: str) -> List[dict]:
+        """解析 CSV 文件（支持多种编码）。"""
+        encodings = ["utf-8", "gbk", "gb2312", "latin1"]
+        for encoding in encodings:
+            try:
+                df = pd.read_csv(file_path, encoding=encoding)
+                records = df.to_dict(orient="records")
+                logger.info("CSV 解析成功，编码: %s，记录数: %d", encoding, len(records))
+                return records
+            except UnicodeDecodeError:
+                continue
+            except Exception as e:
+                logger.error("CSV 解析失败（编码 %s）: %s", encoding, e)
+                continue
+
+        logger.error("CSV 文件编码无法识别")
+        return []
+
+    def _parse_excel_file(self, file_path: str) -> List[dict]:
+        """解析 Excel 文件（.xlsx/.xls）。"""
+        try:
+            df = pd.read_excel(file_path)
+            records = df.to_dict(orient="records")
+            logger.info("Excel 解析成功，记��数: %d", len(records))
+            return records
+        except Exception as e:
+            logger.error("Excel 解析失败: %s", e)
+            return []
+
+    def _parse_parquet_file(self, file_path: str) -> List[dict]:
+        """解析 Parquet 文件。"""
+        try:
+            df = pd.read_parquet(file_path)
+            records = df.to_dict(orient="records")
+            logger.info("Parquet 解析成功，记录数: %d", len(records))
+            return records
+        except Exception as e:
+            logger.error("Parquet 解析失败: %s", e)
+            return []
+
+    def _parse_json_file(self, file_path: str) -> List[dict]:
+        """解析 JSON 文件。"""
+        try:
+            with open(file_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+
+            if isinstance(data, list):
+                logger.info("JSON 解析成功，记录数: %d", len(data))
+                return data
+
+            if isinstance(data, dict):
+                for key in ["data", "records", "items", "results"]:
+                    if key in data and isinstance(data[key], list):
+                        logger.info("JSON 解析成功，从 '%s' 字段提取 %d 条记录", key, len(data[key]))
+                        return data[key]
+
+            logger.warning("JSON 格式无法识别，尝试包装为单条记录")
+            return [data]
+        except Exception as e:
+            logger.error("JSON 解析失败: %s", e)
+            return []
+
+    def _extract_and_parse_zip(self, zip_content: bytes) -> list:
+        """
+        解压 ZIP 文件并解析其中的数据文件。
+
+        支持的数据格式：
+        - CSV (.csv)
+        - Parquet (.parquet)
+        - JSON (.json)
+        - Excel (.xlsx/.xls)
+
+        返回: 包含字典元素的列表，每个字典代表一条记录
+        """
+        data_records = []
+
+        # 使用 io.BytesIO 将 bytes 转为文件对象
+        with zipfile.ZipFile(io.BytesIO(zip_content), "r") as zip_file:
+            # 获取 ZIP 内所有文件名
+            file_names = zip_file.namelist()
+            logger.info("ZIP 文件包含: %s", file_names)
+
+            # 遍历 ZIP 内的文件，寻找可解析的数据文件
+            for file_name in file_names:
+                # 跳过目录和非数据文件
+                if file_name.endswith("/") or not self._is_data_file(file_name):
+                    continue
+
+                logger.info("正在解析数据文件: %s", file_name)
+                file_content = zip_file.read(file_name)
+
+                # 根据文件扩展名选择解析方式（使用内存中的 bytes 解析）
+                if file_name.lower().endswith(".csv"):
+                    data_records = self._parse_csv_bytes(file_content)
+                elif file_name.lower().endswith(".parquet"):
+                    data_records = self._parse_parquet_bytes(file_content)
+                elif file_name.lower().endswith(".json"):
+                    data_records = self._parse_json_bytes(file_content)
+                elif file_name.lower().endswith((".xlsx", ".xls")):
+                    data_records = self._parse_excel_bytes(file_content)
+
+                # 找到第一个有效的数据文件就返回
+                if data_records:
+                    break
+
+        if not data_records:
+            logger.warning("ZIP 文件中未找到可解析的数据文件")
+
+        return data_records
+
+    def _is_data_file(self, file_name: str) -> bool:
+        """判断文件是否为可解析的数据文件。"""
+        data_extensions = {".csv", ".parquet", ".json", ".xlsx", ".xls"}
+        return any(file_name.lower().endswith(ext) for ext in data_extensions)
+
+    def _parse_csv_bytes(self, file_content: bytes) -> list:
+        """解析内存中的 CSV 文件内容（从 ZIP 解压）。"""
+        encodings = ["utf-8", "gbk", "gb2312", "latin1"]
+        for encoding in encodings:
+            try:
+                df = pd.read_csv(io.BytesIO(file_content), encoding=encoding)
+                records = df.to_dict(orient="records")
+                logger.info("CSV 解析成功（内存），编码: %s，记录数: %d", encoding, len(records))
+                return records
+            except UnicodeDecodeError:
+                continue
+            except Exception as e:
+                logger.error("CSV 解析失败（编码 %s）: %s", encoding, e)
+                continue
+        return []
+
+    def _parse_parquet_bytes(self, file_content: bytes) -> list:
+        """解析内存中的 Parquet 文件内容（从 ZIP 解压）。"""
+        try:
+            df = pd.read_parquet(io.BytesIO(file_content))
+            records = df.to_dict(orient="records")
+            logger.info("Parquet 解析成功（内存），记录数: %d", len(records))
+            return records
+        except Exception as e:
+            logger.error("Parquet 解析失败: %s", e)
+            return []
+
+    def _parse_json_bytes(self, file_content: bytes) -> list:
+        """解析内存中的 JSON 文件内容（从 ZIP 解压）。"""
+        try:
+            data = json.loads(file_content.decode("utf-8"))
+            if isinstance(data, list):
+                logger.info("JSON 解析成功（内存），记录数: %d", len(data))
+                return data
+            if isinstance(data, dict):
+                for key in ["data", "records", "items", "results"]:
+                    if key in data and isinstance(data[key], list):
+                        logger.info("JSON 解析成功（内存），从 '%s' 字段提取 %d 条记录", key, len(data[key]))
+                        return data[key]
+            return [data]
+        except Exception as e:
+            logger.error("JSON 解析失败: %s", e)
+            return []
+
+    def _parse_excel_bytes(self, file_content: bytes) -> list:
+        """解析内存中的 Excel 文件内容（从 ZIP 解压）。"""
+        try:
+            df = pd.read_excel(io.BytesIO(file_content))
+            records = df.to_dict(orient="records")
+            logger.info("Excel 解析成功（内存），记录数: %d", len(records))
+            return records
+        except Exception as e:
+            logger.error("Excel 解析失败: %s", e)
+            return []
+
+
+# ===========================================================================
+# Agent 工厂函数：创建配置好的 ToolCallingAgent
+# ===========================================================================
+def create_hf_data_agent(
+    model_id: str = MODEL_ID,
+    api_key: Optional[str] = None,
+    api_base: Optional[str] = None,
+    thinking_enabled: bool = THINKING_MODE_ENABLED,
+    reasoning_effort: str = THINKING_EFFORT,
+    max_steps: int = 5,
+) -> ToolCallingAgent:
+    """
+    创建并返回一个配置好的 Hugging Face 数据集查询智能代理。
+
+    在 Agent 启动时会自动获取数据集文件列表并作为系统提示词告知用户。
+
+    Args:
+        model_id: 模型 ID，默认使用文件顶部 MODEL_ID 常量
+        api_key: API Key，默认使用文件顶部 OPENAI_API_KEY 常量
+        api_base: API 地址，默认使用文件顶部 OPENAI_API_BASE 常量
+        thinking_enabled: 是否启用 DeepSeek 思考模式
+        reasoning_effort: 思考强度（low/medium/high/max）
+        max_steps: Agent 最大推理步骤数
+
+    Returns:
+        配置好的 ToolCallingAgent 实例
+    """
+    key = api_key or OPENAI_API_KEY
+    base = api_base or OPENAI_API_BASE
+
+    # 初始化 LLM 模型（使用 DeepSeekThinkingModel 处理思考模式）
+    model = DeepSeekThinkingModel(
+        model_id=model_id,
+        api_key=key,
+        api_base=base,
+        thinking_enabled=thinking_enabled,
+        reasoning_effort=reasoning_effort,
+    )
+
+    # 实例化工具
+    param_tool = ParameterExtractionTool()
+    hf_data_tool = HFDataQueryTool()
+
+    # 构建包含数据集文件列表的指令（会被插入到系统提示词中）
+    instructions = build_system_prompt_with_file_list()
+    logger.info("Agent 自定义指令: %s", instructions[:200] + "...")  # 只打印前200字符
+
+    # 创建 ToolCallingAgent，使用 instructions 参数设置自定义指令
+    agent = ToolCallingAgent(
+        tools=[param_tool, hf_data_tool],
+        model=model,
+        max_steps=max_steps,
+        instructions=instructions,
+    )
+
+    logger.info("Hugging Face 数据集查询代理创建成功，模型: %s", model_id)
+    return agent
+
+
+# ===========================================================================
+# 主入口：GradioUI 交互式 Web 界面
+# ===========================================================================
+def main():
+    """
+    使用 GradioUI 启动交互式 Web 界面。
+    用户在浏览器中输入自然语言查询，代理实时展示思考过程和数据获取结果。
+    界面基于 gr.ChatInterface，支持流式输出和步骤可视化。
+    """
+    agent = create_hf_data_agent()
+
+    # GradioUI 将 agent 包装为 gr.ChatInterface Web 应用
+    # - share=True: 生成公网可访问的临时链接（72小时有效）
+    # - reset_agent_memory=False: 保留对话上下文，允许多轮交互
+    demo = GradioUI(
+        agent,
+        reset_agent_memory=False,
+    )
+
+    logger.info("正在启动 GradioUI Web 界面...")
+    demo.launch(share=True)
+
+
+if __name__ == "__main__":
+    main()