Kling AI API를 활용한 이미지 생성 방법
요즘 다양한 생성형 AI를 활용하여 이미지를 생성하면서 모델별 성능을 비교하고 있습니다.
각 모델의 기본적인 설정 값과 특성을 파악하는 것이 목표입니다.
웹사이트에서 직접 테스트하는 것은 장기적인 관점에서 비효율적이라고 판단하여 API를 활용한 테스트를 진행해볼까 합니다.
최종 결과물로는 다양한 모델을 한 사이트에서 한 번에 테스트 해볼 수 있게 스트림릿으로 구현해볼까 합니다.
그 과정에서 Kling AI의 API를 사용해 보았고, 공식 문서 외에는 참고할만한 자료가 많지 않아 직접 경험을 공유하고자 이 글을 작성하게 되었습니다.
준비 사항
1. API 키 발급
Kling AI API를 이용하려면 기존 구독제와 별도로 구매해야 합니다.
저는 이미지 생성을 테스트해보기 위해 $2 요금제를 구매하였습니다.
자세한 요금제는 공식 사이트에서 확인할 수 있습니다.
https://klingai.com/dev-center
API를 구매하였다면 Access Key(AK) 및 Secret Key(SK)가 필요합니다.
아래 사이트에 접속하면 키를 발급받을 수 있습니다.
https://console.klingai.com/console/access-control/accesskey-management
2. 환경 설정 및 JWT 토큰 생성
필수 라이브러리를 설치한 후, API 요청을 위해 JWT 토큰을 생성해야 합니다.
라이브러리 설치
다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.
pip install requests pyjwt
API 인증 정보 및 환경 설정
import time
import jwt
import requests
# API 인증 정보
ACCESS_KEY = "YOUR_ACCESS_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"
def encode_jwt_token(ak, sk):
"""JWT 토큰 생성 함수"""
headers = {
"alg": "HS256",
"typ": "JWT"
}
payload = {
"iss": ak,
"exp": int(time.time()) + 1800, # 현재 시간 + 30분
"nbf": int(time.time()) - 5 # 현재 시간 - 5초
}
return jwt.encode(payload, sk, headers=headers)
3. 이미지 생성 요청
JWT 토큰을 사용하여 이미지를 생성할 수 있습니다. 아래 함수들은 텍스트 프롬프트를 기반으로 이미지를 생성하고, 생성이 완료될 때까지 대기한 뒤 결과를 가져옵니다.
def get_task_result(task_id, api_key):
"""태스크 결과 조회 함수"""
url = "https://api.klingai.com/v1/images/generations"
headers = {"Authorization": f"Bearer {api_key}"}
params = {"pageSize": 500}
response = requests.get(url, headers=headers, params=params).json()
for task in response["data"]:
if task["task_id"] == task_id and task["task_status"] == "succeed":
return task["task_result"]["images"]
return None
def wait_for_image(task_id, api_key, max_attempts=30, delay=2):
"""이미지 생성 완료 대기 함수"""
for _ in range(max_attempts):
images = get_task_result(task_id, api_key)
if images:
return images
time.sleep(delay)
raise TimeoutError("태스크 시간 초과")
4. 실행 및 이미지 다운로드
아래 코드로 이미지 생성을 요청하고 결과를 다운로드할 수 있습니다.
try:
# JWT 토큰 생성
authorization = encode_jwt_token(ACCESS_KEY, SECRET_KEY)
print("Generated Token:", authorization)
# API 엔드포인트
API_URL = "https://api.klingai.com/v1/images/generations"
# 헤더 설정
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {authorization}"
}
# 이미지 생성 요청 데이터
data = {
"model": "kling-v1",
"prompt": "person",
"negative_prompt": "",
"n": 1,
"width": 512,
"height": 512
}
# 이미지 생성 요청
print("\n이미지 생성 요청 중...")
response = requests.post(API_URL, headers=headers, json=data)
result = response.json()
print("API 응답:", result)
# 이미지 생성이 성공적으로 요청되었는지 확인
if result.get("code") == 0:
task_id = result["data"]["task_id"]
print(f"\n태스크 ID: {task_id}")
print("이미지 생성 대기 중...")
images = wait_for_image(task_id, authorization)
if images:
print("\n생성된 이미지 URL:")
for idx, img in enumerate(images, 1):
print(f"이미지 {idx}: {img['url']}")
# 이미지 다운로드
image_response = requests.get(img['url'])
if image_response.status_code == 200:
with open(f"generated_image_{idx}.png", "wb") as f:
f.write(image_response.content)
print(f"이미지 {idx} 저장 완료")
else:
print("이미지를 가져오는데 실패했습니다.")
else:
print(f"\n에러 발생: {result.get('message', '알 수 없는 에러')}")
except Exception as e:
print(f"\n오류 발생: {str(e)}")
전체 코드
아래는 위 단계를 모두 합친 최종 코드입니다.
import time
import jwt
import requests
# API 인증 정보
ACCESS_KEY = "YOUR_ACCESS_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"
def encode_jwt_token(ak, sk):
"""JWT 토큰 생성 함수"""
headers = {
"alg": "HS256",
"typ": "JWT"
}
payload = {
"iss": ak,
"exp": int(time.time()) + 1800, # 현재 시간 + 30분
"nbf": int(time.time()) - 5 # 현재 시간 - 5초
}
return jwt.encode(payload, sk, headers=headers)
def get_task_result(task_id, api_key):
"""태스크 결과 조회 함수"""
url = "https://api.klingai.com/v1/images/generations"
headers = {"Authorization": f"Bearer {api_key}"}
params = {"pageSize": 500}
response = requests.get(url, headers=headers, params=params).json()
for task in response["data"]:
if task["task_id"] == task_id and task["task_status"] == "succeed":
return task["task_result"]["images"]
return None
def wait_for_image(task_id, api_key, max_attempts=30, delay=2):
"""이미지 생성 완료 대기 함수"""
for _ in range(max_attempts):
images = get_task_result(task_id, api_key)
if images:
return images
time.sleep(delay)
raise TimeoutError("태스크 시간 초과")
try:
authorization = encode_jwt_token(ACCESS_KEY, SECRET_KEY)
print("Generated Token:", authorization)
API_URL = "https://api.klingai.com/v1/images/generations"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {authorization}"
}
data = {
"model": "kling-v1",
"prompt": "person",
"negative_prompt": "",
"n": 1,
"width": 512,
"height": 512
}
print("\n이미지 생성 요청 중...")
response = requests.post(API_URL, headers=headers, json=data)
result = response.json()
print("API 응답:", result)
if result.get("code") == 0:
task_id = result["data"]["task_id"]
print(f"\n태스크 ID: {task_id}")
print("이미지 생성 대기 중...")
images = wait_for_image(task_id, authorization)
if images:
print("\n생성된 이미지 URL:")
for idx, img in enumerate(images, 1):
print(f"이미지 {idx}: {img['url']}")
image_response = requests.get(img['url'])
if image_response.status_code == 200:
with open(f"generated_image_{idx}.png", "wb") as f:
f.write(image_response.content)
print(f"이미지 {idx} 저장 완료")
else:
print("이미지를 가져오는데 실패했습니다.")
else:
print(f"\n에러 발생: {result.get('message', '알 수 없는 에러')}")
except Exception as e:
print(f"\n오류 발생: {str(e)}")
마무리
이렇게 API를 통해 간단하게 Kling을 사용해 이미지를 생성하는 방법을 알아봤습니다.
다음에는 다른 모델 한 번 시도해보겠습니다리
반응형