大模型function calling：让AI函数调用更智能、更高效

5e5a6b93-1b1f-425a-b01d-d15e85720563

随着大语言模型（LLM）的快速发展，其在实际应用中的能力越来越受到关注。Function Calling 是一种新兴的技术，允许大模型与外部工具或API进行交互，从而扩展了模型的功能边界。本文将深入探讨 Function Calling 的原理、应用场景以及如何通过优化实现更智能、高效的函数调用。

1. 什么是 Function Calling？

unction Calling 是一种机制，允许大语言模型动态调用外部函数或API，以完成特定任务。例如，当用户提问“今天的天气如何？”时，模型可以通过调用天气API获取实时数据并返回结果。这种方式不仅提升了模型的实用性，还使其能够处理复杂的多步骤任务。

关键特点：
- 动态调用：根据用户需求调用合适的函数。
- 上下文感知：结合对话上下文选择最佳函数。
- 扩展性强：支持多种API和服务集成。

2. Function Calling 的核心原理

workflow

用户输入问题或指令。
模型分析输入，生成 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, 发送内容：写着一封来自未来胖虎的问候邮件，主题：来自未来的问候