第二章：小型语言模型（SLMs）中的函数调用

文档摘要

第02节：小型语言模型（SLMs）中的函数调用目录什么是函数调用？函数调用的工作原理应用场景使用Phi-4-mini和Ollama设置函数调用使用Qwen3进行函数调用 Foundry Local集成最佳实践与故障排除高级示例什么是函数调用？函数调用是一种强大的功能，使小型语言模型（SLMs）能够与外部工具、API和服务交互。通过函数调用，SLMs不再局限于其训练数据，而是可以：连接外部API（如天气服务、数据库、搜索引擎）根据用户请求执行特定功能从各种来源获取实时信息通过专业工具执行计算任务将多个操作串联起来以完成复杂工作流这种能力将SLMs从静态文本生成器转变为能够执行实际任务的动态AI代理。

第02节：小型语言模型（SLMs）中的函数调用

什么是函数调用？

函数调用是一种强大的功能，使小型语言模型（SLMs）能够与外部工具、API和服务交互。通过函数调用，SLMs不再局限于其训练数据，而是可以：

连接外部API（如天气服务、数据库、搜索引擎）
根据用户请求执行特定功能
从各种来源获取实时信息
通过专业工具执行计算任务
将多个操作串联起来以完成复杂工作流

这种能力将SLMs从静态文本生成器转变为能够执行实际任务的动态AI代理。

函数调用的工作原理

函数调用过程遵循一个系统化的工作流：

1. 工具集成

外部工具：SLMs可以连接到天气API、数据库、Web服务和其他外部系统
函数定义：每个工具都通过特定的参数、输入/输出格式和描述进行定义
API兼容性：工具通过标准化接口（如REST API、SDK等）集成

2. 函数定义

函数由三个关键组件定义：


{
  "name": "function_name",
  "description": "Clear description of what the function does",
  "parameters": {
    "parameter_name": {
      "description": "What this parameter represents",
      "type": "data_type",
      "default": "default_value"
    }
  }
}

3. 意图识别

自然语言处理：SLM分析用户输入以理解意图
函数匹配：确定需要调用哪些函数以满足请求
参数提取：识别并提取用户消息中所需的参数

4. JSON输出生成

SLM生成结构化的JSON，其中包含：

要调用的函数名称
所需参数及其对应值
执行上下文和元数据

5. 外部执行

参数验证：确保所有必需参数都存在且格式正确
函数执行：应用程序使用提供的参数执行指定函数
错误处理：管理失败、超时和无效响应

6. 响应集成

结果处理：函数输出返回给SLM
上下文整合：SLM将结果整合到其响应中
用户沟通：以自然、对话的形式向用户呈现信息

应用场景

数据检索

将自然语言查询转换为结构化API调用：

“显示我的最近订单” → 带有用户ID和日期过滤器的数据库查询
“东京的天气怎么样？” → 带有位置参数的天气API调用
“查找上周John发来的邮件” → 带有发件人和日期过滤器的邮件服务查询

操作执行

将用户请求转换为特定函数调用：

“安排明天下午2点的会议” → 日历API集成
“给团队发一条消息” → 通信平台API
“备份我的文件” → 文件系统操作

计算任务

处理复杂的数学或逻辑操作：

“计算$10,000以5%利率10年的复利” → 财务计算函数
“分析这个数据集的趋势” → 统计分析工具
“优化这条配送路线” → 路线优化算法

数据处理工作流

为复杂操作串联多个函数调用：

从多个来源检索数据
解析并验证信息
将数据转换为所需格式
将结果存储在适当的系统中
生成报告或可视化

UI/UX集成

支持动态界面更新：

“在仪表板上显示销售数据” → 图表生成和显示
“用新位置更新地图” → 地理空间数据集成
“刷新库存显示” → 实时数据同步

使用Phi-4-mini和Ollama设置函数调用

微软的Phi-4-mini通过Ollama支持单一和并行函数调用。以下是设置步骤：

前置条件

Ollama版本0.5.13或更高
Phi-4-mini模型（推荐：phi4-mini:3.8b-fp16）

安装步骤

1. 安装并运行Phi-4-mini


# Download the model (if not already present)
ollama run phi4-mini:3.8b-fp16

# Verify the model is available
ollama list

2. 创建自定义ModelFile模板

由于Ollama默认模板的限制，需要创建一个自定义ModelFile，模板如下：


TEMPLATE """
{{- if .Messages }}
{{- if or .System .Tools }}<|system|>
{{ if .System }}{{ .System }}
{{- end }}
In addition to plain text responses, you can chose to call one or more of the provided functions.
Use the following rule to decide when to call a function:
* if the response can be generated from your internal knowledge (e.g., as in the case of queries like "What is the capital of Poland?"), do so
* if you need external information that can be obtained by calling one or more of the provided functions, generate a function calls
If you decide to call functions:
* prefix function calls with functools marker (no closing marker required)
* all function calls should be generated in a single JSON list formatted as functools[{"name": [function name], "arguments": [function arguments as JSON]}, ...]
* follow the provided JSON schema. Do not hallucinate arguments or values. Do to blindly copy values from the provided samples
* respect the argument type formatting. E.g., if the type if number and format is float, write value 7 as 7.0
* make sure you pick the right functions that match the user intent
Available functions as JSON spec:
{{- if .Tools }}
{{ .Tools }}
{{- end }}<|end|>
{{- end }}
{{- range .Messages }}
{{- if ne .Role "system" }}<|{{ .Role }}|>
{{- if and .Content (eq .Role "tools") }}
{"result": {{ .Content }}}
{{- else if .Content }}
{{ .Content }}
{{- else if .ToolCalls }}
functools[
{{- range .ToolCalls }}{{ "{" }}"name": "{{ .Function.Name }}", "arguments": {{ .Function.Arguments }}{{ "}" }}
{{- end }}]
{{- end }}<|end|>
{{- end }}
{{- end }}<|assistant|>
{{ else }}
{{- if .System }}<|system|>
{{ .System }}<|end|>{{ end }}{{ if .Prompt }}<|user|>
{{ .Prompt }}<|end|>{{ end }}<|assistant|>
{{ end }}{{ .Response }}{{ if .Response }}<|user|>{{ end }}
"""

3. 创建自定义模型


# Save the template above as 'Modelfile' and run:
ollama create phi4-mini-fc:3.8b-fp16 -f ./Modelfile

单一函数调用示例


import json
import requests

# Define the tool/function
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather information for a location",
        "parameters": {
            "location": {
                "description": "The city or location name",
                "type": "str",
                "default": "New York"
            },
            "units": {
                "description": "Temperature units (celsius or fahrenheit)",
                "type": "str",
                "default": "celsius"
            }
        }
    }
]

# Create the message with system prompt including tools
messages = [
    {
        "role": "system",
        "content": "You are a helpful weather assistant",
        "tools": json.dumps(tools)
    },
    {
        "role": "user",
        "content": "What's the weather like in London today?"
    }
]

# Make request to Ollama API
response = requests.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "phi4-mini-fc:3.8b-fp16",
        "messages": messages,
        "stream": False
    }
)

print(response.json())

并行函数调用示例


import json
import requests

# Define multiple tools for parallel execution
AGENT_TOOLS = {
    "booking_flight": {
        "name": "booking_flight",
        "description": "Book a flight ticket",
        "parameters": {
            "departure": {
                "description": "Departure airport code",
                "type": "str"
            },
            "destination": {
                "description": "Destination airport code", 
                "type": "str"
            },
            "outbound_date": {
                "description": "Departure date (YYYY-MM-DD)",
                "type": "str"
            },
            "return_date": {
                "description": "Return date (YYYY-MM-DD)",
                "type": "str"
            }
        }
    },
    "booking_hotel": {
        "name": "booking_hotel",
        "description": "Book a hotel room",
        "parameters": {
            "city": {
                "description": "City name for hotel booking",
                "type": "str"
            },
            "check_in_date": {
                "description": "Check-in date (YYYY-MM-DD)",
                "type": "str"
            },
            "check_out_date": {
                "description": "Check-out date (YYYY-MM-DD)",
                "type": "str"
            }
        }
    }
}

SYSTEM_PROMPT = """
You are my travel agent with some tools available.
"""

messages = [
    {
        "role": "system",
        "content": SYSTEM_PROMPT,
        "tools": json.dumps(AGENT_TOOLS)
    },
    {
        "role": "user", 
        "content": "I need to travel from London to New York from March 21 2025 to March 27 2025. Please book both flight and hotel."
    }
]

# The model will generate parallel function calls
response = requests.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "phi4-mini-fc:3.8b-fp16",
        "messages": messages,
        "stream": False
    }
)

print(response.json())

使用Qwen3进行函数调用

Qwen3提供了先进的函数调用功能，具有出色的性能和灵活性。以下是实现方法：

使用Qwen-Agent框架

Qwen-Agent提供了一个高层框架，简化了函数调用的实现：

安装


pip install -U "qwen-agent[gui,rag,code_interpreter,mcp]"

基本设置


import os
from qwen_agent.agents import Assistant

# Configure the LLM
llm_cfg = {
    'model': 'Qwen3-8B',
    # Option 1: Use Alibaba Model Studio
    'model_type': 'qwen_dashscope',
    'api_key': os.getenv('DASHSCOPE_API_KEY'),
    
    # Option 2: Use local deployment
    # 'model_server': 'http://localhost:8000/v1',
    # 'api_key': 'EMPTY',
    
    # Optional configuration for thinking mode
    'generate_cfg': {
        'thought_in_content': True,  # Include reasoning in response
    }
}

# Define tools using MCP (Model Context Protocol)
tools = [
    {
        'mcpServers': {
            'time': {
                'command': 'uvx',
                'args': ['mcp-server-time', '--local-timezone=Asia/Shanghai']
            },
            'fetch': {
                'command': 'uvx', 
                'args': ['mcp-server-fetch']
            }
        }
    },
    'code_interpreter',  # Built-in code execution tool
]

# Create the assistant
bot = Assistant(llm=llm_cfg, function_list=tools)

# Example usage
messages = [
    {
        'role': 'user', 
        'content': 'What time is it now? Also, fetch the latest news from https://example.com/news'
    }
]

# Generate response with function calling
for response in bot.run(messages=messages):
    print(response)

自定义函数实现

您还可以为Qwen3定义自定义函数：


import json
from qwen_agent.tools.base import BaseTool

class WeatherTool(BaseTool):
    description = 'Get weather information for a specific location'
    parameters = [
        {
            'name': 'location',
            'type': 'string', 
            'description': 'City or location name',
            'required': True
        },
        {
            'name': 'units',
            'type': 'string',
            'description': 'Temperature units (celsius or fahrenheit)',
            'required': False,
            'default': 'celsius'
        }
    ]
    
    def call(self, params: str, **kwargs) -> str:
        """Execute the weather lookup"""
        params_dict = json.loads(params)
        location = params_dict.get('location')
        units = params_dict.get('units', 'celsius')
        
        # Simulate weather API call
        weather_data = {
            'location': location,
            'temperature': '22°C' if units == 'celsius' else '72°F',
            'condition': 'Partly cloudy',
            'humidity': '65%'
        }
        
        return json.dumps(weather_data)

# Use the custom tool
tools = [WeatherTool()]
bot = Assistant(llm=llm_cfg, function_list=tools)

messages = [{'role': 'user', 'content': 'What\'s the weather in Tokyo?'}]
response = bot.run(messages=messages)
print(list(response)[-1])

Qwen3的高级功能

思维模式控制

Qwen3支持在思维模式和非思维模式之间动态切换：


# Enable thinking mode for complex reasoning
messages = [
    {
        'role': 'user',
        'content': '/think Solve this complex math problem: If a train travels 120 km in 1.5 hours, and another train travels 200 km in 2.5 hours, which train is faster and by how much?'
    }
]

# Disable thinking mode for simple queries
messages = [
    {
        'role': 'user', 
        'content': '/no_think What is the capital of France?'
    }
]

多步函数调用

Qwen3擅长于串联多个函数调用：


# Complex workflow example
messages = [
    {
        'role': 'user',
        'content': '''
        I need to prepare for a business meeting:
        1. Check my calendar for conflicts tomorrow
        2. Get weather forecast for the meeting location (San Francisco)
        3. Find recent news about the client company (TechCorp)
        4. Calculate travel time from my office to their headquarters
        '''
    }
]

# Qwen3 will automatically determine the sequence of function calls needed

Foundry Local集成

微软的Foundry Local提供了一个OpenAI兼容的API，用于在本地运行模型，具有增强的隐私性和性能。

设置与安装

Windows

从Foundry Local发布页面下载安装程序，并按照安装说明操作。

macOS


brew tap microsoft/foundrylocal
brew install foundrylocal

基本用法


import openai
from foundry_local import FoundryLocalManager

# Initialize with model alias
alias = "phi-3.5-mini"  # Or any supported model
manager = FoundryLocalManager(alias)

# Create OpenAI client pointing to local endpoint
client = openai.OpenAI(
    base_url=manager.endpoint,
    api_key=manager.api_key
)

# Define functions for the model
functions = [
    {
        "name": "calculate_tax",
        "description": "Calculate tax amount based on income and rate",
        "parameters": {
            "type": "object",
            "properties": {
                "income": {
                    "type": "number",
                    "description": "Annual income amount"
                },
                "tax_rate": {
                    "type": "number", 
                    "description": "Tax rate as decimal (e.g., 0.25 for 25%)"
                }
            },
            "required": ["income", "tax_rate"]
        }
    }
]

# Make function calling request
response = client.chat.completions.create(
    model=manager.model_info.id,
    messages=[
        {
            "role": "user",
            "content": "Calculate the tax for someone earning $75,000 with a 22% tax rate"
        }
    ],
    functions=functions,
    function_call="auto"
)

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

Foundry Local的高级功能

模型管理


# List available models
foundry model list

# Download specific model
foundry model download phi-3.5-mini

# Run model interactively
foundry model run phi-3.5-mini

# Remove model from cache
foundry model remove phi-3.5-mini

# Delete all cached models
foundry model remove "*"

性能优化

Foundry Local会根据您的硬件自动选择最佳模型变体：

CUDA GPU：下载GPU优化模型
Qualcomm NPU：使用NPU加速变体
仅CPU：选择CPU优化模型

最佳实践与故障排除

函数定义最佳实践

1. 清晰且描述性强的命名


# Good
{
    "name": "get_stock_price",
    "description": "Retrieve current stock price for a given symbol"
}

# Avoid
{
    "name": "get_data", 
    "description": "Gets data"
}

2. 全面的参数定义


{
    "name": "send_email",
    "description": "Send an email message to specified recipients",
    "parameters": {
        "to": {
            "type": "array",
            "items": {"type": "string"},
            "description": "List of recipient email addresses",
            "required": True
        },
        "subject": {
            "type": "string",
            "description": "Email subject line",
            "required": True
        },
        "body": {
            "type": "string", 
            "description": "Email message content",
            "required": True
        },
        "priority": {
            "type": "string",
            "enum": ["low", "normal", "high"],
            "description": "Email priority level",
            "default": "normal",
            "required": False
        }
    }
}

3. 输入验证与错误处理


def execute_function(function_name, parameters):
    try:
        # Validate required parameters
        if function_name == "send_email":
            if not parameters.get("to") or not parameters.get("subject"):
                return {"error": "Missing required parameters: to, subject"}
            
            # Validate email format
            email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            for email in parameters["to"]:
                if not re.match(email_pattern, email):
                    return {"error": f"Invalid email format: {email}"}
        
        # Execute function logic
        result = perform_actual_function(function_name, parameters)
        return {"success": True, "data": result}
        
    except Exception as e:
        return {"error": str(e)}

常见问题及解决方案

问题1：函数未被调用

症状：模型以文本形式响应，而不是调用函数

解决方案：

检查函数描述：确保其清晰地匹配用户意图
验证参数定义：确保所有必需参数已正确定义
检查系统提示：明确说明何时使用函数
尝试显式请求：例如“请使用天气函数获取伦敦的数据”

问题2：参数错误

症状：函数调用时参数错误或缺失

解决方案：

添加参数示例：在参数描述中包含示例值
使用枚举约束：尽可能限制参数值为特定选项
实现回退值：为可选参数提供合理的默认值


{
    "name": "book_restaurant",
    "parameters": {
        "cuisine": {
            "type": "string",
            "enum": ["italian", "chinese", "mexican", "american", "french"],
            "description": "Type of cuisine (example: 'italian' for Italian food)"
        },
        "party_size": {
            "type": "integer",
            "minimum": 1,
            "maximum": 20,
            "description": "Number of people (example: 4 for a family of four)"
        }
    }
}

问题3：并行函数调用失败

症状：仅执行一个函数，而非多个

解决方案：

检查模型支持：确保您的模型支持并行函数调用
更新系统提示：在系统消息中包含“某些工具”或“多个工具”
使用适当的模型版本：推荐使用Phi-4-mini:3.8b-fp16与Ollama

问题4：Ollama模板问题

症状：函数调用在默认Ollama设置下无法工作

解决方案：

使用自定义ModelFile：应用本教程中提供的修正模板
更新Ollama：确保使用版本0.5.13或更高
检查模型量化：较高量化级别（Q8_0, fp16）比重度量化版本效果更好

性能优化

1. 高效的函数设计

保持函数专注：每个函数应有单一、明确的目的
最小化外部依赖：尽量减少API调用和网络请求
缓存结果：存储频繁请求的数据以提高响应速度

2. 批处理与异步操作


import asyncio
import aiohttp

async def batch_function_calls(function_calls):
    """Execute multiple function calls concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for call in function_calls:
            if call["name"] == "fetch_url":
                task = fetch_url_async(session, call["parameters"]["url"])
                tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results

async def fetch_url_async(session, url):
    async with session.get(url) as response:
        return await response.text()

3. 资源管理

连接池化：重用数据库和API连接
速率限制：对外部API实施适当的速率限制
超时处理：为所有外部调用设置合理的超时时间

高级示例

多代理协作系统


import json
from typing import List, Dict
from qwen_agent.agents import Assistant

class MultiAgentSystem:
    def __init__(self):
        # Research Agent
        self.research_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=[
                {'mcpServers': {'search': {'command': 'uvx', 'args': ['mcp-server-search']}}},
                {'mcpServers': {'fetch': {'command': 'uvx', 'args': ['mcp-server-fetch']}}}
            ]
        )
        
        # Analysis Agent
        self.analysis_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=['code_interpreter']
        )
        
        # Communication Agent
        self.comm_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=[self.create_email_tool(), self.create_slack_tool()]
        )
    
    def create_email_tool(self):
        """Custom email sending tool"""
        class EmailTool:
            name = "send_email"
            description = "Send email to specified recipients"
            parameters = {
                "to": {"type": "string", "description": "Recipient email"},
                "subject": {"type": "string", "description": "Email subject"},
                "body": {"type": "string", "description": "Email content"}
            }
            
            def call(self, params):
                # Implement actual email sending logic
                return f"Email sent successfully to {params['to']}"
        
        return EmailTool()
    
    def create_slack_tool(self):
        """Custom Slack messaging tool"""  
        class SlackTool:
            name = "send_slack"
            description = "Send message to Slack channel"
            parameters = {
                "channel": {"type": "string", "description": "Slack channel"},
                "message": {"type": "string", "description": "Message content"}
            }
            
            def call(self, params):
                # Implement actual Slack API call
                return f"Message sent to {params['channel']}"
        
        return SlackTool()
    
    async def process_complex_request(self, user_request: str):
        """Process complex multi-step requests using multiple agents"""
        
        # Step 1: Research phase
        research_prompt = f"Research the following topic and gather relevant information: {user_request}"
        research_results = []
        for response in self.research_agent.run([{'role': 'user', 'content': research_prompt}]):
            research_results.append(response)
        
        # Step 2: Analysis phase
        analysis_prompt = f"Analyze the following research data and provide insights: {research_results[-1]}"
        analysis_results = []
        for response in self.analysis_agent.run([{'role': 'user', 'content': analysis_prompt}]):
            analysis_results.append(response)
        
        # Step 3: Communication phase
        comm_prompt = f"Create a summary report and send it via email: {analysis_results[-1]}"
        comm_results = []
        for response in self.comm_agent.run([{'role': 'user', 'content': comm_prompt}]):
            comm_results.append(response)
        
        return {
            'research': research_results[-1],
            'analysis': analysis_results[-1], 
            'communication': comm_results[-1]
        }

# Usage example
async def main():
    system = MultiAgentSystem()
    
    request = """
    Analyze the impact of remote work on productivity in tech companies. 
    Research recent studies, analyze the data, and send a summary to our team.
    """
    
    results = await system.process_complex_request(request)
    print("Multi-agent processing complete:", results)

# Run the example
# asyncio.run(main())

动态工具选择系统


class DynamicToolSelector:
    def __init__(self):
        self.available_tools = {
            'weather': {
                'description': 'Get weather information',
                'domains': ['weather', 'temperature', 'forecast', 'climate'],
                'function': self.get_weather
            },
            'calculator': {
                'description': 'Perform mathematical calculations',
                'domains': ['math', 'calculate', 'compute', 'arithmetic'],
                'function': self.calculate
            },
            'web_search': {
                'description': 'Search the internet for information',
                'domains': ['search', 'find', 'lookup', 'research'],
                'function': self.web_search
            },
            'file_manager': {
                'description': 'Manage files and directories',
                'domains': ['file', 'directory', 'save', 'load', 'delete'],
                'function': self.manage_files
            }
        }
    
    def analyze_intent(self, user_input: str) -> List[str]:
        """Analyze user input to determine which tools might be needed"""
        user_words = user_input.lower().split()
        relevant_tools = []
        
        for tool_name, tool_info in self.available_tools.items():
            for domain in tool_info['domains']:
                if domain in user_words:
                    relevant_tools.append(tool_name)
                    break
        
        return relevant_tools
    
    def get_tool_definitions(self, tool_names: List[str]) -> List[Dict]:
        """Generate function definitions for selected tools"""
        definitions = []
        
        for tool_name in tool_names:
            if tool_name == 'weather':
                definitions.append({
                    'name': 'get_weather',
                    'description': 'Get current weather information',
                    'parameters': {
                        'location': {'type': 'string', 'description': 'City or location name'},
                        'units': {'type': 'string', 'enum': ['celsius', 'fahrenheit'], 'default': 'celsius'}
                    }
                })
            elif tool_name == 'calculator':
                definitions.append({
                    'name': 'calculate',
                    'description': 'Perform mathematical calculations',
                    'parameters': {
                        'expression': {'type': 'string', 'description': 'Mathematical expression to evaluate'},
                        'precision': {'type': 'integer', 'default': 2, 'description': 'Decimal places for result'}
                    }
                })
            # Add more tool definitions as needed
        
        return definitions
    
    def get_weather(self, location: str, units: str = 'celsius') -> Dict:
        """Mock weather function"""
        return {
            'location': location,
            'temperature': '22°C' if units == 'celsius' else '72°F',
            'condition': 'Sunny',
            'humidity': '60%'
        }
    
    def calculate(self, expression: str, precision: int = 2) -> Dict:
        """Safe mathematical calculation"""
        try:
            # Simple evaluation for demo - in production, use a proper math parser
            import math
            allowed_names = {
                k: v for k, v in math.__dict__.items() if not k.startswith("__")
            }
            allowed_names.update({"abs": abs, "round": round})
            
            result = eval(expression, {"__builtins__": {}}, allowed_names)
            return {
                'expression': expression,
                'result': round(float(result), precision),
                'success': True
            }
        except Exception as e:
            return {
                'expression': expression,
                'error': str(e),
                'success': False
            }
    
    def web_search(self, query: str, max_results: int = 5) -> Dict:
        """Mock web search function"""
        return {
            'query': query,
            'results': [
                {'title': f'Result {i+1} for {query}', 'url': f'https://example{i+1}.com'}
                for i in range(max_results)
            ]
        }
    
    def manage_files(self, action: str, file_path: str, content: str = None) -> Dict:
        """Mock file management function"""
        return {
            'action': action,
            'file_path': file_path,
            'success': True,
            'message': f'Successfully {action}ed file: {file_path}'
        }

# Usage example
def smart_assistant_with_dynamic_tools():
    selector = DynamicToolSelector()
    
    user_requests = [
        "What's the weather like in New York and calculate 15% tip on $50?",
        "Search for recent AI developments and save the results to a file",
        "Calculate the area of a circle with radius 10 and check weather in Tokyo"
    ]
    
    for request in user_requests:
        print(f"\nUser Request: {request}")
        
        # Analyze which tools might be needed
        relevant_tools = selector.analyze_intent(request)
        print(f"Relevant Tools: {relevant_tools}")
        
        # Get function definitions for the LLM
        tool_definitions = selector.get_tool_definitions(relevant_tools)
        print(f"Tool Definitions: {len(tool_definitions)} functions available")
        
        # In a real implementation, you would pass these to your LLM
        # The LLM would then decide which functions to call and with what parameters

### Enterprise Integration Example

```python
import asyncio
import json
from typing import Dict, List, Any
from dataclasses import dataclass
from datetime import datetime

@dataclass
class FunctionResult:
    """所有函数调用的标准结果格式"""
    success: bool
    data: Any = None
    error: str = None
    execution_time: float = 0.0
    timestamp: datetime = None

class EnterpriseAIAgent:
    """具有全面函数调用能力的生产级AI代理"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.functions = {}
        self.audit_log = []
        self.rate_limiters = {}
        
        # 初始化核心业务函数
        self._register_core_functions()
    
    def _register_core_functions(self):
        """注册所有可用的业务函数"""
        
        # CRM函数
        self.register_function(
            name="get_customer_info",
            description="从CRM检索客户信息",
            parameters={
                "customer_id": {"type": "string", "required": True},
                "include_history": {"type": "boolean", "default": False}
            },
            handler=self._get_customer_info,
            rate_limit=100  # 每分钟调用次数
        )
        
        # 销售函数
        self.register_function(
            name="create_sales_opportunity",
            description="创建新的销售机会",
            parameters={
                "customer_id": {"type": "string", "required": True},
                "product_id": {"type": "string", "required": True},
                "estimated_value": {"type": "number", "required": True},
                "expected_close_date": {"type": "string", "required": True}
            },
            handler=self._create_sales_opportunity,
            rate_limit=50
        )
        
        # 分析函数
        self.register_function(
            name="generate_sales_report",
            description="生成销售业绩报告",
            parameters={
                "period": {"type": "string", "enum": ["daily", "weekly", "monthly", "quarterly"]},
                "region": {"type": "string", "required": False},
                "product_category": {"type": "string", "required": False}
            },
            handler=self._generate_sales_report,
            rate_limit=10
        )
        
        # 通知函数
        self.register_function(
            name="send_notification",
            description="向团队成员发送通知",
            parameters={
                "recipients": {"type": "array", "items": {"type": "string"}},
                "message": {"type": "string", "required": True},
                "priority": {"type": "string", "enum": ["low", "medium", "high"], "default": "medium"},
                "channel": {"type": "string", "enum": ["email", "slack", "teams"], "default": "email"}
            },
            handler=self._send_notification,
            rate_limit=200
        )
    
    def register_function(self, name: str, description: str, parameters: Dict, 
                         handler: callable, rate_limit: int = 60):
        """向代理注册新函数"""
        self.functions[name] = {
            'description': description,
            'parameters': parameters,
            'handler': handler,
            'rate_limit': rate_limit,
            'call_count': 0,
            'last_reset': datetime.now()
        }
    
    async def execute_function(self, function_name: str, parameters: Dict) -
请提供需要翻译的 Markdown 文件内容，我将根据规则进行翻译。
"""执行一个具有全面错误处理和日志记录的函数"""
start_time = datetime.now()

try:
    # 验证函数是否存在
    if function_name not in self.functions:
        return FunctionResult(
            success=False,
            error=f"未找到函数 '{function_name}'",
            timestamp=start_time
        )
    
    # 检查速率限制
    if not self._check_rate_limit(function_name):
        return FunctionResult(
            success=False,
            error=f"函数 '{function_name}' 的速率限制已超出",
            timestamp=start_time
        )
    
    # 验证参数
    validation_result = self._validate_parameters(function_name, parameters)
    if not validation_result.success:
        return validation_result
    
    # 执行函数
    func_info = self.functions[function_name]
    handler = func_info['handler']
    
    if asyncio.iscoroutinefunction(handler):
        result_data = await handler(**parameters)
    else:
        result_data = handler(**parameters)
    
    execution_time = (datetime.now() - start_time).total_seconds()
    
    result = FunctionResult(
        success=True,
        data=result_data,
        execution_time=execution_time,
        timestamp=start_time
    )
    
    # 记录成功的执行
    self._log_function_call(function_name, parameters, result)
    
    return result
    
except Exception as e:
    execution_time = (datetime.now() - start_time).total_seconds()
    result = FunctionResult(
        success=False,
        error=str(e),
        execution_time=execution_time,
        timestamp=start_time
    )
    
    # 记录失败的执行
    self._log_function_call(function_name, parameters, result)
    
    return result

def _check_rate_limit(self, function_name: str) -> bool:
    """检查函数调用是否在速率限制内"""
    func_info = self.functions[function_name]
    now = datetime.now()
    
    # 如果已过一分钟，重置计数器
    if (now - func_info['last_reset']).seconds >= 60:
        func_info['call_count'] = 0
        func_info['last_reset'] = now
    
    # 检查是否在限制内
    if func_info['call_count'] >= func_info['rate_limit']:
        return False
    
    func_info['call_count'] += 1
    return True

def _validate_parameters(self, function_name: str, parameters: Dict) -> FunctionResult:
    """验证函数参数"""
    func_params = self.functions[function_name]['parameters']
    
    # 检查必需参数
    for param_name, param_info in func_params.items():
        if param_info.get('required', False) and param_name not in parameters:
            return FunctionResult(
                success=False,
                error=f"缺少必需参数: {param_name}"
            )
    
    # 验证参数类型和约束
    for param_name, value in parameters.items():
        if param_name in func_params:
            param_info = func_params[param_name]
            
            # 类型验证
            expected_type = param_info.get('type')
            if expected_type == 'string' and not isinstance(value, str):
                return FunctionResult(
                    success=False,
                    error=f"参数 '{param_name}' 必须是字符串"
                )
            elif expected_type == 'number' and not isinstance(value, (int, float)):
                return FunctionResult(
                    success=False,
                    error=f"参数 '{param_name}' 必须是数字"
                )
            elif expected_type == 'boolean' and not isinstance(value, bool):
                return FunctionResult(
                    success=False,
                    error=f"参数 '{param_name}' 必须是布尔值"
                )
            
            # 枚举验证
            if 'enum' in param_info and value not in param_info['enum']:
                return FunctionResult(
                    success=False,
                    error=f"参数 '{param_name}' 必须是以下之一: {param_info['enum']}"
                )
    
    return FunctionResult(success=True)

def _log_function_call(self, function_name: str, parameters: Dict, result: FunctionResult):
    """记录函数调用以供审计"""
    log_entry = {
        'timestamp': result.timestamp.isoformat(),
        'function_name': function_name,
        'parameters': parameters,
        'success': result.success,
        'execution_time': result.execution_time,
        'error': result.error if not result.success else None
    }
    
    self.audit_log.append(log_entry)
    
    # 可选：写入外部日志系统
    if self.config.get('enable_external_logging', False):
        self._write_to_external_log(log_entry)

def _write_to_external_log(self, log_entry: Dict):
    """将日志条目写入外部日志系统"""
    # 实现方式取决于您的日志基础设施
    # 例如，发送到 ELK 堆栈、CloudWatch 等
    pass

# 业务功能实现
async def _get_customer_info(self, customer_id: str, include_history: bool = False) -> Dict:
    """从 CRM 系统中检索客户信息"""
    # 模拟数据库/API 调用
    await asyncio.sleep(0.1)  # 模拟网络延迟
    
    customer_data = {
        'customer_id': customer_id,
        'name': 'John Doe',
        'email': 'john.doe@example.com',
        'phone': '+1-555-0123',
        'status': 'active',
        'tier': 'premium'
    }
    
    if include_history:
        customer_data['purchase_history'] = [
            {'date': '2024-01-15', 'product': 'Product A', 'amount': 1500},
            {'date': '2024-03-22', 'product': 'Product B', 'amount': 2300}
        ]
    
    return customer_data

async def _create_sales_opportunity(self, customer_id: str, product_id: str, 
                                  estimated_value: float, expected_close_date: str) -> Dict:
    """创建新的销售机会"""
    # 模拟 CRM API 调用
    await asyncio.sleep(0.2)
    
    opportunity_id = f"OPP-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    return {
        'opportunity_id': opportunity_id,
        'customer_id': customer_id,
        'product_id': product_id,
        'estimated_value': estimated_value,
        'expected_close_date': expected_close_date,
        'status': 'open',
        'created_date': datetime.now().isoformat()
    }

async def _generate_sales_report(self, period: str, region: str = None, 
                               product_category: str = None) -> Dict:
    """生成全面的销售报告"""
    # 模拟数据聚合
    await asyncio.sleep(0.5)
    
    return {
        'report_id': f"RPT-{datetime.now().strftime('%Y%m%d%H%M%S')}",
        'period': period,
        'region': region,
        'product_category': product_category,
        'total_sales': 125000.00,
        'total_opportunities': 45,
        'conversion_rate': 0.67,
        'top_products': [
            {'product_id': 'PROD-001', 'sales': 45000},
            {'product_id': 'PROD-002', 'sales': 32000}
        ],
        'generated_at': datetime.now().isoformat()
    }

async def _send_notification(self, recipients: List[str], message: str, 
                           priority: str = 'medium', channel: str = 'email') -> Dict:
    """通过指定渠道发送通知"""
    # 模拟通知服务调用
    await asyncio.sleep(0.1)
    
    notification_id = f"NOTIF-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    return {
        'notification_id': notification_id,
        'recipients': recipients,
        'channel': channel,
        'priority': priority,
        'status': 'sent',
        'sent_at': datetime.now().isoformat()
    }

def get_function_definitions(self) -> List[Dict]:
    """获取所有已注册函数的 OpenAI 兼容函数定义"""
    definitions = []
    
    for func_name, func_info in self.functions.items():
        definition = {
            'name': func_name,
            'description': func_info['description'],
            'parameters': {
                'type': 'object',
                'properties': {},
                'required': []
            }
        }
        
        for param_name, param_info in func_info['parameters'].items():
            definition['parameters']['properties'][param_name] = {
                'type': param_info['type'],
                'description': param_info.get('description', '')
            }
            
            if 'enum' in param_info:
                definition['parameters']['properties'][param_name]['enum'] = param_info['enum']
            
            if 'default' in param_info:
                definition['parameters']['properties'][param_name]['default'] = param_info['default']
            
            if param_info.get('required', False):
                definition['parameters']['required'].append(param_name)
        
        definitions.append(definition)
    
    return definitions

# 企业集成使用示例
async def enterprise_demo():
    """演示企业 AI 代理的功能"""
    
    config = {
        'enable_external_logging': True,
        'max_concurrent_functions': 10,
        'default_timeout': 30
    }
    
    agent = EnterpriseAIAgent(config)
    
    # 示例 1：客户查询处理
    print("=== 客户查询处理 ===")
    
    # 获取客户信息
    result = await agent.execute_function(
        'get_customer_info',
        {'customer_id': 'CUST-12345', 'include_history': True}
    )
    
    if result.success:
        print(f"客户信息已检索: {result.data['name']}")
        print(f"执行时间: {result.execution_time:.3f}s")
    
    # 示例 2：创建销售机会
    print("\n=== 创建销售机会 ===")
    
    result = await agent.execute_function(
        'create_sales_opportunity',
        {
            'customer_id': 'CUST-12345',
            'product_id': 'PROD-001',
            'estimated_value': 15000.0,
            'expected_close_date': '2025-09-30'
        }
    )
    
    if result.success:
        print(f"销售机会已创建: {result.data['opportunity_id']}")
    
    # 示例 3：批量操作
    print("\n=== 批量操作 ===")
    
    tasks = [
        agent.execute_function('generate_sales_report', {'period': 'monthly'}),
        agent.execute_function('send_notification', {
            'recipients': ['manager@company.com'],
            'message': '新销售机会已创建',
            'priority': 'high',
            'channel': 'email'
        })
    ]
    
    results = await asyncio.gather(*tasks)
    
    for i, result in enumerate(results):
        if result.success:
            print(f"任务 {i+1} 成功完成")
        else:
            print(f"任务 {i+1} 失败: {result.error}")
    
    # 显示审计日志
    print(f"\n=== 审计日志 ({len(agent.audit_log)} 条目) ===")
    for entry in agent.audit_log[-3:]:  # 显示最近 3 条
        print(f"{entry['timestamp']}: {entry['function_name']} - {'成功' if entry['success'] else '失败'}")

# 运行企业演示
# asyncio.run(enterprise_demo())

## 结论

在小型语言模型中调用函数代表了从静态 AI 助手到动态、强大代理的范式转变。本教程涵盖了以下内容：

### 关键要点

1. **基础理解**：函数调用使 SLM 能够通过连接外部工具和服务扩展其能力。
2. **灵活实现**：从自定义模板的低级实现到 Qwen-Agent 和 Foundry Local 等高级框架，存在多种方法。
3. **生产环境考虑**：企业部署需要关注错误处理、速率限制、安全性和审计日志。
4. **性能优化**：合理的函数设计、高效的执行和智能缓存可以显著提高响应速度。

### 未来方向

随着 SLM 技术的不断发展，我们可以期待：

- **更高的函数调用准确性**：更好的意图识别和参数提取
- **增强的并行处理能力**：更复杂的多函数编排
- **更好的集成标准**：工具集成的标准化协议
- **高级安全功能**：更强的身份验证和授权机制
- **扩展的生态系统**：不断增长的预构建函数和集成库

### 入门指南

要在您的项目中实现函数调用：

1. **从简单开始**：从基本的单函数场景入手
2. **选择框架**：在直接实现（Ollama/Phi-4）或框架辅助（Qwen-Agent）之间选择
3. **仔细设计函数**：专注于清晰、文档齐全的函数定义
4. **实现错误处理**：从一开始就构建健壮的错误处理
5. **逐步扩展**：随着经验的积累，从简单场景扩展到复杂场景

函数调用将 SLM 从令人印象深刻的文本生成器转变为能够解决实际问题的实用 AI 代理。通过遵循本教程中概述的模式和实践，您可以构建强大、可靠的 AI 系统，超越传统的聊天界面。

### 资源和参考资料
- **Phi-4 模型**: [Hugging Face 集合](https://huggingface.co/collections/microsoft/phi-4-677e9380e514feb5577a40e4)
- **Qwen3 文档**: [官方 Qwen 文档](https://qwen.readthedocs.io/)
- **Ollama**: [官方网站](https://ollama.com/)
- **Foundry Local**: [GitHub 仓库](https://github.com/microsoft/Foundry-Local)
- **函数调用最佳实践**: [Hugging Face 指南](https://huggingface.co/docs/hugs/en/guides/function-calling)

请记住，函数调用是一个不断发展的领域，及时了解您选择的框架和模型的最新动态将有助于您构建更高效的 AI 代理。


## ➡️ 接下来

- [03: 模型上下文协议 (MCP) 集成](./03.IntroduceMCP.md)

**免责声明**：  
本文档使用AI翻译服务 [Co-op Translator](https://github.com/Azure/co-op-translator) 进行翻译。虽然我们尽力确保翻译的准确性，但请注意，自动翻译可能包含错误或不准确之处。原始语言的文档应被视为权威来源。对于重要信息，建议使用专业人工翻译。我们不对因使用此翻译而产生的任何误解或误读承担责任。