Function Calling函数调用深度解析：从工具描述到结构化输出的工程实现

一、引言：函数调用是Agent的基石

大模型的Function Calling（函数调用）能力是AI Agent实现工具调用的基础。通过将工具描述为结构化的JSON Schema，模型能够理解工具的用途和参数，生成符合格式的调用指令。从OpenAI的Function Calling到Anthropic的Tool Use，从单工具调用到并行多工具调用，函数调用技术持续演进。本文将深入解析Function Calling的原理和实现。

二、函数调用实现

import torch
import torch.nn as nn
import torch.nn.functional as F
import json
import re
from typing import List, Dict, Any, Optional, Callable, Tuple
from dataclasses import dataclass, field
import math

@dataclass
class FunctionDefinition:
    """函数定义"""
    name: str
    description: str
    parameters: Dict[str, Any]  # JSON Schema
    
    def to_dict(self) -> Dict:
        return {
            'type': 'function',
            'function': {
                'name': self.name,
                'description': self.description,
                'parameters': self.parameters
            }
        }

@dataclass
class FunctionCall:
    """函数调用"""
    name: str
    arguments: Dict[str, Any]

@dataclass
class FunctionResult:
    """函数结果"""
    name: str
    result: Any
    success: bool = True
    error: str = ""

class FunctionRegistry:
    """函数注册表"""
    
    def __init__(self):
        self.functions: Dict[str, FunctionDefinition] = {}
        self.handlers: Dict[str, Callable] = {}
    
    def register(self, func_def: FunctionDefinition, handler: Callable):
        """注册函数"""
        self.functions[func_def.name] = func_def
        self.handlers[func_def.name] = handler
    
    def execute(self, call: FunctionCall) -> FunctionResult:
        """执行函数调用"""
        if call.name not in self.handlers:
            return FunctionResult(
                name=call.name,
                result=None,
                success=False,
                error=f"Unknown function: {call.name}"
            )
        
        try:
            result = self.handlers[call.name](**call.arguments)
            return FunctionResult(name=call.name, result=result)
        except Exception as e:
            return FunctionResult(
                name=call.name,
                result=None,
                success=False,
                error=str(e)
            )
    
    def get_definitions(self) -> List[Dict]:
        """获取所有函数定义"""
        return [f.to_dict() for f in self.functions.values()]
    
    def get_definitions_text(self) -> str:
        """获取文本格式的函数定义"""
        lines = []
        for f in self.functions.values():
            lines.append(f"Function: {f.name}")
            lines.append(f"  Description: {f.description}")
            params = f.parameters.get('properties', {})
            required = f.parameters.get('required', [])
            for pname, pdef in params.items():
                req = "(required)" if pname in required else "(optional)"
                lines.append(f"  - {pname}: {pdef.get('type', 'any')} {req} - {pdef.get('description', '')}")
        return '\n'.join(lines)

class FunctionCallParser:
    """函数调用解析器"""
    
    @staticmethod
    def parse_from_text(text: str) -> Optional[FunctionCall]:
        """从文本中解析函数调用"""
        # 尝试JSON格式
        json_match = re.search(r'\{[^{}]*"name"[^{}]*"arguments"[^{}]*\}', text, re.DOTALL)
        if json_match:
            try:
                data = json.loads(json_match.group())
                return FunctionCall(
                    name=data['name'],
                    arguments=data.get('arguments', {})
                )
            except:
                pass
        
        # 尝试函数调用格式
        func_match = re.search(r'(?:function_call|tool_call)\s*:\s*(\w+)\s*\(([^)]*)\)', text, re.IGNORECASE)
        if func_match:
            name = func_match.group(1)
            args_str = func_match.group(2)
            arguments = FunctionCallParser._parse_arguments(args_str)
            return FunctionCall(name=name, arguments=arguments)
        
        # 尝试XML风格
        xml_match = re.search(r'<function_call>\s*<name>(\w+)</name>\s*<arguments>(.*?)</arguments>\s*</function_call>', 
                            text, re.DOTALL | re.IGNORECASE)
        if xml_match:
            name = xml_match.group(1)
            args_str = xml_match.group(2)
            arguments = FunctionCallParser._parse_arguments(args_str)
            return FunctionCall(name=name, arguments=arguments)
        
        return None
    
    @staticmethod
    def _parse_arguments(args_str: str) -> Dict:
        """解析参数字符串"""
        args = {}
        # 尝试JSON
        try:
            return json.loads(args_str)
        except:
            pass
        
        # 尝试key=value格式
        for match in re.finditer(r'(\w+)\s*=\s*([^,]+)', args_str):
            key = match.group(1)
            value = match.group(2).strip().strip('"\'')
            # 尝试转换类型
            try:
                value = int(value)
            except:
                try:
                    value = float(value)
                except:
                    pass
            args[key] = value
        
        return args

class FunctionCallingAgent:
    """支持函数调用的Agent"""
    
    def __init__(self, registry: FunctionRegistry, llm_generate: Callable = None):
        self.registry = registry
        self.llm = llm_generate
        self.conversation: List[Dict] = []
        self.max_iterations = 5
    
    def run(self, user_query: str) -> str:
        """运行Agent"""
        self.conversation.append({'role': 'user', 'content': user_query})
        
        for i in range(self.max_iterations):
            # 构建prompt
            prompt = self._build_prompt(user_query)
            
            # 获取LLM响应
            response = self._generate(prompt)
            
            # 解析函数调用
            call = FunctionCallParser.parse_from_text(response)
            
            if call:
                # 执行函数
                result = self.registry.execute(call)
                
                # 记录到对话
                self.conversation.append({
                    'role': 'assistant',
                    'content': response,
                    'function_call': {'name': call.name, 'arguments': call.arguments}
                })
                self.conversation.append({
                    'role': 'function',
                    'name': call.name,
                    'content': json.dumps({'result': str(result.result), 'success': result.success})
                })
                
                print(f"[调用] {call.name}({call.arguments})")
                print(f"[结果] {result.result}")
                
                # 如果函数返回了最终答案
                if result.success and self._is_final(result.result):
                    return str(result.result)
            else:
                # 没有函数调用，返回响应
                self.conversation.append({'role': 'assistant', 'content': response})
                return response
        
        return "达到最大迭代次数"
    
    def _build_prompt(self, query: str) -> str:
        """构建prompt"""
        functions_text = self.registry.get_definitions_text()
        
        history = ""
        for msg in self.conversation[-5:]:
            if msg['role'] == 'user':
                history += f"User: {msg['content']}\n"
            elif msg['role'] == 'assistant':
                history += f"Assistant: {msg['content']}\n"
            elif msg['role'] == 'function':
                history += f"Function {msg['name']} result: {msg['content']}\n"
        
        return f"""You have access to the following functions:

{functions_text}

When you need to use a function, respond with:
{{"name": "function_name", "arguments": {{"param": "value"}}}}

Conversation:
{history}

User: {query}

Response:"""
    
    def _generate(self, prompt: str) -> str:
        """生成响应"""
        if self.llm:
            return self.llm(prompt)
        
        # 模拟：根据prompt内容生成函数调用
        if 'weather' in prompt.lower():
            return '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
        elif 'calculate' in prompt.lower() or 'math' in prompt.lower():
            return '{"name": "calculator", "arguments": {"expression": "2 + 3 * 4"}}'
        elif 'search' in prompt.lower():
            return '{"name": "search_web", "arguments": {"query": "AI news"}}'
        return "I can help with that. Let me think..."
    
    def _is_final(self, result: Any) -> bool:
        """判断是否为最终答案"""
        return isinstance(result, str) and len(result) > 10

# 预定义函数
def create_default_registry() -> FunctionRegistry:
    """创建默认函数注册表"""
    registry = FunctionRegistry()
    
    # 天气查询
    registry.register(
        FunctionDefinition(
            name="get_weather",
            description="Get current weather for a city",
            parameters={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        ),
        lambda city, unit="celsius": f"Weather in {city}: 25°C, sunny ({unit})"
    )
    
    # 计算器
    registry.register(
        FunctionDefinition(
            name="calculator",
            description="Perform mathematical calculations",
            parameters={
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression"}
                },
                "required": ["expression"]
            }
        ),
        lambda expression: eval(expression, {"__builtins__": {}}, 
                                {"abs": abs, "round": round, "min": min, "max": max,
                                 "sqrt": math.sqrt, "pi": math.pi})
    )
    
    # 网页搜索
    registry.register(
        FunctionDefinition(
            name="search_web",
            description="Search the web for information",
            parameters={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "num_results": {"type": "integer", "description": "Number of results"}
                },
                "required": ["query"]
            }
        ),
        lambda query, num_results=5: f"Found {num_results} results for '{query}': [result1, result2, ...]"
    )
    
    # 时间查询
    registry.register(
        FunctionDefinition(
            name="get_current_time",
            description="Get current date and time",
            parameters={
                "type": "object",
                "properties": {
                    "timezone": {"type": "string", "description": "Timezone (e.g., UTC, Asia/Shanghai)"}
                },
                "required": []
            }
        ),
        lambda timezone="UTC": f"Current time ({timezone}): 2024-01-15 10:30:00"
    )
    
    return registry

def test_function_calling():
    """测试函数调用"""
    registry = create_default_registry()
    agent = FunctionCallingAgent(registry)
    
    queries = [
        "What's the weather in Beijing?",
        "Calculate 2 + 3 * 4",
        "Search for AI news",
        "What time is it?",
    ]
    
    print("=== 函数调用测试 ===\n")
    for query in queries:
        print(f"User: {query}")
        result = agent.run(query)
        print(f"Final: {result}\n")
        agent.conversation = []  # 重置
    
    # 并行函数调用
    print("=== 并行函数调用 ===")
    
    class ParallelFunctionCaller:
        """并行函数调用器"""
        def __init__(self, registry: FunctionRegistry):
            self.registry = registry
        
        def call_parallel(self, calls: List[FunctionCall]) -> List[FunctionResult]:
            """并行执行多个函数调用"""
            import concurrent.futures
            
            results = []
            with concurrent.futures.ThreadPoolExecutor() as executor:
                futures = [executor.submit(self.registry.execute, call) for call in calls]
                for future in concurrent.futures.as_completed(futures):
                    results.append(future.result())
            
            return results
    
    parallel_caller = ParallelFunctionCaller(registry)
    calls = [
        FunctionCall(name="get_weather", arguments={"city": "Beijing"}),
        FunctionCall(name="get_weather", arguments={"city": "Shanghai"}),
        FunctionCall(name="calculator", arguments={"expression": "10 * 10"}),
        FunctionCall(name="get_current_time", arguments={}),
    ]
    
    results = parallel_caller.call_parallel(calls)
    for r in results:
        print(f"  {r.name}: {r.result}")
    
    # 函数调用格式对比
    print("\n=== 函数调用格式 ===")
    formats = [
        ("OpenAI Function Calling", '{"name": "func", "arguments": {"k": "v"}}'),
        ("Anthropic Tool Use", '<tool_use>{"name": "func", "input": {"k": "v"}}</tool_use>'),
        ("XML Style", '<function_call>\n<name>func</name>\n<arguments>{"k": "v"}</arguments>\n</function_call>'),
        ("ReAct Style", 'Action: func\nAction Input: {"k": "v"}'),
    ]
    for name, example in formats:
        print(f"  {name}: {example}")

if __name__ == "__main__":
    test_function_calling()

三、总结

Function Calling通过将工具描述为结构化JSON Schema，使大模型能够理解工具的用途和参数，生成可执行的调用指令。函数调用的核心流程为：注册函数定义 -> LLM生成调用 -> 解析调用格式 -> 执行函数 -> 返回结果给LLM -> LLM基于结果生成最终回复。并行函数调用支持同时执行多个独立工具，提高Agent效率。不同厂商的调用格式略有差异（OpenAI的JSON、Anthropic的XML标签、ReAct的文本格式），但核心逻辑一致。函数调用是AI Agent从"对话"到"行动"的关键技术桥梁，使模型能够感知外部世界和执行真实操作。