目录
这个周六下午,我在键盘前发出了灵魂拷问
只用两段 Prompt,角色从无到有
让它动起来:眨个眼、挥个手、说句话
代码里几个值得拿出来说的细节
1. 指数退避重试,不慌
2. 视频状态轮询,不偷懒
3. 文件下载自动命名
4. 下载和生成解耦
完整Python源码,拷贝即用
一个比喻收尾
这个周六下午,我在键盘前发出了灵魂拷问
周六下午两点半,上海闵行区,窗外下着淅淅沥沥的小雨,我窝在出租屋的书桌前,桌上摆着半杯已经凉透的美式。微信群里朋友丢来一条消息:
"你天天折腾 AI,能不能帮我想个科幻西部题材的角色设计?要那种超细节的,最好能动起来那种。"
我看着这条消息愣了五秒钟。搁以前,我大概会打开 Blender,然后花一个下午建模——前提是我会 Blender(并不会)。或者打开 Midjourney 吭哧吭哧调参数,然后对着静态图片跟朋友说"你脑补一下它会动"。
但那天不一样。我突然想起自己前两天刚写完一个脚本,对接的是 Agnes AI 的图片和视频生成接口,图生视频一条龙,Prompt 进去、成品出来。说干就干。
只用两段 Prompt,角色从无到有
我把朋友的需求拆成了两步:先跑一张概念角色设定图,然后基于这张图生成一小段会说话的视频。整个流程的核心代码不到 200 行,核心思路极其简单:文本模型润色画面描述 → 图片模型出图 → 视频模型基于图片生成动态视频,每个阶段自动下载文件到本地。
整个程序启动后先问一个问题:
choice = input("是否使用自定义提示词?(是y/否n): ").strip().lower()
走 y 分支进入自定义模式,敲两段 Prompt 全部搞定。我把朋友的需求翻译成 第一段 Prompt,直接贴进了终端:
Cinematic character concept art sheet, photorealistic humanoid robot cowboy.
White mechanical body with exposed metallic joints, glowing blue electronic screen
digital smiley face. Wearing a brown cowboy hat, brown leather jacket, red bandana
around the neck, black gloves, western-style belt with a large silver buckle,
and hip holsters. Weathered mechanical texture, sci-fi western aesthetic.
Ultra-detailed, 8K resolution, sharp focus, shot on ARRI Alexa 65,
volumetric lighting. Clean solid color background for character design reference.
回车敲下去的一瞬间,程序立刻调用了 Agnes Image 2.1 Flash:
def generate_image(prompt):
print("=== Generating image with Agnes Image 2.1 Flash ===")
resp = make_request_with_retry(
"POST",
f"{BASE_URL}/images/generations",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": "agnes-image-2.1-flash",
"prompt": prompt,
"n": 1,
"size": "1024×768"
}
)
resp_data = resp.json()
image_url = resp_data["data"][0]["url"]
image_task_id = resp_data.get("id", f"img_{int(time.time())}")
local_image_path = download_file(image_url, image_task_id, "image")
return image_url, image_task_id, local_image_path
控制台刷刷刷跑起来:
=== Generating image with Agnes Image 2.1 Flash ===
Generated image URL: https://platform-outputs.agnes-ai.space/…/8485197f…png
Image task ID: img_1781346290
正在下载 image…
✅ image 已保存到: D:\\PythonProjects\\agnes-demo\\img_1781346290.png
图片秒出,一个戴着棕色牛仔帽、穿着皮夹克、面部是蓝色 LED 数字笑脸的机械牛仔跃然屏幕上。白色机械躯体,暴露的金属关节,西部风的皮套和银扣腰带——朋友描述的那些细节一个不落。
生成图片展示亮相:

让它动起来:眨个眼、挥个手、说句话
图片出来只是第一步。朋友说"能动起来就好了",我嘿嘿一笑,把 第二段 Prompt 敲了进去:
The character blinks naturally by briefly flickering its blue LED face screen,
then raises its right hand in a cheerful wave gesture while its mouth moves
in sync as it speaks: "Hello everyone, I'm Jack." 5-second video clip,
smooth motion, consistent with the reference frame in lighting, composition,
and background.
这段 Prompt 的设计很有意思——你不需要说"生成一个视频"这种废话,而是直接描述你要看到什么动作:眼皮蓝色屏幕闪烁一下表示眨眼、右手抬起做挥手的动作、嘴部同步说台词。"5-second video clip"限定时长,"consistent with the reference frame"保证视频风格和底图一致,不会出现画面突变。
视频生成走的是 Agnes Video V2.0,调用方式如下:
def generate_video(prompt, image_url=None, duration=10):
payload = {
"model": "agnes-video-v2.0",
"prompt": prompt,
"duration": duration
}
if image_url:
payload["image"] = image_url
resp = make_request_with_retry(
"POST",
f"{BASE_URL}/videos",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload
)
task_id = resp.json()["id"]
# 轮询等待视频生成完成
while True:
status_resp = make_request_with_retry(
"GET", f"{BASE_URL}/videos/{task_id}",
headers={"Authorization": f"Bearer {API_KEY}"}
)
status_data = status_resp.json()
status = status_data.get("status", "unknown")
progress = status_data.get("progress", 0)
print(f"状态: {status} | 进度: {progress}%")
if status in ["succeeded", "completed"]:
video_url = status_data.get("remixed_from_video_id", "")
if not video_url:
video_url = status_data.get("url", "")
local_video_path = download_file(video_url, task_id, "video")
return video_url, task_id, local_video_path
elif status in ["failed", "cancelled"]:
raise Exception(…)
time.sleep(5)
视频生成是异步任务,提交之后进入 queued 状态,程序每 5 秒轮询一次进度。这个等待过程有点像等外卖——你盯着屏幕,进度条龟速爬行,心里默念"别超时别失败"。
控制台输出看起来是这样的:
=== Generating video with Agnes Video V2.0 ===
请求失败,2秒后重试 (1/5)…
Video generation task created. Task ID: task_PKJPAZWcF6Sr9J8s1rtmvQlD0SlkeDUoE
Waiting for video generation to complete…
状态: queued | 进度: 0%
状态: queued | 进度: 0%
…(几次轮询后)
状态: completed | 进度: 100%
✅ 视频生成完成!
下载地址: https://platform-outputs.agnes-ai.space/videos/…/video_7e64e992…mp4
正在下载 video…
✅ video 已保存到: D:\\PythonProjects\\agnes-demo\\task_PKJPAZWcF6Sr9J8s1rtmvQlD0SlkeDUoE.mp4
注意中间有一次请求失败,程序自动 2 秒后退避重试成功——这个指数退避重试机制在视频生成这类长耗时任务里简直是救命的。
生成视频展示亮相,你瞅瞅,你瞅瞅,Big Bang!
agnes-video-generate
代码里几个值得拿出来说的细节
1. 指数退避重试,不慌
API 调用最怕的就是网络波动或者服务端临时繁忙。这个脚本里用了一个朴素但有效的重试机制:
def make_request_with_retry(method, url, max_retries=5, timeout=30, **kwargs):
retry_count = 0
while retry_count < max_retries:
try:
resp = requests.request(method, url, timeout=timeout, **kwargs)
resp.raise_for_status()
return resp
except (requests.exceptions.SSLError,
requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
retry_count += 1
if retry_count < max_retries:
wait_time = 2 ** retry_count # 指数退避
print(f"请求失败,{wait_time}秒后重试 ({retry_count}/{max_retries})…")
time.sleep(wait_time)
else:
raise Exception(f"请求失败,已重试{max_retries}次: {str(e)}")
重试间隔是 2^n 秒——2 秒、4 秒、8 秒、16 秒——最多 5 次。这种指数退避策略避免了在对方服务还没恢复时疯狂重试雪上加霜。实际跑的时候刚好触发了一次,2 秒后重试成功,整个过程丝滑无感。
2. 视频状态轮询,不偷懒
视频生成是异步的,提交任务后服务端给一个 task_id,你得自己轮询状态。代码里用一个 while True 循环每 5 秒查一次,状态从 queued 到 completed。轮询过程中打印状态和进度百分比,让你知道它没死机:
状态: queued | 进度: 0%
状态: queued | 进度: 0%
…
状态: completed | 进度: 100%
中间有几次 queued 连着出现,但总共等了大概几十秒就完成了。5 秒间隔是经验值——太短浪费请求次数,太长干等焦虑。
3. 文件下载自动命名
图片和视频下载到本地时,用 task_id 自动命名,不撞名、不覆盖:
def download_file(url, task_id, file_type="image"):
if file_type == "image":
ext = "png"
elif file_type == "video":
ext = "mp4"
filename = f"{task_id}.{ext}"
filepath = os.path.join(os.getcwd(), filename)
with open(filepath, "wb") as f:
f.write(resp.content)
跑完之后本地目录里躺着两个文件:img_1781346290.png 和 task_PKJPAZWcF6Sr9J8srtmvQlD0SlkeDUoE.mp4,一目了然。
4. 下载和生成解耦
下载用的是裸 requests.get,不带 API Key 头,因为生成的下载 URL 是公开可访问的临时地址。这个设计避免了在 URL 参数里带密钥的安全隐患。另外下载超时设了 300 秒,视频文件比较大时也不会半路断开。
完整Python源码,拷贝即用
拷贝下方源码到你的python文件中,修改API_KEY 为你注册的API_KEY,即可。 不知道怎么申请 API_KEY 的同学,移步到我昨天的文章:【白嫖分享】0 Token 消耗,Agnes AI API 实战 —— 免费多模态模型从入门到踩坑全记录-CSDN博客
import sys
import os
import requests
import time
API_KEY = "sk-*************" #替换成你的APIKEY
BASE_URL = "https://apihub.agnes-ai.com/v1"
def fix_encoding():
"""Fix Unicode encoding on Windows"""
if sys.platform == "win32":
sys.stdout.reconfigure(encoding="utf-8")
def download_file(url, task_id, file_type="image"):
"""下载文件到项目根目录,以 task_id 命名"""
try:
print(f"正在下载 {file_type}…", flush=True)
# 直接使用 requests 下载,不带 API Key,因为下载 URL 是公开的
resp = requests.get(url, timeout=300)
resp.raise_for_status()
# 确定文件扩展名
if file_type == "image":
ext = "png"
elif file_type == "video":
ext = "mp4"
else:
ext = "dat"
filename = f"{task_id}.{ext}"
filepath = os.path.join(os.getcwd(), filename)
with open(filepath, "wb") as f:
f.write(resp.content)
print(f"✅ {file_type} 已保存到: {filepath}", flush=True)
return filepath
except Exception as e:
print(f"❌ 下载 {file_type} 失败: {str(e)}", flush=True)
return None
def make_request_with_retry(method, url, max_retries=5, timeout=30, **kwargs):
"""Make HTTP request with retry mechanism"""
retry_count = 0
while retry_count < max_retries:
try:
resp = requests.request(method, url, timeout=timeout, **kwargs)
resp.raise_for_status()
return resp
except (requests.exceptions.SSLError, requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
retry_count += 1
if retry_count < max_retries:
wait_time = 2 ** retry_count # 指数退避
print(f"请求失败,{wait_time}秒后重试 ({retry_count}/{max_retries})…", flush=True)
time.sleep(wait_time)
else:
raise Exception(f"请求失败,已重试{max_retries}次: {str(e)}")
def generate_text(prompt):
"""Generate text using Agnes 2.0 Flash"""
print("=== Generating text with Agnes 2.0 Flash ===", flush=True)
resp = make_request_with_retry(
"POST",
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": "agnes-2.0-flash",
"messages": [{"role": "user", "content": prompt}]
}
)
content = resp.json()["choices"][0]["message"]["content"]
print(f"Generated text:\\n{content}\\n", flush=True)
return content
def generate_image(prompt):
"""Generate image using Agnes Image 2.1 Flash"""
print("=== Generating image with Agnes Image 2.1 Flash ===", flush=True)
resp = make_request_with_retry(
"POST",
f"{BASE_URL}/images/generations",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": "agnes-image-2.1-flash",
"prompt": prompt,
"n": 1,
"size": "1024×768"
}
)
resp_data = resp.json()
image_url = resp_data["data"][0]["url"]
# 图片生成的 task_id 可能在响应中,尝试获取,如果没有则用时间戳
image_task_id = resp_data.get("id", f"img_{int(time.time())}")
print(f"Generated image URL: {image_url}", flush=True)
print(f"Image task ID: {image_task_id}", flush=True)
# 下载图片
local_image_path = download_file(image_url, image_task_id, "image")
print()
return image_url, image_task_id, local_image_path
def generate_video(prompt, image_url=None, duration=10):
"""Generate video using Agnes Video V2.0"""
print("=== Generating video with Agnes Video V2.0 ===", flush=True)
payload = {
"model": "agnes-video-v2.0",
"prompt": prompt,
"duration": duration
}
if image_url:
payload["image"] = image_url
resp = make_request_with_retry(
"POST",
f"{BASE_URL}/videos",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload
)
task_id = resp.json()["id"]
print(f"Video generation task created. Task ID: {task_id}", flush=True)
# Poll for video completion
print("Waiting for video generation to complete…", flush=True)
while True:
try:
status_resp = make_request_with_retry(
"GET",
f"{BASE_URL}/videos/{task_id}",
headers={"Authorization": f"Bearer {API_KEY}"}
)
status_data = status_resp.json()
except Exception as e:
print(f"查询状态时出错: {str(e)},继续尝试…", flush=True)
time.sleep(5)
continue
status = status_data.get("status", "unknown")
progress = status_data.get("progress", 0)
print(f"状态: {status} | 进度: {progress}%", flush=True)
if status in ["succeeded", "completed"]:
# 使用与 check_agnes_video.py 相同的方式获取 URL
video_url = status_data.get("remixed_from_video_id", "")
# 如果 remixed_from_video_id 没有,尝试 url 作为备选
if not video_url:
video_url = status_data.get("url", "")
print(f"\\n✅ 视频生成完成!", flush=True)
print(f"下载地址: {video_url}", flush=True)
# 下载视频
local_video_path = None
if video_url:
local_video_path = download_file(video_url, task_id, "video")
else:
print(f"❌ 无法获取视频URL", flush=True)
print()
return video_url, task_id, local_video_path
elif status in ["failed", "cancelled"]:
error = status_data.get("error", {})
error_msg = error.get("message", str(error)) if isinstance(error, dict) else str(error)
raise Exception(f"视频生成 {status}: {error_msg}")
time.sleep(5)
def main():
fix_encoding()
print("Starting the workflow…\\n", flush=True)
# 询问用户是否使用自定义提示词
while True:
choice = input("是否使用自定义提示词?(是y/否n): ").strip().lower()
if choice in ['是', 'y', 'yes']:
# 用户选择使用自定义提示词
user_prompt = input("请输入图片生成的提示词: ").strip()
if user_prompt:
print(f"\\n使用图片提示词: {user_prompt}\\n", flush=True)
# Generate image using user prompt
image_url, image_task_id, local_image_path = generate_image(user_prompt)
# 让用户输入视频生成的提示词
video_prompt = input("\\n图片生成完成!请输入视频生成的提示词(直接回车使用图片提示词): ").strip()
if not video_prompt:
video_prompt = user_prompt
print(f"使用视频提示词: {video_prompt}\\n", flush=True)
# Generate video using user prompt and image
video_url, video_task_id, local_video_path = generate_video(video_prompt, image_url, duration=10)
print("=== 完整流程完成! ===", flush=True)
print(f" – 图片URL: {image_url}", flush=True)
print(f" – 图片本地路径: {local_image_path}", flush=True)
print(f" – 视频URL: {video_url}", flush=True)
print(f" – 视频本地路径: {local_video_path}", flush=True)
else:
print("提示词不能为空,请重新运行程序。", flush=True)
break
elif choice in ['否', 'n', 'no']:
# 使用默认流程
print("\\n使用默认流程生成自然场景…\\n", flush=True)
# Step 1: Generate detailed scene description using text model
scene_prompt = """
Keep the camera still, the frame size unchanged, and the composition consistent.
The people in the scene appear natural.
Suddenly, a gigantic orange (about a quarter of the screen's size) falls from the sky, crashing onto the zebra crossing. Vehicles swerve to avoid it, creating a large crater in the ground.
Use realistic sound effects.
"""
scene_description = generate_text(scene_prompt)
# Step 2: Generate an image using the scene description
image_prompt = scene_description
image_url, image_task_id, local_image_path = generate_image(image_prompt)
# 让用户输入视频生成的提示词
video_prompt = input("\\n图片生成完成!请输入视频生成的提示词(直接回车使用场景描述): ").strip()
if not video_prompt:
video_prompt = scene_description
print(f"使用视频提示词: {video_prompt}\\n", flush=True)
# Step 3: Generate a 10-second video using the scene description and image
video_url, video_task_id, local_video_path = generate_video(video_prompt, image_url, duration=10)
print("=== 完整流程完成! ===", flush=True)
print(f" – 场景描述: 已生成", flush=True)
print(f" – 图片URL: {image_url}", flush=True)
print(f" – 图片本地路径: {local_image_path}", flush=True)
print(f" – 视频URL: {video_url}", flush=True)
print(f" – 视频本地路径: {local_video_path}", flush=True)
break
else:
print("请输入 是 或 否 (y/n)", flush=True)
if __name__ == "__main__":
main()
除了以上一个完整的程序外,更多我的开发过程代码、分步测试过程代码和控制台输出、方便与与你的实践做对比,我都分享在CSDN平台资源中了,喜欢的去 免积分下载,多谢支持
【免费】0Token、全免费AI多模态、无需GPU无需本地部署、Python代码实现调用Agnes,两端提示词实现图片和视频的完美呈现资源-CSDN下载
https://download.csdn.net/download/xiaoma0529/92977839
一个比喻收尾
如果说传统的角色设计像是在画布上一笔一笔手绘,那这套工作流就像请了一个全自动数字片场:你坐在导演椅上对着对讲机喊两句要求,剧组的 AI 美术指导刷刷刷画好概念图,AI 动画师直接让角色动起来,AI 摄影师保持帧帧一致的画风和光影。你什么都不用画、什么都不用改,两段 Prompt 进去,一个会说话的角色视频就出来了。
从收到朋友微信到成品视频落地,前后不到五分钟。我把视频文件丢过去,朋友回了一串感叹号。
如果你也在折腾 AI 生图生视频,这套代码可以直接拿去改。把 API Key 换成你自己的,图片 Prompt 换成你想做的任何画面,视频 Prompt 描述你想看到的任何动作——剩下的交给 Agnes 去跑。周六下午的雨天,适合敲代码,更适合看 AI 替你画画。
整套代码的核心文件:generate_video.py(约 200 行),依赖 requests 库。需要自行在 Agnes AI 平台 注册获取 API Key,替换代码中的 API_KEY 变量即可运行。
如果你用这套代码做出了什么好玩的东西,评论区丢个链接——我很想看看你的 Prompt 想象力能飞出多高。





