Upload 13 files

README.md

@@ -1,13 +1,19 @@
 ---
-title: Agent
-emoji: 🔥
-colorFrom: yellow
-colorTo: red
+title: GAIA Agent (LangGraph)
+emoji: ⚡
+colorFrom: pink
+colorTo: yellow
 sdk: gradio
-sdk_version: 6.15.2
-python_version: '3.13'
+sdk_version: 5.23.1
 app_file: app.py
 pinned: false
+hf_oauth: true
+tags:
+- langgraph
+- langchain
+- agent
+- tool
+- agent-course
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,471 @@
+"""
+app.py —— 整个项目的"主程序"
+
+这个文件把所有零件组装起来，主要包含四大块：
+    1. 系统提示词 SYSTEM_PROMPT：写给大模型的"工作守则"，告诉它怎么答题、答案要什么格式。
+    2. GAIA Agent 类：真正的"答题机器人"，一个会思考的大模型 + 8 个工具（搜索、看图、读文件…）。
+        它按"思考→调用工具→再思考→…→给出答案"的循环工作。
+    3. 提交相关函数：把答案 POST 给评分服务器，并处理服务器偶尔出错时的重试。
+    4. Gradio 界面：网页上的几个按钮和表格，方便点一下就跑全流程、看结果。
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import tempfile
+
+import gradio as gr    
+import requests         
+import pandas as pd    
+
+# LangChain 里三种"消息"类型：系统消息(给AI定规则)、人类消息(用户的话)、AI消息(AI的回复)。
+from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
+
+# 导入"创建 ReAct agent"的函数。
+try:
+    from langgraph.prebuilt import create_react_agent
+except ImportError:  # newer langchain/langgraph layouts
+    from langchain.agents import create_agent as create_react_agent
+
+# 导入一个特定错误类型：agent 思考步数超上限时会抛它。库里没有就自己定义一个占位的。
+try:
+    from langgraph.errors import GraphRecursionError
+except ImportError:
+    class GraphRecursionError(Exception):
+        pass
+
+# 导入我们自己写的 8 个工具（每个都在 tools/ 文件夹里）。
+from tools.web_search import web_search
+from tools.wikipedia_search import wikipedia_search
+from tools.visit_webpage import visit_webpage
+from tools.read_file import read_file
+from tools.transcribe_audio import transcribe_audio
+from tools.visual_qa import visual_qa
+from tools.youtube_transcript import youtube_transcript
+from tools.python_repl import python_repl
+
+# 导入配置（模型地址/密钥/名字）和标准答案表/题目分类器。
+from config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL_ID
+from answer_key import REFERENCE_ANSWERS, classify_question
+
+# 课程评分服务器的网址（提供取题、下载附件、提交答案三个接口）。
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+# 系统提示词：这是发给大模型的"工作守则"，相当于给它的岗前培训。它直接决定答题质量，
+# 是反复调试出来的成果。大意是：一步步推理、必须先用工具查证再回答、
+# 绝不能没查就瞎猜或说"不知道"、最后必须用固定格式 "FINAL ANSWER: ..." 给出极简答案。
+SYSTEM_PROMPT = (
+    "You are a general AI assistant answering questions from the GAIA benchmark. "
+    "Reason step by step and use your tools to gather and verify facts. "
+    "Available tools: web_search, wikipedia_search, visit_webpage, read_file "
+    "(spreadsheets/PDF/Word/code/text), transcribe_audio, visual_qa (images), "
+    "youtube_transcript, and python_repl (run Python for any maths, string or data work).\n"
+    "Tool guidance:\n"
+    "- For ANY question needing a fact, name, number, date, list, or file content you MUST "
+    "call at least one tool (web_search / wikipedia_search / a file tool) before answering. "
+    "NEVER output 'unknown' or a guess without having searched first.\n"
+    "- Plan multi-hop lookups: search, open the most relevant result with visit_webpage, and "
+    "read it carefully to extract the EXACT value (full names, exact spelling). Cross-check "
+    "when sources conflict.\n"
+    "- A question that contains a YouTube URL: call `youtube_transcript` on that URL.\n"
+    "- A question that says a file is attached: a local path is given in the message — open it "
+    "with read_file / transcribe_audio / visual_qa. Never ask the user to upload anything.\n"
+    "- If a tool fails or returns nothing, try a different tool or a reworded query rather than "
+    "repeating the same call.\n"
+    "- Only after genuinely trying the tools may you, as a last resort, give your single best "
+    "guess. Always commit to a concrete answer — never say you cannot answer.\n\n"
+    "Finish with one line in exactly this template:\n"
+    "FINAL ANSWER: [YOUR FINAL ANSWER]\n"
+    "YOUR FINAL ANSWER must be a number OR as few words as possible OR a comma separated "
+    "list of numbers and/or strings. Do not add anything after it.\n"
+    "- For a number: no thousands separators and no units ($, %, ...) unless asked.\n"
+    "- For a string: no articles, no abbreviations, digits written in plain text unless asked.\n"
+    "- For a list: apply these rules to each element."
+)
+
+
+def build_model():
+    """创建并返回"驱动 agent 的大模型对象"。参数都来自 config.py。这个模型必须支持"调用工具"。"""
+    from langchain_openai import ChatOpenAI
+
+    return ChatOpenAI(
+        model=LLM_MODEL_ID,
+        base_url=LLM_BASE_URL,
+        api_key=LLM_API_KEY,
+        # 答题任务需要确定性，所以设为 0。
+        temperature=float(os.getenv("AGENT_TEMPERATURE", "0")),
+        # 单次回复最多生成多少字，防止回答过长。
+        max_tokens=int(os.getenv("AGENT_MAX_TOKENS", "4096")),
+    )
+
+
+def clean_answer(content) -> str:
+    """把大模型那段啰嗦的最终回复，"提纯"成评分服务器要的、干干净净的标准答案。"""
+    # 有时回复是分段的列表，这里先把它们拼成一整段文字。
+    if isinstance(content, list):
+        content = " ".join(
+            part.get("text", "") if isinstance(part, dict) else str(part) for part in content
+        )
+    text = str(content).strip()
+    # 只保留 "FINAL ANSWER:" 后面的那部分（这是我们要求模型给的最终答案标记）。
+    if "FINAL ANSWER:" in text:
+        text = text.split("FINAL ANSWER:")[-1].strip()
+    # 只取第一行，丢掉模型可能在后面多写的解释。
+    text = text.splitlines()[0].strip() if text else text
+    # 如果答案被引号包住，去掉首尾的引号。
+    if len(text) >= 2 and text[0] == text[-1] and text[0] in ("'", '"'):
+        text = text[1:-1].strip()
+    # GAIA 的标准答案末尾没有句号，所以去掉结尾的句号和空格。
+    return text.rstrip(". ").strip()
+
+
+class GAIAAgent:
+    """答题机器人：一个 LangGraph ReAct agent，外加上网、读文件、听音频、看图等全套工具。"""
+
+    def __init__(self, api_url: str = DEFAULT_API_URL):
+        self.api_url = api_url
+        tools = [
+            web_search,
+            wikipedia_search,
+            visit_webpage,
+            read_file,
+            transcribe_audio,
+            visual_qa,
+            youtube_transcript,
+            python_repl,
+        ]
+        self.model = build_model()                          
+        self.agent = create_react_agent(self.model, tools)  # 把模型和工具组装成会用工具的 agent
+        # 思考步数上限：防止 agent 陷入死循环无限调用工具。默认最多 40 步。
+        self.recursion_limit = int(os.getenv("AGENT_RECURSION_LIMIT", "40"))
+        print("GAIAAgent initialized.")
+
+    def _download_file(self, task_id: str, file_name: str) -> str | None:
+        """把某道题的附件从评分服务器下载到本地临时文件夹，返回本地路径；失败返回 None。"""
+        try:
+            response = requests.get(
+                f"{self.api_url}/files/{task_id}",
+                timeout=30,
+                headers={"User-Agent": "Mozilla/5.0"},
+            )
+            response.raise_for_status()
+        except Exception as e:
+            print(f"Could not download file for task {task_id}: {e}")
+            return None
+        # tempfile.mkdtemp() 新建一个临时文件夹，把下载内容写进去。
+        path = os.path.join(tempfile.mkdtemp(), file_name or f"{task_id}.dat")
+        with open(path, "wb") as f:   # "wb" = 以二进制写入（附件可能是图片/音频等非文本）
+            f.write(response.content)
+        return path
+
+    @staticmethod
+    def _collect_tools(history: list) -> list:
+        """统计这次答题中 agent 实际用过哪些工具（按首次使用的先后顺序），用于结果表格展示。"""
+        used = []
+        for m in history:                                   # 遍历对话历史里的每条消息
+            for tc in (getattr(m, "tool_calls", None) or []):  # 看这条消息有没有"调用工具"的记录
+                name = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                if name and name not in used:               # 没记过的工具名才加进去（去重）
+                    used.append(name)
+        return used
+
+    @staticmethod
+    def _file_hint(path: str) -> str:
+        """根据附件后缀名，明确告诉模型"这个文件该用哪个工具"。
+        （因为模型有时会选错工具——比如对着 mp3 录音却用 read_file 去读，所以这里给个明确提示。）"""
+        ext = os.path.splitext(path)[1].lower()
+        if ext in (".mp3", ".wav", ".m4a", ".flac", ".ogg", ".aac"):
+            return "It is an AUDIO file: call `transcribe_audio` on this path, then answer from the transcript."
+        if ext in (".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"):
+            return "It is an IMAGE: call `visual_qa` on this path with a precise question to read it."
+        return "Call `read_file` on this path to read its contents, then answer."
+
+    @staticmethod
+    def _is_giveup(text: str) -> bool:
+        """判断模型这次是不是"摆烂了"——给了空答案，或说"不知道/做不到"之类的放弃性回答。"""
+        low = str(text).strip().lower()
+        return (
+            not low
+            or low in ("unknown", "nan", "none")
+            or "unable to" in low
+            or "cannot" in low
+            or "i don't" in low
+        )
+
+    @staticmethod
+    def _looks_incomplete(text: str) -> bool:
+        """判断答案是不是"没答完/答歪了"——比如出现"需要更多步骤""我看不到图""请上传文件"等字样。"""
+        low = str(text).strip().lower()
+        if not low:
+            return True
+        markers = (
+            "need more steps",
+            "i cannot see",
+            "i can't see",
+            "please upload",
+            "i don't see any",
+            "i do not see any",
+            "unable to access",
+            "as an ai",
+        )
+        return any(m in low for m in markers)
+
+    def _force_final(self, history: list) -> str:
+        """最后的兜底手段：不给任何工具，逼模型"就用目前已经查到的信息，立刻给一个确定答案"，
+        这样它就没法再用"需要更多步骤"来拖延了。"""
+        # 先把历史里那些"需要更多步骤"的废话清掉，免得干扰。
+        clean_history = [
+            m for m in history
+            if not (isinstance(m, AIMessage) and "need more steps" in str(m.content).lower())
+        ]
+        # 追加一句强硬要求：必须现在就给出 FINAL ANSWER，实在不行就猜，但不准说答不了。
+        clean_history.append(
+            HumanMessage(
+                content=(
+                    "You are out of tool budget. Using only the information already gathered "
+                    "above, give your single best answer now. You MUST output exactly one line "
+                    "'FINAL ANSWER: [answer]' with a concrete value — guess if you must, and "
+                    "never say you cannot answer."
+                )
+            )
+        )
+        # 这里直接调模型（不带工具），拿到它的最终回复。
+        return self.model.invoke(clean_history).content
+
+    def _run(self, messages: list, task_id):
+        """跑一次完整的答题流程，返回 (答案, 用过的工具列表)。"""
+        try:
+            # 让 agent 开跑：它会自己循环"思考→用工具→再思考"，直到给出答案或达到步数上限。
+            result = self.agent.invoke(
+                {"messages": messages}, config={"recursion_limit": self.recursion_limit}
+            )
+            history = result["messages"]                 # 全过程的对话记录
+            tools_used = self._collect_tools(history)    # 统计用过哪些工具
+            answer = history[-1].content                 # 最后一条消息就是最终答案
+            # 如果答案看起来"没答完/答歪"，就用兜底手段再逼它给个确定答案。
+            if self._looks_incomplete(answer):
+                answer = self._force_final(history)
+            return clean_answer(answer), tools_used
+        except GraphRecursionError:
+            # 思考步数超了上限：也走兜底，硬要一个答案。
+            try:
+                return clean_answer(self._force_final(messages)), self._collect_tools(messages)
+            except Exception as e:
+                print(f"Forced-answer failed on task {task_id}: {e}")
+                return "unknown", []
+        except Exception as e:
+            # 其它任何意外错误：返回 "unknown"，保证整批题不会因一道题崩掉而中断。
+            print(f"Agent error on task {task_id}: {e}")
+            return "unknown", []
+
+    def __call__(self, question: str, task_id: str | None = None, file_name: str | None = None):
+        """让这个机器人对象能像函数一样被"调用"来答一道题。返回 (答案, 用过的工具)。"""
+        user_content = question
+        # 如果这道题带附件：先下载，再把本地路径和"该用哪个工具"的提示一起拼进给模型的消息里。
+        if file_name:
+            path = self._download_file(task_id, file_name)
+            if path:
+                user_content += f"\n\nA file is attached at local path: {path}\n{self._file_hint(path)}"
+            else:
+                # 下载失败就告诉模型"附件下不下来，尽量只凭题目文字作答"。
+                user_content += (
+                    "\n\n(Note: the attached file could not be downloaded. Answer as best you "
+                    "can from the question text alone.)"
+                )
+        # 组装成两条消息：系统守则 + 用户的问题，交给 _run 去跑。
+        messages = [SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=user_content)]
+        answer, tools_used = self._run(messages, task_id)
+
+        # 补救机制：如果模型"一个工具都没用"就摆烂了，强制它再答一次，并要求这次必须先搜索。
+        if not tools_used and self._is_giveup(answer):
+            retry = messages + [
+                HumanMessage(
+                    content=(
+                        "You answered without using any tool, which is not allowed. Call "
+                        "web_search or wikipedia_search now, read the results, and then give "
+                        "the FINAL ANSWER."
+                    )
+                )
+            ]
+            answer2, tools2 = self._run(retry, task_id)
+            # 只有当重试确实用了工具、或给出了非放弃的答案时，才采用重试结果。
+            if tools2 or not self._is_giveup(answer2):
+                answer, tools_used = answer2, tools2
+        return answer, tools_used
+
+
+# 一个"缓存"：存住最近一次算好的答案。这样如果提交失败（评分服务器偶尔返回 500 错误），
+# 可以直接重新提交缓存里的答案，而不必让又慢又花钱的 agent 重跑一遍。
+LAST_SUBMISSION: dict = {}
+
+
+def _submit_with_retry(payload: dict, retries: int = 4):
+    """把答案 POST 提交到 /submit 接口。遇到 5xx 服务器错误或网络问题会自动重试；
+    遇到 4xx（我们这边请求有问题）则把具体原因返回，方便人去修。"""
+    submit_url = f"{DEFAULT_API_URL}/submit"
+    last = None
+    for attempt in range(retries):
+        try:
+            resp = requests.post(submit_url, json=payload, timeout=120)
+            if resp.status_code >= 500:   # 5xx = 服务器自己出毛病了，值得重试
+                last = f"{resp.status_code} server error: {resp.text[:200]}"
+                print(f"submit attempt {attempt + 1}: {last}")
+                time.sleep(4 * (attempt + 1))   # 等一会儿再试，等待时间逐次拉长
+                continue
+            resp.raise_for_status()
+            return True, resp.json()   # 成功：返回 (True, 服务器给的结果)
+        except requests.exceptions.HTTPError as e:
+            # 4xx 错误：是我们的请求有问题，重试也没用，直接把原因返回。
+            detail = e.response.text[:300]
+            try:
+                detail = e.response.json().get("detail", detail)
+            except Exception:
+                pass
+            return False, f"HTTP {e.response.status_code}: {detail}"
+        except Exception as e:
+            # 网络异常等：记下错误，等一会儿继续重试。
+            last = str(e)
+            print(f"submit attempt {attempt + 1} failed: {last}")
+            time.sleep(4 * (attempt + 1))
+    return False, f"all {retries} attempts failed (last error: {last})"
+
+
+def _format_result(result_data: dict) -> str:
+    """把评分服务器返回的结果，整理成一段人类易读的文字（用户名、总分、对了几道、附言）。"""
+    return (
+        f"Submission Successful!\n"
+        f"User: {result_data.get('username')}\n"
+        f"Overall Score: {result_data.get('score', 'N/A')}% "
+        f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+        f"Message: {result_data.get('message', 'No message received.')}"
+    )
+
+
+def run_and_submit_all(profile: gr.OAuthProfile | None):
+    """【一键全流程】取回所有题目 → 逐题让 agent 作答 → 缓存答案 → 提交。这是主按钮触发的函数。"""
+    space_id = os.getenv("SPACE_ID")   # 当前 Hugging Face Space 的标识
+
+    # 第一步：必须先登录 Hugging Face（评分要记到你账号名下）。
+    if profile:
+        username = f"{profile.username}"
+        print(f"User logged in: {username}")
+    else:
+        return "Please Login to Hugging Face with the button.", None
+
+    # 第二步：检查 SPACE_ID。评分服务器会校验提交者的 Space 代码链接，缺了它常导致 500 错误。
+    if not space_id:
+        return (
+            "SPACE_ID not found. Run this on your public Hugging Face Space — the scoring "
+            "server validates the agent_code link and a missing/invalid Space can cause a 500.",
+            None,
+        )
+    # 拼出本项目代码的公开地址，提交时要一并交上去（证明这答案是这套代码产出的）。
+    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
+    print(agent_code)
+
+    # 第三步：从服务器取回全部题目。
+    try:
+        response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15)
+        response.raise_for_status()
+        questions_data = response.json()
+        if not questions_data:
+            return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+    except Exception as e:
+        return f"Error fetching questions: {e}", None
+
+    # 第四步：创建答题机器人。
+    try:
+        agent = GAIAAgent(api_url=DEFAULT_API_URL)
+    except Exception as e:
+        return f"Error initializing agent: {e}", None
+
+    # 第五步：逐题作答。results_log 存给人看的结果表格；answers_payload 存要提交给服务器的答案。
+    results_log = []
+    answers_payload = []
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        file_name = item.get("file_name") or ""
+        if not task_id or question_text is None:   # 题目数据不完整就跳过
+            continue
+        answer, tools_used = agent(question_text, task_id=task_id, file_name=file_name)
+        answer = (str(answer).strip() or "unknown")   # 答案不能为空（空答案会让服务器 500）
+        answers_payload.append({"task_id": task_id, "submitted_answer": answer})
+        results_log.append(
+            {
+                "Task ID": task_id,
+                "Type": classify_question(question_text, file_name),   # 题型标签
+                "Question": question_text,
+                "Reference Answer": REFERENCE_ANSWERS.get(task_id, ""),  # 标准答案（仅供对照）
+                "Submitted Answer": answer,                              # 我们提交的答案
+                "Tools Used": ", ".join(tools_used) if tools_used else "(none)",
+            }
+        )
+
+    if not answers_payload:
+        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+
+    # 第六步：把答案打包，存进缓存，然后提交。
+    payload = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
+    LAST_SUBMISSION.update(payload=payload, results_log=results_log)
+    print(f"Submitting {len(answers_payload)} answers for user '{username}'...")
+
+    df = pd.DataFrame(results_log)   # 把结果整理成表格，显示在网页上
+    ok, data = _submit_with_retry(payload)
+    if ok:
+        return _format_result(data), df
+    # 提交失败时：答案已缓存，提示用户修好问题（最常见是把 Space 设为公开）后点"重新提交"即可。
+    return (
+        f"Submission Failed: {data}\n\n"
+        "Your answers are cached — fix the issue (most often: make the Space Public) and click "
+        "'Re-submit last answers' to retry WITHOUT re-running the agent.",
+        df,
+    )
+
+
+def submit_only(profile: gr.OAuthProfile | None):
+    """【重新提交】只把缓存里的答案再交一次，不重新跑 agent（省时省钱）。"""
+    if not LAST_SUBMISSION.get("payload"):
+        return "No cached answers yet — run the evaluation first.", None
+    df = pd.DataFrame(LAST_SUBMISSION.get("results_log", []))
+    ok, data = _submit_with_retry(LAST_SUBMISSION["payload"])
+    if ok:
+        return _format_result(data), df
+    return f"Submission Failed again: {data}", df
+
+
+# --- Gradio 网页界面 ---
+# 下面用 Gradio 搭一个简单网页：一段说明 + 登录按钮 + 两个操作按钮 + 状态框 + 结果表格。
+with gr.Blocks() as demo:
+    gr.Markdown("# GAIA Agent Evaluation Runner (LangGraph)")
+    gr.Markdown(
+        """
+        1. Log in to your Hugging Face account with the button below.
+        2. Click 'Run Evaluation & Submit All Answers' to fetch the questions, run the agent
+           and submit the answers. This can take several minutes.
+        3. If submission fails (the scoring server sometimes returns 500), click
+           'Re-submit last answers' to retry without re-running the agent.
+
+        The model endpoint is preconfigured in `config.py`, so no secrets are required.
+        Make sure this Space is **Public**, otherwise the scoring server can reject the
+        submission with a 500.
+        """
+    )
+
+    gr.LoginButton()   # Hugging Face 登录按钮
+    with gr.Row():     # 把两个按钮排在同一行
+        run_button = gr.Button("Run Evaluation & Submit All Answers", variant="primary")
+        resubmit_button = gr.Button("Re-submit last answers")
+    status_output = gr.Textbox(label="Run Status / Submission Result", lines=6, interactive=False)
+    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
+
+    # 把"按钮点击"和"要运行的函数"绑定起来：点主按钮跑全流程，点另一个只重新提交。
+    run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
+    resubmit_button.click(fn=submit_only, outputs=[status_output, results_table])
+
+
+# 只有"直接运行这个文件"时才启动网页（被别处导入时不会启动）。
+if __name__ == "__main__":
+    demo.launch(debug=True, share=False)

config.py

@@ -0,0 +1,10 @@
+import os
+
+LLM_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.agicto.cn/v1")
+LLM_API_KEY = os.getenv("OPENAI_API_KEY", "sk-8B2kHRZwRdwnMAtEKKZpDUHsS5tPK31Ibq5shXbGLkolzsih")
+
+LLM_MODEL_ID = os.getenv("MODEL_ID", "gpt-5.4")        
+VLM_MODEL_ID = os.getenv("VLM_MODEL_ID", "gpt-4o")     
+ASR_MODEL_ID = os.getenv("ASR_MODEL_ID", "whisper-1")   
+
+TAVILY_API_KEY = os.getenv("TAVILY_API_KEY", "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")

requirements.txt

@@ -0,0 +1,14 @@
+langgraph
+langchain
+langchain-core
+langchain-openai
+openai
+ddgs
+requests
+markdownify
+pandas
+openpyxl
+pypdf
+python-docx
+youtube-transcript-api
+gradio[oauth]

tools/python_repl.py

@@ -0,0 +1,33 @@
+"""
+tools/python_repl.py —— 工具⑧：运行 Python 代码（计算器/小程序）
+
+大模型自己算数、处理表格、倒写字符串时容易出错。这个工具给它一个"草稿纸"：
+它可以写一段 Python 代码交给本工具真正运行，再把运行结果拿回去。
+适合：算术、字符串处理（如把句子倒过来）、解析表格、集合/列表运算、日期计算等。
+"""
+
+import io
+import contextlib
+
+from langchain_core.tools import tool
+
+
+@tool
+def python_repl(code: str) -> str:
+    """Execute Python code and return everything it prints. Use this for any computation:
+    arithmetic, string manipulation (e.g. reversing text), parsing tables/CSV, set and list
+    operations, date math, etc. You MUST `print(...)` the values you want to see. You may
+    import standard libraries plus pandas and numpy."""
+    buffer = io.StringIO()   # 一个"内存里的纸"，用来接住代码 print 出来的所有文字
+    namespace: dict = {}     # 代码运行时用的独立变量空间，避免污染本程序自身的变量
+    try:
+        # redirect_stdout：把代码里 print 的内容，从"打印到屏幕"改成"写进上面的 buffer"。
+        # exec：真正执行那段代码字符串。
+        with contextlib.redirect_stdout(buffer):
+            exec(code, namespace)
+    except Exception as e:
+        # 代码出错时，连同"出错前已经打印的内容"一起返回，方便大模型排查问题。
+        return f"Error: {e}\nOutput before error:\n{buffer.getvalue()}"
+    output = buffer.getvalue()   # 取出代码打印的全部内容
+    # 如果代码跑成功但什么都没打印，就提醒大模型"记得用 print 输出结果"。
+    return output if output.strip() else "Code ran successfully but printed nothing. Remember to print() your result."

tools/read_file.py

@@ -0,0 +1,69 @@
+"""
+tools/read_file.py —— 工具④：读取本地文件（万能读取器）
+
+有些题目会附带一个文件（Excel 表格、PDF、Word 文档、Python 代码、纯文本等）。
+这个工具负责把这些文件的内容读成文字交给大模型。它会先看文件后缀名，再决定用什么方式读：
+不同格式的文件读法不一样，所以下面用一连串 if 分别处理。
+
+注意：图片要用 visual_qa（看图）、音频要用 transcribe_audio（转写），它们不归这个工具管。
+"""
+
+import os
+
+from langchain_core.tools import tool
+
+
+@tool
+def read_file(file_path: str) -> str:
+    """Read a local file and return its content as text. Handles spreadsheets
+    (.xlsx/.xls/.csv/.tsv), PDFs (.pdf), Word documents (.docx) and any plain-text or code
+    file (.txt/.py/.json/.md/...). For images use `visual_qa`; for audio use
+    `transcribe_audio`. Returns the full text so you can reason over it or parse it with
+    `python_repl`."""
+    # 先确认文件真的存在，不存在就直接返回提示（避免后面读取时报错崩溃）。
+    if not os.path.exists(file_path):
+        return f"File not found: {file_path}"
+    ext = os.path.splitext(file_path)[1].lower()   # 取出后缀名（如 ".xlsx"），转小写
+    try:
+        # —— Excel 表格 ——
+        if ext in (".xlsx", ".xls"):
+            import pandas as pd   # pandas 是处理表格数据的常用库
+
+            # sheet_name=None 表示"读取工作簿里的所有工作表"，结果是 {表名: 表格数据} 的字典。
+            sheets = pd.read_excel(file_path, sheet_name=None)
+            # 把每张表都转成 CSV 文字（逗号分隔），拼起来一起返回。
+            return "\n\n".join(
+                f"## Sheet: {name}\n{df.to_csv(index=False)}" for name, df in sheets.items()
+            )
+
+        # —— CSV / TSV 文本表格 ——
+        if ext in (".csv", ".tsv"):
+            import pandas as pd
+
+            # CSV 用逗号分隔，TSV 用制表符(Tab)分隔，这里据后缀选对分隔符。
+            sep = "\t" if ext == ".tsv" else ","
+            return pd.read_csv(file_path, sep=sep).to_csv(index=False)
+
+        # —— PDF 文档 ——
+        if ext == ".pdf":
+            from pypdf import PdfReader
+
+            reader = PdfReader(file_path)
+            # 逐页抽取文字再用换行拼起来（有的页面抽不出文字就当空字符串处理）。
+            return "\n".join((page.extract_text() or "") for page in reader.pages)
+
+        # —— Word 文档 ——
+        if ext == ".docx":
+            import docx
+
+            document = docx.Document(file_path)
+            # 逐段落取文字再拼起来。
+            return "\n".join(p.text for p in document.paragraphs)
+
+        # —— 其它情况：当作普通纯文本/代码文件，直接按文本读 ——
+        # errors="replace" 表示遇到无法识别的字符时用占位符替代，而不是报错。
+        with open(file_path, "r", encoding="utf-8", errors="replace") as f:
+            return f.read()
+    except Exception as e:
+        # 任何读取错误都转成一句说明返回，保证程序不崩。
+        return f"Error reading file '{file_path}': {e}"

tools/transcribe_audio.py

@@ -0,0 +1,36 @@
+"""
+tools/transcribe_audio.py —— 工具⑤：把录音转成文字（语音识别）
+
+有的题目附带一段录音（比如"语音备忘录""课堂录音"），问的内容藏在话里。
+这个工具把音频文件交给专门的语音识别模型（whisper-1），让它"听写"成文字，
+然后大模型就能从文字里找答案了。
+"""
+
+import os
+
+from langchain_core.tools import tool
+
+# 复用 config.py 里配置好的服务地址、密钥，以及语音识别专用模型的名字。
+from config import LLM_BASE_URL, LLM_API_KEY, ASR_MODEL_ID
+
+
+@tool
+def transcribe_audio(file_path: str) -> str:
+    """Transcribe a local audio file (.mp3/.wav/.m4a/...) to text. Use it whenever a
+    question references a voice memo or recording. The transcript is returned as plain
+    text."""
+    # 先确认音频文件存在。
+    if not os.path.exists(file_path):
+        return f"File not found: {file_path}"
+    try:
+        from openai import OpenAI
+
+        # 创建一个连接到我们模型服务的客户端。
+        client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
+        # 以"二进制"方式打开音频文件（"rb" = read binary），把它上传给语音识别模型。
+        with open(file_path, "rb") as f:
+            result = client.audio.transcriptions.create(model=ASR_MODEL_ID, file=f)
+        # 取出识别出的文字。getattr(...) 是稳妥写法：能取到 text 就用 text，取不到就退而求其次。
+        return getattr(result, "text", None) or str(result)
+    except Exception as e:
+        return f"Error transcribing audio '{file_path}': {e}"

tools/visit_webpage.py

@@ -0,0 +1,36 @@
+"""
+tools/visit_webpage.py —— 工具③：打开并阅读一个网页
+
+上面的搜索工具只能给出"标题 + 简短摘要"，信息量不够。这个工具负责"点进去看全文"：
+给它一个网址，它把那个网页的完整内容抓下来，转成干净的纯文字交给大模型阅读。
+
+典型配合：先用 web_search 搜到一批结果 → 挑最相关的网址 → 用本工具打开它读详情。
+"""
+
+import re
+
+from langchain_core.tools import tool
+
+
+@tool
+def visit_webpage(url: str) -> str:
+    """Fetch a web page and return its content as markdown text. Use this to read a page
+    found via `web_search` or a url given in the question."""
+    import requests
+    from markdownify import markdownify   # 把网页的 HTML 代码转成清爽的 Markdown 文字
+
+    # 访问网页。User-Agent 是伪装成普通浏览器的标识，否则有些网站会拒绝程序访问。
+    try:
+        response = requests.get(url, timeout=25, headers={"User-Agent": "Mozilla/5.0"})
+        response.raise_for_status()   # 访问失败（如 404）就报错
+    except Exception as e:
+        return f"Error fetching the webpage: {e}"   # 出错时返回错误说明，而不是让程序崩溃
+
+    # 把网页源代码转成纯文字，并去掉首尾空白。
+    content = markdownify(response.text).strip()
+    # 把连续 3 个以上的换行压缩成 2 个，让排版更紧凑、好读。
+    content = re.sub(r"\n{3,}", "\n\n", content)
+    # 同样地，网页太长就截断到 4 万字，避免内容过多。
+    if len(content) > 40000:
+        content = content[:40000] + "\n...[truncated]"
+    return content

tools/visual_qa.py

@@ -0,0 +1,55 @@
+"""
+tools/visual_qa.py —— 工具⑥：看图回答问题
+
+有的题目附带一张图片（比如国际象棋棋盘截图），需要"看懂图"才能答。
+这个工具把图片交给会"看图"的模型（gpt-4o），并附上一个具体问题，让它看图作答。
+
+关键技巧：图片不能直接当文字发送，要先把它编码成一长串文本（Base64 编码），
+再拼成一种叫 data URI 的特殊格式，模型才能"收到"这张图。下面的代码就是在做这件事。
+"""
+
+import os
+import base64      # 用于把图片转成可传输的文本编码
+import mimetypes   # 用于猜测图片的具体类型（png/jpeg...）
+
+from langchain_core.tools import tool
+
+from config import LLM_BASE_URL, LLM_API_KEY, VLM_MODEL_ID
+
+
+@tool
+def visual_qa(file_path: str, question: str) -> str:
+    """Answer a question about a local image file using a vision-language model. Pass the
+    image path and a precise question, e.g. 'What chess move should black play? Answer in
+    algebraic notation.' or 'Transcribe all text in this image.'"""
+    if not os.path.exists(file_path):
+        return f"File not found: {file_path}"
+    try:
+        from openai import OpenAI
+
+        # 猜测图片类型（如 image/png）；猜不出就默认按 png 处理。
+        mime = mimetypes.guess_type(file_path)[0] or "image/png"
+        # 以二进制读入图片，编码成 Base64 文本，再拼成 data URI（模型能识别的"图片文本"格式）。
+        with open(file_path, "rb") as f:
+            b64 = base64.b64encode(f.read()).decode("utf-8")
+        data_uri = f"data:{mime};base64,{b64}"
+
+        client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
+        # 一条消息里同时塞进两样东西：要问的问题(text) + 那张图(image_url)，一起发给看图模型。
+        response = client.chat.completions.create(
+            model=VLM_MODEL_ID,
+            max_tokens=1024,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": question},
+                        {"type": "image_url", "image_url": {"url": data_uri}},
+                    ],
+                }
+            ],
+        )
+        # 取出模型看图后给出的回答文字。
+        return response.choices[0].message.content
+    except Exception as e:
+        return f"Error analysing image '{file_path}': {e}"

tools/web_search.py

@@ -0,0 +1,84 @@
+"""
+tools/web_search.py —— 工具①：联网搜索
+
+给 agent 一个"上网搜东西"的能力。
+"""
+
+import time
+
+from langchain_core.tools import tool
+
+from config import TAVILY_API_KEY
+
+
+def _tavily(query: str):
+    """用 Tavily 搜索。函数名以下划线开头，表示这是"内部辅助函数"，只给本文件自己调用。"""
+    # 没有配密钥就直接返回 None（表示"用不了 Tavily"），让外层去用备胎 DuckDuckGo。
+    if not TAVILY_API_KEY:
+        return None
+    import requests
+
+    # 向 Tavily 的接口发一个请求，带上：密钥、搜索词、最多要 8 条结果、并让它顺便给个总结性答案。
+    r = requests.post(
+        "https://api.tavily.com/search",
+        json={"api_key": TAVILY_API_KEY, "query": query, "max_results": 8, "include_answer": True},
+        timeout=30,   # 最多等 30 秒，避免卡死
+    )
+    r.raise_for_status()   # 如果服务器返回错误状态码，这里会主动报错
+    data = r.json()        # 把返回的 JSON 文本解析成 Python 能用的数据
+    # 下面把搜索结果拼成一段整齐的文字返回（标题、网址、摘要），方便大模型阅读。
+    parts = []
+    if data.get("answer"):
+        parts.append("Answer: " + data["answer"])
+    for it in data.get("results", []):
+        parts.append(f"[{it.get('title')}]({it.get('url')})\n{it.get('content', '')}")
+    return "## Search Results\n\n" + "\n\n".join(parts) if parts else None
+
+
+def _duckduckgo(query: str):
+    """备用搜索引擎：DuckDuckGo。同样是内部辅助函数。"""
+    # 不同版本的库名字不一样，try/except 是为了"哪个能导入就用哪个"，增强兼容性。
+    try:
+        from ddgs import DDGS
+    except ImportError:
+        from duckduckgo_search import DDGS
+
+    results = DDGS().text(query, max_results=10)   # 搜索，最多取 10 条
+    if not results:
+        return None
+    # 同样把结果拼成整齐文字返回。
+    return "## Search Results\n\n" + "\n\n".join(
+        f"[{r.get('title')}]({r.get('href') or r.get('url')})\n{r.get('body') or r.get('content', '')}"
+        for r in results
+    )
+
+
+@tool
+def web_search(query: str) -> str:
+    """Search the web and return the top results (title, url, snippet). Use for general,
+    up-to-date research; follow up with `visit_webpage` to read a result in full. For
+    encyclopedic facts prefer `wikipedia_search`, which is more reliable."""
+    # 第一步：先试 Tavily。成功拿到结果就直接返回。
+    try:
+        out = _tavily(query)
+        if out:
+            return out
+    except Exception:
+        pass   # Tavily 出错也不报错中断，继续往下用
+
+    # 第二步：用 DuckDuckGo，并且最多重试 4 次（免费引擎容易临时失败，多试几次更稳）。
+    last_err = None
+    for attempt in range(4):
+        try:
+            out = _duckduckgo(query)
+            if out:
+                return out
+        except Exception as e:
+            last_err = e
+        # 每次失败后等一会儿再试，且等待时间越来越长（1.5秒、3秒、4.5秒...），避免被对方当成攻击而封锁。
+        time.sleep(1.5 * (attempt + 1))
+    # 两套引擎都没搜到，就返回一句提示，告诉大模型"换个工具或换个说法再试"。
+    return (
+        f"Web search returned nothing (last error: {last_err}). "
+        "Try `wikipedia_search`, rephrase the query, or `visit_webpage` on a known url."
+    )

tools/wikipedia_search.py

@@ -0,0 +1,69 @@
+"""
+tools/wikipedia_search.py —— 工具②：维基百科检索
+
+专门去英文维基百科搜词条，并把整篇文章的正文文字取回来。
+为什么单独做一个维基工具，而不是都用上面的网页搜索？因为很多 GAIA 题问的是"百科知识"
+（人物、地点、历史等），维基百科上的答案比普通网页搜索更权威、更稳定。
+
+整个过程分两步（这是维基百科官方接口的标准用法）：
+  第一步：用关键词搜，找到最匹配的那篇文章【标题】；
+  第二步：再用这个标题，去把那篇文章的【正文】抓下来。
+"""
+
+from langchain_core.tools import tool
+
+# 维基百科官方的数据接口地址，以及一个"自报家门"的标识（礼貌地告诉对方我们是谁）。
+_API = "https://en.wikipedia.org/w/api.php"
+_HEADERS = {"User-Agent": "langgraph-gaia-agent/1.0"}
+
+
+@tool
+def wikipedia_search(query: str) -> str:
+    """Search Wikipedia and return the best matching article as plain text (title, url and
+    full text extract). More reliable than web_search for encyclopedic facts. If you need a
+    table that is missing from the extract, open the returned url with `visit_webpage`."""
+    import requests
+
+    # === 第一步：搜索，拿到最匹配文章的标题 ===
+    search = requests.get(
+        _API,
+        params={
+            "action": "query",
+            "list": "search",
+            "srsearch": query,   # 要搜的关键词
+            "srlimit": 1,        # 只要最匹配的 1 篇
+            "format": "json",
+        },
+        headers=_HEADERS,
+        timeout=20,
+    ).json()
+    hits = search.get("query", {}).get("search", [])
+    if not hits:   # 一篇都没搜到
+        return f"No Wikipedia article found for '{query}'."
+    title = hits[0]["title"]   # 取第一篇（最匹配的）的标题
+
+    # === 第二步：用标题把这篇文章的正文取回来 ===
+    page = requests.get(
+        _API,
+        params={
+            "action": "query",
+            "prop": "extracts",     # 要"正文摘录"
+            "explaintext": 1,       # 要纯文字（去掉网页里的格式标签）
+            "redirects": 1,         # 自动跟随"重定向"（比如搜"美国"自动跳到"美利坚合众国"）
+            "titles": title,
+            "format": "json",
+        },
+        headers=_HEADERS,
+        timeout=20,
+    ).json()
+    pages = page.get("query", {}).get("pages", {})
+    extract = ""
+    for p in pages.values():
+        extract = p.get("extract", "") or ""   # 取出正文文字
+    # 由文章标题拼出它的网页地址（把空格换成下划线，这是维基百科网址的规则）。
+    url = "https://en.wikipedia.org/wiki/" + title.replace(" ", "_")
+
+    # 文章太长会占用太多空间、拖慢大模型，所以超过 4 万字就截断，并标注"已截断"。
+    if len(extract) > 40000:
+        extract = extract[:40000] + "\n...[truncated]"
+    return f"# {title}\nURL: {url}\n\n{extract}"

tools/youtube_transcript.py

@@ -0,0 +1,36 @@
+"""
+tools/youtube_transcript.py —— 工具⑦：抓取 YouTube 视频字幕
+
+有的题目给一个 YouTube 视频链接，问"视频里某人说了什么/出现了什么数字"。
+这个工具不去真的"看"视频（那太慢），而是直接抓取视频自带的字幕文字，
+把整段台词拼成一段文字交给大模型，从中找答案。
+"""
+
+import re
+
+from langchain_core.tools import tool
+
+
+@tool
+def youtube_transcript(url: str) -> str:
+    """Return the spoken transcript of a YouTube video given its URL (or video id). Use it
+    to answer questions about what is said in a video."""
+    # 每个 YouTube 视频都有一个 11 位的唯一编号(video id)。下面用正则从各种格式的链接里把它揪出来：
+    # 比如 youtube.com/watch?v=XXXX、youtu.be/XXXX、/shorts/XXXX 等都能匹配。
+    match = re.search(r"(?:v=|youtu\.be/|/shorts/|/embed/)([0-9A-Za-z_-]{11})", url)
+    # 匹配到就用匹配出的编号；没匹配到就假设传进来的本身就是编号。
+    video_id = match.group(1) if match else url.strip()
+    try:
+        from youtube_transcript_api import YouTubeTranscriptApi
+
+        # 这个库新旧版本用法不同，下面两种写法是为了兼容：哪种可用就用哪种。
+        if hasattr(YouTubeTranscriptApi, "get_transcript"):
+            # 旧版用法：返回一串"片段"，每段是 {"text": 这句话, ...}，把所有句子拼起来。
+            chunks = YouTubeTranscriptApi.get_transcript(video_id)
+            return " ".join(c["text"] for c in chunks)
+        # 新版用法：写法略有不同，但同样是把每段台词拼成整段文字。
+        fetched = YouTubeTranscriptApi().fetch(video_id)
+        return " ".join(snippet.text for snippet in fetched)
+    except Exception as e:
+        # 抓不到字幕（比如视频没字幕、或服务器 IP 被限制）就返回错误说明。
+        return f"Could not fetch transcript for '{url}': {e}"