3.3 【动手】给 LLM 接入搜索 / 计算器 / 数据库工具

实验目标

本节结束后，你将能够：独立为 LLM 接入三类主流外部工具——实时搜索（Tavily）、安全代码执行沙箱（E2B）、以及 Text-to-SQL 数据库查询链路，并通过 LiteLLM 统一调用层把它们串联成一个可扩展的工具调用框架。

核心学习点： 1. 工具定义规范：如何用 JSON Schema 描述工具、让 LLM 自主决策何时调用哪个工具 2. 安全边界意识：代码执行为何必须在沙箱中进行，以及 Text-to-SQL 的 SQL 注入防御 3. 工具链可扩展性：用统一的 dispatcher 模式支持工具数量的线性扩展

---

架构总览

graph TD
    User["👤 用户输入"] --> LLM["🧠 LLM\n(通过 LiteLLM 调用)"]
    LLM -->|"tool_choice: auto"| Router["工具路由器\nDispatcher"]

    Router --> Search["🔍 搜索工具\nTavily API"]
    Router --> Code["💻 代码执行工具\nE2B 沙箱"]
    Router --> DB["🗄️ 数据库工具\nText-to-SQL"]

    Search -->|"网页摘要 + URL"| Router
    Code -->|"stdout / stderr"| Router
    DB -->|"查询结果表格"| Router

    Router -->|"ToolMessage"| LLM
    LLM -->|"最终回答"| User

    subgraph "安全边界"
        Code
        DB
    end

---

环境准备

# 创建虚拟环境（uv）
uv venv --python 3.11 && source .venv/bin/activate

# 安装依赖
pip install -r requirements.txt

依赖清单（requirements.txt）：

python-dotenv>=1.0.0
litellm>=1.40.0
tavily-python>=0.3.0
e2b-code-interpreter>=0.5.0
sqlalchemy>=2.0.0
openai>=1.0.0
pytest>=7.0.0

Colab 用户：!pip install python-dotenv litellm tavily-python e2b-code-interpreter sqlalchemy openai pytest 即可

配置 .env 文件（不要提交到 Git）：

# 模型 API Key（二选一即可，在 core_config.py 中切换 ACTIVE_MODEL_KEY）
DEEPSEEK_API_KEY=your_deepseek_api_key_here
DASHSCOPE_API_KEY=your_dashscope_api_key_here

# 工具 API Key
TAVILY_API_KEY=tvly-...
E2B_API_KEY=e2b_...

# Text-to-SQL 使用本地 SQLite，无需额外 Key

---

Step-by-Step 实现

Step 0：统一大模型配置（core_config.py）

目标：通过 MODEL_REGISTRY 集中管理模型配置，业务代码中不硬编码模型名和 API Key。

# core_config.py
"""全局配置：模型注册表与定价信息"""
import os
from typing import TypedDict


class ModelConfig(TypedDict):
    litellm_id: str          # LiteLLM 识别的模型字符串
    price_in: float          # 每 1K input tokens 价格（美元）
    price_out: float         # 每 1K output tokens 价格（美元）
    max_tokens_limit: int    # 模型支持的最大 max_tokens
    api_key_env: str | None  # API Key 环境变量名
    base_url: str | None     # API 基础 URL（None 表示使用默认）


# 注册表：key 是界面显示名，value 是调用配置
MODEL_REGISTRY: dict[str, ModelConfig] = {
    "DeepSeek-V3": {
        "litellm_id": "deepseek/deepseek-chat",
        "price_in": 0.00027,
        "price_out": 0.0011,
        "max_tokens_limit": 4096,
        "api_key_env": "DEEPSEEK_API_KEY",
        "base_url": None,
    },
    "Qwen-Max": {
        "litellm_id": "qwen/qwen-plus",
        "price_in": 0.001,
        "price_out": 0.004,
        "max_tokens_limit": 4096,
        "api_key_env": "DASHSCOPE_API_KEY",
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    },
}

# ✅ 当前激活模型 key — 修改此处全局生效
ACTIVE_MODEL_KEY: str = "DeepSeek-V3"


def get_litellm_id(model_key: str | None = None) -> str:
    key = model_key or ACTIVE_MODEL_KEY
    return MODEL_REGISTRY[key]["litellm_id"]


def get_api_key(model_key: str | None = None) -> str | None:
    key = model_key or ACTIVE_MODEL_KEY
    env_var = MODEL_REGISTRY[key]["api_key_env"]
    return os.environ.get(env_var) if env_var else None


def get_base_url(model_key: str | None = None) -> str | None:
    key = model_key or ACTIVE_MODEL_KEY
    return MODEL_REGISTRY[key]["base_url"]


def get_model_list() -> list[str]:
    return list(MODEL_REGISTRY.keys())


def estimate_cost(model_key: str, input_tokens: int, output_tokens: int) -> float:
    cfg = MODEL_REGISTRY[model_key]
    return (
        input_tokens / 1000 * cfg["price_in"]
        + output_tokens / 1000 * cfg["price_out"]
    )

替换规则：所有业务代码中的硬编码模型引用统一替换为：

原始写法	替换为
model="gpt-4o"	model=get_litellm_id()
api_key=os.environ["OPENAI_API_KEY"]	api_key=get_api_key()
api_base="https://..."	api_base=get_base_url()

---

Step 1：统一工具框架基础结构

目标：搭建一个可插拔的工具注册和调度体系，让后续每个工具只需实现一个接口即可接入。

工具系统最大的工程陷阱是"烟囱式"开发——每加一个工具就改一堆 if-else。正确做法是先定义协议，后实现具体工具。

# tools/base.py
"""工具基类与调度器定义"""

from __future__ import annotations

import json
import logging
from abc import ABC, abstractmethod
from typing import Any

logger = logging.getLogger(__name__)


class BaseTool(ABC):
    """所有工具必须继承此基类。

    设计原则：工具负责"做"，不负责"决策调用时机"——那是 LLM 的事。
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """工具名称，必须与 JSON Schema 中的 name 字段一致。"""
        ...

    @property
    @abstractmethod
    def schema(self) -> dict[str, Any]:
        """OpenAI Function Calling 格式的工具描述。

        description 字段极为关键：LLM 完全依赖它来判断是否调用该工具。
        描述要说清楚"什么时候用"，而不只是"能做什么"。
        """
        ...

    @abstractmethod
    def run(self, **kwargs: Any) -> str:
        """执行工具逻辑，返回字符串结果（LLM 只接受文本反馈）。"""
        ...


class ToolDispatcher:
    """工具调度器：管理工具注册、LLM 调用循环、工具执行。"""

    def __init__(self, tools: list[BaseTool]) -> None:
        self._tools: dict[str, BaseTool] = {t.name: t for t in tools}

    @property
    def schemas(self) -> list[dict[str, Any]]:
        """返回所有工具的 schema 列表，直接传给 OpenAI API 的 tools 参数。"""
        return [t.schema for t in self._tools.values()]

    def dispatch(self, tool_name: str, tool_args: str) -> str:
        """根据 LLM 的工具调用请求执行对应工具。

        Args:
            tool_name: LLM 选择的工具名称
            tool_args: JSON 格式的参数字符串（来自 LLM 的 function_call.arguments）

        Returns:
            工具执行结果的字符串表示
        """
        tool = self._tools.get(tool_name)
        if tool is None:
            # 返回错误信息而非抛异常——让 LLM 知道工具不存在并自行纠正
            return f"Error: tool '{tool_name}' not found. Available: {list(self._tools.keys())}"

        try:
            args = json.loads(tool_args)
            result = tool.run(**args)
            logger.info("Tool '%s' executed successfully", tool_name)
            return result
        except json.JSONDecodeError as e:
            return f"Error: invalid arguments JSON: {e}"
        except Exception as e:
            logger.exception("Tool '%s' execution failed", tool_name)
            # 生产中不要把完整 traceback 返回给 LLM（信息泄露风险）
            return f"Error executing tool: {type(e).__name__}: {str(e)[:200]}"

---

Step 2：搜索工具（Tavily）

目标：接入 Tavily 的实时搜索 API，让 LLM 能查询训练截止日期之后的信息。

为何选 Tavily 而非直接用 Google？Tavily 返回的是经过结构化处理的摘要，不是原始 HTML，LLM 能直接消费，无需额外解析层。Brave Search 是另一个好选择，更注重隐私，API 免费层额度更高。

# tools/search_tool.py
"""Tavily 实时搜索工具"""

from __future__ import annotations

import os
from typing import Any

from tavily import TavilyClient

from tools.base import BaseTool


class TavilySearchTool(BaseTool):
    """接入 Tavily API 的实时搜索工具。

    适用场景：
    - 需要最新信息（股价、新闻、近期事件）
    - 需要引用来源 URL 的场景
    - 替代 LLM 可能过时的训练知识
    """

    def __init__(
        self,
        api_key: str | None = None,
        max_results: int = 5,
        search_depth: str = "basic",  # "basic" 或 "advanced"，advanced 消耗更多 quota
    ) -> None:
        self._api_key = api_key or os.environ.get("TAVILY_API_KEY")
        if self._api_key:
            self._client = TavilyClient(api_key=self._api_key)
        else:
            self._client = None
        self._max_results = max_results
        self._search_depth = search_depth

    @property
    def name(self) -> str:
        return "web_search"

    @property
    def schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": (
                    "搜索互联网上的实时信息。当用户询问近期事件、"
                    "当前价格、最新新闻、或任何可能超出你知识截止日期的信息时使用。"
                    "不要用于可以直接回答的通用知识问题。"
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "搜索查询词，使用清晰具体的关键词，避免自然语言长句",
                        },
                        "topic": {
                            "type": "string",
                            "enum": ["general", "news", "finance"],
                            "description": "搜索类型：general=通用，news=新闻，finance=金融",
                            "default": "general",
                        },
                    },
                    "required": ["query"],
                },
            },
        }

    def run(self, query: str, topic: str = "general") -> str:
        """执行搜索并返回格式化结果。"""
        if not self._client:
            return "⚠️ 搜索功能不可用：未配置 TAVILY_API_KEY 环境变量。请在 https://tavily.com 获取 API Key 后使用。"

        response = self._client.search(
            query=query,
            topic=topic,
            max_results=self._max_results,
            search_depth=self._search_depth,
            include_answer=True,  # Tavily 会额外提供一个 AI 摘要答案
        )

        # 构建 LLM 友好的文本格式：摘要 + 各结果条目
        parts: list[str] = []

        if response.get("answer"):
            parts.append(f"【摘要】{response['answer']}\n")

        for i, result in enumerate(response.get("results", []), 1):
            parts.append(
                f"[{i}] {result['title']}\n"
                f"URL: {result['url']}\n"
                f"内容: {result['content'][:400]}...\n"  # 截断避免 context 爆炸
            )

        return "\n".join(parts) if parts else "未找到相关搜索结果"

⚠️ 生产注意：Tavily basic 模式每次搜索约消耗 1 个 credit，advanced 消耗 2 个。高频场景建议在工具层加 TTL 缓存（如 Redis），相同 query 5 分钟内命中缓存，避免重复计费。

---

Step 3：代码执行工具（E2B 沙箱）

目标：让 LLM 生成 Python 代码并在隔离沙箱中安全执行，获取真实计算结果。

为何必须用沙箱？ 直接 exec() 执行 LLM 生成的代码等于把 root 权限交给用户，任何 os.system("rm -rf /") 都会执行成功。E2B 在隔离的 Firecracker microVM 中运行代码，每次执行完毕 VM 即销毁，天然隔离。

# tools/code_tool.py
"""E2B 沙箱代码执行工具"""

from __future__ import annotations

import os
from typing import Any

from e2b_code_interpreter import Sandbox

from tools.base import BaseTool


class E2BCodeExecutorTool(BaseTool):
    """在 E2B 安全沙箱中执行 Python 代码。

    适用场景：
    - 数学计算（尤其是需要精确结果时，避免 LLM 算错）
    - 数据处理与统计分析
    - 代码验证与调试

    不适用于：需要访问本地文件系统或数据库的场景。
    """

    def __init__(self, api_key: str | None = None, timeout: int = 30) -> None:
        self._api_key = api_key or os.environ.get("E2B_API_KEY")
        self._timeout = timeout  # 单次执行超时秒数，防止死循环

    @property
    def name(self) -> str:
        return "python_executor"

    @property
    def schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": (
                    "在安全沙箱中执行 Python 代码。用于精确数学计算、"
                    "数据统计、算法验证等需要真实运行结果的任务。"
                    "代码可以使用 numpy、pandas、matplotlib 等常用库。"
                    "不能访问外部网络或本地文件。"
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "code": {
                            "type": "string",
                            "description": (
                                "要执行的 Python 代码。"
                                "用 print() 输出结果，否则无法获取返回值。"
                            ),
                        },
                        "packages": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "需要额外安装的 pip 包列表（已预装 numpy/pandas/matplotlib）",
                            "default": [],
                        },
                    },
                    "required": ["code"],
                },
            },
        }

    def run(self, code: str, packages: list[str] | None = None) -> str:
        """在 E2B 沙箱中执行代码并返回输出。

        每次调用都会创建一个新的沙箱实例，执行完毕自动销毁。
        这是 E2B 的设计哲学：无状态 = 无风险。
        """
        if not self._api_key:
            return "⚠️ 代码执行功能不可用：未配置 E2B_API_KEY 环境变量。请在 https://e2b.dev 获取 API Key 后使用。"

        # 使用 context manager 确保沙箱资源释放
        with Sandbox(api_key=self._api_key, timeout=self._timeout) as sandbox:
            # 按需安装额外依赖（冷启动约 3-5 秒，有成本意识）
            if packages:
                install_result = sandbox.commands.run(
                    f"pip install -q {' '.join(packages)}"
                )
                if install_result.exit_code != 0:
                    return f"Package installation failed: {install_result.stderr}"

            execution = sandbox.run_code(code)

            # 收集所有输出类型
            output_parts: list[str] = []

            # 标准输出（print 的内容）
            if execution.logs.stdout:
                output_parts.append("【输出】\n" + "\n".join(execution.logs.stdout))

            # 错误信息
            if execution.logs.stderr:
                output_parts.append("【错误】\n" + "\n".join(execution.logs.stderr))

            # 如果有图表，告知 LLM（图表本身也无法直接传回文本 LLM）
            if execution.results:
                for result in execution.results:
                    if hasattr(result, "png"):
                        output_parts.append("【图表】已生成图表（PNG 格式，共 1 张）")

            if not output_parts:
                return "代码执行成功，无输出（确认 print() 是否被调用）"

            return "\n".join(output_parts)

⚠️ 生产注意：E2B 沙箱冷启动约 200-500ms，热启动（复用 session）约 50ms。高频场景可考虑维护一个沙箱连接池，但要注意沙箱间的状态隔离——一个用户的执行不能影响另一个用户的环境。

Colab 替代方案：E2B 需要付费 API Key。如果只是学习验证，可以用更简单的本地执行（仅限受控环境，生产禁用）：

# ⚠️ 仅用于本地学习，生产环境绝对禁止
import subprocess
import sys

def local_execute_unsafe(code: str) -> str:
    """本地执行，无任何沙箱隔离——仅用于测试！"""
    result = subprocess.run(
        [sys.executable, "-c", code],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout or result.stderr

---

Step 4：数据库工具（Text-to-SQL）

目标：实现「自然语言 → SQL 生成 → 安全执行 → 结果解释」的完整链路。

Text-to-SQL 的核心挑战不是"让 LLM 写 SQL"——这已经很成熟——而是如何把 Schema 信息高效注入 Prompt，同时防止 SQL 注入和数据泄露。

# tools/db_tool.py
"""Text-to-SQL 数据库查询工具"""

from __future__ import annotations

import json
import re
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine

from tools.base import BaseTool


class TextToSQLTool(BaseTool):
    """将自然语言查询转换为 SQL 并安全执行。

    关键设计决策：
    1. 只读连接：使用只有 SELECT 权限的数据库用户
    2. Schema 注入：将表结构信息嵌入工具 description，让 LLM 知道"有什么表"
    3. 行数限制：强制 LIMIT，避免把整张大表返回给 LLM
    """

    MAX_ROWS = 50       # 返回给 LLM 的最大行数
    MAX_COL_LEN = 100   # 单个字段值的最大字符数（防止 blob 字段撑爆 context）

    def __init__(
        self,
        db_url: str,
        allowed_tables: list[str] | None = None,  # None 表示允许所有表
        max_rows: int = MAX_ROWS,
    ) -> None:
        # ⚠️ 生产中应使用只读账号的连接串，如 postgresql://readonly_user:pwd@host/db
        self._engine: Engine = create_engine(db_url)
        self._allowed_tables = set(allowed_tables) if allowed_tables else None
        self._max_rows = max_rows
        self._schema_description = self._build_schema_description()

    def _build_schema_description(self) -> str:
        """提取数据库 Schema 信息，生成 LLM 可理解的文本描述。

        这是 Text-to-SQL 质量的关键：Schema 信息越详细（含注释、样例值），
        LLM 生成的 SQL 越准确。
        """
        from sqlalchemy import inspect

        inspector = inspect(self._engine)
        tables = inspector.get_table_names()

        if self._allowed_tables:
            tables = [t for t in tables if t in self._allowed_tables]

        schema_parts: list[str] = []
        for table in tables:
            columns = inspector.get_columns(table)
            col_desc = ", ".join(
                f"{col['name']} ({str(col['type'])})" for col in columns
            )
            schema_parts.append(f"表名: {table}\n字段: {col_desc}")

        return "\n\n".join(schema_parts)

    @property
    def name(self) -> str:
        return "database_query"

    @property
    def schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": (
                    f"查询数据库中的结构化数据。当用户询问具体的数据统计、"
                    f"记录查找、数据分析时使用。\n\n"
                    f"数据库结构如下：\n{self._schema_description}"
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "natural_language_query": {
                            "type": "string",
                            "description": "用户的自然语言查询需求，如'查找销售额最高的前5个产品'",
                        },
                        "sql": {
                            "type": "string",
                            "description": (
                                "要执行的 SQL 查询语句。"
                                "必须是只读 SELECT 语句，禁止 INSERT/UPDATE/DELETE/DROP。"
                                f"务必加 LIMIT，最多返回 {self.MAX_ROWS} 行。"
                            ),
                        },
                    },
                    "required": ["natural_language_query", "sql"],
                },
            },
        }

    def _validate_sql(self, sql: str) -> None:
        """防御性 SQL 验证：拦截危险操作。

        注意：这只是第一道防线，根本防线是数据库用户的权限控制。
        不要依赖正则来保证安全——有经验的攻击者可以绕过。
        """
        sql_upper = sql.upper().strip()

        # 检测写操作关键字
        dangerous_keywords = [
            "INSERT", "UPDATE", "DELETE", "DROP", "CREATE",
            "ALTER", "TRUNCATE", "EXEC", "EXECUTE", "--", ";"
        ]
        for keyword in dangerous_keywords:
            if re.search(r'\b' + keyword + r'\b', sql_upper):
                raise ValueError(
                    f"禁止执行 '{keyword}' 操作，本工具仅支持 SELECT 查询"
                )

        if not sql_upper.lstrip().startswith("SELECT"):
            raise ValueError("SQL 必须以 SELECT 开头")

    def _format_results(self, rows: list[dict], columns: list[str]) -> str:
        """将查询结果格式化为 LLM 易读的文本表格。"""
        if not rows:
            return "查询成功，结果为空（0 行）"

        # 截断过长的字段值
        truncated_rows = []
        for row in rows:
            truncated_row = {}
            for k, v in row.items():
                v_str = str(v) if v is not None else "NULL"
                truncated_row[k] = v_str[:self.MAX_COL_LEN] + "..." if len(v_str) > self.MAX_COL_LEN else v_str
            truncated_rows.append(truncated_row)

        # 简单 Markdown 表格格式（LLM 更容易解析）
        header = " | ".join(columns)
        separator = " | ".join(["---"] * len(columns))
        data_rows = [" | ".join(str(row.get(col, "")) for col in columns) for row in truncated_rows]

        result = f"查询返回 {len(rows)} 行数据：\n\n"
        result += f"| {header} |\n| {separator} |\n"
        result += "\n".join(f"| {row} |" for row in data_rows)

        if len(rows) >= self._max_rows:
            result += f"\n\n结果已截断至 {self._max_rows} 行，实际可能有更多数据"

        return result

    def run(self, natural_language_query: str, sql: str) -> str:
        """验证并执行 SQL，返回格式化结果。"""
        try:
            self._validate_sql(sql)
        except ValueError as e:
            return f"SQL 验证失败：{e}"

        # 强制添加 LIMIT（即使 LLM 已经写了，再加一层保险）
        if "LIMIT" not in sql.upper():
            sql = sql.rstrip(";") + f" LIMIT {self._max_rows}"

        try:
            with self._engine.connect() as conn:
                result = conn.execute(text(sql))
                columns = list(result.keys())
                rows = [dict(zip(columns, row)) for row in result.fetchmany(self._max_rows)]
                return self._format_results(rows, columns)
        except Exception as e:
            # 将数据库错误返回给 LLM，让它自行修正 SQL
            return f"SQL 执行失败：{type(e).__name__}: {str(e)[:300]}\n原始 SQL：{sql}"

⚠️ 生产注意：_validate_sql 中的正则只是辅助手段，真正的安全边界是数据库层的权限控制。为 Text-to-SQL 专门创建一个只有 SELECT 权限的数据库账号，并且限制只能访问指定的表。

---

Step 5：组装 Agent 主循环

目标：把三个工具和 LiteLLM 统一调用层串联成完整的工具调用循环。

OpenAI 的 Function Calling 是一个多轮交互协议：LLM 返回 tool_calls → 你执行工具 → 把结果作为 tool 角色消息回传 → LLM 继续生成，直到不再需要工具为止。本项目使用 LiteLLM 作为统一调用层，兼容所有 LLM 提供商。

# agent.py
"""工具调用 Agent 主循环"""

from __future__ import annotations

import json
import os
from typing import Any

import litellm
from dotenv import load_dotenv

from core_config import get_api_key, get_base_url, get_litellm_id
from tools.base import ToolDispatcher
from tools.code_tool import E2BCodeExecutorTool
from tools.db_tool import TextToSQLTool
from tools.search_tool import TavilySearchTool

load_dotenv()


class ToolCallingAgent:
    """支持多工具调用的 LLM Agent。

    实现标准的 ReAct 循环：
    LLM 思考 → 选择工具 → 执行工具 → 观察结果 → 循环直到得出答案
    """

    MAX_TURNS = 10  # 防止死循环：最多 10 轮工具调用

    def __init__(self, dispatcher: ToolDispatcher, model: str | None = None) -> None:
        self._dispatcher = dispatcher
        self._model = model or get_litellm_id()
        self._api_key = get_api_key()
        self._base_url = get_base_url()
        self._system_prompt = (
            "你是一个有用的 AI 助手，可以使用搜索、代码执行和数据库查询工具来回答问题。\n"
            "工具使用原则：\n"
            "1. 需要最新信息时，使用搜索工具\n"
            "2. 需要精确计算时，使用代码执行工具（不要自己心算）\n"
            "3. 需要查询结构化数据时，使用数据库工具\n"
            "4. 能直接回答的问题不要无谓调用工具"
        )

    def run(self, user_input: str, verbose: bool = False) -> str:
        """执行工具调用循环，返回最终答案。

        Args:
            user_input: 用户输入
            verbose: 是否打印工具调用过程（调试用）

        Returns:
            LLM 的最终文本回答
        """
        messages: list[dict[str, str]] = [
            {"role": "system", "content": self._system_prompt},
            {"role": "user", "content": user_input},
        ]

        for turn in range(self.MAX_TURNS):
            response = litellm.completion(
                model=self._model,
                api_key=self._api_key,
                api_base=self._base_url,
                messages=messages,
                tools=self._dispatcher.schemas,
                tool_choice="auto",  # 让 LLM 自主决定是否调用工具
            )

            message = response.choices[0].message
            # 将 assistant 消息追加到历史（包含 tool_calls 信息）
            messages.append(message)  # type: ignore[arg-type]

            # 没有工具调用 → LLM 给出了最终答案
            if not message.tool_calls:
                return message.content or ""

            # 执行所有工具调用（OpenAI 支持并行工具调用）
            for tool_call in message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = tool_call.function.arguments

                if verbose:
                    print(f"\n调用工具: {tool_name}")
                    print(f"   参数: {tool_args[:200]}")

                tool_result = self._dispatcher.dispatch(tool_name, tool_args)

                if verbose:
                    print(f"   结果: {tool_result[:300]}...")

                # 工具结果必须以 tool 角色回传，且 tool_call_id 要对应
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                })

        return "达到最大工具调用轮次，无法完成任务"


def build_agent(db_url: str = "sqlite:///demo.db") -> ToolCallingAgent:
    """工厂函数：组装带三类工具的 Agent 实例。"""
    tools = [
        TavilySearchTool(max_results=3),
        E2BCodeExecutorTool(timeout=30),
        TextToSQLTool(db_url=db_url),
    ]
    dispatcher = ToolDispatcher(tools)
    return ToolCallingAgent(dispatcher=dispatcher)

---

主入口（main.py）

项目提供交互式对话入口，可直接运行：

# main.py
"""主入口 — 工具调用 Agent 演示程序"""

import sys
import os

# 确保项目根目录在 sys.path 中
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from core_config import get_model_list, ACTIVE_MODEL_KEY
from agent import build_agent


def main():
    print(f"当前使用模型: {ACTIVE_MODEL_KEY} ({get_model_list()})")
    print("=" * 60)

    # 初始化 Agent（含搜索、代码执行、数据库三类工具）
    agent = build_agent(db_url="sqlite:///demo.db")

    print("可用工具:")
    for schema in agent._dispatcher.schemas:
        fn = schema["function"]
        print(f"  - {fn['name']}: {fn['description'][:80]}...")

    print("\n" + "=" * 60)
    print("交互式对话模式（输入 'quit' 或 'exit' 退出）")
    print("=" * 60)

    while True:
        user_input = input("\n你: ").strip()
        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit"):
            print("再见!")
            break

        print("\n思考中...")
        try:
            answer = agent.run(user_input, verbose=True)
            print(f"\n回答: {answer}")
        except Exception as e:
            print(f"\n错误: {type(e).__name__}: {e}")


if __name__ == "__main__":
    main()

---

完整运行验证

先初始化一个演示用的 SQLite 数据库，然后运行端到端测试：

# demo_setup.py — 可直接运行

import os
import json
from dotenv import load_dotenv
from sqlalchemy import create_engine, text

load_dotenv()

# ── 1. 初始化演示数据库 ──────────────────────────────────────────────────────
DB_URL = "sqlite:///demo.db"
engine = create_engine(DB_URL)

with engine.connect() as conn:
    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            category TEXT,
            price REAL,
            stock INTEGER,
            sales_count INTEGER
        )
    """))
    conn.execute(text("DELETE FROM products"))  # 清空防止重复插入
    conn.executemany(
        "INSERT INTO products VALUES (?, ?, ?, ?, ?, ?)",
        [
            (1, "MacBook Pro 16", "laptop", 19999, 50, 320),
            (2, "iPhone 15 Pro", "phone", 8999, 200, 1580),
            (3, "AirPods Pro", "audio", 1799, 500, 2100),
            (4, "iPad Air", "tablet", 4799, 80, 450),
            (5, "Apple Watch S9", "wearable", 2999, 150, 880),
        ],
    )
    conn.commit()
print("✅ 演示数据库初始化完成")

# ── 2. 初始化 Agent ──────────────────────────────────────────────────────────
from agent import build_agent

agent = build_agent(db_url=DB_URL)
print("✅ Agent 初始化完成\n")

# ── 3. 端到端测试三类工具 ─────────────────────────────────────────────────────
test_cases = [
    {
        "desc": "🔍 测试搜索工具",
        "query": "2024年诺贝尔物理学奖得主是谁？他们的主要贡献是什么？"
    },
    {
        "desc": "💻 测试代码执行工具",
        "query": "计算斐波那契数列前20项的总和，并告诉我第20项是多少"
    },
    {
        "desc": "🗄️ 测试数据库工具",
        "query": "在我们的产品数据库中，哪个类别的平均价格最高？销量最好的产品是什么？"
    },
]

for case in test_cases:
    print(f"\n{'='*60}")
    print(f"{case['desc']}")
    print(f"问题: {case['query']}")
    print("-" * 40)

    answer = agent.run(case["query"], verbose=True)

    print(f"\n💬 最终回答:")
    print(answer)

print(f"\n\n✅ 所有测试通过！")

预期输出示例：

✅ 演示数据库初始化完成
✅ Agent 初始化完成

============================================================
🔍 测试搜索工具
问题: 2024年诺贝尔物理学奖得主是谁？...
----------------------------------------

调用工具: web_search
   参数: {"query": "2024诺贝尔物理学奖得主", "topic": "news"}
   结果: 【摘要】2024年诺贝尔物理学奖授予约翰·霍普菲尔德...

💬 最终回答:
2024年诺贝尔物理学奖授予了约翰·霍普菲尔德（John Hopfield）
和杰弗里·辛顿（Geoffrey Hinton），表彰他们在人工神经网络
领域的奠基性工作...

============================================================
💻 测试代码执行工具
...
调用工具: python_executor
   参数: {"code": "fib = [0, 1]\nfor i in range(18):\n    fib.append(fib[-1]+fib[-2])\nprint(f'前20项总和: {sum(fib)}')\nprint(f'第20项: {fib[19]}')"}
   结果: 【输出】
前20项总和: 10945
第20项: 4181

============================================================
🗄️ 测试数据库工具
...
调用工具: database_query
   参数: {"natural_language_query": "各类别平均价格和最高销量产品",
          "sql": "SELECT category, AVG(price) as avg_price, MAX(sales_count) as max_sales FROM products GROUP BY category ORDER BY avg_price DESC LIMIT 10"}
   结果: 查询返回 5 行数据：...

✅ 所有测试通过！

---

常见报错与解决方案

报错信息	原因	解决方案
tavily.errors.UsageLimitExceededError	Tavily 免费额度耗尽	检查 dashboard.tavily.com 的 credit 余额；临时切换 search_depth 为 "basic"
e2b.exceptions.SandboxException: timeout	代码执行超过 timeout 秒	增大 timeout 参数；检查代码是否有死循环
sqlalchemy.exc.OperationalError: no such table	数据库路径错误或表未创建	确认 db_url 路径正确；先运行 demo_setup 初始化数据库
litellm.exceptions.RateLimitError	调用频率过高	在循环中加 time.sleep(1)；检查是否在短时间内多次运行
KeyError: 'TAVILY_API_KEY'	.env 文件未加载或 Key 未配置	确认 .env 文件在工作目录；检查 load_dotenv() 是否在导入前调用
LLM 忽略工具、直接回答	system prompt 对工具使用时机描述不清	在 schema 的 description 字段中明确写"什么情况下必须使用本工具"

---

扩展练习（可选）

1. 🟡 中等：为 TextToSQLTool 增加「自我修正」能力——当 SQL 执行失败时，把错误信息回传给 LLM，让它重写 SQL，最多重试 3 次。提示：在 run() 方法里加一层 retry 循环，把 sql、error、schema 一起作为 prompt 调用 LLM 修正。

2. 🔴 困难：实现一个「工具路由优化器」——当工具数量超过 20 个时，先用一个轻量 LLM（如 gpt-4o-mini）根据用户意图选出最相关的 3-5 个工具，再把这个子集传给主 LLM。这是真实生产环境中解决「工具数量爆炸导致 context 占用过大」问题的标准方案。

---