54 lines
2.5 KiB
Python
54 lines
2.5 KiB
Python
from pydantic.main import BaseModel
|
|
|
|
from app.utils.prompts.chatgpt_prompt import ChatgptService
|
|
from app.utils.prompts.prompts import image_autotag_prompt
|
|
from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommended
|
|
|
|
import asyncio
|
|
|
|
# medium 추론은 출력 비용의 80%를 차지하고, minimal은 A/B 비교(4회)에서 narrative 점수가
|
|
# welcome 단계로 편향되고 태그를 과다 선택하는 패턴이 반복돼 low로 고정한다.
|
|
IMAGE_TAG_REASONING_EFFORT = "low"
|
|
|
|
async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_list
|
|
chatgpt = ChatgptService(model_type="gpt")
|
|
image_input_data = {
|
|
"img_url" : image_url,
|
|
"industry" : industry,
|
|
"space_type" : list(SpaceType),
|
|
"subject" : list(Subject),
|
|
"camera" : list(Camera),
|
|
"motion_recommended" : list(MotionRecommended)
|
|
}
|
|
|
|
image_result = await chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_url, True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT)
|
|
return image_result
|
|
|
|
async def autotag_images(image_url_list : list[str], industry: str = "") -> list[dict]: #tag_list
|
|
chatgpt = ChatgptService(model_type="gpt")
|
|
image_input_data_list = [{
|
|
"img_url" : image_url,
|
|
"industry" : industry,
|
|
"space_type" : list(SpaceType),
|
|
"subject" : list(Subject),
|
|
"camera" : list(Camera),
|
|
"motion_recommended" : list(MotionRecommended)
|
|
}for image_url in image_url_list]
|
|
|
|
image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], True, silent = True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT) for image_input_data in image_input_data_list]
|
|
image_result_list: list[BaseModel | BaseException] = await asyncio.gather(*image_result_tasks, return_exceptions=True)
|
|
MAX_RETRY = 2
|
|
for _ in range(MAX_RETRY):
|
|
failed_idx = [i for i, r in enumerate(image_result_list) if isinstance(r, Exception)]
|
|
# print("Failed", failed_idx)
|
|
if not failed_idx:
|
|
break
|
|
retried = await asyncio.gather(
|
|
*[chatgpt.generate_structured_output(image_autotag_prompt, image_input_data_list[i], image_input_data_list[i]['img_url'], True, silent=True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT) for i in failed_idx],
|
|
return_exceptions=True
|
|
)
|
|
for i, result in zip(failed_idx, retried):
|
|
image_result_list[i] = result
|
|
|
|
# print("Failed", failed_idx)
|
|
return image_result_list |