Tools 工具集成

函数工具：Python 函数即工具

Agents SDK 最优雅的设计之一：将普通 Python 函数转换为 Agent 可调用的工具，无需手写 JSON Schema。函数的文档字符串自动变成工具描述，类型注解自动生成参数 schema。

from agents import function_tool, Agent, Runner
from datetime import datetime
import asyncio

# ── 最简单的工具：同步函数 ─────────────────────────────────
@function_tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间。

    Args:
        timezone: 时区名称，例如 'Asia/Shanghai'、'America/New_York'
    Returns:
        格式化的当前时间字符串
    """
    from zoneinfo import ZoneInfo
    tz = ZoneInfo(timezone)
    now = datetime.now(tz)
    return now.strftime("%Y-%m-%d %H:%M:%S %Z")

# ── 异步工具：调用外部 API ─────────────────────────────────
@function_tool
async def fetch_weather(city: str) -> str:
    """查询城市当前天气。

    Args:
        city: 城市名称（中文或英文均可）
    Returns:
        天气信息字符串，包含温度、湿度、天气状况
    """
    import httpx
    async with httpx.AsyncClient() as client:
        # 使用免费天气 API（替换为实际 API）
        resp = await client.get(
            f"https://wttr.in/{city}?format=3",
            timeout=10
        )
        return resp.text.strip()

# ── 挂载到 Agent ───────────────────────────────────────────
time_weather_agent = Agent(
    name="时间天气助手",
    instructions="你可以查询时间和天气，准确回答用户问题。",
    model="gpt-4o-mini",
    tools=[get_current_time, fetch_weather]
)

result = Runner.run_sync(time_weather_agent, "北京现在几点，天气怎么样？")
print(result.final_output)

@function_tool 的工作原理

装饰器在底层做了三件事，值得深入理解：

Pydantic 模型作为工具参数

对于复杂的结构化输入，使用 Pydantic 模型作为参数类型，自动生成嵌套的 JSON Schema：

from pydantic import BaseModel, Field
from agents import function_tool
from typing import Literal

# ── 定义参数模型 ───────────────────────────────────────────
class EmailParams(BaseModel):
    to: list[str] = Field(description="收件人邮箱列表")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文（支持 Markdown）")
    priority: Literal["high", "normal", "low"] = "normal"
    cc: list[str] = Field(default=[], description="抄送列表（可选）")

class DatabaseQuery(BaseModel):
    table: str = Field(description="查询的数据库表名")
    filters: dict[str, str] = Field(default={}, description="过滤条件 key-value 对")
    limit: int = Field(default=10, ge=1, le=100, description="返回记录数上限")
    order_by: str = Field(default="created_at", description="排序字段")

# ── 接受 Pydantic 模型的工具 ───────────────────────────────
@function_tool
async def send_email(params: EmailParams) -> str:
    """发送邮件。支持多收件人、优先级设置和抄送功能。"""
    # SDK 自动将 LLM 的 JSON 反序列化为 EmailParams 对象
    print(f"发送邮件到 {params.to}，主题：{params.subject}")
    # 实际调用 SMTP/SendGrid 等邮件服务...
    return f"邮件已发送给 {', '.join(params.to)}"

@function_tool
async def query_database(query: DatabaseQuery) -> str:
    """查询数据库。返回符合条件的记录。"""
    # 参数已经是类型安全的 Pydantic 对象
    sql = f"SELECT * FROM {query.table}"
    if query.filters:
        conditions = " AND ".join([f"{k}='{v}'" for k, v in query.filters.items()])
        sql += f" WHERE {conditions}"
    sql += f" ORDER BY {query.order_by} LIMIT {query.limit}"
    # 执行实际查询...
    return f"执行查询：{sql}，返回 {query.limit} 条结果"

工具错误处理

工具执行失败时，默认行为是将异常信息传递给 LLM，LLM 可能陷入"无限重试"死循环。正确的错误处理策略是返回结构化错误信息：

from agents import function_tool
from agents.exceptions import ToolCallError

# ── 方式 1：返回错误字符串（LLM 会看到并据此决策）───────────
@function_tool
async def search_database(query: str, table: str) -> str:
    """在数据库中搜索记录。"""
    try:
        # 白名单校验：防止 SQL 注入和误操作
        allowed_tables = {"users", "products", "orders"}
        if table not in allowed_tables:
            return f"错误：表 '{table}' 不在允许列表中。可用表：{allowed_tables}"

        results = await db_search(table, query)  # 实际数据库查询

        if not results:
            return "查询成功，但没有找到匹配记录。请尝试更宽泛的搜索词。"

        return f"找到 {len(results)} 条记录：\n" + "\n".join(str(r) for r in results[:5])

    except ConnectionError:
        return "数据库连接失败。请稍后重试，或告知用户系统暂时不可用。"
    except Exception as e:
        return f"查询失败：{type(e).__name__}: {str(e)[:200]}"

# ── 方式 2：failure_error_function（工具级错误处理器）───────
def handle_tool_error(error: Exception, tool_name: str) -> str:
    """统一的工具错误处理函数，返回 LLM 友好的错误描述"""
    error_map = {
        ConnectionError: "服务暂时不可用，请稍后重试",
        TimeoutError: "请求超时，可能网络较慢",
        PermissionError: "权限不足，无法执行此操作",
        ValueError: f"参数错误：{str(error)}",
    }
    for exc_type, msg in error_map.items():
        if isinstance(error, exc_type):
            return f"工具 {tool_name} 执行失败：{msg}"
    return f"工具 {tool_name} 执行失败（未知错误）"

# 在 Agent 级别配置全局错误处理
resilient_agent = Agent(
    name="健壮助手",
    instructions="工具失败时根据错误信息决定是重试还是告知用户。",
    model="gpt-4o-mini",
    tools=[search_database],
)

内置工具：WebSearch

SDK 内置了 WebSearch 工具，无需任何额外配置（需要 OpenAI 平台支持）：

from agents import Agent, Runner
from agents.tools import WebSearchTool

# WebSearchTool 使用 OpenAI 平台的搜索能力
# 无需 API Key，计费在 OpenAI 账单中
search_agent = Agent(
    name="搜索助手",
    instructions="""你是网络搜索助手。用户提问时，先搜索最新信息，
    再综合搜索结果给出准确回答。注明信息来源。""",
    model="gpt-4o",
    tools=[WebSearchTool()]
)

import asyncio
result = asyncio.run(
    Runner.run(search_agent, "OpenAI Agents SDK 最新版本是什么？有哪些新功能？")
)
print(result.final_output)

内置工具：FileSearch（向量存储检索）

FileSearch 工具允许 Agent 搜索你上传到 OpenAI 平台的文件，实现 RAG 检索功能：

from agents import Agent
from agents.tools import FileSearchTool
from openai import AsyncOpenAI

async def setup_rag_agent():
    client = AsyncOpenAI()

    # 步骤 1：创建 Vector Store
    vector_store = await client.beta.vector_stores.create(
        name="产品文档库"
    )

    # 步骤 2：上传文档
    with open("product_docs.pdf", "rb") as f:
        await client.beta.vector_stores.files.upload_and_poll(
            vector_store_id=vector_store.id,
            file=f
        )

    # 步骤 3：创建带 FileSearch 的 Agent
    rag_agent = Agent(
        name="产品知识助手",
        instructions="""你是产品知识助手，回答问题时先搜索产品文档库。
        如果文档中没有相关信息，明确说明"文档中未找到该信息"。""",
        model="gpt-4o",
        tools=[
            FileSearchTool(
                vector_store_ids=[vector_store.id],
                max_num_results=5   # 最多检索 5 个片段
            )
        ]
    )
    return rag_agent

工具链设计原则