前提

需要大模型支持工具调用

工具提供

# 天气工具 https://open-meteo.com/ 贵阳经纬度 26.647 106.63
def get_weather(latitude, longitude):
    url = f"https://api.open-meteo.com/v1/forecast"
    params = {
        'latitude': latitude,
        'longitude': longitude,
        'current': 'temperature_2m,relative_humidity_2m,wind_speed_10m',
        'hourly': 'temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m,cloud_cover',
        'timezone': 'Asia/Shanghai',
        'forecast_days': 1
    }

    response = httpx.get(url, params=params)
    data = response.json()

    return f"当前温度: {data['current']['temperature_2m']}°C"


# 工具描述
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "根据指定坐标提供当地摄氏温度",
        "parameters": {
            "type": "object",
            "properties": {
                "latitude": {"type": "number"},
                "longitude": {"type": "number"},
            },
            "required": ["latitude", "longitude"],
            "additionalProperties": False # 禁止额外参数
        },
        "strict": True, # 严格模式：AI 必须严格遵守参数定义
    }
}]

关键点

completion = client.chat.completions.create(
    # 模型
    model=OPENAI_MODEL_NAME,
    # 消息
    messages=message_history,
    # 温度
    temperature=0.5,
    # 流式
    stream=True,
    # 工具
    tools=tools
)

首次AI模型会返回是否调用函数工具，当使用"role": "tool"时，必须提供tool_call_id来关联对应的工具调用

代码示例

from openai import OpenAI
import httpx
import json

OPENAI_BASE_URL = "https://api.deepseek.com"
OPENAI_API_KEY = "sk-xxxx"
OPENAI_MODEL_NAME = "deepseek-v4-pro"

# OPENAI_BASE_URL = "http://localhost:11434/v1"
# OPENAI_API_KEY = "ollama"
# OPENAI_MODEL_NAME = "qwen2.5-coder:7b"

client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)


# 天气工具
def get_weather(latitude, longitude):
    url = f"https://api.open-meteo.com/v1/forecast"
    params = {
        'latitude': latitude,
        'longitude': longitude,
        'current': 'temperature_2m,relative_humidity_2m,wind_speed_10m',
        'hourly': 'temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m,cloud_cover',
        'timezone': 'Asia/Shanghai',
        'forecast_days': 1
    }

    response = httpx.get(url, params=params)
    data = response.json()

    return f"当前温度: {data['current']['temperature_2m']}°C"


# 工具描述
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "根据指定坐标提供当地摄氏温度",
        "parameters": {
            "type": "object",
            "properties": {
                "latitude": {"type": "number"},
                "longitude": {"type": "number"},
            },
            "required": ["latitude", "longitude"],
            "additionalProperties": False
        },
        "strict": True,
    }
}]

message_history = []


def parse_tool_call(message):
    """兼容解析不同模型的工具调用格式"""
    # 标准 OpenAI 格式
    if message.tool_calls:
        return message.tool_calls

    # Qwen2.5-Coder 格式：工具调用在 content 中以 JSON 形式返回
    if message.content and '"name": "get_weather"' in message.content:
        try:
            # 尝试解析 JSON
            content = message.content.strip()
            # 处理可能的 markdown 代码块
            if content.startswith('```json'):
                content = content[7:]
            if content.startswith('```'):
                content = content[3:]
            if content.endswith('```'):
                content = content[:-3]

            data = json.loads(content)
            if data.get('name') == 'get_weather':
                # 构造标准格式的 tool_call
                class ToolCall:
                    def __init__(self, id, name, arguments):
                        self.id = id
                        self.function = type('obj', (object,), {
                            'name': name,
                            'arguments': json.dumps(arguments)
                        })()
                        self.type = 'function'

                return [ToolCall(
                    id="call_" + str(hash(str(data))),
                    name=data['name'],
                    arguments=data.get('arguments', {})
                )]
        except:
            pass

    return None


def chat(msg=None):
    # 如果有新消息，添加到历史
    if msg:
        message_history.append({
            "role": "user",
            "content": msg
        })

    # 调用模型
    completion = client.chat.completions.create(
        model=OPENAI_MODEL_NAME,
        messages=message_history,
        temperature=0.5,
        tools=tools
    )

    chatResp = completion
    message = chatResp.choices[0].message

    # 打印详细信息（兼容处理）
    print(chatResp.model_dump_json(indent=2))
    print(f"对话ID：{chatResp.id}")
    print(f"对话人：{message.role}")

    if hasattr(message, 'reasoning_content') and message.reasoning_content:
        print(f"思考：{message.reasoning_content}")

    print(f"回复内容：{message.content}")

    # 兼容获取工具调用
    tool_calls = parse_tool_call(message)
    print(f"工具：{tool_calls}")
    print(f"使用模型：{chatResp.model}")
    print(f"输入token：{chatResp.usage.prompt_tokens}")
    print(f"回复token：{chatResp.usage.completion_tokens}")
    print(f"交互总token：{chatResp.usage.total_tokens}")

    # 兼容处理 usage 中可能不存在的字段
    if hasattr(chatResp.usage, 'prompt_cache_hit_tokens') and chatResp.usage.prompt_cache_hit_tokens:
        print(f"命中缓存token：{chatResp.usage.prompt_cache_hit_tokens}")
    if hasattr(chatResp.usage, 'prompt_cache_miss_tokens') and chatResp.usage.prompt_cache_miss_tokens:
        print(f"未被命中token：{chatResp.usage.prompt_cache_miss_tokens}")

    print("=" * 60)

    # 检查是否有工具调用
    if tool_calls:
        # 添加助手的消息到历史
        message_history.append({
            "role": "assistant",
            "content": message.content
        })

        # 执行工具并添加结果
        for tool_call in tool_calls:
            if tool_call.function.name == "get_weather":
                func_arg = json.loads(tool_call.function.arguments)
                print(f"📊 调用天气API，参数: {func_arg}")
                func_result = get_weather(**func_arg)
                print(f"📊 工具结果: {func_result}")

                # 添加工具结果
                message_history.append({
                    "role": "user",  # Qwen 模型使用 user 角色继续对话
                    "content": f"天气查询结果：{func_result}"
                })

        # 继续对话，让模型根据工具结果生成最终回复
        print("\n🔄 生成最终回复...\n")
        chat()  # 不传参数，继续使用现有的历史

    else:
        # 没有工具调用，这是最终回复
        print(f"\n✅ 最终回复: {message.content}")

        # 保存助手回复到历史
        if message.content:
            # 避免重复添加（因为已经在上面添加过了）
            if not message_history or message_history[-1].get('content') != message.content:
                message_history.append({
                    "role": "assistant",
                    "content": message.content
                })

        return message.content


if __name__ == '__main__':
    result = chat("你好，今天天气如何，纬度26.647, 经度106.63")
    print(f"\n最终结果: {result}")