[알토르] #7 openai assistants api를 사용해 이력서 기반 챗봇 만들기

Hyungjun·2024년 5월 19일

알토르

목록 보기

8/23

Mission

openai 어시스턴트 란?
이력서 샘플을 업로드해보고 테스트 해보기

배경지식

openai 어시스턴트 사용법
openai 사의 chatgpt를 넘어서서 나만의 assistant를 만들 수 있다.
1) openai playground는 assistant를 만들고 테스트 해볼 수 있는 유저 인터페이스다. 코드 없이 글과 버튼 몇개만 누르면 나만의 assistant를 만들 수 있다.
2) openai assistants API를 사용해서 코드 상에서 assistant를 생성할 수 있다.
openai 어시스턴트 API란?
애플리케이션 내에 AI 기반 어시스턴트를 만들 수 있는 toolkit이다. 단순히 사용자 질문에 대한 답을 하는 것에 넘어서서 코드 해석기(Code Interpreter), 검색(Retrieval), 기능 호출(Function calling)과 같은 Tool을 이용하여 AI의 상호작용 능력을 향상시킬 수 있다.

#1 Assistants API의 기본요소

assistants : 모델, instruction, tools, files를 캡슐화 한다.
threads : 하나의 대화 채널이다. message들이 이 안에 담긴다.
runs : assistant와 thread에서의 실행을 구동한다.

Tools

1) code interpreter
어시스턴트가 코드를 작성하고 실행해준다.

2) retrieval
제공된 문서를 기반으로 답변할 수 있게하는 기능.

3) functions
어시스턴트에게 사용자 정의 함수를 지정할 수 있다.

실습

#2 이력서 챗봇 만들기

retrieval 도구를 사용하여 내 이력서를 바탕으로 답변해주는 챗봇을 만들어 보자

과정

assistants를 만든다.
thread를 만든다.
Thread에 Message 추가: 사용자가 AI 어시스턴트에게 입력한 Message나 어시스턴트의 답변을 추가한다.
해당 Thread에서 어시스턴트 실행(Run): Thread에서 어시스턴트를 실행하여 사용자에게 제공할 답변을 생성한다. 이 과정에서 관련 도구들이 자동으로 호출된다.

코드

from openai import OpenAI
from dotenv import load_dotenv
import os
import json
import time


def display(obj):
  print(json.dumps(obj, indent=4))

def show_json(obj):
	display(json.loads(obj.model_dump_json()))

load_dotenv()


client = OpenAI(
	api_key = os.environ.get("OPENAI_API_KEY"),
)

file_ids=[
    "file-9DdWlanIt6JwTEuUTP3mHQGs",
]


assistant = client.beta.assistants.update(
    os.environ.get("ASSISTANT_ID"),
    tools=[
      {"type": "code_interpreter"},
      {"type": "file_search"},
	],
    instructions="You are a chatbot that provides information about Jun's resume.",
)

def create_new_thread():
    # 새로운 스레드를 생성합니다.
    thread = client.beta.threads.create()
    return thread



def wait_on_run(run, thread):
    # 주어진 실행(run)이 완료될 때까지 대기합니다.
    # status 가 "queued" 또는 "in_progress" 인 경우에는 계속 polling 하며 대기합니다.
    while run.status == "queued" or run.status == "in_progress":
        # run.status 를 업데이트합니다.
        run = client.beta.threads.runs.retrieve(
            thread_id=os.environ.get("THREAD_ID"),
            run_id=run.id,
        )
        # API 요청 사이에 잠깐의 대기 시간을 두어 서버 부하를 줄입니다.
        time.sleep(0.5)
    return run

def submit_message(assistant_id, thread_id, user_message):
    # 3-1. 스레드에 종속된 메시지를 '추가' 합니다.
    client.beta.threads.messages.create(
        thread_id=os.environ.get("THREAD_ID"), role="user", content=user_message
    )
    # 3-2. 스레드를 실행합니다.
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant_id,
    )
    return run

def get_response(thread_id):
    # 3-4. 스레드에 종속된 메시지를 '조회' 합니다.
    return client.beta.threads.messages.list(thread_id=thread_id, order="asc")

def print_message(response):
    for res in response:
        print(f"[{res.role.upper()}]\n{res.content[0].text.value}\n")

def ask(assistant_id, thread_id, user_message):
    run = submit_message(
        assistant_id,
        thread_id,
        user_message,
    )
    # 실행이 완료될 때까지 대기합니다.
    run = wait_on_run(run, thread_id)
    print_message(get_response(thread_id).data[-2:])
    return run

def upload_files(files):
    uploaded_files = []
    for filepath in files:
        file = client.files.create(
            file=open(
                # 업로드할 파일의 경로를 지정합니다.
                filepath,  # 파일경로. (예시) data/sample.pdf
                "rb",
            ),
            purpose="assistants",
        )
        uploaded_files.append(file)
        print(f"[업로드한 파일 ID]\n{file.id}")
    return uploaded_files

run = ask(
    os.environ.get("ASSISTANT_ID"),
    os.environ.get("THREAD_ID"),
    "한형준의 나이를 파일에서 검색하여 알려주세요. ",
)