File size: 4,440 Bytes
d3cadd5 |
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 |
#!/usr/bin/env python3
"""
测试思考功能
"""
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())
|