Azure OpenAI Assistant - Part 3
Azure OpenAI를 사용하기 위한 서비스를 생성합니다.
먼저 각자의 Azure Open API Key와 Endpoint를 설정하고,
API Version은 Azure Open AI의 Assistant를 지원하는 API 버전인 2024-02-15-preview 버전을 사용합니다.
그리고, Assisant에서 호출할 함수를 2개를 만들 예정인 데,
그 중 하나가 Bing을 통한 검색 정보를 가져오는 함수입니다.
해당 함수 사용을 위해서 Bing 검색용 Key와 URL도 함께 설정합니다.
import os
os.environ["AZURE_OPENAI_ENDPOINT"] = AZURE_OPENAI_ENDPOINT 주소
os.environ["AZURE_OPENAI_API_KEY"] = AZURE_OPENAI_API_KEY 값
bing_search_subscription_key = Bing 검색 키 값
bing_search_url = "https://api.bing.microsoft.com/v7.0/search"
참고로, Bing 검색용 리소스는 Azure에서 아래와 같이 만들 수 있습니다.
구독, 리소스 그룹, 이름을 설정하고, 가격 책정 정책을 선택합니다.
초당 1회, 1달에 1천건까지 가능한 Bing 검색용 리소스는 F1으로 무료 사용이 가능합니다.
Bing 검색용 리소스를 생성하고, 키 및 엔드포인트 메뉴를 보면
아래와 같이 키와 엔드포인트 확인이 가능합니다.
다음으로 Assistant에서 사용 할 Function을 만듭니다.
첫 번째, 함수는 특정 지역의 시간을 확인하는 함수입니다.
시간을 확인 할 지역 정보를 Location으로 받아서, 해당 시간을 확인합니다.
import pytz
from datetime import datetime
def get_current_time(location):
try:
# Get the timezone for the city
timezone = pytz.timezone(location)
# Get the current time in the timezone
now = datetime.now(timezone)
current_time = now.strftime("%I:%M:%S %p")
return current_time
except:
return "Sorry, I couldn't find the timezone for that location."
두 번쨰 함수로 Bing 검색을 할 수 있는 함수를 만듭니다.
검색어를 문자열로 받아서, 검색한 내용에 대해서 리스트로 응답합니다.
import json
import requests
import time
from openai import AzureOpenAI
from pathlib import Path
from typing import Optional
from datetime import datetime
def search(query: str) -> list:
headers = {"Ocp-Apim-Subscription-Key": bing_search_subscription_key}
params = {"q": query, "textDecorations": False}
response = requests.get(bing_search_url, headers=headers, params=params)
response.raise_for_status()
search_results = response.json()
output = []
for result in search_results["webPages"]["value"]:
output.append({ "title": result["name"], "link": result["url"], "snippet": result["snippet"]})
return json.dumps(output)
이제 2개의 함수를 모두 만들고나면,
Assistant에서 사용 할 도구가 함수이므로, 'type'을 function으로 하고
Assistant에서 Function을 사용하기 위해서 Function 정의서를 작성합니다.
함수명과, 해당 함수가 어떤 함수 인지에 대해서 작성하고,
함수 호출 시 필요한 매개변수에 대한 내용을 지정합니다.
tools = [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "Get the current time in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The location name. The pytz is used to get the timezone for that location. Location names should be in a format like America/New_York, Asia/Bangkok, Europe/London",
}
},
"required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "search_bing",
"description": "Searches bing to get up-to-date information from the web.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query",
}
},
"required": ["query"],
},
},
}
]
Assistant에서 Function 실행 시에,
현재 Function에 대한 기능이 구현되어서 실행이 가능한 Fucntion인지 확인 하기 위해서
현재 사용 가능한 Function 리스트를 만듭니다.
available_functions = {
"get_current_time": get_current_time,
"search_bing": search
}
다음은
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")
)
이제 Azure OpenAI Assistant를 생성합니다.
앞서 구현한 Function에 맞춰서 적절한 Instructions을 선언합니다.
이번 포스팅에서는 Function Callilng을 사용하는 Assistant를 만들고 있기 때문에
tools에 'function' type으로 Function을 정의한 tools 배열로 설정합니다.
model은 Azure Open AI에서 Assistant를 지원하는 버전인 gpt-4-1106-preview 모델을 사용합니다.
assistant = client.beta.assistants.create(
name="다양한 정보를 제공해주는 Assistant",
instructions="""너는 세계 시간과 최신 정보를 검색해서 알려주는 AI Assistant 입니다.
검색을 사용한 경우에는 답변을 위해서 참고한 링크도 함께 알려주셔야 합니다.
모든 답변은 한국어로 합니다.""",
tools= tools,
model="gpt-4-1106-preview"
)
thread = client.beta.threads.create()
다음은 Assistant를 Function Calling을 위해 필요한 기능을 추가한 함수를 정의합니다.
Thread를 실행하고, run의 상태가 함수 호출 시에 전환되는 'requires_action'인지 확인 한 후,
사용 가능한 함수를 체크해서 함수를 호출합니다.
tool_calls 루프 내의 print는 실제 실행되는 내용들을 확인하기 위해서
부가적으로 넣어놓은 내용이라, print문은 모두 빠져 있어도 무방합니다.
def poll_run_till_completion(
client: AzureOpenAI,
thread_id: str,
run_id: str,
available_functions: dict,
verbose: bool,
max_steps: int = 20,
wait: int = 5,
) -> None:
if (client is None and thread_id is None) or run_id is None:
print("Client, Thread ID and Run ID are required.")
return
try:
cnt = 0
while cnt < max_steps:
run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
if verbose:
print("Poll {}: {}".format(cnt, run.status))
cnt += 1
if run.status == "requires_action":
tool_responses = []
if (
run.required_action.type == "submit_tool_outputs"
and run.required_action.submit_tool_outputs.tool_calls is not None
):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
for call in tool_calls:
print(f"Call Type : {type(call)}")
print(f"Call : {call}")
if call.type == "function":
if call.function.name not in available_functions:
raise Exception("Function requested by the model does not exist")
print(f"{call.function.name} 함수를 호출하고 있습니다")
print(f"function.name의 Type : {type(call.function.name)}")
print(f"인자 값으로 {call.function.arguments}을 사용합니다.")
print(f"인자 값 Type : {type(call.function.arguments)}")
function_to_call = available_functions[call.function.name]
tool_response = function_to_call(**json.loads(call.function.arguments))
tool_responses.append({"tool_call_id": call.id, "output": tool_response})
print(tool_responses)
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
)
if run.status == "failed":
print("Run failed.")
break
if run.status == "completed":
break
time.sleep(wait)
except Exception as e:
print(e)
참고로,
Run의 lifecyle은 아래의 도표 이미지와 같습니다.
Assistant에서 Function calling 도구 사용 시, 모델이 호출할 함수와 인수(argument)를 결정하면
Run의 상태가 in_progress에서 requires_action으로 이동합니다.
Run Cycle의 변화 모습은 앞서 정의한 함수를 실행한 뒷 부분에서 확인이 가능합니다.
Assistant에 질의 할 내용을 설정합니다.
먼저, 뉴욕에 시간을 묻는 질문입니다.
query = "What time is it in New York?"
thread에 'user' role로 뉴욕 시간을 묻는 query를 추가합니다.
해당 Thread를 앞서 설정한 Assistant로 시작하는 프로세스를 만듭니다.
Function Calling을 위해 필요한 기능을 추가한 함수인 poll_run_till_completion을 호출합니다.
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=query
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
poll_run_till_completion(
client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=True
)
run이 queued 상태에서, Function Calling을 위해서 requires_action 상태로 변경된 것을 볼 수 있고,
해당 Call Type과 Assistant가 호출할 함수가 'get_currnet_time' 인 것을 확인 할 수 있습니다.
그리고 사용자가 질의를 토대로 해당 함수 호출 시에 사용할 매개변수로 America/New_York 라는 값을 설정한 것을 볼 수 있습니다.
최종적으로 Function Calling에 대한 결괏 값으로 09:50:25 PM이 응답 받게 됩니다.
이제 run 상태가 completed 됐기 때문에 thread 내에 있는 메시지를 가져와서,
ChatGPT의 최종 응답 값인 마지막 메시지를 확인합니다.
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
result = messages.model_dump_json()
import json
result_dict = json.loads(result)
result_dict["data"][0]["content"][0]["text"]["value"]
답변 내용은 사용자의 질의 내용과 get_current_time 함수로 응답 받은 '09:50:25 PM' 값을 통해서
ChatGPT에서 답변한 내용이며, 아래와 같습니다.
다음은, Bing 검색을 통한 정보를 가져오는 함수를 확인해 보겠습니다.
질의 내용을 'nvidia 최신 AI Chip 아키텍처는?' 으로 바꿉니다.
query = "nvidia 최신 AI Chip 아키텍처는?"
새로운 질의를 Thread에 추가하고,
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=query
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
poll_run_till_completion(
client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=True
)
이번에는 Bing 함수를 호출하고 있고,
인자 값으로, Bing에서 검색 할 내용을 매개변수로 가져오는 데
최초 질의한 내용에 대한 검색 결과를 찾을 수 있도록, 'latest NVIDIA AI Chip architecture'로 된 것을 볼 수 있습니다.
이제 Bing 검색기를 통해서 확인한 내용을 기반으로 chatGPT가 작성한 메시지 확인을 위해서
Thread에서 메시지 리스트를 가져와서, 마지막 메시지를 확인합니다.
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
result = messages.model_dump_json()
import json
result_dict = json.loads(result)
print(result_dict["data"][0]["content"][0]["text"]["value"])
아래와 같이 NVIDIA의 최신 AI 칩에 대한 정보를 BING 검색 정보를 통해서 만들어 준 것을 확인 할 수 있습니다.
