构建AI智能体：七、Function Calling - 解锁大语言模型的实际行动力+案例解析

import json
import os
import dashscope
from dashscope.api_entities.dashscope_response import Role
# 从环境变量中，获取 DASHSCOPE_API_KEY
api_key = os.environ.get('DASHSCOPE_API_KEY')
dashscope.api_key = api_key
# 编写天气函数，调用本地的函数看看Function Call的效果，此处可替换为天气的接口看真实数据
def get_current_weather(location, unit="摄氏度"):
    # 获取指定地点的天气
    temperature = -1
    if '杭州' in location or 'Hangzhou' in location:
        temperature = 10
    if '上海' in location or 'Shanghai' in location:
        temperature = 36
    if '深圳' in location or 'Shenzhen' in location:
        temperature = 37
    weather_info = {
        "location": location,
        "temperature": temperature,
        "unit": unit,
        "forecast": ["晴天", "微风"],
    }
    return json.dumps(weather_info)
# 封装模型响应函数
def get_response(messages):
    try:
        response = dashscope.Generation.call(
            model='qwen-plus',
            messages=messages,
            functions=functions,
            result_format='message'
        )
        return response
    except Exception as e:
        print(f"API调用出错: {str(e)}")
        return None
# 使用function call进行QA
def run_conversation():
    query = "杭州的天气怎样"
    messages=[{"role": "user", "content": query}]
    # 得到第一次响应
    response = get_response(messages)
    if not response or not response.output:
        print("获取响应失败")
        return None
    print('response=', response)
    message = response.output.choices[0].message
    messages.append(message)
    print('message=', message)
    # Step 2, 判断用户是否要call function
    if hasattr(message, 'function_call') and message.function_call:
        function_call = message.function_call
        tool_name = function_call['name']
        # Step 3, 执行function call
        arguments = json.loads(function_call['arguments'])
        print('arguments=', arguments)
        tool_response = get_current_weather(
            location=arguments.get('location'),
            unit=arguments.get('unit'),
        )
        tool_info = {"role": "function", "name": tool_name, "content": tool_response}
        print('tool_info=', tool_info)
        messages.append(tool_info)
        print('messages=', messages)
        #Step 4, 得到第二次响应
        response = get_response(messages)
        if not response or not response.output:
            print("获取第二次响应失败")
            return None
        print('response=', response)
        message = response.output.choices[0].message
        return message
    return message
functions = [
    {
      'name': 'get_current_weather',
      'description': '获取指定地点，指定温度单位的天气信息',
      'parameters': {
            'type': 'object',
            'properties': {
                'location': {
                    'type': 'string',
                    'description': '城市名称，例如：杭州，上海，深圳。'
                },
                'unit': {'type': 'string', 'enum': ['celsius', 'fahrenheit']}
            },
        'required': ['location']
      }
    }
]
if __name__ == "__main__":
    result = run_conversation()
    if result:
        print("最终结果:", result)
    else:
        print("对话执行失败")

最终结果: {"role": "assistant", "content": "杭州当前的温度是10摄氏度，天气晴朗，微风。"}

{
  "function_call": {
    "name": "get_current_weather",
    "arguments": "{\"location\":\"杭州\"}"
  }
}

def get_current_weather(location, unit="摄氏度"):
    # 模拟真实天气API（根据城市返回预设温度）
    temperature = -1
    if '杭州' in location: temperature = 10
    if '上海' in location: temperature = 36
    if '深圳' in location: temperature = 37
    return json.dumps({
        "location": location,
        "temperature": temperature,
        "unit": unit,
        "forecast": ["晴天", "微风"]
    })

functions = [{
    'name': 'get_current_weather',
    'description': '获取指定地点，指定温度单位的天气信息',
    'parameters': {
        'type': 'object',
        'properties': {
            'location': {
                'type': 'string',
                'description': '城市名称，例如：杭州，上海，深圳。'
            },
            'unit': {'type': 'string', 'enum': ['celsius', 'fahrenheit']}
        },
        'required': ['location']
    }
}]

def run_conversation():
    # 1. 用户输入
    query = "杭州的天气怎样"
    messages = [{"role": "user", "content": query}]
    # 2. 首次模型调用
    response = get_response(messages)
    message = response.output.choices[0].message
    # 3. 检测函数调用
    if hasattr(message, 'function_call'):
        # 解析参数
        arguments = json.loads(message.function_call['arguments'])
        # 4. 执行天气函数
        tool_response = get_current_weather(**arguments)
        # 5. 将结果加入上下文
        messages.append({
            "role": "function", 
            "name": "get_current_weather",
            "content": tool_response
        })
        # 6. 二次模型调用生成最终回复
        final_response = get_response(messages)
        return final_response.output.choices[0].message