1

用户提问 → LLM 生成结构化的函数调用（JSON）→ 应用代码执行函数 → 结果返回给 LLM → LLM 生成最终回答

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

┌─────────────────────────────────────────────────────────┐
│                      用户输入                            │
│           "北京今天天气怎么样？顺便看看苹果股价"              │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                   LLM 推理阶段                            │
│                                                         │
│  模型分析用户意图，决定需要调用两个工具：                     │
│  1. get_weather(city="北京")                             │
│  2. get_stock_price(symbol="AAPL")                      │
│                                                         │
│  返回结构化的 tool_calls JSON                             │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                  应用代码执行阶段                           │
│                                                         │
│  解析 tool_calls，执行对应的函数                           │
│  get_weather("北京") → {"temp": 28, "weather": "晴"}     │
│  get_stock_price("AAPL") → {"price": 198.50, ...}       │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│               结果返回给 LLM                              │
│                                                         │
│  将工具执行结果作为 tool 角色的消息添加到对话中               │
│  LLM 基于这些真实数据生成最终回答                           │
│                                                         │
│  "北京今天晴天，气温28°C。苹果股价目前为$198.50，..."       │
└─────────────────────────────────────────────────────────┘

1

pip install openai>=2.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

"""
基础 Function Calling 示例
使用 GPT-5.4 查询天气
"""
import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 第一步：定义工具（Tool）的 JSON Schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 '北京'、'上海'、'New York'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认摄氏度"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 第二步：模拟工具的实际执行（真实项目中这里会调用天气 API）
def get_weather(city: str, unit: str = "celsius") -> dict:
    """模拟天气查询函数"""
    # 这里用模拟数据，实际项目中应调用真实 API
    mock_data = {
        "北京": {"temp": 28, "weather": "晴", "humidity": 45, "wind": "北风3级"},
        "上海": {"temp": 30, "weather": "多云", "humidity": 70, "wind": "东南风2级"},
        "广州": {"temp": 33, "weather": "阵雨", "humidity": 80, "wind": "南风3级"},
    }
    data = mock_data.get(city, {"temp": 25, "weather": "未知", "humidity": 50, "wind": "微风"})
    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32
    return {"city": city, **data}

# 第三步：发送请求，让模型决定是否需要调用工具
messages = [
    {"role": "system", "content": "你是一个有用的天气助手。如果用户询问天气，请调用 get_weather 函数获取实时数据。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

response = client.chat.completions.create(
    model="gpt-5.4",
    messages=messages,
    tools=tools,
    tool_choice="auto",  # auto: 模型自主决定 | required: 强制调用 | none: 不调用
)

# 第四步：检查模型是否要求调用工具
assistant_message = response.choices[0].message
print(f"模型回复内容: {assistant_message.content}")
print(f"工具调用请求: {assistant_message.tool_calls}")

if assistant_message.tool_calls:
    # 执行工具调用
    for tool_call in assistant_message.tool_calls:
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        print(f"\n执行函数: {function_name}")
        print(f"参数: {arguments}")

        # 调用实际函数
        result = get_weather(**arguments)
        print(f"结果: {result}")

        # 将工具执行结果添加到消息历史
        messages.append(assistant_message.model_dump())
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False)
        })

    # 第五步：用工具结果继续对话
    final_response = client.chat.completions.create(
        model="gpt-5.4",
        messages=messages,
        tools=tools,
    )
    print(f"\n最终回答: {final_response.choices[0].message.content}")
else:
    print(f"\n模型直接回答: {assistant_message.content}")

1
2
3
4
5
6
7
8

模型回复内容: None
工具调用请求: [ChatCompletionMessageToolCall(id='call_abc123', function=Function(arguments='{"city": "北京", "unit": "celsius"}', name='get_weather'), type='function')]

执行函数: get_weather
参数: {'city': '北京', 'unit': 'celsius'}
结果: {'city': '北京', 'temp': 28, 'weather': '晴', 'humidity': 45, 'wind': '北风3级'}

最终回答: 北京今天天气晴朗，气温 28°C，湿度 45%，北风3级。适合外出活动，建议做好防晒。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 1. auto - 模型自主判断是否需要工具（最常用）
tool_choice="auto"

# 2. none - 强制模型不使用工具
tool_choice="none"

# 3. required - 强制模型必须使用至少一个工具
tool_choice="required"

# 4. 指定特定函数 - 强制调用特定函数
tool_choice={"type": "function", "function": {"name": "get_weather"}}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282

"""
多工具 Agent：天气查询 + 股票价格查询
使用 GPT-5.4 构建，支持多轮对话
"""
import json
import os
import datetime
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


# ===================== 工具定义 =====================

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的当前天气信息，包括温度、天气状况、湿度和风力",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    },
                    "date": {
                        "type": "string",
                        "description": "查询日期，格式 YYYY-MM-DD，默认今天"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "查询股票的实时价格信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "股票代码，如 AAPL（苹果）、GOOGL（谷歌）、000001.SS（上证指数）"
                    },
                    "market": {
                        "type": "string",
                        "enum": ["us", "cn", "hk"],
                        "description": "市场：us（美股）、cn（A股）、hk（港股）"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算，支持加减乘除、百分比、汇率换算等",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '28 * 9/5 + 32' 或 '100 * (1 - 0.15)'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]


# ===================== 工具实现 =====================

def get_weather(city: str, date: str = None) -> dict:
    """查询天气（模拟实现）"""
    import random
    if date is None:
        date = datetime.date.today().isoformat()
    
    # 模拟不同城市的天气
    weather_patterns = {
        "北京": {"temp_range": (15, 35), "conditions": ["晴", "多云", "阴", "小雨"]},
        "上海": {"temp_range": (18, 33), "conditions": ["多云", "阴", "小雨", "晴"]},
        "深圳": {"temp_range": (22, 36), "conditions": ["晴", "多云", "雷阵雨", "阴"]},
        "东京": {"temp_range": (10, 28), "conditions": ["晴", "多云", "小雨"]},
        "New York": {"temp_range": (5, 30), "conditions": ["晴", "多云", "Rain", "Snow"]},
    }
    
    pattern = weather_patterns.get(city, {"temp_range": (15, 30), "conditions": ["晴", "多云"]})
    temp = random.randint(*pattern["temp_range"])
    condition = random.choice(pattern["conditions"])
    
    return {
        "city": city,
        "date": date,
        "temperature": f"{temp}°C",
        "condition": condition,
        "humidity": f"{random.randint(30, 90)}%",
        "wind": f"{random.choice(['微风', '轻风', '和风', '劲风'])} {random.randint(1, 6)}级"
    }


def get_stock_price(symbol: str, market: str = "us") -> dict:
    """查询股价（模拟实现）"""
    import random
    
    # 模拟股价数据
    mock_stocks = {
        "AAPL": {"name": "苹果公司", "base_price": 198.50},
        "GOOGL": {"name": "谷歌", "base_price": 175.20},
        "MSFT": {"name": "微软", "base_price": 420.80},
        "NVDA": {"name": "英伟达", "base_price": 880.50},
        "000001.SS": {"name": "上证指数", "base_price": 3250.00},
    }
    
    stock = mock_stocks.get(symbol, {"name": symbol, "base_price": 100.00})
    change_pct = random.uniform(-5, 5)
    current_price = round(stock["base_price"] * (1 + change_pct / 100), 2)
    
    return {
        "symbol": symbol,
        "name": stock["name"],
        "market": market,
        "current_price": f"${current_price}" if market == "us" else f"¥{current_price}",
        "change_pct": f"{change_pct:+.2f}%",
        "volume": f"{random.randint(10, 500)}M",
        "timestamp": datetime.datetime.now().isoformat()
    }


def calculate(expression: str) -> dict:
    """安全执行数学计算"""
    import ast
    import operator as op
    
    # 安全的运算符映射
    safe_ops = {
        ast.Add: op.add,
        ast.Sub: op.sub,
        ast.Mult: op.mul,
        ast.Div: op.truediv,
        ast.Pow: op.pow,
        ast.Mod: op.mod,
    }
    
    def _eval(node):
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = _eval(node.left)
            right = _eval(node.right)
            return safe_ops[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            return -_eval(node.operand)
        else:
            raise ValueError(f"不支持的表达式: {ast.dump(node)}")
    
    try:
        tree = ast.parse(expression, mode='eval')
        result = _eval(tree.body)
        return {"expression": expression, "result": result, "success": True}
    except Exception as e:
        return {"expression": expression, "error": str(e), "success": False}


# ===================== 工具路由 =====================

TOOL_MAP = {
    "get_weather": get_weather,
    "get_stock_price": get_stock_price,
    "calculate": calculate,
}


def execute_tool(tool_call) -> str:
    """执行工具调用并返回结果"""
    func_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    func = TOOL_MAP.get(func_name)
    if func is None:
        return json.dumps({"error": f"未知工具: {func_name}"})
    
    try:
        result = func(**arguments)
        return json.dumps(result, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": f"工具执行失败: {str(e)}"})


# ===================== Agent 循环 =====================

def run_agent(user_input: str, conversation_history: list = None, max_rounds: int = 5) -> str:
    """
    运行 Agent，支持多轮工具调用
    
    Args:
        user_input: 用户输入
        conversation_history: 对话历史
        max_rounds: 最大工具调用轮数（防止无限循环）
    
    Returns:
        最终回复文本
    """
    if conversation_history is None:
        conversation_history = [
            {
                "role": "system",
                "content": (
                    "你是一个智能助手，可以查询天气、股票价格，并进行数学计算。"
                    "请用自然、友好的方式回答用户的问题。"
                    "如果需要获取实时数据，请调用相应的工具。"
                    "你可以同时调用多个工具来获取完整信息。"
                )
            }
        ]
    
    conversation_history.append({"role": "user", "content": user_input})
    
    for round_num in range(max_rounds):
        response = client.chat.completions.create(
            model="gpt-5.4",
            messages=conversation_history,
            tools=tools,
            tool_choice="auto",
        )
        
        message = response.choices[0].message
        conversation_history.append(message.model_dump())
        
        # 如果没有工具调用，返回最终回答
        if not message.tool_calls:
            return message.content
        
        # 执行所有工具调用
        print(f"[第{round_num + 1}轮] 模型请求调用 {len(message.tool_calls)} 个工具:")
        for tc in message.tool_calls:
            print(f"  - {tc.function.name}({tc.function.arguments})")
        
        for tool_call in message.tool_calls:
            result = execute_tool(tool_call)
            print(f"  → 结果: {result[:100]}...")
            
            conversation_history.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
    
    return "抱歉，处理过程中出现了问题，请稍后再试。"


# ===================== 运行示例 =====================

if __name__ == "__main__":
    # 示例1：单工具调用
    print("=" * 60)
    print("示例1：查询天气")
    print("=" * 60)
    answer = run_agent("北京今天天气怎么样？")
    print(f"回答: {answer}\n")
    
    # 示例2：多工具并行调用
    print("=" * 60)
    print("示例2：同时查询天气和股票")
    print("=" * 60)
    answer = run_agent("帮我看看北京天气，还有苹果和英伟达的股价")
    print(f"回答: {answer}\n")
    
    # 示例3：需要计算的调用
    print("=" * 60)
    print("示例3：带计算的查询")
    print("=" * 60)
    answer = run_agent("如果我在北京穿短袖出门，需要多少度以上才合适？另外帮我算算华氏温度")
    print(f"回答: {answer}\n")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

"""
并行工具调用示例
同时获取多个城市的天气，使用 asyncio 并发执行
"""
import asyncio
import json
import time
from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor

client = OpenAI()

# 同样的工具定义...
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    }
]


def get_weather(city: str) -> dict:
    """模拟耗时的 API 调用"""
    import random
    time.sleep(0.5)  # 模拟网络延迟
    return {"city": city, "temp": random.randint(15, 35), "condition": "晴"}


# ===================== 方案1：串行执行 =====================
def run_serial(tool_calls) -> list:
    """逐个执行工具调用"""
    results = []
    for tc in tool_calls:
        args = json.loads(tc.function.arguments)
        result = get_weather(**args)
        results.append({"tool_call_id": tc.id, "content": json.dumps(result)})
    return results


# ===================== 方案2：并行执行（推荐） =====================
def run_parallel(tool_calls) -> list:
    """使用线程池并行执行工具调用"""
    results = []
    
    def execute(tc):
        args = json.loads(tc.function.arguments)
        result = get_weather(**args)
        return {"tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)}
    
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(execute, tc): tc for tc in tool_calls}
        for future in futures:
            results.append(future.result())
    
    return results


# ===================== 性能对比 =====================
def compare_performance():
    """对比串行和并行执行的性能"""
    # 模拟模型返回3个并行工具调用
    mock_tool_calls = [
        type('ToolCall', (), {
            'id': f'call_{i}',
            'function': type('Function', (), {
                'name': 'get_weather',
                'arguments': json.dumps({"city": city})
            })()
        })()
        for i, city in enumerate(["北京", "上海", "广州"])
    ]
    
    # 串行执行
    start = time.time()
    serial_results = run_serial(mock_tool_calls)
    serial_time = time.time() - start
    
    # 并行执行
    start = time.time()
    parallel_results = run_parallel(mock_tool_calls)
    parallel_time = time.time() - start
    
    print(f"串行执行耗时: {serial_time:.2f}s")
    print(f"并行执行耗时: {parallel_time:.2f}s")
    print(f"加速比: {serial_time/parallel_time:.1f}x")


if __name__ == "__main__":
    compare_performance()
    # 串行执行耗时: 1.50s
    # 并行执行耗时: 0.52s
    # 加速比: 2.9x

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

"""
异步并行工具调用
适合 FastAPI / aiohttp 等异步框架
"""
import asyncio
import json
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def async_get_weather(city: str) -> dict:
    """异步天气查询（实际中应使用 aiohttp 调用 API）"""
    import random
    await asyncio.sleep(0.5)  # 模拟异步网络请求
    return {"city": city, "temp": random.randint(15, 35), "condition": "晴"}


async def run_async_agent(user_input: str) -> str:
    """异步 Agent 主循环"""
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "查询指定城市的天气",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string", "description": "城市名称"}
                    },
                    "required": ["city"]
                }
            }
        }
    ]
    
    messages = [
        {"role": "system", "content": "你是一个天气助手。"},
        {"role": "user", "content": user_input}
    ]
    
    while True:
        response = await async_client.chat.completions.create(
            model="gpt-5.4",
            messages=messages,
            tools=tools,
        )
        
        message = response.choices[0].message
        messages.append(message.model_dump())
        
        if not message.tool_calls:
            return message.content
        
        # 并行执行所有工具调用
        tasks = []
        for tc in message.tool_calls:
            args = json.loads(tc.function.arguments)
            tasks.append(async_get_weather(**args))
        
        results = await asyncio.gather(*tasks)
        
        for tc, result in zip(message.tool_calls, results):
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result, ensure_ascii=False)
            })


# 使用示例
async def main():
    answer = await run_async_agent("北京、上海、广州天气分别怎么样？")
    print(answer)

# asyncio.run(main())

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

"""
流式响应 + 工具调用
处理 chunk 增量拼接的正确方式
"""
import json
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    }
]


def get_weather(city: str) -> dict:
    import random
    return {"city": city, "temp": random.randint(15, 35), "condition": "晴"}


def process_streaming_tool_calls(user_input: str) -> str:
    """
    处理包含工具调用的流式响应
    
    关键点：
    - tool_calls 的 arguments 是增量发送的，需要拼接
    - tool_calls 通过 index 字段区分不同的调用
    """
    messages = [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": user_input}
    ]
    
    # 存储所有工具调用的累积数据
    tool_calls_data = {}  # index -> {id, name, arguments_json}
    accumulated_content = ""
    
    stream = client.chat.completions.create(
        model="gpt-5.4",
        messages=messages,
        tools=tools,
        tool_choice="auto",
        stream=True,
    )
    
    for chunk in stream:
        if not chunk.choices:
            continue
        
        delta = chunk.choices[0].delta
        
        # 处理文本内容（正常回复）
        if delta.content:
            accumulated_content += delta.content
            print(delta.content, end="", flush=True)
        
        # 处理工具调用（增量拼接）
        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                
                if idx not in tool_calls_data:
                    tool_calls_data[idx] = {
                        "id": tc.id or "",
                        "name": "",
                        "arguments": ""
                    }
                
                # 拼接 ID
                if tc.id:
                    tool_calls_data[idx]["id"] = tc.id
                
                # 拼接函数名
                if tc.function and tc.function.name:
                    tool_calls_data[idx]["name"] = tc.function.name
                
                # 拼接参数（这是增量的！）
                if tc.function and tc.function.arguments:
                    tool_calls_data[idx]["arguments"] += tc.function.arguments
    
    print()  # 换行
    
    # 检查是否有工具调用
    if tool_calls_data:
        print(f"\n检测到 {len(tool_calls_data)} 个工具调用:")
        
        messages.append({
            "role": "assistant",
            "content": accumulated_content or None,
            "tool_calls": [
                {
                    "id": data["id"],
                    "type": "function",
                    "function": {
                        "name": data["name"],
                        "arguments": data["arguments"]
                    }
                }
                for data in tool_calls_data.values()
            ]
        })
        
        # 执行工具
        for idx, data in tool_calls_data.items():
            print(f"  调用: {data['name']}({data['arguments']})")
            args = json.loads(data["arguments"])
            result = get_weather(**args)
            print(f"  结果: {result}")
            
            messages.append({
                "role": "tool",
                "tool_call_id": data["id"],
                "content": json.dumps(result, ensure_ascii=False)
            })
        
        # 用工具结果继续
        print("\n最终回答: ", end="")
        final_stream = client.chat.completions.create(
            model="gpt-5.4",
            messages=messages,
            tools=tools,
            stream=True,
        )
        
        final_answer = ""
        for chunk in final_stream:
            if chunk.choices and chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                final_answer += content
                print(content, end="", flush=True)
        
        print()
        return final_answer
    
    return accumulated_content


# 使用
if __name__ == "__main__":
    answer = process_streaming_tool_calls("今天北京天气如何？")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242

"""
健壮的工具调用错误处理与重试
"""
import json
import time
import logging
from functools import wraps
from openai import OpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = OpenAI()


# ===================== 重试装饰器 =====================

def retry_with_backoff(max_retries=3, backoff_factor=2, retryable_exceptions=(Exception,)):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        wait_time = backoff_factor ** attempt
                        logger.warning(
                            f"工具调用失败 (尝试 {attempt + 1}/{max_retries + 1}): {e}. "
                            f"{wait_time}s 后重试..."
                        )
                        time.sleep(wait_time)
                    else:
                        logger.error(f"工具调用最终失败: {e}")
            raise last_exception
        return wrapper
    return decorator


# ===================== 带错误处理的工具 =====================

class ToolExecutionError(Exception):
    """工具执行错误"""
    def __init__(self, tool_name: str, message: str, retryable: bool = False):
        self.tool_name = tool_name
        self.message = message
        self.retryable = retryable
        super().__init__(f"{tool_name}: {message}")


@retry_with_backoff(max_retries=2, retryable_exceptions=(TimeoutError, ConnectionError))
def get_weather_with_retry(city: str) -> dict:
    """带重试的天气查询"""
    import random
    # 模拟偶发失败
    if random.random() < 0.3:  # 30% 失败率
        raise TimeoutError(f"天气 API 请求超时: {city}")
    
    return {"city": city, "temp": random.randint(15, 35), "condition": "晴"}


@retry_with_backoff(max_retries=2, retryable_exceptions=(TimeoutError,))
def get_stock_with_retry(symbol: str) -> dict:
    """带重试的股价查询"""
    import random
    if random.random() < 0.2:
        raise TimeoutError(f"股价 API 请求超时: {symbol}")
    
    return {"symbol": symbol, "price": round(random.uniform(50, 500), 2)}


# ===================== 参数验证 =====================

def validate_tool_arguments(tool_name: str, arguments: dict, schema: dict) -> dict:
    """
    验证工具参数
    
    Args:
        tool_name: 工具名称
        arguments: 实际参数
        schema: 参数 schema
    
    Returns:
        验证后的参数
    
    Raises:
        ValueError: 参数验证失败
    """
    properties = schema.get("parameters", {}).get("properties", {})
    required = schema.get("parameters", {}).get("required", [])
    
    # 检查必需参数
    for param in required:
        if param not in arguments:
            raise ValueError(f"工具 {tool_name} 缺少必需参数: {param}")
    
    # 检查参数类型
    for param, value in arguments.items():
        if param in properties:
            expected_type = properties[param].get("type")
            if expected_type == "string" and not isinstance(value, str):
                raise ValueError(f"参数 {param} 应为字符串类型")
            elif expected_type == "number" and not isinstance(value, (int, float)):
                raise ValueError(f"参数 {param} 应为数字类型")
    
    return arguments


# ===================== 完整的 Agent 循环（带错误处理） =====================

def run_robust_agent(user_input: str) -> str:
    """带完整错误处理的 Agent"""
    
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "查询天气",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string", "description": "城市名"}
                    },
                    "required": ["city"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "get_stock_price",
                "description": "查询股价",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "symbol": {"type": "string", "description": "股票代码"}
                    },
                    "required": ["symbol"]
                }
            }
        }
    ]
    
    TOOL_MAP = {
        "get_weather": get_weather_with_retry,
        "get_stock_price": get_stock_with_retry,
    }
    
    messages = [
        {"role": "system", "content": "你是一个助手。如果工具调用失败，请告知用户并建议重试。"},
        {"role": "user", "content": user_input}
    ]
    
    MAX_ROUNDS = 5
    MAX_RETRIES_PER_TOOL = 2
    
    for round_num in range(MAX_ROUNDS):
        try:
            response = client.chat.completions.create(
                model="gpt-5.4",
                messages=messages,
                tools=tools,
                tool_choice="auto",
            )
        except Exception as e:
            logger.error(f"LLM API 调用失败: {e}")
            return "抱歉，AI 服务暂时不可用，请稍后再试。"
        
        message = response.choices[0].message
        messages.append(message.model_dump())
        
        if not message.tool_calls:
            return message.content
        
        # 执行工具调用
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            try:
                arguments = json.loads(tool_call.function.arguments)
            except json.JSONDecodeError as e:
                # JSON 解析失败 - 返回错误给模型
                logger.error(f"参数 JSON 解析失败: {e}")
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps({
                        "error": f"参数格式错误: {tool_call.function.arguments}",
                        "suggestion": "请重新生成正确的参数"
                    })
                })
                continue
            
            tool_func = TOOL_MAP.get(func_name)
            if tool_func is None:
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps({"error": f"未知工具: {func_name}"})
                })
                continue
            
            # 带本地重试的执行
            last_error = None
            for attempt in range(MAX_RETRIES_PER_TOOL):
                try:
                    result = tool_func(**arguments)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })
                    last_error = None
                    break
                except Exception as e:
                    last_error = e
                    logger.warning(
                        f"工具 {func_name} 执行失败 (尝试 {attempt + 1}): {e}"
                    )
            
            # 如果所有重试都失败
            if last_error is not None:
                error_result = {
                    "error": str(last_error),
                    "tool": func_name,
                    "retries_exhausted": True
                }
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(error_result, ensure_ascii=False)
                })
    
    return "处理超时，请简化您的问题后重试。"


# 运行测试
if __name__ == "__main__":
    print(run_robust_agent("北京天气怎么样？苹果股价多少？"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

┌──────────────────────────────────────────────────────┐
│               错误处理决策树                            │
│                                                      │
│  工具调用失败？                                        │
│  ├─ JSON 解析错误 → 返回错误信息，让模型修正参数          │
│  ├─ 参数验证失败 → 返回具体错误原因                      │
│  ├─ 网络/超时错误 → 重试（指数退避）                      │
│  ├─ 4xx 客户端错误 → 不重试，返回错误信息                 │
│  ├─ 5xx 服务端错误 → 重试 2-3 次                        │
│  └─ 重试用尽 → 返回最终错误，让模型告知用户               │
└──────────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# ===================== OpenAI GPT-5.4 格式 =====================
openai_tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"}
                },
                "required": ["city"]
            }
        }
    }
]


# ===================== Anthropic Claude 格式 =====================
claude_tools = [
    {
        "name": "get_weather",
        "description": "查询天气",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名"}
            },
            "required": ["city"]
        }
    }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

"""
OpenAI vs Anthropic 工具调用对比
"""
import os

# ---------- OpenAI 方式 ----------
from openai import OpenAI

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# OpenAI: 工具在 tools 参数中，结果通过 role="tool" 消息返回
openai_response = openai_client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "北京天气？"}],
    tools=openai_tools,
)
# tool_call 结果: response.choices[0].message.tool_calls


# ---------- Anthropic 方式 ----------
import anthropic

claude_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# Anthropic: 工具在 tools 参数中，结果通过 user 角色的 tool_result 返回
claude_response = claude_client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "北京天气？"}],
    tools=claude_tools,
)

# Claude 工具调用的返回格式不同
for content_block in claude_response.content:
    if content_block.type == "tool_use":
        print(f"工具: {content_block.name}")
        print(f"参数: {content_block.input}")
        print(f"ID: {content_block.id}")