최근 인공지능 분야에서 대규모 언어 모델(LLM), 비전 언어 모델 등과 더불어 AI 에이전트가 폭발적으로 성장하고 있습니다. 전통적인 규칙 기반 애플리케이션과 달리, 이러한 에이전트는 복잡한 작업을 자율적으로 또는 최소한의 사용자 개입만으로 수행하기 위해 추론 및 의사 결정 기능을 활용할 수 있습니다. 이를 위해 AI 에이전트는 종종 코드를 동적으로 생성하고 실행하거나 심지어 전체 가상 머신을 제어해야 하며, Novita AI 샌드박스는 이러한 기능을 제공합니다.
이 튜토리얼에서는 데이터 분석가 역할을 할 AI 에이전트를 생성할 것입니다. 인간 분석가처럼, 에이전트는 지시에 따라 브라우저를 사용하여 데이터셋을 찾고 다운로드할 수 있으며, 요청 시 분석, 시각화, 코드를 실행하여 도출된 인사이트를 사용자에게 제공할 수 있습니다.
AI 에이전트란 무엇인가?
AI 에이전트는 특정 목표를 달성하기 위해 인공지능을 사용하여 추론하고, 계획을 세우고, 행동을 수행하는 소프트웨어 프로그램입니다. 이러한 에이전트는 정보를 수집하거나 작업을 수행하기 위해 API, 코드 실행, 웹 검색 등의 도구를 사용할 수 있습니다. 때로는 Deep Research와 같은 더 복잡한 목표를 위해 다른 에이전트와 협력하기도 합니다.
AI 에이전트의 일반적인 구성 요소:
- 추론/계획: 목표가 주어지면, 에이전트는 이를 달성하기 위한 단계별 계획을 생성합니다. 이 과정은 하위 작업을 완료하고 다음 행동을 안내할 추가 정보를 수집하는 것을 포함할 수 있습니다. 이는 특히 완료하는 데 시간이 오래 걸리는 복잡한 작업에서 흔히 볼 수 있으며, 추론 모델이 이러한 상황에서 잘 작동합니다.
- 도구 사용: AI 에이전트는 외부 도구 및 서비스를 호출할 수 있습니다. 여기에는 기능, API 또는 기능을 확장하는 기타 리소스가 포함될 수 있습니다.
- 조정: 여러 에이전트가 책임을 공유하여 함께 작업할 수 있으며, 각각은 공유된 복잡하거나 장기적인 목표를 달성하는 데 기여합니다.
Novita AI 샌드박스, LLM 제품 및 브라우저 사용 개요
샌드박스란 무엇이며 왜 중요한가?
샌드박스는 신뢰할 수 없는 코드를 호스트 시스템에 영향을 주지 않고 실행할 수 있는 안전하고 격리된 런타임 환경입니다. 기본적으로 AI 에이전트가 코드, 명령을 실행하고, 파일을 생성하는 등의 작업을 할 수 있는 가벼운 가상 컴퓨터입니다.
Novita AI는 에이전트가 필요에 따라 빠르게 액세스할 수 있도록 클라우드에서 이 샌드박스를 제공하며, 사용된 리소스에 따라 유연한 초 단위 과금을 적용합니다.
Novita 샌드박스의 주요 기능:
- 보안 격리: 각 샌드박스는 격리된 파일 시스템과 환경을 가지므로 데이터가 보호되고 의도하지 않은 상호 작용이 방지됩니다.
- 빠른 시작: 샌드박스 인스턴스는 평균 약 200ms 이내에 시작되므로, 저지연 시나리오에 이상적입니다.
- 다중 언어 지원: Python, JavaScript, TypeScript 등 여러 프로그래밍 언어로 코드를 실행할 수 있습니다.
- 빠른 일시 중지 및 재개: 언제든지 샌드박스를 일시 중지했다가 필요할 때 재개할 수 있으며, 파일 시스템 및 프로세스 상태가 완전히 복원됩니다.
- 백그라운드 실행: 백그라운드 작업 실행을 지원하며, 결과를 기다려야 하는 시나리오에 적합합니다.
Novita 모델 API:
Novita는 OpenAI, Google, DeepSeek, Qwen 등 주요 연구실의 방대한 오픈 소스 AI 모델 라이브러리를 제공합니다. 여기에는 언어, 비전, 오디오, 비디오 및 임베딩 모델이 포함됩니다. 당사의 언어 모델은 OpenAI SDK와 완전히 호환되므로, OpenAI에서 Novita로 전환하려면 클라이언트의 기본 URL과 API 키만 업데이트한 다음 Novita 모델을 선택하면 됩니다.
| from openai import OpenAI client = OpenAI( base_url=“https://api.novita.ai/v3/openai”, api_key=“ ) |
브라우저 사용 개요:
Browser Use는 자연어 명령(예: “오늘 뉴욕의 날씨를 확인하세요”)을 사용하여 AI 에이전트가 웹 브라우저와 상호 작용할 수 있도록 하는 오픈 소스 Python 라이브러리입니다. 복잡한 규칙 기반 선택자를 작성하는 대신 AI가 요소 찾기 및 상호 작용을 처리합니다. 또한 Novita의 OpenAI 호환 모델 API를 사용하면 어떤 LLM 또는 VLM(비전 언어 모델)을 사용하여 Browser Use를 구동할 수 있습니다.
개발 환경 설정
시작하려면 GitHub 리포지토리를 복제하고, 깨끗한 Python 환경을 설정하고, 모든 필수 종속성을 설치하고, Novita AI 키를 가져오겠습니다.
uv를 사용하여 GitHub 리포지토리 복제 및 종속성 설치
1. uv (가벼운 Python 패키지 관리자) 설치
| pip install uv |
2. 리포지토리(GitHub 리포지토리)를 복제하고 해당 디렉토리로 이동합니다.
| git clone https://github.com/Studio1HQ/AI-sandbox.git cd AI-sandbox |
3. uv 가상 환경 생성 및 활성화
| # 가상 환경 생성 uv venv # 가상 환경 활성화 source .venv/bin/activate # Mac/Linux의 경우 # 또는 .venv\Scripts\activate # Windows의 경우 |
4. 프로젝트 종속성 설치
| # 종속성 설치 uv sync |
Novita AI 계정 생성 및 API 키 획득
1. novita.ai에 가입합니다.
2. 대시보드에서 사용자 프로필 아이콘 위에 마우스를 올린 다음 팝업에서 API Keys를 클릭합니다.
3. 키 관리 페이지에서 Add New Key를 클릭합니다. 팝업에서 키 이름을 입력하고 Confirm을 클릭한 다음 생성된 키를 복사합니다.
4. 이제 프로젝트 디렉토리 내에서 .env 파일을 생성하고 아래 내용을 붙여넣습니다.
| NOVITA_API_KEY=“<여기에 NOVITA API 키를 붙여넣으세요>” NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1” |
Novita AI 계정에 크레딧을 추가하세요.
Novita 샌드박스를 사용하려면 계정에 크레딧을 추가해야 합니다. 대시보드 탭에서 **‘Billing’**을 클릭합니다. 그 다음 청구 페이지에서 결제 수단을 추가하고 최소 $10의 크레딧을 충전합니다.
탐색적 데이터 분석(EDA) 에이전트 구축
이제 환경이 설정되고 API 키를 확보했으니, 에이전트를 시작해 보겠습니다.
에이전트 브라우저 사용을 통한 데이터셋(또는 파일) 다운로드:
인간 분석가처럼, 에이전트가 지시를 받아 브라우저를 사용하여 데이터셋을 가져올 수 있기를 원합니다. 따라서 browser_agent.py에서 볼 수 있듯이, 먼저 에이전트의 최종 출력을 위한 Pydantic 데이터 구조를 생성합니다. 이 구조는 다운로드된 파일 이름과 완료된 작업 결과를 파일에 작성하는 데 사용됩니다.
| … # 기존 코드 아래 class TaskFile(BaseModel): “”“스크랩된 데이터나 조사된 데이터 등 작업 결과로 생성된 파일을 나타냅니다.”“” filename: str = Field(…, description=“확장자를 포함한 파일 이름”) content: str = Field(…, description=“파일에 작성할 텍스트 내용”) class AgentOutput(BaseModel): “”“브라우저 에이전트 실행의 최종 집계 출력입니다.”“” downloaded_files: Optional[list[str]] = Field( None, description=“다운로드된 파일 이름 목록(확장자 포함), 해당하는 경우” ) task_files: Optional[list[TaskFile]] = Field( None, description=“사용자 작업에서 생성된 파일(예: 스크랩 또는 조사된 데이터), 해당하는 경우” ) |
아래는 에이전트 브라우저가 작동하는 방식의 샘플입니다. 브라우저 웹 탐색에는 대규모 언어 모델(LLM)을 사용할 것이며, 이는 대부분의 사용 사례를 커버할 수 있습니다. 그러나 캡차 해결과 같은 더 복잡한 UI 작업의 경우 비전 언어 모델(VLM)을 사용하는 것이 권장됩니다. VLM을 사용할 때는 vision detail 수준(‘high’, ‘low’, ‘auto’)을 설정하여 토큰 비용과 비전 명확성의 균형을 맞출 수 있습니다.
그 다음 브라우저 세션을 생성하고, 다운로드 경로 (./Download) 및 user_data_dir를 구성 프로필에 설정합니다(user_data_dir는 인코그니토 모드로 설정하려면 None으로 설정). 그리고 에이전트 출력을 구조화하기 위해 Pydantic 모델을 컨트롤러로 설정합니다. 그 다음 **await agent.run()**으로 에이전트를 시작하고, 최종 결과를 구문 분석하여 다운로드된 파일 이름을 가져옵니다.
| # 에이전트 브라우저 작동 방식 샘플 agent = Agent( task=“허깅페이스로 이동하여 An-j96/SuperstoreData를 검색하고 해당 페이지를 연 다음, Files 탭으로 이동하여 데이터 CSV 파일을 다운로드하세요.”, llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key), use_vision=False, vision_detail_level=“auto”, # 사용 가능한 옵션 [‘low’, ‘high’, ‘auto’] browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=“./Download”, user_data_dir=None ) ), # 브라우저의 다운로드 디렉토리 경로를 설정합니다. controller=Controller( output_model=AgentOutput ), # 에이전트가 작업 완료 시 다운로드된 파일 이름을 출력하도록 합니다. ) all_results = await agent.run() final_output = AgentOutput.model_validate_json( all_results.final_result() ) # 최종 에이전트 결과를 구문 분석합니다. |
(참고: 로컬 머신에서 브라우저 에이전트를 실행하면 실제 브라우저 인스턴스가 시작되어 작동하는 모습을 직접 확인할 수 있습니다.) 전체 메서드는 다운로드된 파일 이름을 수집하고, 작업 결과를 파일에 작성한 다음, 나중에 업로드하기 위해 모든 파일 경로를 반환합니다.
| … # 기존 코드 아래 async def downloading_task_for_browser_agent( task: str, api_key: str, model: str, model_api_base_url: str, use_vision: bool, download_dir_path: str = “./Download”, ) -> Tuple[str, list[str]]: “”“ 브라우저 사용을 통해 사용자의 다운로드 작업을 수행하고 다운로드 디렉토리 경로와 다운로드된 파일 이름을 반환합니다. 반환값: (다운로드_디렉토리, 확장자가_포함된_파일_이름) 튜플 ”“” agent = Agent( task=task, llm=ChatOpenAI( base_url=model_api_base_url, model=model, api_key=api_key, max_completion_tokens=20_000, frequency_penalty=0, # 이 페널티는 도구 사용에 약간의 영향을 줄 수 있으므로 0으로 유지하세요. ), use_vision=use_vision, vision_detail_level=“auto”, # 사용 가능한 옵션 [‘low’, ‘high’, ‘auto’] browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=download_dir_path, user_data_dir=None, # “./browser_user_data” ) ), # 브라우저의 다운로드 디렉토리 경로를 설정합니다. controller=Controller( output_model=AgentOutput ), # 에이전트가 AgentOutput 스키마에 따라 작업 실행 결과를 출력하도록 합니다. max_failures=5, ) try: # 에이전트를 실행하고 출력을 구조화합니다. all_results = await agent.run() final_output: AgentOutput = AgentOutput.model_validate_json( all_results.final_result() ) if final_output.task_files: console.print( Panel( f"[bold yellow]작업 결과를 파일에 작성 중입니다…[/bold yellow] {final_output.task_files}“, title=“작업 결과”, border_style=“white”, ) ) # 각 작업 결과를 다운로드 디렉토리의 파일에 작성합니다. task_result_files: list[str] = [] for task_file in final_output.task_files or [] file_path = Path(task_file.filename) # 경로 탐색 공격이나 안전하지 않은 절대 경로를 방지합니다. if file_path.is_absolute() or “…” in file_path.parts: raise ValueError( f"에이전트가 안전하지 않은 파일 경로를 파일 이름으로 전달했습니다: {file_path}” ) # 파일 경로를 다운로드 디렉토리 내부로 지정합니다. file_path = Path(download_dir_path) / file_path # 다운로드 디렉토리가 존재하지 않으면 생성합니다. file_path.parent.mkdir(parents=True, exist_ok=True) # 작업 결과 내용을 파일에 작성합니다. with open(file_path, “w”, encoding=“utf-8”) as f: f.write(task_file.content) task_result_files.append(task_file.filename) if task_result_files: console.print( Panel( f"[bold green]작업 결과가 다음 파일에 작성되었습니다:[/bold green] {task_result_files}“, title=“작업 결과”, border_style=“white”, ) ) # 다운로드된 파일과 작업 결과 파일을 결합합니다. file_results = (final_output.downloaded_files or []) + task_result_files if file_results: console.print( Panel( f”[bold green]사용 가능한 파일:[/bold green] {file_results} (위치: {download_dir_path})“, title=“다운로드된 파일”, border_style=“green”, ) ) else: raise RuntimeError(“다운로드되거나 작성된 파일이 없습니다.”) except Exception as e: file_results = None console.print( Panel( f”[bold red]오류:[/bold red] {str(e)}\ ", title=“실행 오류”, border_style=“red”, ) ) return (download_dir_path, file_results) |
에이전트 호출은 try-except 블록으로 래핑되어 있으므로, 실패가 발생하면 예외가 발생하고 file_results가 None으로 설정됩니다. 마지막으로 메서드는 download_dir_path와 file_results 경로를 모두 반환합니다.
EDA 에이전트에 사용 가능한 도구의 스키마 정의:
이제 브라우저 사용이 완료되었으니, EDA 에이전트에 사용 가능한 도구의 스키마를 정의할 차례입니다. 다음 4가지 도구를 제공할 것입니다:
- run_python_code: 에이전트가 샌드박스 내에서 Python 코드를 실행할 수 있도록 합니다.
- run_on_command_line: 에이전트가 샌드박스 터미널에서 명령을 실행할 수 있도록 합니다(예: Python 패키지 설치).
- sync_with_user: 에이전트가 샌드박스에서 생성하거나 업데이트한 파일 및 디렉토리를 로컬 동기화 폴더와 동기화할 수 있도록 합니다.
- delete_from_user_sync_folder: 에이전트가 로컬 동기화 폴더에서 파일이나 디렉토리를 제거할 수 있도록 합니다.
이러한 모든 도구를 통해 에이전트는 코드 실행, 터미널 사용, 샌드박스와 로컬 시스템 간의 파일 동기화를 완전히 제어할 수 있습니다.
sandbox_eda.py에서 도구의 스키마를 확인할 수 있습니다:
- run_python_code는 실행할 Python 코드를 입력 매개변수로 받아 결과가 있으면 반환합니다.
| { “type”: “function”, “function”: { “name”: “run_python_code”, “description”: “Runs the python code and returns the result if any.”, “parameters”: { “type”: “object”, “properties”: { “python_code”: { “type”: “string”, “description”: “The Python code to run.”, } }, “required”: [“python_code”], }, }, }, |
- run_on_command_line은 실행할 명령을 입력 매개변수로 받아 결과가 있으면 반환합니다.
| { “type”: “function”, “function”: { “name”: “run_on_command_line”, “description”: “Runs the command on the command line and returns the result if any.”, “parameters”: { “type”: “object”, “properties”: { “command”: { “type”: “string”, “description”: “The command to run on the command line.”, } }, “required”: [“command”], }, }, }, |
- sync_with_user는 두 가지 입력 매개변수를 받습니다:
- sandbox_path: 샌드박스 내부의 파일 또는 디렉토리 경로입니다.
- path_on_user_sync_folder: 사용자의 동기화 폴더 내에서 파일 또는 디렉토리가 취할 경로입니다.
이 두 번째 경로는 다음 형식이 됩니다(예: /new.txt), 동기화 폴더가 상위 디렉토리라고 가정합니다. 예를 들어, 나중에 /new.txt를 sync_folder/new.txt로 해석합니다.
| { “type”: “function”, “function”: { “name”: “sync_with_user”, “description”: “Will sync a file or directory on sandbox, to the sync folder on the user’s computer”, “parameters”: { “type”: “object”, “properties”: { “sandbox_path”: { “type”: “string”, “description”: “Path to the file or directory on the sandbox.”, }, “path_on_user_sync_folder”: { “type”: “string”, “description”: “Relative path where the file or directory will be placed inside the user’s sync folder. For example, ‘/hello.txt’ goes directly in the sync folder, while ‘/run1/hello.txt’ will be placed in a ‘run1’ subfolder within the sync folder.”, }, }, “required”: [“sandbox_path”, “path_on_user_sync_folder”], }, }, }, |
- delete_from_user_sync_folder는 하나의 입력 매개변수만 받습니다:
- path_on_user_sync_folder: 사용자의 동기화 폴더에서 에이전트가 삭제하려는 파일 또는 디렉토리의 경로입니다.
이는 sync_with_user와 동일한 경로 가정을 따른다는 점에 유의하세요.
| { “type”: “function”, “function”: { “name”: “delete_from_user_sync_folder”, “description”: “Will delete a file or directory from the sync folder on the user’s computer”, “parameters”: { “type”: “object”, “properties”: { “path_on_user_sync_folder”: { “type”: “string”, “description”: “Relative path to the file or directory on the user’s sync folder. For example, ‘/hello.txt’ will delete it directly from the sync folder, while ‘/run1/hello.txt’ will delete it directly from ‘run1’ subfolder within the sync folder.”, } }, “required”: [“path_on_user_sync_folder”], }, }, }, |
사용 가능한 기능 구현:
먼저 다음 매개변수를 받는 SandboxEDA 클래스가 있습니다:
- sandbox: 샌드박스 인스턴스입니다.
- model_base_url 및 model_api_key: Novita 모델에 연결하기 위한 것입니다.
- max_consecutive_function_calls_allowed: 기본값은 30로, 나중에 보게 되겠지만 에이전트의 무한 기능 호출 루프를 방지하기 위한 것입니다.
| … # 기존 코드 아래 class SandboxEDA: def __init__( self, sandbox: Sandbox, model_api_base_url: str, model_api_key: str, max_consecutive_function_calls_allowed: int = 30, ): self.sandbox = sandbox self.model_api_base_url = model_api_base_url self.model_api_key = model_api_key self.max_consecutive_function_calls_allowed = ( max_consecutive_function_calls_allowed ) |
그 다음 SandboxEDA 클래스의 메서드인 run_python_code입니다. 이 메서드는 Python 코드를 입력으로 받아 sandbox 인스턴스를 사용하여 실행합니다. 출력이 반환되면, 이미지 출력(참고: base64로 인코딩됨)은 temp_image_output 디렉토리에 저장됩니다. 마지막으로 메서드는 이미지 출력, 기타 출력, 로그 및 오류를 포함하는 사전을 반환합니다.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def run_python_code(self, python_code: str) -> dict: “”“ 샌드박스에서 Python 코드를 실행하고, 이미지가 있으면 로컬에 저장합니다. 매개변수: python_code (str): 실행할 Python 코드입니다. 반환값: dict: base64 이미지 출력과 기타 출력(표준 출력, 로그, 오류 등)을 포함합니다. ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # base64로 인코딩된 이미지를 반복하고 ./temp_image_output 디렉토리에 temp-{timestamp}.png 형식으로 저장합니다. for b64_image in image_outputs: timestamp = int(time.time_ns()) image_filename = Path(f"./temp_image_output/temp-{timestamp}.png") # temp_image_output 디렉토리가 아직 존재하지 않으면 생성합니다. image_filename.parent.mkdir(parents=True, exist_ok=True) with open(image_filename, “wb”) as f: f.write(base64.b64decode(b64_image)) return { “image_outputs”: image_outputs, “other_outputs”: { “outputs”: [result for result in execution.results if not result.png], “logs”: execution.logs, “error”: execution.error, }, } |
다음으로 run_on_command_line 메서드입니다. 이 메서드도 마찬가지로 샌드박스 인스턴스에서 명령을 실행한 다음 출력을 포함하는 사전을 반환합니다. 실행에 실패하면 출력에 대해 None을 반환하고 "실행 오류"를 오류 메시지로 설정합니다.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def run_on_command_line(self, command: str) -> dict: “”“ 샌드박스에서 명령을 실행합니다. 매개변수: command (str): 실행할 명령입니다. 반환값: dict: 명령의 출력과 실행 오류(해당하는 경우)를 포함합니다. ”“” try: result = self.sandbox.commands.run(command) return { “output”: { “stdout”: result.stdout, “stderr”: result.stderr, “exit_code”: result.exit_code, “error”: result.error, }, “execution error”: None, } except Exception as e: return {“output”: None, “execution error”: str(e)} |
그리고 sync_with_user 메서드입니다. 앞서 설명한 대로, 이 메서드는 sandbox_path와 path_on_user_sync_folder를 받습니다. 샌드박스 경로가 파일을 가리키면, 샌드박스에서 해당 파일을 다운로드하여 대응하는 동기화 폴더 위치에 저장합니다. 디렉토리인 경우, 메서드는 모든 하위 내용을 재귀적으로 탐색하여 각 파일을 대응하는 위치에 다운로드합니다. 성공하면 "동기화 성공"을 반환하고, 그렇지 않으면 예외 메시지를 반환합니다.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def sync_with_user(self, sandbox_path, path_on_user_sync_folder): “”“ 샌드박스의 파일이나 디렉토리를 사용자의 동기화 폴더로 다운로드합니다. 매개변수: sandbox_path (str): 샌드박스에서 동기화할 파일 또는 디렉토리의 경로입니다. path_on_user_sync_folder (str): 사용자의 동기화 폴더 내에서 파일 또는 디렉토리가 배치될 상대 경로입니다. 반환값: str: 파일 또는 디렉토리가 성공적으로 동기화된 경우 “Sync Successful”, 그렇지 않으면 오류 메시지입니다. ”“” try: path_info = self.sandbox.files.get_info(sandbox_path) if path_info.type == FileType.DIR: # 디렉토리인 경우 내용을 반복하며 다운로드합니다. dir_contents = self.sandbox.files.list(sandbox_path) for content in dir_contents: path_to_content_in_sync_folder = Path( path_on_user_sync_folder ).joinpath(content.name) self.sync_with_user(content.path, path_to_content_in_sync_folder) elif path_info.type == FileType.FILE: # 파일이 항상 ./sync_folder 내부에 있도록 합니다. sandbox_path_obj = Path(path_on_user_sync_folder) # 루트나 드라이브 구성 요소를 제거하여 경로를 상대 경로로 만듭니다. relative_path = sandbox_path_obj.relative_to( sandbox_path_obj.anchor or “.” ) # sync_folder 내부의 최종 경로 file_path = Path(“sync_folder”) / relative_path # 경로에 존재하지 않는 모든 디렉토리를 생성합니다. file_path.parent.mkdir(parents=True, exist_ok=True) # 파일을 동기화 폴더에 다운로드합니다. file_content = self.sandbox.files.read(sandbox_path, “bytes”) with open(file_path, “wb”) as f: f.write(file_content) return “Sync Successful” except Exception as e: return str(e) |
마지막으로 delete_from_user_sync_folder 메서드입니다. 앞서 설명한 대로 path_on_user_sync_folder를 받습니다. 그 다음 해당 파일 또는 디렉토리가 존재하면 삭제하고 "삭제 성공"을 반환하며, 그렇지 않으면 예외 메시지를 반환합니다.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def delete_from_user_sync_folder(self, path_on_user_sync_folder): “”“ 사용자 동기화 폴더에서 파일이나 디렉토리를 삭제합니다. 매개변수: path_on_user_sync_folder (str): 사용자 동기화 폴더에서 삭제할 파일 또는 디렉토리의 경로입니다. 반환값: str: 파일 또는 디렉토리가 성공적으로 삭제된 경우 “Deletion Successful”, 그렇지 않으면 오류 메시지입니다. ”“” # 파일이 항상 ./sync_folder 내부에 있도록 합니다. sandbox_path_obj = Path(path_on_user_sync_folder) # 루트나 드라이브 구성 요소를 제거하여 경로를 상대 경로로 만듭니다. relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”) # sync_folder 내부의 최종 경로 delete_path = Path(“sync_folder”) / relative_path try: if not delete_path.exists(): raise Exception( f"동기화 폴더에 {path_on_user_sync_folder} 위치에 파일 또는 디렉토리가 존재하지 않습니다." ) if delete_path.is_file(): delete_path.unlink() elif delete_path.is_dir(): shutil.rmtree(str(delete_path)) return “Deletion Successful” except Exception as e: return str(e) |
기타 샌드박스 기능:
1. 샌드박스에 파일 업로드 메서드.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def upload_files_to_sandbox( self, file_paths: list[str], file_names_in_sandbox: list[str] ): “”“ 파일을 샌드박스에 업로드합니다. 매개변수: file_paths (list[str]): 업로드할 파일의 경로(예: [”./Download/data.csv", “./Download/data2.csv”]). file_names_in_sandbox (list[str]): 샌드박스에서 파일이 가질 이름(예: [“data.csv”, “data2.csv”]). 참고: 파일은 샌드박스의 /home/user 디렉토리에 업로드됩니다(예: ./home/user/data.csv, ./home/user/data2.csv). “”“ console.print( f”[yellow]{file_paths}의 파일을 샌드박스에 업로드 중입니다[/yellow] (id: {self.sandbox.sandbox_id})“ ) for file_path, file_name_in_sandbox in zip(file_paths, file_names_in_sandbox): with open(file_path, “rb”) as file: self.sandbox.files.write(file_name_in_sandbox, file) console.print( f”[bold cyan]파일 {file_paths}가 샌드박스에 업로드되었습니다[/bold cyan] (id: {self.sandbox.sandbox_id})" ) |
2. 샌드박스 기본 디렉토리(/home/user) 내용 나열 메서드.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def list_files_in_sandbox_main_dir(self) -> list[str]: return [i.name for i in self.sandbox.files.list(“/home/user”)] |
에이전트와의 상호 작용 구축:
이제 에이전트와 상호 작용하는 메서드를 살펴보겠습니다. 그 전에 prompts/system_prompt.py에 있는 매개변수화된 시스템 프롬프트 지시문을 확인해 보겠습니다.
| SYSTEM_PROMPT = “”“ 당신은 탐색적 데이터 분석(EDA) 에이전트이며, 인터넷 액세스가 가능한 샌드박스에 액세스할 수 있습니다. 여기서 다음 작업을 수행할 수 있습니다: - run_python_code 함수 호출을 사용하여 Python 코드를 실행합니다. - run_on_command_line 또는 run_python_code 함수 호출을 통해 리눅스 머신에서 할 수 있는 거의 모든 작업을 수행할 수 있습니다. - sync_with_user 함수 호출을 통해 샌드박스에서 생성, 작성 또는 업데이트한 모든 디렉토리(구조가 필요한 경우 웹사이트 등)나 파일을 사용자의 로컬 머신에 있는 동기화 폴더와 동기화할 수 있습니다. - delete_from_user_sync_folder 함수 호출을 통해 사용자의 로컬 머신에 있는 동기화 폴더에서 해당 디렉토리나 파일을 삭제할 수 있습니다. 현재 작업 디렉토리(PWD)는 '/home/user’이며, 아래는 해당 디렉토리의 파일 목록입니다. {list_sandbox_files} 참고: - 샌드박스에는 일반적인 데이터 분석 패키지가 사전 설치되어 있지만, 설치되어 있는지 확실하지 않은 패키지나 import 오류로 인해 누락된 패키지가 있는 경우, 설치되어 있는지 확인하고 설치되어 있지 않으면 설치할 수 있습니다. - 이미지 출력(예: 데이터 시각화)의 경우 PNG 형식이어야 합니다. 함수 호출 가이드라인: - run_on_command_line을 사용해야 하는 경우(예: 패키지 설치 등)가 아니라면 항상 run_python_code를 사용하여 작업을 수행하세요. - 필요한 경우 함수 호출을 연결하세요: 한 함수 호출의 결과를 받은 후, 추가 정보가 필요하면 즉시 추가 호출을 수행하세요. - 필요한 정보만 먼저 수집하세요: 함수 호출에서 충분한 정보를 얻었을 때만 사용자에게 응답하세요. - 효율적으로 작업하세요: 최대 {max_consecutive_function_calls_allowed}번의 연속 함수 호출 제한이 있지만, 충분한 정보를 얻기 위해 가능한 한 적은 호출을 수행하세요. - 사용자가 도구 호출의 출력을 읽을 것이라고 가정하지 말고, 응답을 제공하세요. 사용자는 아마도 (/home/user/) 디렉토리에 있는 데이터셋 파일({downloaded_dataset_names})에 대해 EDA를 수행하려고 할 것입니다. 다음 함수 호출을 수행할 수 있습니다: {available_function_calls_schema} ”“” |
이제 SandboxEDA 클래스의 eda_chat 메서드로 넘어가겠습니다. 이 메서드는 샌드박스에 업로드된 파일 이름(나중에 보겠지만 채팅을 시작하기 전에 업로드할 것입니다)과 사용할 Novita 모델 이름을 받습니다.
1. 먼저 OpenAI 클라이언트를 기본 URL을 통해 Novita를 가리키도록 설정하고, 첫 번째 메시지로 시스템 프롬프트를 사용하여 대화를 초기화합니다.
| … # 기존 코드 아래 class SandboxEDA: … # 기존 코드 아래 def eda_chat( self, downloaded_dataset_names: list[str], model_for_eda: str, ): “”“ 코드 실행 및 터미널 명령이 가능한 AI 에이전트와의 대화형 EDA 세션 매개변수: downloaded_dataset_names (list[str]): 다운로드된 데이터셋의 이름입니다. model_for_eda (str, 선택 사항): 사용할 기본 모델입니다. ”“” console.print( Panel( “[bold green]EDA 세션이 시작되었습니다[/bold green]\ 종료하려면 'quit()'를 입력하세요.”, title=“탐색적 데이터 분석”, border_style=“green”, ) ) client = OpenAI( base_url=self.model_api_base_url, api_key=self.model_api_key, ) # 시스템 프롬프트로 대화 초기화 messages = [ { “role”: “system”, “content”: SYSTEM_PROMPT.format( downloaded_dataset_names=str(downloaded_dataset_names), list_sandbox_files=str(self.list_files_in_sandbox_main_dir()), available_function_calls_schema=str( AVAILABLE_FUNCTION_CALL_SCHEMAS ), max_consecutive_function_calls_allowed=self.max_consecutive_function_calls_allowed, ), } ] |
2. 다음으로, 메인 채팅 루프는 while 루프이며, 여기서 사용자의 메시지를 받아 기존 메시지 기록에 추가합니다. 그리고 연속 도구 호출로 인한 무한 루프를 방지하기 위해 while 루프 내에 제한까지의 for 루프가 있으며, 제한에 도달하면 예외가 발생합니다. 그 다음 메시지로 모델에 프롬프트를 전송합니다.
| … # 기존 코드 아래 # 메인 채팅 루프 while True: user_input = Prompt.ask(“\ [bold yellow]>>> 사용자 메시지[/bold yellow]”) if user_input.lower().strip() == “quit()”: break messages.append({“role”: “user”, “content”: user_input}) # 연속 도구 호출을 안전 제한으로 처리하여 무한 루프 방지 for i in range(self.max_consecutive_function_calls_allowed + 1): if i == self.max_consecutive_function_calls_allowed: raise Exception( f"에이전트의 연속 도구 호출은 {self.max_consecutive_function_calls_allowed}을(를) 초과할 수 없습니다." ) response = client.chat.completions.create( model=model_for_eda, messages=messages, tools=AVAILABLE_FUNCTION_CALL_SCHEMAS, frequency_penalty=0, ) response_message = response.choices[0].message |
3. 다음으로 모델이 도구 호출을 결정했는지 확인합니다. 그렇다면 모델이 제공한 인수를 사용하여 도구를 실행하고, 터미널에 출력을 인쇄하거나(해당하는 경우 이미지 표시) 출력을 모델에 반환합니다. 먼저 run_python_code 도구 호출부터 시작합니다.
| … # 기존 코드 아래 tool_calls = response_message.tool_calls if tool_calls: messages.append( response_message ) # 도구 호출을 트리거한 어시스턴트 메시지 추가 # 요청된 각 도구 호출 실행 for tool_call in tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments) if name == “run_python_code”: console.print( Panel( args[“python_code”], title=“에이전트가 Python 코드 실행 중”, border_style=“blue”, ) ) code_result = self.run_python_code(args[“python_code”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, “name”: name, # 이미지 출력(예: 데이터 시각화)이 있는 경우, 아직 도구 호출에서 이미지를 반환할 수 없으므로 # 에이전트에 이미지가 사용자에게 표시되었음을 알립니다. “content”: [ { “type”: “text”, “text”: ( f"이미지가 이미 터미널의 사용자에게 표시되었으며 사용자 컴퓨터의 ./temp_image_output 디렉토리에 temp-{{timestamp}}.png 등의 임시 파일로 저장되었습니다. 다른 출력은 아래와 같습니다\ {code_result[‘other_outputs’]}“ if code_result[“image_outputs”] else f”{code_result[‘other_outputs’]}" ), } ], } ) display_sandbox_code_output(code_result) |
3b. run_on_command_line 도구 호출.
| … # 기존 코드 아래 elif name == “run_on_command_line”: console.print( Panel( args[“command”], title=“에이전트가 터미널에서 명령 실행 중”, border_style=“blue”, ) ) command_result = self.run_on_command_line(args[“command”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # 이 메시지가 도구 사용에서 나왔음을 나타냅니다. “name”: name, “content”: str(command_result), } ) display_sandbox_command_output(command_result) |
3c. sync_with_user 도구 호출.
| … # 기존 코드 아래 elif name == “sync_with_user”: console.print( Panel( f"[bold yellow]에이전트가 {args[‘sandbox_path’]}을(를) 사용자의 동기화 폴더({args[‘path_on_user_sync_folder’]})에 동기화하기 시작했습니다[/bold yellow]", title=“파일 동기화”, border_style=“white”, ) ) sync_result = self.sync_with_user( args[“sandbox_path”], args[“path_on_user_sync_folder”] ) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # 이 메시지가 도구 사용에서 나왔음을 나타냅니다. “name”: name, “content”: sync_result, } ) … # 간략하게 생략 |
3d. delete_from_user_sync_folder 도구 호출 및 존재하지 않는 함수 호출의 경우 알 수 없는 오류 발생.
| … # 기존 코드 아래 elif name == “delete_from_user_sync_folder”: console.print( Panel( f"[bold yellow]에이전트가 사용자의 동기화 폴더({args[‘path_on_user_sync_folder’]})에서 파일을 삭제하고 있습니다[/bold yellow]“, title=“파일 동기화”, border_style=“white”, ) ) delete_result = self.delete_from_user_sync_folder( args[“path_on_user_sync_folder”] ) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # 이 메시지가 도구 사용에서 나왔음을 나타냅니다. “name”: name, “content”: delete_result, } ) … # 간략하게 생략 else: raise ValueError(f"알 수 없는 함수 호출: {name}”) |
4. 에이전트의 최신 응답이 도구 호출이 아닌 경우, 응답을 사용자에게 인쇄하고 연속 도구 호출 제한 루프를 종료합니다.
| … # 기존 코드 아래 else: # 도구 호출이 없으므로 어시스턴트 응답을 메시지에 추가한 후 표시합니다. messages.append( {“role”: “assistant”, “content”: response_message.content} ) console.print( f"[bold green]>>> 어시스턴트 응답: {response_message.content} [/]" ) break |
에이전트 플로우 오케스트레이션:
마지막으로 main.py가 애플리케이션의 진입점으로, 모든 것을 통합합니다. 내부에 있는 start_eda 메서드는 새로운 샌드박스 세션을 시작합니다. sandbox_timeout 매개변수는 샌드박스가 자동으로 종료되기 전에 활성 상태로 유지되는 시간을 결정합니다. 데모를 위해 900초(약 15분)로 설정하겠습니다.
샌드박스가 생성된 후, 파일을 업로드한 다음 eda_chat 메서드를 시작하여 에이전트와의 상호 작용을 시작합니다.
| … # 기존 코드 아래 def start_eda( model_for_eda: str, dataset_paths: list[str], dataset_file_names: list[str], api_key_for_sandbox_and_model: str, model_api_base_url: str, sandbox_domain: str, sandbox_template: str, sandbox_timeout: int, ): with Sandbox( template=sandbox_template, api_key=api_key_for_sandbox_and_model, domain=sandbox_domain, timeout=sandbox_timeout, ) as sandbox: try: sandbox_eda = SandboxEDA( sandbox, model_api_base_url, api_key_for_sandbox_and_model ) console.print( f"[bold cyan]샌드박스가 시작되었습니다[/bold cyan] (id: {sandbox.sandbox_id})“ ) sandbox_eda.upload_files_to_sandbox(dataset_paths, dataset_file_names) sandbox_eda.eda_chat(dataset_file_names, model_for_eda) console.print( f”\ \ [bold cyan]------ 샌드박스(id: {sandbox.sandbox_id})에 대한 EDA 세션이 완료되었습니다 ------[/]“ ) finally: console.print( f”[bold cyan]----- 샌드박스가 종료되었습니다 (id: {sandbox.sandbox_id})-----[/cyan]\ " ) |
애플리케이션의 시작점인 main 메서드 아래입니다:
| … # 기존 코드 아래 async def main( api_key_for_sandbox_and_model: str, model_api_base_url: str, model_for_browser_agent: str, enable_vision_for_browser_agent: bool, model_for_eda: str, sandbox_domain: str, sandbox_template: str, sandbox_timeout_seconds: int, ): while True: # 환영 배너 console.print( Panel( “[bold white]에이전트 기반 탐색적 데이터 분석에 오신 것을 환영합니다[/bold white]\ \ “ ”[grey]진행 방법을 선택하세요:[/grey]\ “ ”[grey]1.[/grey] 먼저 데이터셋을 다운로드합니다. ”[grey]2.[/grey] 이미 다운로드된 데이터셋으로 진행합니다. “[grey]3.[/grey] 종료”, title=“메인 메뉴”, border_style=“green”, width=70, ) ) choice = Prompt.ask( “\ [bold yellow]선택지를 입력하세요[/bold yellow]”, choices=[“1”, “2”, “3”] ).strip() if choice == “1”: result = await choice_download_dataset( api_key_for_sandbox_and_model, model_api_base_url, model_for_browser_agent, enable_vision_for_browser_agent, ) if result: download_path, filenames = result DATASET_PATHS = [ str(Path(download_path) / filename) for filename in filenames ] DATASET_FILE_NAMES = filenames else: continue # 사용자가 메인 메뉴로 돌아감 elif choice == “2”: result = choice_proceed_with_already_downloaded_datasets() if result: DATASET_PATHS = result DATASET_FILE_NAMES = [os.path.basename(path) for path in result] else: continue # 사용자가 메인 메뉴로 돌아가기를 클릭했으므로. elif choice == “3”: break # EDA 세션 시작 start_eda( model_for_eda, DATASET_PATHS, DATASET_FILE_NAMES, api_key_for_sandbox_and_model, model_api_base_url, sandbox_domain, sandbox_template, sandbox_timeout_seconds, ) |
main.py는 스크립트로 실행되므로, 아래 코드를 추가하고 환경 변수, 샌드박스 타임아웃 및 사용할 Novita 모델을 전달합니다:
| … # 기존 코드 아래 if __name__ == “__main__”: NOVITA_API_KEY = os.getenv(“NOVITA_API_KEY”) NOVITA_BASE_URL = os.getenv(“NOVITA_BASE_URL”) NOVITA_E2B_DOMAIN = os.getenv(“NOVITA_E2B_DOMAIN”) NOVITA_E2B_TEMPLATE = os.getenv(“NOVITA_E2B_TEMPLATE”) NOVITA_MODEL_FOR_BROWSER_AGENT = “qwen/qwen3-coder-480b-a35b-instruct” ENABLE_VISION_FOR_BROWSER_AGENT = ( False # True로 설정하면 브라우저 에이전트 모델에 비전 기능이 있는지 확인하세요. ) NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct” NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900초(15분), 샌드박스 인스턴스는 이 시간 후에 자동으로 종료됩니다. asyncio.run( main( NOVITA_API_KEY, NOVITA_BASE_URL, NOVITA_MODEL_FOR_BROWSER_AGENT, ENABLE_VISION_FOR_BROWSER_AGENT, NOVITA_MODEL_FOR_EDA, NOVITA_E2B_DOMAIN, NOVITA_E2B_TEMPLATE, NOVITA_SANDBOX_TIMEOUT_SECONDS, ) ) |
EDA 에이전트 테스트 실행:
터미널에서 다음 명령을 실행하세요:
| uv run main.py |
결론
탐색적 데이터 분석 에이전트를 구축한 것을 축하합니다. 이제 어떤 데이터셋 파일의 분석/인사이트를 사용하여 웹사이트, PowerPoint 등을 생성하도록 요청할 수 있으며, 결과가 로컬 컴퓨터에 직접 동기화됩니다.
간단히 요약하면, 이 글에서는 지시를 받아 브라우저를 사용하여 웹을 탐색하고 파일을 다운로드하며, Novita 샌드박스에서 코드와 명령을 실행하고, 파일과 디렉토리를 로컬 컴퓨터와 동기화하는 에이전트를 구축하는 방법을 배웠습니다.
이것은 빙산의 일각에 불과합니다. 에이전트를 데이터베이스에 연결하거나, MCP를 통해 Google Docs 등의 도구와 통합하는 등 다양한 확장이 가능합니다. Novita에서 아이디어를 현실로 만들어 보세요!
Novita AI는 개발자가 간단한 API를 사용하여 AI 모델을 쉽게 배포할 수 있는 AI 클라우드 플랫폼으로, AI 구축 및 확장을 위한 합리적인 가격의 신뢰할 수 있는 GPU 클라우드도 제공합니다.
