Function Calling 框架完整实现
import json
import requests
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import inspect
from abc import ABC, abstractmethod
class ToolParameterType(Enum):
"""工具参数类型"""
STRING = "string"
NUMBER = "number"
INTEGER = "integer"
BOOLEAN = "boolean"
ARRAY = "array"
OBJECT = "object"
@dataclass
class ToolParameter:
"""工具参数定义"""
name: str
type: ToolParameterType
description: str
required: bool = True
enum: Optional[List[Any]] = None
default: Optional[Any] = None
@dataclass
class ToolSchema:
"""工具 Schema 定义"""
name: str
description: str
parameters: List[ToolParameter]
def to_openai_format(self) -> Dict:
"""转换为 OpenAI Function Calling 格式"""
properties = {}
required = []
for param in self.parameters:
prop = {
"type": param.type.value,
"description": param.description
}
if param.enum:
prop["enum"] = param.enum
if param.default is not None:
prop["default"] = param.default
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
class Tool(ABC):
"""工具基类"""
def __init__(self, schema: ToolSchema):
self.schema = schema
@abstractmethod
def execute(self, **kwargs) -> Any:
"""执行工具"""
pass
def validate_input(self, **kwargs) -> bool:
"""验证输入参数"""
for param in self.schema.parameters:
if param.required and param.name not in kwargs:
raise ValueError(f"Missing required parameter: {param.name}")
if param.name in kwargs:
value = kwargs[param.name]
expected_type = param.type.value
# 简单类型检查
if expected_type == "string" and not isinstance(value, str):
raise TypeError(f"Parameter {param.name} must be string")
elif expected_type in ["number", "integer"] and not isinstance(value, (int, float)):
raise TypeError(f"Parameter {param.name} must be number")
elif expected_type == "boolean" and not isinstance(value, bool):
raise TypeError(f"Parameter {param.name} must be boolean")
# Enum 检查
if param.enum and value not in param.enum:
raise ValueError(f"Parameter {param.name} must be one of {param.enum}")
return True
class WeatherTool(Tool):
"""天气查询工具示例"""
def __init__(self, api_key: str):
schema = ToolSchema(
name="get_current_weather",
description="Get the current weather in a given location",
parameters=[
ToolParameter(
name="location",
type=ToolParameterType.STRING,
description="The city and state, e.g. San Francisco, CA",
required=True
),
ToolParameter(
name="unit",
type=ToolParameterType.STRING,
description="Temperature unit",
required=False,
enum=["celsius", "fahrenheit"],
default="celsius"
)
]
)
super().__init__(schema)
self.api_key = api_key
def execute(self, **kwargs) -> Dict:
"""执行天气查询"""
self.validate_input(**kwargs)
location = kwargs.get("location")
unit = kwargs.get("unit", "celsius")
# 模拟 API 调用（实际应调用真实天气 API）
# response = requests.get(
# f"https://api.weather.com/v1/current?q={location}&units={unit}",
# headers={"Authorization": f"Bearer {self.api_key}"}
# )
# return response.json()
# 模拟响应
return {
"location": location,
"temperature": 22 if unit == "celsius" else 72,
"unit": unit,
"condition": "Sunny",
"humidity": 65,
"wind_speed": 12
}
class CalculatorTool(Tool):
"""计算器工具示例"""
def __init__(self):
schema = ToolSchema(
name="calculate",
description="Perform mathematical calculations",
parameters=[
ToolParameter(
name="expression",
type=ToolParameterType.STRING,
description="Mathematical expression to evaluate, e.g. '2 + 2 * 3'",
required=True
)
]
)
super().__init__(schema)
def execute(self, **kwargs) -> Dict:
"""执行计算"""
self.validate_input(**kwargs)
expression = kwargs.get("expression")
# 安全计算（避免 eval 风险）
try:
# 使用 ast 安全解析
import ast
import operator
operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg
}
def eval_expr(node):
if isinstance(node, ast.Num):
return node.n
elif isinstance(node, ast.BinOp):
left = eval_expr(node.left)
right = eval_expr(node.right)
return operators[type(node.op)](left, right)
elif isinstance(node, ast.UnaryOp):
operand = eval_expr(node.operand)
return operators[type(node.op)](operand)
else:
raise TypeError(f"Unsupported operation: {type(node)}")
tree = ast.parse(expression, mode='eval')
result = eval_expr(tree.body)
return {
"expression": expression,
"result": result,
"success": True
}
except Exception as e:
return {
"expression": expression,
"error": str(e),
"success": False
}
class FunctionCaller:
"""
Function Calling 执行器
核心流程:
1. 注册工具（定义 Schema）
2. LLM 生成工具调用请求
3. 解析并验证调用
4. 执行工具
5. 返回结果给 LLM
"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.schema.name] = tool
print(f"Registered tool: {tool.schema.name}")
def get_tools_schema(self) -> List[Dict]:
"""获取所有工具的 Schema（用于 LLM 调用）"""
return [tool.schema.to_openai_format() for tool in self.tools.values()]
def execute_tool_call(self, tool_name: str, arguments: Dict) -> Any:
"""
执行工具调用
Args:
tool_name: 工具名称
arguments: 参数字典
Returns:
工具执行结果
"""
if tool_name not in self.tools:
raise ValueError(f"Unknown tool: {tool_name}. Available tools: {list(self.tools.keys())}")
tool = self.tools[tool_name]
try:
result = tool.execute(**arguments)
return {
"success": True,
"tool": tool_name,
"result": result
}
except Exception as e:
return {
"success": False,
"tool": tool_name,
"error": str(e)
}
def process_llm_response(self, llm_response: Dict) -> Any:
"""
处理 LLM 响应（包含工具调用）
Args:
llm_response: LLM 响应，格式如:
{
"role": "assistant",
"tool_calls": [
{
"id": "call_123",
"function": {
"name": "get_current_weather",
"arguments": '{"location": "Beijing"}'
}
}
]
}
Returns:
工具执行结果列表
"""
if "tool_calls" not in llm_response or not llm_response["tool_calls"]:
return None
results = []
for tool_call in llm_response["tool_calls"]:
tool_name = tool_call["function"]["name"]
arguments_str = tool_call["function"]["arguments"]
# 解析参数
try:
arguments = json.loads(arguments_str)
except json.JSONDecodeError as e:
results.append({
"success": False,
"tool": tool_name,
"error": f"Invalid JSON arguments: {e}"
})
continue
# 执行工具
result = self.execute_tool_call(tool_name, arguments)
result["call_id"] = tool_call.get("id")
results.append(result)
return results
# 使用示例
if __name__ == "__main__":
# 创建 Function Caller
caller = FunctionCaller()
# 注册工具
weather_tool = WeatherTool(api_key="demo_key")
calculator_tool = CalculatorTool()
caller.register_tool(weather_tool)
caller.register_tool(calculator_tool)
print("\n=== 已注册工具 Schema ===")
for schema in caller.get_tools_schema():
print(json.dumps(schema, indent=2))
print("\n=== 模拟 LLM 工具调用 ===")
# 模拟 LLM 响应（工具调用）
llm_response = {
"role": "assistant",
"tool_calls": [
{
"id": "call_001",
"function": {
"name": "get_current_weather",
"arguments": json.dumps({"location": "Beijing, China", "unit": "celsius"})
}
},
{
"id": "call_002",
"function": {
"name": "calculate",
"arguments": json.dumps({"expression": "2 + 2 * 3"})
}
}
]
}
# 处理 LLM 响应
results = caller.process_llm_response(llm_response)
print("\n=== 工具执行结果 ===")
for result in results:
print(f"\n工具：{result['tool']}")
print(f"调用 ID: {result.get('call_id')}")
print(f"成功：{result['success']}")
if result['success']:
print(f"结果：{json.dumps(result['result'], indent=2, ensure_ascii=False)}")
else:
print(f"错误：{result.get('error')}")
print("\n关键观察:")
print("1. Function Calling 将自然语言转换为结构化函数调用")
print("2. Schema 定义：工具名称、描述、参数（类型、必填、枚举）")
print("3. 执行流程：LLM 生成调用 → 解析验证 → 执行工具 → 返回结果")
print("4. 错误处理：参数验证、执行异常、JSON 解析错误")
print("5. 可扩展性：轻松添加新工具，无需修改 LLM")