Function Calling 与工具使用

# 传统方式：让 LLM 生成代码执行
user_input = "帮我计算 2 + 2"
response = llm.invoke(user_input)
# LLM 可能返回：
# "答案是 4" ✅ 正确
# "让我想想...2 加 2 等于... 5" ❌ 幻觉

# Function Calling：结构化调用
response = llm.invoke(
    user_input,
    tools=[calculate_tool]
)
# LLM 返回：
# {"name": "calculate", "arguments": {"expression": "2 + 2"}}
# 代码执行 calculate("2 + 2") = 4
# LLM 基于结果生成："2 + 2 = 4" ✅ 准确

from openai import OpenAI
import json

client = OpenAI()

# 定义工具函数
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称，如：北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# 发起请求
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=tools
)

# 检查是否需要调用函数
tool_calls = response.choices[0].message.tool_calls

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

        # 执行函数
        if function_name == "get_weather":
            result = get_weather(
                location=function_args.get("location"),
                unit=function_args.get("unit", "celsius")
            )

            # 将结果返回给 LLM
            follow_up = client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {"role": "user", "content": "北京今天天气怎么样？"},
                    response.choices[0].message,
                    {
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result)
                    }
                ]
            )

            print(follow_up.choices[0].message.content)

# ✅ 好的工具定义
good_tool = {
    "name": "search_products",
    "description": "搜索产品目录，根据关键词查找匹配的产品",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词，如：蓝牙耳机、运动鞋"
            },
            "category": {
                "type": "string",
                "description": "产品类别，如：电子产品、服装、食品"
            },
            "min_price": {
                "type": "number",
                "description": "最低价格"
            },
            "max_price": {
                "type": "number",
                "description": "最高价格"
            }
        },
        "required": ["query"]
    }
}

# ❌ 差的工具定义
bad_tool = {
    "name": "search",
    "description": "搜索",  # 太简单
    "parameters": {
        "type": "object",
        "properties": {
            "args": {  # 不明确
                "type": "string"
            }
        }
    }
}

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_time",
            "description": "获取指定时区的当前时间",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "时区，如：Asia/Shanghai"
                    }
                },
                "required": ["timezone"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "convert_currency",
            "description": "货币汇率转换",
            "parameters": {
                "type": "object",
                "properties": {
                    "amount": {"type": "number"},
                    "from": {"type": "string"},
                    "to": {"type": "string"}
                },
                "required": ["amount", "from", "to"]
            }
        }
    }
]

# LLM 可能并行调用多个工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "帮我查一下北京天气、纽约时间，还有100美元换人民币是多少？"}
    ],
    tools=tools
)

# 处理所有工具调用
tool_calls = response.choices[0].message.tool_calls

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

    # 根据名称调用不同函数
    if function_name == "get_weather":
        result = get_weather(**function_args)
    elif function_name == "get_time":
        result = get_time(**function_args)
    elif function_name == "convert_currency":
        result = convert_currency(**function_args)

    tool_results.append({
        "tool_call_id": tool_call.id,
        "role": "tool",
        "content": json.dumps(result)
    })

# 将所有结果返回给 LLM
final_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "帮我查一下北京天气、纽约时间，还有100美元换人民币是多少？"},
        response.choices[0].message,
        *tool_results
    ]
)

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_function

# 定义工具
@tool
def get_weather(location: str, unit: str = "celsius") -> str:
    """获取指定城市的当前天气"""
    # 实际调用天气 API
    return f"{location}: 晴天，25°{unit[0].upper()}"

@tool
def get_time(timezone: str) -> str:
    """获取指定时区的当前时间"""
    # 实际获取时间
    return f"{timezone}: 14:30"

@tool
def convert_currency(amount: float, from_: str, to: str) -> str:
    """货币汇率转换"""
    # 实际调用汇率 API
    rate = 7.2 if to == "CNY" else 1
    return f"{amount} {from_} = {amount * rate:.2f} {to}"

# 绑定工具到 LLM
llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather, get_time, convert_currency]
llm_with_tools = llm.bind_functions(tools)

# 调用
response = llm_with_tools.invoke(
    "帮我查一下北京天气、纽约时间，还有100美元换人民币是多少？"
)

# 处理工具调用
from langchain.tools.render import render_text_description

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

        for tool in tools:
            if tool.name == tool_name:
                result = tool.func(**tool_args)
                print(f"{tool_name}: {result}")

class AgentWithChaining:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}

    def run(self, query: str, max_iterations: int = 5):
        messages = [{"role": "user", "content": query}]

        for iteration in range(max_iterations):
            # 调用 LLM
            response = self.llm.invoke(
                messages,
                tools=[tool["function"] for tool in self.tools.values()]
            )

            messages.append(response.message)

            # 检查是否需要调用工具
            if not response.tool_calls:
                break

            # 执行工具调用
            for tool_call in response.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                if tool_name in self.tools:
                    result = self.tools[tool_name](**tool_args)

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })

        return messages[-1]["content"]

# 使用
agent = AgentWithChaining(llm, tools)
result = agent.run("帮我查北京天气，如果温度超过30度，告诉我需要注意什么")

@tool
def find_restaurants(location: str, cuisine: str = None) -> str:
    """查找餐厅"""
    # 返回餐厅列表
    return "['海底捞(王府井店)', '全聚德(前门店)', '鼎泰丰(国贸店)']"

@tool
def get_restaurant_details(restaurant_name: str) -> str:
    """获取餐厅详细信息"""
    # 返回餐厅详情
    return f"{restaurant_name}: 地址xxx, 电话xxx, 人均150元"

@tool
def make_reservation(restaurant_name: str, time: str, people: int) -> str:
    """预订餐厅"""
    # 调用预订 API
    return f"已预订 {restaurant_name} {time} {people}人位"

# LLM 会自动处理工具依赖
agent = AgentWithChaining(llm, [
    find_restaurants,
    get_restaurant_details,
    make_reservation
])

result = agent.run("帮我在王府井附近找个川菜餐厅，今晚7点2个人")

from functools import wraps
import json

def handle_tool_errors(func):
    """工具错误处理装饰器"""
    @wraps(func)
    def wrapper(**kwargs):
        try:
            return func(**kwargs)
        except ValueError as e:
            return {
                "error": "invalid_arguments",
                "message": str(e)
            }
        except Exception as e:
            return {
                "error": "execution_failed",
                "message": f"工具执行失败: {str(e)}"
            }
    return wrapper

@tool
@handle_tool_errors
def get_weather(location: str) -> str:
    """获取天气信息"""
    if not location:
        raise ValueError("location 参数不能为空")

    # 实际 API 调用
    try:
        response = requests.get(f"https://api.weather.com/{location}")
        return response.json()
    except requests.RequestException as e:
        raise Exception(f"天气 API 调用失败: {e}")

from pydantic import BaseModel, Field, validator

class WeatherParams(BaseModel):
    """天气参数验证"""
    location: str = Field(..., min_length=2, description="城市名称")
    unit: str = Field("celsius", regex="^(celsius|fahrenheit)$")

    @validator('location')
    def location_must_not_be_empty(cls, v):
        if not v or v.strip() == "":
            raise ValueError('location 不能为空')
        return v.strip()

@tool
def get_weather_validated(location: str, unit: str = "celsius") -> str:
    """获取天气信息（带参数验证）"""
    # 验证参数
    params = WeatherParams(location=location, unit=unit)

    # 实际调用
    return f"{params.location}: 晴天，25°{params.unit[0].upper()}"

@tool
def retry_on_failure(func, max_retries=3):
    """工具重试装饰器"""
    @wraps(func)
    def wrapper(**kwargs):
        last_error = None
        for attempt in range(max_retries):
            try:
                return func(**kwargs)
            except Exception as e:
                last_error = e
                if attempt < max_retries - 1:
                    continue
                raise Exception(f"重试 {max_retries} 次后仍失败: {e}")
    return wrapper

@tool
@retry_on_failure
def get_weather_with_retry(location: str) -> str:
    """获取天气信息（带重试）"""
    # 可能失败的操作
    response = requests.get(f"https://api.weather.com/{location}")
    response.raise_for_status()
    return response.json()

from openai import OpenAI
import json
from typing import List, Dict, Any

class CustomerServiceAgent:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.tools = self._init_tools()
        self.conversation_history = []

    def _init_tools(self) -> List[Dict]:
        """初始化工具"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "search_faq",
                    "description": "搜索常见问题知识库",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "搜索关键词"
                            }
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_order_status",
                    "description": "查询订单状态",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "order_id": {
                                "type": "string",
                                "description": "订单编号"
                            }
                        },
                        "required": ["order_id"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "cancel_order",
                    "description": "取消订单",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "order_id": {
                                "type": "string",
                                "description": "订单编号"
                            },
                            "reason": {
                                "type": "string",
                                "description": "取消原因"
                            }
                        },
                        "required": ["order_id"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "transfer_to_human",
                    "description": "转接人工客服",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "reason": {
                                "type": "string",
                                "description": "转接原因"
                            },
                            "priority": {
                                "type": "string",
                                "enum": ["low", "medium", "high"],
                                "description": "优先级"
                            }
                        },
                        "required": ["reason"]
                    }
                }
            }
        ]

    def _execute_tool(self, tool_name: str, tool_args: Dict) -> Any:
        """执行工具函数"""
        # 这里实现实际的工具逻辑
        if tool_name == "search_faq":
            return self._search_faq(tool_args["query"])
        elif tool_name == "get_order_status":
            return self._get_order_status(tool_args["order_id"])
        elif tool_name == "cancel_order":
            return self._cancel_order(tool_args["order_id"], tool_args.get("reason"))
        elif tool_name == "transfer_to_human":
            return self._transfer_to_human(tool_args["reason"], tool_args.get("priority", "medium"))

    def _search_faq(self, query: str) -> str:
        """搜索 FAQ"""
        # 实际实现会连接知识库
        return f"找到与'{query}'相关的常见问题..."

    def _get_order_status(self, order_id: str) -> str:
        """查询订单状态"""
        # 实际实现会连接订单系统
        return f"订单 {order_id} 状态：已发货"

    def _cancel_order(self, order_id: str, reason: str = None) -> str:
        """取消订单"""
        # 实际实现会调用订单 API
        return f"订单 {order_id} 已取消"

    def _transfer_to_human(self, reason: str, priority: str) -> str:
        """转接人工"""
        # 实际实现会创建工单
        return f"已转接人工客服，原因：{reason}，优先级：{priority}"

    def chat(self, user_message: str) -> str:
        """处理用户消息"""
        # 添加用户消息到历史
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        # 调用 LLM
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=self.conversation_history,
            tools=self.tools
        )

        assistant_message = response.choices[0].message
        self.conversation_history.append(assistant_message)

        # 处理工具调用
        max_iterations = 5
        for _ in range(max_iterations):
            if not assistant_message.tool_calls:
                break

            # 执行所有工具调用
            tool_messages = []
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                result = self._execute_tool(tool_name, tool_args)

                tool_messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                })

            # 添加工具结果到历史
            self.conversation_history.extend(tool_messages)

            # 继续对话
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=self.conversation_history,
                tools=self.tools
            )

            assistant_message = response.choices[0].message
            self.conversation_history.append(assistant_message)

        return assistant_message.content

# 使用
agent = CustomerServiceAgent(api_key="your-api-key")

# 示例对话
print(agent.chat("我的订单12345什么时候到？"))
print(agent.chat("我想取消这个订单"))
print(agent.chat("你们支持七天无理由退货吗？"))

# ✅ 好：每个工具做一件事
@tool
def get_user_name(user_id: str) -> str:
    """获取用户姓名"""
    pass

@tool
def get_user_email(user_id: str) -> str:
    """获取用户邮箱"""
    pass

# ❌ 差：一个工具做多件事
@tool
def get_user_info(user_id: str, field: str) -> str:
    """获取用户信息"""
    pass

@tool
def calculate_discount(price: float, discount_percent: float) -> float:
    """计算折扣后价格"""
    return price * (1 - discount_percent / 100)

@tool
def calculate_tax(price: float, tax_rate: float) -> float:
    """计算税费"""
    return price * tax_rate

@tool
def calculate_final_price(base_price: float, discount: float, tax_rate: float) -> float:
    """计算最终价格（组合其他工具）"""
    discounted = calculate_discount(base_price, discount)
    with_tax = calculate_tax(discounted, tax_rate)
    return discounted + with_tax

@tool
def create_document(document_id: str, content: str) -> str:
    """创建文档（幂等）"""
    if document_exists(document_id):
        return update_document(document_id, content)
    else:
        return insert_document(document_id, content)