Claude代码解析：如何根据问题智能调用不同工具与Agent的原理与实践

1次阅读

共计 2562 个字符，预计需要花费 7 分钟才能阅读完成。

在复杂业务场景下，手动管理多工具调用会面临几个主要挑战：

上下文保持困难 ：当需要跨多个工具处理一个复杂请求时，手动维护和传递上下文信息会变得非常复杂，容易导致信息丢失或混乱。
性能开销大 ：每次调用工具都需要手动判断选择哪个工具，这种决策过程会带来额外的性能开销，特别是在高并发场景下。
错误处理复杂 ：不同工具可能有不同的错误类型和返回格式，手动处理这些差异会使代码变得冗长且难以维护。

以下是三种常见路由方案的对比：

方案类型	QPS	准确率	维护成本	适用场景
规则引擎	高	中等	中	规则明确且变化少的场景
机器学习模型	中	高	高	需要高准确率的复杂场景
向量检索	高	中高	低	工具数量多且变化频繁的场景

Claude 的请求分流采用双层过滤机制：

意图识别层 ：分析用户请求的意图，确定大致的工具类别。
工具能力匹配层 ：在确定的类别中，找到最能满足用户具体需求的工具。

工具的描述采用标准化的 JSON Schema 格式，例如：

{
  "name": "weather_checker",
  "description": "获取指定城市的天气信息",
  "input_schema": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称"
      }
    },
    "required": ["city"]
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "temperature": {
        "type": "number",
        "description": "当前温度"
      },
      "condition": {
        "type": "string",
        "description": "天气状况"
      }
    }
  }
}

以下是 Python 实现的工具路由核心类：

from typing import Dict, List, Optional
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer


class ToolSelector:
    def __init__(self, tools: List[Dict]):
        self.tools = tools
        self.vectorizer = TfidfVectorizer()
        self._fit_vectorizer()

    def _fit_vectorizer(self):
        descriptions = [tool['description'] for tool in self.tools]
        self.vectorizer.fit(descriptions)

    async def select_tool(self, query: str) -> Optional[Dict]:
        query_vec = self.vectorizer.transform([query])

        max_sim = -1
        selected_tool = None

        for tool in self.tools:
            tool_vec = self.vectorizer.transform([tool['description']])
            sim = self._cosine_similarity(query_vec, tool_vec)
            if sim > max_sim:
                max_sim = sim
                selected_tool = tool

        return selected_tool if max_sim > 0.3 else None

    @staticmethod
    def _cosine_similarity(vec1, vec2) -> float:
        return (vec1 * vec2.T).toarray()[0][0] / (np.linalg.norm(vec1.toarray()) * np.linalg.norm(vec2.toarray()))


# 单元测试
import unittest


class TestToolSelector(unittest.TestCase):
    def setUp(self):
        self.tools = [
            {
                'name': 'weather',
                'description': '获取天气信息'
            },
            {
                'name': 'calculator',
                'description': '数学计算'
            }
        ]
        self.selector = ToolSelector(self.tools)

    async def test_weather_query(self):
        tool = await self.selector.select_tool('今天北京天气怎么样')
        self.assertEqual(tool['name'], 'weather')

    async def test_calculator_query(self):
        tool = await self.selector.select_tool('计算 1 加 1 等于几')
        self.assertEqual(tool['name'], 'calculator')

实现工具热加载可以让系统在不重启的情况下更新工具集合：

使用文件监听机制（如 watchdog）监控工具描述文件的变更
当文件变更时，重新加载工具描述
更新向量化模型

配置限流和熔断参数可以提高系统的稳定性：

circuit_breaker:
  failure_threshold: 5
  recovery_timeout: 60
  expected_exceptions:
    - TimeoutError
    - ConnectionError
rate_limit:
  max_calls: 100
  period: 60

对于长对话上下文，可以采用以下压缩策略：