4. 多工具选择与路由

任务目标

构建一个Agent核心路由模块,使其能够根据用户自然语言请求自动选择合适的工具,并处理多工具按序调用的场景。本案例聚焦于路由决策逻辑与调用编排,不涉及具体工具的内部实现细节。

核心实现

import re
from typing import Callable, Dict, List, Tuple

# ---------- 工具定义 ----------
def weather_tool(city: str) -> str:
    """模拟天气查询工具"""
    return f"{city}天气:晴,25°C"

def calculator_tool(expression: str) -> str:
    """模拟计算器工具,仅支持四则运算"""
    # 安全校验:仅允许数字和运算符
    if not re.fullmatch(r'[\d+\-*/().\s]+', expression):
        return "错误:表达式包含非法字符"
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception:
        return "错误:表达式无法计算"

def time_tool() -> str:
    """模拟时间查询工具"""
    return "当前时间:2025-01-15 14:30:00"

# ---------- 路由与编排核心 ----------
class ToolRouter:
    def __init__(self):
        self.tools: Dict[str, Callable] = {
            "weather": weather_tool,
            "calculator": calculator_tool,
            "time": time_tool,
        }
        # 关键词 → 工具名映射(简化意图识别)
        self.keyword_map: List[Tuple[List[str], str]] = [
            (["天气", "气温", "下雨", "晴"], "weather"),
            (["计算", "等于", "多少"], "calculator"),
            (["时间", "几点", "日期"], "time"),
        ]

    def _match_tool(self, query: str) -> List[str]:
        """根据用户请求匹配需要调用的工具列表(按优先级排序)"""
        matched = []
        for keywords, tool_name in self.keyword_map:
            if any(kw in query for kw in keywords):
                if tool_name not in matched:
                    matched.append(tool_name)
        return matched

    def _extract_args(self, query: str, tool_name: str) -> tuple:
        """从请求中提取工具调用参数(简化版)"""
        if tool_name == "weather":
            # 提取城市名(简单假设:'天气'前的内容)
            match = re.search(r'([\u4e00-\u9fa5]{2,4})天气', query)
            return (match.group(1),) if match else ("北京",)
        elif tool_name == "calculator":
            # 提取包含数字和运算符的表达式
            match = re.search(r'[\d+\-*/().\s]+', query)
            return (match.group(0).strip(),) if match else ("0",)
        return ()

    def execute(self, query: str) -> List[str]:
        """路由主入口:匹配工具 → 提取参数 → 按序调用"""
        tool_names = self._match_tool(query)
        if not tool_names:
            return ["未找到匹配的工具,请尝试包含'天气'、'计算'或'时间'的请求"]

        results = []
        for tool_name in tool_names:
            tool_func = self.tools[tool_name]
            args = self._extract_args(query, tool_name)
            result = tool_func(*args)
            results.append(f"[{tool_name}] {result}")
        return results

# ---------- 最小调用入口 ----------
if __name__ == "__main__":
    router = ToolRouter()
    test_queries = [
        "北京天气怎么样?",
        "请计算 (15+3)*2 等于多少",
        "现在几点了?顺便查一下上海天气",
        "讲个笑话",
    ]
    for q in test_queries:
        print(f"用户请求:{q}")
        for r in router.execute(q):
            print(f"  → {r}")
        print()

运行与验证

运行方式:将上述代码保存为 tool_router.py,执行 python tool_router.py。

预期输出(确定性逻辑推导):

用户请求:北京天气怎么样?
  → [weather] 北京天气:晴,25°C

用户请求:请计算 (15+3)*2 等于多少
  → [calculator] (15+3)*2 = 36

用户请求:现在几点了?顺便查一下上海天气
  → [time] 当前时间:2025-01-15 14:30:00
  → [weather] 上海天气:晴,25°C

用户请求:讲个笑话
  → 未找到匹配的工具,请尝试包含'天气'、'计算'或'时间'的请求

验证判定标准:

  • 单工具请求(天气/计算/时间)能正确匹配并返回对应结果;
  • 复合请求(如“时间+天气”)按关键词映射顺序依次调用多个工具,输出顺序与匹配顺序一致;
  • 无匹配请求返回提示信息,不抛出异常;
  • 计算器工具对非法表达式(如包含字母)返回错误提示而非崩溃。

关键说明

  • 意图识别采用关键词包含匹配,实际生产环境可替换为LLM意图分类或更复杂的NLU模块,但路由编排逻辑保持不变。
  • 参数提取为最小实现:天气默认提取“XX天气”前的2-4个中文字符作为城市,计算器提取首个连续数学表达式。真实场景应接入实体识别或函数调用协议。
  • 工具注册表 self.tools 与关键词映射 self.keyword_map 分离设计,便于扩展新工具时只需增加注册项和关键词规则。