Appearance
调用说明
本站点大部分模型都可以直接通过OpenAI的 API 进行调用, 您只需要更改模型名称即可。
更多详细参数,请参考我们的API 文档。
OpenAI
将 OpenAI 的 API 基础地址https://api.openai.com改为 UniAPI 的接口地址:https://api.uniapi.io,参照 OpenAI 官方文档搭配令牌页面生成的 Key 进行调用即可快速调用。
Python 示例
bash
pip install openaipython
from openai import OpenAI
client = OpenAI(
base_url="https://api.uniapi.io/v1",
api_key="sk-xxxxxx"
)
completion = client.chat.completions.create(
model="gpt-4o-mini",
max_tokens=16384,
messages=[
{"role": "user", "content": "hi"}
]
)
print(completion)Curl 示例
聊天接口
bash
curl --request POST \
--url https://api.uniapi.io/v1/chat/completions \
--header 'Authorization: Bearer sk-替换为你的key' \
-H "Content-Type: application/json" \
--data '{
"max_tokens": 1200,
"model": "gpt-3.5-turbo",
"temperature": 0.8,
"top_p": 1,
"presence_penalty": 1,
"messages": [
{
"role": "system",
"content": "You are ChatGPT, a large language model trained by OpenAI. Answer as concisely as possible."
},
{
"role": "user",
"content": "hi~"
}
]
}'DALL-E
bash
curl https://api.uniapi.io/v1/images/generations \
-H 'Authorization: Bearer sk-替换为你的key' \
-H "Content-Type: application/json" \
-d '{
"model": "dall-e-3",
"prompt": "a white siamese cat",
"n": 1,
"size": "1024x1024"
}'vision
bash
curl https://api.uniapi.io/v1/chat/completions \
-H 'Authorization: Bearer sk-替换为你的key' \
-H "Content-Type: application/json" \
-d '{
"max_tokens": 1200,
"model": "gpt-4-vision-preview",
"messages": [
{
"role": "system",
"content": "You are an expert Tailwind developer"
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "将图片生成网页代码"
},
{
"type": "image_url",
"image_url": {
"url": "data:image/jpeg;base64,xxxx=图片链接或者图片base64"
}
}
]
}
]
}'whisper
bash
curl --request POST \
--url https://api.uniapi.io/v1/audio/transcriptions \
--header 'Authorization: Bearer sk-你的key' \
--header 'Content-Type: multipart/form-data' \
--form file=@/path/to/file/openai.mp3 \
--form model=whisper-1tts
bash
curl https://api.uniapi.io/v1/audio/speech \
-H "Authorization: Bearer sk-你的key" \
-H "Content-Type: application/json" \
-d '{
"model": "tts-1",
"input": "你说点什么 包括中文!",
"voice": "alloy"
}' \
--output speech.mp3Claude
您使用OpenAI SDK或者使用OpenAI API格式,您只需要将model改成Claude的模型名称即可调用Claude的模型。
如果需要使用claudeAPI, 请在 base_url 中添加/claude后缀,例如:https://api.uniapi.io/claude
⚠️ 注意
Claude 默认渠道请优先使用 OpenAI API 进行请求。 稳定渠道无限制。
python 示例
bash
pip install anthropicpython
import anthropic
client = Anthropic(
api_key="sk-xxxxxxx",
base_url="https://api.uniapi.io/claude",
)
message = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
messages=[
{"role": "user", "content": "Hello, Claude"}
]
)
print(message.content)Curl 示例
更多示例,请查看Anthropic 官方文档
聊天接口
bash
curl --request POST \
--url https://api.uniapi.io/claude/v1/messages \
-H "Content-Type: application/json" \
-H "x-api-key: sk-替换为你的key" \
--data '{
"model": "claude-3-haiku-20240307",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "Hello, world"
}
]
}'Gemini
使用OpenAI SDK或者使用OpenAI API格式,您只需要将model改成Gemini的模型名称即可调用Gemini的模型。
更多示例,请查看Gemini 官方文档
Python 示例
bash
pip install -q -U google-genaipython
from google import genai
from google.genai import types
prompt = "hi~"
budget = 1024 # You can set this variable to any value between 0 and 24k
client = genai.Client(
http_options=types.HttpOptions(base_url='https://api.uniapi.io/gemini'),
api_key="sk-替换为你的key",
)
response = client.models.generate_content(
model="gemini-2.5-flash-preview-04-17",
contents=prompt,
config=types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(thinking_budget=budget, include_thoughts=True)
),
)
print(response.text)Curl 示例
bash
curl --request POST \
--url 'https://api.uniapi.io/gemini/v1alpha/models/gemini-1.5-flash-latest:generateContent?alt=sse' \
--header 'Content-Type: application/json' \
--header 'x-goog-api-key: sk-替换为你的key' \
--data '{
"contents": [
{
"parts": [
{
"text": "hi~"
}
],
"role": "user"
}
]
}'OCR
目前接入的是mistral的 OCR 模型。 详情请参考OCR 官方文档
Curl 示例
bash
curl --request POST \
--url https://api.uniapi.io/ocr/v1/ocr \
--header 'Authorization: Bearer sk-替换为你的key' \
--header 'Content-Type: application/json' \
--data '{
"model": "mistral-ocr-latest",
"document": {
"type": "document_url",
"document_url": "https://assets.anthropic.com/m/1cd9d098ac3e6467/original/Claude-3-Model-Card-October-Addendum.pdf"
},
"include_image_base64": true
}'Midjourney
支持 midjourney-proxy-plus 接口协议, 接口地址使用:
https://hk.uniapi.io(境内网路)或https://api.uniapi.io(境外网路)Relax 模式路径:
/mj-relax/mjFast 模式路径:/mj或/mj-fast/mjTurbo 模式路径:/mj-turbo/mj
详见接口说明。
python 示例
python
import requests
import time
import json
# 配置信息
# Relax模式请求路径:/mj-relax
# Fast模式请求路径:/mj-fast
# Turbo模式请求路径:/mj-turbo
# 如果不填写,默认是Fast模式
BASE_URL = "https://api.uniapi.io/mj-relax"
BEARER_TOKEN = "your_token_here"
HEADERS = {
"mj-api-secret:": f"{BEARER_TOKEN}",
"Content-Type": "application/json"
}
class MJImageGenerator:
def __init__(self):
self.base_url = BASE_URL
self.headers = HEADERS
def submit_task(self, prompt):
"""提交图片生成任务"""
url = f"{self.base_url}/mj/submit/imagine"
payload = {
"base64Array": [], # 这个是可选的,如果垫图,需要传入垫图的base64, 可以使用多张垫图 最多5张, 单张图片不超过1M
"prompt": prompt, # 这个是必填的,prompt 是必填的
"botType": "MID_JOURNEY" # 这个是必填的,botType 是必填的,MID_JOURNEY 表示 Midjourney 模型,NIJI_JOURNEY 表示 Niji 模型, 详细参考 https://docs.midjourney.com/docs/models
}
response = requests.post(url, headers=self.headers, json=payload)
data = response.json()
if data["code"] not in [1, 22]:
raise Exception(f"提交任务失败: {data['description']}")
return data["result"]
def action(self, customId, taskId):
"""执行操作"""
url = f"{self.base_url}/mj/submit/action"
payload = {
"customId": customId,
"taskId": taskId
}
response = requests.post(url, headers=self.headers, json=payload)
data = response.json()
if data["code"] not in [1, 22]:
raise Exception(f"提交任务失败: {data['description']}")
return data["result"]
def modal(self, taskId, prompt):
"""执行modal操作"""
url = f"{self.base_url}/mj/submit/modal"
payload = {
"prompt": prompt,
"taskId": taskId
}
response = requests.post(url, headers=self.headers, json=payload)
data = response.json()
if data["code"] not in [1, 22]:
raise Exception(f"提交任务失败: {data['description']}")
return data["result"]
def describe(self, base64, botType):
"""执行图生文操作"""
url = f"{self.base_url}/mj/submit/describe"
payload = {
"base64": base64,
"botType": botType
}
response = requests.post(url, headers=self.headers, json=payload)
data = response.json()
if data["code"] not in [1, 22]:
raise Exception(f"提交任务失败: {data['description']}")
return data["result"]
def shorten(self, prompt, botType):
"""执行缩短提示词的操作"""
url = f"{self.base_url}/mj/submit/shorten"
payload = {
"prompt": prompt,
"botType": botType
}
response = requests.post(url, headers=self.headers, json=payload)
data = response.json()
if data["code"] not in [1, 22]:
raise Exception(f"提交任务失败: {data['description']}")
return data["result"]
def fetch_task_status(self, task_id):
"""获取任务状态"""
url = f"{self.base_url}/mj/task/{task_id}/fetch"
response = requests.get(url, headers=self.headers)
return response.json()
def wait_for_completion(self, task_id):
"""轮询等待任务完成"""
while True:
data = self.fetch_task_status(task_id)
status = data["status"]
progress = data["progress"]
print(f"当前进度: {progress}")
if status == "SUCCESS":
print("任务完成!")
print(f"图片链接: {data['imageUrl']}")
print(f"提示词: {data['prompt']}")
print(f"提示词EN: {data['promptEn']}")
print("\n可用的操作按钮:")
print("\按钮说明, 顺序,从左往右,从上往下。 Ux:表示你想要操作哪张图片。 🔄 表示重新绘制。 Vx:表示你想要变化哪一张图片 ")
# 参考
# https://docs.midjourney.com/docs/midjourney-discord
# https://docs.midjourney.com/docs/variations
for button in data["buttons"]:
label = button['label'] if button['label'] != "" else button['emoji']
print(f"{label}: {button['customId']}")
return data
elif status == "FAILURE":
raise Exception(f"任务失败: {data['failReason']}")
time.sleep(10) # 每10秒轮询一次
def main():
try:
generator = MJImageGenerator()
# 示例prompt
# Cyberpunk 风格, 可以不填,可选值: 賽博朋克: Cyberpunk,星際: Warframe,動漫: ACGN,日本漫畫: Japanese comics/manga,水墨畫風格: Ink Wash Painting Style,原創: Original,風景畫: landscape,插畫: illustration,漫畫: Manga,現代自然: modern organic,創世紀: Genesis,海報風格: posterstyle,超現實主義: surrealism,素描: sketch,寫實: realism,水彩畫: Watercolor painting,立體主義: Cubism,黑白: black and white,膠片攝影風格: fm photography,電影化: cinematic,清晰的面部特徵: dlear facial features
# Wide view 宽视角, 可以不填, 可选值: 寬視角: Wide view,鳥瞰視角: Aerial view,頂視角: Top view,仰視角: Upview,正面視角: Front view,頭部特寫: Headshot,超廣角視角: Ultrawideshot,中景: Medium Shot(MS),遠景: Long Shot(LS),景深: depth offield(dof)
# Face Shot (VCU) 人物特写, 可以不填, 可选值: 臉部特寫: Face Shot (VCU),大特寫: Big Close-Up(BCU),特寫: Close-Up(CU),腰部以上: Waist Shot(WS),膝蓋以上: KneeShot(KS),全身照: Full Length Shot(FLS),極遠景: Extra Long Shot(ELS)
# Cold light 灯光, 可以不填,可选值: 冷光: Cold light,暖光: Warm light,硬光: hard lighting,戲劇性光線: Dramatic light,反射光: reflection light,薄霧: Misty foggy,自然光: Natural light,陽光: Sun light,情緒化: moody
# 可爱的猫, 描述你想要生成的图片
# --q 2 表示质量, 可选值: 0.25,0.5,1,2, 参考 https://docs.midjourney.com/docs/quality
# --s 50 表示艺术程度,可选值: 0-1000,参考 https://docs.midjourney.com/docs/stylize
# --v 6.1 表示版本, 可选值:1、2、3、4、5、5.0、5.1、5.2、6 和 6.1, 如果使用niji模型,则需要改为 --niji 版本号, 参考 https://docs.midjourney.com/docs/models
# --ar 4:3 表示宽高比
# 更多请参考 https://docs.midjourney.com/docs/parameter-list
prompt = "Cyberpunk,Wide view,Face Shot (VCU),Cold light,可爱的猫 --q 2 --s 50 --v 6.1 --ar 4:3"
print("提交任务中...")
task_id = generator.submit_task(prompt)
print(f"任务ID: {task_id}")
print("等待任务完成...")
generator.wait_for_completion(task_id)
# 按钮操作, 每次有按钮,都按照这个操作执行。
# action_id = generator.action("MJ::JOB::upsample::1::4f9c53b7-dc2a-441f-a7e0-e8b65dd2ce6d", task_id)
# print("等待任务完成...")
# generator.wait_for_completion(action_id)
# ------------------------------------------------------------
# 特殊操作, CustomZoom, 先 执行 按钮操作, 然后 执行 modal 操作
# 执行按钮操作
# action_id = generator.action("MJ::CustomZoom::ec92d09b-e6c4-458c-952f-d34e87b090a8", task_id)
# 执行 modal 操作
# 填写变焦值, 例如 1.8
# prompt += " --zoom 1.8"
# modal_id = generator.modal(prompt, action_id)
# print("等待任务完成...")
# generator.wait_for_completion(modal_id)
# 执行图生文操作
# task_id = generator.describe("data:image/jpeg;base64,xxxxx", "MID_JOURNEY")
# print(f"任务ID: {task_id}")
# print("等待任务完成...")
# generator.wait_for_completion(task_id)
# 执行缩短提示词操作
# task_id = generator.shorten("Please create a whimsical majestic tower of donuts, intricately crafted and adorned with a mesmerizing array of colorful sprinkles. Bring this sugary masterpiece to life, ensuring every detail is rendered in stunning magical realism. Thank you!", "MID_JOURNEY")
# print(f"任务ID: {task_id}")
# print("等待任务完成...")
# generator.wait_for_completion(task_id)
except Exception as e:
print(f"发生错误: {str(e)}")
if __name__ == "__main__":
main()Suno
Python 示例
python
import requests
import time
from typing import Dict, Any
key = "sk-xxxxx" # 替换为你的key
BASE_URL = "https://api.uniapi.io/suno"
def get_headers() -> Dict[str, str]:
return {
"Authorization": f"Bearer {key}",
"Content-Type": "application/json"
}
def handle_response(response: requests.Response) -> Dict[str, Any]:
response_data = response.json()
if response.status_code != 200:
raise Exception(f"请求失败,状态码: {response.status_code},响应信息: {response_data}")
if response_data.get("code") != "success":
raise Exception(f"操作失败,响应信息: {response_data}")
return response_data["data"]
def submit_lyrics(prompt: str) -> str:
url = f"{BASE_URL}/submit/lyrics"
data = {"prompt": prompt}
response = requests.post(url, headers=get_headers(), json=data)
return handle_response(response)
def fetch(task_id: str) -> Dict[str, Any]:
url = f"{BASE_URL}/fetch/{task_id}"
response = requests.get(url, headers=get_headers())
return handle_response(response)
def submit_song(payload: Dict[str, Any]) -> str:
url = f"{BASE_URL}/submit/music"
response = requests.post(url, headers=get_headers(), json=payload)
return handle_response(response)
def main():
prompt = "愉快的,摇滚的,学猫叫"
try:
lyrics_task_id = submit_lyrics(prompt)
lyrics_text = ""
while True:
lyrics_fetch = fetch(lyrics_task_id)
task_status = lyrics_fetch["status"]
if task_status == "FAILURE":
raise Exception("歌词生成失败")
if task_status == "SUCCESS":
lyrics_text = lyrics_fetch['data']['text']
break
print(f"歌词生成状态: {task_status},请等待2s...")
time.sleep(2)
print("歌词内容:" + lyrics_text)
if not lyrics_text:
raise Exception("歌词为空终止调用")
# 组装歌曲生成请求
payload = {
"prompt": lyrics_text,
"tags": "emotional punk",
"mv": "chirp-v3-5",
"title": "学猫叫"
}
# 提交歌曲生成请求
song_task_id = submit_song(payload)
print("歌曲任务ID:" + song_task_id)
isStream = False
# 轮询查询歌曲生成状态
while True:
task_data = fetch(song_task_id)
task_status = task_data["status"]
if task_status == "FAILURE":
raise Exception("歌曲生成失败")
if task_status == "SUCCESS":
break
if task_status == "IN_PROGRESS" and isStream == False:
isStream = True
print(f"歌曲已经可以通过流播放,但不可下载")
for song in task_data["data"]:
print(f"歌曲名称: {song['title']}")
print(f"歌曲封面: {song['image_url']}")
print(f"音频地址: {song['audio_url']}")
print("-" * 40)
print(f"歌曲生成状态: {task_status},请等待15s...")
time.sleep(15)
# 打印歌曲信息
for song in task_data["data"]:
print(f"歌曲名称: {song['title']}")
print(f"歌曲封面: {song['image_url']}")
print(f"音频地址: {song['audio_url']}")
print(f"视频地址: {song['video_url']}")
print("-" * 40)
except Exception as e:
print(f"发生错误: {e}")
if __name__ == "__main__":
main()FalAI
视频和图像调用方法一致。
⚠️ 注意
根据 FalAI 官方文档规则,如因为您输入参数错误导致的请求失败,积分不予退还。 详情见FalAI 官方文档
Python 示例
python
import requests
import time
from typing import Dict, Any
key = "sk-xxx" # 替换为实际的 Bearer Token
BASE_URL = "https://api.uniapi.io/fal-ai"
def get_headers() -> Dict[str, str]:
return {
"Authorization": f"Key {key}",
"Content-Type": "application/json"
}
def handle_response(response: requests.Response) -> Dict[str, Any]:
response_data = response.json()
if response.status_code != 200:
raise Exception(f"请求失败,状态码: {response.status_code},响应信息: {response_data}")
if response_data.get("status") == "FAILURE":
raise Exception(f"操作失败,响应信息: {response_data}")
return response_data
def submit(model: str, data: Dict[str, Any]) -> str:
url = f"{BASE_URL}/{model}"
response = requests.post(url, headers=get_headers(), json=data)
return handle_response(response)
def fetch(url: str) -> Dict[str, Any]:
response = requests.get(url, headers=get_headers())
return handle_response(response)
def main():
model = "flux-pro/kontext/text-to-image"
data = {
"prompt": "Extreme close-up of a single tiger eye, direct frontal view. Detailed iris and pupil. Sharp focus on eye texture and color. Natural lighting to capture authentic eye shine and depth. The word \"FLUX\" is painted over it in big, white brush strokes with visible texture.",
"guidance_scale": 1,
"num_images": 1,
"safety_tolerance": "2",
"output_format": "jpeg",
"aspect_ratio": "9:21"
}
try:
task = submit(model, data)
status_url = task['status_url']
response_url = task['response_url']
while True:
status_fetch = fetch(status_url)
task_status = status_fetch["status"]
if task_status != "COMPLETED":
print(f"视频生成状态: {task_status},请等待2s...")
time.sleep(2)
continue
response_fetch = fetch(response_url)
print("响应:")
print(response_fetch)
break
except Exception as e:
print(f"发生错误: {e}")
if __name__ == "__main__":
main()图像模型调用
可以直接通过 OpenAI 的 image generation 接口调用.
curl
curl --request POST \
--url https://api.uniapi.io/v1/images/generations \
--header 'Authorization: Bearer sk-xxx' \
--header 'Content-Type: application/json' \
--data '{
"model": "flux-pro/kontext/text-to-image",
"prompt": "Extreme close-up of a single tiger eye, direct frontal view. Detailed iris and pupil. Sharp focus on eye texture and color. Natural lighting to capture authentic eye shine and depth. The word \"FLUX\" is painted over it in big, white brush strokes with visible texture.",
"guidance_scale": 1,
"n": 1,
"num_images": 2,
"safety_tolerance": "2",
"output_format": "jpeg",
"aspect_ratio": "1:1"
}
'可灵
图像模型调用
python
"""
Kling Image Generation API SDK
支持图像生成任务的创建和查询
"""
import time
import requests
from typing import Optional, Dict, Any, List
from enum import Enum
class ModelName(str, Enum):
"""模型名称枚举"""
KLING_V1 = "kling-v1"
KLING_V1_5 = "kling-v1-5"
KLING_V2 = "kling-v2"
KLING_V2_1 = "kling-v2-1"
class ImageReference(str, Enum):
"""图片参考类型枚举"""
SUBJECT = "subject" # 角色特征参考
FACE = "face" # 人物长相参考
class Resolution(str, Enum):
"""分辨率枚举"""
K1 = "1k" # 1K标清
K2 = "2k" # 2K高清
class AspectRatio(str, Enum):
"""画面纵横比枚举"""
RATIO_16_9 = "16:9"
RATIO_9_16 = "9:16"
RATIO_1_1 = "1:1"
RATIO_4_3 = "4:3"
RATIO_3_4 = "3:4"
RATIO_3_2 = "3:2"
RATIO_2_3 = "2:3"
RATIO_21_9 = "21:9"
class TaskStatus(str, Enum):
"""任务状态枚举"""
SUBMITTED = "submitted" # 已提交
PROCESSING = "processing" # 处理中
SUCCEED = "succeed" # 成功
FAILED = "failed" # 失败
class KlingImageSDK:
"""Kling 图像生成 API SDK"""
def __init__(self, api_key: str, base_url: str = "https://api.uniapi.io/kling"):
"""
初始化 SDK
Args:
api_key: API 密钥
base_url: API 基础地址
"""
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
})
def create_generation_task(
self,
prompt: str,
model_name: str = ModelName.KLING_V1,
negative_prompt: Optional[str] = None,
image: Optional[str] = None,
image_reference: Optional[str] = None,
image_fidelity: float = 0.5,
human_fidelity: float = 0.45,
resolution: str = Resolution.K1,
n: int = 1,
aspect_ratio: str = AspectRatio.RATIO_16_9,
callback_url: Optional[str] = None
) -> Dict[str, Any]:
"""
创建图像生成任务
Args:
prompt: 正向文本提示词,不能超过2500个字符
model_name: 模型名称,默认 kling-v1
negative_prompt: 负向文本提示词,不能超过2500个字符
image: 参考图像(Base64编码或URL)
image_reference: 图片参考类型(subject/face)
image_fidelity: 对上传图片的参考强度,取值范围[0,1]
human_fidelity: 面部参考强度,取值范围[0,1]
resolution: 生成图片的清晰度(1k/2k)
n: 生成图片数量,取值范围[1,9]
aspect_ratio: 画面纵横比
callback_url: 任务结果回调通知地址
Returns:
Dict: 包含任务信息的响应
"""
url = f"{self.base_url}/v1/images/generations"
payload = {
"model_name": model_name,
"prompt": prompt,
"resolution": resolution,
"n": n,
"aspect_ratio": aspect_ratio
}
# 添加可选参数
if negative_prompt:
payload["negative_prompt"] = negative_prompt
if image:
payload["image"] = image
if image_reference:
payload["image_reference"] = image_reference
if image_fidelity != 0.5:
payload["image_fidelity"] = image_fidelity
if human_fidelity != 0.45:
payload["human_fidelity"] = human_fidelity
if callback_url:
payload["callback_url"] = callback_url
try:
response = self.session.post(url, json=payload)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {
"code": -1,
"message": f"请求失败: {str(e)}",
"request_id": None,
"data": None
}
def query_task(self, task_id: str) -> Dict[str, Any]:
"""
查询单个任务状态
Args:
task_id: 任务ID
Returns:
Dict: 包含任务详细信息的响应
"""
url = f"{self.base_url}/v1/images/generations/{task_id}"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {
"code": -1,
"message": f"请求失败: {str(e)}",
"request_id": None,
"data": None
}
def wait_for_completion(
self,
task_id: str,
poll_interval: int = 3,
max_wait_time: int = 300
) -> Dict[str, Any]:
"""
轮询等待任务完成
Args:
task_id: 任务ID
poll_interval: 轮询间隔(秒),默认3秒
max_wait_time: 最大等待时间(秒),默认300秒
Returns:
Dict: 最终的任务结果
"""
start_time = time.time()
while True:
# 检查是否超时
if time.time() - start_time > max_wait_time:
return {
"code": -1,
"message": f"等待超时(超过{max_wait_time}秒)",
"request_id": None,
"data": None
}
# 查询任务状态
result = self.query_task(task_id)
# 检查响应是否有效
if result.get("code") != 0:
return result
# 获取任务状态
task_status = result.get("data", {}).get("task_status")
# 任务完成
if task_status == TaskStatus.SUCCEED:
return result
# 任务失败
if task_status == TaskStatus.FAILED:
return result
# 任务进行中,继续等待
time.sleep(poll_interval)
def generate_image(
self,
prompt: str,
model_name: str = ModelName.KLING_V1,
negative_prompt: Optional[str] = None,
image: Optional[str] = None,
image_reference: Optional[str] = None,
image_fidelity: float = 0.5,
human_fidelity: float = 0.45,
resolution: str = Resolution.K1,
n: int = 1,
aspect_ratio: str = AspectRatio.RATIO_16_9,
callback_url: Optional[str] = None,
wait: bool = True,
poll_interval: int = 3,
max_wait_time: int = 300
) -> Dict[str, Any]:
"""
生成图像(一站式方法)
创建任务并可选择等待完成
Args:
prompt: 正向文本提示词
model_name: 模型名称
negative_prompt: 负向文本提示词
image: 参考图像
image_reference: 图片参考类型
image_fidelity: 图片参考强度
human_fidelity: 面部参考强度
resolution: 分辨率
n: 生成数量
aspect_ratio: 画面比例
callback_url: 回调地址
wait: 是否等待任务完成,默认True
poll_interval: 轮询间隔(秒)
max_wait_time: 最大等待时间(秒)
Returns:
Dict: 任务结果
"""
# 创建任务
create_result = self.create_generation_task(
prompt=prompt,
model_name=model_name,
negative_prompt=negative_prompt,
image=image,
image_reference=image_reference,
image_fidelity=image_fidelity,
human_fidelity=human_fidelity,
resolution=resolution,
n=n,
aspect_ratio=aspect_ratio,
callback_url=callback_url
)
# 检查创建是否成功
if create_result.get("code") != 0:
return create_result
# 获取任务ID
task_id = create_result.get("data", {}).get("task_id")
if not task_id:
return create_result
# 如果不需要等待,直接返回创建结果
if not wait:
return create_result
# 等待任务完成
return self.wait_for_completion(
task_id=task_id,
poll_interval=poll_interval,
max_wait_time=max_wait_time
)
def get_image_urls(self, result: Dict[str, Any]) -> List[str]:
"""
从结果中提取图片URL列表
Args:
result: query_task 或 generate_image 返回的结果
Returns:
List[str]: 图片URL列表
"""
if result.get("code") != 0:
return []
task_result = result.get("data", {}).get("task_result", {})
images = task_result.get("images", [])
return [img.get("url") for img in images if img.get("url")]
# 使用示例
if __name__ == "__main__":
# 初始化 SDK
api_key = "sk-your-api-key" # 替换为你的 API Key
base_url = "https://api.uniapi.io/kling" # 根据实际情况修改
sdk = KlingImageSDK(api_key=api_key, base_url=base_url)
# 示例1: 简单生成图像(自动等待完成)
print("示例1: 简单生成图像")
result = sdk.generate_image(
prompt="一只可爱的小猫在花园里玩耍",
model_name=ModelName.KLING_V2_1,
resolution=Resolution.K1,
aspect_ratio=AspectRatio.RATIO_16_9,
n=1
)
if result.get("code") == 0:
print("生成成功!")
image_urls = sdk.get_image_urls(result)
for idx, url in enumerate(image_urls):
print(f"图片 {idx + 1}: {url}")
else:
print(f"生成失败: {result.get('message')}")
print("\n" + "="*50 + "\n")
# 示例2: 图生图 - 使用参考图像生成
print("示例2: 图生图(使用URL)")
result_img2img = sdk.generate_image(
prompt="将这张图片转换为油画风格",
model_name=ModelName.KLING_V1_5,
image="https://example.com/reference-image.jpg", # 参考图片URL
image_reference=ImageReference.SUBJECT, # 角色特征参考
image_fidelity=0.7, # 参考强度
resolution=Resolution.K1,
aspect_ratio=AspectRatio.RATIO_1_1,
n=1
)
if result_img2img.get("code") == 0:
print("图生图成功!")
image_urls = sdk.get_image_urls(result_img2img)
for idx, url in enumerate(image_urls):
print(f"图片 {idx + 1}: {url}")
else:
print(f"图生图失败: {result_img2img.get('message')}")
print("\n" + "="*50 + "\n")
# 示例3: 图生图 - 使用Base64编码的图片
print("示例3: 图生图(使用Base64)")
# 读取本地图片并转换为Base64
import base64
# 假设你有一个本地图片文件
# with open("reference_image.jpg", "rb") as image_file:
# image_base64 = base64.b64encode(image_file.read()).decode('utf-8')
# 示例Base64(实际使用时替换为真实图片的Base64)
image_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAUAAAAFCAYAAACNbyblAAAAHElEQVQI12P4//8/w38GIAXDIBKE0DHxgljNBAAO9TXL0Y4OHwAAAABJRU5ErkJggg=="
result_img2img_base64 = sdk.generate_image(
prompt="保持人物特征,背景改为城市街道",
model_name=ModelName.KLING_V1_5,
image=image_base64, # Base64编码的图片(不要加data:image前缀)
image_reference=ImageReference.FACE, # 人物长相参考
image_fidelity=0.8,
human_fidelity=0.75, # 面部参考强度
resolution=Resolution.K1,
aspect_ratio=AspectRatio.RATIO_9_16,
n=1,
wait=True # 等待完成
)
if result_img2img_base64.get("code") == 0:
print("图生图成功!")
image_urls = sdk.get_image_urls(result_img2img_base64)
for idx, url in enumerate(image_urls):
print(f"图片 {idx + 1}: {url}")
else:
print(f"图生图失败: {result_img2img_base64.get('message')}")
print("\n" + "="*50 + "\n")
# 示例4: 创建任务后手动轮询
print("示例4: 创建任务后手动轮询")
create_result = sdk.create_generation_task(
prompt="日落时分的海滩风景",
model_name=ModelName.KLING_V2_1,
resolution=Resolution.K2,
n=2
)
if create_result.get("code") == 0:
task_id = create_result["data"]["task_id"]
print(f"任务创建成功,任务ID: {task_id}")
# 等待任务完成
final_result = sdk.wait_for_completion(task_id)
if final_result.get("code") == 0:
task_status = final_result["data"]["task_status"]
if task_status == TaskStatus.SUCCEED:
print("任务完成!")
image_urls = sdk.get_image_urls(final_result)
for idx, url in enumerate(image_urls):
print(f"图片 {idx + 1}: {url}")
else:
print(f"任务失败: {final_result['data'].get('task_status_msg')}")
else:
print(f"查询失败: {final_result.get('message')}")
else:
print(f"创建任务失败: {create_result.get('message')}")Replicate
Python 示例
官方SDK 调用示例
python
from replicate.client import Client
client = Client(api_token="sk-xxxxx", base_url="https://api.uniapi.io/replicate")
input = {
"prompt": "The photo: Create a cinematic, photorealistic medium shot capturing the dynamic energy of a high-octane action film. The focus is a young woman with wind-swept dark hair streaked with pink highlights and determined features, looking directly and intently into the camera lens, she is slightly off-center. She wears a fitted pink and gold racing jacket over a black tank top with \"Imagen 4 Fast\" in motion-stylized lettering and on the next line \"on Replicate\" emblazoned across the chest and aviator sunglasses pushed up on her head. The lighting is dramatic with motion blur streaks and neon reflections from passing city lights, creating dynamic lens flares and light trails (they do not cover her face). The background shows a blurred urban nightscape with streaking car headlights and illuminated skyscrapers rushing past, rendered with heavy motion blur and shallow depth of field. High contrast lighting, vibrant neon color palette with deep blues and electric yellows, and razor-sharp focus on her intense eyes enhance the fast-paced, electrifying atmosphere. She is illuminated while the background is darker.",
"aspect_ratio": "4:3"
}
output = client.run(
"google/imagen-4-fast",
input=input
)
# To access the file URL:
print(output)直接调用示例
python
import time
import requests
from typing import Dict, Any, List, Union, Optional
from pathlib import Path
import os
class ReplicateAPIError(Exception):
"""Replicate API 错误"""
def __init__(self, status_code: int, detail: str, title: str = ""):
self.status_code = status_code
self.detail = detail
self.title = title
error_message = f"API 错误 [{status_code}]"
if title:
error_message += f" - {title}"
if detail:
error_message += f": {detail}"
super().__init__(error_message)
@classmethod
def from_response(cls, response: requests.Response):
"""从响应对象创建异常"""
try:
data = response.json()
return cls(
status_code=data.get('status', response.status_code),
detail=data.get('detail', ''),
title=data.get('title', '')
)
except Exception:
# 如果无法解析 JSON,使用默认错误信息
return cls(
status_code=response.status_code,
detail=response.text or '未知错误',
title=''
)
class ReplicateOutput:
"""处理 Replicate API 返回的输出结果"""
def __init__(self, output: Union[str, List[str], None], task_id: str):
self._output = output
self._task_id = task_id
def url(self) -> List[str]:
"""
返回输出 URL 数组
Returns:
List[str]: URL 列表
"""
if self._output is None:
return []
if isinstance(self._output, str):
return [self._output]
if isinstance(self._output, list):
return self._output
return []
def save(self, path: str) -> List[str]:
"""
保存输出文件到指定路径
Args:
path: 保存目录路径
Returns:
List[str]: 保存的文件路径列表
"""
urls = self.url()
if not urls:
raise ValueError("没有可保存的输出文件")
# 确保目录存在
os.makedirs(path, exist_ok=True)
saved_files = []
for index, url in enumerate(urls):
# 从 URL 中提取文件扩展名
file_ext = self._get_file_extension(url)
# 构建文件名: task_id_{index}.ext
filename = f"{self._task_id}_{index}{file_ext}"
filepath = os.path.join(path, filename)
# 下载并保存文件
response = requests.get(url, stream=True)
response.raise_for_status()
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
saved_files.append(filepath)
print(f"文件已保存: {filepath}")
return saved_files
def _get_file_extension(self, url: str) -> str:
"""从 URL 中提取文件扩展名"""
# 移除查询参数
url_without_params = url.split('?')[0]
# 获取扩展名
ext = os.path.splitext(url_without_params)[1]
# 如果没有扩展名,尝试从 URL 路径推断
if not ext:
if 'mp4' in url.lower():
ext = '.mp4'
elif 'jpg' in url.lower() or 'jpeg' in url.lower():
ext = '.jpg'
elif 'png' in url.lower():
ext = '.png'
else:
ext = '.bin' # 默认扩展名
return ext
class ReplicateClient:
"""自定义 Replicate API 客户端"""
def __init__(self, api_token: str, base_url: str = "https://api.uniapi.io/replicate"):
"""
初始化客户端
Args:
api_token: API 令牌
base_url: API 基础 URL
"""
self.api_token = api_token
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_token}',
'Content-Type': 'application/json'
})
def run(
self,
model: str,
input: Dict[str, Any],
poll_interval: float = 1.0,
timeout: Optional[float] = None
) -> ReplicateOutput:
"""
运行模型并等待结果
Args:
model: 模型名称,格式为 "owner/model"
input: 模型输入参数
poll_interval: 轮询间隔(秒),默认 1 秒
timeout: 超时时间(秒),None 表示无限等待
Returns:
ReplicateOutput: 输出结果对象
"""
# 1. 创建任务
task_id = self._create_prediction(model, input)
print(f"任务已创建: {task_id}")
# 2. 轮询获取结果
result = self._poll_prediction(task_id, poll_interval, timeout)
# 3. 返回输出对象
return ReplicateOutput(result['output'], task_id)
def _create_prediction(self, model: str, input: Dict[str, Any]) -> str:
"""
创建预测任务
Args:
model: 模型名称
input: 输入参数
Returns:
str: 任务 ID
"""
url = f"{self.base_url}/v1/models/{model}/predictions"
payload = {
"input": input
}
response = self.session.post(url, json=payload)
# 检查响应状态码
if response.status_code != 200:
raise ReplicateAPIError.from_response(response)
data = response.json()
return data['id']
def _poll_prediction(
self,
task_id: str,
poll_interval: float,
timeout: Optional[float]
) -> Dict[str, Any]:
"""
轮询获取预测结果
Args:
task_id: 任务 ID
poll_interval: 轮询间隔
timeout: 超时时间
Returns:
Dict: 完整的任务结果
"""
url = f"{self.base_url}/v1/predictions/{task_id}"
start_time = time.time()
while True:
# 检查超时
if timeout and (time.time() - start_time) > timeout:
raise TimeoutError(f"任务 {task_id} 超时")
# 获取任务状态
response = self.session.get(url)
# 检查响应状态码
if response.status_code != 200:
raise ReplicateAPIError.from_response(response)
data = response.json()
status = data['status']
print(f"任务状态: {status}")
# 检查任务状态
if status == 'succeeded':
print("任务成功完成")
return data
elif status == 'failed':
error_msg = data.get('error', '未知错误')
raise RuntimeError(f"任务失败: {error_msg}")
elif status in ['starting', 'processing']:
# 继续等待
time.sleep(poll_interval)
else:
# 未知状态
print(f"警告: 未知状态 {status}")
time.sleep(poll_interval)
def get_prediction(self, task_id: str) -> Dict[str, Any]:
"""
获取指定任务的详情
Args:
task_id: 任务 ID
Returns:
Dict: 任务详情
"""
url = f"{self.base_url}/v1/predictions/{task_id}"
response = self.session.get(url)
# 检查响应状态码
if response.status_code != 200:
raise ReplicateAPIError.from_response(response)
return response.json()
if __name__ == "__main__":
# 初始化客户端
client = ReplicateClient(
api_token="sk-xxxxx",
base_url="https://api.uniapi.io/replicate"
)
# 输入参数
input = {
"prompt": "The photo: Create a cinematic, photorealistic medium shot capturing the dynamic energy of a high-octane action film. The focus is a young woman with wind-swept dark hair streaked with pink highlights and determined features, looking directly and intently into the camera lens, she is slightly off-center. She wears a fitted pink and gold racing jacket over a black tank top with \"Imagen 4 Fast\" in motion-stylized lettering and on the next line \"on Replicate\" emblazoned across the chest and aviator sunglasses pushed up on her head. The lighting is dramatic with motion blur streaks and neon reflections from passing city lights, creating dynamic lens flares and light trails (they do not cover her face). The background shows a blurred urban nightscape with streaking car headlights and illuminated skyscrapers rushing past, rendered with heavy motion blur and shallow depth of field. High contrast lighting, vibrant neon color palette with deep blues and electric yellows, and razor-sharp focus on her intense eyes enhance the fast-paced, electrifying atmosphere. She is illuminated while the background is darker.",
"aspect_ratio": "4:31"
}
# 运行模型
output = client.run(
"google/imagen-4-fast",
input=input
)
# 获取输出 URL
print("输出 URL:")
print(output.url())
# 保存文件到磁盘
saved_files = output.save("./output")
print(f"\n已保存文件: {saved_files}")BFL
Python 示例
python
import time
import requests
import json
import os
from typing import Dict, Any, List, Union, Optional
class BFLAPIError(Exception):
"""BFL API Error"""
def __init__(self, status_code: int, message: str):
self.status_code = status_code
self.message = message
super().__init__(f"API Error [{status_code}]: {message}")
@classmethod
def from_response(cls, response: requests.Response):
"""Create exception from response object"""
try:
data = response.json()
if 'detail' in data and data['detail']:
message = data['detail']
elif 'details' in data and data['details']:
message = json.dumps(data['details'], ensure_ascii=False)
else:
message = response.text or "Unknown Error"
except Exception:
message = response.text or "Unknown Error"
return cls(
status_code=response.status_code,
message=message
)
class BFLOutput:
"""Handle BFL API Output"""
def __init__(self, result: Dict[str, Any], task_id: str):
self._result = result
self._task_id = task_id
self._sample_url = result.get('sample')
def url(self) -> Optional[str]:
"""Return the sample URL"""
return self._sample_url
def save(self, path: str) -> str:
"""
Save the output file to the specified path
Args:
path: Directory path to save the file
Returns:
str: Path of the saved file
"""
if not self._sample_url:
raise ValueError("No sample URL available to save")
os.makedirs(path, exist_ok=True)
# Extract extension or default to .jpg
if '.png' in self._sample_url.lower():
ext = '.png'
elif '.jpg' in self._sample_url.lower() or '.jpeg' in self._sample_url.lower():
ext = '.jpg'
else:
ext = '.jpg' # Default for BFL usually
filename = f"{self._task_id}{ext}"
filepath = os.path.join(path, filename)
response = requests.get(self._sample_url, stream=True)
response.raise_for_status()
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return filepath
class BFLClient:
"""BFL API Client"""
def __init__(self, api_key: str, base_url: str = "https://api.uniapi.io/bfl"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
'x-key': api_key,
'Content-Type': 'application/json'
})
def run(
self,
model: str,
input: Dict[str, Any],
poll_interval: float = 1.0,
timeout: Optional[float] = None
) -> BFLOutput:
"""
Run model and wait for result
Args:
model: Model name (e.g. "flux-pro-1.1")
input: Input parameters
poll_interval: Polling interval in seconds
timeout: Timeout in seconds
Returns:
BFLOutput: Output result object
"""
# 1. Create Task
task_id = self._create_task(model, input)
print(f"Task created: {task_id}")
# 2. Poll Result
result_data = self._poll_result(task_id, poll_interval, timeout)
# 3. Return Output
return BFLOutput(result_data['result'], task_id)
def _create_task(self, model: str, input: Dict[str, Any]) -> str:
"""Create generation task"""
url = f"{self.base_url}/v1/{model}"
response = self.session.post(url, json=input)
if response.status_code != 200:
raise BFLAPIError.from_response(response)
data = response.json()
return data['id']
def _poll_result(
self,
task_id: str,
poll_interval: float,
timeout: Optional[float]
) -> Dict[str, Any]:
"""Poll for task result"""
url = f"{self.base_url}/v1/get_result"
params = {'id': task_id}
start_time = time.time()
while True:
if timeout and (time.time() - start_time) > timeout:
raise TimeoutError(f"Task {task_id} timed out")
response = self.session.get(url, params=params)
# Handle 404 by ignoring and continuing
if response.status_code == 404:
time.sleep(poll_interval)
continue
if response.status_code != 200:
raise BFLAPIError.from_response(response)
data = response.json()
status = data.get('status')
if status == 'Ready':
result = data.get('result')
if not result or 'sample' not in result:
raise RuntimeError("Task Ready but result or sample missing")
return data
elif status in ['Request Moderated', 'Content Moderated', 'Error']:
details = data.get('details')
if not details:
# If details is missing/empty, check detail?
# Prompt says: "read details to see reason, if details has no value, return error + status code"
# But BFLAPIError handles 'detail' vs 'details'.
# Here we are inside the JSON body where status is Error.
# The prompt says: "If status is ... Error ... need to read details ... if details has no value, directly return error + status code."
# It implies raising an exception with the info we have.
raise RuntimeError(f"Task failed with status {status}. Details: {data.get('detail') or 'No details provided'}")
# If details exists, it might be a dict or list, serialize it.
if isinstance(details, (dict, list)):
details_str = json.dumps(details, ensure_ascii=False)
else:
details_str = str(details)
raise RuntimeError(f"Task failed with status {status}: {details_str}")
elif status == 'Pending':
time.sleep(poll_interval)
elif status == 'Task not found':
# Treat as pending/retry if it's a consistency issue, or error?
# Given 404 is ignored, maybe this is similar.
# But if it persists, it's bad.
# I'll log and wait.
print(f"Status: {status}, waiting...")
time.sleep(poll_interval)
else:
print(f"Unknown status: {status}, waiting...")
time.sleep(poll_interval)
if __name__ == "__main__":
# Initialize Client
client = BFLClient(
api_key="sk-xxxxxx",
base_url="https://api.uniapi.io/bfl"
)
# Input parameters
input_data = {
"prompt": "A futuristic city with flying cars, neon lights, cyberpunk style",
"width": 1024,
"height": 768
}
try:
# Run model
# Replace 'flux-pro-1.1' with the actual model name you want to use
output = client.run(
"flux-pro-1.1",
input=input_data
)
# Get output URL
print("Output URL:")
print(output.url())
# Save to disk
saved_file = output.save("./output")
print(f"\nSaved file: {saved_file}")
except Exception as e:
print(f"An error occurred: {e}")