Agent工具使用

工具带来的能力扩展

• 获取实时数据（天气、新闻、股价等）
• 执行精确计算和数据处理
• 访问和操作文件系统
• 调用外部API和服务
• 执行代码和脚本
• 与数据库交互
• 发送邮件、消息等通知
• 控制物理设备（通过API）

OpenAI Function Calling 格式

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "获取指定城市的当前天气信息。当用户询问天气相关问题时使用此工具。",
    "parameters": {
      "type": "object",
      "properties": {
        "city": {
          "type": "string",
          "description": "城市名称，如'北京'、'上海'"
        },
        "unit": {
          "type": "string",
          "enum": ["celsius", "fahrenheit"],
          "description": "温度单位，默认摄氏度",
          "default": "celsius"
        }
      },
      "required": ["city"]
    }
  }
}

Claude Tool Use 格式

{
  "name": "get_weather",
  "description": "获取指定城市的当前天气信息",
  "input_schema": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称"
      },
      "unit": {
        "type": "string",
        "enum": ["celsius", "fahrenheit"],
        "description": "温度单位"
      }
    },
    "required": ["city"]
  }
}

复杂参数示例

{
  "name": "search_products",
  "description": "搜索商品信息",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "搜索关键词"
      },
      "filters": {
        "type": "object",
        "description": "筛选条件",
        "properties": {
          "price_range": {
            "type": "object",
            "properties": {
              "min": { "type": "number", "description": "最低价格" },
              "max": { "type": "number", "description": "最高价格" }
            }
          },
          "category": {
            "type": "string",
            "description": "商品分类"
          },
          "brand": {
            "type": "string",
            "description": "品牌名称"
          }
        }
      },
      "sort_by": {
        "type": "string",
        "enum": ["price_asc", "price_desc", "rating", "sales"],
        "description": "排序方式"
      },
      "limit": {
        "type": "integer",
        "description": "返回结果数量",
        "default": 10,
        "minimum": 1,
        "maximum": 100
      }
    },
    "required": ["query"]
  }
}

文件操作工具示例

# 文件读取工具
{
  "name": "read_file",
  "description": "读取文件内容",
  "parameters": {
    "type": "object",
    "properties": {
      "path": { "type": "string", "description": "文件路径" },
      "encoding": { "type": "string", "default": "utf-8" }
    },
    "required": ["path"]
  }
}

# 文件写入工具
{
  "name": "write_file",
  "description": "写入内容到文件",
  "parameters": {
    "type": "object",
    "properties": {
      "path": { "type": "string", "description": "文件路径" },
      "content": { "type": "string", "description": "文件内容" },
      "mode": { 
        "type": "string", 
        "enum": ["write", "append"],
        "default": "write"
      }
    },
    "required": ["path", "content"]
  }
}

工具选择流程

1. 分析任务需求
   ↓
2. 匹配工具描述
   - 遍历可用工具列表
   - 将任务需求与工具description匹配
   ↓
3. 选择最合适的工具
   - 可能有多个候选工具
   - 根据上下文选择最佳匹配
   ↓
4. 构造参数
   - 从上下文中提取参数值
   - 验证参数类型和约束
   ↓
5. 执行工具调用
   ↓
6. 解析返回结果

完整执行流程

# Agent工具执行循环

while not task_complete:
    # 1. LLM分析当前状态，决定下一步行动
    response = llm.generate(
        messages=conversation_history,
        tools=available_tools
    )
    
    # 2. 检查是否需要调用工具
    if response.tool_calls:
        for tool_call in response.tool_calls:
            # 3. 验证工具和参数
            tool = get_tool(tool_call.name)
            validated_params = validate(
                tool_call.arguments,
                tool.parameters_schema
            )
            
            # 4. 执行工具
            try:
                result = tool.execute(validated_params)
            except ToolExecutionError as e:
                result = {"error": str(e)}
            
            # 5. 将结果添加到对话历史
            conversation_history.append({
                "role": "tool",
                "name": tool_call.name,
                "content": serialize(result)
            })
    else:
        # 没有工具调用，返回最终答案
        return response.content

错误处理策略

def execute_with_retry(tool, params, max_retries=3):
    """带重试的工具执行"""
    for attempt in range(max_retries):
        try:
            # 参数验证
            validated = validate_params(params, tool.schema)
            
            # 执行工具
            result = tool.execute(validated)
            
            # 结果验证
            if validate_result(result):
                return {"success": True, "data": result}
            else:
                return {"success": False, "error": "Invalid result"}
                
        except ValidationError as e:
            return {"success": False, "error": f"参数错误: {e}"}
            
        except TimeoutError:
            if attempt < max_retries - 1:
                continue  # 重试
            return {"success": False, "error": "执行超时"}
            
        except Exception as e:
            return {"success": False, "error": str(e)}

LangChain工具定义示例

from langchain.tools import tool
from pydantic import BaseModel, Field

# 使用Pydantic定义参数模型
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    num_results: int = Field(default=5, description="返回结果数量")

# 使用装饰器定义工具
@tool("search_web", args_schema=SearchInput)
def search_web(query: str, num_results: int = 5) -> str:
    """
    搜索互联网获取最新信息。
    当需要查询实时数据、新闻、具体事实时使用此工具。
    """
    # 实际搜索逻辑
    results = perform_search(query, num_results)
    return format_results(results)

# 使用工具
from langchain.agents import create_openai_functions_agent
from langchain_openai import ChatOpenAI

tools = [search_web]
llm = ChatOpenAI(model="gpt-4-turbo")
agent = create_openai_functions_agent(llm, tools)