| |
| """ |
| 测试思考功能 |
| """ |
| import asyncio |
| import json |
| import httpx |
|
|
| async def test_thinking_feature(): |
| """测试思考功能""" |
| |
| |
| test_data = { |
| "model": "claude-sonnet-4.5", |
| "messages": [ |
| { |
| "role": "user", |
| "content": "请解释什么是递归，并给出一个简单的例子。" |
| } |
| ], |
| "thinking": { |
| "thinking_type": "enabled", |
| "budget_tokens": 5000 |
| }, |
| "stream": True, |
| "max_tokens": 1000 |
| } |
| |
| print("发送思考功能测试请求...") |
| print(f"请求内容: {json.dumps(test_data, indent=2, ensure_ascii=False)}") |
| print("\n" + "="*60 + "\n") |
| |
| try: |
| async with httpx.AsyncClient(timeout=60) as client: |
| async with client.stream( |
| "POST", |
| "http://localhost:8080/v1/messages", |
| headers={ |
| "Content-Type": "application/json", |
| "x-api-key": "any", |
| "anthropic-version": "2023-06-01" |
| }, |
| json=test_data |
| ) as response: |
| |
| if response.status_code != 200: |
| print(f"错误: {response.status_code}") |
| print(await response.aread()) |
| return |
| |
| print("收到响应:\n") |
| |
| thinking_content = [] |
| text_content = [] |
| current_block = None |
| |
| async for line in response.aiter_lines(): |
| if line.startswith("data: "): |
| data_str = line[6:] |
| |
| if data_str == "[DONE]": |
| break |
| |
| try: |
| data = json.loads(data_str) |
| event_type = data.get("type") |
| |
| if event_type == "content_block_start": |
| block_type = data.get("content_block", {}).get("type") |
| current_block = block_type |
| print(f"\n[开始 {block_type} 块]") |
| |
| elif event_type == "content_block_delta": |
| delta = data.get("delta", {}) |
| |
| if delta.get("type") == "thinking_delta": |
| thinking = delta.get("thinking", "") |
| thinking_content.append(thinking) |
| print(thinking, end="", flush=True) |
| |
| elif delta.get("type") == "text_delta": |
| text = delta.get("text", "") |
| text_content.append(text) |
| print(text, end="", flush=True) |
| |
| elif event_type == "content_block_stop": |
| print(f"\n[结束块]") |
| current_block = None |
| |
| elif event_type == "message_stop": |
| print("\n\n[响应完成]") |
| break |
| |
| elif event_type == "error": |
| print("\n\n[错误]") |
| print(json.dumps(data.get("error", data), ensure_ascii=False, indent=2)) |
| break |
| |
| except json.JSONDecodeError as e: |
| print(f"\n解析错误: {e}") |
| continue |
| |
| print("\n" + "="*60) |
| print("\n思考内容汇总:") |
| print("".join(thinking_content)) |
| |
| print("\n" + "-"*40) |
| print("\n回答内容汇总:") |
| print("".join(text_content)) |
| |
| except Exception as e: |
| print(f"请求失败: {e}") |
|
|
| if __name__ == "__main__": |
| asyncio.run(test_thinking_feature()) |
|
|
