Azure OpenAI 모델 연결

Azure에 배포한 OpenAI 채팅 모델을 외부 모델 연결로 감쌉니다. 이 예제는 API 키 인증과 AzureOpenAI 클라이언트를 사용합니다. Model Adapter Library에서 어댑터를 게시하고 태그를 만든 뒤 URL, 연결 설정과 암호화할 API 키를 입력합니다.

현재 OpenAI Python SDK를 사용할 때에는 Python 3.10 이상과 openai, pandas 및 palantir_models가 필요합니다. 각 모델 인스턴스에 별도 클라이언트를 두며 SDK의 전역 API 키나 주소를 바꾸지 않습니다.

설정 값

위치와 키	값
URL / base_url	Azure OpenAI 리소스의 HTTPS 엔드포인트입니다. 예: https://YOUR_RESOURCE.openai.azure.com. 배포별 호출 경로를 붙이지 않은 리소스 주소를 사용합니다.
Connection Configuration의 api_version	연결한 Azure API에서 지원하는 버전입니다.
Connection Configuration의 engine	Azure에 만든 채팅 모델 배포 이름입니다. SDK 호출의 model 인자로 전달합니다.
Credentials Configuration의 api_key	해당 Azure 리소스를 호출할 API 키입니다.

AzureOpenAI 클래스로 Azure 연결을 선택하므로 api_type을 별도 클라이언트 설정으로 전달하지 않습니다. 배포 이름과 기반 모델 이름은 같지 않을 수 있으므로 engine에는 실제 배포 이름을 넣습니다.

텍스트 입력과 출력

입력 df_in은 prompt 문자열 열을 갖습니다. 행마다 user 메시지를 만들어 채팅 완료 API를 호출하고 df_out의 prediction에 텍스트를 기록합니다. 이 예제는 텍스트 응답을 반환하는 배포를 대상으로 합니다. 텍스트가 없는 응답을 빈 성공값으로 대신하지 않습니다.

import logging
import openai
import palantir_models as pm

logger = logging.getLogger(__name__)

class OpenAIModelAdapter(pm.ExternalModelAdapter):
    def __init__(self, base_url, api_version, engine, api_key):
        self.engine = engine
        self.client = openai.AzureOpenAI(
            azure_endpoint=base_url,
            api_version=api_version,
            api_key=api_key,
        )

    @classmethod
    def init_external(cls, external_context):
        return cls(
            base_url=external_context.base_url,
            api_version=external_context.connection_config["api_version"],
            engine=external_context.connection_config["engine"],
            api_key=external_context.resolved_credentials["api_key"],
        )

    @classmethod
    def api(cls):
        return (
            {"df_in": pm.Pandas(columns=[("prompt", str)])},
            {"df_out": pm.Pandas(columns=[("prompt", str), ("prediction", str)])},
        )

    def predict(self, df_in):
        predictions = []
        for prompt in df_in["prompt"]:
            try:
                response = self.client.chat.completions.create(
                    model=self.engine,
                    messages=[{"role": "user", "content": prompt}],
                )
            except openai.APITimeoutError:
                logger.error("OpenAI API request timed out.")
                raise
            except openai.APIConnectionError:
                logger.error("OpenAI API request failed to connect.")
                raise
            except openai.BadRequestError:
                logger.error("OpenAI API request was invalid.")
                raise
            except openai.AuthenticationError:
                logger.error("OpenAI API request was not authorized.")
                raise
            except openai.PermissionDeniedError:
                logger.error("OpenAI API request was not permitted.")
                raise
            except openai.RateLimitError:
                logger.error("OpenAI API request exceeded a rate limit.")
                raise
            except openai.APIError:
                logger.error("OpenAI API returned an API error.")
                raise
            content = response.choices[0].message.content
            if content is None:
                raise ValueError("The chat response has no text content for prediction.")
            predictions.append(content)
        df_in["prediction"] = predictions
        return df_in

오류 처리

예외	확인할 조건
APITimeoutError	요청 시간 초과입니다.
APIConnectionError	서버 연결과 송신 경로를 확인합니다.
BadRequestError	메시지와 요청 설정을 확인합니다.
AuthenticationError	API 키와 인증 대상 리소스를 확인합니다.
PermissionDeniedError	호출 주체의 권한을 확인합니다.
RateLimitError	제공자의 요청 또는 사용량 한도를 확인합니다.
APIError	위에 구분하지 않은 API 오류이며 원래 예외를 호출자에게 전달합니다.