工具调用 Tool Use 深度

工具调用的工作原理

工具调用（Tool Use / Function Calling）本质上是 LLM 与外部系统交互的协议。LLM 不能直接执行代码，但它可以输出结构化的"调用意图"，由宿主程序代为执行，再将结果反馈给 LLM。

工具定义规范

LangChain 提供了多种工具定义方式，推荐使用 @tool 装饰器配合 Pydantic 模型，以获得最佳的类型安全和文档生成：

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Optional, Literal

# ── 方式1：简单函数 + docstring（推荐用于简单工具）────────
@tool
def get_word_count(text: str) -> int:
    """统计文本中的单词数量。

    Args:
        text: 要统计的文本内容
    Returns:
        单词数量
    """
    return len(text.split())

# ── 方式2：Pydantic Schema（推荐用于复杂参数）────────────
class WebSearchInput(BaseModel):
    query: str = Field(description="搜索关键词，使用英文可获得更好结果")
    num_results: int = Field(default=5, ge=1, le=20, description="返回结果数量")
    time_range: Literal["day", "week", "month", "year"] = Field(
        default="week",
        description="时间范围过滤"
    )

@tool("web_search", args_schema=WebSearchInput)
def search_web(query: str, num_results: int = 5,
               time_range: str = "week") -> str:
    """在网络上搜索最新信息，适用于需要实时数据的问题。"""
    # 实际使用 Tavily API 或 SerpAPI
    return f"[搜索结果：{query}，{num_results}条，时间：{time_range}]"

# 查看工具的 JSON Schema（会被发送给 LLM）
print(search_web.tool_call_schema.schema())

工具调用的完整实现

下面展示如何手动处理工具调用循环，帮助你理解框架内部的工作原理：

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage, AIMessage, ToolMessage, SystemMessage
)
import json

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [search_web, get_word_count]
llm_with_tools = llm.bind_tools(tools)

# 工具名 → 函数的映射表
tool_map = {t.name: t for t in tools}

def run_agent_loop(user_input: str, max_iterations: int = 10) -> str:
    messages = [
        SystemMessage(content="你是一个智能助手，可以使用工具来回答问题。"),
        HumanMessage(content=user_input),
    ]

    for i in range(max_iterations):
        response = llm_with_tools.invoke(messages)
        messages.append(response)

        # 没有工具调用 → Agent 完成
        if not response.tool_calls:
            return response.content

        # 执行每个工具调用
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_id   = tool_call["id"]

            print(f"  → 调用工具: {tool_name}({tool_args})")

            try:
                tool_fn = tool_map[tool_name]
                result = tool_fn.invoke(tool_args)
            except KeyError:
                result = f"错误：工具 '{tool_name}' 不存在"
            except Exception as e:
                result = f"工具执行失败：{e}"

            # 将工具结果作为 ToolMessage 追加
            messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_id
            ))

    return "已达到最大迭代次数，任务未完成。"

result = run_agent_loop("搜索 LangGraph 0.2 的新特性，并统计摘要字数")
print(result)

结构化输出（Structured Output）

有时你需要 LLM 返回特定格式的数据（如 JSON 对象），而不是工具调用。with_structured_output 是最佳方案：

from pydantic import BaseModel, Field
from typing import List

class ResearchReport(BaseModel):
    """研究报告结构"""
    title: str = Field(description="报告标题")
    summary: str = Field(description="200字以内的执行摘要")
    key_findings: List[str] = Field(description="3-5条核心发现")
    confidence: float = Field(ge=0, le=1, description="置信度 0-1")
    sources: List[str] = Field(description="参考来源URL列表")

# 绑定结构化输出 Schema
structured_llm = llm.with_structured_output(ResearchReport)

report = structured_llm.invoke(
    "请总结 2025 年 AI Agent 框架的发展趋势"
)

print(report.title)          # str
print(report.confidence)     # float
print(report.key_findings)   # list[str]

# 在 Agent 中作为"路由判断"工具使用
class RouteDecision(BaseModel):
    next_action: Literal["search", "analyze", "generate", "finish"]
    reasoning: str
    priority: int = Field(ge=1, le=5)

router_llm = llm.with_structured_output(RouteDecision)

def structured_router(state):
    decision = router_llm.invoke([
        SystemMessage(content="分析当前任务状态，决定下一步行动。"),
        *state["messages"]
    ])
    return decision.next_action

工具选择策略

错误处理最佳实践

from langchain_core.tools import tool, ToolException
from tenacity import retry, stop_after_attempt, wait_exponential
import httpx

# ── 方案1：工具内部处理，返回错误信息给 LLM ───────────────
@tool
def robust_search(query: str) -> str:
    """搜索网络信息。如果失败，会返回错误描述供 Agent 决策。"""
    try:
        result = call_search_api(query)
        return result
    except httpx.TimeoutException:
        return "搜索超时，建议尝试更简短的关键词或稍后重试"
    except httpx.HTTPStatusError as e:
        return f"搜索API返回错误 {e.response.status_code}，请尝试其他方式"
    except Exception as e:
        return f"搜索失败：{type(e).__name__}: {str(e)}"

# ── 方案2：自动重试（适合网络不稳定的工具）───────────────
@tool
@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=2, max=10))
def reliable_api_call(endpoint: str, params: str) -> str:
    """调用外部 API，自动重试最多3次。"""
    response = httpx.get(endpoint, params=json.loads(params), timeout=10)
    response.raise_for_status()
    return response.text

# ── 方案3：ToolNode 级别的统一错误处理 ───────────────────
from langgraph.prebuilt import ToolNode

# handle_tool_errors=True：工具异常会被捕获并作为 ToolMessage 返回
# 让 LLM 能看到错误并决定如何处理（重试/换方法/告知用户）
tool_node = ToolNode(tools, handle_tool_errors=True)

# 自定义错误处理器
def custom_error_handler(error: Exception) -> str:
    error_type = type(error).__name__
    if isinstance(error, httpx.TimeoutException):
        return "工具调用超时(10s)，请简化请求或换一种方式完成任务"
    elif isinstance(error, ValueError):
        return f"参数格式错误：{error}，请检查参数类型"
    return f"{error_type}：{str(error)[:200]}"

tool_node_robust = ToolNode(tools, handle_tool_errors=custom_error_handler)

并行工具调用

GPT-4o 和 Claude 3.5 支持一次输出多个工具调用。LangGraph 的 ToolNode 会自动并行执行：

import asyncio
from langchain_core.tools import tool

# 声明异步工具（支持并发执行）
@tool
async def async_search(query: str) -> str:
    """异步搜索工具，支持并发调用。"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.tavily.com/search",
            params={"query": query},
            timeout=10
        )
        return resp.json()["results"][0]["content"]

# ToolNode 自动并行执行多个工具调用
# 当 LLM 输出：[search("A"), search("B"), calculate("1+1")]
# ToolNode 会并发执行三个工具，汇总结果后返回
tool_node = ToolNode([async_search, calculate])

# 运行时使用异步版本
async def run():
    result = await graph.ainvoke({...})
    return result