Skip to content

阶段 2：工具调用与函数式 Agent

📖 本阶段目标

  • 理解 Function Calling 的工作原理
  • 学会将业务 API、本地模块包装为 LLM 可调用的工具
  • 掌握工具调用的错误处理和重试机制
  • 实践构建多工具聚合的 Agent

🔧 Function Calling 核心概念

什么是 Function Calling？

Function Calling 允许 LLM 在需要时调用外部函数/API，从而： - 获取实时数据（天气、股票、新闻等） - 执行具体操作（发送邮件、创建任务等） - 访问数据库或内部系统 - 扩展 LLM 的能力边界

工作流程

用户输入："北京明天的天气怎么样？"
    ↓
LLM 理解意图，决定调用工具
    ↓
返回 Function Call 请求：
{
  "name": "get_weather",
  "arguments": {
    "city": "北京",
    "date": "明天"
  }
}
    ↓
系统执行实际的 API 调用
    ↓
将结果返回给 LLM
    ↓
LLM 生成自然语言回复：
"北京明天晴，最高温度25°C，最低温度15°C..."

🛠️ 实现 Function Calling

1. 基础示例：天气查询

# examples/stage2/weather_agent.py
import json
from openai import OpenAI

client = OpenAI()

# 定义工具函数
def get_current_weather(location, unit="celsius"):
    """获取指定地点的天气（模拟）"""
    # 实际应用中这里会调用真实的天气API
    weather_data = {
        "北京": {"temperature": 22, "condition": "晴", "humidity": 45},
        "上海": {"temperature": 26, "condition": "多云", "humidity": 60},
        "深圳": {"temperature": 28, "condition": "阴", "humidity": 75},
    }

    weather = weather_data.get(location, {
        "temperature": 20,
        "condition": "未知",
        "humidity": 50
    })

    return json.dumps({
        "location": location,
        "temperature": weather["temperature"],
        "unit": unit,
        "condition": weather["condition"],
        "humidity": weather["humidity"]
    }, ensure_ascii=False)

# 工具定义（告诉 LLM 有哪些工具可用）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称，例如：北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

def run_conversation(user_message):
    """运行对话，处理函数调用"""
    messages = [
        {"role": "user", "content": user_message}
    ]

    # 第一次调用：让 LLM 决定是否需要调用函数
    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # 让模型自动决定
    )

    response_message = response.choices[0].message
    tool_calls = response_message.tool_calls

    # 如果 LLM 决定调用函数
    if tool_calls:
        messages.append(response_message)

        # 执行所有函数调用
        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            # 调用实际的函数
            if function_name == "get_current_weather":
                function_response = get_current_weather(**function_args)

            # 将函数结果添加到消息历史
            messages.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": function_response
            })

        # 第二次调用：让 LLM 根据函数结果生成回复
        second_response = client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=messages
        )

        return second_response.choices[0].message.content

    # 如果不需要调用函数，直接返回
    return response_message.content

# 测试
if __name__ == "__main__":
    result = run_conversation("北京现在的天气怎么样？")
    print(result)

2. 多工具集成：待办事项管理

# examples/stage2/todo_agent.py
import json
from datetime import datetime
from openai import OpenAI

client = OpenAI()

# 简单的内存存储（实际应用应使用数据库）
todo_list = []

def add_todo(task, priority="medium", due_date=None):
    """添加待办事项"""
    todo_item = {
        "id": len(todo_list) + 1,
        "task": task,
        "priority": priority,
        "due_date": due_date,
        "completed": False,
        "created_at": datetime.now().isoformat()
    }
    todo_list.append(todo_item)
    return json.dumps({"success": True, "item": todo_item}, ensure_ascii=False)

def list_todos(filter_by=None):
    """列出待办事项"""
    if filter_by == "completed":
        items = [t for t in todo_list if t["completed"]]
    elif filter_by == "pending":
        items = [t for t in todo_list if not t["completed"]]
    else:
        items = todo_list

    return json.dumps({"count": len(items), "items": items}, ensure_ascii=False)

def complete_todo(task_id):
    """标记待办事项为完成"""
    for item in todo_list:
        if item["id"] == task_id:
            item["completed"] = True
            return json.dumps({"success": True, "item": item}, ensure_ascii=False)

    return json.dumps({"success": False, "error": "任务不存在"}, ensure_ascii=False)

def delete_todo(task_id):
    """删除待办事项"""
    global todo_list
    original_len = len(todo_list)
    todo_list = [t for t in todo_list if t["id"] != task_id]

    success = len(todo_list) < original_len
    return json.dumps({"success": success}, ensure_ascii=False)

# 定义所有工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "add_todo",
            "description": "添加新的待办事项",
            "parameters": {
                "type": "object",
                "properties": {
                    "task": {
                        "type": "string",
                        "description": "待办任务的描述"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                        "description": "任务优先级"
                    },
                    "due_date": {
                        "type": "string",
                        "description": "截止日期，格式：YYYY-MM-DD"
                    }
                },
                "required": ["task"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "list_todos",
            "description": "列出待办事项",
            "parameters": {
                "type": "object",
                "properties": {
                    "filter_by": {
                        "type": "string",
                        "enum": ["all", "completed", "pending"],
                        "description": "筛选条件"
                    }
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "complete_todo",
            "description": "标记待办事项为已完成",
            "parameters": {
                "type": "object",
                "properties": {
                    "task_id": {
                        "type": "integer",
                        "description": "任务ID"
                    }
                },
                "required": ["task_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "delete_todo",
            "description": "删除待办事项",
            "parameters": {
                "type": "object",
                "properties": {
                    "task_id": {
                        "type": "integer",
                        "description": "任务ID"
                    }
                },
                "required": ["task_id"]
            }
        }
    }
]

# 函数映射
available_functions = {
    "add_todo": add_todo,
    "list_todos": list_todos,
    "complete_todo": complete_todo,
    "delete_todo": delete_todo
}

def run_agent(user_input):
    """运行待办事项 Agent"""
    messages = [
        {
            "role": "system",
            "content": "你是一个待办事项管理助手，帮助用户管理他们的任务。"
        },
        {"role": "user", "content": user_input}
    ]

    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )

    response_message = response.choices[0].message
    tool_calls = response_message.tool_calls

    if tool_calls:
        messages.append(response_message)

        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            # 调用对应的函数
            function_to_call = available_functions[function_name]
            function_response = function_to_call(**function_args)

            messages.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": function_response
            })

        second_response = client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=messages
        )

        return second_response.choices[0].message.content

    return response_message.content

# 测试
if __name__ == "__main__":
    # 添加任务
    print(run_agent("帮我添加一个高优先级的任务：完成项目报告，截止日期是2024-12-31"))

    # 列出任务
    print(run_agent("给我看看所有待办任务"))

    # 完成任务
    print(run_agent("把第1个任务标记为完成"))

3. 综合项目：多工具聚合 Agent

# examples/stage2/multi_tool_agent.py
import json
import requests
from datetime import datetime
from openai import OpenAI

client = OpenAI()

class MultiToolAgent:
    """多工具聚合 Agent"""

    def __init__(self):
        self.conversation_history = []
        self.tools = self._define_tools()
        self.functions = {
            "get_weather": self.get_weather,
            "get_exchange_rate": self.get_exchange_rate,
            "calculate": self.calculate,
            "get_current_time": self.get_current_time
        }

    def get_weather(self, city):
        """获取天气（使用真实API或模拟）"""
        # 这里应该调用真实的天气API，如 OpenWeatherMap
        # 为了演示，使用模拟数据
        mock_data = {
            "北京": {"temp": 22, "condition": "晴"},
            "上海": {"temp": 26, "condition": "多云"},
        }
        data = mock_data.get(city, {"temp": 20, "condition": "未知"})
        return json.dumps(data, ensure_ascii=False)

    def get_exchange_rate(self, from_currency, to_currency):
        """获取汇率"""
        # 实际应该调用汇率API
        # 模拟数据
        rates = {
            ("USD", "CNY"): 7.2,
            ("CNY", "USD"): 0.14,
            ("EUR", "CNY"): 7.8,
        }
        rate = rates.get((from_currency, to_currency), 1.0)
        return json.dumps({"from": from_currency, "to": to_currency, "rate": rate})

    def calculate(self, expression):
        """计算数学表达式"""
        try:
            # 注意：eval 有安全风险，生产环境应使用专门的数学解析库
            result = eval(expression, {"__builtins__": {}}, {})
            return json.dumps({"expression": expression, "result": result})
        except Exception as e:
            return json.dumps({"error": str(e)})

    def get_current_time(self, timezone="Asia/Shanghai"):
        """获取当前时间"""
        now = datetime.now()
        return json.dumps({
            "timezone": timezone,
            "datetime": now.isoformat(),
            "formatted": now.strftime("%Y-%m-%d %H:%M:%S")
        })

    def _define_tools(self):
        """定义所有可用工具"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "获取指定城市的天气信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "city": {"type": "string", "description": "城市名称"}
                        },
                        "required": ["city"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_exchange_rate",
                    "description": "获取货币汇率",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "from_currency": {"type": "string", "description": "源货币代码，如USD"},
                            "to_currency": {"type": "string", "description": "目标货币代码，如CNY"}
                        },
                        "required": ["from_currency", "to_currency"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "calculate",
                    "description": "计算数学表达式，支持加减乘除和括号",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {"type": "string", "description": "数学表达式，如 '(10 + 5) * 2'"}
                        },
                        "required": ["expression"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_current_time",
                    "description": "获取当前时间",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "timezone": {"type": "string", "description": "时区，默认Asia/Shanghai"}
                        }
                    }
                }
            }
        ]

    def chat(self, user_input):
        """处理用户输入"""
        self.conversation_history.append({"role": "user", "content": user_input})

        response = client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[
                {"role": "system", "content": "你是一个多功能助手，可以查天气、查汇率、计算和查时间。"},
                *self.conversation_history
            ],
            tools=self.tools,
            tool_choice="auto"
        )

        response_message = response.choices[0].message
        tool_calls = response_message.tool_calls

        if tool_calls:
            self.conversation_history.append(response_message)

            for tool_call in tool_calls:
                function_name = tool_call.function.name
                function_args = json.loads(tool_call.function.arguments)

                # 调用对应函数
                function_response = self.functions[function_name](**function_args)

                self.conversation_history.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": function_name,
                    "content": function_response
                })

            # 获取最终回复
            final_response = client.chat.completions.create(
                model="gpt-4-turbo-preview",
                messages=[
                    {"role": "system", "content": "你是一个多功能助手。"},
                    *self.conversation_history
                ]
            )

            assistant_message = final_response.choices[0].message.content
        else:
            assistant_message = response_message.content

        self.conversation_history.append({"role": "assistant", "content": assistant_message})
        return assistant_message

# 测试
if __name__ == "__main__":
    agent = MultiToolAgent()

    print("=== 多工具 Agent 测试 ===\n")

    # 测试1：天气查询
    print("用户：北京今天天气怎么样？")
    print("助手：", agent.chat("北京今天天气怎么样？"))
    print()

    # 测试2：汇率查询
    print("用户：100美元等于多少人民币？")
    print("助手：", agent.chat("100美元等于多少人民币？"))
    print()

    # 测试3：计算
    print("用户：帮我算一下 (123 + 456) * 2")
    print("助手：", agent.chat("帮我算一下 (123 + 456) * 2"))
    print()

    # 测试4：复杂查询（可能需要多个工具）
    print("用户：现在几点了？北京天气如何？")
    print("助手：", agent.chat("现在几点了？北京天气如何？"))

📋 工具设计最佳实践

1. 工具定义原则

✅ DO（应该）： - 清晰的函数名称（使用动词开头） - 详细的描述说明 - 明确的参数类型和说明 - 合理的必填/可选参数设置

❌ DON'T（不应该）： - 模糊的函数名 - 缺少参数描述 - 参数类型不明确 - 过于复杂的参数结构

2. 错误处理

def safe_tool_call(func, **kwargs):
    """安全的工具调用包装器"""
    try:
        result = func(**kwargs)
        return {
            "success": True,
            "data": result
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "error_type": type(e).__name__
        }

3. 重试机制

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator

@retry_on_failure(max_retries=3)
def call_external_api(url):
    """调用外部API（带重试）"""
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

✅ 阶段完成检查清单

  • [ ] 理解 Function Calling 的工作原理和流程
  • [ ] 能够定义标准的工具函数描述
  • [ ] 掌握单工具和多工具的集成方法
  • [ ] 实现了至少一个实际的工具调用项目
  • [ ] 了解工具调用的错误处理机制
  • [ ] 能够处理多轮对话中的工具调用
  • [ ] 理解工具调用的最佳实践

🎯 下一步

完成本阶段后，进入 阶段 3：任务分解与多步推理 Agent，学习如何让 Agent 自主进行多步推理和任务分解。

💡 小贴士：工具调用是 Agent 的"手和脚"，好的工具设计能让 Agent 更强大。记住：保持工具的单一职责，让每个工具只做好一件事。