大模型function calling：让AI函数调用更智能、更高效
随着大语言模型（LLM）的快速发展，其在实际应用中的能力越来越受到关注。Function Calling 是一种新兴的技术，允许大模型与外部工具或API进行交互，从而扩展了模型的功能边界。本文将深入探讨 Function Calling 的原理、应用场景以及如何通过优化实现更智能、高效的函数调用。
1. 什么是 Function Calling？
unction Calling 是一种机制，允许大语言模型动态调用外部函数或API，以完成特定任务。例如，当用户提问“今天的天气如何？”时，模型可以通过调用天气API获取实时数据并返回结果。这种方式不仅提升了模型的实用性，还使其能够处理复杂的多步骤任务。
- 关键特点：
- 动态调用：根据用户需求调用合适的函数。
- 上下文感知：结合对话上下文选择最佳函数。
- 扩展性强：支持多种API和服务集成。
2. Function Calling 的核心原理
- 用户输入问题或指令。
- 模型分析输入，生成 JSON 格式的函数调用请求。
- 后端解析请求，调用对应函数或API。
- 将函数返回的结果传递回模型。
- 模型基于结果生成最终回答。
3.基本操作
llm使用的是llama3.2:3b的模型，有一些模型是不支持tools的。
# 安装依赖包
pip isntall ollama
pip install yfinance
"""
1.将问题输出到llm模型中
2.模型解析问题中参数，将需要回调的函数和参数输出。
3.将回调结果作为上下文输入llm模型中验证
"""
import ollama
from ollama import Client
client = Client(
host='http://192.168.3.203:11434/',
headers={'x-some-header': 'some-value'}
)
def add_two_numbers(a: int, b: int) -> int:
"""
Add two numbers
Args:
a (int): The first number
b (int): The second number
Returns:
int: The sum of the two numbers
"""
return int(a) + int(b)
def subtract_two_numbers(a: int, b: int) -> int:
"""
Subtract two numbers
"""
return int(a) - int(b)
# Tools can still be manually defined and passed into chat
subtract_two_numbers_tool = {
'type': 'function',
'function': {
'name': 'subtract_two_numbers',
'description': 'Subtract two numbers',
'parameters': {
'type': 'object',
'required': ['a', 'b'],
'properties': {
'a': {'type': 'integer', 'description': 'The first number'},
'b': {'type': 'integer', 'description': 'The second number'},
},
},
},
}
messages = [{'role': 'user', 'content': 'What is three plus one?'}]
print('Prompt:', messages[0]['content'])
available_functions = {
'add_two_numbers': add_two_numbers,
'subtract_two_numbers': subtract_two_numbers,
}
def main():
response = client.chat(
'llama3.2:3b',
messages=messages,
tools=[add_two_numbers, subtract_two_numbers_tool],
)
if response.message.tool_calls:
# There may be multiple tool calls in the response
for tool in response.message.tool_calls:
# Ensure the function is available, and then call it
if function_to_call := available_functions.get(tool.function.name):
print('Calling function:', tool.function.name)
print('Arguments:', tool.function.arguments)
output = function_to_call(**tool.function.arguments)
print('Function output:', output)
else:
print('Function', tool.function.name, 'not found')
# Only needed to chat with the model using the tool call results
if response.message.tool_calls:
# Add the function response to messages for the model to use
messages.append(response.message)
messages.append({'role': 'tool', 'content': str(output), 'name': tool.function.name})
# Get final response from model with function outputs
final_response = client.chat('llama3.2:3b', messages=messages)
print('Final response:', final_response.message.content)
else:
print('No tool calls returned from model')
if __name__ == '__main__':
main()
"""
结果
Prompt: What is three plus one?
Calling function: add_two_numbers
Arguments: {'a': 3, 'b': 1}
Function output: 4
"""
"""
1.通过llm模型获取解析回调函数和函数的参数
2.通过yfinance库获取公司最新股票信息
"""
from ollama import Client
import yfinance as yf
client = Client(
host='http://192.168.3.203:11434',
headers={'x-some-header': 'some-value'}
)
def get_current_stock_price(ticker_symbol):
# 定义函数
stock = yf.Ticker(ticker_symbol)
current_price = stock.history(period='1d')['Close'].iloc[0]
return current_price
# 本地测试
# data = get_current_stock_price("MSFT")
# print(data)
# 工具函数请求参数
tools = [{'type': 'function',
'function': {
'name': 'get_current_stock_price',
'description': 'Get the current price for a stock',
'parameters': {
'type': 'object',
'properties': {
'ticker_symbol': {
'type': 'string',
'description': 'The ticker symbol of the stock'
}
}
},
'required': ['ticker_symbol']
}
}]
response = client.chat(model='llama3.2:3b',
messages=[{
'role': 'user',
'content': 'What is the current price of MSFT'
}],
# provide a tool to get the current price of a stock
tools=tools
)
print(response['message']['tool_calls'])
# [ToolCall(function=Function(name='get_current_stock_price', arguments={'ticker_symbol': 'MSFT'}))]
# 模拟函数库
function_map = {'get_current_stock_price': get_current_stock_price}
def call_function_safely(response, function_map):
# 模型结果回调函数
tool_call = response['message']['tool_calls'][0]
function_name = tool_call['function']['name']
arguments = tool_call['function']['arguments']
function_to_call = function_map.get(function_name)
if function_to_call:
try:
result = function_to_call(**arguments)
print(f"The current price of {arguments['ticker_symbol']} is : {result}")
except TypeError as e:
print(f"Argument error: {e}")
else:
print(f"{function_name} is not a recognized function")
call_function_safely(response, function_map)
4.实现自动发送邮件
import os
from dotenv import load_dotenv
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import json
from openai import OpenAI
GPT_MODEL = "llama3.2:3b"
load_dotenv(dotenv_path="/Users/wuzhibin/workspace/pythonDemo/ollama_demo/.env")
# 大模型密钥
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
# 邮箱授权码
AUTHORIZATION_CODE = os.getenv("AUTHORIZATION_CODE")
print("1111", OPENAI_API_KEY, AUTHORIZATION_CODE)
client = OpenAI(base_url="http://192.168.3.203:11434/v1", api_key=OPENAI_API_KEY)
tools = [
{
"type": "function",
"function": {
"name": "send_email",
"description": "Send an email to the specified email with the subject and content",
"parameters":{
"type": "object",
"properties": {
"FromEmail": {
"type": "string",
"description": "The email address, eg., rememeber0101@126.com",
},
"Subject": {
"type": "string",
"description": "Subject of the email",
},
"Body": {
"type": "string",
"description": "The content of the email",
},
"Recipients": {
"type": "string",
"description": "The recipients' email addresses",
}
},
"required": ["FromEmail", "Subject", "Body", "Recipients"],
},
}
}
]
def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice=tool_choice,
)
return response
except Exception as e:
print("Unable to generate ChatCompletion response")
print(f"Exception: {e}")
return e
def send_email(sender_email, sender_authorization_code, recipient_email, subject, body):
# 创建 MIMEMultipart 对象
message = MIMEMultipart()
message["From"] = sender_email
message["To"] = recipient_email
message["Subject"] = subject
message.attach(MIMEText(body, "plain"))
# 创建 SMTP_SSL 会话
with smtplib.SMTP_SSL("smtp.126.com", 465) as server:
server.login(sender_email, sender_authorization_code)
text = message.as_string()
server.sendmail(sender_email, recipient_email, text)
def main():
messages = []
while True:
msg = input("【You】: ")
messages.append({"role": "user", "content": msg})
response = chat_completion_request(
messages=messages,
tools=tools
)
print(response)
if content := response.choices[0].message.content:
print(f"【AI】: {content}")
messages.append({"role": "assistant", "content": content})
else:
fn_name = response.choices[0].message.tool_calls[0].function.name
fn_args = response.choices[0].message.tool_calls[0].function.arguments
# print(f"【Debug info】: fn_name - {fn_name}")
# print(f"【Debug info】: fn_args - {fn_args}")
if fn_name == "send_email":
try:
args = json.loads(fn_args)
# 返回将要发送的邮件内容给用户确认
print("【AI】: 邮件内容如下：")
print(f"发件人: {args['FromEmail']}")
print(f"收件人: {args['Recipients']}")
print(f"主题: {args['Subject']}")
print(f"内容: {args['Body']}")
confirm = input("AI: 确认发送邮件吗？ (yes/no): ").strip().lower()
if confirm == "yes":
send_email(
sender_email=args["FromEmail"],
sender_authorization_code=AUTHORIZATION_CODE,
recipient_email=args["Recipients"],
subject=args["Subject"],
body=args["Body"],
)
print("邮件已发送，还需要什么帮助吗？")
messages.append({"role": "assistant", "content": "邮件已发送，还需要什么帮助吗？"})
else:
print("邮件发送已取消，还需要什么帮助吗？")
messages.append({"role": "assistant", "content": "邮件发送已取消，还需要什么帮助吗？"})
except Exception as e:
print(f"发送邮件时出错：{e}")
messages.append({"role": "assistant", "content": "抱歉，功能异常！"})
if __name__ == "__main__":
main()
# 帮我发送一封邮件 发件人: xxxxxxx@126.com, 收信人：xxxxxxx@qq.com, 发送内容：写着一封来自未来胖虎的问候邮件，主题：来自未来的问候
