feat: 添加 ad_creator_agent 和 video_generator_agent

- ad_creator_agent: 多模态广告创意生成,支持 Gemini/GPT Image/DALL-E 图片生成和 Sora 视频生成
- video_generator_agent: 根据描述生成图片并拼接为视频
- 包含 Dockerfile、API 文档和测试

Made-with: Cursor
This commit is contained in:
zhanggangyong
2026-03-02 15:17:02 +00:00
parent 33955b68dc
commit a34d5a081e
19 changed files with 3846 additions and 0 deletions
@@ -0,0 +1,466 @@
# Ad Creator Agent - API 文档
多模态广告创意生成 Agent,通过素材(文字描述/参考图片)生成广告图片或视频。
## 基本信息
| 项目 | 值 |
|------|------|
| 镜像 | `agnettaiji.azurecr.io/ai-agents/ad-creator-agent:latest` |
| 端口 | `8000` |
| 模板名 | `ad_creator_agent` |
| 框架 | API (FastAPI) |
## 支持的模型
| 用途 | 模型 | 备注 |
|------|------|------|
| 图片生成(默认) | `taiji/gemini-3-pro-image-preview` | 支持参考图片输入 |
| 图片生成 | `taiji/gpt-image-1` | OpenAI GPT Image |
| 图片生成 | `taiji/gpt-image-1-mini` | 轻量版,速度更快 |
| 图片生成 | `taiji/dall-e-3` | DALL-E 3 |
| 文案生成 | `taiji/gpt-4o-mini` | 广告文案 + 图片 prompt |
| 视频生成 | `taiji/sora-2` | Sora 视频生成 |
## 认证方式
所有写操作端点均需传入 API Key,支持以下两种方式:
```
api-key: sk-xxx
```
```
Authorization: Bearer sk-xxx
```
如果部署时配置了 `LLM_API_KEY` 环境变量,可省略请求头中的 Key。
## 环境变量
| 变量名 | 说明 | 默认值 |
|--------|------|--------|
| `LLM_API_KEY` | LiteLLM API Key | (必填或请求头传入) |
| `LLM_BASE_URL` | LiteLLM Base URL | `https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1` |
| `DEFAULT_IMAGE_MODEL` | 默认图片模型 | `taiji/gemini-3-pro-image-preview` |
| `DEFAULT_TEXT_MODEL` | 默认文案模型 | `taiji/gpt-4o-mini` |
| `DEFAULT_VIDEO_MODEL` | 默认视频模型 | `taiji/sora-2` |
| `SERVICE_PORT` | 服务端口 | `8000` |
| `OUTPUT_DIR` | 文件输出目录 | `/app/outputs` |
---
## API 端点
### 1. 健康检查
**GET** `/health`
```bash
curl http://<AGENT_URL>/health
```
**响应示例:**
```json
{
"status": "healthy",
"service": "Ad Creator Agent",
"pod_name": "test-ad-creator",
"models": {
"image": "taiji/gemini-3-pro-image-preview",
"text": "taiji/gpt-4o-mini",
"video": "taiji/sora-2"
},
"callback_enabled": false,
"timestamp": "2026-03-02T14:52:15.109589"
}
```
---
### 2. 生成广告图片
**POST** `/api/v1/generate-image`
通过文字描述生成广告图片,可指定模型、风格、宽高比等。
**请求体:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `prompt` | string | 是 | 广告图片描述/创意需求 |
| `model` | string | 否 | 模型名称,默认 `taiji/gemini-3-pro-image-preview` |
| `aspect_ratio` | string | 否 | 宽高比: `1:1`, `16:9`, `9:16`, `4:3`, `3:4`(Gemini) |
| `size` | string | 否 | 图片尺寸(仅 GPT/DALL-E): `1024x1024`, `1024x1792`, `1792x1024` |
| `quality` | string | 否 | 质量: `low`, `medium`, `high`(默认 `high`) |
| `style` | string | 否 | 广告风格: `modern`, `minimalist`, `luxury`, `playful`, `tech`, `vintage` |
| `brand_name` | string | 否 | 品牌名称 |
| `reference_image_b64` | string | 否 | 参考图片 base64(仅 Gemini 支持) |
**示例 - Gemini 生成:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-image \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"prompt": "A premium headphone floating against dark gradient background with golden light accents",
"aspect_ratio": "1:1",
"quality": "high",
"style": "luxury",
"brand_name": "SoundElite"
}'
```
**示例 - GPT Image 生成:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-image \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"prompt": "A vibrant Instagram ad for a coffee brand with warm morning light",
"model": "taiji/gpt-image-1",
"size": "1024x1024",
"quality": "high"
}'
```
**响应示例:**
```json
{
"success": true,
"file_path": "/app/outputs/images/ad_gemini_20260302_145310_209307.jpg",
"filename": "ad_gemini_20260302_145310_209307.jpg",
"url": "/api/v1/files/ad_gemini_20260302_145310_209307.jpg",
"model": "taiji/gemini-3-pro-image-preview"
}
```
---
### 3. 上传参考图片并生成广告图
**POST** `/api/v1/generate-image-upload`
支持 `multipart/form-data` 上传参考图片,结合文字描述生成广告图。
**表单字段:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `prompt` | string | 是 | 广告图片描述 |
| `reference_image` | file | 否 | 参考图片文件 |
| `model` | string | 否 | 模型名称 |
| `aspect_ratio` | string | 否 | 宽高比 |
| `quality` | string | 否 | 质量 |
| `style` | string | 否 | 广告风格 |
| `brand_name` | string | 否 | 品牌名称 |
**示例:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-image-upload \
-H "api-key: sk-xxx" \
-F "prompt=基于这张产品图,生成一张高端产品广告海报" \
-F "reference_image=@product_photo.jpg" \
-F "style=luxury" \
-F "aspect_ratio=16:9"
```
---
### 4. 生成广告文案
**POST** `/api/v1/generate-copy`
根据产品信息,由 LLM 生成结构化广告文案(标题、正文、CTA、hashtags)以及用于图片生成的英文 prompt。
**请求体:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `product` | string | 是 | 产品/服务描述 |
| `target_audience` | string | 否 | 目标受众 |
| `tone` | string | 否 | 语气: `professional`, `casual`, `humorous`, `urgent`, `luxury` |
| `platform` | string | 否 | 投放平台: `instagram`, `facebook`, `tiktok`, `billboard`, `general` |
| `language` | string | 否 | 语言: `zh`, `en`, `ja`(默认 `zh`) |
**示例:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-copy \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"product": "高端无线降噪耳机,主打沉浸式音乐体验",
"target_audience": "音乐爱好者和商务人士",
"tone": "luxury",
"platform": "instagram",
"language": "zh"
}'
```
**响应示例:**
```json
{
"success": true,
"headline": "沉浸高端音质",
"body_copy": "体验非凡音质,尽享音乐带来的宁静与专注...",
"cta": "立即体验",
"image_prompt": "A luxurious setting featuring a sleek wireless headphone...",
"hashtags": ["#高端耳机", "#沉浸音乐", "#商务生活"]
}
```
---
### 5. 一键生成完整广告(文案 + 图片)
**POST** `/api/v1/generate-ad`
自动生成广告文案,并基于文案中的图片 prompt 自动生成配图。
**请求体:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `product` | string | 是 | 产品/服务描述 |
| `image_model` | string | 否 | 图片生成模型 |
| `aspect_ratio` | string | 否 | 宽高比 |
| `style` | string | 否 | 广告风格 |
| `brand_name` | string | 否 | 品牌名称 |
| `target_audience` | string | 否 | 目标受众 |
| `tone` | string | 否 | 语气 |
| `platform` | string | 否 | 投放平台 |
| `language` | string | 否 | 语言 |
| `reference_image_b64` | string | 否 | 参考图片 base64 |
**示例:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-ad \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"product": "新能源电动汽车,零排放、高续航、智能驾驶",
"target_audience": "环保意识强的中产家庭",
"tone": "professional",
"platform": "facebook",
"language": "zh",
"style": "tech",
"brand_name": "GreenDrive"
}'
```
**响应示例:**
```json
{
"success": true,
"copy": {
"success": true,
"headline": "开启绿色出行新生活",
"body_copy": "选择我们的新能源电动汽车...",
"cta": "立即了解更多",
"image_prompt": "A futuristic electric vehicle...",
"hashtags": ["#新能源车", "#绿色出行", "#智能驾驶"]
},
"image": {
"success": true,
"filename": "ad_gemini_20260302_145504_262223.jpg",
"url": "/api/v1/files/ad_gemini_20260302_145504_262223.jpg",
"model": "taiji/gemini-3-pro-image-preview"
},
"timestamp": "2026-03-02T14:55:04.262223"
}
```
---
### 6. 生成广告视频
**POST** `/api/v1/generate-video`
使用 Sora 模型生成广告短视频。
**请求体:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `prompt` | string | 是 | 视频描述/创意需求 |
| `model` | string | 否 | 视频模型(默认 `taiji/sora-2`) |
| `aspect_ratio` | string | 否 | 宽高比: `16:9`, `9:16`, `1:1` |
| `duration` | string | 否 | 视频时长秒数(默认 `5`) |
**示例:**
```bash
curl -X POST http://<AGENT_URL>/api/v1/generate-video \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"prompt": "A sleek electric car driving through a futuristic city at sunset, cinematic style",
"aspect_ratio": "16:9",
"duration": "5"
}'
```
---
### 7. 智能对话
**POST** `/chat`
与 AI 广告创意总监对话。系统会理解需求,自动决定是否生成图片。
**请求体:**
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `message` | string | 是 | 用户消息 |
**示例:**
```bash
curl -X POST http://<AGENT_URL>/chat \
-H "Content-Type: application/json" \
-H "api-key: sk-xxx" \
-d '{
"message": "帮我为一款蓝牙音箱做一个抖音封面图,要有科技感"
}'
```
**响应示例:**
```json
{
"response": "为这款蓝牙音箱设计封面图的建议...",
"image": {
"success": true,
"filename": "ad_gemini_20260302_145539_866923.jpg",
"url": "/api/v1/files/ad_gemini_20260302_145539_866923.jpg",
"model": "taiji/gemini-3-pro-image-preview"
},
"timestamp": "2026-03-02T14:55:39.866923"
}
```
---
### 8. 下载生成的文件
**GET** `/api/v1/files/{filename}`
```bash
curl -O http://<AGENT_URL>/api/v1/files/ad_gemini_20260302_145310_209307.jpg
```
---
### 9. 列出已生成的文件
**GET** `/api/v1/list-files?file_type=all`
参数 `file_type` 可选值: `all`, `image`, `video`
```bash
curl http://<AGENT_URL>/api/v1/list-files
```
**响应示例:**
```json
{
"images": [
{
"filename": "ad_gemini_20260302_145539_866923.jpg",
"url": "/api/v1/files/ad_gemini_20260302_145539_866923.jpg",
"size_bytes": 589722,
"created_at": "2026-03-02T14:55:39.865520"
}
],
"videos": []
}
```
---
### 10. 清理旧文件
**POST** `/api/v1/cleanup?max_age_hours=24`
删除超过指定时间的旧文件。
```bash
curl -X POST "http://<AGENT_URL>/api/v1/cleanup?max_age_hours=24"
```
---
### 11. 状态查看
**GET** `/status`
```bash
curl http://<AGENT_URL>/status
```
**响应示例:**
```json
{
"status": "running",
"pod_name": "test-ad-creator",
"generated_images": 4,
"generated_videos": 0,
"timestamp": "2026-03-02T15:01:43.636444"
}
```
---
## 通过 Agent Manager 部署
### 1. 注册模板
```bash
curl -X POST http://20.212.121.126/templates/create \
-H "Content-Type: application/json" \
-d '{
"name": "ad_creator_agent",
"display_name": "Ad Creator Agent",
"description": "多模态广告创意生成 Agent",
"image": "agnettaiji.azurecr.io/ai-agents/ad-creator-agent:latest",
"port": 8000,
"agent_type": "platform",
"agent_framework": "api",
"env_requirements": {
"LLM_API_KEY": "LiteLLM API Key"
}
}'
```
### 2. 创建实例
```bash
curl -X POST http://20.212.121.126/agents \
-H "Content-Type: application/json" \
-d '{
"name": "my-ad-creator",
"template": "ad_creator_agent",
"config": { "user_id": "your-user-id" },
"env": {
"LLM_API_KEY": "sk-your-api-key"
}
}'
```
### 3. 删除实例
```bash
curl -X DELETE http://20.212.121.126/agents/my-ad-creator
```
@@ -0,0 +1,35 @@
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir \
fastapi==0.109.0 \
uvicorn[standard]==0.27.0 \
pydantic==2.5.3 \
aiohttp>=3.9.0 \
python-multipart>=0.0.6
COPY common/agent_callback_utils.py /app/common/
RUN touch /app/common/__init__.py
COPY agents/ad_creator_agent/ad_creator_agent.py /app/
ENV PYTHONUNBUFFERED=1
ENV SERVICE_HOST=0.0.0.0
ENV SERVICE_PORT=8000
ENV OUTPUT_DIR=/app/outputs
ENV AGENT_CALLBACK_URL=http://mcp-server:8002/api/v1/billing/agent-callback
RUN mkdir -p /app/outputs/images /app/outputs/videos
HEALTHCHECK --interval=30s --timeout=10s --start-period=20s --retries=3 \
CMD python3 -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()" || exit 1
EXPOSE 8000
CMD ["python3", "-u", "ad_creator_agent.py"]
@@ -0,0 +1,793 @@
"""
Ad Creator Agent - 多模态广告创意生成 Agent
通过素材(文字描述/参考图片)生成广告图片或视频
支持模型:Gemini 3 Pro Image / GPT Image 1 / DALL-E 3 / Sora 2
"""
import os
import sys
import re
import uuid
import json
import base64
import logging
import aiohttp
from typing import Optional, List, Dict, Any
from datetime import datetime
from pathlib import Path
from enum import Enum
from fastapi import FastAPI, HTTPException, Header, Depends, UploadFile, File, Form, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, Field
import uvicorn
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
try:
from common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager
CALLBACK_ENABLED = True
except ImportError:
CALLBACK_ENABLED = False
AgentCallbackHandler = None
CallbackContextManager = None
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# ==================== 环境变量 ====================
SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0")
SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8000"))
POD_NAME = os.getenv("POD_NAME", "ad-creator-agent")
USER_ID = os.getenv("USER_ID", "")
OUTPUT_DIR = os.getenv("OUTPUT_DIR", "/app/outputs")
LLM_BASE_URL = os.getenv(
"LLM_BASE_URL",
"https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1"
)
LLM_API_KEY = os.getenv("LLM_API_KEY", "")
DEFAULT_IMAGE_MODEL = os.getenv("DEFAULT_IMAGE_MODEL", "taiji/gemini-3-pro-image-preview")
DEFAULT_TEXT_MODEL = os.getenv("DEFAULT_TEXT_MODEL", "taiji/gpt-4o-mini")
DEFAULT_VIDEO_MODEL = os.getenv("DEFAULT_VIDEO_MODEL", "taiji/sora-2")
Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
Path(f"{OUTPUT_DIR}/images").mkdir(parents=True, exist_ok=True)
Path(f"{OUTPUT_DIR}/videos").mkdir(parents=True, exist_ok=True)
# ==================== 模型枚举 ====================
class ImageModel(str, Enum):
GEMINI = "taiji/gemini-3-pro-image-preview"
GPT_IMAGE = "taiji/gpt-image-1"
GPT_IMAGE_MINI = "taiji/gpt-image-1-mini"
DALLE3 = "taiji/dall-e-3"
class AspectRatio(str, Enum):
SQUARE = "1:1"
LANDSCAPE = "16:9"
PORTRAIT = "9:16"
WIDE = "4:3"
TALL = "3:4"
# ==================== 请求/响应模型 ====================
class GenerateImageRequest(BaseModel):
prompt: str = Field(..., description="广告图片描述/创意需求")
model: Optional[str] = Field(None, description="图片生成模型,默认 gemini-3-pro-image-preview")
aspect_ratio: Optional[str] = Field("1:1", description="宽高比: 1:1, 16:9, 9:16, 4:3, 3:4")
size: Optional[str] = Field("1024x1024", description="图片尺寸(仅 GPT/DALL-E)")
quality: Optional[str] = Field("high", description="质量: low, medium, high")
style: Optional[str] = Field(None, description="广告风格: modern, minimalist, luxury, playful, tech, vintage")
brand_name: Optional[str] = Field(None, description="品牌名称")
reference_image_b64: Optional[str] = Field(None, description="参考图片 base64 (用于风格参考或产品素材)")
user_id: Optional[str] = Field(None, description="用户ID")
class GenerateAdCopyRequest(BaseModel):
product: str = Field(..., description="产品/服务描述")
target_audience: Optional[str] = Field(None, description="目标受众")
tone: Optional[str] = Field("professional", description="语气: professional, casual, humorous, urgent, luxury")
platform: Optional[str] = Field("general", description="投放平台: instagram, facebook, tiktok, billboard, general")
language: Optional[str] = Field("zh", description="语言: zh, en, ja")
user_id: Optional[str] = Field(None, description="用户ID")
class GenerateAdRequest(BaseModel):
"""完整广告生成(文案 + 图片)"""
product: str = Field(..., description="产品/服务描述")
image_model: Optional[str] = Field(None, description="图片生成模型")
aspect_ratio: Optional[str] = Field("1:1", description="宽高比")
style: Optional[str] = Field(None, description="广告风格")
brand_name: Optional[str] = Field(None, description="品牌名称")
target_audience: Optional[str] = Field(None, description="目标受众")
tone: Optional[str] = Field("professional", description="语气")
platform: Optional[str] = Field("general", description="投放平台")
language: Optional[str] = Field("zh", description="语言")
reference_image_b64: Optional[str] = Field(None, description="参考图片 base64")
user_id: Optional[str] = Field(None, description="用户ID")
class GenerateVideoRequest(BaseModel):
prompt: str = Field(..., description="视频描述/创意需求")
model: Optional[str] = Field(None, description="视频模型, 默认 sora-2")
aspect_ratio: Optional[str] = Field("16:9", description="宽高比")
duration: Optional[str] = Field("5", description="视频时长秒数")
user_id: Optional[str] = Field(None, description="用户ID")
class ChatRequest(BaseModel):
message: str = Field(..., description="用户消息")
user_id: Optional[str] = Field(None, description="用户ID")
# ==================== FastAPI ====================
app = FastAPI(
title="Ad Creator Agent",
description="多模态广告创意生成 - 通过素材生成广告图片或视频(支持 Gemini / GPT Image / DALL-E / Sora)",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
callback_handler: Optional[Any] = None
# ==================== 依赖 ====================
async def get_api_key(
api_key: Optional[str] = Header(None, alias="api-key"),
authorization: Optional[str] = Header(None),
) -> str:
if api_key and api_key.strip():
return api_key.strip()
if authorization:
key = authorization[7:].strip() if authorization.startswith("Bearer ") else authorization.strip()
if key:
return key
if LLM_API_KEY:
return LLM_API_KEY
raise HTTPException(status_code=401, detail="请在请求头提供 api-key 或 Authorization Bearer token")
# ==================== 生命周期 ====================
@app.on_event("startup")
async def startup_event():
global callback_handler
if CALLBACK_ENABLED:
callback_handler = AgentCallbackHandler(agent_name=POD_NAME, user_id=USER_ID)
logger.info(f"回调处理器已初始化: {callback_handler.callback_url}")
else:
logger.warning("回调模块未加载")
logger.info(f"Ad Creator Agent 启动: port={SERVICE_PORT}, output={OUTPUT_DIR}")
logger.info(f"默认模型: image={DEFAULT_IMAGE_MODEL}, text={DEFAULT_TEXT_MODEL}, video={DEFAULT_VIDEO_MODEL}")
# ==================== 核心:图片生成 ====================
async def generate_image_gemini(
prompt: str,
api_key: str,
aspect_ratio: str = "1:1",
quality: str = "high",
reference_image_b64: str = None,
) -> dict:
"""通过 Gemini chat/completions 生成图片,返回 {success, file_path, filename, url}"""
quality_map = {"low": "1K", "medium": "1K", "high": "2K"}
image_size = quality_map.get(quality, "2K")
messages_content: Any
if reference_image_b64:
messages_content = [
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{reference_image_b64}"},
},
{"type": "text", "text": prompt},
]
else:
messages_content = prompt
payload = {
"model": DEFAULT_IMAGE_MODEL,
"stream": False,
"messages": [{"role": "user", "content": messages_content}],
"extra_body": {
"google": {
"image_config": {
"aspect_ratio": aspect_ratio,
"image_size": image_size,
}
}
},
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
base = LLM_BASE_URL.rstrip("/")
url = f"{base}/chat/completions"
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=120)) as resp:
if resp.status != 200:
text = await resp.text()
return {"success": False, "error": f"Gemini API error {resp.status}: {text[:500]}"}
data = await resp.json()
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
match = re.search(r"data:image/([\w+]+);base64,([A-Za-z0-9+/=\s]+)", content)
if not match:
return {"success": False, "error": "Gemini 未返回图片数据", "text_response": content[:500]}
img_format = match.group(1).replace("+xml", "")
ext = "jpg" if img_format == "jpeg" else img_format
b64_data = match.group(2).replace("\n", "").replace(" ", "")
ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
filename = f"ad_gemini_{ts}.{ext}"
file_path = os.path.join(OUTPUT_DIR, "images", filename)
with open(file_path, "wb") as f:
f.write(base64.b64decode(b64_data))
logger.info(f"Gemini 图片已生成: {file_path} ({os.path.getsize(file_path)} bytes)")
return {
"success": True,
"file_path": file_path,
"filename": filename,
"url": f"/api/v1/files/{filename}",
"model": DEFAULT_IMAGE_MODEL,
}
async def generate_image_openai(
prompt: str,
api_key: str,
model: str = "taiji/gpt-image-1",
size: str = "1024x1024",
quality: str = "high",
) -> dict:
"""通过 OpenAI images/generations 接口生成图片(GPT Image / DALL-E)"""
quality_map = {"low": "low", "medium": "medium", "high": "high"}
payload = {
"model": model,
"prompt": prompt,
"n": 1,
"size": size,
"quality": quality_map.get(quality, "high"),
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
base = LLM_BASE_URL.rstrip("/")
url = f"{base}/images/generations"
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=120)) as resp:
if resp.status != 200:
text = await resp.text()
return {"success": False, "error": f"OpenAI API error {resp.status}: {text[:500]}"}
data = await resp.json()
items = data.get("data", [])
if not items:
return {"success": False, "error": "API 未返回图片数据"}
item = items[0]
b64_data = item.get("b64_json")
image_url = item.get("url")
ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
model_tag = model.split("/")[-1].replace("-", "")
filename = f"ad_{model_tag}_{ts}.png"
file_path = os.path.join(OUTPUT_DIR, "images", filename)
if b64_data:
with open(file_path, "wb") as f:
f.write(base64.b64decode(b64_data))
elif image_url:
async with session.get(image_url, timeout=aiohttp.ClientTimeout(total=30)) as dl_resp:
if dl_resp.status != 200:
return {"success": False, "error": f"下载图片失败: HTTP {dl_resp.status}"}
with open(file_path, "wb") as f:
f.write(await dl_resp.read())
else:
return {"success": False, "error": "API 响应中无图片数据"}
logger.info(f"OpenAI 图片已生成: {file_path} ({os.path.getsize(file_path)} bytes)")
return {
"success": True,
"file_path": file_path,
"filename": filename,
"url": f"/api/v1/files/{filename}",
"model": model,
}
async def generate_image_dispatch(
prompt: str,
api_key: str,
model: str = None,
aspect_ratio: str = "1:1",
size: str = "1024x1024",
quality: str = "high",
reference_image_b64: str = None,
) -> dict:
"""根据模型分发到对应生成函数"""
model = model or DEFAULT_IMAGE_MODEL
if "gemini" in model.lower():
return await generate_image_gemini(
prompt=prompt,
api_key=api_key,
aspect_ratio=aspect_ratio,
quality=quality,
reference_image_b64=reference_image_b64,
)
else:
return await generate_image_openai(
prompt=prompt,
api_key=api_key,
model=model,
size=size,
quality=quality,
)
# ==================== 核心:文案生成 ====================
async def generate_ad_copy(
product: str,
api_key: str,
target_audience: str = None,
tone: str = "professional",
platform: str = "general",
language: str = "zh",
) -> dict:
"""用 LLM 生成广告文案和图片 prompt"""
lang_map = {"zh": "中文", "en": "English", "ja": "日本語"}
lang_label = lang_map.get(language, language)
system_prompt = f"""你是一位顶尖的广告创意总监。请根据产品信息生成广告创意方案。
回复必须使用 {lang_label}。
回复格式必须严格为以下 JSON(不包含 markdown 代码块标记):
{{
"headline": "广告标题(10字以内)",
"body_copy": "广告正文(30-60字)",
"cta": "行动号召按钮文字",
"image_prompt": "用于 AI 生成广告配图的英文详细描述(100-200 words, 描述画面构图、色调、元素)",
"hashtags": ["标签1", "标签2", "标签3"]
}}"""
user_msg = f"产品/服务: {product}"
if target_audience:
user_msg += f"\n目标受众: {target_audience}"
user_msg += f"\n语气风格: {tone}\n投放平台: {platform}"
payload = {
"model": DEFAULT_TEXT_MODEL,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_msg},
],
"max_tokens": 800,
"temperature": 0.8,
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
base = LLM_BASE_URL.rstrip("/")
url = f"{base}/chat/completions"
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
if resp.status != 200:
text = await resp.text()
return {"success": False, "error": f"LLM API error {resp.status}: {text[:300]}"}
data = await resp.json()
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
try:
cleaned = content.strip()
if cleaned.startswith("```"):
cleaned = re.sub(r"^```\w*\n?", "", cleaned)
cleaned = re.sub(r"\n?```$", "", cleaned)
copy_data = json.loads(cleaned)
except json.JSONDecodeError:
return {
"success": True,
"raw_text": content,
"headline": "",
"body_copy": content[:200],
"cta": "",
"image_prompt": "",
"hashtags": [],
}
return {
"success": True,
"headline": copy_data.get("headline", ""),
"body_copy": copy_data.get("body_copy", ""),
"cta": copy_data.get("cta", ""),
"image_prompt": copy_data.get("image_prompt", ""),
"hashtags": copy_data.get("hashtags", []),
}
# ==================== 核心:视频生成 ====================
async def generate_video_sora(
prompt: str,
api_key: str,
model: str = None,
aspect_ratio: str = "16:9",
duration: str = "5",
) -> dict:
"""通过 Sora 生成视频(OpenAI 兼容接口)"""
model = model or DEFAULT_VIDEO_MODEL
size_map = {
"16:9": "1920x1080",
"9:16": "1080x1920",
"1:1": "1080x1080",
}
size = size_map.get(aspect_ratio, "1920x1080")
payload = {
"model": model,
"input": prompt,
"size": size,
"duration": int(duration),
"n": 1,
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
base = LLM_BASE_URL.rstrip("/")
url = f"{base}/videos/generations"
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=300)) as resp:
if resp.status != 200:
text = await resp.text()
return {"success": False, "error": f"Sora API error {resp.status}: {text[:500]}"}
data = await resp.json()
items = data.get("data", [])
if not items:
return {"success": False, "error": "Sora 未返回视频数据"}
video_url = items[0].get("url")
b64_data = items[0].get("b64_json")
ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
filename = f"ad_video_{ts}.mp4"
file_path = os.path.join(OUTPUT_DIR, "videos", filename)
if b64_data:
with open(file_path, "wb") as f:
f.write(base64.b64decode(b64_data))
elif video_url:
async with session.get(video_url, timeout=aiohttp.ClientTimeout(total=120)) as dl_resp:
if dl_resp.status != 200:
return {"success": False, "error": f"下载视频失败: HTTP {dl_resp.status}"}
with open(file_path, "wb") as f:
f.write(await dl_resp.read())
else:
return {"success": False, "error": "Sora 响应中无视频数据"}
logger.info(f"视频已生成: {file_path} ({os.path.getsize(file_path)} bytes)")
return {
"success": True,
"file_path": file_path,
"filename": filename,
"url": f"/api/v1/files/{filename}",
"model": model,
}
# ==================== API 端点 ====================
@app.get("/")
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"service": "Ad Creator Agent",
"pod_name": POD_NAME,
"models": {
"image": DEFAULT_IMAGE_MODEL,
"text": DEFAULT_TEXT_MODEL,
"video": DEFAULT_VIDEO_MODEL,
},
"callback_enabled": CALLBACK_ENABLED,
"timestamp": datetime.utcnow().isoformat(),
}
@app.get("/status")
async def status():
images = list(Path(f"{OUTPUT_DIR}/images").glob("*"))
videos = list(Path(f"{OUTPUT_DIR}/videos").glob("*"))
return {
"status": "running",
"pod_name": POD_NAME,
"generated_images": len(images),
"generated_videos": len(videos),
"timestamp": datetime.utcnow().isoformat(),
}
# ---------- 图片生成 ----------
@app.post("/api/v1/generate-image")
async def api_generate_image(request: GenerateImageRequest, api_key: str = Depends(get_api_key)):
"""生成广告图片"""
prompt = request.prompt
if request.style:
prompt = f"[{request.style} style] {prompt}"
if request.brand_name:
prompt = f"{prompt}. Brand: {request.brand_name}"
result = await generate_image_dispatch(
prompt=prompt,
api_key=api_key,
model=request.model,
aspect_ratio=request.aspect_ratio or "1:1",
size=request.size or "1024x1024",
quality=request.quality or "high",
reference_image_b64=request.reference_image_b64,
)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "图片生成失败"))
return result
@app.post("/api/v1/generate-image-upload")
async def api_generate_image_with_upload(
prompt: str = Form(..., description="广告图片描述"),
model: Optional[str] = Form(None),
aspect_ratio: Optional[str] = Form("1:1"),
quality: Optional[str] = Form("high"),
style: Optional[str] = Form(None),
brand_name: Optional[str] = Form(None),
reference_image: Optional[UploadFile] = File(None, description="参考图片文件"),
api_key: str = Depends(get_api_key),
):
"""上传参考图片并生成广告图(multipart/form-data)"""
ref_b64 = None
if reference_image:
content = await reference_image.read()
ref_b64 = base64.b64encode(content).decode("utf-8")
full_prompt = prompt
if style:
full_prompt = f"[{style} style] {full_prompt}"
if brand_name:
full_prompt = f"{full_prompt}. Brand: {brand_name}"
result = await generate_image_dispatch(
prompt=full_prompt,
api_key=api_key,
model=model,
aspect_ratio=aspect_ratio or "1:1",
size="1024x1024",
quality=quality or "high",
reference_image_b64=ref_b64,
)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "图片生成失败"))
return result
# ---------- 文案生成 ----------
@app.post("/api/v1/generate-copy")
async def api_generate_copy(request: GenerateAdCopyRequest, api_key: str = Depends(get_api_key)):
"""生成广告文案"""
result = await generate_ad_copy(
product=request.product,
api_key=api_key,
target_audience=request.target_audience,
tone=request.tone or "professional",
platform=request.platform or "general",
language=request.language or "zh",
)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "文案生成失败"))
return result
# ---------- 完整广告生成(文案 + 图片) ----------
@app.post("/api/v1/generate-ad")
async def api_generate_ad(request: GenerateAdRequest, api_key: str = Depends(get_api_key)):
"""一键生成完整广告(文案 + 配图)"""
copy_result = await generate_ad_copy(
product=request.product,
api_key=api_key,
target_audience=request.target_audience,
tone=request.tone or "professional",
platform=request.platform or "general",
language=request.language or "zh",
)
image_prompt = copy_result.get("image_prompt", "")
if not image_prompt:
image_prompt = f"Advertisement for: {request.product}"
if request.style:
image_prompt = f"[{request.style} style] {image_prompt}"
if request.brand_name:
image_prompt = f"{image_prompt}. Brand: {request.brand_name}"
image_result = await generate_image_dispatch(
prompt=image_prompt,
api_key=api_key,
model=request.image_model,
aspect_ratio=request.aspect_ratio or "1:1",
quality="high",
reference_image_b64=request.reference_image_b64,
)
return {
"success": True,
"copy": copy_result,
"image": image_result,
"timestamp": datetime.utcnow().isoformat(),
}
# ---------- 视频生成 ----------
@app.post("/api/v1/generate-video")
async def api_generate_video(request: GenerateVideoRequest, api_key: str = Depends(get_api_key)):
"""生成广告视频"""
result = await generate_video_sora(
prompt=request.prompt,
api_key=api_key,
model=request.model,
aspect_ratio=request.aspect_ratio or "16:9",
duration=request.duration or "5",
)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "视频生成失败"))
return result
# ---------- 文件管理 ----------
@app.get("/api/v1/files/{filename}")
async def download_file(filename: str):
"""下载生成的文件"""
for subdir in ["images", "videos"]:
path = os.path.join(OUTPUT_DIR, subdir, filename)
if os.path.exists(path):
ext = filename.rsplit(".", 1)[-1].lower()
media_types = {
"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg",
"webp": "image/webp", "mp4": "video/mp4", "gif": "image/gif",
}
return FileResponse(path, media_type=media_types.get(ext, "application/octet-stream"), filename=filename)
raise HTTPException(status_code=404, detail="文件不存在")
@app.get("/api/v1/list-files")
async def list_files(file_type: str = "all"):
"""列出已生成的文件"""
result = {"images": [], "videos": []}
if file_type in ("all", "image"):
img_dir = Path(f"{OUTPUT_DIR}/images")
for f in sorted(img_dir.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True):
result["images"].append({
"filename": f.name,
"url": f"/api/v1/files/{f.name}",
"size_bytes": f.stat().st_size,
"created_at": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
})
if file_type in ("all", "video"):
vid_dir = Path(f"{OUTPUT_DIR}/videos")
for f in sorted(vid_dir.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True):
result["videos"].append({
"filename": f.name,
"url": f"/api/v1/files/{f.name}",
"size_bytes": f.stat().st_size,
"created_at": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
})
return result
@app.post("/api/v1/cleanup")
async def cleanup_files(max_age_hours: int = 24):
"""清理超过指定时间的旧文件"""
import time
cutoff = time.time() - max_age_hours * 3600
deleted = 0
for subdir in ["images", "videos"]:
d = Path(f"{OUTPUT_DIR}/{subdir}")
for f in d.glob("*"):
if f.stat().st_mtime < cutoff:
f.unlink()
deleted += 1
return {"deleted": deleted, "max_age_hours": max_age_hours}
# ---------- 智能对话 ----------
@app.post("/chat")
async def chat(request: ChatRequest, api_key: str = Depends(get_api_key)):
"""智能对话 - 理解需求后自动生成广告创意"""
system_prompt = """你是一位资深广告创意总监 AI 助手。你可以帮用户:
1. 分析产品卖点,构思广告创意方案
2. 生成广告文案和配图描述
3. 推荐合适的广告风格和投放策略
如果用户想生成图片,在回复末尾添加一行:[GENERATE_IMAGE: 你的英文图片生成 prompt]
如果用户只是聊天咨询,正常回复即可。"""
payload = {
"model": DEFAULT_TEXT_MODEL,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": request.message},
],
"max_tokens": 1000,
"temperature": 0.8,
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
base = LLM_BASE_URL.rstrip("/")
url = f"{base}/chat/completions"
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
if resp.status != 200:
text = await resp.text()
raise HTTPException(status_code=500, detail=f"LLM error {resp.status}: {text[:300]}")
data = await resp.json()
llm_reply = data.get("choices", [{}])[0].get("message", {}).get("content", "")
image_result = None
img_match = re.search(r"\[GENERATE_IMAGE:\s*(.+?)\]", llm_reply)
if img_match:
img_prompt = img_match.group(1)
llm_reply = llm_reply.replace(img_match.group(0), "").strip()
image_result = await generate_image_dispatch(prompt=img_prompt, api_key=api_key)
return {
"response": llm_reply,
"image": image_result,
"timestamp": datetime.utcnow().isoformat(),
}
# ==================== 主入口 ====================
def main():
logger.info(f"启动 Ad Creator Agent - {POD_NAME}")
logger.info(f"回调功能: {'已启用' if CALLBACK_ENABLED else '未启用'}")
uvicorn.run(app, host=SERVICE_HOST, port=SERVICE_PORT, log_level="info")
if __name__ == "__main__":
main()
@@ -0,0 +1,24 @@
__pycache__
*.pyc
*.pyo
*.pyd
.Python
*.so
*.egg
*.egg-info
dist
build
.pytest_cache
.coverage
htmlcov
.venv
venv
outputs/
*.mp4
*.png
*.jpg
*.jpeg
.DS_Store
.env
.vscode
.idea
@@ -0,0 +1,294 @@
# Video Generator Agent - 部署和测试指南
## ✅ 已完成的工作
### 1. 核心功能实现
- ✅ 图片生成工具 (`image_generator.py`)
- ✅ 视频处理工具 (`video_processor.py`) - 使用 FFmpeg
- ✅ 文件管理工具 (`file_manager.py`)
- ✅ MCP 服务器 (`mcp_server.py`) - 4个工具
- ✅ API 服务器 (`api_server.py`) - REST API + MCP HTTP/SSE
### 2. Docker 配置
- ✅ Dockerfile(包含 FFmpeg)
- ✅ requirements.txt
- ✅ .dockerignore
- ✅ 健康检查配置
### 3. 测试和文档
- ✅ 测试脚本 (`test_video_agent.py`)
- ✅ 核心功能测试 (`test_core_functions.py`)
- ✅ README.md 完整文档
- ✅ Docker 镜像构建成功
- ✅ FFmpeg 验证通过
- ✅ 模块导入验证通过
## 🚀 快速部署
### 方法 1: Docker 部署(推荐)
```bash
# 1. 构建镜像
cd agent_templates/agents/video_generator_agent
docker build -t video-generator-agent:latest .
# 2. 运行容器
docker run -d \
--name video-generator \
-p 8000:8000 \
-e OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-e MODEL_NAME="dall-e-3" \
-v $(pwd)/outputs:/app/outputs \
video-generator-agent:latest
# 3. 查看日志
docker logs -f video-generator
# 4. 测试健康检查
curl http://localhost:8000/health
```
### 方法 2: 本地运行
```bash
# 1. 创建虚拟环境
python3 -m venv venv
source venv/bin/activate
# 2. 安装依赖
pip install -r requirements.txt
# 3. 安装 FFmpeg(如果未安装)
# Ubuntu/Debian:
sudo apt-get install ffmpeg
# macOS:
brew install ffmpeg
# 4. 设置环境变量
export OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg"
export MODEL_NAME="dall-e-3"
# 5. 启动服务
python run_api_server.py
```
## 🧪 测试
### 1. 健康检查
```bash
curl http://localhost:8000/health
```
预期输出:
```json
{
"status": "healthy",
"service": "Video Generator Agent"
}
```
### 2. 查看服务信息
```bash
curl http://localhost:8000/
```
### 3. 生成单张图片
```bash
curl -X POST http://localhost:8000/api/v1/generate-image \
-H "api-key: sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-H "Content-Type: application/json" \
-d '{
"description": "a beautiful sunset over mountains",
"size": "1024x1024"
}'
```
### 4. 生成视频
```bash
curl -X POST http://localhost:8000/api/v1/generate-video \
-H "api-key: sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-H "Content-Type: application/json" \
-d '{
"descriptions": [
"sunrise over mountains",
"a peaceful lake at noon",
"starry night sky"
],
"duration_per_image": 3,
"transition": "fade"
}'
```
### 5. 运行完整测试套件
```bash
# 使用 Docker
docker run --rm \
-e API_BASE_URL=http://host.docker.internal:8000 \
-e API_KEY=sk-i9AwAgXDqqxsA9Ym4AjSPg \
--network host \
video-generator-agent:latest \
python test_video_agent.py
# 或本地运行
python test_video_agent.py
```
## ⚠️ 重要注意事项
### 1. 模型配置
**当前问题**: 模型 `taiji/gemini-3-pro-image-preview` 不支持图片生成。
**解决方案**: 使用以下支持的模型之一:
```bash
# OpenAI DALL-E 模型
export MODEL_NAME="dall-e-3"
export MODEL_NAME="dall-e-2"
# 或其他 LiteLLM 支持的图片生成模型
```
### 2. API Key
确保使用有效的 API Key:
```bash
export OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg"
```
### 3. FFmpeg 依赖
视频拼接功能需要 FFmpeg。Docker 镜像已包含,本地运行需要手动安装。
验证 FFmpeg:
```bash
ffmpeg -version
```
### 4. 存储空间
生成的图片和视频会占用存储空间。建议:
- 定期清理旧文件
- 使用 Docker volume 持久化数据
- 监控磁盘使用情况
## 📊 测试结果
### Docker 构建
- ✅ 镜像构建成功
- ✅ FFmpeg 7.1.3 已安装
- ✅ Python 依赖已安装
- ✅ 健康检查配置正确
### 模块验证
- ✅ 所有 Python 模块导入成功
- ✅ FFmpeg 可执行
- ✅ 服务启动正常
### 功能状态
- ✅ REST API 端点正常
- ✅ MCP 端点正常
- ✅ 健康检查通过
- ⚠️ 图片生成需要正确的模型配置
- ✅ 视频拼接功能就绪(FFmpeg)
- ✅ 文件管理功能正常
## 🔧 故障排除
### 问题 1: 图片生成失败
**错误**: `not supported model for image generation`
**解决**:
```bash
# 更改模型为支持的模型
docker run -d \
-e MODEL_NAME="dall-e-3" \
...
```
### 问题 2: 端口被占用
**错误**: `Bind for 0.0.0.0:8000 failed: port is already allocated`
**解决**:
```bash
# 使用不同端口
docker run -d -p 8765:8000 ...
```
### 问题 3: FFmpeg 未找到
**错误**: `FFmpeg not available`
**解决**:
```bash
# Ubuntu/Debian
sudo apt-get install ffmpeg
# macOS
brew install ffmpeg
```
## 📝 下一步
1. **配置正确的图片生成模型**
- 联系 LiteLLM Gateway 管理员
- 确认可用的图片生成模型
- 更新 MODEL_NAME 环境变量
2. **运行完整测试**
```bash
python test_video_agent.py
```
3. **集成到 Agent Manager**
- 在 `k8s_manager.py` 中添加配置
- 在 `app.py` 中注册模板
- 部署到 Kubernetes
4. **生产环境优化**
- 配置持久化存储
- 设置资源限制
- 配置日志收集
- 添加监控告警
## 📚 相关文档
- [README.md](./README.md) - 完整使用文档
- [test_video_agent.py](./test_video_agent.py) - 测试脚本
- [Dockerfile](./Dockerfile) - Docker 配置
## ✨ 功能亮点
1. **完整的 MCP 支持** - 4个工具,支持 HTTP 和 SSE
2. **FFmpeg 视频处理** - 支持多种转场效果
3. **文件管理** - 自动清理、存储统计
4. **Docker 化** - 开箱即用,包含所有依赖
5. **完善的文档** - README、测试脚本、部署指南
## 🎯 总结
Video Generator Agent 已经完成开发和基础测试:
✅ **已完成**:
- 核心功能实现(图片生成、视频拼接、文件管理)
- MCP 和 REST API 服务器
- Docker 配置和构建
- 测试脚本和文档
- FFmpeg 集成和验证
⚠️ **待配置**:
- 正确的图片生成模型(当前模型不支持)
🚀 **可以部署**:
- Docker 镜像已就绪
- 服务可以启动
- 视频拼接功能完整
- 只需配置正确的模型即可使用图片生成功能
@@ -0,0 +1,33 @@
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
# 安装系统依赖(包括 FFmpeg)
RUN apt-get update && apt-get install -y \
ffmpeg \
gcc \
curl \
&& rm -rf /var/lib/apt/lists/*
# 验证 FFmpeg 安装
RUN ffmpeg -version
# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建输出目录
RUN mkdir -p /app/outputs/images /app/outputs/videos
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
CMD ["python", "run_api_server.py"]
@@ -0,0 +1,372 @@
# Video Generator Agent
根据文本描述生成图片并拼接为视频的 AI Agent。
## 功能特性
- 🎨 **图片生成**: 使用 Gemini 3 Pro Image Preview 根据描述生成高质量图片
- 🎬 **视频拼接**: 使用 FFmpeg 将多张图片拼接成视频
- ✨ **转场效果**: 支持多种转场效果(fade, wipeleft, wiperight, slideup, slidedown)
- 📁 **文件管理**: 自动管理生成的图片和视频文件
- 🔌 **MCP 协议**: 支持 MCP (Model Context Protocol) 工具调用
- 🌐 **REST API**: 提供完整的 HTTP API 接口
## 快速开始
### 1. 本地运行
```bash
# 安装依赖
pip install -r requirements.txt
# 设置环境变量
export OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg"
export MODEL_NAME="taiji/gemini-3-pro-image-preview"
# 启动服务
python run_api_server.py
```
服务将在 `http://localhost:8000` 启动。
### 2. Docker 运行
```bash
# 构建镜像
docker build -t video-generator-agent:latest .
# 运行容器
docker run -d \
-p 8000:8000 \
-e OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-e MODEL_NAME="taiji/gemini-3-pro-image-preview" \
-v $(pwd)/outputs:/app/outputs \
video-generator-agent:latest
```
### 3. 测试
```bash
# 运行测试脚本
python test_video_agent.py
# 或指定自定义 API URL
API_BASE_URL=http://localhost:8000 python test_video_agent.py
```
## API 使用示例
### 生成单张图片
```bash
curl -X POST http://localhost:8000/api/v1/generate-image \
-H "api-key: sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-H "Content-Type: application/json" \
-d '{
"description": "a beautiful sunset over mountains",
"size": "1024x1024",
"quality": "standard"
}'
```
响应示例:
```json
{
"success": true,
"file_path": "/app/outputs/images/image_20260226_123456.png",
"filename": "image_20260226_123456.png",
"url": "/api/v1/files/image_20260226_123456.png",
"description": "a beautiful sunset over mountains"
}
```
### 生成视频
```bash
curl -X POST http://localhost:8000/api/v1/generate-video \
-H "api-key: sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-H "Content-Type: application/json" \
-d '{
"descriptions": [
"sunrise over mountains",
"a peaceful lake at noon",
"starry night sky"
],
"duration_per_image": 3,
"fps": 30,
"transition": "fade"
}'
```
响应示例:
```json
{
"success": true,
"video": {
"file_path": "/app/outputs/videos/video_20260226_123456.mp4",
"url": "/api/v1/files/video_20260226_123456.mp4",
"filename": "video_20260226_123456.mp4",
"duration": 9.0,
"image_count": 3
},
"images": [
{
"index": 1,
"description": "sunrise over mountains",
"file_path": "/app/outputs/images/image_20260226_123456_1.png",
"url": "/api/v1/files/image_20260226_123456_1.png"
}
],
"settings": {
"duration_per_image": 3,
"fps": 30,
"transition": "fade"
}
}
```
### 下载文件
```bash
# 下载图片
curl -O http://localhost:8000/api/v1/files/image_20260226_123456.png
# 下载视频
curl -O http://localhost:8000/api/v1/files/video_20260226_123456.mp4
```
### 列出文件
```bash
# 列出所有文件
curl http://localhost:8000/api/v1/list-files?file_type=all
# 只列出图片
curl http://localhost:8000/api/v1/list-files?file_type=image
# 只列出视频
curl http://localhost:8000/api/v1/list-files?file_type=video
```
## MCP 工具
Agent 提供以下 MCP 工具:
### 1. generate_image
根据描述生成单张图片。
```json
{
"name": "generate_image",
"arguments": {
"description": "a futuristic city at night",
"size": "1024x1024",
"quality": "standard"
}
}
```
### 2. generate_video
根据多个描述生成视频。
```json
{
"name": "generate_video",
"arguments": {
"descriptions": [
"scene 1 description",
"scene 2 description",
"scene 3 description"
],
"duration_per_image": 3,
"fps": 30,
"transition": "fade"
}
}
```
### 3. list_generated_files
列出已生成的文件。
```json
{
"name": "list_generated_files",
"arguments": {
"file_type": "all"
}
}
```
### 4. cleanup_old_files
清理旧文件。
```json
{
"name": "cleanup_old_files",
"arguments": {
"max_age_hours": 24
}
}
```
## MCP 端点
### HTTP 端点
```bash
# 初始化
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"id": 1
}'
# 列出工具
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "tools/list",
"id": 2
}'
# 调用工具
curl -X POST http://localhost:8000/mcp \
-H "api-key: sk-i9AwAgXDqqxsA9Ym4AjSPg" \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "generate_image",
"arguments": {
"description": "a beautiful landscape"
}
},
"id": 3
}'
```
### SSE 端点
```bash
# 连接 SSE
curl -N http://localhost:8000/mcp/sse
```
## 环境变量
| 变量 | 必需 | 默认值 | 说明 |
|------|------|--------|------|
| `OPENAI_API_KEY` | 是 | `sk-i9AwAgXDqqxsA9Ym4AjSPg` | API Key |
| `OPENAI_BASE_URL` | 否 | LiteLLM Gateway URL | API Base URL |
| `MODEL_NAME` | 否 | `taiji/gemini-3-pro-image-preview` | 模型名称 |
| `API_PORT` | 否 | `8000` | 服务端口 |
| `OUTPUT_DIR` | 否 | `/app/outputs` | 输出目录 |
## 转场效果
支持的转场效果:
- `fade` - 淡入淡出
- `wipeleft` - 左擦除
- `wiperight` - 右擦除
- `slideup` - 上滑
- `slidedown` - 下滑
## 图片尺寸
支持的图片尺寸:
- `256x256`
- `512x512`
- `1024x1024` (默认)
- `1792x1024`
- `1024x1792`
## 项目结构
```
video_generator_agent/
├── Dockerfile # Docker 配置
├── requirements.txt # Python 依赖
├── run_api_server.py # 启动脚本
├── test_video_agent.py # 测试脚本
├── README.md # 文档
└── src/
├── __init__.py
├── utils/ # 工具模块
│ ├── __init__.py
│ ├── image_generator.py # 图片生成
│ ├── video_processor.py # 视频处理
│ └── file_manager.py # 文件管理
└── server/ # 服务器模块
├── __init__.py
├── mcp_server.py # MCP 工具定义
└── api_server.py # FastAPI 服务器
```
## 技术栈
- **Pydantic AI**: Agent 框架
- **FastMCP**: MCP 服务器
- **FastAPI**: REST API 框架
- **FFmpeg**: 视频处理
- **Gemini 3 Pro Image Preview**: 图片生成模型
- **aiohttp**: 异步 HTTP 客户端
- **Pillow**: 图片处理
## 故障排除
### FFmpeg 未安装
如果遇到 FFmpeg 相关错误,请确保已安装 FFmpeg:
```bash
# Ubuntu/Debian
sudo apt-get install ffmpeg
# macOS
brew install ffmpeg
# 验证安装
ffmpeg -version
```
### API Key 错误
确保设置了正确的 API Key:
```bash
export OPENAI_API_KEY="sk-i9AwAgXDqqxsA9Ym4AjSPg"
```
### 端口被占用
如果端口 8000 被占用,可以更改端口:
```bash
export API_PORT=8080
python run_api_server.py
```
## 性能建议
- 图片生成通常需要 10-30 秒
- 视频拼接时间取决于图片数量(每张图片约 2-5 秒)
- 建议每个视频不超过 10 个场景
- 使用转场效果会增加处理时间
## 许可证
MIT License
## 作者
Video Generator Agent Team
@@ -0,0 +1,19 @@
# Pydantic AI
pydantic-ai>=0.0.14
# MCP
mcp>=0.9.0
fastmcp>=0.1.0
# FastAPI
fastapi>=0.109.0
uvicorn[standard]>=0.27.0
# HTTP Client
aiohttp>=3.9.0
# Image Processing
Pillow>=10.0.0
# OpenAI (for image generation)
openai>=1.0.0
@@ -0,0 +1,20 @@
#!/usr/bin/env python
"""启动 Video Generator Agent API 服务器"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
if __name__ == '__main__':
from src.server.api_server import app
import uvicorn
import os
host = os.getenv('API_HOST', '0.0.0.0')
port = int(os.getenv('API_PORT', '8000'))
print(f"🚀 启动 Video Generator Agent API: http://{host}:{port}")
print(f"📖 API 文档: http://{host}:{port}/docs")
print(f"🔧 MCP 端点: http://{host}:{port}/mcp")
uvicorn.run(app, host=host, port=port, log_level="info")
@@ -0,0 +1,2 @@
"""Video Generator Agent - 视频生成 Agent"""
__version__ = "1.0.0"
@@ -0,0 +1 @@
"""Server modules"""
@@ -0,0 +1,345 @@
"""
HTTP API 服务器 - 视频生成 Agent
提供 REST API 和 MCP HTTP/SSE 端点。
"""
import json
import uuid
import os
import sys
from typing import Optional, Dict, Any, AsyncGenerator, List
from contextlib import asynccontextmanager
from pathlib import Path
# 添加父目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from fastapi import FastAPI, HTTPException, Request, Header, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse, JSONResponse, FileResponse
from pydantic import BaseModel, Field
from src.server.mcp_server import TOOL_MAP, TOOL_LIST
from src.utils.file_manager import FileManager
# ==================== 配置 ====================
SERVER_NAME = "Video Generator Agent"
OUTPUT_DIR = os.getenv('OUTPUT_DIR', '/app/outputs')
file_manager = FileManager(base_dir=OUTPUT_DIR)
# ==================== FastAPI 应用 ====================
@asynccontextmanager
async def lifespan(app: FastAPI):
print(f"🚀 {SERVER_NAME} 启动")
print(f"📁 输出目录: {OUTPUT_DIR}")
yield
print(f"🛑 {SERVER_NAME} 关闭")
app = FastAPI(
title=SERVER_NAME,
description="根据描述生成图片并拼接为视频",
version="1.0.0",
lifespan=lifespan
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==================== API Key 验证 ====================
async def verify_api_key(
api_key: Optional[str] = Header(None, alias="api-key"),
authorization: Optional[str] = Header(None)
) -> str:
"""验证 API Key"""
if api_key and api_key.strip() and api_key.strip() != "sk":
return api_key.strip()
if authorization:
key = authorization[7:].strip() if authorization.startswith("Bearer ") else authorization.strip()
if key and key != "sk":
return key
raise HTTPException(status_code=401, detail="缺少 API Key")
def get_api_key_from_request(request: Request) -> Optional[str]:
"""从请求头提取 API Key(不验证)"""
api_key = request.headers.get("api-key") or request.headers.get("api_key")
if not api_key:
auth = request.headers.get("Authorization")
if auth:
api_key = auth[7:] if auth.startswith("Bearer ") else auth
return api_key
# ==================== 健康检查 ====================
@app.get("/")
async def root():
storage_info = file_manager.get_storage_info()
return {
"service": SERVER_NAME,
"status": "running",
"version": "1.0.0",
"tools": list(TOOL_MAP.keys()),
"storage": storage_info
}
@app.get("/health")
async def health():
return {"status": "healthy", "service": SERVER_NAME}
# ==================== MCP 端点 ====================
sessions: Dict[str, Dict] = {}
async def handle_mcp_request(data: Dict, session_id: str = None, api_key: str = None) -> Dict:
"""处理 MCP JSON-RPC 请求"""
method = data.get("method")
params = data.get("params", {})
req_id = data.get("id")
# tools/call 需要验证 API Key
if method == "tools/call" and (not api_key or api_key == "sk"):
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32001, "message": "缺少 API Key"}}
try:
if method == "initialize":
session_id = session_id or str(uuid.uuid4())
sessions[session_id] = {"initialized": True}
return {
"jsonrpc": "2.0", "id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": SERVER_NAME, "version": "1.0.0"}
}
}
elif method == "tools/list":
return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOL_LIST}}
elif method == "tools/call":
tool_name = params.get("name")
args = params.get("arguments", {})
if tool_name not in TOOL_MAP:
raise ValueError(f"Unknown tool: {tool_name}")
# 设置 API Key 到环境变量
old_key = os.environ.get('OPENAI_API_KEY')
if api_key:
os.environ['OPENAI_API_KEY'] = api_key
try:
result = await TOOL_MAP[tool_name](**args)
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
return {
"jsonrpc": "2.0", "id": req_id,
"result": {"content": [{"type": "text", "text": str(result)}]}
}
elif method == "ping":
return {"jsonrpc": "2.0", "id": req_id, "result": {}}
else:
raise ValueError(f"Unknown method: {method}")
except Exception as e:
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(e)}}
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""MCP HTTP 端点"""
try:
body = await request.json()
session_id = request.headers.get("x-mcp-session-id")
api_key = get_api_key_from_request(request)
response = await handle_mcp_request(body, session_id, api_key)
return JSONResponse(content=response, headers={"x-mcp-session-id": session_id or ""})
except Exception as e:
return JSONResponse(status_code=400, content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}})
@app.get("/mcp/sse")
async def mcp_sse(request: Request):
"""MCP SSE 端点"""
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
async def stream() -> AsyncGenerator[str, None]:
yield f"data: {json.dumps({'type': 'connection', 'sessionId': session_id})}\n\n"
import asyncio
while True:
await asyncio.sleep(30)
yield f"data: {json.dumps({'type': 'ping'})}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id})
@app.post("/mcp/sse")
async def mcp_sse_post(request: Request):
"""MCP SSE POST 端点"""
try:
body = await request.json()
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
api_key = get_api_key_from_request(request)
async def stream() -> AsyncGenerator[str, None]:
response = await handle_mcp_request(body, session_id, api_key)
yield f"data: {json.dumps(response)}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "x-mcp-session-id": session_id})
except Exception as e:
return JSONResponse(status_code=400, content={"jsonrpc": "2.0", "error": {"code": -32700, "message": str(e)}})
# ==================== 业务 API ====================
class ImageGenerationRequest(BaseModel):
"""图片生成请求"""
description: str = Field(..., description="图片描述")
size: str = Field("1024x1024", description="图片尺寸")
quality: str = Field("standard", description="图片质量")
class VideoGenerationRequest(BaseModel):
"""视频生成请求"""
descriptions: List[str] = Field(..., description="场景描述列表")
duration_per_image: int = Field(3, description="每张图片显示秒数")
fps: int = Field(30, description="视频帧率")
transition: Optional[str] = Field(None, description="转场效果")
@app.post("/api/v1/generate-image")
async def api_generate_image(request: ImageGenerationRequest, api_key: str = Depends(verify_api_key)):
"""生成单张图片"""
try:
# 设置 API Key
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result_str = await TOOL_MAP['generate_image'](
description=request.description,
size=request.size,
quality=request.quality
)
result = json.loads(result_str)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Unknown error"))
return result
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/generate-video")
async def api_generate_video(request: VideoGenerationRequest, api_key: str = Depends(verify_api_key)):
"""生成视频"""
try:
# 设置 API Key
old_key = os.environ.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = api_key
try:
result_str = await TOOL_MAP['generate_video'](
descriptions=request.descriptions,
duration_per_image=request.duration_per_image,
fps=request.fps,
transition=request.transition
)
result = json.loads(result_str)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Unknown error"))
return result
finally:
if old_key:
os.environ['OPENAI_API_KEY'] = old_key
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/list-files")
async def api_list_files(file_type: str = "all"):
"""列出文件"""
try:
result_str = await TOOL_MAP['list_generated_files'](file_type=file_type)
result = json.loads(result_str)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/files/{filename}")
async def api_download_file(filename: str):
"""下载文件"""
try:
file_path = file_manager.get_file_path(filename)
if not file_path or not os.path.exists(file_path):
raise HTTPException(status_code=404, detail="File not found")
# 确定 media type
if filename.endswith('.mp4'):
media_type = "video/mp4"
elif filename.endswith('.png'):
media_type = "image/png"
elif filename.endswith('.jpg') or filename.endswith('.jpeg'):
media_type = "image/jpeg"
else:
media_type = "application/octet-stream"
return FileResponse(
file_path,
media_type=media_type,
filename=filename
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/cleanup")
async def api_cleanup(max_age_hours: int = 24):
"""清理旧文件"""
try:
result_str = await TOOL_MAP['cleanup_old_files'](max_age_hours=max_age_hours)
result = json.loads(result_str)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
@@ -0,0 +1,308 @@
"""
MCP 服务器 - 视频生成 Agent 工具
"""
import json
import os
import sys
from typing import Optional, List
from pathlib import Path
# 添加父目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
from src.utils.image_generator import ImageGenerator
from src.utils.video_processor import VideoProcessor
from src.utils.file_manager import FileManager
# ==================== 配置 ====================
# LiteLLM Gateway 配置
_BASE_URL = os.getenv('OPENAI_BASE_URL',
os.getenv('LLM_BASE_URL', 'https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1'))
_API_KEY = os.getenv('OPENAI_API_KEY', 'sk-i9AwAgXDqqxsA9Ym4AjSPg')
os.environ.setdefault('OPENAI_API_KEY', _API_KEY)
os.environ.setdefault('OPENAI_BASE_URL', _BASE_URL)
# 模型名称
def _get_model_name() -> str:
model = os.getenv('MODEL_NAME', os.getenv('LITELLM_MODEL', 'taiji/gemini-3-pro-image-preview'))
return model if ':' in model else f'openai:{model}'
MODEL_NAME = _get_model_name()
# 输出目录
OUTPUT_DIR = os.getenv('OUTPUT_DIR', '/app/outputs')
# ==================== 初始化工具 ====================
image_generator = ImageGenerator(api_key=_API_KEY, base_url=_BASE_URL, model='taiji/gemini-3-pro-image-preview')
video_processor = VideoProcessor()
file_manager = FileManager(base_dir=OUTPUT_DIR)
# ==================== MCP 服务器 ====================
server = FastMCP('Video Generator Agent')
SYSTEM_PROMPT = '''你是一个专业的视频生成 AI 助手。
你可以根据用户的描述生成图片,并将多张图片拼接成视频。
请根据用户的需求提供帮助。'''
def get_agent() -> Agent:
"""创建 Agent 实例"""
return Agent(MODEL_NAME, system_prompt=SYSTEM_PROMPT)
# ==================== MCP 工具定义 ====================
@server.tool()
async def generate_image(
description: str,
size: str = "1024x1024",
quality: str = "standard"
) -> str:
"""
根据描述生成单张图片
Args:
description: 图片描述(英文效果更好)
size: 图片尺寸,可选 256x256, 512x512, 1024x1024, 1792x1024, 1024x1792
quality: 图片质量,可选 standard, hd
Returns:
JSON 格式的结果,包含图片路径和下载 URL
"""
try:
result = await image_generator.generate_image(
description=description,
output_dir=os.path.join(OUTPUT_DIR, "images"),
size=size,
quality=quality
)
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
}, ensure_ascii=False)
@server.tool()
async def generate_video(
descriptions: List[str],
duration_per_image: int = 3,
fps: int = 30,
transition: Optional[str] = None
) -> str:
"""
根据多个描述生成视频
Args:
descriptions: 场景描述列表(每个描述对应一张图片)
duration_per_image: 每张图片显示秒数(默认 3 秒)
fps: 视频帧率(默认 30)
transition: 转场效果,可选 fade, wipeleft, wiperight, slideup, slidedown
Returns:
JSON 格式的结果,包含视频路径和下载 URL
"""
try:
if not descriptions or len(descriptions) == 0:
return json.dumps({
"success": False,
"error": "至少需要提供一个场景描述"
}, ensure_ascii=False)
# 1. 生成所有图片
image_paths = []
generated_images = []
for i, desc in enumerate(descriptions):
print(f"生成图片 {i+1}/{len(descriptions)}: {desc[:50]}...")
result = await image_generator.generate_image(
description=desc,
output_dir=os.path.join(OUTPUT_DIR, "images")
)
if not result.get("success"):
return json.dumps({
"success": False,
"error": f"生成第 {i+1} 张图片失败: {result.get('error')}",
"generated_images": generated_images
}, ensure_ascii=False)
image_paths.append(result["file_path"])
generated_images.append({
"index": i + 1,
"description": desc,
"file_path": result["file_path"],
"url": result["url"]
})
# 2. 拼接视频
print(f"拼接视频: {len(image_paths)} 张图片...")
video_result = await video_processor.create_video_from_images(
image_paths=image_paths,
output_dir=os.path.join(OUTPUT_DIR, "videos"),
duration_per_image=duration_per_image,
fps=fps,
transition=transition
)
if not video_result.get("success"):
return json.dumps({
"success": False,
"error": f"视频拼接失败: {video_result.get('error')}",
"generated_images": generated_images
}, ensure_ascii=False)
# 3. 返回结果
return json.dumps({
"success": True,
"video": {
"file_path": video_result["file_path"],
"url": video_result["url"],
"filename": video_result["filename"],
"duration": video_result["duration"],
"image_count": video_result["image_count"]
},
"images": generated_images,
"settings": {
"duration_per_image": duration_per_image,
"fps": fps,
"transition": transition
}
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
}, ensure_ascii=False)
@server.tool()
async def list_generated_files(file_type: str = "all") -> str:
"""
列出已生成的文件
Args:
file_type: 文件类型,可选 all, image, video
Returns:
JSON 格式的文件列表
"""
try:
files = file_manager.list_files(file_type=file_type)
storage_info = file_manager.get_storage_info()
return json.dumps({
"success": True,
"files": files,
"storage": storage_info,
"total_count": len(files)
}, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
}, ensure_ascii=False)
@server.tool()
async def cleanup_old_files(max_age_hours: int = 24) -> str:
"""
清理旧文件
Args:
max_age_hours: 保留最近多少小时的文件(默认 24 小时)
Returns:
JSON 格式的清理结果
"""
try:
result = file_manager.cleanup_old_files(max_age_hours=max_age_hours)
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
}, ensure_ascii=False)
# ==================== 工具映射(供 API 使用)====================
TOOL_MAP = {
'generate_image': generate_image,
'generate_video': generate_video,
'list_generated_files': list_generated_files,
'cleanup_old_files': cleanup_old_files,
}
TOOL_LIST = [
{
"name": "generate_image",
"description": "根据描述生成单张图片",
"inputSchema": {
"type": "object",
"properties": {
"description": {"type": "string", "description": "图片描述"},
"size": {"type": "string", "description": "图片尺寸", "default": "1024x1024"},
"quality": {"type": "string", "description": "图片质量", "default": "standard"}
},
"required": ["description"]
}
},
{
"name": "generate_video",
"description": "根据多个描述生成视频",
"inputSchema": {
"type": "object",
"properties": {
"descriptions": {
"type": "array",
"items": {"type": "string"},
"description": "场景描述列表"
},
"duration_per_image": {"type": "integer", "description": "每张图片显示秒数", "default": 3},
"fps": {"type": "integer", "description": "视频帧率", "default": 30},
"transition": {"type": "string", "description": "转场效果"}
},
"required": ["descriptions"]
}
},
{
"name": "list_generated_files",
"description": "列出已生成的文件",
"inputSchema": {
"type": "object",
"properties": {
"file_type": {"type": "string", "description": "文件类型 (all/image/video)", "default": "all"}
}
}
},
{
"name": "cleanup_old_files",
"description": "清理旧文件",
"inputSchema": {
"type": "object",
"properties": {
"max_age_hours": {"type": "integer", "description": "保留最近多少小时的文件", "default": 24}
}
}
}
]
if __name__ == '__main__':
server.run()
@@ -0,0 +1,6 @@
"""Utility modules for video generation"""
from .image_generator import ImageGenerator
from .video_processor import VideoProcessor
from .file_manager import FileManager
__all__ = ['ImageGenerator', 'VideoProcessor', 'FileManager']
@@ -0,0 +1,218 @@
"""
文件管理工具
管理生成的图片和视频文件
"""
import os
import logging
from pathlib import Path
from typing import List, Dict, Optional
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
class FileManager:
"""文件管理器"""
def __init__(self, base_dir: str = "/app/outputs"):
"""
初始化文件管理器
Args:
base_dir: 基础目录
"""
self.base_dir = base_dir
self.images_dir = os.path.join(base_dir, "images")
self.videos_dir = os.path.join(base_dir, "videos")
# 创建目录
Path(self.images_dir).mkdir(parents=True, exist_ok=True)
Path(self.videos_dir).mkdir(parents=True, exist_ok=True)
logger.info(f"FileManager initialized: base_dir={base_dir}")
def get_file_path(self, filename: str) -> Optional[str]:
"""
获取文件完整路径
Args:
filename: 文件名
Returns:
str: 文件路径,如果不存在返回 None
"""
# 先在 images 目录查找
image_path = os.path.join(self.images_dir, filename)
if os.path.exists(image_path):
return image_path
# 再在 videos 目录查找
video_path = os.path.join(self.videos_dir, filename)
if os.path.exists(video_path):
return video_path
return None
def list_files(self, file_type: str = "all") -> List[Dict]:
"""
列出文件
Args:
file_type: 文件类型 (all, image, video)
Returns:
List[Dict]: 文件列表
"""
files = []
try:
if file_type in ["all", "image"]:
files.extend(self._list_directory(self.images_dir, "image"))
if file_type in ["all", "video"]:
files.extend(self._list_directory(self.videos_dir, "video"))
# 按修改时间排序(最新的在前)
files.sort(key=lambda x: x["modified_time"], reverse=True)
except Exception as e:
logger.error(f"Error listing files: {e}")
return files
def _list_directory(self, directory: str, file_type: str) -> List[Dict]:
"""
列出目录中的文件
Args:
directory: 目录路径
file_type: 文件类型标签
Returns:
List[Dict]: 文件信息列表
"""
files = []
try:
if not os.path.exists(directory):
return files
for filename in os.listdir(directory):
file_path = os.path.join(directory, filename)
if not os.path.isfile(file_path):
continue
stat = os.stat(file_path)
files.append({
"filename": filename,
"type": file_type,
"size": stat.st_size,
"size_mb": round(stat.st_size / 1024 / 1024, 2),
"modified_time": stat.st_mtime,
"modified_date": datetime.fromtimestamp(stat.st_mtime).isoformat(),
"url": f"/api/v1/files/{filename}"
})
except Exception as e:
logger.error(f"Error listing directory {directory}: {e}")
return files
def cleanup_old_files(self, max_age_hours: int = 24) -> Dict:
"""
清理旧文件
Args:
max_age_hours: 最大保留时间(小时)
Returns:
Dict: 清理结果
"""
try:
cutoff_time = datetime.now() - timedelta(hours=max_age_hours)
cutoff_timestamp = cutoff_time.timestamp()
deleted_count = 0
freed_space = 0
# 清理图片
for filename in os.listdir(self.images_dir):
file_path = os.path.join(self.images_dir, filename)
if os.path.isfile(file_path):
stat = os.stat(file_path)
if stat.st_mtime < cutoff_timestamp:
freed_space += stat.st_size
os.remove(file_path)
deleted_count += 1
logger.info(f"Deleted old image: {filename}")
# 清理视频
for filename in os.listdir(self.videos_dir):
file_path = os.path.join(self.videos_dir, filename)
if os.path.isfile(file_path):
stat = os.stat(file_path)
if stat.st_mtime < cutoff_timestamp:
freed_space += stat.st_size
os.remove(file_path)
deleted_count += 1
logger.info(f"Deleted old video: {filename}")
return {
"success": True,
"deleted_count": deleted_count,
"freed_space_mb": round(freed_space / 1024 / 1024, 2)
}
except Exception as e:
logger.error(f"Error cleaning up files: {e}")
return {
"success": False,
"error": str(e)
}
def get_storage_info(self) -> Dict:
"""
获取存储信息
Returns:
Dict: 存储信息
"""
try:
total_size = 0
image_count = 0
video_count = 0
# 统计图片
if os.path.exists(self.images_dir):
for filename in os.listdir(self.images_dir):
file_path = os.path.join(self.images_dir, filename)
if os.path.isfile(file_path):
total_size += os.path.getsize(file_path)
image_count += 1
# 统计视频
if os.path.exists(self.videos_dir):
for filename in os.listdir(self.videos_dir):
file_path = os.path.join(self.videos_dir, filename)
if os.path.isfile(file_path):
total_size += os.path.getsize(file_path)
video_count += 1
return {
"total_size_mb": round(total_size / 1024 / 1024, 2),
"image_count": image_count,
"video_count": video_count,
"total_files": image_count + video_count
}
except Exception as e:
logger.error(f"Error getting storage info: {e}")
return {
"total_size_mb": 0,
"image_count": 0,
"video_count": 0,
"total_files": 0,
"error": str(e)
}
@@ -0,0 +1,197 @@
"""
图片生成工具
使用 OpenAI 兼容接口调用 Gemini 3 Pro Image Preview
"""
import os
import aiohttp
import logging
from pathlib import Path
from typing import Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class ImageGenerator:
"""图片生成器"""
def __init__(self, api_key: str = None, base_url: str = None, model: str = None):
"""
初始化图片生成器
Args:
api_key: API Key
base_url: API Base URL
model: 模型名称
"""
self.api_key = api_key or os.getenv('OPENAI_API_KEY', 'sk')
self.base_url = base_url or os.getenv('OPENAI_BASE_URL',
'https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1')
self.model = model or os.getenv('MODEL_NAME', 'taiji/gemini-3-pro-image-preview')
# 确保 base_url 不以 /v1 结尾(我们会手动添加)
if self.base_url.endswith('/v1'):
self.base_url = self.base_url[:-3]
logger.info(f"ImageGenerator initialized: model={self.model}, base_url={self.base_url}")
async def generate_image(
self,
description: str,
output_dir: str = "/app/outputs/images",
size: str = "1024x1024",
quality: str = "standard"
) -> dict:
"""
生成图片
Args:
description: 图片描述
output_dir: 输出目录
size: 图片尺寸
quality: 图片质量
Returns:
dict: {
"success": bool,
"file_path": str,
"url": str,
"description": str
}
"""
try:
# 创建输出目录
Path(output_dir).mkdir(parents=True, exist_ok=True)
# 调用 API 生成图片
logger.info(f"Generating image: {description[:50]}...")
async with aiohttp.ClientSession() as session:
# 使用 OpenAI 兼容的图片生成接口
url = f"{self.base_url}/v1/images/generations"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"prompt": description,
"n": 1,
"size": size,
"quality": quality
}
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=60)) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"API error: {response.status} - {error_text}")
return {
"success": False,
"error": f"API error: {response.status} - {error_text}"
}
data = await response.json()
# 获取图片 URL
if not data.get("data") or len(data["data"]) == 0:
return {
"success": False,
"error": "No image data returned from API"
}
image_url = data["data"][0].get("url")
if not image_url:
return {
"success": False,
"error": "No image URL in response"
}
# 下载图片
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
filename = f"image_{timestamp}.png"
file_path = os.path.join(output_dir, filename)
download_result = await self._download_image(image_url, file_path)
if not download_result["success"]:
return download_result
logger.info(f"Image generated successfully: {file_path}")
return {
"success": True,
"file_path": file_path,
"filename": filename,
"url": f"/api/v1/files/{filename}",
"description": description
}
except Exception as e:
logger.error(f"Error generating image: {e}", exc_info=True)
return {
"success": False,
"error": str(e)
}
async def _download_image(self, url: str, save_path: str) -> dict:
"""
下载图片到本地
Args:
url: 图片 URL
save_path: 保存路径
Returns:
dict: {"success": bool, "file_path": str, "error": str}
"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
if response.status != 200:
return {
"success": False,
"error": f"Failed to download image: HTTP {response.status}"
}
content = await response.read()
# 保存文件
with open(save_path, 'wb') as f:
f.write(content)
# 验证文件
if not os.path.exists(save_path) or os.path.getsize(save_path) == 0:
return {
"success": False,
"error": "Downloaded file is empty or not saved"
}
return {
"success": True,
"file_path": save_path
}
except Exception as e:
logger.error(f"Error downloading image: {e}")
return {
"success": False,
"error": str(e)
}
def validate_image(self, file_path: str) -> bool:
"""
验证图片文件是否有效
Args:
file_path: 文件路径
Returns:
bool: 是否有效
"""
try:
from PIL import Image
with Image.open(file_path) as img:
img.verify()
return True
except Exception as e:
logger.error(f"Image validation failed: {e}")
return False
@@ -0,0 +1,327 @@
"""
视频处理工具
使用 FFmpeg 将图片拼接成视频
"""
import os
import subprocess
import logging
from pathlib import Path
from typing import List, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class VideoProcessor:
"""视频处理器"""
def __init__(self):
"""初始化视频处理器"""
# 检查 FFmpeg 是否可用
try:
result = subprocess.run(['ffmpeg', '-version'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0:
logger.info("FFmpeg is available")
else:
logger.warning("FFmpeg check returned non-zero exit code")
except Exception as e:
logger.error(f"FFmpeg not available: {e}")
async def create_video_from_images(
self,
image_paths: List[str],
output_dir: str = "/app/outputs/videos",
duration_per_image: int = 3,
fps: int = 30,
transition: Optional[str] = None
) -> dict:
"""
从图片创建视频
Args:
image_paths: 图片路径列表
output_dir: 输出目录
duration_per_image: 每张图片显示秒数
fps: 帧率
transition: 转场效果 (fade, wipeleft, wiperight, slideup, slidedown)
Returns:
dict: {
"success": bool,
"file_path": str,
"url": str,
"duration": float,
"image_count": int
}
"""
try:
# 验证输入
if not image_paths or len(image_paths) == 0:
return {
"success": False,
"error": "No images provided"
}
# 验证所有图片文件存在
for img_path in image_paths:
if not os.path.exists(img_path):
return {
"success": False,
"error": f"Image not found: {img_path}"
}
# 创建输出目录
Path(output_dir).mkdir(parents=True, exist_ok=True)
# 生成输出文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_filename = f"video_{timestamp}.mp4"
output_path = os.path.join(output_dir, output_filename)
logger.info(f"Creating video from {len(image_paths)} images...")
# 根据是否需要转场效果选择不同的方法
if transition and len(image_paths) > 1:
result = await self._create_video_with_transitions(
image_paths, output_path, duration_per_image, fps, transition
)
else:
result = await self._create_simple_video(
image_paths, output_path, duration_per_image, fps
)
if not result["success"]:
return result
# 获取视频信息
video_info = self.get_video_info(output_path)
return {
"success": True,
"file_path": output_path,
"filename": output_filename,
"url": f"/api/v1/files/{output_filename}",
"duration": video_info.get("duration", 0),
"image_count": len(image_paths)
}
except Exception as e:
logger.error(f"Error creating video: {e}", exc_info=True)
return {
"success": False,
"error": str(e)
}
async def _create_simple_video(
self,
image_paths: List[str],
output_path: str,
duration_per_image: int,
fps: int
) -> dict:
"""
创建简单视频(无转场效果)
使用 concat demuxer 方法
"""
try:
# 创建临时文件列表
temp_dir = os.path.dirname(output_path)
concat_file = os.path.join(temp_dir, f"concat_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt")
# 写入文件列表
with open(concat_file, 'w') as f:
for img_path in image_paths:
# FFmpeg concat 格式
f.write(f"file '{img_path}'\n")
f.write(f"duration {duration_per_image}\n")
# 最后一张图片需要再写一次(FFmpeg 要求)
f.write(f"file '{image_paths[-1]}'\n")
# 构建 FFmpeg 命令
cmd = [
'ffmpeg',
'-f', 'concat',
'-safe', '0',
'-i', concat_file,
'-vsync', 'vfr',
'-pix_fmt', 'yuv420p',
'-c:v', 'libx264',
'-r', str(fps),
'-y', # 覆盖输出文件
output_path
]
logger.info(f"Running FFmpeg command: {' '.join(cmd)}")
# 执行命令
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300 # 5分钟超时
)
# 清理临时文件
try:
os.remove(concat_file)
except:
pass
if result.returncode != 0:
logger.error(f"FFmpeg error: {result.stderr}")
return {
"success": False,
"error": f"FFmpeg failed: {result.stderr[:500]}"
}
# 验证输出文件
if not os.path.exists(output_path) or os.path.getsize(output_path) == 0:
return {
"success": False,
"error": "Output video file is empty or not created"
}
logger.info(f"Video created successfully: {output_path}")
return {"success": True}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": "FFmpeg command timed out"
}
except Exception as e:
logger.error(f"Error in _create_simple_video: {e}")
return {
"success": False,
"error": str(e)
}
async def _create_video_with_transitions(
self,
image_paths: List[str],
output_path: str,
duration_per_image: int,
fps: int,
transition: str
) -> dict:
"""
创建带转场效果的视频
使用 xfade filter
"""
try:
# 转场持续时间(秒)
transition_duration = 1
# 构建 filter_complex
# 每张图片显示时间 = duration_per_image
# 转场开始时间 = duration_per_image - transition_duration
inputs = []
filter_parts = []
# 添加所有输入
for i, img_path in enumerate(image_paths):
inputs.extend(['-loop', '1', '-t', str(duration_per_image), '-i', img_path])
# 构建 xfade filter chain
if len(image_paths) == 2:
# 两张图片的简单情况
offset = duration_per_image - transition_duration
filter_complex = f"[0][1]xfade=transition={transition}:duration={transition_duration}:offset={offset}[v]"
else:
# 多张图片需要链式 xfade
current_label = "0"
for i in range(1, len(image_paths)):
offset = i * duration_per_image - i * transition_duration
next_label = f"v{i}" if i < len(image_paths) - 1 else "v"
filter_parts.append(f"[{current_label}][{i}]xfade=transition={transition}:duration={transition_duration}:offset={offset}[{next_label}]")
current_label = next_label
filter_complex = ";".join(filter_parts)
# 构建完整命令
cmd = ['ffmpeg'] + inputs + [
'-filter_complex', filter_complex,
'-map', '[v]',
'-c:v', 'libx264',
'-pix_fmt', 'yuv420p',
'-r', str(fps),
'-y',
output_path
]
logger.info(f"Running FFmpeg with transitions: {' '.join(cmd[:20])}...")
# 执行命令
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.returncode != 0:
logger.error(f"FFmpeg error: {result.stderr}")
# 如果转场失败,回退到简单模式
logger.warning("Falling back to simple video creation")
return await self._create_simple_video(image_paths, output_path, duration_per_image, fps)
# 验证输出文件
if not os.path.exists(output_path) or os.path.getsize(output_path) == 0:
return {
"success": False,
"error": "Output video file is empty or not created"
}
logger.info(f"Video with transitions created successfully: {output_path}")
return {"success": True}
except Exception as e:
logger.error(f"Error in _create_video_with_transitions: {e}")
# 回退到简单模式
return await self._create_simple_video(image_paths, output_path, duration_per_image, fps)
def get_video_info(self, video_path: str) -> dict:
"""
获取视频信息
Args:
video_path: 视频路径
Returns:
dict: 视频信息
"""
try:
cmd = [
'ffprobe',
'-v', 'quiet',
'-print_format', 'json',
'-show_format',
'-show_streams',
video_path
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if result.returncode == 0:
import json
data = json.loads(result.stdout)
duration = float(data.get('format', {}).get('duration', 0))
size = int(data.get('format', {}).get('size', 0))
return {
"duration": duration,
"size": size,
"format": data.get('format', {}).get('format_name', 'unknown')
}
except Exception as e:
logger.error(f"Error getting video info: {e}")
return {
"duration": 0,
"size": 0,
"format": "unknown"
}
@@ -0,0 +1,134 @@
#!/usr/bin/env python
"""
快速测试脚本 - 验证核心功能
"""
import asyncio
import os
import sys
from pathlib import Path
# 添加路径
sys.path.insert(0, str(Path(__file__).parent))
# 设置环境变量
os.environ['OPENAI_API_KEY'] = 'sk-i9AwAgXDqqxsA9Ym4AjSPg'
os.environ['OPENAI_BASE_URL'] = 'https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1'
os.environ['OUTPUT_DIR'] = './outputs'
from src.utils.video_processor import VideoProcessor
from src.utils.file_manager import FileManager
async def test_video_processor():
"""测试视频处理器(使用模拟图片)"""
print("\n" + "="*60)
print("测试: 视频处理器(FFmpeg)")
print("="*60)
# 创建测试图片
from PIL import Image
import os
os.makedirs("./outputs/images", exist_ok=True)
image_paths = []
colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255)] # 红、绿、蓝
for i, color in enumerate(colors):
img = Image.new('RGB', (1024, 1024), color)
path = f"./outputs/images/test_image_{i}.png"
img.save(path)
image_paths.append(path)
print(f"✅ 创建测试图片: {path}")
# 测试视频拼接
processor = VideoProcessor()
result = await processor.create_video_from_images(
image_paths=image_paths,
output_dir="./outputs/videos",
duration_per_image=2,
fps=30,
transition="fade"
)
if result.get("success"):
print(f"\n✅ 视频生成成功!")
print(f" 文件: {result['file_path']}")
print(f" 时长: {result['duration']} 秒")
print(f" 图片数: {result['image_count']}")
return True
else:
print(f"\n❌ 视频生成失败: {result.get('error')}")
return False
async def test_file_manager():
"""测试文件管理器"""
print("\n" + "="*60)
print("测试: 文件管理器")
print("="*60)
manager = FileManager(base_dir="./outputs")
# 列出文件
files = manager.list_files("all")
print(f"\n📁 文件列表 (共 {len(files)} 个):")
for file in files:
print(f" - {file['filename']} ({file['type']}, {file['size_mb']} MB)")
# 存储信息
storage = manager.get_storage_info()
print(f"\n💾 存储信息:")
print(f" - 总大小: {storage['total_size_mb']} MB")
print(f" - 图片数: {storage['image_count']}")
print(f" - 视频数: {storage['video_count']}")
return True
async def main():
"""运行测试"""
print("\n" + "="*60)
print("🧪 Video Generator Agent - 核心功能测试")
print("="*60)
results = []
# 测试视频处理
results.append(("视频处理器", await test_video_processor()))
# 测试文件管理
results.append(("文件管理器", await test_file_manager()))
# 汇总
print("\n" + "="*60)
print("📊 测试结果")
print("="*60)
passed = sum(1 for _, result in results if result)
total = len(results)
for name, result in results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{status} - {name}")
print(f"\n总计: {passed}/{total} 通过")
if passed == total:
print("\n🎉 核心功能测试通过!")
print("\n📝 注意: 图片生成功能需要正确的模型配置")
print(" 当前模型 'taiji/gemini-3-pro-image-preview' 可能不支持")
print(" 建议使用: dall-e-3, dall-e-2 或其他支持的模型")
return 0
else:
return 1
if __name__ == "__main__":
try:
exit_code = asyncio.run(main())
sys.exit(exit_code)
except Exception as e:
print(f"\n❌ 测试失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
@@ -0,0 +1,252 @@
#!/usr/bin/env python
"""
Video Generator Agent 测试脚本
测试图片生成和视频拼接功能
"""
import asyncio
import aiohttp
import json
import sys
import os
# 配置
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000")
API_KEY = os.getenv("API_KEY", "sk-i9AwAgXDqqxsA9Ym4AjSPg")
async def test_health_check():
"""测试健康检查"""
print("\n" + "="*60)
print("测试 1: 健康检查")
print("="*60)
try:
async with aiohttp.ClientSession() as session:
async with session.get(f"{API_BASE_URL}/health") as response:
data = await response.json()
print(f"✅ 状态: {response.status}")
print(f"📊 响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
return response.status == 200
except Exception as e:
print(f"❌ 错误: {e}")
return False
async def test_generate_single_image():
"""测试生成单张图片"""
print("\n" + "="*60)
print("测试 2: 生成单张图片")
print("="*60)
try:
payload = {
"description": "a beautiful sunset over mountains with orange and purple sky",
"size": "1024x1024",
"quality": "standard"
}
headers = {
"api-key": API_KEY,
"Content-Type": "application/json"
}
print(f"📝 请求: {json.dumps(payload, indent=2, ensure_ascii=False)}")
async with aiohttp.ClientSession() as session:
async with session.post(
f"{API_BASE_URL}/api/v1/generate-image",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=120)
) as response:
data = await response.json()
print(f"✅ 状态: {response.status}")
print(f"📊 响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
if data.get("success"):
print(f"🖼️ 图片 URL: {API_BASE_URL}{data.get('url')}")
return True
else:
print(f"❌ 失败: {data.get('error')}")
return False
except Exception as e:
print(f"❌ 错误: {e}")
return False
async def test_generate_video():
"""测试生成视频"""
print("\n" + "="*60)
print("测试 3: 生成视频(3个场景)")
print("="*60)
try:
payload = {
"descriptions": [
"a peaceful morning sunrise over a calm lake with mist",
"a busy city street at noon with people and cars",
"a starry night sky with the milky way visible"
],
"duration_per_image": 3,
"fps": 30,
"transition": "fade"
}
headers = {
"api-key": API_KEY,
"Content-Type": "application/json"
}
print(f"📝 请求: {json.dumps(payload, indent=2, ensure_ascii=False)}")
print("⏳ 生成视频中,这可能需要 1-3 分钟...")
async with aiohttp.ClientSession() as session:
async with session.post(
f"{API_BASE_URL}/api/v1/generate-video",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=300)
) as response:
data = await response.json()
print(f"✅ 状态: {response.status}")
print(f"📊 响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
if data.get("success"):
video_info = data.get("video", {})
print(f"\n🎬 视频信息:")
print(f" - URL: {API_BASE_URL}{video_info.get('url')}")
print(f" - 时长: {video_info.get('duration')} 秒")
print(f" - 图片数: {video_info.get('image_count')}")
return True
else:
print(f"❌ 失败: {data.get('error')}")
return False
except Exception as e:
print(f"❌ 错误: {e}")
return False
async def test_list_files():
"""测试列出文件"""
print("\n" + "="*60)
print("测试 4: 列出生成的文件")
print("="*60)
try:
async with aiohttp.ClientSession() as session:
async with session.get(f"{API_BASE_URL}/api/v1/list-files?file_type=all") as response:
data = await response.json()
print(f"✅ 状态: {response.status}")
if data.get("success"):
files = data.get("files", [])
storage = data.get("storage", {})
print(f"\n📁 存储信息:")
print(f" - 总大小: {storage.get('total_size_mb')} MB")
print(f" - 图片数: {storage.get('image_count')}")
print(f" - 视频数: {storage.get('video_count')}")
print(f"\n📄 文件列表 (共 {len(files)} 个):")
for i, file in enumerate(files[:5], 1): # 只显示前5个
print(f" {i}. {file['filename']} ({file['type']}, {file['size_mb']} MB)")
if len(files) > 5:
print(f" ... 还有 {len(files) - 5} 个文件")
return True
else:
print(f"❌ 失败: {data.get('error')}")
return False
except Exception as e:
print(f"❌ 错误: {e}")
return False
async def test_mcp_tools():
"""测试 MCP 工具调用"""
print("\n" + "="*60)
print("测试 5: MCP 工具调用")
print("="*60)
try:
# 测试 tools/list
print("\n📋 测试 tools/list:")
payload = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": 1
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{API_BASE_URL}/mcp",
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
data = await response.json()
print(f"✅ 状态: {response.status}")
if "result" in data:
tools = data["result"].get("tools", [])
print(f"🔧 可用工具 ({len(tools)} 个):")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
return True
else:
print(f"❌ 失败: {data.get('error')}")
return False
except Exception as e:
print(f"❌ 错误: {e}")
return False
async def main():
"""运行所有测试"""
print("\n" + "="*60)
print("🧪 Video Generator Agent 测试套件")
print("="*60)
print(f"🌐 API URL: {API_BASE_URL}")
print(f"🔑 API Key: {API_KEY[:20]}...")
results = []
# 运行测试
results.append(("健康检查", await test_health_check()))
results.append(("生成单张图片", await test_generate_single_image()))
results.append(("生成视频", await test_generate_video()))
results.append(("列出文件", await test_list_files()))
results.append(("MCP 工具", await test_mcp_tools()))
# 汇总结果
print("\n" + "="*60)
print("📊 测试结果汇总")
print("="*60)
passed = sum(1 for _, result in results if result)
total = len(results)
for name, result in results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{status} - {name}")
print(f"\n总计: {passed}/{total} 通过")
if passed == total:
print("\n🎉 所有测试通过!")
return 0
else:
print(f"\n⚠️ {total - passed} 个测试失败")
return 1
if __name__ == "__main__":
try:
exit_code = asyncio.run(main())
sys.exit(exit_code)
except KeyboardInterrupt:
print("\n\n⚠️ 测试被中断")
sys.exit(1)
except Exception as e:
print(f"\n\n❌ 测试失败: {e}")
sys.exit(1)