Azure OpenAI를 사용하기 위한 서비스를 생성합니다.
먼저 각자의 Azure Open API Key와 Endpoint를 설정하고,
API Version은 Azure Open AI의 Assistant를 지원하는 API 버전인 2024-02-15-preview 버전을 사용합니다.
import os
import time
import json
from openai import AzureOpenAI
client = AzureOpenAI(
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
api_version="2024-02-15-preview", #Azure OpenAI Assistants를 위한 API 버전
azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
)
다음은 Assistant를 생성합니다.
본 포스팅에서 'AI 수학 도우미2'라는 Code Interpreter를 사용하는 Assistant를 만듭니다.
수학 도우미를 위한 적절헌 Instructions를 입력하고,
model은 Azure Open AI에서 Assistant를 지원하는 버전인 gpt-4-1106-preview 모델을 사용합니다.
참고로, 현재 Azure Open AI에서 Assistant 기능은 Preview이기 때문에
패키지에 beta라고 포함되어 있으며, 향후 GA가 된 이후에는 beta가 빠진 패키지 명이 될 것으로 생각됩니다.
# assistant 생성
assistant = client.beta.assistants.create(
name="AI 수학 도우미2",
instructions="당신은 수학 문제에 답하는 데 도움이 되는 코드를 작성할 수 있는 AI 도우미입니다..",
tools=[{"type": "code_interpreter"}],
# tools를 dict형태의 Assistant type을 배열 내에 다수 등록 가능.
model="gpt-4-1106-preview"
)
다음은 Assistant와 사용자 간의 대화 세션인 Thread를 생성합니다.
일종의 대화를 주고 받는 방이라고 보시면 됩니다.
그리고, Thread에 실제 대화 내용인 Message를 추가합니다.
Message를 추가 할 때의 누구의 메시지인지를 위한 role과 실제 대화내용인 content를 넣고,
이 대화가 어떤 Assistant와 이뤄질지를 위해서 thread.id를 지정합니다.
즉, Message를 분리하여 다양한 Assistant와 대화 히스토리를 이어나갈 수도 있습니다.
# thread 생성
thread = client.beta.threads.create()
content = "어떤 정사각형의 가로를 3cm, 세로를 2cm만큼 늘였더니 넓이가 처음의 2배가 되었다. 처음 정사각형의 한 변의 길이는?"
# thread에 사용자 message를 추가
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=content
)
Thread를 실행하여, Assistant와 대화를 시작하게 합니다.
"어떤 정사각형의 가로를 3cm, 세로를 2cm만큼 늘였더니 넓이가 처음의 2배가 되었다.
처음 정사각형의 한 변의 길이는?"
라는 사용자 대화에 답을 하기 위해 Code Interpreter가 해당 문제를 풀기 위한
코드를 작성하여, 답을 하게 됩니다.
전체 동작이 완료까지 시간이 걸리기 때문에 run.status로 상태 값을 확인하고,
완료(completed) 될 때까지 기다립니다.
# thread를 실행한 Assistant 활성화
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
status = run.status
# Wait till the assistant has responded
while status not in ["completed", "cancelled", "expired", "failed"]:
time.sleep(5)
run = client.beta.threads.runs.retrieve(thread_id=thread.id,run_id=run.id)
status = run.status
Assistant의 답변을 확인하기 위해서 Thread에서 message list를 가져옵니다.
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
전체 message list를 json 형태로 변경해서, dictionary 타입으로 가져옵니다.
message list에는 기존 대화 내역들이 기록되는 데,
앞서 문자를 질의한 사용자의 대화가 있고, Assistant의 답변이 있으면
가장 최근 대화가 배열의 앞에 위치하게 됩니다.
따라서, result_dict["data"][0]을 가져옵니다.
실제 답변을 뽑기 위해서 위치한 내용만 출력합니다.
result = messages.model_dump_json()
import json
result_dict = json.loads(result)
print(result_dict["data"][0]["content"][0]["text"]["value"])
출력결과는 다음과 같습니다. (colab에서 동작 시킨 코드 포함)
실제 Code Interpreter가 어떤 코드를 작성해서 문제를 풀었는지를 확인 할 수도 있습니다.
message list에 문제에 사용된 코드를 요청하는 사용자 메시지를 추가합니다.
그리고, 다시 대화를 시작합니다.
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="이 문제를 풀기 위해서 사용한 코드를 알려줘"
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
status = run.status
# Wait till the assistant has responded
while status not in ["completed", "cancelled", "expired", "failed"]:
time.sleep(5)
run = client.beta.threads.runs.retrieve(thread_id=thread.id,run_id=run.id)
status = run.status
Assistant의 답변을 모두 완료하여 다시, 메시지 리스트를 가져와서
Assistant의 답변인 가장 최근의 메시지를 가져와서 출력합니다..
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
data = json.loads(messages.model_dump_json(indent=2))
code = data['data'][0]['content'][0]['text']['value']
print(code)
출력한 결과 값은 아래와 같습니다.
최초에 남긴 사용자 질문을 풀기 위한 Python 코드를 확인 할 수 있습니다.
실제 최초 질문을 남기면 답변을 위해서 이러한 Python 코드를 작성하고
Python 코드로 실행한 결과 값을 도출하고 이를 토대로 답변까지 작성하게 되는 것입니다.
이러한 코드 작성을 통해 답변을 생성할 수 있게 되면서,
문제를 푸는 것 뿐만 아니라, 주어진 데이터 분석 및 분석된 데이터에 그래프 생성 등의
다양한 역할을 할 수 있게 도와줍니다.
