1. 资源准备
2. 安装依赖
pip install openai
3. 实现代码
import sys
import openai
# 这里填写您在 https://yibuapi.com 上创建的 API Key
api_key = "sk-xxxxx" # 建议用环境变量管理，避免明文
# 这里填写 https://yibuapi.com/v1
base_url = "https://yibuapi.com/v1"
prompt = """
an astronaut golden retriever named Sora levitates around an intergalactic pup-themed space station with a tiny jet back that propels him. gorgeous specular lighting and comets fly through the sky, retro-future astro-themed music plays in the background. light glimmers off the dog's eyes. the dog initially propels towards the space station with the doors opening to let him in. the shot then changes. now inside the space station, many tennis balls are flying around in zero gravity. the dog's astronaut helmet opens up so he can grab one. 35mm film, the intricate details and texturing of the dog's hair are clearly visible and the light of the comets shimmers off the fur.
"""
def get_sora_response(prompt, api_key, base_url, use_stream=True):
try:
client = openai.OpenAI(api_key=api_key, base_url=base_url)
# 许多 OpenAI 兼容网关会默认开启 stream，
# 这里显式传入 use_stream 来避免歧义
resp = client.chat.completions.create(
model="sora-2",
max_tokens=1000,
stream=use_stream,
messages=[{"role": "user", "content": prompt}],
)
# 情况 1：流式返回（resp 可迭代）
if use_stream:
full_text = []
try:
for chunk in resp:
# 兼容不同实现的字段：delta.content / message / content
part = None
try:
part = chunk.choices[0].delta.get("content")
except Exception:
# 某些实现可能是对象属性而不是字典
if hasattr(chunk.choices[0], "delta") and hasattr(chunk.choices[0].delta, "content"):
part = chunk.choices[0].delta.content
elif hasattr(chunk.choices[0], "message") and hasattr(chunk.choices[0].message, "content"):
part = chunk.choices[0].message.content
elif hasattr(chunk.choices[0], "text"):
part = chunk.choices[0].text
if part:
sys.stdout.write(part)
sys.stdout.flush()
full_text.append(part)
except Exception:
# 有些 SDK 的 Stream 对象也支持 .text / .content
pass
print() # 换行
return "".join(full_text) if full_text else "[流式输出结束]"
# 情况 2：非流式返回（普通对象）
# 兼容某些实现可能把内容放在不同字段
if hasattr(resp, "choices") and resp.choices:
choice0 = resp.choices[0]
if hasattr(choice0, "message") and hasattr(choice0.message, "content") and choice0.message.content:
return choice0.message.content
if hasattr(choice0, "text") and choice0.text:
return choice0.text
# 兜底：直接转字符串看看
return str(resp)
except Exception as e:
return f"请求失败: {str(e)}"
if __name__ == "__main__":
# 明确用流式，避免网关默认为流式但你按非流式取值导致的报错
result = get_sora_response(prompt, api_key, base_url, use_stream=True)
print(f"\n\n—— 汇总内容 ——\n{result}\n")
4. 运行效果
