이는 Python SDK를 사용하여 OpenAI의 Assistant API로 에이전트를 구축하는 방법에 대한 여러 부분으로 구성된 시리즈 중 첫 번째 부분입니다.

에이전트란 무엇입니까?

내가 보기에 에이전트는 실제로 LLM(대형 언어 모델)을 활용하고 인간 행동을 모방하려는 소프트웨어의 일부일 뿐입니다. 이는 언어를 대화하고 이해할 수 있을 뿐만 아니라 현실 세계에 영향을 미치는 행동을 수행할 수도 있다는 것을 의미합니다. 이러한 작업을 일반적으로 도구라고 합니다.

이 블로그 게시물에서는 Python SDK를 사용하여 OpenAI의 Assistant API를 사용하여 에이전트를 구축하는 방법을 살펴보겠습니다. 파트 1은 단지 어시스턴트의 뼈대일 뿐입니다. 즉, 대화 부분입니다.

저는 프레임워크에 구애받지 않기 위해 의도적으로 CLI 앱을 구축하기로 결정했습니다. 우리는 의도적으로 구현을 에이전트라고 부르고 OpenAI SDK 구현을 보조자로 참조하여 둘을 쉽게 구별할 것입니다.

에이전트가 호출할 수 있는 기능에 관해서는 도구 와 기능 이라는 용어를 같은 의미로 사용합니다. 2부에서는 함수 호출에 대해 더 자세히 다룰 것입니다.

전제조건

이 튜토리얼을 진행하려면 다음이 필요합니다.

• 컴퓨터에 설치된 Python3
• OpenAI API 키
• Python 프로그래밍에 대한 기본 지식

OpenAI 어시스턴트 개념

Assistant : Assistants API의 Assistant는 사용자 메시지에 응답하도록 구성된 엔터티입니다. 지침, 선택한 모델 및 도구를 사용하여 기능과 상호 작용하고 답변을 제공합니다.

스레드 : 스레드는 Assistants API의 대화를 나타냅니다. 각 사용자 상호 작용에 대해 생성되며 진행 중인 대화의 컨테이너 역할을 하는 여러 메시지를 포함할 수 있습니다.

메시지 : 메시지는 스레드의 통신 단위입니다. 여기에는 텍스트(및 잠재적으로 향후 파일)가 포함되어 있으며 스레드 내에서 사용자 쿼리 또는 보조 응답을 전달하는 데 사용됩니다.

Run : Run은 스레드를 처리하는 Assistant의 인스턴스입니다. 여기에는 스레드 읽기, 도구 호출 여부 결정, 스레드 메시지에 대한 모델 해석을 기반으로 응답 생성이 포함됩니다.

개발 환경 설정

첫 번째 단계는 venv 사용하여 가상 환경을 만들고 활성화하는 것입니다. 이렇게 하면 종속성이 시스템 Python 설치로부터 격리됩니다.

 python3 -m venv venv source venv/bin/activate

유일한 종속성인 openai 패키지를 설치해 보겠습니다.

 pip install openai

main.py 파일을 만듭니다. CLI 앱에 대한 몇 가지 기본 런타임 논리를 채워 보겠습니다.

 while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the assistant...") break print(f"Assistant: You said {user_input}")

python3 main.py 실행하여 시도해 보세요.

 python3 main.py User: hi Assistant: You said hi

보시다시피 CLI는 사용자 메시지를 입력으로 받아들이고 우리의 천재 어시스턴트는 아직 두뇌🧠가 없으므로 메시지를 바로 반복합니다. 아직은 그렇게 똑똑하지 않습니다.

에이전트

이제 재미 😁 (또는 두통 🤕)이 시작됩니다. 지금 바로 최종 클래스에 필요한 모든 가져오기를 제공할 것이므로 간결성을 위해 코드 샘플에서 가져오기를 유지했기 때문에 항목이 어디서 오는지 고민하지 않아도 됩니다. 새 파일 agent.py Agent 클래스를 작성하는 것부터 시작하겠습니다.

 import time import openai from openai.types.beta.threads.run import Run class Agent: def __init__(self, name: str, personality: str): self.name = name self.personality = personality self.client = openai.OpenAI(api_key="sk-*****") self.assistant = self.client.beta.assistants.create( name=self.name, model="gpt-4-turbo-preview" )

클래스 생성자에서 OpenAI API 키를 전달하여 OpenAI 클라이언트를 클래스 속성으로 초기화합니다. 다음으로 새로 생성된 어시스턴트에 매핑되는 assistant 클래스 속성을 만듭니다. 나중에 사용할 수 있도록 name 과 personality 클래스 속성으로 저장합니다.

create 메소드에 전달하는 name 인수는 OpenAI 대시보드에서 Assistant를 식별하기 위한 것일 뿐이며 AI는 이 시점에서 실제로 이를 인식하지 못합니다. 실제로 나중에 보게 될 instructions 에 이름을 전달해야 합니다.

어시스턴트를 생성할 때 이미 instructions 설정할 수 있지만 실제로는 어시스턴트가 동적 변경에 덜 유연하게 됩니다.

client.beta.assistants.update 호출하여 Assistant를 업데이트할 수 있지만 Runs에 도달할 때 보게 될 동적 값을 전달하는 더 좋은 곳이 있습니다.

달리기를 생성할 때 여기에 instructions 전달한 다음 다시 실행하면 보조자의 instructions 실행 instructions 으로 덮어쓰여진다는 점에 유의하십시오. 이들은 서로를 보완하지 않으므로 필요에 따라 하나를 선택하십시오. 정적 지침의 경우 보조 수준, 동적 지침의 경우 실행 수준입니다.

모델로는 이 시리즈의 2부에서 함수 호출을 추가할 수 있도록 gpt-4-turbo-preview 모델을 선택했습니다. 도구를 구현할 때 순수한 좌절의 편두통을 겪으면서도 몇 푼도 절약하고 싶다면 gpt-3.5-turbo 사용할 수 있습니다.

GPT 3.5는 도구 호출에 있어서 형편없습니다. 내가 그것을 처리하려고 노력하면서 잃어버린 시간은 내가 그렇게 말할 수 있게 해줍니다. 😝 그건 이쯤 하고 나중에 더 자세히 다루겠습니다.

스레드 생성, 메시지 추가 및 마지막 메시지 검색

에이전트를 생성한 후에는 대화 스레드를 시작해야 합니다.

 class Agent: # ... (rest of code) def create_thread(self): self.thread = self.client.beta.threads.create()

그리고 우리는 해당 스레드에 메시지를 추가하는 방법을 원할 것입니다.

 class Agent: # ... (rest of code) def add_message(self, message): self.client.beta.threads.messages.create( thread_id=self.thread.id, role="user", content=message )

현재로서는 user 역할이 있는 메시지만 추가할 수 있습니다. 저는 OpenAI가 향후 릴리스에서 이를 변경할 계획이라고 생각합니다. 이는 상당히 제한적이기 때문입니다.

이제 스레드의 마지막 메시지를 얻을 수 있습니다.

 class Agent: # ... (rest of code) def get_last_message(self): return self.client.beta.threads.messages.list( thread_id=self.thread.id ).data[0].content[0].text.value

다음으로, 지금까지 가지고 있는 것을 테스트하기 위해 진입점 run_agent 메소드를 생성합니다. 현재 run_agent 메소드는 스레드의 마지막 메시지만 반환합니다. 실제로 실행을 수행하지는 않습니다. 아직 정신이 없는 상태입니다.

 class Agent: # ... (rest of code) def run_agent(self): message = self.get_last_message() return message

main.py 로 돌아가서 에이전트와 첫 번째 스레드를 만듭니다. 스레드에 메시지를 추가합니다. 그런 다음 동일한 메시지를 사용자에게 다시 반환합니다. 단, 이번에는 해당 라이브 스레드에서 전송됩니다.

 from agent import Agent agent = Agent(name="Bilbo Baggins", personality="You are the accomplished and renowned adventurer from The Hobbit. You act like you are a bit of a homebody, but you are always up for an adventure. You worry a bit too much about breakfast.") agent.create_thread() while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the agent...") break agent.add_message(user_input) answer = agent.run_agent() print(f"Assistant: {answer}")

실행해보자:

 python3 main.py User: hi Assistant: hi

아직은 그다지 똑똑하지 않습니다. 호빗보다는 앵무새🦜에 더 가깝습니다. 다음 섹션에서는 진짜 재미가 시작됩니다.

실행 생성 및 폴링

실행을 생성할 때 정기적으로 Run 개체를 검색하여 실행 상태를 확인해야 합니다. 이것을 폴링(Polling)이라고 하는데, 정말 짜증납니다. 에이전트가 다음에 수행해야 할 작업을 결정하려면 폴링해야 합니다. OpenAI는 이를 더욱 단순화하기 위해 스트리밍 지원을 추가할 계획입니다. 그동안 다음 섹션에서 폴링을 설정하는 방법을 보여 드리겠습니다.

메서드가 내부용으로 사용되며 외부 코드에서 직접 액세스해서는 안 됨을 나타내는 Python의 표준인 다음 메서드 이름의 _ 에 유의하세요.

먼저 Run 생성을 위한 도우미 메서드 _create_run 만들고 run_agent 업데이트하여 이 메서드를 호출해 보겠습니다.

 class Agent: # ... (rest of code) def get_breakfast_count_from_db(self): return 1 def _create_run(self): count = self.get_breakfast_count_from_db() return self.client.beta.threads.runs.create( thread_id=self.thread.id, assistant_id=self.assistant.id, instructions=f""" Your name is: {self.name} Your personality is: {self.personality} Metadata related to this conversation: {{ "breakfast_count": {count} }} """, ) def run_agent(self): run = self._create_run() # add this line message = self.get_last_message() return message

실행을 생성하기 위해 thread.id 및 assistant.id 전달하는 방법에 주목하세요.

처음에 동적 명령과 데이터를 전달할 더 좋은 장소가 있다고 말한 것을 기억하십니까? 이는 실행을 생성할 때 instructions 매개변수가 됩니다. 우리의 경우 데이터베이스에서 아침 식사 count 가져올 수 있습니다. 이렇게 하면 답변을 트리거할 때마다 다양한 관련 동적 데이터를 쉽게 전달할 수 있습니다.

이제 에이전트는 주변 세계의 변화를 인식하고 이에 따라 조치를 취할 수 있습니다. 나는 관련 동적 컨텍스트를 유지하는 지침에 메타데이터 JSON 개체를 갖고 싶습니다. 이를 통해 덜 장황하면서도 LLM이 잘 이해할 수 있는 형식으로 데이터를 전달할 수 있습니다.

아직 실행하지 마세요. 마지막 메시지를 받을 때 실행이 완료되기를 기다리지 않기 때문에 작동하지 않으므로 여전히 마지막 사용자 메시지가 됩니다.

폴링 메커니즘을 구축하여 이 문제를 해결해 보겠습니다. 먼저, 실행을 반복적이고 쉽게 검색할 수 있는 방법이 필요하므로 _retrieve_run 메서드를 추가해 보겠습니다.

 class Agent: # ... (rest of code) def _retrieve_run(self, run: Run): return self.client.beta.threads.runs.retrieve( run_id=run.id, thread_id=self.thread.id)

특정 실행을 찾기 위해 run.id 와 thread.id 모두 전달해야 하는 방법에 주목하세요.

Agent 클래스에 _poll_run 메소드를 추가합니다:

 class Agent: # ... (rest of code) def _cancel_run(self, run: Run): self.client.beta.threads.runs.cancel( run_id=run.id, thread_id=self.thread.id) def _poll_run(self, run: Run): status = run.status start_time = time.time() while status != "completed": if status == 'failed': raise Exception(f"Run failed with error: {run.last_error}") if status == 'expired': raise Exception("Run expired.") time.sleep(1) run = self._retrieve_run(run) status = run.status elapsed_time = time.time() - start_time if elapsed_time > 120: # 2 minutes self._cancel_run(run) raise Exception("Run took longer than 2 minutes.")

🥵 휴, 그거 많구나... 풀어보자.

_poll_run Run 객체를 인수로 받아 현재 Run status 추출합니다. 사용 가능한 모든 상태는 OpenAI 문서 에서 확인할 수 있습니다. 현재 목적에 맞는 몇 가지만 사용하겠습니다.

이제 몇 가지 오류 시나리오를 처리하면서 완료된 상태를 확인하기 위해 while 루프를 실행합니다. Assistant API의 실제 청구는 약간 어둡기 때문에 안전을 위해 2분 후에 실행을 취소하기로 결정했습니다.

OpenAI가 10분 후에 실행을 취소하는 경우 expired 상태가 있더라도 마찬가지입니다. 실행하는 데 2분 이상 걸리면 어쨌든 문제가 있을 수 있습니다.

또한 몇 밀리초마다 폴링하고 싶지 않기 때문에 2분 표시에 도달하고 실행을 취소할 때까지 1초마다 폴링하여 요청을 제한합니다. 당신이 적합하다고 생각하는대로 이것을 조정할 수 있습니다.

지연 후 반복할 때마다 실행 상태를 다시 가져옵니다.

이제 모든 것을 run_agent 메소드에 연결해 보겠습니다. 먼저 _create_run 사용하여 실행을 생성한 다음 응답을 얻거나 오류가 발생할 때까지 _poll_run 사용하여 폴링하고, 마지막으로 폴링이 완료되면 이제 에이전트에서 전송될 스레드에서 마지막 메시지를 검색합니다.

그런 다음 메시지를 런타임 루프로 반환하여 사용자에게 다시 보낼 수 있습니다.

 class Agent: # ... (rest of code) def run_agent(self): run = self._create_run() self._poll_run(run) # add this line message = self.get_last_message() return message

짜잔, 이제 에이전트를 다시 실행하면 친절한 에이전트로부터 다음과 같은 답장을 받게 됩니다.

 python3 main.py User: hi Assistant: Hello there! What adventure can we embark on today? Or perhaps, before we set out, we should think about breakfast. Have you had yours yet? I've had mine, of course – can't start the day without a proper breakfast, you know. User: how many breakfasts have you had? Assistant: Ah, well, I've had just 1 breakfast today. But the day is still young, and there's always room for a second, isn't there? What about you? How can I assist you on this fine day?

2부에서는 에이전트가 도구를 호출하는 기능을 추가할 것입니다.

내 GitHub 에서 전체 코드를 찾을 수 있습니다.

읽어주셔서 감사합니다. 댓글로 어떤 생각이나 피드백이라도 듣고 싶습니다. 다음과 같은 더 많은 콘텐츠를 보려면 Linkedin에서 저를 팔로우하세요: https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/