| |
| |
| """ |
| 测试工具调用功能 |
| """ |
|
|
| import json |
| import requests |
|
|
| |
| BASE_URL = "http://localhost:8080" |
| API_KEY = "your-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() |