|
import requests
import json
def chat_with_openclaw(use_stream=False):
"""
与 Openclaw 服务进行对话的完整示例
参数:
use_stream (bool): 是否使用流式响应,默认为 False
"""
# 1. 配置 API 端点
url = "http://localhost:7860/api/v1/chat/completions"
# 2. 设置请求头
headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
# 3. 构建请求数据
data = {
"model": "doubao-pro-256k", # 使用的模型名称
"messages": [
{"role": "system", "content": "你是一个专业的 AI 助手。"}, # 系统提示词
{"role": "user", "content": "请用中文简要介绍一下 Openclaw 项目。"} # 用户消息
],
"temperature": 0.7, # 控制回答的随机性 (0.0-1.0)
"max_tokens": 500, # 限制生成的最大 token 数
"stream": use_stream # 是否启用流式响应
}
try:
# 4. 发送 POST 请求
print(f"正在发送请求到 {url}...")
print(f"使用流式响应: {use_stream}")
if use_stream:
# 流式响应处理
print("开始接收流式响应:")
print("-" * 40)
response = requests.post(
url,
headers=headers,
data=json.dumps(data),
stream=True # 启用流式传输
)
# 检查响应状态
if response.status_code != 200:
print(f"请求失败,状态码: {response.status_code}")
print(f"错误信息: {response.text}")
return None
# 逐行处理流式响应
full_response = ""
for line in response.iter_lines():
if line:
line_str = line.decode('utf-8')
# 跳过 SSE 格式中的 "data: " 前缀
if line_str.startswith("data: "):
line_str = line_str[6:]
# 检查是否为结束标记
if line_str == "[DONE]":
print("\n流式响应结束")
break
# 解析 JSON 数据
try:
chunk_data = json.loads(line_str)
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
delta = chunk_data["choices"][0].get("delta", {})
content = delta.get("content", "")
if content:
print(content, end="", flush=True)
full_response += content
except json.JSONDecodeError:
# 忽略非 JSON 数据行
continue
print(f"\n{'='*40}")
print(f"完整响应内容:\n{full_response}")
return full_response
else:
# 非流式响应处理
response = requests.post(
url,
headers=headers,
data=json.dumps(data),
timeout=30 # 设置超时时间(秒)
)
# 5. 检查响应状态
response.raise_for_status() # 如果状态码不是 200,抛出异常
# 6. 解析响应数据
result = response.json()
print("请求成功!")
print(f"状态码: {response.status_code}")
print(f"响应时间: {response.elapsed.total_seconds():.2f}秒")
# 7. 提取并显示回答内容
if "choices" in result and len(result["choices"]) > 0:
message = result["choices"][0].get("message", {})
content = message.get("content", "无内容")
print(f"\nAI 回答:")
print("-" * 40)
print(content)
print("-" * 40)
# 显示使用统计(如果存在)
if "usage" in result:
usage = result["usage"]
print(f"\n使用统计:")
print(f" 提示词 tokens: {usage.get('prompt_tokens', 'N/A')}")
print(f" 完成 tokens: {usage.get('completion_tokens', 'N/A')}")
print(f" 总 tokens: {usage.get('total_tokens', 'N/A')}")
return result
except requests.exceptions.Timeout:
print("错误: 请求超时,请检查网络连接或服务状态")
except requests.exceptions.ConnectionError:
print("错误: 连接失败,请确保 Openclaw 服务正在运行")
except requests.exceptions.HTTPError as e:
print(f"HTTP 错误: {e}")
if response is not None:
print(f"错误详情: {response.text}")
except json.JSONDecodeError:
print("错误: 响应不是有效的 JSON 格式")
if response is not None:
print(f"原始响应: {response.text}")
except Exception as e:
print(f"未知错误: {type(e).__name__}: {e}")
return None
if __name__ == "__main__":
print("=== Openclaw API 调用示例 ===")
print("\n1. 非流式调用示例:")
result1 = chat_with_openclaw(use_stream=False)
print("\n\n2. 流式调用示例:")
result2 = chat_with_openclaw(use_stream=True)
print("\n=== 示例执行完成 ===")
|