更新为llm判断工具函数调用

README.md

@@ -14,11 +14,12 @@ hf_oauth_expiration_minutes: 480
 
 这是 Hugging Face Agents Course Final Assignment 的基础 Space，用于运行并提交 GAIA 风格问题的 Agent 作答结果。
 
-当前版本已经改成工具路由 Agent：
+当前版本已经改成 LLM 工具调用 Agent：
 
 - `app.py` 会通过课程评测接口拉取问题并提交答案。
-- `agent.py` 中的 `GaiaAgent` 会按题型路由到专用 solver。
-- `tools/` 目录包含规则题、附件下载、Python 执行、Excel 计算、结构化网页和体育数据工具。
+- `agent.py` 中的 `GaiaAgent` 会让 LLM planner 输出 JSON action，决定下一步调用哪个工具。
+- `tools/executor.py` 和 `tools/tool_specs.py` 负责注册、描述并执行工具。
+- `tools/` 目录包含规则题、附件下载、Python 执行、Excel 计算、结构化网页、网页检索和体育数据工具。
 - 登录 Hugging Face 后，界面会使用当前 HF 用户名提交答案。
 - 提交时会携带当前 Space 的代码仓库链接，便于评测系统记录实现来源。
 
@@ -44,7 +45,9 @@ Space 配置参考：https://huggingface.co/docs/hub/spaces-config-reference
 │   ├── code_runner.py
 │   ├── spreadsheet_solver.py
 │   ├── structured_web_tools.py
-│   └── sports_solver.py
+│   ├── sports_solver.py
+│   ├── tool_specs.py
+│   └── executor.py
 ├── data/
 │   ├── questions.json
 │   ├── gold_local.jsonl
@@ -65,7 +68,21 @@ Space 配置参考：https://huggingface.co/docs/hub/spaces-config-reference
 6.  维护 gold 文件：`gold_local.jsonl` 每行包含 `task_id`、`expected_answer`、`match_type` 和备注。没有标准答案的问题只做流程测试，不纳入正确率统计。
 7.  分层运行：先跑 1 到 3 道 smoke test，再跑全部本地题；确认稳定后再通过 Space 按官方流程提交。
 
-当前代码采用 40% 优先策略：为了先冲过最低分，不处理音频和视频题，只集中处理文本、网页、Python、Excel、体育统计和少量确定性规则题。
+当前代码采用 LLM JSON action loop：LLM 每一步只能输出一个 JSON 对象，要么调用工具，要么返回最终答案。Python 不再按固定顺序手动选择 solver，而是只负责执行 LLM 指定的工具。
+
+工具调用格式：
+
+```json
+{"thought":"需要先查 Wikipedia 表格","action":"wikipedia_tool","args":{"query":"1928 Summer Olympics athletes"}}
+```
+
+最终答案格式：
+
+```json
+{"thought":"工具已给出高置信答案","action":"final_answer","answer":"CUB","confidence":"high"}
+```
+
+当前仍采用 40% 优先策略：为了先冲过最低分，不处理音频和视频题，只集中处理文本、网页、Python、Excel、体育统计和少量确定性规则题。
 
 当前优先覆盖的题型：
 
@@ -85,6 +102,8 @@ Space 配置参考：https://huggingface.co/docs/hub/spaces-config-reference
 python3 gaia_local_eval.py
 ```
 
+注意：现在本地回归会调用 LLM planner，因此需要 `HF_TOKEN`，并会消耗少量 Hugging Face Inference Providers 额度。这个 token 必须具备 `Make calls to Inference Providers` 权限，否则第一步 planner 请求会返回 403。`MAX_AGENT_STEPS` 可用于限制每题最多工具调用步数。
+
 如果本地没有安装依赖，先执行：
 
 ```bash

agent.py

@@ -1,94 +1,142 @@
-from pathlib import Path
-
-from config import HF_TEXT_MODEL, HF_VISION_MODEL
-from tools.attachment_loader import download_task_file
-from tools.code_runner import solve_python_output
-from tools.common import (
-    AUDIO_VIDEO_EXTENSIONS,
-    IMAGE_EXTENSIONS,
-    TEXT_EXTENSIONS,
-    extract_urls,
-    is_youtube_url,
-    normalize_answer,
-    read_plain_file,
-)
-from tools.direct_rules import solve_direct
-from tools.llm_client import answer_with_light_model
-from tools.sports_solver import solve_sports
-from tools.spreadsheet_solver import solve_excel_food_sales
-from tools.structured_web_tools import solve_structured_web
-from tools.types import SolverResult, unresolved
-from tools.web_tools import collect_web_evidence, fetch_url_text
+import json
+from typing import Any
+
+from config import HF_TEXT_MODEL, HF_VISION_MODEL, MAX_AGENT_STEPS, MAX_TOOL_OUTPUT_CHARS
+from tools.common import normalize_answer, truncate_text
+from tools.executor import execute_tool, tool_prompt
+from tools.llm_client import AgentAction, plan_next_action
+from tools.types import SolverResult
+
+
+SYSTEM_PROMPT = """
+You are a GAIA tool-using agent. You must decide which tool to call based on the question, file name,
+and prior tool observations.
+
+Rules:
+1. You control tool selection. Do not ask the user for clarification.
+2. Use low-cost deterministic tools before generic web search when they fit.
+3. Do not guess. If enabled tools cannot solve a task, return final_answer "无法确定".
+4. Audio, video, and image/chess tools are currently disabled. If a task requires them, return "无法确定".
+5. Every response must be one valid JSON object and nothing else.
+6. To call a tool, output:
+   {"thought":"...","action":"tool_name","args":{...}}
+7. To finish, output:
+   {"thought":"...","action":"final_answer","answer":"...","confidence":"high|medium|low"}
+8. Final answers must obey the requested format exactly. Do not include explanations.
+""".strip()
 
 
 class GaiaAgent:
-    """面向当前 GAIA 20 题的工具优先 Agent。"""
+    """LLM 主导的 JSON action-loop Agent。"""
 
     def __init__(self):
-        print("GAIA 工具路由 Agent 已初始化。")
+        print("GAIA LLM 工具调用 Agent 已初始化。")
         print(f"文本模型：{HF_TEXT_MODEL}")
         print(f"视觉模型：{HF_VISION_MODEL or '未启用'}")
+        print(f"最大 Agent 步数：{MAX_AGENT_STEPS}")
 
     def answer_task(self, question: str, task_id: str = "", file_name: str = "") -> SolverResult:
         print(f"Agent 收到问题（前 100 个字符）：{question[:100]}...")
+        context = {
+            "question": question,
+            "task_id": task_id,
+            "file_name": file_name,
+        }
+        messages = self._initial_messages(context)
+        trace: list[dict[str, Any]] = []
+        best_tool_answer: dict[str, Any] | None = None
+
+        for step_index in range(1, MAX_AGENT_STEPS + 1):
+            try:
+                action = plan_next_action(messages)
+            except Exception as exc:
+                trace.append(
+                    {
+                        "step": step_index,
+                        "event": "planner_error",
+                        "error": str(exc),
+                    }
+                )
+                return SolverResult(
+                    "无法确定",
+                    source="llm_agent.planner_error",
+                    confidence="low",
+                    evidence=self._trace_text(trace),
+                    error=str(exc),
+                )
 
-        media_result = self._skip_media_if_needed(question, file_name)
-        if media_result.solved:
-            return media_result
-
-        # 1. 纯规则题：不联网、不调用模型。
-        result = solve_direct(question)
-        if result.solved:
-            return result
-
-        # 2. 附件类确定性题。
-        result = solve_python_output(question, task_id, file_name)
-        if result.solved:
-            return result
-
-        result = solve_excel_food_sales(question, task_id, file_name)
-        if result.solved:
-            return result
+            trace.append(
+                {
+                    "step": step_index,
+                    "event": "planner_action",
+                    "action": action.action,
+                    "args": action.args,
+                    "thought": action.thought,
+                }
+            )
 
-        # 3. 结构化网页/体育数据：优先走专用解析器。
-        result = solve_sports(question)
-        if result.solved:
-            return result
+            if action.is_final:
+                final_answer = self._extract_final_answer(action)
+                return SolverResult(
+                    final_answer or "无法确定",
+                    source="llm_agent.final",
+                    confidence=action.confidence,
+                    evidence=self._trace_text(trace),
+                )
 
-        result = solve_structured_web(question)
-        if result.solved:
-            return result
+            observation = execute_tool(action.action, action.args, context)
+            if observation.get("answer"):
+                best_tool_answer = observation
+            compact_observation = self._compact_observation(observation)
+            trace.append(
+                {
+                    "step": step_index,
+                    "event": "tool_observation",
+                    "observation": compact_observation,
+                }
+            )
 
-        # 4. 图片题暂时不投入，除非后续补视觉/棋局 solver。
-        if file_name and Path(file_name).suffix.lower() in IMAGE_EXTENSIONS:
-            return SolverResult(
-                "无法确定",
-                source="agent.skip_image",
-                confidence="low",
-                evidence="当前 40% 策略暂不处理图片/棋局题。",
+            messages.append(
+                {
+                    "role": "assistant",
+                    "content": json.dumps(
+                        {
+                            "thought": action.thought,
+                            "action": action.action,
+                            "args": action.args,
+                        },
+                        ensure_ascii=False,
+                    ),
+                }
+            )
+            messages.append(
+                {
+                    "role": "user",
+                    "content": (
+                        "TOOL_OBSERVATION:\n"
+                        + json.dumps(compact_observation, ensure_ascii=False)
+                        + "\n\nDecide the next tool call or return final_answer as JSON."
+                    ),
+                }
             )
 
-        # 5. 最后才构造证据交给轻量模型兜底。
-        evidence = self._build_fallback_evidence(question, task_id, file_name)
-        try:
-            answer = answer_with_light_model(question, evidence)
-            if answer:
-                return SolverResult(
-                    normalize_answer(answer),
-                    source="llm_fallback",
-                    confidence="medium",
-                    evidence=evidence,
-                )
-        except Exception as exc:
+        if best_tool_answer and best_tool_answer.get("answer"):
+            # 达到步数上限时，使用模型已经选择过的工具返回的候选答案兜底。
             return SolverResult(
-                "无法确定",
-                source="llm_fallback.error",
-                confidence="low",
-                evidence=evidence,
-                error=f"{exc}",
+                normalize_answer(str(best_tool_answer["answer"])),
+                source=f"llm_agent.step_limit_best_tool:{best_tool_answer.get('source')}",
+                confidence=str(best_tool_answer.get("confidence", "medium")),
+                evidence=self._trace_text(trace),
+                error="达到 MAX_AGENT_STEPS，使用最佳工具候选答案。",
             )
 
-        return unresolved("agent", "所有 solver 均未给出答案。", evidence)
+        return SolverResult(
+            "无法确定",
+            source="llm_agent.step_limit",
+            confidence="low",
+            evidence=self._trace_text(trace),
+            error="达到 MAX_AGENT_STEPS，且没有工具候选答案。",
+        )
 
     def __call__(self, question: str, task_id: str = "", file_name: str = "") -> str:
         result = self.answer_task(question, task_id=task_id, file_name=file_name)
@@ -98,59 +146,40 @@ class GaiaAgent:
         )
         return normalize_answer(result.answer or "无法确定")
 
-    def _skip_media_if_needed(self, question: str, file_name: str) -> SolverResult:
-        extension = Path(file_name).suffix.lower() if file_name else ""
-        if extension in AUDIO_VIDEO_EXTENSIONS:
-            return SolverResult(
-                "无法确定",
-                source="agent.skip_audio_video_attachment",
-                confidence="low",
-                evidence=f"当前 40% 策略跳过音视频附件：{file_name}",
-            )
-
-        urls = extract_urls(question)
-        if any(is_youtube_url(url) for url in urls):
-            return SolverResult(
-                "无法确定",
-                source="agent.skip_youtube",
-                confidence="low",
-                evidence="当前 40% 策略跳过 YouTube 视频题。",
-            )
-        return unresolved("agent.media_skip")
-
-    def _build_fallback_evidence(self, question: str, task_id: str, file_name: str) -> str:
-        evidence_parts = []
-
-        if file_name:
-            file_path, note = download_task_file(task_id, file_name)
-            evidence_parts.append(note)
-            if file_path:
-                extension = Path(file_name).suffix.lower()
-                if extension in TEXT_EXTENSIONS:
-                    evidence_parts.append(f"文本附件内容：\n{read_plain_file(file_path)}")
-                else:
-                    evidence_parts.append(f"附件类型 {extension} 没有专用兜底处理。")
-
-        urls = extract_urls(question)
-        for url in urls:
-            if not is_youtube_url(url):
-                evidence_parts.append(f"网页 {url} 内容：\n{fetch_url_text(url)}")
-
-        if not urls and not self._is_self_contained(question):
-            evidence_parts.append(f"搜索证据：\n{collect_web_evidence(question)}")
-
-        return "\n\n".join(evidence_parts) or "没有额外工具证据。"
-
-    def _is_self_contained(self, question: str) -> bool:
-        lower_question = question.lower()
-        if "|---|" in question or "given this table" in lower_question:
-            return True
-        if "grocery list" in lower_question or "here's the list" in lower_question:
-            return True
-        reversed_question = question[::-1].lower()
-        if "if you understand this sentence" in reversed_question:
-            return True
-        return False
+    def _initial_messages(self, context: dict[str, str]) -> list[dict[str, str]]:
+        task_payload = {
+            "task_id": context.get("task_id", ""),
+            "file_name": context.get("file_name", ""),
+            "question": context["question"],
+        }
+        user_prompt = (
+            "Available tools:\n"
+            + tool_prompt()
+            + "\n\nTask:\n"
+            + json.dumps(task_payload, ensure_ascii=False)
+            + "\n\nChoose the first tool call or return final_answer. Output JSON only."
+        )
+        return [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": user_prompt},
+        ]
+
+    def _extract_final_answer(self, action: AgentAction) -> str:
+        answer = action.answer
+        if answer is None:
+            answer = action.args.get("answer")
+        if answer is None:
+            return "无法确定"
+        return normalize_answer(str(answer))
+
+    def _compact_observation(self, observation: dict[str, Any]) -> dict[str, Any]:
+        compact = dict(observation)
+        if compact.get("evidence"):
+            compact["evidence"] = truncate_text(str(compact["evidence"]), MAX_TOOL_OUTPUT_CHARS)
+        return compact
+
+    def _trace_text(self, trace: list[dict[str, Any]]) -> str:
+        return truncate_text(json.dumps(trace, ensure_ascii=False, indent=2), MAX_TOOL_OUTPUT_CHARS)
 
 
 # 兼容原模板里的 BasicAgent 名称。

app.py

@@ -150,19 +150,19 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
 
 
 with gr.Blocks() as demo:
-    gr.Markdown("# 工具路由 GAIA Agent 评测运行器")
+    gr.Markdown("# LLM 工具调用 GAIA Agent 评测运行器")
     gr.Markdown(
         f"""
         **当前策略：**
 
-        1.  目标是先稳定覆盖 40% 左右的低成本题型，而不是全多模态。
-        2.  规则题、Python、Excel、Wikipedia/表格、棒球统计会优先走专用 solver。
-        3.  音频、视频和棋局图暂时跳过，返回“无法确定”。
-        4.  只有专用 solver 失败时，才调用轻量文本模型 `{HF_TEXT_MODEL}` 兜底。
-        5.  图片题默认不启用视觉模型；当前视觉模型：`{HF_VISION_MODEL or "未启用"}`。
+        1.  LLM planner 会输出 JSON action，决定调用哪个工具。
+        2.  工具执行器会调用规则、Python、Excel、Wikipedia、体育统计、网页搜索等工具。
+        3.  每题最多运行有限步数，避免无限搜索和过度消耗额度。
+        4.  音频、视频和棋局图工具当前禁用；LLM 遇到这类题应返回“无法确定”。
+        5.  当前文本模型：`{HF_TEXT_MODEL}`；当前视觉模型：`{HF_VISION_MODEL or "未启用"}`。
 
         **提交前检查：**
-        Space Secrets 至少需要 `HF_TOKEN`，否则 LLM 兜底不可用；确定性 solver 不依赖它。
+        Space Secrets 至少需要 `HF_TOKEN`，因为每道题都需要 LLM planner 决定工具调用。
         """
     )
 
@@ -192,5 +192,5 @@ if __name__ == "__main__":
         print("未找到 SPACE_ID 环境变量（可能是在本地运行）。无法确定仓库地址。")
 
     print("-" * (60 + len(" 应用启动中 ")) + "\n")
-    print("正在启动工具路由 GAIA Agent 评测 Gradio 界面...")
+    print("正在启动 LLM 工具调用 GAIA Agent 评测 Gradio 界面...")
     demo.launch(debug=True, share=False)

config.py

@@ -8,6 +8,8 @@ HF_ROUTER_URL = "https://router.huggingface.co/v1/chat/completions"
 HF_TEXT_MODEL = os.getenv("HF_TEXT_MODEL", "openai/gpt-oss-20b:cheapest")
 HF_VISION_MODEL = os.getenv("HF_VISION_MODEL", "")
 MAX_EVIDENCE_CHARS = int(os.getenv("MAX_EVIDENCE_CHARS", "18000"))
+MAX_AGENT_STEPS = int(os.getenv("MAX_AGENT_STEPS", "4"))
+MAX_TOOL_OUTPUT_CHARS = int(os.getenv("MAX_TOOL_OUTPUT_CHARS", "8000"))
 CACHE_DIR = Path(os.getenv("GAIA_CACHE_DIR", tempfile.gettempdir())) / "gaia_agent_files"
 REQUEST_HEADERS = {
     "User-Agent": (
@@ -20,8 +22,15 @@ REQUEST_HEADERS = {
 
 def get_hf_token() -> str | None:
     """读取 Hugging Face Token。Space Secrets 中建议使用 HF_TOKEN。"""
-    return (
+    token = (
         os.getenv("HF_TOKEN")
         or os.getenv("HUGGING_FACE_HUB_TOKEN")
         or os.getenv("HUGGINGFACEHUB_API_TOKEN")
     )
+    if token:
+        return token
+
+    local_token_path = Path.home() / ".cache" / "huggingface" / "token"
+    if local_token_path.exists():
+        return local_token_path.read_text(encoding="utf-8").strip()
+    return None

tools/executor.py

@@ -0,0 +1,316 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from config import MAX_TOOL_OUTPUT_CHARS
+from tools.attachment_loader import download_task_file
+from tools.code_runner import solve_python_output
+from tools.common import normalize_answer, read_plain_file, truncate_text
+from tools.direct_rules import solve_direct
+from tools.sports_solver import solve_sports
+from tools.spreadsheet_solver import solve_excel_food_sales
+from tools.structured_web_tools import solve_structured_web, wikipedia_parse_html, wikipedia_search_titles
+from tools.tool_specs import ToolSpec
+from tools.web_tools import fetch_url_text, search_web
+
+
+def solver_result_to_observation(result, tool_name: str) -> dict[str, Any]:
+    return {
+        "tool": tool_name,
+        "ok": result.solved,
+        "answer": result.answer,
+        "confidence": result.confidence,
+        "source": result.source,
+        "evidence": truncate_text(result.evidence, MAX_TOOL_OUTPUT_CHARS),
+        "error": result.error,
+    }
+
+
+def disabled_tool_observation(tool_name: str, reason: str) -> dict[str, Any]:
+    return {
+        "tool": tool_name,
+        "ok": False,
+        "answer": None,
+        "confidence": "low",
+        "source": f"{tool_name}.disabled",
+        "evidence": "",
+        "error": reason,
+    }
+
+
+def direct_answer_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    question = args.get("question") or context["question"]
+    return solver_result_to_observation(solve_direct(question), "direct_answer_tool")
+
+
+def python_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    return solver_result_to_observation(
+        solve_python_output(
+            context["question"],
+            args.get("task_id") or context.get("task_id", ""),
+            args.get("file_name") or context.get("file_name", ""),
+        ),
+        "python_tool",
+    )
+
+
+def spreadsheet_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    return solver_result_to_observation(
+        solve_excel_food_sales(
+            context["question"],
+            args.get("task_id") or context.get("task_id", ""),
+            args.get("file_name") or context.get("file_name", ""),
+        ),
+        "spreadsheet_tool",
+    )
+
+
+def wikipedia_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    question = args.get("question") or context["question"]
+    result = solve_structured_web(question)
+    if result.solved:
+        return solver_result_to_observation(result, "wikipedia_tool")
+
+    query = args.get("query") or question
+    try:
+        titles = wikipedia_search_titles(query, limit=5)
+        evidence_parts = [f"search_titles={titles}"]
+        if titles:
+            html = wikipedia_parse_html(titles[0])
+            evidence_parts.append(truncate_text(html, 5000))
+        return {
+            "tool": "wikipedia_tool",
+            "ok": bool(titles),
+            "answer": None,
+            "confidence": "medium" if titles else "low",
+            "source": "wikipedia_tool.search",
+            "evidence": truncate_text("\n\n".join(evidence_parts), MAX_TOOL_OUTPUT_CHARS),
+            "error": "" if titles else "Wikipedia search returned no titles.",
+        }
+    except Exception as exc:
+        return {
+            "tool": "wikipedia_tool",
+            "ok": False,
+            "answer": None,
+            "confidence": "low",
+            "source": "wikipedia_tool.error",
+            "evidence": "",
+            "error": str(exc),
+        }
+
+
+def sports_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    question = args.get("question") or context["question"]
+    return solver_result_to_observation(solve_sports(question), "sports_tool")
+
+
+def web_search_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    query = args.get("query") or context["question"]
+    results = search_web(query, max_results=int(args.get("max_results", 5)))
+    return {
+        "tool": "web_search_tool",
+        "ok": bool(results),
+        "answer": None,
+        "confidence": "medium" if results else "low",
+        "source": "web_search_tool",
+        "evidence": truncate_text(json.dumps(results, ensure_ascii=False), MAX_TOOL_OUTPUT_CHARS),
+        "error": "" if results else "No search results.",
+    }
+
+
+def web_read_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    url = args.get("url", "")
+    if not url:
+        return {
+            "tool": "web_read_tool",
+            "ok": False,
+            "answer": None,
+            "confidence": "low",
+            "source": "web_read_tool",
+            "evidence": "",
+            "error": "Missing url.",
+        }
+    text = fetch_url_text(url, limit=MAX_TOOL_OUTPUT_CHARS)
+    return {
+        "tool": "web_read_tool",
+        "ok": bool(text and not text.startswith("无法读取网页")),
+        "answer": None,
+        "confidence": "medium",
+        "source": "web_read_tool",
+        "evidence": truncate_text(text, MAX_TOOL_OUTPUT_CHARS),
+        "error": "" if not text.startswith("无法读取网页") else text,
+    }
+
+
+def attachment_text_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    task_id = args.get("task_id") or context.get("task_id", "")
+    file_name = args.get("file_name") or context.get("file_name", "")
+    file_path, note = download_task_file(task_id, file_name)
+    if not file_path:
+        return {
+            "tool": "attachment_text_tool",
+            "ok": False,
+            "answer": None,
+            "confidence": "low",
+            "source": "attachment_text_tool",
+            "evidence": note,
+            "error": note,
+        }
+    text = read_plain_file(Path(file_path), limit=MAX_TOOL_OUTPUT_CHARS)
+    return {
+        "tool": "attachment_text_tool",
+        "ok": True,
+        "answer": None,
+        "confidence": "medium",
+        "source": "attachment_text_tool",
+        "evidence": f"{note}\n{text}",
+        "error": "",
+    }
+
+
+def final_format_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    answer = normalize_answer(str(args.get("answer", "")))
+    return {
+        "tool": "final_format_tool",
+        "ok": bool(answer),
+        "answer": answer or None,
+        "confidence": args.get("confidence", "medium"),
+        "source": "final_format_tool",
+        "evidence": "Normalized final answer.",
+        "error": "" if answer else "Missing answer.",
+    }
+
+
+def audio_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    return disabled_tool_observation("audio_tool", "当前版本未启用音频转写工具。")
+
+
+def video_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    return disabled_tool_observation("video_tool", "当前版本未启用视频/YouTube 工具。")
+
+
+def vision_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    return disabled_tool_observation("vision_tool", "当前版本未启用图片/棋局视觉工具。")
+
+
+TOOL_REGISTRY: dict[str, ToolSpec] = {
+    "direct_answer_tool": ToolSpec(
+        name="direct_answer_tool",
+        description="Low-cost deterministic solver for self-contained logic questions: reversed text, non-commutative table, botanical vegetable list.",
+        input_schema={"question": "optional string; omit to use current question"},
+        run=direct_answer_tool,
+        cost_level="free",
+    ),
+    "python_tool": ToolSpec(
+        name="python_tool",
+        description="Download and execute/analyze an attached Python file, then return the final numeric/text output if recoverable.",
+        input_schema={"task_id": "optional string", "file_name": "optional string"},
+        run=python_tool,
+        cost_level="free",
+    ),
+    "spreadsheet_tool": ToolSpec(
+        name="spreadsheet_tool",
+        description="Download and analyze an attached Excel spreadsheet; currently solves food-sales-not-drinks style calculations.",
+        input_schema={"task_id": "optional string", "file_name": "optional string"},
+        run=spreadsheet_tool,
+        cost_level="free",
+    ),
+    "wikipedia_tool": ToolSpec(
+        name="wikipedia_tool",
+        description="Use MediaWiki/Wikipedia APIs and structured parsers for Wikipedia-style questions.",
+        input_schema={"query": "optional search query", "question": "optional question override"},
+        run=wikipedia_tool,
+        cost_level="low",
+    ),
+    "sports_tool": ToolSpec(
+        name="sports_tool",
+        description="Use sports data APIs; currently handles the 1977 Yankees walks/at-bats question.",
+        input_schema={"question": "optional question override"},
+        run=sports_tool,
+        cost_level="low",
+    ),
+    "web_search_tool": ToolSpec(
+        name="web_search_tool",
+        description="Search the web and return result titles and URLs.",
+        input_schema={"query": "string", "max_results": "optional integer, default 5"},
+        run=web_search_tool,
+        cost_level="low",
+    ),
+    "web_read_tool": ToolSpec(
+        name="web_read_tool",
+        description="Read a specific URL and return extracted page text.",
+        input_schema={"url": "string"},
+        run=web_read_tool,
+        cost_level="low",
+    ),
+    "attachment_text_tool": ToolSpec(
+        name="attachment_text_tool",
+        description="Download an attachment and read it as text when it is a plain text/CSV/JSON/Markdown-like file.",
+        input_schema={"task_id": "optional string", "file_name": "optional string"},
+        run=attachment_text_tool,
+        cost_level="free",
+    ),
+    "final_format_tool": ToolSpec(
+        name="final_format_tool",
+        description="Normalize a candidate final answer string to submission-safe format.",
+        input_schema={"answer": "string", "confidence": "optional string"},
+        run=final_format_tool,
+        cost_level="free",
+    ),
+    "audio_tool": ToolSpec(
+        name="audio_tool",
+        description="Disabled audio transcription tool. Calling it reports unavailable.",
+        input_schema={},
+        run=audio_tool,
+        enabled=False,
+        cost_level="disabled",
+    ),
+    "video_tool": ToolSpec(
+        name="video_tool",
+        description="Disabled video/YouTube analysis tool. Calling it reports unavailable.",
+        input_schema={},
+        run=video_tool,
+        enabled=False,
+        cost_level="disabled",
+    ),
+    "vision_tool": ToolSpec(
+        name="vision_tool",
+        description="Disabled image/chess vision tool. Calling it reports unavailable.",
+        input_schema={},
+        run=vision_tool,
+        enabled=False,
+        cost_level="disabled",
+    ),
+}
+
+
+def tool_prompt() -> str:
+    return "\n".join(spec.prompt_block() for spec in TOOL_REGISTRY.values())
+
+
+def execute_tool(action: str, args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
+    spec = TOOL_REGISTRY.get(action)
+    if not spec:
+        return {
+            "tool": action,
+            "ok": False,
+            "answer": None,
+            "confidence": "low",
+            "source": "tool_executor.unknown_tool",
+            "evidence": "",
+            "error": f"Unknown tool: {action}",
+        }
+    if not spec.enabled:
+        return disabled_tool_observation(action, f"Tool {action} is disabled.")
+    try:
+        return spec.run(args or {}, context)
+    except Exception as exc:
+        return {
+            "tool": action,
+            "ok": False,
+            "answer": None,
+            "confidence": "low",
+            "source": f"{action}.exception",
+            "evidence": "",
+            "error": str(exc),
+        }

tools/llm_client.py

@@ -1,3 +1,6 @@
+import json
+import re
+from dataclasses import dataclass
 from typing import Any
 
 import requests
@@ -6,7 +9,25 @@ from config import HF_ROUTER_URL, HF_TEXT_MODEL, get_hf_token
 from tools.common import normalize_answer
 
 
-def call_hf_chat(messages: list[dict[str, Any]], model: str = HF_TEXT_MODEL, max_tokens: int = 512) -> str:
+@dataclass
+class AgentAction:
+    action: str
+    args: dict[str, Any]
+    thought: str = ""
+    answer: str | None = None
+    confidence: str = "medium"
+
+    @property
+    def is_final(self) -> bool:
+        return self.action == "final_answer"
+
+
+def call_hf_chat(
+    messages: list[dict[str, Any]],
+    model: str = HF_TEXT_MODEL,
+    max_tokens: int = 512,
+    temperature: float = 0.1,
+) -> str:
     token = get_hf_token()
     if not token:
         raise RuntimeError("未配置 HF_TOKEN。")
@@ -20,16 +41,64 @@ def call_hf_chat(messages: list[dict[str, Any]], model: str = HF_TEXT_MODEL, max
         json={
             "model": model,
             "messages": messages,
-            "temperature": 0.1,
+            "temperature": temperature,
             "max_tokens": max_tokens,
         },
         timeout=180,
     )
-    response.raise_for_status()
+    if not response.ok:
+        raise RuntimeError(
+            f"HF chat 请求失败：status={response.status_code}, body={response.text[:500]}"
+        )
     data = response.json()
     return data["choices"][0]["message"]["content"]
 
 
+def extract_json_object(text: str) -> dict[str, Any]:
+    text = text.strip()
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?", "", text, flags=re.IGNORECASE).strip()
+        text = re.sub(r"```$", "", text).strip()
+
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
+
+    match = re.search(r"\{.*\}", text, flags=re.DOTALL)
+    if not match:
+        raise ValueError(f"模型没有输出 JSON 对象：{text[:300]}")
+    return json.loads(match.group(0))
+
+
+def parse_agent_action(text: str) -> AgentAction:
+    data = extract_json_object(text)
+    action = data.get("action") or data.get("tool")
+    if not isinstance(action, str) or not action.strip():
+        raise ValueError(f"模型 JSON 缺少 action 字段：{data}")
+
+    args = data.get("args") or {}
+    if not isinstance(args, dict):
+        raise ValueError(f"action args 必须是对象：{data}")
+
+    answer = data.get("answer")
+    if answer is not None:
+        answer = str(answer)
+
+    return AgentAction(
+        action=action.strip(),
+        args=args,
+        thought=str(data.get("thought", "")).strip(),
+        answer=answer,
+        confidence=str(data.get("confidence", "medium")).strip() or "medium",
+    )
+
+
+def plan_next_action(messages: list[dict[str, Any]]) -> AgentAction:
+    raw = call_hf_chat(messages, max_tokens=700, temperature=0.0)
+    return parse_agent_action(raw)
+
+
 def answer_with_light_model(question: str, evidence: str) -> str:
     prompt = f"""
 你是一个 GAIA 问答 Agent。你会收到问题和工具已经整理好的证据。

tools/tool_specs.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+from typing import Any, Callable
+
+
+ToolContext = dict[str, Any]
+ToolArgs = dict[str, Any]
+ToolRunner = Callable[[ToolArgs, ToolContext], dict[str, Any]]
+
+
+@dataclass(frozen=True)
+class ToolSpec:
+    name: str
+    description: str
+    input_schema: dict[str, Any]
+    run: ToolRunner
+    enabled: bool = True
+    cost_level: str = "low"
+
+    def prompt_block(self) -> str:
+        enabled_text = "enabled" if self.enabled else "disabled"
+        return (
+            f"- {self.name} ({enabled_text}, cost={self.cost_level})\n"
+            f"  description: {self.description}\n"
+            f"  args_schema: {self.input_schema}"
+        )