Function Calling

Function Calling（函数调用）是大语言模型的一项强大功能，允许模型根据用户输入自动调用预定义的函数，从而实现与外部系统、API 和工具的集成。

概述

Function Calling 使模型能够：

• 理解何时需要调用函数
• 提取函数所需的参数
• 生成结构化的函数调用
• 处理函数返回结果

工作原理

1. 函数定义：定义可用的函数及其参数
2. 意图识别：模型判断是否需要调用函数
3. 参数提取：从用户输入中提取函数参数
4. 函数执行：执行函数调用
5. 结果处理：模型处理函数返回结果并生成回复

基本用法

定义函数

import openai
import json

client = openai.OpenAI(
    api_key="your_api_key",
    base_url="https://realmrouter.cn/v1"
)

# 定义函数
functions = [
    {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位"
                }
            },
            "required": ["city"]
        }
    }
]

基本函数调用

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "北京今天天气怎么样？"
        }
    ],
    functions=functions,
    function_call="auto"
)

# 检查是否需要调用函数
if response.choices[0].message.function_call:
    function_call = response.choices[0].message.function_call
    function_name = function_call.name
    function_args = json.loads(function_call.arguments)
    
    print(f"调用函数: {function_name}")
    print(f"参数: {function_args}")
    
    # 执行函数（这里用模拟函数）
    if function_name == "get_weather":
        result = get_weather(function_args["city"], function_args.get("unit", "celsius"))
        
        # 将函数结果返回给模型
        second_response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "user", "content": "北京今天天气怎么样？"},
                response.choices[0].message,  # 模型的函数调用消息
                {
                    "role": "function",
                    "name": function_name,
                    "content": json.dumps(result)
                }
            ]
        )
        
        print(second_response.choices[0].message.content)

实际函数实现

def get_weather(city, unit="celsius"):
    """模拟天气API调用"""
    # 这里应该是实际的API调用
    weather_data = {
        "北京": {"temp": 25, "condition": "晴朗", "humidity": 60},
        "上海": {"temp": 28, "condition": "多云", "humidity": 70},
        "广州": {"temp": 32, "condition": "阴天", "humidity": 80}
    }
    
    if city in weather_data:
        data = weather_data[city]
        if unit == "fahrenheit":
            data["temp"] = data["temp"] * 9/5 + 32
        
        return {
            "city": city,
            "temperature": data["temp"],
            "unit": unit,
            "condition": data["condition"],
            "humidity": data["humidity"]
        }
    else:
        return {"error": "城市未找到"}

高级功能

多函数调用

functions = [
    {
        "name": "get_weather",
        "description": "获取天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"}
            },
            "required": ["city"]
        }
    },
    {
        "name": "get_stock_price",
        "description": "获取股票价格",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码"},
                "exchange": {"type": "string", "description": "交易所"}
            },
            "required": ["symbol"]
        }
    },
    {
        "name": "send_email",
        "description": "发送邮件",
        "parameters": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "收件人邮箱"},
                "subject": {"type": "string", "description": "邮件主题"},
                "body": {"type": "string", "description": "邮件内容"}
            },
            "required": ["to", "subject", "body"]
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "帮我查一下北京的天气，然后发送邮件给user@example.com告诉他天气情况"
        }
    ],
    functions=functions,
    function_call="auto"
)

流式函数调用

def stream_function_call(client, messages, functions):
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        functions=functions,
        function_call="auto",
        stream=True
    )
    
    function_call_data = {}
    current_function = None
    
    for chunk in stream:
        delta = chunk.choices[0].delta
        
        if delta.function_call:
            if delta.function_call.name:
                current_function = delta.function_call.name
                function_call_data[current_function] = {"arguments": ""}
            
            if delta.function_call.arguments:
                function_call_data[current_function]["arguments"] += delta.function_call.arguments
        
        elif delta.content:
            print(delta.content, end="", flush=True)
    
    return function_call_data

# 使用示例
messages = [{"role": "user", "content": "现在几点了？"}]
function_calls = stream_function_call(client, messages, functions)

并行函数调用

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "同时查一下北京、上海、广州的天气"
        }
    ],
    functions=functions,
    function_call="auto"
)

# 处理多个函数调用
if response.choices[0].message.function_call:
    # 并行执行多个函数调用
    import concurrent.futures
    
    def execute_function_call(function_call):
        function_name = function_call.name
        function_args = json.loads(function_call.arguments)
        
        if function_name == "get_weather":
            return get_weather(function_args["city"])
        # 其他函数...
    
    # 如果有多个函数调用，可以并行执行
    # 这里简化处理，实际可能需要解析多个函数调用

实际应用场景

1. 数据库查询

functions = [
    {
        "name": "query_database",
        "description": "查询数据库信息",
        "parameters": {
            "type": "object",
            "properties": {
                "table": {"type": "string", "description": "表名"},
                "conditions": {"type": "object", "description": "查询条件"},
                "limit": {"type": "integer", "description": "返回数量限制"}
            },
            "required": ["table"]
        }
    }
]

def query_database(table, conditions=None, limit=10):
    """实际数据库查询函数"""
    # 实现数据库查询逻辑
    pass

2. API 集成

functions = [
    {
        "name": "create_calendar_event",
        "description": "创建日历事件",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "事件标题"},
                "start_time": {"type": "string", "description": "开始时间"},
                "end_time": {"type": "string", "description": "结束时间"},
                "attendees": {"type": "array", "items": {"type": "string"}, "description": "参与者"}
            },
            "required": ["title", "start_time", "end_time"]
        }
    }
]

def create_calendar_event(title, start_time, end_time, attendees=None):
    """调用日历API创建事件"""
    # 实现日历API调用
    pass

3. 文件操作

functions = [
    {
        "name": "read_file",
        "description": "读取文件内容",
        "parameters": {
            "type": "object",
            "properties": {
                "filepath": {"type": "string", "description": "文件路径"},
                "encoding": {"type": "string", "description": "文件编码"}
            },
            "required": ["filepath"]
        }
    },
    {
        "name": "write_file",
        "description": "写入文件内容",
        "parameters": {
            "type": "object",
            "properties": {
                "filepath": {"type": "string", "description": "文件路径"},
                "content": {"type": "string", "description": "文件内容"},
                "encoding": {"type": "string", "description": "文件编码"}
            },
            "required": ["filepath", "content"]
        }
    }
]

最佳实践

1. 函数设计原则

# 好的函数定义
good_function = {
    "name": "search_products",
    "description": "搜索产品信息，支持按类别、价格范围等条件筛选",
    "parameters": {
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "description": "产品类别，如：电子产品、服装、食品等"
            },
            "min_price": {
                "type": "number",
                "description": "最低价格（可选）"
            },
            "max_price": {
                "type": "number",
                "description": "最高价格（可选）"
            },
            "sort_by": {
                "type": "string",
                "enum": ["price", "rating", "sales"],
                "description": "排序方式"
            }
        },
        "required": ["category"]
    }
}

# 避免的函数定义
bad_function = {
    "name": "do_something",
    "description": "做一些事情",
    "parameters": {
        "type": "object",
        "properties": {
            "data": {"type": "object", "description": "数据"}
        }
    }
}

2. 错误处理

def safe_function_call(client, messages, functions, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                functions=functions,
                function_call="auto"
            )
            
            if response.choices[0].message.function_call:
                function_call = response.choices[0].message.function_call
                function_name = function_call.name
                function_args = json.loads(function_call.arguments)
                
                # 验证参数
                if not validate_function_args(function_name, function_args):
                    raise ValueError(f"无效的函数参数: {function_args}")
                
                # 执行函数
                result = execute_function(function_name, function_args)
                
                return {
                    "success": True,
                    "result": result,
                    "function_call": function_call
                }
            else:
                return {
                    "success": True,
                    "content": response.choices[0].message.content
                }
                
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            else:
                return {
                    "success": False,
                    "error": str(e)
                }

3. 性能优化

# 缓存函数结果
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_get_weather(city, unit="celsius"):
    """带缓存的天气查询"""
    return get_weather(city, unit)

# 批量函数调用
def batch_function_calls(calls):
    """批量执行函数调用"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = []
        
        for call in calls:
            future = executor.submit(execute_function, call["name"], call["args"])
            futures.append(future)
        
        results = []
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                results.append({"error": str(e)})
        
        return results

限制和注意事项

1. 函数数量限制：单次请求中定义的函数数量有限制
2. 参数复杂度：过于复杂的参数结构可能影响识别准确率
3. 执行时间：函数执行时间会影响整体响应时间
4. 安全性：确保函数调用的安全性，避免执行恶意代码
5. 错误处理：妥善处理函数执行失败的情况

调试和测试

def test_function_calling():
    """测试函数调用功能"""
    test_cases = [
        {
            "input": "北京今天天气怎么样？",
            "expected_function": "get_weather",
            "expected_args": {"city": "北京"}
        },
        {
            "input": "帮我查一下AAPL的股价",
            "expected_function": "get_stock_price",
            "expected_args": {"symbol": "AAPL"}
        }
    ]
    
    for case in test_cases:
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": case["input"]}],
            functions=functions,
            function_call="auto"
        )
        
        if response.choices[0].message.function_call:
            function_call = response.choices[0].message.function_call
            assert function_call.name == case["expected_function"]
            # 更多断言...
        
        print(f"测试通过: {case['input']}")

# 运行测试
test_function_calling()