File size: 4,527 Bytes
962759f |
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 |
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
测试工具调用功能
"""
import json
import requests
# 配置
BASE_URL = "http://localhost:8080"
API_KEY = "your-api-key" # 替换为实际的 API key
def test_tool_call():
"""测试工具调用功能"""
# 定义一个简单的工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，例如：北京、上海"
},
"unit": {
"type": "string",
"description": "温度单位",
"enum": ["celsius", "fahrenheit"]
}
},
"required": ["location"]
}
}
}
]
# 构建请求
request_data = {
"model": "GLM-4.5",
"messages": [
{
"role": "user",
"content": "北京的天气怎么样？"
}
],
"tools": tools,
"tool_choice": "auto",
"stream": False
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
print("=" * 60)
print("测试工具调用 (非流式)")
print("=" * 60)
# 发送请求
response = requests.post(
f"{BASE_URL}/v1/chat/completions",
json=request_data,
headers=headers
)
print(f"状态码: {response.status_code}")
if response.status_code == 200:
result = response.json()
print("\n响应内容:")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 检查是否有工具调用
if result.get("choices"):
choice = result["choices"][0]
if choice.get("message", {}).get("tool_calls"):
print("\n✅ 检测到工具调用!")
for tc in choice["message"]["tool_calls"]:
print(f" - 函数: {tc.get('function', {}).get('name')}")
print(f" 参数: {tc.get('function', {}).get('arguments')}")
else:
print("\n⚠️ 未检测到工具调用")
if choice.get("message", {}).get("content"):
print(f"内容: {choice['message']['content'][:200]}")
else:
print(f"\n错误响应: {response.text}")
# 测试流式响应
print("\n" + "=" * 60)
print("测试工具调用 (流式)")
print("=" * 60)
request_data["stream"] = True
response = requests.post(
f"{BASE_URL}/v1/chat/completions",
json=request_data,
headers=headers,
stream=True
)
print(f"状态码: {response.status_code}")
if response.status_code == 200:
print("\n流式响应:")
tool_calls_detected = False
for line in response.iter_lines():
if line:
line_str = line.decode('utf-8')
if line_str.startswith("data: "):
data = line_str[6:]
if data == "[DONE]":
print("流结束")
break
try:
chunk = json.loads(data)
if chunk.get("choices"):
delta = chunk["choices"][0].get("delta", {})
if delta.get("tool_calls"):
tool_calls_detected = True
print(f"检测到工具调用: {json.dumps(delta['tool_calls'], ensure_ascii=False)}")
elif delta.get("content"):
print(f"内容: {delta['content']}", end="")
except json.JSONDecodeError:
pass
if tool_calls_detected:
print("\n\n✅ 流式响应中检测到工具调用!")
else:
print("\n\n⚠️ 流式响应中未检测到工具调用")
else:
print(f"\n错误响应: {response.text}")
if __name__ == "__main__":
test_tool_call() |