为 AI Agent 编写高效工具

Posted on

为 AI Agent 编写高效工具

来源: Anthropic Engineering Blog
作者: Anthropic Engineering Team
发布日期: 2025 年 6 月 18 日
类型: 技术指南
阅读时间: 约 12 分钟

概述

本文详细介绍了如何为 AI Agent 设计和编写高效工具。工具是 AI Agent 与外部世界交互的桥梁，良好的工具设计可以显著提升 Agent 的性能和可靠性。我们分享了工具设计的核心原则、接口设计规范、错误处理策略、性能优化技巧，以及实际案例分析。遵循这些最佳实践，开发者可以创建出易于理解、可靠执行、高效运行的 AI Agent 工具。

为什么工具设计很重要

工具的作用

工具是 AI Agent 能力的延伸：

  • 感知扩展：让 Agent 获取外部信息
  • 执行能力：让 Agent 执行实际操作
  • 记忆增强：让 Agent 访问持久化数据
  • 计算补充：让 Agent 使用专门计算能力

设计不良的影响

模糊的工具定义：

1
2
3
工具名：process_data
参数：data, options
返回：结果

问题：Agent 不知道 data 的格式、options 的含义、结果的结构

清晰的工具定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
工具名：analyze_sales_data
功能：分析销售数据，生成统计报告
参数：
  - csv_file: 销售数据 CSV 文件路径
  - date_range: 可选，日期范围 {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"}
  - metrics: 可选，要计算的指标列表 ["revenue", "units", "growth"]
返回：
  {
    "total_revenue": number,
    "total_units": number,
    "growth_rate": number,
    "by_month": [...]
  }

工具设计原则

1. 单一职责

每个工具只做一件事，并做好：

不推荐：

1
2
3
def process_data(operation, input_data, options=None):
    # 可以做查询、更新、删除、分析...
    # Agent 难以理解何时使用什么操作

推荐：

1
2
3
4
5
6
7
8
9
10
11
def query_database(sql: str) -> list:
    """执行数据库查询"""

def update_record(table: str, id: int, data: dict) -> bool:
    """更新数据库记录"""

def delete_record(table: str, id: int) -> bool:
    """删除数据库记录"""

def analyze_trends(data: list, metric: str) -> dict:
    """分析数据趋势"""

2. 自解释接口

工具接口应该清晰表达意图：

命名规范：

1
2
3
4
好：search_knowledge_base, create_ticket, send_email
差：search, create, send (太通用)
好：get_user_by_id, list_recent_orders
差：get_data, list_items (不够具体)

参数设计：

1
2
3
4
5
6
7
8
9
10
11
12
13
# 好的参数设计
def schedule_meeting(
    title: str,
    attendees: list[str],
    start_time: str,  # ISO 8601 格式
    duration_minutes: int,
    timezone: str = "UTC"
) -> MeetingConfirmation:
    """安排会议并发送邀请"""

# 差的参数设计
def schedule(title, people, time, length, tz="UTC"):
    # 类型不清，格式不明

3. 类型明确

使用明确的类型定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from typing import TypedDict, Literal, Optional

class SearchParams(TypedDict):
    query: str
    filters: Optional[dict]
    limit: int
    offset: int

class SearchResult(TypedDict):
    total: int
    items: list[dict]
    has_more: bool

def search_documents(params: SearchParams) -> SearchResult:
    """搜索文档"""

4. 错误处理

提供清晰的错误信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class ToolError(Exception):
    def __init__(self, code: str, message: str, details: dict = None):
        self.code = code
        self.message = message
        self.details = details or {}

def call_external_api(url: str) -> dict:
    """调用外部 API

    可能错误:
    - NETWORK_ERROR: 网络连接失败
    - AUTH_ERROR: 认证失败
    - RATE_LIMITED: 请求频率超限
    - API_ERROR: API 返回错误
    """
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        raise ToolError("NETWORK_ERROR", "请求超时")
    except requests.HTTPError as e:
        if e.response.status_code == 401:
            raise ToolError("AUTH_ERROR", "认证失败")
        elif e.response.status_code == 429:
            raise ToolError("RATE_LIMITED", "请求频率超限")
        else:
            raise ToolError("API_ERROR", f"API 错误：{e.response.status_code}")

工具实现最佳实践

1. 丰富的文档

为每个工具提供完整文档：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def analyze_sentiment(text: str) -> dict:
    """
    分析文本情感

    这个工具使用 NLP 模型分析文本的情感倾向，返回情感分类和置信度。

    参数:
        text: 要分析的文本，最多 10000 字符

    返回:
        dict: {
            "sentiment": "positive" | "negative" | "neutral",
            "confidence": float (0-1),
            "scores": {
                "positive": float,
                "negative": float,
                "neutral": float
            }
        }

    示例:
        >>> analyze_sentiment("I love this product!")
        {
            "sentiment": "positive",
            "confidence": 0.95,
            "scores": {"positive": 0.95, "negative": 0.02, "neutral": 0.03}
        }

    注意:
        - 文本必须是英文
        - 结果仅供参考，不应作为唯一决策依据
    """

2. 使用示例

提供多个使用示例：

1
2
3
4
5
6
7
8
9
# 示例 1: 基本使用
result = analyze_sentiment("Great product, fast shipping!")

# 示例 2: 处理中性文本
result = analyze_sentiment("The package arrived on time.")

# 示例 3: 批量分析
texts = ["Love it!", "Hate it.", "It's okay."]
results = [analyze_sentiment(t) for t in texts]

3. 边界情况处理

明确处理边界情况：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def analyze_sentiment(text: str) -> dict:
    # 处理空输入
    if not text or not text.strip():
        return {"sentiment": "neutral", "confidence": 1.0, "scores": {...}}

    # 处理过长输入
    if len(text) > 10000:
        text = text[:10000]  # 截断并记录警告

    # 处理不支持的语言
    if not detect_english(text):
        raise ToolError("UNSUPPORTED_LANGUAGE", "仅支持英文分析")

    # 正常处理
    ...

4. 日志和追踪

添加详细日志用于调试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import logging

logger = logging.getLogger("tools.sentiment")

def analyze_sentiment(text: str, request_id: str = None) -> dict:
    request_id = request_id or generate_id()

    logger.info(f"[{request_id}] 开始情感分析", extra={
        "text_length": len(text),
        "request_id": request_id
    })

    try:
        result = model.predict(text)
        logger.info(f"[{request_id}] 分析完成", extra={
            "sentiment": result["sentiment"],
            "confidence": result["confidence"]
        })
        return result
    except Exception as e:
        logger.error(f"[{request_id}] 分析失败：{e}")
        raise

性能优化

1. 缓存策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from functools import lru_cache
import hashlib

def cache_key(func, *args, **kwargs):
    """生成缓存键"""
    key_data = f"{func.__name__}:{args}:{kwargs}"
    return hashlib.md5(key_data.encode()).hexdigest()

@lru_cache(maxsize=1000)
def analyze_sentiment_cached(text_hash: str, text: str) -> dict:
    """带缓存的情感分析"""
    return _analyze_sentiment(text)

def analyze_sentiment(text: str) -> dict:
    text_hash = cache_key("analyze_sentiment", text)
    return analyze_sentiment_cached(text_hash, text)

2. 批量处理

1
2
3
4
5
6
7
8
9
10
11
12
def analyze_sentiment_batch(texts: list[str]) -> list[dict]:
    """批量情感分析，比逐个调用更高效"""
    # 合并请求，减少 API 调用次数
    batch_size = 100
    results = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        batch_results = _batch_analyze(batch)
        results.extend(batch_results)

    return results

3. 异步执行

1
2
3
4
5
6
7
8
9
10
11
import asyncio

async def analyze_sentiment_async(text: str) -> dict:
    """异步情感分析"""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, _analyze_sentiment, text)

async def analyze_multiple(texts: list[str]) -> list[dict]:
    """并发分析多个文本"""
    tasks = [analyze_sentiment_async(text) for text in texts]
    return await asyncio.gather(*tasks)

工具注册和发现

MCP 工具定义

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
{
  "name": "analyze_sentiment",
  "description": "分析文本情感",
  "inputSchema": {
    "type": "object",
    "properties": {
      "text": {
        "type": "string",
        "description": "要分析的文本，最多 10000 字符"
      }
    },
    "required": ["text"]
  },
  "outputSchema": {
    "type": "object",
    "properties": {
      "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
      "confidence": {"type": "number", "minimum": 0, "maximum": 1},
      "scores": {
        "type": "object",
        "properties": {
          "positive": {"type": "number"},
          "negative": {"type": "number"},
          "neutral": {"type": "number"}
        }
      }
    }
  }
}

工具目录

维护工具目录便于发现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# tools_catalog.yaml
tools:
  - name: analyze_sentiment
    category: nlp
    description: 分析文本情感
    tags: [sentiment, nlp, text-analysis]

  - name: search_knowledge_base
    category: search
    description: 搜索知识库
    tags: [search, knowledge, retrieval]

  - name: create_ticket
    category: support
    description: 创建支持工单
    tags: [support, ticket, customer-service]

测试策略

单元测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def test_analyze_sentiment_positive():
    result = analyze_sentiment("I love this!")
    assert result["sentiment"] == "positive"
    assert result["confidence"] > 0.8

def test_analyze_sentiment_negative():
    result = analyze_sentiment("This is terrible.")
    assert result["sentiment"] == "negative"

def test_analyze_sentiment_empty():
    result = analyze_sentiment("")
    assert result["sentiment"] == "neutral"

def test_analyze_sentiment_long():
    long_text = "test " * 5000
    result = analyze_sentiment(long_text)
    assert "sentiment" in result

集成测试

1
2
3
4
5
6
7
def test_sentiment_tool_integration():
    """测试工具与 Agent 的集成"""
    agent = create_agent(tools=[analyze_sentiment])

    response = agent.run("分析这句话的情感：'Great product!'")

    assert "positive" in response.lower()

关键要点总结

  1. 单一职责：每个工具只做一件事
  2. 自解释接口：命名清晰，类型明确
  3. 丰富文档：完整文档和多个示例
  4. 错误处理：清晰的错误码和消息
  5. 性能优化：缓存、批量、异步

个人评价

工具设计是 AI Agent 开发中的关键技能：

优点：

  1. 提升性能：良好工具设计提升 Agent 表现
  2. 增强可靠性：清晰接口减少误用
  3. 便于维护：模块化设计易于更新

总体评价：

这是 AI Agent 开发的必备技能。投资工具设计会带来显著的长期收益。

本文内容翻译自 Anthropic Engineering Blog 官方博客。

© 2026 Generative AI Discovery All Rights Reserved.
Theme by hiero