Gemini API 병렬처리 사용 예제

vertexai 1.43.0

 

Gemini API를 사용하여 코드상에서 응답을 받을 수 있는 예제입니다. 결제계정 연결은 구글 vertexai를 참고하시면 좋을 것 같습니다.

 

먼저 1개씩 처리하는 코드입니다.

import os
from vertexai.preview.generative_models import GenerativeModel, GenerationConfig
from vertexai import generative_models

def Gemini_call():
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "******"
    gemini_vision = GenerativeModel("gemini-1.5-pro")
    gemini_config = GenerationConfig(temperature=0.3)

    prompt = "안녕 Gemini, 구글이라는 회사에 대해 설명해줘."
    prompt = [prompt]

    # 프롬프트에 넣을 이미지가 있을 경우 경로로 리스트를 만들어 놓을 수 있음.
    # for img_path in img_path_list:
    #     prompt.append(img_path)
    
    # API call
    try:
        safety_config = {
                generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_NONE,
        }
        response = gemini_vision.generate_content(prompt, generation_config=gemini_config, safety_settings=safety_config)
        gemini_pred = response.text

    # api error 예외 처리
    except Exception as api_error:
        response = None
        print(f"gemini api error: {api_error}")
        gemini_pred = "gemini_api_error"

    print(gemini_pred)
    

if __name__ == '__main__':
    Gemini_call()

 

그러면 아웃풋이 아래와 같이 나옵니다.

 

다음으로 multi-processing 코드입니다. 프롬프트를 리스트로 넣었습니다. max_workers 숫자 수정으로 프로세서가 수용가능한 범위에서, Gemini API 한도 범위 내에서 원하는 만큼 던질 수 있습니다.

import os
from vertexai.preview.generative_models import GenerativeModel, GenerationConfig
from vertexai import generative_models
import concurrent.futures

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "******"
gemini_vision = GenerativeModel("gemini-1.5-pro")
gemini_config = GenerationConfig(temperature=0.3)

def Gemini_call(prompt):
    prompt = [prompt]

    # 프롬프트에 넣을 이미지가 있을 경우 경로로 리스트를 만들어 놓음
    # for img_path in img_path_list:
    #     prompt.append(img_path)
    
    # API call
    try:
        safety_config = {
                generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_NONE,
                generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_NONE,
        }
        response = gemini_vision.generate_content(prompt, generation_config=gemini_config, safety_settings=safety_config)
        gemini_pred = response.text

    # api error 예외 처리
    except Exception as api_error:
        response = None
        print(f"gemini api error: {api_error}")
        gemini_pred = "gemini_api_error"

    return gemini_pred


def multi_process():
    prompts = [
        "구글이라는 회사에 대해 100자 이내로 설명해줘.",
        "OpenAI라는 회사에 대해 100자 이내로 설명해줘."
    ]
    # max_workers에 병렬처리를 어느 수준으로 할지 지정
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(Gemini_call, x) for x in prompts]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]
    
    print(results)

if __name__ == '__main__':
    multi_process()

 

병렬처리 아웃풋

 

 

 

도움이 되셨다면 아무 광고나 클릭 한 번 부탁드립니다👍