AI 大模型与 Agent 开发实战

Appearance

13.6 工具调用代码实战 ​

生活类比：你让助理帮你订机票。助理不是自己有飞机，而是会"打电话给航空公司"（调用工具）。你不需要告诉助理具体怎么打电话，只要说"订一张明天去北京的机票"，助理自己决定该打哪个电话、说什么。Function Calling 就是让 Agent 学会"打电话"的能力。

本节从 Function Calling 的完整实现讲到 MCP Server 开发，覆盖工具定义、调用循环、错误处理的全流程。

13.6.1 Function Calling 完整流程 ​

Function Calling 的核心是"两轮对话"：

第一轮：用户提问 → LLM 决定调哪个工具 → 返回工具调用请求
第二轮：执行工具 → 把结果交给 LLM → LLM 生成最终回答

下面是完整代码实现：

python
import json
from openai import OpenAI

client = OpenAI()                    # 初始化同步客户端

# ========== 第一步：定义工具的 JSON Schema ==========
# 这段 schema 告诉 LLM"你有哪些工具可用，每个工具怎么调"
tools = [
    {
        "type": "function",           # 工具类型：函数调用
        "function": {
            "name": "get_weather",   # 工具名称
            "description": "获取指定城市的天气信息",  # 描述越清晰，模型选择越准确
            "parameters": {           # 参数定义，用 JSON Schema 格式
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],  # 限定可选值
                        "description": "温度单位"
                    }
                },
                "required": ["city"]  # city 是必填，unit 可选
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索网页信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "num_results": {"type": "integer", "description": "返回结果数量", "default": 5}
                },
                "required": ["query"]
            }
        }
    }
]

# ========== 第二步：实现工具函数 ==========
def get_weather(city: str, unit: str = "celsius") -> str:
    """模拟天气查询（实际应用中对接天气 API）"""
    weather_data = {
        "北京": {"celsius": "25°C 晴", "fahrenheit": "77°F Sunny"},
        "上海": {"celsius": "28°C 多云", "fahrenheit": "82°F Cloudy"},
    }
    return weather_data.get(city, {}).get(unit, f"未找到{city}的天气信息")


def search_web(query: str, num_results: int = 5) -> str:
    """模拟网页搜索（实际应用中对接搜索 API）"""
    return f"搜索'{query}'的结果：找到了{num_results}条相关信息"


# 工具名称到函数的映射表
available_functions = {
    "get_weather": get_weather,       # 字符串名 → 实际函数
    "search_web": search_web
}


# ========== 第三步：Agent 主循环 ==========
def run_agent_with_tools(user_query: str) -> str:
    """运行带工具调用的 Agent"""
    messages = [{"role": "user", "content": user_query}]  # 初始消息只有用户输入

    # --- 第一轮：让模型决定是否调用工具 ---
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,                  # 传入工具定义
        tool_choice="auto"             # auto = 模型自己决定调不调工具
    )

    response_message = response.choices[0].message
    messages.append(response_message)  # 把模型的回复加入消息历史

    # --- 检查模型是否要调用工具 ---
    if response_message.tool_calls:
        # 遍历所有工具调用请求
        for tool_call in response_message.tool_calls:
            function_name = tool_call.function.name          # 要调的函数名
            function_args = json.loads(tool_call.function.arguments)  # 参数（JSON 字符串→字典）

            print(f"调用工具: {function_name}({function_args})")

            # 从映射表找到对应函数并执行
            function_to_call = available_functions[function_name]
            function_result = function_to_call(**function_args)  # ** 展开参数

            # 把工具执行结果加入消息历史
            messages.append({
                "role": "tool",                         # 角色是 tool（工具）
                "tool_call_id": tool_call.id,           # 关联到对应的调用 ID
                "name": function_name,                  # 工具名
                "content": str(function_result)          # 执行结果
            })

        # --- 第二轮：让模型基于工具结果生成最终回答 ---
        final_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages                            # 包含了工具结果的完整历史
        )
        return final_response.choices[0].message.content

    # 如果模型没有调用工具，直接返回它的回复
    return response_message.content


# ========== 测试 ==========
print(run_agent_with_tools("北京今天天气怎么样？"))   # 会调用 get_weather
print("---")
print(run_agent_with_tools("搜索 Python 最新版本"))    # 会调用 search_web

13.6.2 多轮工具调用 ​

有时候一个任务需要连续调用多个工具。例如"北京天气怎么样？顺便搜一下附近有什么好吃的"——需要先调天气工具，再调搜索工具。上面的代码已经支持这种情况：模型可以在一次回复中返回多个 tool_calls，循环会逐个执行。

13.6.3 MCP Server 开发：标准化工具服务 ​

生活类比：Function Calling 是"自己造工具自己用"，MCP 是"造一个标准化的工具箱，任何人都能来用"。MCP 就像 USB 标准——之前每个设备都要专用接口，现在统一一个 USB-C。

MCP（Model Context Protocol）把工具定义为独立的服务，任何支持 MCP 的客户端都能调用。

python
# mcp_server.py
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.types import Tool, TextContent
import mcp.server.stdio

# 创建 MCP Server 实例
server = Server("my-agent-tools")     # 服务名称


# 注册工具列表——告诉客户端"我有哪些工具"
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",          # 工具名
            description="执行数学计算",
            inputSchema={             # 参数 schema（与 Function Calling 相同格式）
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        ),
        Tool(
            name="read_file",
            description="读取文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"]
            }
        )
    ]


# 注册工具执行逻辑——客户端调用工具时执行这里
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """根据工具名执行对应的逻辑"""
    if name == "calculate":
        expression = arguments["expression"]
        try:
            result = eval(expression)               # 计算表达式（生产环境慎用 eval！）
            return [TextContent(type="text", text=f"计算结果: {result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"计算错误: {str(e)}")]

    elif name == "read_file":
        path = arguments["path"]
        try:
            with open(path, "r") as f:
                content = f.read(1000)              # 最多读 1000 字符
            return [TextContent(type="text", text=content)]
        except Exception as e:
            return [TextContent(type="text", text=f"读取失败: {str(e)}")]

    return [TextContent(type="text", text=f"未知工具: {name}")]


# 启动 Server——通过 stdio 通信（标准输入输出）
async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,                            # 读取客户端请求
            write_stream,                           # 写入响应
            InitializationCapabilities(
                sampling={},
                experimental={},
                roots={}
            ),
            NotificationOptions()
        )

if __name__ == "__main__":
    asyncio.run(main())

13.6.4 MCP Client 调用 ​

python
# mcp_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def call_mcp_tool():
    """调用 MCP Server 的工具"""
    # 定义如何启动 Server 进程
    server_params = StdioServerParameters(
        command="python",            # 启动命令
        args=["mcp_server.py"]       # 启动参数（Server 脚本路径）
    )

    # 通过 stdio 连接 Server
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()             # 初始化连接

            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具:", [t.name for t in tools.tools])

            # 调用 calculate 工具
            result = await session.call_tool("calculate", {"expression": "2 + 3 * 4"})
            print("计算:", result.content[0].text)  # 输出: 计算结果: 14

            # 调用 read_file 工具
            result = await session.call_tool("read_file", {"path": "/tmp/test.txt"})
            print("文件:", result.content[0].text)

asyncio.run(call_mcp_tool())

13.6.5 健壮的工具执行器：超时 + 重试 ​

生产环境中工具调用会失败（网络超时、服务不可用等），需要重试机制：

python
import asyncio
from typing import Any

class RobustToolExecutor:
    """
    健壮的工具执行器：带超时和重试

    就像一个靠谱的快递系统：
    - 超时：30 分钟没送达就标记失败
    - 重试：送失败了明天再送，最多试 3 次
    - 退避：每次重试间隔越来越长
    """

    def __init__(self, max_retries: int = 3, timeout: float = 30.0):
        self.max_retries = max_retries     # 最大重试次数
        self.timeout = timeout            # 单次超时时间（秒）

    async def execute(self, tool_name: str, tool_fn: callable,
                      args: dict) -> dict:
        """执行工具，带重试和超时"""
        for attempt in range(self.max_retries):
            try:
                # asyncio.to_thread：把同步函数放到线程池执行（不阻塞事件循环）
                # asyncio.wait_for：加超时限制
                result = await asyncio.wait_for(
                    asyncio.to_thread(tool_fn, **args),
                    timeout=self.timeout
                )
                return {
                    "success": True,
                    "result": result,
                    "attempts": attempt + 1          # 第几次尝试成功的
                }
            except asyncio.TimeoutError:              # 超时
                if attempt == self.max_retries - 1:   # 最后一次了，不再重试
                    return {"success": False, "error": "timeout", "attempts": self.max_retries}
                print(f"  [{tool_name}] 超时，重试中...")
            except Exception as e:                    # 其他异常
                if attempt == self.max_retries - 1:
                    return {"success": False, "error": str(e), "attempts": self.max_retries}
                print(f"  [{tool_name}] 错误: {e}，重试中...")

            # 指数退避等待
            await asyncio.sleep(2 ** attempt)          # 1, 2, 4 秒...

        return {"success": False, "error": "max_retries_exceeded"}

# 使用示例
# executor = RobustToolExecutor(max_retries=3, timeout=30)
# result = await executor.execute("get_weather", get_weather, {"city": "北京"})
# print(result)  # {"success": True, "result": "25°C 晴", "attempts": 1}

13.6.6 工具调用常见问题排查 ​

问题可能原因解决方案
模型不调用工具工具描述太模糊在 description 中写清楚"什么情况下用这个工具"
参数类型错误Schema 定义与实际不符确认 type、enum、required 与函数签名一致
工具执行超时外部服务慢设置合理 timeout，加重试机制
循环调用模型反复调用同一工具限制最大工具调用次数（如 5 次）
JSON 解析失败模型返回的 arguments 格式不对用 try-except 包裹 json.loads

常见误区 ​

  1. "工具描述不重要"：非常重要！模型是靠 description 决定"什么时候用哪个工具"的。描述不清会导致选错工具或不调用工具。
  2. "eval 是安全的"：上面示例用了 eval 做计算，这在生产环境是严重安全漏洞——攻击者可以注入恶意表达式。应使用 ast.literal_eval 或专门的数学表达式库。
  3. "MCP 就是另一种 Function Calling"：不完全一样。MCP 是协议层面的标准化，目的是让不同框架的 Agent 共享同一套工具。Function Calling 是模型层面的能力。
  4. "工具越多越好"：工具太多会导致模型选择困难，且每次调用都要把所有工具定义发给模型，增加 token 消耗。5-10 个精心设计的工具通常比 50 个粗糙的工具好用。

本节小结 ​

要点说明
Function Calling定义 tools schema，模型自动决定调用哪个工具
工具调用循环调用 → 获取结果 → 再次调用模型 → 最终回答
MCP Server标准化工具服务，跨框架复用
MCP Client通过 stdio 或 SSE 连接 MCP Server
错误处理超时、重试、指数退避
安全注意eval 等危险函数要替换，工具执行要沙箱化

核心心法：工具是 Agent 的"手"。定义工具时多花 5 分钟写清楚 description 和 parameters，能省下后面 5 小时的调试时间。

延伸阅读 ​

  1. OpenAI Function Calling Guide：https://platform.openai.com/docs/guides/function-calling
  2. MCP Protocol Specification：https://modelcontextprotocol.io
  3. Anthropic Tool Use：https://docs.anthropic.com/en/docs/build-with-claude/tool-use
京ICP备17024326号-1Copyright © 2025 AI 大模型与 Agent 开发实战