Python 快速开始
- AI 快速开始
- 完整开发指引
准备工作
在开始之前,请确保您已完成以下准备:
- 开通云开发环境:开通云开发环境
- 获取 API 访问凭证:在 CloudBase 控制台 获取环境 ID 和 API 密钥
Client
配置你的 AI 工具以连接 CloudBase 能力 支持本地与托管两种连接方式,详见 连接方式。
步骤 1:安装 / 配置 CloudBase
步骤 2:和 AI 对话
在 AI 对话中依次输入以下内容:
prompt
安装 CloudBase Skills:命令 npx skills add tencentcloudbase/cloudbase-skills -y使用 CloudBase Skills: 使用 CloudBase Skills:在 Python 后端服务中集成 CloudBase,包括数据库和云存储准备工作
在开始之前,请确保您已完成以下准备:
- 开通云开发环境:开通云开发环境
- 获取 API 访问凭证:在 CloudBase 控制台 获取环境 ID 和 API 密钥
更多详情请参考:Python 完整文档
身份认证
from cloudbase_client import cloudbase
def sign_in(username, password):
"""账号密码登录"""
result = cloudbase.request("POST", "/auth/v1/signin",
json={"username": username, "password": password})
if result:
access_token = result.get("access_token")
refresh_token = result.get("refresh_token")
user_id = result.get("sub")
print(f"登录成功! 用户ID: {user_id}")
print(f"访问令牌: {access_token[:20]}...")
return result
return None
# 使用示例
if __name__ == "__main__":
result = sign_in("your_username", "your_password")
云存储
- 上传文件
- 获取文件链接
- 下载文件
- 删除文件
import os
import requests
from datetime import datetime
from cloudbase_client import cloudbase
def upload_file(file_path, object_id=None):
"""上传文件到云存储"""
if not object_id:
filename = os.path.basename(file_path)
timestamp = int(datetime.now().timestamp() * 1000)
object_id = f"uploads/{timestamp}-{filename}"
# 1. 获取上传信息
upload_info = cloudbase.request("POST", "/v1/storages/get-objects-upload-info",
json=[{"objectId": object_id}])
if not upload_info:
return None
upload_info = upload_info[0]
upload_url = upload_info["uploadUrl"]
try:
# 2. 上传文件
upload_headers = {
"Authorization": upload_info["authorization"],
"X-Cos-Security-Token": upload_info["token"],
"X-Cos-Meta-Fileid": upload_info["cloudObjectMeta"]
}
with open(file_path, "rb") as f:
file_data = f.read()
upload_response = requests.put(upload_url, headers=upload_headers, data=file_data)
upload_response.raise_for_status()
result = {
"cloudObjectId": upload_info["cloudObjectId"],
"downloadUrl": upload_info["downloadUrl"],
"objectId": object_id
}
print("文件上传成功:")
print(f"- 对象ID: {result['objectId']}")
print(f"- 下载URL: {result['downloadUrl']}")
return result
except FileNotFoundError:
print(f"文件不存在: {file_path}")
return None
except Exception as e:
print(f"文件上传失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
result = upload_file("./example.jpg")
from cloudbase_client import cloudbase
def get_file_url(cloud_object_id):
"""获取云存储文件的临时访问链接"""
result = cloudbase.request("POST", "/v1/storages/get-objects-download-info",
json=[{"cloudObjectId": cloud_object_id}])
if result:
download_url = result[0].get("downloadUrl")
print("文件链接:", download_url)
return download_url
return None
# 使用示例
if __name__ == "__main__":
file_url = get_file_url("cloud://xxx.png")
import os
import requests
from cloudbase_client import cloudbase
def download_file(cloud_object_id, save_path="./"):
"""下载云存储文件到本地"""
# 1. 获取下载链接
result = cloudbase.request("POST", "/v1/storages/get-objects-download-info",
json=[{"cloudObjectId": cloud_object_id}])
if not result:
return False
download_url = result[0].get("downloadUrl")
try:
# 2. 从URL中提取文件名
filename = download_url.split("/")[-1].split("?")[0]
# 3. 如果save_path是目录,则拼接文件名
if os.path.isdir(save_path) or save_path.endswith("/"):
full_path = os.path.join(save_path, filename)
else:
full_path = save_path
# 4. 下载文件
file_response = requests.get(download_url)
file_response.raise_for_status()
# 5. 保存到本地
with open(full_path, "wb") as f:
f.write(file_response.content)
print(f"下载成功! 文件已保存到: {full_path}")
return True
except Exception as e:
print(f"下载失败: {e}")
return False
# 使用示例
if __name__ == "__main__":
# 下载到当前目录,使用原文件名
result = download_file("cloud://xxx.png")
# 下载到指定目录
result = download_file("cloud://xxx.png", "./downloads/")
# 下载并重命名
result = download_file("cloud://xxx.png", "./my-image.png")
from cloudbase_client import cloudbase
def delete_file(cloud_object_ids):
"""删除云存储文件"""
# 如果传入的是单个字符串,转换为列表
if isinstance(cloud_object_ids, str):
cloud_object_ids = [cloud_object_ids]
data = [{"cloudObjectId": obj_id} for obj_id in cloud_object_ids]
result = cloudbase.request("POST", "/v1/storages/delete-objects", json=data)
if result:
print("删除成功!")
return True
return False
# 使用示例
if __name__ == "__main__":
result = delete_file("cloud://xxx.png")
云函数
from cloudbase_client import cloudbase
def call_function(function_name, data=None):
"""调用云函数"""
result = cloudbase.request("POST", f"/v1/functions/{function_name}", json=data or {})
if result:
print("云函数调用结果:", result)
return result
# 使用示例
if __name__ == "__main__":
result = call_function("{%FUNCTION_NAME%}")
大模型
- 生文模型
- 生图模型
import requests
import json
from cloudbase_client import cloudbase
def stream_text(model, sub_model, messages):
"""流式文本生成"""
payload = {
"model": sub_model,
"messages": messages,
"stream": True
}
url = f"{cloudbase.base_url}/v1/ai/{model}/chat/completions"
headers = cloudbase.headers.copy()
headers["Accept"] = "text/event-stream"
try:
response = requests.post(url, headers=headers, json=payload, stream=True)
response.raise_for_status()
print("AI 流式响应:")
full_content = ""
for line in response.iter_lines():
if line:
line_str = line.decode("utf-8")
if line_str.startswith("data: "):
data_str = line_str[6:]
if data_str.strip() != "[DONE]":
try:
chunk_data = json.loads(data_str)
content = chunk_data.get("choices", [{}])[0].get("delta", {}).get("content", "")
if content:
print(content, end="", flush=True)
full_content += content
except json.JSONDecodeError:
continue
print() # 换行
return full_content
except Exception as e:
print(f"AI 调用失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
response = stream_text(
"{%AI_MODEL_NAME%}",
"{%AI_SUB_MODEL_NAME%}",
[
{"role": "system", "content": "请严格按照七言绝句或七言律诗的格律要求创作,平仄需符合规则,押韵要和谐自然,韵脚字需在同一韵部。"},
{"role": "user", "content": "春天"}
]
)
生图模型通过调用云函数来实现,在生图模型页面点击「一键创建云函数」,函数调用示例如下:
import requests
from cloudbase_client import cloudbase
def generate_image(prompt):
"""调用生图云函数"""
url = f"{cloudbase.base_url}/v1/functions/<YOUR_FUNCTION_NAME>/invoke"
headers = cloudbase.headers
payload = {
"prompt": prompt
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
result = response.json()
if result.get("success"):
# 生成成功
print("生成成功!")
print(f"图片URL: {result.get('imageUrl')}")
print(f"优化后的提示词: {result.get('revised_prompt')}")
# 使用图片
# 注意:图片URL有效期为24小时,请及时保存或转存
return result
else:
# 生成失败
print(f"生成失败: {result.get('code')} {result.get('message')}")
return None
except Exception as e:
print(f"调用失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
result = generate_image("一只可爱的猫咪在阳光下玩耍")
if result:
print(f"图片URL: {result.get('imageUrl')}")