工具调用：Function Calling 实战

# 大模型的三个根本性局限
# 1. 知识截止：无法获取训练后的新信息
print(model.predict("2026年世界杯冠军是谁？"))
# → 模型不知道，因为训练数据只到2024年

# 2. 无法执行操作：不能实际发邮件
print(model.predict("帮我给张三发一封邮件"))
# → 模型只能生成邮件草稿，无法真正发送

# 3. 数学计算不准
print(model.predict("2847 × 9361 = ?"))
# → 经常给出错误答案，不如计算器可靠

# Function Calling 的解决方案：让模型决定调用工具
def handle_user_query(user_input: str) -> str:
    """增强型对话处理：模型决定何时调用工具"""
    # 第一步：让模型判断是否需要工具调用
    response = llm.generate_with_tools(
        prompt=user_input,
        tools=[get_weather, send_email, calculate]
    )
    
    if response.tool_call:
        # 第二步：执行工具获取真实结果
        result = execute_tool(response.tool_call)
        # 第三步：将结果反馈给模型生成最终回答
        return llm.generate(user_input, tool_result=result)
    else:
        return response.text  # 普通对话直接回复

import openai

client = openai.OpenAI()

# 第一阶段：定义工具并发起请求
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    }]
)

# 第二阶段：解析模型的调用决策
tool_call = response.choices[0].message.tool_calls[0]
func_name = tool_call.function.name  # "get_weather"
func_args = json.loads(tool_call.function.arguments)
# → {"city": "北京", "unit": "celsius"}

# 第三阶段：执行工具并将结果注入对话
weather_result = get_weather_api(**func_args)
messages = [
    {"role": "user", "content": "北京今天天气怎么样？"},
    response.choices[0].message,  # 模型的tool_call响应
    {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(weather_result)
    }
]

# 让模型基于工具结果生成最终回答
final = client.chat.completions.create(
    model="gpt-4o", messages=messages
)
print(final.choices[0].message.content)

// 好的工具定义示例
{
  "name": "search_database",
  "description": "从用户数据库中查询信息，适用于需要检索用户资料、订单记录或统计数据的场景",
  "parameters": {
    "type": "object",
    "properties": {
      "query_type": {
        "type": "string",
        "enum": ["user_profile", "order_history", "statistics"],
        "description": "查询类型：user_profile查用户资料，order_history查订单，statistics查统计数据"
      },
      "user_id": {
        "type": "string",
        "description": "用户ID，格式为UUID，如 '550e8400-e29b-41d4-a716-446655440000'"
      },
      "date_range": {
        "type": "string",
        "description": "日期范围，格式 YYYY-MM-DD 至 YYYY-MM-DD，如 '2024-01-01 至 2024-12-31'"
      }
    },
    "required": ["query_type", "user_id"]
  }
}

# 使用 Pydantic 定义工具 Schema（推荐做法）
from pydantic import BaseModel, Field
from enum import Enum

class QueryType(str, Enum):
    USER_PROFILE = "user_profile"
    ORDER_HISTORY = "order_history"
    STATISTICS = "statistics"

class SearchDatabaseArgs(BaseModel):
    query_type: QueryType = Field(
        description="查询类型：user_profile查用户资料，order_history查订单，statistics查统计数据"
    )
    user_id: str = Field(
        description="用户ID，格式为UUID"
    )
    date_range: str | None = Field(
        default=None,
        description="日期范围，格式 YYYY-MM-DD 至 YYYY-MM-DD"
    )

# 自动转换为 OpenAI 格式
schema = SearchDatabaseArgs.model_json_schema()
tool_def = {
    "type": "function",
    "function": {
        "name": "search_database",
        "description": "从用户数据库中查询信息",
        "parameters": schema
    }
}

# 多工具定义与路由示例
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市当前的实时天气状况。不适用于查询历史天气或未来天气预报。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather_forecast",
            "description": "获取指定城市未来7天的天气预报。不适用于查询当前实时天气。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "days": {"type": "integer", "description": "预报天数，1-7"},
                },
                "required": ["city", "days"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_historical_weather",
            "description": "查询指定日期的历史天气记录。不适用于查询当前或未来天气。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "date": {"type": "string", "description": "日期，格式YYYY-MM-DD"},
                },
                "required": ["city", "date"]
            }
        }
    }
]

# 工具路由质量检查
def evaluate_tool_selection(user_input: str, 
                            selected_tool: str,
                            expected_tool: str) -> dict:
    """评估模型的工具选择是否正确"""
    return {
        "input": user_input,
        "selected": selected_tool,
        "expected": expected_tool,
        "correct": selected_tool == expected_tool,
    }

# 测试用例
test_cases = [
    ("北京现在多少度？", "get_current_weather"),
    ("下周去上海要带伞吗", "get_weather_forecast"),
    ("去年今天北京下雪了吗", "get_historical_weather"),
    ("深圳明天和后天天气", "get_weather_forecast"),
]

for query, expected in test_cases:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
        tools=tools
    )
    selected = response.choices[0].message.tool_calls[0].function.name
    result = evaluate_tool_selection(query, selected, expected)
    print(f"{'✅' if result['correct'] else '❌'} {query} → {selected}")

import requests
from typing import Any

def execute_tool_call(tool_call) -> str:
    """执行工具调用并返回结果字符串"""
    func_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    
    # 参数验证
    if func_name == "get_weather":
        city = args.get("city", "")
        if not city:
            return json.dumps({"error": "缺少城市参数"})
        
        # 执行 + 错误处理
        try:
            resp = requests.get(
                f"https://api.weather.com/v1/{city}",
                timeout=5
            )
            resp.raise_for_status()
            data = resp.json()
            # 结果格式化：只保留关键字段
            return json.dumps({
                "temperature": data["temp"],
                "description": data["description"],
                "humidity": data["humidity"]
            }, ensure_ascii=False)
        except requests.Timeout:
            return json.dumps({"error": "天气服务超时，请稍后重试"})
        except requests.HTTPError as e:
            return json.dumps({"error": f"天气服务错误: {e}"})

# 完整的多轮工具调用循环
def chat_with_tools(user_message: str, 
                    tools: list,
                    tool_registry: dict,
                    max_tool_calls: int = 5) -> str:
    """支持多轮工具调用的完整对话循环"""
    messages = [{"role": "user", "content": user_message}]
    
    for _ in range(max_tool_calls):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools
        )
        msg = response.choices[0].message
        
        if not msg.tool_calls:
            return msg.content  # 模型决定不再调用工具
        
        # 执行所有工具调用
        messages.append(msg)
        for tc in msg.tool_calls:
            func_name = tc.function.name
            if func_name in tool_registry:
                result = tool_registry[func_name](tc)
            else:
                result = json.dumps({"error": f"未知工具: {func_name}"})
            
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result
            })
    
    return "已达到最大工具调用次数"

# 安全工具执行器：参数校验 + 权限控制 + 审计日志
import logging
from functools import wraps

logger = logging.getLogger("tool_executor")

def safe_tool_executor(func):
    """装饰器：为工具函数添加安全防护"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        tool_name = func.__name__
        logger.info(f"[TOOL_CALL] {tool_name} args={args} kwargs={kwargs}")
        
        # 参数校验
        validated = validate_params(tool_name, args, kwargs)
        if not validated["valid"]:
            logger.warning(f"[BLOCKED] {tool_name}: {validated['reason']}")
            return {"error": f"参数校验失败: {validated['reason']}"}
        
        # 权限检查
        if not check_permission(tool_name, kwargs.get("user_id")):
            logger.warning(f"[DENIED] {tool_name} user={kwargs.get('user_id')}")
            return {"error": "权限不足，无法执行此操作"}
        
        # 敏感操作需要二次确认
        if is_sensitive_operation(tool_name):
            if not request_confirmation(tool_name, kwargs):
                return {"error": "操作已取消：需要用户确认"}
        
        try:
            result = func(*args, **kwargs)
            logger.info(f"[SUCCESS] {tool_name}")
            return result
        except Exception as e:
            logger.error(f"[ERROR] {tool_name}: {e}")
            return {"error": "工具执行失败"}
    
    return wrapper

# 数据脱敏：根据用户权限过滤工具返回结果
def sanitize_tool_result(tool_name: str, 
                         result: dict, 
                         user_role: str) -> dict:
    """根据用户角色脱敏工具返回数据"""
    # 定义每个工具的字段权限级别
    field_permissions = {
        "get_user_profile": {
            "name": "all",
            "email": "authenticated",
            "phone": "admin",
            "address": "admin",
            "id_number": "admin",
            "password_hash": "never",  # 永远不返回
        },
        "search_database": {
            "order_id": "all",
            "amount": "authenticated",
            "payment_method": "admin",
            "card_last_four": "admin",
        }
    }
    
    if tool_name not in field_permissions:
        return result  # 未定义的工具不过滤
    
    rules = field_permissions[tool_name]
    sanitized = {}
    
    for field, value in result.items():
        required_role = rules.get(field, "never")
        role_levels = ["all", "authenticated", "admin"]
        
        if user_role == "never":
            continue  # 永远不返回
        if required_role == "all":
            sanitized[field] = value
        elif role_levels.index(user_role) >= role_levels.index(required_role):
            sanitized[field] = value
    
    return sanitized

# ===== 工具定义 =====
WEATHER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的当前天气状况",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称，如北京、上海"}
            },
            "required": ["city"]
        }
    }
}

CALCULATOR_TOOL = {
    "type": "function",
    "function": {
        "name": "calculate",
        "description": "执行数学计算，支持加减乘除、幂运算、开方",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string", 
                    "description": "数学表达式，如 '2 + 2'、'3 ** 4'、'sqrt(144)'"
                }
            },
            "required": ["expression"]
        }
    }
}

SEARCH_TOOL = {
    "type": "function",
    "function": {
        "name": "web_search",
        "description": "搜索互联网获取最新信息，适用于查询实时新闻、最新事件或训练数据之外的知识",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"},
                "num_results": {"type": "integer", "description": "返回结果数量，默认5"}
            },
            "required": ["query"]
        }
    }
}

ALL_TOOLS = [WEATHER_TOOL, CALCULATOR_TOOL, SEARCH_TOOL]

# ===== 工具实现 + Agent 核心 =====
import openai
import requests
import math

client = openai.OpenAI()

TOOL_REGISTRY = {
    "get_weather": lambda args: _get_weather(args["city"]),
    "calculate": lambda args: {"result": _calc(args["expression"])},
    "web_search": lambda args: _search(args["query"], args.get("num_results", 5)),
}

def _get_weather(city: str) -> dict:
    resp = requests.get(f"https://wttr.in/{city}?format=j1", timeout=5)
    data = resp.json()["current_condition"][0]
    return {"city": city, "temp": data["temp_C"], "desc": data["weatherDesc"][0]["value"]}

def _calc(expr: str) -> str:
    allowed = set("0123456789+-*/.() ")
    if not all(c in allowed for c in expr):
        return "错误：表达式包含非法字符"
    return str(eval(expr, {"__builtins__": {}, "sqrt": math.sqrt, "pow": pow}))

def _search(query: str, n: int = 5) -> list:
    resp = requests.get(f"https://api.duckduckgo.com/?q={query}&format=json")
    return [{"title": r["Text"], "url": r["FirstURL"]} for r in resp.json().get("Results", [])[:n]]

# Agent 协调层
def agent_run(user_input: str, max_rounds: int = 5) -> str:
    messages = [{"role": "user", "content": user_input}]
    for _ in range(max_rounds):
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages, tools=ALL_TOOLS
        )
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)
        for tc in msg.tool_calls:
            result = TOOL_REGISTRY[tc.function.name](
                json.loads(tc.function.arguments)
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
    return "已达到最大调用次数"

# 测试
print(agent_run("北京天气怎么样？"))
print(agent_run("2的32次方等于多少？"))
print(agent_run("今天有什么重要国际新闻？"))