Nunchaku-flux-1-dev与数据库联动MySQL管理生成结果与用户偏好最近在折腾AI图像生成特别是Nunchaku-flux-1-dev这个模型效果确实挺惊艳的。但用久了就发现一个问题生成的图片越来越多想找之前某张图或者看看自己最喜欢哪种风格简直是大海捞针。每次都得在文件夹里翻来翻去时间一长自己都记不清了。这让我想到能不能像管理音乐歌单或者电商收藏夹一样来管理这些AI生成的作品呢把每次生成的任务信息、图片链接还有用户的喜好比如收藏、评分都存起来。这样不仅能随时查看历史记录还能根据大家的喜好智能推荐一些热门的生成风格。说干就干我决定用最熟悉的MySQL数据库来搭这个后台。这篇文章我就来分享一下怎么设计这个系统以及如何用Python把它和Nunchaku-flux-1-dev模型联动起来实现一个带历史记录和智能推荐功能的AI图像生成应用。1. 系统设计与数据库搭建在开始写代码之前得先把整个系统的架子搭好核心就是数据库怎么设计。我们的目标是记录两类主要信息一是每次生成任务的“过程”二是用户对生成结果的“反馈”。1.1 核心数据表设计我设计了四张核心表它们之间的关系就像是一个简单的社交应用有用户、有作品生成任务、有对作品的互动收藏。生成任务表 (generation_tasks)这是最核心的表记录每一次图像生成的“出生证明”。CREATE TABLE generation_tasks ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(100) NOT NULL COMMENT ‘用户标识可以是用户名或ID‘, prompt TEXT NOT NULL COMMENT ‘用户输入的描述文本‘, negative_prompt TEXT COMMENT ‘用户指定的负面描述不希望出现的元素‘, style_preset VARCHAR(50) COMMENT ‘使用的风格预设如‘anime‘, ‘photographic‘‘, steps INT DEFAULT 20 COMMENT ‘生成步数‘, cfg_scale DECIMAL(3,1) DEFAULT 7.5 COMMENT ‘提示词相关性强度‘, seed BIGINT COMMENT ‘随机种子用于复现相同结果‘, image_url VARCHAR(500) NOT NULL COMMENT ‘生成图片的存储URL或路径‘, status ENUM(‘pending‘, ‘success‘, ‘failed‘) DEFAULT ‘pending‘ COMMENT ‘任务状态‘, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘任务创建时间‘ );这张表把一次生成请求的所有参数和结果都存下来了。特别是seed字段有了它只要参数一样你就能随时“复刻”出同一张图这对创作非常有用。用户收藏表 (user_favorites)用来记录用户“点赞”或“收藏”了哪些作品。这是实现个性化推荐的数据基础。CREATE TABLE user_favorites ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(100) NOT NULL, task_id INT NOT NULL COMMENT ‘收藏的生成任务ID‘, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_id) REFERENCES generation_tasks(id) ON DELETE CASCADE, UNIQUE KEY unique_favorite (user_id, task_id) -- 防止重复收藏 );这里设置了一个唯一约束确保同一个用户不会对同一作品重复收藏。风格标签表 (style_tags) 任务-标签关联表 (task_style_tags)为了能做风格推荐我们需要对作品打标签。我采用了一种灵活的方式先有一个所有可用风格的标签库然后再关联具体任务。-- 风格标签库 CREATE TABLE style_tags ( id INT AUTO_INCREMENT PRIMARY KEY, tag_name VARCHAR(50) NOT NULL UNIQUE COMMENT ‘风格标签名如‘赛博朋克‘, ‘水墨风‘‘ ); -- 关联表一个任务可以有多个风格标签 CREATE TABLE task_style_tags ( task_id INT NOT NULL, tag_id INT NOT NULL, PRIMARY KEY (task_id, tag_id), FOREIGN KEY (task_id) REFERENCES generation_tasks(id) ON DELETE CASCADE, FOREIGN KEY (tag_id) REFERENCES style_tags(id) ON DELETE CASCADE );为什么不直接把风格存在任务表里因为一张图可能融合多种风格比如“赛博朋克水墨风”用关联表的方式更灵活也方便后续的统计和查询。1.2 本地MySQL环境快速配置为了让大家能快速上手我们先用最方便的方式在本地把MySQL跑起来。这里推荐使用Docker省去繁琐的安装和配置。如果你还没安装Docker可以去官网下载安装。安装好后打开终端或命令行执行下面这条命令docker run -d \ --name mysql-for-ai \ -p 3306:3306 \ -e MYSQL_ROOT_PASSWORDyour_strong_password \ -e MYSQL_DATABASEai_gallery \ mysql:8.0这条命令做了几件事启动一个MySQL 8.0的容器命名为mysql-for-ai把容器的3306端口映射到你电脑的3306端口设置root用户的密码记得把your_strong_password换成你自己的复杂密码同时创建一个叫ai_gallery的数据库。容器启动后你可以用任何MySQL客户端比如命令行工具mysql或者图形化工具DBeaver、Navicat连接它。主机localhost或127.0.0.1端口3306用户名root密码你刚才设置的密码连接成功后就可以执行前面提供的那些CREATE TABLE语句来创建表了。建议按顺序创建先generation_tasks再style_tags接着task_style_tags最后user_favorites因为后面几张表依赖前面的表。2. 用Python连接与操作数据库数据库搭好了下一步就是让我们的Python程序能和它“对话”。我会用一个叫SQLAlchemy的库它能把数据库表变成Python类用起来非常直观。2.1 建立数据库连接与模型定义首先安装必要的库pip install sqlalchemy pymysql。pymysql是Python连接MySQL的驱动。然后我们创建一个database.py文件在这里定义数据库连接和那些表对应的Python类我们称之为“模型”。from sqlalchemy import create_engine, Column, Integer, String, Text, Float, BigInteger, Enum, TIMESTAMP, ForeignKey, UniqueConstraint from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import enum # 定义数据库连接地址格式为mysqlpymysql://用户名:密码主机:端口/数据库名 DATABASE_URL “mysqlpymysql://root:your_strong_passwordlocalhost:3306/ai_gallery“ # 创建引擎和基类 engine create_engine(DATABASE_URL, echoTrue) # echoTrue 方便查看执行的SQL语句 Base declarative_base() SessionLocal sessionmaker(bindengine) # 定义任务状态枚举 class TaskStatus(enum.Enum): PENDING ‘pending‘ SUCCESS ‘success‘ FAILED ‘failed‘ # 定义生成任务模型对应 generation_tasks 表 class GenerationTask(Base): __tablename__ ‘generation_tasks‘ id Column(Integer, primary_keyTrue, indexTrue) user_id Column(String(100), nullableFalse) prompt Column(Text, nullableFalse) negative_prompt Column(Text) style_preset Column(String(50)) steps Column(Integer, default20) cfg_scale Column(Float, default7.5) seed Column(BigInteger) image_url Column(String(500), nullableFalse) status Column(Enum(TaskStatus), defaultTaskStatus.PENDING) created_at Column(TIMESTAMP, defaultdatetime.utcnow) # 定义关系一个任务可以有多个收藏也可以关联多个风格标签 favorites relationship(“UserFavorite“, back_populates“task“, cascade“all, delete-orphan“) style_tags relationship(“StyleTag“, secondary“task_style_tags“, back_populates“tasks“) # 定义用户收藏模型 class UserFavorite(Base): __tablename__ ‘user_favorites‘ __table_args__ (UniqueConstraint(‘user_id‘, ‘task_id‘, name‘unique_favorite‘),) id Column(Integer, primary_keyTrue, indexTrue) user_id Column(String(100), nullableFalse) task_id Column(Integer, ForeignKey(‘generation_tasks.id‘, ondelete‘CASCADE‘), nullableFalse) created_at Column(TIMESTAMP, defaultdatetime.utcnow) # 定义关系一个收藏属于一个任务 task relationship(“GenerationTask“, back_populates“favorites“) # 定义风格标签模型 class StyleTag(Base): __tablename__ ‘style_tags‘ id Column(Integer, primary_keyTrue, indexTrue) tag_name Column(String(50), uniqueTrue, nullableFalse) # 定义关系一个标签可以被多个任务使用 tasks relationship(“GenerationTask“, secondary“task_style_tags“, back_populates“style_tags“) # 注意task_style_tags 是关联表SQLAlchemy会自动处理通常不需要单独定义模型。 # 创建所有表如果不存在的话 Base.metadata.create_all(bindengine)这段代码的核心是把数据库表“映射”成了Python类。Column定义字段relationship定义表与表之间的关系。运行后如果数据库里没有这些表SQLAlchemy会自动帮你创建。2.2 实现核心数据操作函数光有模型还不够我们需要一些具体的函数来增删改查数据。在同一个目录下创建crud.pyCRUD代表创建、读取、更新、删除。from sqlalchemy.orm import Session from database import SessionLocal, GenerationTask, UserFavorite, StyleTag, TaskStatus from typing import List, Optional # 获取数据库会话的依赖函数 def get_db(): db SessionLocal() try: yield db finally: db.close() # --- 生成任务相关操作 --- def create_generation_task(db: Session, user_id: str, prompt: str, image_url: str, negative_prompt: Optional[str] None, style_preset: Optional[str] None, steps: int 20, cfg_scale: float 7.5, seed: Optional[int] None): 创建一条新的生成任务记录 db_task GenerationTask( user_iduser_id, promptprompt, negative_promptnegative_prompt, style_presetstyle_preset, stepssteps, cfg_scalecfg_scale, seedseed, image_urlimage_url, statusTaskStatus.SUCCESS # 假设调用此函数时任务已成功 ) db.add(db_task) db.commit() db.refresh(db_task) # 刷新以获取自动生成的ID等字段 return db_task def get_user_tasks(db: Session, user_id: str, skip: int 0, limit: int 100): 查询某个用户的所有生成任务支持分页 return db.query(GenerationTask)\ .filter(GenerationTask.user_id user_id)\ .order_by(GenerationTask.created_at.desc())\ .offset(skip).limit(limit).all() def get_task_by_id(db: Session, task_id: int): 根据ID查询单个任务 return db.query(GenerationTask).filter(GenerationTask.id task_id).first() # --- 用户收藏相关操作 --- def add_to_favorites(db: Session, user_id: str, task_id: int): 用户收藏一个生成任务 # 先检查是否已收藏 existing db.query(UserFavorite).filter( UserFavorite.user_id user_id, UserFavorite.task_id task_id ).first() if existing: return existing # 已收藏直接返回 db_fav UserFavorite(user_iduser_id, task_idtask_id) db.add(db_fav) db.commit() db.refresh(db_fav) return db_fav def get_user_favorites(db: Session, user_id: str): 获取用户收藏的所有任务 return db.query(GenerationTask)\ .join(UserFavorite, GenerationTask.id UserFavorite.task_id)\ .filter(UserFavorite.user_id user_id)\ .order_by(UserFavorite.created_at.desc())\ .all() # --- 风格标签相关操作 --- def get_or_create_style_tag(db: Session, tag_name: str): 获取或创建一个风格标签 tag db.query(StyleTag).filter(StyleTag.tag_name tag_name).first() if not tag: tag StyleTag(tag_nametag_name) db.add(tag) db.commit() db.refresh(tag) return tag def add_style_tag_to_task(db: Session, task_id: int, tag_name: str): 为生成任务添加一个风格标签 task get_task_by_id(db, task_id) tag get_or_create_style_tag(db, tag_name) if tag not in task.style_tags: task.style_tags.append(tag) db.commit() return task这些函数封装了最基本的数据库操作。比如create_generation_task它接收生成图片的所有参数然后像填表格一样把这些信息存到数据库里并返回存好的记录。3. 整合Nunchaku-flux-1-dev与数据库现在我们要把AI模型和数据库这两块“拼图”组合起来。想象一个完整的流程用户输入描述 - 模型生成图片 - 图片上传到图床 - 所有信息存入数据库。3.1 构建完整的图像生成与存储流程我设计了一个ImageGenerator类它负责调度整个流程。这里假设你已经有了Nunchaku-flux-1-dev模型的调用方式可能是通过API或本地库并且有一个上传图片到云存储如阿里云OSS、腾讯云COS或图床的服务。import requests from typing import Optional from crud import create_generation_task, add_style_tag_to_task, get_db from database import SessionLocal class ImageGenerator: def __init__(self, model_endpoint: str, upload_service): :param model_endpoint: Nunchaku-flux-1-dev模型的API地址 :param upload_service: 图片上传服务的实例需有upload(image_data)方法返回URL self.model_endpoint model_endpoint self.upload_service upload_service def generate_and_save(self, user_id: str, prompt: str, **kwargs): 完整的生成与保存流程 :param user_id: 用户ID :param prompt: 生成描述 :param kwargs: 其他模型参数如negative_prompt, steps等 :return: 保存到数据库后的任务对象 # 1. 调用模型生成图片 print(f“用户 {user_id} 正在生成: {prompt}“) image_data self._call_model(prompt, **kwargs) # 2. 上传图片到存储服务获取可访问的URL print(“图片生成成功正在上传...“) image_url self.upload_service.upload(image_data) # 3. 获取数据库会话保存任务记录 db SessionLocal() try: db_task create_generation_task( dbdb, user_iduser_id, promptprompt, image_urlimage_url, negative_promptkwargs.get(‘negative_prompt‘), style_presetkwargs.get(‘style_preset‘), stepskwargs.get(‘steps‘, 20), cfg_scalekwargs.get(‘cfg_scale‘, 7.5), seedkwargs.get(‘seed‘) ) print(f“任务记录已保存ID: {db_task.id}“) # 4. (可选) 根据预设风格或分析结果为任务添加标签 style_preset kwargs.get(‘style_preset‘) if style_preset: add_style_tag_to_task(db, db_task.id, style_preset) print(f“已添加风格标签: {style_preset}“) db.commit() return db_task except Exception as e: db.rollback() print(f“保存到数据库时出错: {e}“) raise finally: db.close() def _call_model(self, prompt: str, **kwargs): 调用Nunchaku-flux-1-dev模型生成图片。 这里需要替换为你实际调用模型的方式HTTP请求或本地库调用。 返回图片的二进制数据。 # 示例假设模型提供HTTP API payload { “prompt“: prompt, “negative_prompt“: kwargs.get(‘negative_prompt‘, ‘’), “steps“: kwargs.get(‘steps‘, 20), “cfg_scale“: kwargs.get(‘cfg_scale‘, 7.5), “seed“: kwargs.get(‘seed‘), “style_preset“: kwargs.get(‘style_preset‘) } # 移除值为None的项 payload {k: v for k, v in payload.items() if v is not None} response requests.post(self.model_endpoint, jsonpayload, timeout60) response.raise_for_status() # 假设API返回的是图片二进制流 return response.content # 一个简单的上传服务模拟实际使用时替换为阿里云OSS、七牛云等SDK class MockUploadService: def upload(self, image_data): # 模拟上传返回一个假URL。真实场景应调用云存储SDK import uuid fake_filename f“{uuid.uuid4()}.png“ return f“https://your-cloud-storage.example.com/{fake_filename}“这个ImageGenerator类就像一个流水线工人。你给它用户ID和描述它就去调用模型干活拿到图片后存到云上最后把这次任务的“档案”规规矩矩地存进数据库。MockUploadService是一个模拟类真实项目中你需要换成阿里云OSS、腾讯云COS这类服务的SDK。3.2 实现用户历史与收藏功能有了数据库作为后台实现历史记录和收藏功能就水到渠成了。我们可以创建一个简单的用户界面比如用Flask框架来展示这些功能。from flask import Flask, request, jsonify from crud import get_user_tasks, add_to_favorites, get_user_favorites, get_db from database import SessionLocal app Flask(__name__) app.route(‘/api/user/user_id/tasks‘, methods[‘GET‘]) def get_user_history(user_id): 获取用户的历史生成记录 db SessionLocal() try: tasks get_user_tasks(db, user_id) # 将任务对象转换为字典列表方便JSON序列化 result [] for task in tasks: task_dict { “id“: task.id, “prompt“: task.prompt, “image_url“: task.image_url, “style_preset“: task.style_preset, “created_at“: task.created_at.isoformat() if task.created_at else None } result.append(task_dict) return jsonify({“user_id“: user_id, “tasks“: result}) finally: db.close() app.route(‘/api/user/user_id/favorites‘, methods[‘POST‘]) def favorite_task(user_id): 用户收藏一个任务 data request.get_json() task_id data.get(‘task_id‘) if not task_id: return jsonify({“error“: “Missing task_id“}), 400 db SessionLocal() try: favorite add_to_favorites(db, user_id, task_id) return jsonify({“message“: “Added to favorites“, “favorite_id“: favorite.id}) except Exception as e: return jsonify({“error“: str(e)}), 500 finally: db.close() app.route(‘/api/user/user_id/favorites‘, methods[‘GET‘]) def list_favorites(user_id): 列出用户收藏的所有任务 db SessionLocal() try: tasks get_user_favorites(db, user_id) result [] for task in tasks: task_dict { “id“: task.id, “prompt“: task.prompt, “image_url“: task.image_url, “favorited_at“: None # 这里需要从关联关系获取收藏时间略作简化 } result.append(task_dict) return jsonify({“user_id“: user_id, “favorites“: result}) finally: db.close() if __name__ ‘__main__‘: app.run(debugTrue)这个简单的API服务提供了三个端点查看历史、收藏作品、查看收藏列表。前端页面可以通过调用这些接口轻松展示用户的历史作品墙和个人收藏夹。4. 从数据到智能热门风格推荐数据库存了数据如果只是查询和展示那就太浪费了。我们可以利用这些数据做一些简单的分析比如找出最受大家欢迎的生成风格。4.1 基于收藏数据的统计分析思路很简单被收藏越多的作品说明其风格或内容越受欢迎。我们可以统计每个风格标签下的作品被收藏的总次数。from sqlalchemy import func from crud import get_db from database import SessionLocal, StyleTag, UserFavorite, task_style_tags def get_popular_styles(db: SessionLocal, limit: int 5): 获取最热门的风格标签按被收藏次数排序 # 使用SQLAlchemy查询关联 style_tags, task_style_tags, generation_tasks, user_favorites 表 # 按标签分组统计其下所有任务被收藏的总次数然后排序 popular_styles db.query( StyleTag.tag_name, func.count(UserFavorite.id).label(‘favorite_count‘) )\ .join(task_style_tags, StyleTag.id task_style_tags.c.tag_id)\ .join(UserFavorite, task_style_tags.c.task_id UserFavorite.task_id)\ .group_by(StyleTag.id, StyleTag.tag_name)\ .order_by(func.count(UserFavorite.id).desc())\ .limit(limit)\ .all() return [{tag: style.tag_name, count: style.favorite_count} for style in popular_styles] # 使用示例 db SessionLocal() try: top_styles get_popular_styles(db, limit3) print(“当前热门风格“) for style in top_styles: print(f“ - {style[‘tag‘]}: 被收藏 {style[‘count‘]} 次“) finally: db.close()这段代码的核心是一个相对复杂的SQL查询用SQLAlchemy表达了出来。它把四张表连起来数出每个风格标签下的作品总共被收藏了多少次。这个结果可以展示在应用的“发现”或“热门”页面给新用户提供创作灵感。4.2 构建简单的推荐接口基于上面的统计我们可以提供一个推荐接口。例如当用户准备生成图片时可以推荐当前最热门的风格或者根据用户自己常使用的风格推荐相似的其他风格。app.route(‘/api/recommendations/popular-styles‘, methods[‘GET‘]) def recommend_popular_styles(): 推荐热门风格 db SessionLocal() try: styles get_popular_styles(db, limit5) return jsonify({“recommendations“: styles}) finally: db.close() app.route(‘/api/user/user_id/recommendations‘, methods[‘GET‘]) def recommend_for_user(user_id): 基于用户历史行为推荐风格简化版推荐他没用过的热门风格 db SessionLocal() try: # 1. 获取用户使用过的风格 user_used_styles db.query(StyleTag.tag_name)\ .join(task_style_tags, StyleTag.id task_style_tags.c.tag_id)\ .join(GenerationTask, task_style_tags.c.task_id GenerationTask.id)\ .filter(GenerationTask.user_id user_id)\ .distinct().all() used_style_set {style[0] for style in user_used_styles} # 2. 获取所有热门风格 all_popular get_popular_styles(db, limit10) # 3. 过滤掉用户已使用的返回前3个 recommendations [s for s in all_popular if s[‘tag‘] not in used_style_set][:3] return jsonify({“user_id“: user_id, “recommendations“: recommendations}) finally: db.close()第一个接口直接返回全局热门风格。第二个接口稍微智能一点它会先查一下这个用户过去用过哪些风格然后从热门榜里剔除这些推荐一些他还没尝试过但很受欢迎的风格给他帮助他拓展创作边界。5. 总结把这个系统跑起来后感觉整个AI图像生成的体验就完整了。以前生成图片就像撒网捕鱼捞上来是什么就是什么很难整理。现在有了数据库这个“仓库”每次生成的作品都被妥善归档附上了详细的“出生信息”。用户可以随时翻看自己的创作历史把满意的作品加入收藏夹甚至还能从社区的热门风格里找到新的灵感。从技术实现上看用MySQL来管理这种结构化的元数据非常合适查询效率高关系清晰。SQLAlchemy让Python操作数据库变得像操作普通对象一样简单。整个系统的扩展性也不错比如未来想增加图片评分、评论功能或者做更复杂的协同过滤推荐都可以在现有的数据模型上平滑地添加。当然这只是个起点。你可以根据实际需求继续丰富它比如增加任务队列管理异步生成、对生成提示词进行分词和情感分析以便做内容推荐、或者用更复杂的算法来挖掘风格之间的关联。希望这个案例能给你带来启发让你手中的AI模型不仅是一个生成工具更能成为一个有记忆、可互动、能学习的创作伙伴。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。