用戶提問 → 模型思考 → 返回文本回答

用戶提問 → 模型思考 → 決定調(diào)用什么工具 → 執(zhí)行工具 → 獲取結(jié)果 → 返回最終回答

┌─────────┐     ┌─────────┐     ┌──────────┐     ┌─────────┐
│  用戶    │     │  大模型  │     │  工具函數(shù) │     │  大模型  │
│  提問    │────?│  判斷    │────?│  執(zhí)行    │────?│  總結(jié)    │──? 最終回答
│         │     │  需要工具 │     │  并返回   │     │  回答    │
└─────────┘     └─────────┘     └──────────┘     └─────────┘

pip install openai python-dotenv

OPENAI_API_KEY=sk-your-key-here
# 也可以用兼容 OpenAI 接口的國(guó)內(nèi)模型
# OPENAI_BASE_URL=https://api.deepseek.com

# weather_tool.py
import json
from datetime import datetime

# 模擬天氣數(shù)據(jù)（實(shí)際項(xiàng)目中替換為真實(shí) API）
WEATHER_DATA = {
    "北京": {"temp": 22, "weather": "晴", "humidity": 45, "wind": "北風(fēng)3級(jí)"},
    "上海": {"temp": 26, "weather": "多云", "humidity": 72, "wind": "東南風(fēng)2級(jí)"},
    "深圳": {"temp": 30, "weather": "陣雨", "humidity": 85, "wind": "南風(fēng)3級(jí)"},
    "成都": {"temp": 20, "weather": "陰", "humidity": 68, "wind": "微風(fēng)"},
    "杭州": {"temp": 24, "weather": "晴轉(zhuǎn)多云", "humidity": 60, "wind": "東風(fēng)2級(jí)"},
}

def get_weather(city: str) -> str:
    """查詢指定城市的天氣信息

    Args:
        city: 城市名稱，如"北京"、"上海"

    Returns:
        天氣信息的 JSON 字符串
    """
    if city in WEATHER_DATA:
        data = WEATHER_DATA[city]
        return json.dumps({
            "city": city,
            "temp": data["temp"],
            "weather": data["weather"],
            "humidity": data["humidity"],
            "wind": data["wind"],
            "update_time": datetime.now().strftime("%Y-%m-%d %H:%M")
        }, ensure_ascii=False)
    return json.dumps({"error": f"未找到城市：{city}"}, ensure_ascii=False)

# tools定義：告訴模型有哪些工具可用
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查詢指定城市的實(shí)時(shí)天氣信息，包括溫度、天氣狀況、濕度、風(fēng)力等",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "要查詢的城市名稱，如'北京'、'上海'、'深圳'"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# main.py
import json
from openai import OpenAI
from weather_tool import get_weather

client = OpenAI()  # 自動(dòng)讀取環(huán)境變量

# 工具映射表：函數(shù)名 → 實(shí)際函數(shù)
TOOL_MAP = {
    "get_weather": get_weather,
}

def chat_with_tools(user_message: str) -> str:
    """帶工具調(diào)用能力的對(duì)話"""
    messages = [
        {"role": "system", "content": "你是一個(gè)智能助手，可以幫用戶查詢天氣信息?；卮饡r(shí)請(qǐng)用簡(jiǎn)潔自然的語(yǔ)言。"},
        {"role": "user", "content": user_message}
    ]

    # 第一次調(diào)用：讓模型判斷是否需要工具
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # 自動(dòng)決定是否調(diào)用工具
    )

    message = response.choices[0].message

    # 如果模型決定不調(diào)用工具，直接返回回答
    if not message.tool_calls:
        return message.content

    # 模型決定調(diào)用工具，逐個(gè)執(zhí)行
    messages.append(message)  # 把模型的工具調(diào)用記錄加入對(duì)話

    for tool_call in message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)

        print(f"  [工具調(diào)用] {func_name}({func_args})")

        # 執(zhí)行對(duì)應(yīng)工具
        if func_name in TOOL_MAP:
            result = TOOL_MAP[func_name](**func_args)
        else:
            result = json.dumps({"error": f"未知工具：{func_name}"})

        # 把工具結(jié)果加入對(duì)話
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result
        })

    # 第二次調(diào)用：讓模型根據(jù)工具結(jié)果生成最終回答
    final_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages
    )

    return final_response.choices[0].message.content


# 測(cè)試
if __name__ == "__main__":
    questions = [
        "北京今天天氣怎么樣？",
        "上海和深圳哪個(gè)城市更熱？",
        "推薦一個(gè)適合出游的城市"
    ]

    for q in questions:
        print(f"\n提問：{q}")
        answer = chat_with_tools(q)
        print(f"回答：{answer}")
        print("-" * 50)

提問：北京今天天氣怎么樣？
  [工具調(diào)用] get_weather({'city': '北京'})
回答：北京今天天氣晴朗，氣溫22°C，濕度45%，北風(fēng)3級(jí)。整體來(lái)說天氣不錯(cuò)，
適合外出活動(dòng)，建議適當(dāng)補(bǔ)水，風(fēng)力稍大注意防護(hù)。
--------------------------------------------------
提問：上海和深圳哪個(gè)城市更熱？
  [工具調(diào)用] get_weather({'city': '上海'})
  [工具調(diào)用] get_weather({'city': '深圳'})
回答：深圳更熱！深圳目前氣溫30°C，而上海是26°C，深圳比上海高了4度。
不過上海濕度更大（72% vs 85%），體感上深圳更悶熱一些。
--------------------------------------------------

# tools.py
import json
import math
from datetime import datetime

# ========== 工具 1：數(shù)學(xué)計(jì)算器 ==========
def calculator(expression: str) -> str:
    """安全地計(jì)算數(shù)學(xué)表達(dá)式

    Args:
        expression: 數(shù)學(xué)表達(dá)式，如 "2 + 3 * 4" 或 "sqrt(16)"
    """
    # 安全白名單：只允許數(shù)學(xué)函數(shù)和數(shù)字運(yùn)算
    ALLOWED_NAMES = {
        "abs": abs, "round": round, "min": min, "max": max,
        "sqrt": math.sqrt, "pow": pow, "pi": math.pi, "e": math.e,
        "sin": math.sin, "cos": math.cos, "tan": math.tan,
        "log": math.log, "log10": math.log10, "ceil": math.ceil,
        "floor": math.floor,
    }
    try:
        result = eval(expression, {"__builtins__": {}}, ALLOWED_NAMES)
        return json.dumps({"expression": expression, "result": result})
    except Exception as e:
        return json.dumps({"error": f"計(jì)算錯(cuò)誤：{str(e)}"})

# ========== 工具 2：時(shí)間查詢 ==========
def get_current_time(timezone_offset: int = 8) -> str:
    """獲取當(dāng)前時(shí)間

    Args:
        timezone_offset: 時(shí)區(qū)偏移量，默認(rèn)8（北京時(shí)間 UTC+8）
    """
    from datetime import timezone, timedelta
    tz = timezone(timedelta(hours=timezone_offset))
    now = datetime.now(tz)
    return json.dumps({
        "current_time": now.strftime("%Y-%m-%d %H:%M:%S"),
        "weekday": ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][now.weekday()],
        "timezone": f"UTC+{timezone_offset}"
    }, ensure_ascii=False)

# ========== 工具 3：?jiǎn)挝粨Q算 ==========
def unit_convert(value: float, from_unit: str, to_unit: str) -> str:
    """單位換算

    Args:
        value: 原始值
        from_unit: 原始單位（km, mile, kg, lb, celsius, fahrenheit 等）
        to_unit: 目標(biāo)單位
    """
    conversions = {
        ("km", "mile"): lambda v: v * 0.621371,
        ("mile", "km"): lambda v: v * 1.60934,
        ("kg", "lb"): lambda v: v * 2.20462,
        ("lb", "kg"): lambda v: v * 0.453592,
        ("celsius", "fahrenheit"): lambda v: v * 9/5 + 32,
        ("fahrenheit", "celsius"): lambda v: (v - 32) * 5/9,
        ("meter", "feet"): lambda v: v * 3.28084,
        ("feet", "meter"): lambda v: v * 0.3048,
    }

    key = (from_unit.lower(), to_unit.lower())
    if key in conversions:
        result = conversions[key](value)
        return json.dumps({
            "original": f"{value} {from_unit}",
            "converted": f"{round(result, 4)} {to_unit}"
        }, ensure_ascii=False)
    return json.dumps({"error": f"不支持的轉(zhuǎn)換：{from_unit} → {to_unit}"})

# 工具定義
ALL_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "計(jì)算數(shù)學(xué)表達(dá)式，支持加減乘除、三角函數(shù)、對(duì)數(shù)、冪運(yùn)算等。當(dāng)需要精確數(shù)值計(jì)算時(shí)使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "要計(jì)算的數(shù)學(xué)表達(dá)式，如 '2**10'、'sqrt(144)'、'sin(pi/4)'"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "獲取當(dāng)前的日期和時(shí)間",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone_offset": {
                        "type": "integer",
                        "description": "時(shí)區(qū)偏移量，默認(rèn)為8（北京時(shí)間）"
                    }
                },
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "unit_convert",
            "description": "在不同度量單位之間進(jìn)行換算，如長(zhǎng)度、重量、溫度等",
            "parameters": {
                "type": "object",
                "properties": {
                    "value": {
                        "type": "number",
                        "description": "要換算的數(shù)值"
                    },
                    "from_unit": {
                        "type": "string",
                        "description": "原始單位，如 km、mile、kg、lb、celsius、fahrenheit"
                    },
                    "to_unit": {
                        "type": "string",
                        "description": "目標(biāo)單位"
                    }
                },
                "required": ["value", "from_unit", "to_unit"]
            }
        }
    }
]

# 工具映射
TOOL_MAP = {
    "calculator": calculator,
    "get_current_time": get_current_time,
    "unit_convert": unit_convert,
}

# agent.py
import json
from openai import OpenAI

client = OpenAI()

def agent_chat(user_message: str, system_prompt: str = None, max_rounds: int = 5) -> str:
    """
    通用 Function Calling 引擎

    Args:
        user_message: 用戶輸入
        system_prompt: 系統(tǒng)提示詞
        max_rounds: 最大工具調(diào)用輪數(shù)（防止死循環(huán)）

    Returns:
        模型的最終回答
    """
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": user_message})

    for round_num in range(max_rounds):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=ALL_TOOLS,
            tool_choice="auto"
        )

        message = response.choices[0].message

        # 沒有工具調(diào)用，返回最終回答
        if not message.tool_calls:
            return message.content

        # 有工具調(diào)用，執(zhí)行所有工具
        messages.append(message)

        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"  ?? [調(diào)用] {func_name}({json.dumps(func_args, ensure_ascii=False)})")

            if func_name in TOOL_MAP:
                result = TOOL_MAP[func_name](**func_args)
            else:
                result = json.dumps({"error": f"未知工具：{func_name}"})

            print(f"  ?? [結(jié)果] {result}")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    return "抱歉，工具調(diào)用輪數(shù)已達(dá)上限。"


# 測(cè)試
if __name__ == "__main__":
    system_prompt = """你是一個(gè)全能AI助手，具備計(jì)算、時(shí)間查詢和單位換算能力。
回答問題時(shí)：
1. 需要精確計(jì)算時(shí)，使用計(jì)算器工具
2. 需要知道當(dāng)前時(shí)間時(shí)，使用時(shí)間查詢工具
3. 需要單位換算時(shí)，使用換算工具
4. 用簡(jiǎn)潔自然的中文回答"""

    test_cases = [
        "幫我算一下 2 的 20 次方是多少",
        "100華氏度等于多少攝氏度？",
        "現(xiàn)在幾點(diǎn)了？距離2026年國(guó)慶節(jié)還有多少天？",
        "一個(gè)馬拉松42.195公里等于多少英里？跑完大概消耗多少卡路里（按每公里60大卡算）？",
    ]

    for question in test_cases:
        print(f"\n{'='*60}")
        print(f"?? 提問：{question}")
        print(f"{'='*60}")
        answer = agent_chat(question, system_prompt)
        print(f"\n?? 回答：{answer}")

============================================================
?? 提問：100華氏度等于多少攝氏度？
============================================================
  ?? [調(diào)用] unit_convert({"value": 100, "from_unit": "fahrenheit", "to_unit": "celsius"})
  ?? [結(jié)果] {"original": "100 fahrenheit", "converted": "37.7778 celsius"}

?? 回答：100華氏度約等于 **37.78攝氏度**，接近人體正常體溫的溫度。
============================================================

# chain_agent.py
import json
from openai import OpenAI

client = OpenAI()

def chain_agent(user_message: str) -> str:
    """
    支持多輪工具調(diào)用的智能 Agent

    模型可以在一輪中調(diào)用多個(gè)工具，
    也可以在拿到結(jié)果后決定是否繼續(xù)調(diào)用更多工具。
    """
    messages = [
        {
            "role": "system",
            "content": """你是一個(gè)智能規(guī)劃助手。面對(duì)復(fù)雜問題時(shí)：
1. 先分析需要哪些信息
2. 逐步調(diào)用工具獲取信息
3. 基于工具結(jié)果進(jìn)行推理和回答

你必須主動(dòng)調(diào)用工具獲取數(shù)據(jù)，不要憑空猜測(cè)。"""
        },
        {"role": "user", "content": user_message}
    ]

    for round_num in range(10):  # 最多 10 輪工具調(diào)用
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=ALL_TOOLS,
            tool_choice="auto"
        )

        message = response.choices[0].message

        if not message.tool_calls:
            return message.content

        messages.append(message)
        print(f"\n--- 第 {round_num + 1} 輪工具調(diào)用 ---")

        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            result = TOOL_MAP[func_name](**func_args)
            print(f"  {func_name}({json.dumps(func_args, ensure_ascii=False)}) => {result}")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    return "工具調(diào)用輪數(shù)已達(dá)上限"


# 測(cè)試復(fù)雜鏈?zhǔn)秸{(diào)用
if __name__ == "__main__":
    result = chain_agent(
        "我想做一個(gè)蛋糕。幫我算一下：如果配方需要把350華氏度的烤箱預(yù)熱，"
        "那是多少攝氏度？另外幫我算一下，2.5磅面粉等于多少公斤？"
    )
    print(f"\n最終回答：\n{result}")

# 強(qiáng)制模型必須調(diào)用某個(gè)工具
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "get_weather"}}
)

# 禁止調(diào)用任何工具（純對(duì)話模式）
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    tools=tools,
    tool_choice="none"
)

# 當(dāng)用戶問"北京和上海天氣對(duì)比"時(shí)，
# 模型會(huì)在一個(gè) tool_calls 中返回兩個(gè)調(diào)用：
# tool_calls[0]: get_weather({"city": "北京"})
# tool_calls[1]: get_weather({"city": "上海"})

def safe_tool_call(func, **kwargs, max_retries=2):
    """帶重試的安全工具調(diào)用"""
    for attempt in range(max_retries + 1):
        try:
            result = func(**kwargs)
            return result
        except Exception as e:
            if attempt == max_retries:
                return json.dumps({"error": f"工具執(zhí)行失敗（已重試{max_retries}次）：{str(e)}"})
            print(f"  ?? 工具執(zhí)行失敗，正在重試（{attempt + 1}/{max_retries}）...")

定義工具 → 注冊(cè)給模型 → 模型自主決策 → 代碼執(zhí)行工具 → 結(jié)果返回模型 → 生成最終回答