极速AI绘画:Dify+Grok一键生图工作流搭建全攻略

Tuesday, March 25, 2025 - AI Agent - Grok Dify AI Art

嗨,大家好!最近X官方开放了Grok模型的图像生成API,为AI绘画领域带来了新的可能性。今天给大家分享的是如何结合Dify与Grok的图像生成能力,打造一个简单易用的一键AI生图助手。

效果预览:
![](/images/c1031b7d17c23df33a80b56dcde701aa.png)

完整工作流概览:(DSL见文末地址)

![](/images/def66dac7dc45d701a2416a629e61248.png)

整个一键生图助手由三部分组成:

1. Grok图像生成服务端:调用Grok的grok-2-image API 生成图像,可选配置Cloudflare R2存储
2. Dify自定义工具:通过OpenAPI schema规范将我们的服务集成到Dify平台
3. Dify工作流:结合问题分类器和图像生成逻辑的完整应用流程

Grok生成图片的服务端应用搭建

1. 申请Grok API密钥

第一步:在Grok官方(https://console.x.ai/)申请API KEY,我在这里创建了一个专门用于图片生成的key,便于权限管理。

![](/images/53f520734c408c1611019e58c11d3c06.png)

2. 配置Cloudflare R2存储(可选)

为了提供稳定的图片访问地址,我们可以使用Cloudflare R2对象存储:

1. 打开Cloudflare控制台(https://dash.cloudflare.com/),在左侧菜单中找到【R2对象存储】,然后创建存储桶

![](/images/87fc7850f549b9d988a9c4349127592e.png)

2. 设置允许公开访问, 或者你有域名的话,可以绑定到自定义的二级域名。

!公开访问设置

3. 创建R2存储的API令牌。

![](/images/40f0de9dcad1e79d8ddad22172af9025.png)

4. 创建之后,会生成访问密钥ID、机密访问密钥以及R2存储的访问url,这里注意保存,后续配置会用到这三个值;

![](/images/ff0a365ab9c0936288af2c10f77d3acf.png)

3. 服务端代码实现

我在这里创建了一个使用FastAPI框架的图片生成服务端应用,并提供了两种方式:
* 调用Grok的grok-2-image API生成图片,然后将图片上传到Cloudflare R2存储, 并返回Cloudflare图片可访问的公开地址;
* 直接返回调用Grok的grok-2-image API生成的图片url。

服务端的完整代码如下:

import os
import base64
import json
import requests
import uuid
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import boto3
from botocore.config import Config
from datetime import datetime
import io

app = FastAPI()

Grok API配置


GROK_API_KEY = os.environ.get("GROK_API_KEY", "your_grok_api_key")
GROK_API_URL = "https://api.x.ai/v1/images/generations"

Cloudflare R2配置


R2_ACCESS_KEY_ID = os.environ.get("R2_ACCESS_KEY_ID", "your_r2_access_key_id")
R2_SECRET_ACCESS_KEY = os.environ.get("R2_SECRET_ACCESS_KEY", "your_r2_secret_access_key")
R2_ENDPOINT_URL = os.environ.get("R2_ENDPOINT_URL", "https://your-account.r2.cloudflarestorage.com")
R2_BUCKET_NAME = os.environ.get("R2_BUCKET_NAME", "your-bucket-name")

初始化R2客户端


s3_client = boto3.client(
's3',
endpoint_url=R2_ENDPOINT_URL,
aws_access_key_id=R2_ACCESS_KEY_ID,
aws_secret_access_key=R2_SECRET_ACCESS_KEY,
config=Config(signature_version='s3v4')
)

class ImageRequest(BaseModel):
prompt: str
n: int = 1
response_format: str = "url"
model: str = "grok-2-image" # 添加模型参数,默认为grok-2-image

@app.post("/generate-image")
async def generate_image(request: ImageRequest, background_tasks: BackgroundTasks):
try:
# 调用Grok API生成图片
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {GROK_API_KEY}"
}

payload = {
"prompt": request.prompt,
"n": request.n,
"response_format": request.response_format,
"model": request.model # 添加模型参数到请求负载
}

response = requests.post(
GROK_API_URL,
headers=headers,
data=json.dumps(payload)
)

if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=f"Grok API error: {response.text}")

response_data = response.json()


image_urls = []
for i, image_data in enumerate(response_data.get("data", [])):
# 处理并上传图片到R2
if request.response_format == "b64_json":
image_content = base64.b64decode(image_data.get("b64_json", ""))

# 生成唯一文件名
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
file_name = f"{timestamp}_{uuid.uuid4()}.png"

# 上传到R2
background_tasks.add_task(
upload_to_r2,
image_content,
file_name
)

# 构建自定义域名URL
image_url = f"https://image.lskun.top/{file_name}"
image_urls.append(image_url)
elif request.response_format == "url":
# 如果响应格式是url,直接获取API返回的URL
if "url" in image_data:
image_urls.append(image_data["url"])

return JSONResponse(content={"status": "success", "message": "图片已生成", "image_urls": image_urls})

except Exception as e:
raise HTTPException(status_code=500, detail=f"处理请求时出错: {str(e)}")

def upload_to_r2(image_content, file_name):
"""将图片上传到Cloudflare R2"""
try:
s3_client.put_object(
Bucket=R2_BUCKET_NAME,
Key=file_name,
Body=io.BytesIO(image_content),
ContentType="image/png"
)
print(f"成功上传图片到R2: {file_name}")
except Exception as e:
print(f"上传图片到R2时出错: {str(e)}")

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)


说明:如果不使用Cloudflare R2存储,只需替换代码中的GROK_API_KEY环境变量即可。

4. 服务端部署与测试

首先,需要确保安装了相关的依赖包

pip install fastapi uvicorn python-multipart requests boto3

运行服务:
python grok_image_generator.py

测试验证:
curl -X POST "http://localhost:8000/generate-image" \
-H "Content-Type: application/json" \
-d '{"prompt": "一只可爱的猫咪在阳光下玩耍", "n": 1}'

看到正常返回,说明服务运行成功。

![](/images/0c5d11448c43089641612a4cce59130c.png)

配置Dify自定义工具


接下来,我们进入Dify的【工具】页面,点击创建自定义工具

![](/images/13e1d9bb79c5c0ba8a88de176e9cc84b.png)

这里我们需要将符合dify规范的schema json填入,注意这里的url替换成你本地ip地址
![](/images/13e583e199554d70d23292108d532e6b.png)

我在这里给出了对应的schema json:

{
"openapi": "3.0.2",
"info": {
"title": "Grok 图像生成 API",
"description": "使用 Grok API 生成图像并上传到 Cloudflare R2 存储",
"version": "1.0.0"
},
"servers": [
{
"url": "http://192.168.1.3:8000",
"description": "默认服务器"
}
],
"paths": {
"/generate-image": {
"post": {
"summary": "生成图像",
"description": "根据提供的提示词生成图像,并将其上传到 Cloudflare R2 存储",
"operationId": "generate_image",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ImageRequest"
}
}
},
"required": true
}
}
}
},
"components": {
"schemas": {
"ImageRequest": {
"type": "object",
"required": ["prompt"],
"properties": {
"prompt": {
"type": "string",
"description": "用于生成图像的提示词"
},
"n": {
"type": "integer",
"description": "要生成的图像数量",
"default": 1
},
"response_format": {
"type": "string",
"description": "响应格式",
"default": "url"
},
"model": {
"type": "string",
"description": "用于生成图像的模型",
"default": "grok-2-image"
}
}
}
}
}
}

自定义接口可以增加一个鉴权,增加安全性。

![](/images/e55cc57d6b86ec73816be5b0a866147d.png)

配置完成后,我们可以点击测试进行验证,可以看到验证通过了

![](/images/e3c5fdaa6358f325336974ab94f4f234.png)

创建Dify工作流

1. 创建Chatflow类型工作流

接下来,我们在dify中创建一个chatflow类型的工作流

![](/images/55fc014006629bada0058b27e4b69ae5.png)

2. 配置问题分类器

在start节点后创建一个问题分类器节点,我这里使用的是deepseek-v3模型,主要是根据用户输入的查询语句判定是AI绘画还是其他,从而走不同的条件分支;

* 如果不是AI绘画,则走到QwQ-32b的推理模型进行普通聊天问答;

* 如果是AI绘画,则走到下一步配置的Grok 图像生成工具。

3. 配置Grok图像生成工具

在问题分类器后面添加节点-工具-自定义工具-Grok 图像生成工具,在环境变量中配置:

* response_format: url
* model: grok-2-image

![](/images/49b82d0d6f1e66b46e0e656101e7ab10.png)

4. 配置代码执行节点

下一步配置代码执行节点,这里主要是提取返回json结构的image_url并转换成markdown格式,用于在聊天框中正常显示。

![](/images/19f4f76499152a4d3bd10b9709aa54cc.png)

代码如下:

def parse_image_response(response_json: str) -> list:

try:
data = json.loads(response_json)
if data.get("status") == "success" and "image_urls" in data:
return data["image_urls"]
return []
except json.JSONDecodeError:
print("JSON解析错误")
return []
except Exception as e:
print(f"解析响应时出错: {str(e)}")
return []

def main(arg1: str) -> dict:

image_urls = parse_image_response(arg1)
return {
"result": f"![]({image_urls[0]})"
}

工作流测试与应用


最后进行完整的chatflow工作流验证,我们点击chatflow 右上角的预览按钮,在左下角输入提示词 比如:帮我画一只狗和猫在公园的草地上玩耍,整个工作流执行效果如下:

![](/images/f35ba202f410cc30065dd1bfebca83e0.gif)

写在最后


通过上面的案例演示,我们成功构建了一个基于Dify和Grok的一键生图助手,实现了从文本描述到高质量的图像生成。好了,今天的分享就到这里啦!期待在未来的文章中与大家分享更多Dify的高级应用技巧!如果你有任何使用心得或疑问,也欢迎在评论区留言交流。

获取资源:关注公众号,回复【一键生图】获取完整的Dify+Grok一键生图工作流DSL配置文件。