多Agent协作商品图片/视频自动生成系统

商品视觉生成系统 - 操作文档多Agent协作的商品图片/视频自动生成系统使用指南仓库https://github.com/JianFeiGan/agent_part/tree/master一、项目概述1.1 项目简介本项目是一个基于 LangGraph 构建的多Agent协作系统能够自动化生成商品营销视觉内容包括商品主图- 白底产品图、场景展示图卖点图- 核心卖点可视化营销视频- 分镜设计 视频合成1.2 系统架构┌─────────────────────────────────────────────────────────────────┐ │ 用户请求 (商品信息) │ └─────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ OrchestratorAgent (编排调度) │ │ 任务分解、流程协调 │ └─────────────────────────────────────────────────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ RequirementAnalyzer│ │ CreativePlanner │ │ VisualDesigner │ │ (需求分析) │ │ (创意策划) │ │ (视觉设计) │ │ - 商品分析 │ │ - 主题设计 │ │ - 构图设计 │ │ - 卖点提取 │ │ - 风格推荐 │ │ - 提示词生成 │ │ - 风格建议 │ │ - 配色方案 │ │ - 分镜脚本 │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ ┌───────────────┴───────────────┐ ▼ ▼ ┌───────────────────┐ ┌───────────────────┐ │ ImageGenerator │ │ VideoGenerator │ │ (图片生成) │ │ (视频生成) │ │ - 通义万象API │ │ - 可灵AI API │ │ - 批量生成 │ │ - 场景合成 │ └───────────────────┘ └───────────────────┘ │ │ └───────────────┬───────────────┘ ▼ ┌───────────────────┐ │ QualityReviewer │ │ (质量审核) │ │ - 质量评分 │ │ - 合规检查 │ └───────────────────┘ │ ▼ ┌───────────────────┐ │ 输出结果 │ │ - 图片URLs │ │ - 视频URLs │ │ - 质量报告 │ └───────────────────┘1.3 技术栈组件技术选型版本PythonCPython3.11LLM 框架LangChain0.3.0Agent 编排LangGraph0.2.0主力 LLM通义千问 (qwen-max)-图像生成通义万象 (wanx-v1)-视频生成可灵AI (kling-v1)-数据验证Pydantic2.0包管理uv最新版二、环境准备2.1 系统要求操作系统: macOS / Linux / WindowsPython 版本: 3.11 或更高版本内存: 建议 4GB 以上2.2 安装 uv 工具uv 是一个现代化的 Python 包管理工具比 pip 快 10-100 倍。# macOS/Linuxcurl-LsSfhttps://astral.sh/uv/install.sh|sh# Windows (PowerShell)powershell-ExecutionPolicyByPass-cirm https://astral.sh/uv/install.ps1 | iex# 使用 Homebrew (macOS)brewinstalluv验证安装uv--version# 输出: uv 0.x.x2.3 获取 API Key本项目需要以下 API KeyAPI Key用途获取地址DASHSCOPE_API_KEY通义千问 LLM 通义万象图像生成https://dashscope.console.aliyun.com/KLING_API_KEY可灵AI 视频生成https://klingai.kuaishou.com/LANGCHAIN_API_KEYLangSmith 追踪可选https://smith.langchain.com/三、安装配置3.1 克隆项目gitclonerepository-urlcdagent_part3.2 创建虚拟环境并安装依赖# 创建虚拟环境uv venv# 激活虚拟环境# macOS/Linuxsource.venv/bin/activate# Windows.venv\Scripts\activate# 安装依赖uvsync# 安装开发依赖测试、代码检查等uvsync--all-extras3.3 配置环境变量# 复制环境变量模板cp.env.example .env# 编辑 .env 文件vim.env# 或使用其他编辑器.env文件配置说明# 必需配置 # 阿里云 DashScope API Key (通义千问、通义万象)DASHSCOPE_API_KEYsk-xxxxxxxxxxxxxxxx# 可灵AI API Key (视频生成)KLING_API_KEYxxxxxxxxxxxxxxxx# 可选配置 # LangSmith 追踪配置 (用于调试和监控)LANGCHAIN_TRACING_V2falseLANGCHAIN_API_KEYlsv2_pt_xxxxxxxxLANGCHAIN_PROJECTproduct-visual-generator# 服务配置HOST0.0.0.0PORT8000DEBUGtrue# 存储配置STORAGE_TYPElocalSTORAGE_PATH./output# 日志配置LOG_LEVELINFO3.4 验证安装# 验证核心组件导入uv run python-c from langchain_core.prompts import ChatPromptTemplate from langgraph.graph import StateGraph print(✅ LangChain 核心组件导入成功) # 验证配置加载uv run python-c from src.config import get_settings settings get_settings() print(f✅ 配置加载成功) print(f - LLM 模型: {settings.llm_model}) print(f - API Key 已配置: {bool(settings.dashscope_api_key)}) # 验证 Agent 导入uv run python-c from src.agents import OrchestratorAgent from src.graph import ProductVisualWorkflow print(✅ 所有模块导入成功) 四、项目启动方式4.0 快速运行推荐项目提供了开箱即用的示例脚本# 运行完整演示包含详细输出uv run python run_workflow.py# 运行快速示例仅生成图片uv run python run_workflow.py--quick4.1 方式一Python 脚本运行创建运行脚本run_workflow.py#!/usr/bin/env python运行商品视觉生成工作流示例。importasynciofromsrc.graphimportGenerationRequest,ProductVisualWorkflowfromsrc.models.productimportProduct,ProductCategory,SellingPointasyncdefmain()-None:主函数。# 1. 创建商品信息productProduct(product_idprod_001,name智能运动手表 Pro,brandTechWatch,categoryProductCategory.DIGITAL,description 全新智能运动手表支持心率监测、睡眠追踪、运动数据分析。 采用 1.4 英寸 AMOLED 高清屏幕IP68 级防水。 续航长达 14 天支持快速充电。 ,selling_points[SellingPoint(title心率实时监测,description24小时不间断心率监测异常提醒,priority5,),SellingPoint(title14天超长续航,description典型使用场景下续航 14 天,priority4,),SellingPoint(titleIP68防水,description游泳、洗澡无需摘下,priority3,),],)# 2. 创建生成请求requestGenerationRequest(task_typeimage_and_video,# 生成图片和视频image_types[main,scene,selling_point],image_count_per_type1,video_duration30.0,video_styleprofessional,style_preference科技感、现代简约,)# 3. 运行工作流print(*60)print(开始执行商品视觉生成工作流...)print(*60)workflowProductVisualWorkflow()resultawaitworkflow.run(product,request)# 4. 输出结果print(\n*60)print(执行完成结果摘要)print(*60)print(f商品名称:{result.product_info.nameifresult.product_infoelseN/A})print(f生成图片数量:{len(result.generated_images)})print(f生成视频:{是ifresult.generated_videoelse否})print(f质量评分:{result.quality_score})print(f发现问题:{len(result.issues)}个)ifresult.final_results:print(f\n改进建议:{result.final_results.get(recommendation,N/A)})# 5. 输出生成的资源print(\n-*40)print(生成的图片)fori,imginenumerate(result.generated_images,1):print(f{i}. [{img.image_type}]{img.urlorURL待生成})ifresult.generated_video:print(f\n生成的视频)print(f -{result.generated_video.urlorURL待生成})if__name____main__:asyncio.run(main())运行uv run python run_workflow.py4.2 方式二交互式 Python 环境# 进入 Python 交互环境uv run python# 或使用 IPython需安装uv run ipython在交互环境中importasynciofromsrc.graphimportProductVisualWorkflow,GenerationRequestfromsrc.models.productimportProduct,ProductCategory# 创建商品productProduct(name便携式蓝牙音箱,categoryProductCategory.DIGITAL,description小巧便携音质出色续航持久,)# 运行工作流asyncdefrun():workflowProductVisualWorkflow()returnawaitworkflow.run(product)resultasyncio.run(run())print(result.get_summary())4.3 方式三Jupyter Notebook项目支持在 Jupyter Notebook 中运行# 安装 Jupyteruvaddjupyter# 启动 Jupyteruv run jupyter notebook在 Notebook 中%load_ext autoreload%autoreload2fromsrc.graphimportProductVisualWorkflowfromsrc.models.productimportProduct,ProductCategory# ... 运行代码五、使用示例5.1 仅生成图片fromsrc.graphimportGenerationRequest,ProductVisualWorkflowfromsrc.models.productimportProduct,ProductCategory productProduct(name纯棉T恤,categoryProductCategory.CLOTHING,description100%纯棉舒适透气,)requestGenerationRequest(task_typeimage_only,# 仅生成图片image_types[main,scene],image_count_per_type2,)workflowProductVisualWorkflow()resultawaitworkflow.run(product,request)5.2 仅生成视频requestGenerationRequest(task_typevideo_only,# 仅生成视频video_duration15.0,# 15秒视频video_stylelifestyle,)resultawaitworkflow.run(product,request)5.3 自定义风格偏好requestGenerationRequest(task_typeimage_and_video,style_preference高端奢华黑金配色,color_preference金色,quality_levelpremium,# 高质量)resultawaitworkflow.run(product,request)5.4 使用会话状态管理# 首次运行workflowProductVisualWorkflow()result1awaitworkflow.run(product,request,thread_idsession_001)# 获取当前状态stateawaitworkflow.get_state(thread_idsession_001)print(f当前步骤:{state.current_step})# 继续同一会话result2awaitworkflow.run(another_product,request,thread_idsession_001)六、API 接口6.1 启动 API 服务# 启动后端 API 服务uv run python main.py# 或使用 uvicornuv run uvicorn main:app--host0.0.0.0--port8000--reload# 启动前端开发服务器cdfrontendnpminstallnpmrun dev服务启动后API 文档: http://localhost:8000/docs (Swagger UI)ReDoc 文档: http://localhost:8000/redoc前端界面: http://localhost:51736.2 REST API 端点商品管理方法路径说明POST/api/v1/products创建商品GET/api/v1/products获取商品列表分页GET/api/v1/products/{product_id}获取商品详情PUT/api/v1/products/{product_id}更新商品DELETE/api/v1/products/{product_id}删除商品POST/api/v1/products/{product_id}/images上传商品图片任务管理方法路径说明POST/api/v1/tasks创建生成任务异步GET/api/v1/tasks获取任务列表分页GET/api/v1/tasks/{task_id}获取任务详情GET/api/v1/tasks/{task_id}/status获取任务状态/进度POST/api/v1/tasks/{task_id}/cancel取消任务WebSocket路径说明/api/v1/tasks/{task_id}/ws任务状态实时推送6.3 请求示例# 创建商品curl-XPOST http://localhost:8000/api/v1/products\-HContent-Type: application/json\-d{ name: 智能运动手表 Pro, brand: TechWatch, category: digital, description: 支持心率监测的智能运动手表, selling_points: [ {title: 心率监测, description: 24小时实时监测, priority: 5} ] }# 创建生成任务curl-XPOST http://localhost:8000/api/v1/tasks\-HContent-Type: application/json\-d{ product_id: prod_xxxxx, task_type: image_and_video, image_types: [main, scene], video_duration: 30.0 }# 查询任务状态curlhttp://localhost:8000/api/v1/tasks/task_xxxxx/status6.4 前端管理界面前端采用Vue 3 Element Plus TypeScript构建提供以下功能页面路径功能仪表盘/dashboard统计概览、最近任务商品列表/products商品管理、搜索、分页创建商品/products/create商品创建表单编辑商品/products/:id/edit商品编辑任务列表/tasks任务管理、状态筛选创建任务/tasks/create选择商品、配置任务参数任务详情/tasks/:id实时进度、结果预览七、开发指南7.1 项目结构agent_part/ ├── .env # 环境变量配置 ├── .env.example # 环境变量模板 ├── pyproject.toml # 项目配置 ├── uv.lock # 依赖锁定文件 ├── main.py # FastAPI 应用入口 ├── run_workflow.py # 工作流运行脚本 ├── README.md # 项目说明 │ ├── src/ │ ├── __init__.py │ │ │ ├── agents/ # Agent 实现 │ │ ├── __init__.py │ │ ├── base.py # Agent 基类 │ │ ├── orchestrator.py # 编排调度 Agent │ │ ├── requirement_analyzer.py # 需求分析 Agent │ │ ├── creative_planner.py # 创意策划 Agent │ │ ├── visual_designer.py # 视觉设计 Agent │ │ ├── image_generator.py # 图片生成 Agent │ │ ├── video_generator.py # 视频生成 Agent │ │ └── quality_reviewer.py # 质量审核 Agent │ │ │ ├── graph/ # 状态图定义 │ │ ├── __init__.py │ │ ├── state.py # 状态定义 │ │ └── workflow.py # 工作流构建 │ │ │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ ├── product.py # 商品模型 │ │ ├── creative.py # 创意模型 │ │ ├── storyboard.py # 分镜模型 │ │ └── assets.py # 资源模型 │ │ │ ├── tools/ # 工具集成 │ │ └── __init__.py # 工具导出 │ │ │ ├── api/ # API 接口 │ │ ├── __init__.py │ │ ├── deps.py # 依赖注入 │ │ ├── router/ # 路由模块 │ │ │ ├── __init__.py │ │ │ ├── health.py # 健康检查 │ │ │ ├── products.py # 商品管理 │ │ │ └── tasks.py # 任务管理 WebSocket │ │ ├── schema/ # DTO 定义 │ │ │ ├── __init__.py │ │ │ ├── common.py # 通用响应 │ │ │ ├── product.py # 商品 DTO │ │ │ ├── task.py # 任务 DTO │ │ │ └── asset.py # 资源 DTO │ │ └── service/ # 服务层 │ │ ├── __init__.py │ │ ├── redis_client.py # Redis 存储 │ │ └── task_manager.py # 任务管理器 │ │ │ └── config/ # 配置管理 │ ├── __init__.py │ └── settings.py # 配置类 │ ├── tests/ # 测试用例 │ ├── test_agents/ │ ├── test_graph/ │ ├── test_models/ │ └── test_tools/ │ ├── frontend/ # Vue 3 前端项目 │ ├── package.json # 依赖配置 │ ├── vite.config.ts # Vite 构建配置 │ └── src/ │ ├── main.ts # 应用入口 │ ├── App.vue # 根组件 │ ├── api/ # API 调用封装 │ ├── views/ # 页面组件 │ │ ├── Dashboard.vue │ │ ├── products/ │ │ └── tasks/ │ ├── stores/ # Pinia 状态管理 │ ├── router/ # 路由配置 │ └── types/ # TypeScript 类型 │ └── output/ # 输出目录 ├── images/ # 生成的图片 └── videos/ # 生成的视频7.2 运行测试# 运行所有测试uv run pytest# 运行带覆盖率的测试uv run pytest--covsrc --cov-reporthtml# 运行指定测试文件uv run pytest tests/test_agents/test_base.py-v# 运行指定测试类uv run pytest tests/test_graph/test_state.py::TestWorkflowBuilder-v7.3 代码质量检查# 格式化代码uv run ruffformat.# Lint 检查uv run ruff check.# 自动修复 Lint 问题uv run ruff check.--fix# 类型检查uv run mypy src/# 完整检查流程uv run ruffformat.uv run ruff check.uv run mypy src/uv run pytest7.4 添加新依赖# 添加运行时依赖uvaddpackage-name# 添加开发依赖uvadd--devpackage-name# 示例uvaddhttpx uvadd--devpytest-asyncio八、常见问题8.1 循环导入错误问题ImportError: cannot import name xxx from partially initialized module原因模块之间存在循环依赖。解决确保使用延迟导入。在workflow.py中Agent 的导入放在方法内部defadd_agent_nodes(self)-WorkflowBuilder:# 延迟导入以避免循环导入fromsrc.agents.orchestratorimportOrchestratorAgent# ...8.2 API Key 无效问题Invalid API Key或401 Unauthorized检查uv run python-c from src.config import get_settings s get_settings() print(fAPI Key 长度: {len(s.dashscope_api_key)}) print(fAPI Key 前缀: {s.dashscope_api_key[:10]}...) 8.3 测试失败问题测试运行失败解决# 重新同步依赖uvsync--all-extras# 清理缓存uv cache clean# 重新运行测试uv run pytest-v--tblong8.4 生成内容为空问题生成的图片/视频 URL 为空原因当前为模拟模式API 调用返回占位符。解决在 Agent 中实现真实的 API 调用src/agents/image_generator.py- 实现_call_image_api方法src/agents/video_generator.py- 实现_call_video_api方法九、下一步开发9.1 已完成功能REST API 接口FastAPIRedis 任务状态持久化WebSocket 实时状态推送Vue 3 前端管理界面商品管理 CRUD异步任务管理9.2 待完成功能接入真实的图像生成 API通义万象接入真实的视频生成 API可灵AI用户认证与权限管理文件上传存储OSS/本地完善错误处理和重试机制单元测试和集成测试9.2 性能优化建议并发控制限制同时运行的 Agent 数量缓存策略缓存 LLM 响应和生成结果异步优化使用asyncio.gather并行执行独立任务资源池复用 HTTP 连接和 LLM 实例十、联系与支持项目文档:documents/目录问题反馈: 提交 Issue技术支持: 联系项目维护者文档版本: 1.0.0最后更新: 2026-03-25