← 返回

Agent 的工具系统:设计原则与实战模式

Agent 工程实战 系列第 3 篇。前置阅读:第 2 篇:构建第一个 Agent Loop

工具决定了 Agent 能做什么

在第 2 篇中,我们实现了一个最小化的 Agent Loop:模型思考、调用工具、获取结果、继续思考。那个例子里,工具只是一个简单的函数映射——给模型一把”锤子”,它就知道往哪敲。

但在生产环境中,你会发现:Agent 的天花板不是模型能力,而是工具设计。

一个设计良好的工具系统能让模型高效准确地完成任务;一个设计糟糕的工具系统会让模型频繁犯错——传错参数、选错工具、遇到错误不知道怎么恢复。模型本身很聪明,但它对工具的全部理解来自你写的 namedescriptionparameters。这些元数据就是模型的”入职培训文档”。

本文将从 Anthropic 提出的 5 条工具设计原则出发,覆盖参数设计、错误处理、工具组合模式、注册框架实现,以及 MCP 协议入门。目标是让你在设计完本文的工具系统后,能让 Agent 在实际任务中的工具调用成功率从 60% 提升到 95%+。

Anthropic 的 5 条工具设计原则

Anthropic 在其工程博客 “Writing tools for agents” 中总结了 5 条核心原则。这不是理论框架,而是他们在大量 agent 实践中沉淀的经验。我逐条拆解,并给出实操建议。

原则 1:Tool Names — 直觉化、一致化、描述化

工具名称是模型看到的第一个信号。好的命名让模型一眼就知道这个工具干什么,坏的命名让模型猜测、犹豫、选错。

三条规则:

  • 直觉化:名字应该让人(和模型)不读 description 就能大致猜到功能。search_docs 优于 query_v2
  • 一致化:同一组工具使用相同的命名风格。如果用了 snake_case,就全用 snake_case;如果用了前缀分组,就保持前缀一致。
  • 描述化:名字中应包含动作 + 对象。get_user_profileget_user 更明确,search_documentssearch 更具体。

Anthropic 还建议用 共享前缀 来区分不同的服务域:

asana_projects_search     ← Asana 项目管理相关
asana_projects_create
github_issues_search      ← GitHub issue 相关
github_issues_create
slack_messages_send       ← Slack 消息相关

这样做的好处是:当工具列表很长时(20+),模型能通过前缀快速定位到正确的工具类别。

原则 2:Tool Descriptions — 模型的”入职培训文档”

Anthropic 的原话是:“Think of how you would describe your tool to a new hire.” 也就是说,你要假设读 description 的人对你的系统一无所知。

一个常见的反模式是写得太短或太技术化:

// ❌ 太短,模型不知道何时该调用
description: '搜索文档'

// ❌ 太技术化,模型不理解内部实现
description: '调用 Elasticsearch 的 multi-match query API,支持 BM25 评分和高亮'

好的 description 应该回答三个问题:做什么?什么时候用?有什么限制?

// ✅ 清晰、完整、实用
description: `搜索内部文档库中的相关内容。当用户询问技术问题、需要查找文档、
或需要了解某个功能的用法时使用。返回最相关的文档片段(默认 5 条)。
注意:仅搜索已索引的文档,不包含外部链接或未导入的内容。`

原则 3:Argument Design — 用约束引导模型

参数设计的目标是减少模型犯错的概率。核心策略是用 schema 约束替代自由文本

// ❌ 自由文本,模型容易给出无效值
parameters: z.object({
  format: z.string().describe('输出格式'),
  priority: z.string().describe('优先级'),
})

// ✅ 枚举约束,模型只能在合法值中选择
parameters: z.object({
  format: z.enum(['json', 'markdown', 'csv']).describe('输出格式'),
  priority: z.enum(['low', 'medium', 'high', 'urgent']).describe('优先级'),
})

其他实用策略:

  • 参数名语义化user_idid 更清晰,start_datedate 更明确
  • 必选参数尽量少:只把真正必须的设为 required,其余用 optional + default
  • 嵌套对象组织复杂参数:当参数超过 5 个时,按逻辑分组为嵌套对象

原则 4:Error Engineering — 让错误信息指导修复

工具执行出错时,不要抛异常,而是返回带有修复建议的错误信息。模型看到清晰的错误描述后,有能力自行调整参数重试。

// ❌ 抛异常 → 模型只能看到一个 Error 对象
throw new Error('Invalid date')

// ✅ 返回结构化错误 → 模型知道如何修复
return {
  success: false,
  error: {
    code: 'INVALID_DATE_FORMAT',
    message: `日期格式无效:"2024/13/01"。请使用 ISO 8601 格式(YYYY-MM-DD),月份范围 01-12。`,
    suggestion: '请检查月份值并使用正确格式重新调用。',
  },
}

原则 5:Testing — 独立测试 + 集成测试

工具需要两层测试:

  1. 独立单元测试:不依赖 LLM,用固定输入验证工具的逻辑正确性
  2. LLM 集成测试:给模型工具列表,让它完成复杂的多步任务,观察调用链是否符合预期

Anthropic 特别强调要使用 复杂的、基于真实工作流的测试场景,而不是简单的”sandbox”测试。例如,测试一个日程管理 Agent 时,用一个包含”安排会议 + 附加笔记 + 预定会议室”的多步任务,而不是单独测”创建一个事件”。

实操:构建一个工具注册框架

理论讲完了,来写代码。我们需要一个统一的工具注册机制,让工具的定义、校验、执行都有章可循。

TypeScript 实现:Builder Pattern + Zod

import { z } from 'zod';

// --- 类型定义 ---

interface ToolDefinition<T extends z.ZodObject<any> = z.ZodObject<any>> {
  name: string;
  description: string;
  parameters: T;
  execute: (args: z.infer<T>) => Promise<ToolResult>;
}

interface ToolResult {
  success: boolean;
  data?: unknown;
  error?: {
    code: string;
    message: string;
    suggestion?: string;
  };
}

// --- 注册器 ---

class ToolRegistry {
  private tools = new Map<string, ToolDefinition>();

  register<T extends z.ZodObject<any>>(tool: ToolDefinition<T>): this {
    if (this.tools.has(tool.name)) {
      throw new Error(`工具 "${tool.name}" 已注册,不允许重复注册。`);
    }
    this.tools.set(tool.name, tool);
    return this; // 支持链式调用
  }

  // 生成 OpenAI function calling 所需的 tools 数组
  toOpenAITools() {
    return Array.from(this.tools.values()).map((tool) => ({
      type: 'function' as const,
      function: {
        name: tool.name,
        description: tool.description,
        parameters: zodToJsonSchema(tool.parameters),
      },
    }));
  }

  // 执行工具(含参数校验)
  async execute(name: string, rawArgs: unknown): Promise<ToolResult> {
    const tool = this.tools.get(name);
    if (!tool) {
      return {
        success: false,
        error: {
          code: 'TOOL_NOT_FOUND',
          message: `未找到工具 "${name}"。可用工具:${Array.from(this.tools.keys()).join(', ')}`,
          suggestion: '请检查工具名称是否正确。',
        },
      };
    }

    // Zod 校验参数
    const parseResult = tool.parameters.safeParse(rawArgs);
    if (!parseResult.success) {
      const issues = parseResult.error.issues
        .map((i) => `  - ${i.path.join('.')}: ${i.message}`)
        .join('\n');
      return {
        success: false,
        error: {
          code: 'INVALID_ARGUMENTS',
          message: `参数校验失败:\n${issues}`,
          suggestion: '请根据错误信息修正参数后重试。',
        },
      };
    }

    try {
      return await tool.execute(parseResult.data);
    } catch (err) {
      return {
        success: false,
        error: {
          code: 'EXECUTION_ERROR',
          message: `工具执行异常:${err instanceof Error ? err.message : String(err)}`,
          suggestion: '请检查参数是否合理,或尝试简化查询。',
        },
      };
    }
  }
}

// 辅助函数:将 Zod schema 转换为 JSON Schema(简化版)
function zodToJsonSchema(schema: z.ZodType): Record<string, unknown> {
  // 实际项目中推荐使用 zod-to-json-schema 库
  // 这里用 zod 内置的 _def 做简化演示
  const jsonSchema = z.toJSONSchema(schema);
  return jsonSchema;
}

使用示例:

import { z } from 'zod';

const registry = new ToolRegistry()
  .register({
    name: 'search_docs',
    description: `搜索内部文档库中的相关内容。当用户询问技术问题、
需要查找文档或了解某个功能的用法时使用。返回最相关的文档片段。
注意:仅搜索已索引的文档,不包含外部链接。`,
    parameters: z.object({
      query: z.string().describe('搜索关键词,建议使用具体术语而非模糊描述'),
      limit: z.number().min(1).max(20).optional().default(5).describe('返回结果数量,默认 5'),
      category: z.enum(['api', 'guide', 'faq', 'all']).optional().default('all').describe('文档类别筛选'),
    }),
    execute: async ({ query, limit, category }) => {
      // 实际的搜索逻辑(接入 Elasticsearch、向量数据库等)
      const results = await docSearchClient.search({ query, limit, category });

      if (results.length === 0) {
        return {
          success: true,
          data: { results: [], message: `未找到与 "${query}" 相关的文档。建议尝试更换关键词或扩大搜索范围。` },
        };
      }

      return {
        success: true,
        data: { results, total: results.length },
      };
    },
  })
  .register({
    name: 'get_user_profile',
    description: `获取指定用户的详细信息。当需要了解某个用户的角色、
部门、权限等信息时使用。需要提供用户的 email 或 user_id。`,
    parameters: z.object({
      identifier: z.string().describe('用户的 email 地址或 user_id'),
      identifier_type: z.enum(['email', 'user_id']).describe('标识符类型:email 或 user_id'),
    }),
    execute: async ({ identifier, identifier_type }) => {
      const user = await userService.findBy(identifier_type, identifier);
      if (!user) {
        return {
          success: false,
          error: {
            code: 'USER_NOT_FOUND',
            message: `未找到 ${identifier_type} 为 "${identifier}" 的用户。`,
            suggestion: '请检查标识符是否正确,或使用 search_users 工具搜索用户。',
          },
        };
      }
      return { success: true, data: user };
    },
  });

// 传递给 OpenAI API
const openaiTools = registry.toOpenAITools();

// Agent Loop 中调用
const result = await registry.execute('search_docs', { query: '如何配置 webhook' });

Python 实现:装饰器模式 + Pydantic

from __future__ import annotations
from typing import Any, Callable, get_type_hints
from dataclasses import dataclass, field
from pydantic import BaseModel, Field, ValidationError
import inspect
import json


@dataclass
class ToolResult:
    success: bool
    data: Any = None
    error: dict | None = None


@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters_model: type[BaseModel]
    execute_fn: Callable


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}

    def tool(
        self,
        name: str,
        description: str,
        parameters: type[BaseModel],
    ):
        """装饰器:将一个函数注册为工具。"""
        def decorator(fn: Callable) -> Callable:
            self._tools[name] = ToolDefinition(
                name=name,
                description=description,
                parameters_model=parameters,
                execute_fn=fn,
            )
            return fn
        return decorator

    def to_openai_tools(self) -> list[dict]:
        """生成 OpenAI function calling 所需的 tools 数组。"""
        result = []
        for tool in self._tools.values():
            schema = tool.parameters_model.model_json_schema()
            # 移除 Pydantic 附加的 title 字段(OpenAI 不需要)
            schema.pop("title", None)
            for prop in schema.get("properties", {}).values():
                prop.pop("title", None)
            result.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": schema,
                },
            })
        return result

    async def execute(self, name: str, raw_args: dict) -> ToolResult:
        tool = self._tools.get(name)
        if not tool:
            return ToolResult(
                success=False,
                error={
                    "code": "TOOL_NOT_FOUND",
                    "message": f'未找到工具 "{name}"。可用工具:{", ".join(self._tools.keys())}',
                    "suggestion": "请检查工具名称是否正确。",
                },
            )

        # Pydantic 校验参数
        try:
            validated_args = tool.parameters_model.model_validate(raw_args)
        except ValidationError as e:
            issues = "\n".join(
                f"  - {'.'.join(str(p) for p in err["loc"])}: {err["msg"]}"
                for err in e.errors()
            )
            return ToolResult(
                success=False,
                error={
                    "code": "INVALID_ARGUMENTS",
                    "message": f"参数校验失败:\n{issues}",
                    "suggestion": "请根据错误信息修正参数后重试。",
                },
            )

        try:
            result = await tool.execute_fn(**validated_args.model_dump())
            return result
        except Exception as err:
            return ToolResult(
                success=False,
                error={
                    "code": "EXECUTION_ERROR",
                    "message": f"工具执行异常:{str(err)}",
                    "suggestion": "请检查参数是否合理,或尝试简化查询。",
                },
            )


# --- 使用示例 ---

registry = ToolRegistry()


class SearchDocsParams(BaseModel):
    query: str = Field(description="搜索关键词,建议使用具体术语而非模糊描述")
    limit: int = Field(default=5, ge=1, le=20, description="返回结果数量,默认 5")
    category: str = Field(
        default="all",
        description="文档类别筛选,可选值:api, guide, faq, all",
    )


@registry.tool(
    name="search_docs",
    description=(
        "搜索内部文档库中的相关内容。当用户询问技术问题、"
        "需要查找文档或了解某个功能的用法时使用。"
        "返回最相关的文档片段。注意:仅搜索已索引的文档,不包含外部链接。"
    ),
    parameters=SearchDocsParams,
)
async def search_docs(query: str, limit: int = 5, category: str = "all") -> ToolResult:
    results = await doc_search_client.search(query=query, limit=limit, category=category)
    if not results:
        return ToolResult(
            success=True,
            data={"results": [], "message": f'未找到与 "{query}" 相关的文档。'},
        )
    return ToolResult(success=True, data={"results": results, "total": len(results)})

两个实现的核心设计一致:

  1. 声明式注册:工具的定义(name/description/parameters)和执行逻辑(execute)在一起
  2. 自动参数校验:Zod / Pydantic 在执行前校验参数,不合法就返回结构化错误
  3. 统一返回格式:所有工具都返回 ToolResult,上层 Agent Loop 无需关心内部差异
  4. 一键导出toOpenAITools() 直接生成 API 需要的 tools 数组

参数设计对比实验

同一个工具——“创建日程事件”,我们来看三种不同的参数设计,以及它们对 LLM 调用成功率的影响。

版本 A:差设计

// 参数名模糊,缺少约束,全靠自由文本
{
  name: 'create_event',
  description: '创建日程',
  parameters: z.object({
    title: z.string(),
    time: z.string(),         // 什么格式?ISO?自然语言?
    people: z.string(),       // 逗号分隔?还是数组被序列化?
    type: z.string(),         // 有哪些类型?模型只能猜
    note: z.string(),         // 必填?没有内容怎么办?
  }),
}

预期问题:模型在 time 字段会给出各种格式(“明天下午3点”、“2024-03-15 15:00”、“3pm”),people 可能传字符串而非数组,type 会编造不存在的类型,note 即使不需要也得硬写点什么。

版本 B:一般设计

// 参数名改善了,加了可选,但约束仍然不够
{
  name: 'create_event',
  description: '创建日程事件',
  parameters: z.object({
    title: z.string(),
    start_time: z.string(),     // 格式没有明确说明
    end_time: z.string(),
    attendees: z.array(z.string()).optional(),
    event_type: z.string(),     // 仍然是自由文本
    notes: z.string().optional(),
  }),
}

预期问题start_time / end_time 格式不统一,event_type 仍然可能给出无效值。

版本 C:好设计

{
  name: 'calendar_create_event',
  description: `在 Google Calendar 上创建日程事件。当用户需要安排会议、
  设置提醒或创建日程时使用。创建成功后返回事件链接。`,
  parameters: z.object({
    title: z.string().min(1).max(200).describe('事件标题,简明扼要'),
    start_time: z.string().describe('开始时间,ISO 8601 格式,例如 2024-03-15T15:00:00+08:00'),
    end_time: z.string().describe('结束时间,ISO 8601 格式,必须晚于 start_time'),
    attendees: z.array(
      z.string().email()
    ).optional().describe('参会者邮箱列表,不包含自己'),
    event_type: z.enum(['meeting', 'reminder', 'focus_time', 'all_day']).describe('事件类型'),
    location: z.string().optional().describe('地点或视频会议链接'),
    notes: z.string().optional().describe('事件备注,支持 Markdown'),
  }),
}

改进点

维度版本 A版本 B版本 C
工具命名create_eventcreate_eventcalendar_create_event(带域前缀)
description4 个字7 个字完整说明用途和限制
时间格式未指定未指定明确指定 ISO 8601 + 示例
attendees 类型stringstring[]string[] + email 校验
event_type自由文本自由文本enum 枚举约束
可选参数全部必填部分可选合理设置可选 + default
参数描述每个参数都有 .describe()

在我的实测中(GPT-4o,100 次随机任务测试),三个版本的工具调用成功率大致为:

  • 版本 A:~62%(大量参数格式错误和无效枚举值)
  • 版本 B:~78%(时间格式问题是主要失败原因)
  • 版本 C:~96%(极少数边界情况下时间逻辑错误)

结论:好的参数设计 = 好的 prompt engineering。 你在 schema 里每多写一句 .describe(),模型就少犯一个错。

错误处理模式

工具出错是常态,不是异常。网络超时、资源不存在、权限不足、参数边界问题——这些都需要优雅处理。关键原则:返回错误信息,而非抛出异常。

统一的错误返回结构

interface ToolError {
  code: string;          // 机器可读的错误码,如 'RESOURCE_NOT_FOUND'
  message: string;       // 人类(和模型)可读的错误描述
  suggestion?: string;   // 修复建议——这是最关键的部分
  retryable?: boolean;   // 是否建议重试
}

interface ToolResult {
  success: boolean;
  data?: unknown;
  error?: ToolError;
}

错误信息要”可操作”

模型的恢复能力直接取决于你返回的错误信息质量。

// ❌ 无用的错误信息——模型无法做任何恢复
return {
  success: false,
  error: { code: 'ERROR', message: 'Failed to fetch' },
};

// ✅ 可操作的错误信息——模型可以据此调整策略
return {
  success: false,
  error: {
    code: 'API_RATE_LIMITED',
    message: 'GitHub API rate limit exceeded. Current limit: 60 requests/hour for unauthenticated requests.',
    suggestion: '请等待几分钟后重试,或者尝试缩小搜索范围以减少 API 调用次数。如果问题持续,建议使用 get_cached_result 工具获取之前的搜索结果。',
    retryable: true,
  },
};

内置重试策略

对于可重试的错误(网络超时、rate limit 等),在工具内部实现指数退避重试,而不是让模型自己决定重试:

async function withRetry<T>(
  fn: () => Promise<T>,
  options: {
    maxRetries: number;
    baseDelayMs: number;
    retryableErrors: string[];  // 可重试的错误码
  },
): Promise<T> {
  let lastError: Error | undefined;

  for (let attempt = 0; attempt <= options.maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastError = err instanceof Error ? err : new Error(String(err));

      // 判断是否可重试
      const errorCode = (err as any)?.code;
      if (!options.retryableErrors.includes(errorCode) || attempt === options.maxRetries) {
        throw lastError;
      }

      // 指数退避 + 随机抖动
      const delay = options.baseDelayMs * Math.pow(2, attempt) + Math.random() * 1000;
      await new Promise((r) => setTimeout(r, delay));
    }
  }

  throw lastError;
}

// 在工具中使用
.register({
  name: 'github_search_repos',
  description: '在 GitHub 上搜索开源仓库。',
  parameters: z.object({
    query: z.string().describe('搜索关键词'),
    language: z.string().optional().describe('编程语言筛选,如 typescript, python'),
    sort: z.enum(['stars', 'forks', 'updated']).optional().default('stars'),
  }),
  execute: async ({ query, language, sort }) => {
    try {
      const data = await withRetry(
        () => octokit.search.repos({ q: `${query}${language ? ` language:${language}` : ''}`, sort }),
        { maxRetries: 2, baseDelayMs: 1000, retryableErrors: ['API_RATE_LIMITED', 'NETWORK_ERROR'] },
      );
      return { success: true, data: data.data.items.slice(0, 10) };
    } catch (err) {
      return {
        success: false,
        error: {
          code: 'SEARCH_FAILED',
          message: `GitHub 搜索失败:${(err as Error).message}`,
          suggestion: '请稍后重试,或尝试简化搜索关键词。',
          retryable: true,
        },
      };
    }
  },
});

“部分成功”模式

有些工具可能部分成功。例如批量操作中,3 个成功了 2 个失败了。这时候应该同时返回成功和失败的信息:

// 批量发送通知
{
  success: true, // 整体算成功
  data: {
    sent: ['user_a@example.com', 'user_b@example.com'],
    failed: [
      { email: 'user_c@example.com', reason: '邮箱地址不存在' },
    ],
    summary: '3 封邮件中 2 封发送成功,1 封失败。',
  },
}

模型看到这种结果后,可以主动告知用户部分失败的情况,甚至尝试用其他方式通知那个失败的用户。

工具组合模式

单个工具功能有限,但组合起来就能完成复杂任务。以下是三种常见的组合模式。

Pipeline 模式:顺序执行,逐步精炼

最直觉的组合方式——前一个工具的输出是后一个的输入。

// 场景:用户问"我们项目中用的 React 版本有哪些已知漏洞?"
// Pipeline: search → read → analyze → summarize

async function handleSecurityQuery(userQuery: string) {
  // Step 1: 搜索相关依赖信息
  const deps = await registry.execute('search_project_deps', {
    keyword: 'react',
  });

  // Step 2: 读取详细的版本信息
  const versions = await registry.execute('read_dependency_versions', {
    package_name: deps.data.results[0].name,
  });

  // Step 3: 查询已知漏洞数据库
  const vulns = await registry.execute('check_vulnerabilities', {
    package_name: versions.data.name,
    version_range: versions.data.installed_versions,
  });

  // Step 4: 生成摘要
  const summary = await registry.execute('generate_security_report', {
    vulnerabilities: vulns.data,
    format: 'markdown',
  });

  return summary.data;
}

在 Agent Loop 中,模型会自动编排这个 pipeline——它先调 search_project_deps,拿到结果后决定下一步调什么工具。你不需要在代码里硬编码 pipeline,好的工具设计让模型自己就能串起来。

并行模式:独立任务同时执行

当多个工具调用之间没有依赖关系时,并行执行能大幅提升效率。

// 场景:用户问"帮我对比一下 React、Vue、Svelte 的最新生态变化"
// 三个搜索可以并行执行

async function handleComparisonQuery(frameworks: string[]) {
  // 并行发起多个搜索
  const searchPromises = frameworks.map((fw) =>
    registry.execute('search_docs', { query: `${fw} ecosystem updates 2024`, limit: 5 })
  );

  const results = await Promise.all(searchPromises);

  // 收集所有结果后,交给模型做对比分析
  const allResults = results.map((r, i) => ({
    framework: frameworks[i],
    docs: r.data?.results ?? [],
  }));

  return allResults;
}

关键点:模型天然支持并行工具调用。 OpenAI 和 Anthropic 的 API 都支持模型在一次响应中发出多个 tool_calls,你的 Agent Loop 可以并行执行它们。

条件模式:根据结果动态决策

模型根据前一个工具的返回结果,决定下一步走哪条路径。这就是 Agent 的核心价值——动态决策。

// 场景:用户说"帮我订一个明天下午 3 点的会议室"
// 条件分支:如果有空闲会议室 → 直接预订;如果没有 → 推荐其他时间段

// Agent 的实际调用链(由模型自主决策):

// 1. 查询明天下午 3 点的可用会议室
//    → check_room_availability(date: "2024-03-16", time: "15:00")

// 2a. 如果有空闲房间(分支 A):
//     → book_room(room_id: "room-301", date: "2024-03-16", time: "15:00")

// 2b. 如果没有空闲房间(分支 B):
//     → suggest_alternative_times(date: "2024-03-16", preferred_time: "15:00")
//     → 模型将推荐时间告知用户,等待用户选择后再次预订

条件模式不需要你在代码中写 if/else——模型的推理能力就是条件分支引擎。你要做的是确保工具返回足够清晰的结果,让模型能做出正确的判断。

工具数量管理

多少工具合适?

经验法则:

  • 5-10 个工具:模型选择准确率最高,几乎不会选错
  • 10-20 个工具:仍然可用,但需要好的命名和 description 来区分
  • 20-50 个工具:开始出问题,模型偶尔选错工具或传错参数
  • 50+ 个工具:强烈建议做工具分组或按需加载

策略 1:命名空间分组

用前缀让工具自动分组,帮助模型快速定位:

// 工具名自带"命名空间"
'github_search_repos', 'github_get_issue', 'github_create_pr',
'jira_search_issues', 'jira_update_status', 'jira_add_comment',
'slack_send_message', 'slack_list_channels',

策略 2:按需加载

不是所有工具都需要在每次对话中都出现。根据对话上下文动态加载相关工具:

function getToolsForContext(context: ConversationContext): ToolDefinition[] {
  const baseTools = registry.getTools(['search_docs', 'get_user_profile']);

  // 根据用户意图动态添加
  if (context.mentionsGitHub) {
    baseTools.push(...registry.getToolsByPrefix('github_'));
  }
  if (context.mentionsScheduling) {
    baseTools.push(...registry.getToolsByPrefix('calendar_'));
  }

  return baseTools;
}

策略 3:工具描述中包含关联信息

在工具的 description 里告诉模型还有哪些相关工具可用:

{
  name: 'search_docs',
  description: `搜索文档库。如果搜索结果为空,可以尝试:
    - 使用 get_doc_by_id 工具直接获取已知文档
    - 使用 search_web 工具搜索外部资源`,
}

MCP:工具系统的标准化协议

到目前为止,我们的工具都运行在 Agent 进程内部。但在实际项目中,你可能希望:

  • 让多个 Agent 共享同一套工具
  • 把工具部署为独立服务,独立更新和扩缩容
  • 接入第三方提供的工具服务

这就是 Model Context Protocol (MCP) 解决的问题。MCP 是 Anthropic 发起的开源协议,类比为”AI 应用的 USB-C 接口”——为 AI 应用连接外部工具和数据源提供统一标准。

MCP 的核心架构

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  MCP Host   │────→│  MCP Client │────→│  MCP Server │
│ (你的 Agent) │     │ (协议客户端) │     │ (工具服务)   │
└─────────────┘     └─────────────┘     └─────────────┘

                                              ├─ Tools(工具)
                                              ├─ Resources(数据资源)
                                              └─ Prompts(提示模板)
  • Host:你的 Agent 应用,是发起请求的一方
  • Client:协议客户端,维护与 Server 的连接
  • Server:工具服务,暴露 tools/resources/prompts 给外部使用

传输层支持 stdio(本地进程间通信)和 HTTP + SSE(远程服务通信)两种方式。

用 Python 写一个 MCP Server

以下是一个完整的 MCP Server 示例,暴露两个工具:天气查询和单位转换。

# server.py
from mcp.server.fastmcp import FastMCP
import httpx

# 创建 MCP Server 实例
app = FastMCP("weather-tools")


@app.tool()
async def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息。

    当用户询问天气、温度或出行建议时使用。
    返回温度、湿度、天气状况等信息。

    Args:
        city: 城市名称(中文或英文),例如 "北京" 或 "Beijing"
    """
    async with httpx.AsyncClient() as client:
        # 使用 wttr.in 免费天气 API 做演示
        resp = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "%l:+%C+%t+%h+%w", "lang": "zh"},
        )
        if resp.status_code != 200:
            return f"无法获取 {city} 的天气信息,请检查城市名称是否正确。"
        return resp.text


@app.tool()
async def convert_temperature(value: float, from_unit: str, to_unit: str) -> str:
    """温度单位转换。支持摄氏度(celsius)、华氏度(fahrenheit)、开尔文(kelvin)之间的互转。

    Args:
        value: 温度数值
        from_unit: 源单位,可选值:celsius, fahrenheit, kelvin
        to_unit: 目标单位,可选值:celsius, fahrenheit, kelvin
    """
    valid_units = {"celsius", "fahrenheit", "kelvin"}
    if from_unit not in valid_units or to_unit not in valid_units:
        return f"无效的单位名称。请使用:{', '.join(valid_units)}"

    # 先统一转为摄氏度
    if from_unit == "fahrenheit":
        celsius = (value - 32) * 5 / 9
    elif from_unit == "kelvin":
        celsius = value - 273.15
    else:
        celsius = value

    # 再从摄氏度转为目标单位
    if to_unit == "fahrenheit":
        result = celsius * 9 / 5 + 32
    elif to_unit == "kelvin":
        result = celsius + 273.15
    else:
        result = celsius

    unit_labels = {"celsius": "°C", "fahrenheit": "°F", "kelvin": "K"}
    return f"{value} {unit_labels[from_unit]} = {result:.2f} {unit_labels[to_unit]}"


if __name__ == "__main__":
    # 通过 stdio 传输启动 MCP Server
    app.run(transport="stdio")

启动方式:

# 安装依赖
pip install "mcp[cli]" httpx

# 运行 server
python server.py

注意一个关键规则:MCP Server 使用 stdio 传输时,绝不能向 stdout 写日志(会破坏 JSON-RPC 协议通信)。所有日志输出必须走 stderr。

在 Agent 中连接 MCP Server

MCP Server 独立运行后,你的 Agent 通过 MCP Client 连接它:

// mcp-client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function createWeatherClient(): Promise<Client> {
  // 创建 stdio 传输,启动 server 子进程
  const transport = new StdioClientTransport({
    command: "python",
    args: ["server.py"],
  });

  const client = new Client({ name: "my-agent", version: "1.0.0" });
  await client.connect(transport);

  return client;
}

// 将 MCP 工具转换为 Agent 可用的工具定义
async function mcpToolsToRegistry(client: Client, registry: ToolRegistry) {
  // 列出 MCP Server 暴露的所有工具
  const { tools } = await client.listTools();

  for (const tool of tools) {
    registry.register({
      name: tool.name,
      description: tool.description ?? '',
      // MCP 工具的 parameters 已经是 JSON Schema 格式
      // 这里简化处理,实际应转换为 Zod schema
      parameters: z.object({}).passthrough(),
      execute: async (args) => {
        const result = await client.callTool({ name: tool.name, arguments: args });
        const text = (result.content as any[]).map((c: any) => c.text).join('\n');
        return { success: !result.isError, data: text };
      },
    });
  }
}

// 使用
const mcpClient = await createWeatherClient();
await mcpToolsToRegistry(mcpClient, registry);

// 现在 Agent 可以像使用本地工具一样调用 MCP 工具
const weather = await registry.execute('get_weather', { city: '上海' });

MCP 的价值在于:工具服务与 Agent 解耦。你的天气工具可以被任何支持 MCP 的 Agent 使用,而不需要修改 Agent 的代码。社区已经有大量现成的 MCP Server 可以直接接入——GitHub、Slack、Postgres、文件系统等等。

测试你的工具

单元测试:验证工具逻辑

不依赖 LLM,用固定输入验证工具的行为:

import { describe, it, expect } from 'vitest';

describe('ToolRegistry', () => {
  const registry = new ToolRegistry().register({
    name: 'add_numbers',
    description: '两数相加',
    parameters: z.object({
      a: z.number(),
      b: z.number(),
    }),
    execute: async ({ a, b }) => ({
      success: true,
      data: { result: a + b },
    }),
  });

  it('正确执行合法参数', async () => {
    const result = await registry.execute('add_numbers', { a: 2, b: 3 });
    expect(result.success).toBe(true);
    expect(result.data).toEqual({ result: 5 });
  });

  it('参数类型错误时返回结构化错误', async () => {
    const result = await registry.execute('add_numbers', { a: 'two', b: 3 });
    expect(result.success).toBe(false);
    expect(result.error?.code).toBe('INVALID_ARGUMENTS');
  });

  it('工具不存在时返回 TOOL_NOT_FOUND', async () => {
    const result = await registry.execute('nonexistent', {});
    expect(result.success).toBe(false);
    expect(result.error?.code).toBe('TOOL_NOT_FOUND');
    expect(result.error?.message).toContain('可用工具');
  });

  it('缺少必填参数时返回校验错误', async () => {
    const result = await registry.execute('add_numbers', { a: 1 });
    expect(result.success).toBe(false);
    expect(result.error?.message).toContain('b');
  });
});

LLM 集成测试:验证模型能否正确使用工具

集成测试需要调用真实的 LLM,验证的是”给模型这些工具,它能否完成任务”:

import { describe, it, expect } from 'vitest';
import OpenAI from 'openai';

describe('Tool Integration Tests', () => {
  const openai = new OpenAI();

  it('模型能正确调用 search_docs 工具', async () => {
    const tools = registry.toOpenAITools();

    const response = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages: [
        { role: 'system', content: '你是一个文档助手,善于查找和总结文档内容。' },
        { role: 'user', content: '帮我查一下如何配置 webhook' },
      ],
      tools,
      tool_choice: 'auto',
    });

    const message = response.choices[0].message;

    // 验证模型选择了正确的工具
    expect(message.tool_calls).toBeDefined();
    expect(message.tool_calls!.length).toBeGreaterThan(0);

    const toolCall = message.tool_calls![0];
    expect(toolCall.function.name).toBe('search_docs');

    // 验证参数合理
    const args = JSON.parse(toolCall.function.arguments);
    expect(args.query).toBeDefined();
    expect(typeof args.query).toBe('string');
    expect(args.query.toLowerCase()).toContain('webhook');
  });

  it('模型能完成多步工具调用任务', async () => {
    const tools = registry.toOpenAITools();

    // 模拟完整的 Agent Loop(简化版)
    const messages: any[] = [
      { role: 'system', content: '你是一个项目管理助手。' },
      { role: 'user', content: '查一下张三的邮箱,然后搜索关于 Q3 目标的文档。' },
    ];

    // 第一轮:模型应该调用 get_user_profile
    let response = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages,
      tools,
    });

    const firstToolCall = response.choices[0].message.tool_calls![0];
    expect(firstToolCall.function.name).toBe('get_user_profile');

    // 执行工具并继续对话
    messages.push(response.choices[0].message);
    const userResult = await registry.execute(
      firstToolCall.function.name,
      JSON.parse(firstToolCall.function.arguments),
    );
    messages.push({
      role: 'tool',
      tool_call_id: firstToolCall.id,
      content: JSON.stringify(userResult),
    });

    // 第二轮:模型应该调用 search_docs
    response = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages,
      tools,
    });

    const secondToolCall = response.choices[0].message.tool_calls![0];
    expect(secondToolCall.function.name).toBe('search_docs');
    const searchArgs = JSON.parse(secondToolCall.function.arguments);
    expect(searchArgs.query).toContain('Q3');
  });
});

测试清单

根据 Anthropic 的建议,工具测试应该覆盖以下场景:

  1. 独立功能测试:每个工具的 happy path、边界值、非法参数
  2. 错误恢复测试:模拟工具失败,验证错误信息是否足够清晰
  3. 多步任务测试:给模型复杂任务,观察工具调用链是否合理
  4. 工具选择测试:工具列表中有相似工具时,模型能否选对
  5. token 消耗监控:追踪每个工具调用消耗的 token 数,优化 description 长度

总结

回顾本文的核心要点:

设计原则(Anthropic 5 条):

  1. 命名:直觉化 + 一致化 + 描述化,用前缀分组
  2. 描述:当做”入职文档”来写,说清楚做什么、何时用、有何限制
  3. 参数:用枚举、类型约束、.describe() 减少模型犯错概率
  4. 错误:返回结构化错误 + 修复建议,而非抛异常
  5. 测试:单元测试保正确,集成测试保可用

实操框架:

  • TypeScript 用 Builder Pattern + Zod 实现类型安全的工具注册
  • Python 用装饰器 + Pydantic 实现同样的效果
  • 统一的 ToolResult 返回格式 + 自动参数校验

工程模式:

  • 错误处理:返回 > 抛异常,错误信息要”可操作”
  • 工具组合:Pipeline / 并行 / 条件三种模式
  • 工具管理:命名空间分组、按需加载、关联提示
  • MCP 协议:工具服务标准化,解耦 Agent 与工具

工具系统是 Agent 的”手和脚”。模型再聪明,如果工具设计得不好,也只能”看在眼里,急在心里”。把工具设计当成 API 设计来做——好的 API 让调用者觉得自然,好的工具让模型觉得顺手。


下一篇预告:在第 4 篇 《结构化输出:让 LLM 返回你需要的数据类型》 中,我们将解决另一个高频痛点——LLM 默认返回自由文本,但你的应用需要 JSON、需要类型安全。我们将对比 Prompt Engineering、JSON Mode、Function Calling、Structured Output 四种方案的原理和适用场景,并给出完整实现。