2026년 2월 20일 금요일

월드랩과 오토데스크 협업을 통한 공간 AI 개발 동향

이 글은 월드랩과 오토데스크 협업을 통한 공간 AI 개발 동향을 조사한 글이다.

오토데스크 마블(Autodesk Marble) 기술적 배경

마블(Marble)은 오토데스크가 직접 개발한 제품이 아니다. 이 모델은 'AI의 대모'라 불리는 페이페이 리(Fei-Fei Li) 교수가 설립한 AI 스타트업 월드랩스(World Labs)가 개발한 핵심 생성형 3D 월드 모델이다. 오토데스크는 2026년 2월, 월드랩스에 대규모 전략적 투자를 단행하며 자사 소프트웨어와의 통합 파트너십을 발표했다.

마블의 구체적인 첫 코드 작성일이 공식적으로 공개되지는 않았으나, 회사의 설립과 주요 제품 마일스톤을 통해 개발 타임라인을 충분히 추론할 수 있다.

  • 초기 R&D 및 시작 (2024년 1월): 페이페이 리 교수를 비롯한 최고 수준의 AI 연구진들이 3D 환경 생성과 실시간 시뮬레이션을 목표로 2024년 1월에 월드랩스를 공동 창립했다. 마블의 근간이 되는 '공간 지능(Spatial Intelligence)' 연구와 코어 모델 개발은 이때부터 본격적으로 시작되었을 가능성이 높다.

  • 프로토타입 및 베타 (2025년 9월): 약 1년 8개월의 딥테크 연구 기간을 거쳐, 2025년 9월에 마블의 첫 번째 제한적 베타 버전이 세상에 공개되었다.

  • 정식 출시 (2025년 11월): 2025년 11월 12일, 텍스트, 이미지, 비디오 등을 입력받아 상호작용 가능한 3D 환경을 즉석에서 구축하는 마블 프론티어 모델이 일반 대중에게 정식으로 론칭되었다.

기술 스택

마블은 단순히 2D 이미지를 이어 붙이는 비디오 생성 AI가 아니라, 물리적 공간의 3차원 구조를 완벽히 이해하는 거대 월드 모델(LWM, Large World Models) 아키텍처를 채택하고 있다.

  • 3D 표현 포맷 (3D Gaussian Splatting): 마블은 시점이 변하면 형태가 무너지는 기존 생성 모델들의 한계를 극복하고, 변형 없이 영구적으로 보존되는 3D 환경을 생성한다. 생성된 결과물은 3D 가우시안 스플랫(Gaussian Splats)이나 메쉬(Mesh) 형태로 다운로드하여 언리얼, 유니티 등 다른 게임 엔진으로 내보낼 수 있다.

  • 실시간 프레임 모델 (RTFM, Real-Time Frame Model): 2025년 10월에 도입된 핵심 렌더링 기술이다. 단일 GPU 환경에서도 실시간으로 월드를 생성하고 상호작용할 수 있도록, 기존 프레임들을 일종의 '공간 메모리'로 활용하여 높은 디테일을 유지한다.

  • 웹 렌더링 엔진 (SparkJS.dev): 별도의 무거운 클라이언트 없이 웹 브라우저 환경에서 매끄러운 3D 렌더링을 구현하기 위해 Three.js를 기반으로 한 독자적인 렌더러인 'SparkJS.dev'를 사용한다. 이는 가우시안 스플랫과 전통적인 WebGL 에셋(glTF 모델 등)을 한 화면에 자연스럽게 혼합해 준다.

  • 공간 편집 도구 (Chisel): 사용자가 직접 상자나 평면 같은 단순한 원시 도형(Primitive)으로 3D 뼈대를 잡으면, AI가 그 맥락을 파악해 그 위에 시각적 디테일과 텍스처를 입히는 하이브리드 3D 편집 도구를 지원한다.


기존의 스테이블 디퓨전 기반 3D 생성이 단일 '객체(Object)'를 깎아내는 데 집중했다면, 월드랩스의 '마블(Marble)'은 단일 이미지나 텍스트에서 거대한 3D 가상 세계(World) 전체를 생성해 내는 기술입이다. 이를 오토데스크의 기존 생태계와 결합하는 것이 핵심이다.

A. 백본 모델 (Backbone Models)

  • Large World Models (LWM) / 공간 지능(Spatial Intelligence): 단순 2D 픽셀의 패턴을 모방하는 것을 넘어, 3D 공간의 기하학(Geometry), 재질, 빛의 반사, 물리 법칙을 스스로 추론하는 거대 세계 모델을 백본으로 사용한다.

  • NeRF 및 차세대 뉴럴 렌더링: 월드랩스의 핵심 개발진(NeRF의 창시자인 벤 밀든홀 등)의 기술적 배경을 고려할 때, 마블의 코어 엔진에는 고도화된 Neural Radiance Fields(NeRF) 기반 기술이나 가우시안 스플래팅 개념이 결합되어 시점 변화에 완벽히 대응하는 일관된 3D 씬을 연산한다.

B. 학습 데이터 종류 (Training Data)

  • 일반적인 2D 이미지 쌍을 넘어서, 3D 레이아웃, 공간 깊이(Depth) 데이터, 카메라 트래킹(Pose)이 포함된 다중 시점 영상, 그리고 오토데스크가 강점을 가진 기하학적/물리적 CAD 시뮬레이션 데이터 등 공간을 이해하기 위한 복합적인 고차원 데이터로 학습된다.

C. 오토데스크와의 통합 파이프라인 (Integration Workflow)

  • 편집 가능한 3D 씬 (Editable 3D Environments): 마블은 단순한 비디오 영상(예: OpenAI Sora)을 생성하는 것이 아니라, 구조화되고 상호작용 가능한 3D 환경 자체를 출력한다.

  • 라스트 마일 편집(Last-mile Editing) 생태계: 마블이 프롬프트로 전체 공간의 초안을 순식간에 생성하면, 이를 오토데스크의 Maya, 3ds Max, Revit 같은 전통적인 소프트웨어로 바로 넘길 수 있다. 여기서 아티스트나 엔지니어가 직접 폴리곤 토폴로지, 리깅, 정밀한 재질 수정을 거쳐 최종 결과물(M&E 및 AEC 분야)을 완성하게 된다.


유사한 오픈소스 3D/월드 생성 모델

마블과 같은 강력한 상용 월드 모델에 대항하여, 연구자들과 개발자들이 투명하게 활용할 수 있는 오픈소스 생태계의 3D 생성 기술들도 빠르게 발전하고 있다.

  • DiamondWM: 구글의 'Genie'나 마블과 유사한 성격을 지닌 대표적인 오픈소스 월드 모델이다. 대량의 FPS 게임 플레이 영상을 시각적으로 학습하여 개발되었으며, 사용자의 로컬 데스크톱 GPU에서도 직접 구동하며 실시간으로 상호작용할 수 있는 점이 특징이다.

  • NVIDIA Isaac Sim (로보틱스 및 시뮬레이션): 프롬프트 한 줄로 세상 전체를 즉석에서 그려내는 마법 같은 생성형 AI는 아니지만, 오픈소스 기반의 확장 가능한 레퍼런스 프레임워크 역할을 한다. 주로 AI 로봇 모델 훈련을 위한 합성 데이터를 대량으로 생성하고, 물리 법칙이 적용된 가상 환경을 정밀하게 시뮬레이션하는 데 핵심적으로 쓰인다.

  • Tencent Hunyuan 3D 시리즈: 텍스트나 단일 이미지를 고품질 3D 에셋으로 변환하는 오픈 웨이트 기반의 생성 모델이다. 2025년 1월 버전 2.0 출시에 이어 최신 3.0 버전은 복잡한 건축물 생성 등에 폭넓게 활용되며 3D 아티스트들의 모델링 시간을 크게 단축시키고 있다.

아울러, 다음과 같은 백본 기술을 살펴볼 필요가 있다.

1. 가장 빠르고 완벽한 Image-to-3D Mesh

2. 3D 가우시안 스플래팅 + 스테이블 디퓨전(생성형 AI)의 융합

  • DreamGaussian: 스테이블 디퓨전의 상상력과 3DGS를 결합해, 이미지를 먼저 가우시안으로 빠르게 만든 뒤 실질적으로 활용 가능한 Mesh로 변환하는 선구적인 프로젝트이다.

  • Threestudio: 네르프(NeRF), 3DGS, 스테이블 디퓨전을 이용한 3D 생성 연구를 한곳에 모아둔 텍스트-to-3D 통합 프레임워크이다.


3. 원본 렌더링 기술


최근 발표된 월드랩스의 마블과 오토데스크의 만남은 기존의 3D 제작 파이프라인(기획, 모델링, 렌더링)을 'AI 초안 생성, 디테일 모델 수정'이라는 차원으로 바꿔놓고 있다.

결론적으로, 오토데스크가 왜 월드랩스에 그토록 막대한 자본을 투자했는지 그 전략적 배경은 명확하다. 수십 시간에 달하던 기존 CAD 및 3D 그래픽 설계자들의 수작업을 마블의 압도적인 '공간 지능'이 획기적으로 대체하고 보조할 수 있기 때문이다.


레퍼런스
작성자: 시간: 댓글 없음:
라벨:

2026년 2월 17일 화요일

고속학습과 모델추론을 지원하는 Unsloth 기반 파인튜닝모델 개발

이 글은 고속학습과 모델추론을 지원하는 Unsloth 기반 모델 파인튜닝 개발 방법을 나눔한다. 


unsloth  사용 순서
1. 환경 구성 및 의존성 설치
Unsloth는 최신 GPU 아키텍처에서 최적의 성능을 발휘하며, 라이브러리 설치 후에는 반드시 PyTorch 및 관련 의존성 패키지의 버전을 확인해야 한다. 특히 가속화된 연산을 위해 xformers  bitsandbytes 등을 함께 설치하는 것이 필수적이다. 이는 메모리 사용량을 70% 가까이 절감하면서도 훈련 속도를 2배 이상 높이는 핵심적인 기반이 된다. 

unsloth 설치는 다음과 같이 명령 입력하면 된다(리눅스에서 동작). 
pip install "unsloth @ git+https://github.com/unslothai/unsloth.git"

2. 모델 및 토크나이저 로드
Unsloth는 FastLanguageModel 클래스를 통해 Llama, Mistral, Gemma 등 주요 오픈 소스 모델을 빠르게 불러오는 기능을 제공한다. 4비트 양자화(4-bit Quantization)를 기본적으로 지원하여 VRAM이 제한적인 환경에서도 대규모 언어 모델을 로드할 수 있는 구조이다. 로딩 시 max_seq_length 와 dtype 등을 설정하여 프로젝트의 하드웨어 사양에 최적화된 상태를 유지하는 것이 중요하다.

3. LoRA 어댑터 설정 및 적용
모델 전체를 훈련시키는 대신, 특정 레이어에 하위 행렬을 추가하여 학습하는 LoRA(Low-Rank Adaptation) 기술을 적용한다. get_peft_model 함수를 호출하여 r(rank), alpha, target_modules 등의 하이퍼파라미터를 설정하는 과정이 핵심이다. 이 방식은 파인튜닝이 필요한 가중치의 양을 획기적으로 줄여 학습 속도를 비약적으로 향상시키고 과적합(Overfitting) 위험을 방지하는 전략이다.

4. 데이터셋 준비 및 포맷팅
신뢰성 있는 학습을 위해서는 데이터의 질과 형식이 정밀하게 관리되어야 한다. 주로 Alpaca나 ChatML 형식을 따르며, Unsloth에서 제공하는 표준화된 템플릿을 사용하여 데이터를 구성하는 것이 일반적이다. 데이터셋 내의 질문과 답변 쌍을 모델이 이해할 수 있는 토큰 형태로 변환하고, 패딩(Padding) 처리를 통해 배치 학습 효율을 높이는 과정이 필요하다.

5. SFTTrainer를 이용한 모델 학습
Hugging Face의 SFTTrainer 와 결합하여 실제 학습을 수행한다. 학습률(Learning Rate), 에폭(Epoch), 배치 사이즈(Batch Size) 등의 파라미터를 설정하고 unsloth 고유의 최적화 커널을 활성화한다. 학습 과정 중 손실(Loss) 변화를 모니터링하며 가중치가 안정적으로 수렴하는지를 확인하는 작업은 모델의 신뢰성을 확보하는 필수 단계이다.

6. 모델 저장 및 배포 준비
학습이 완료된 모델은 LoRA 어댑터 형태로 저장하거나 전체 모델과 병합(Merge)하여 배포할 수 있다. 특히 Unsloth는 GGUF 포맷으로의 내보내기 기능을 강력하게 지원하여, 파인튜닝된 모델을 llama.cpp 등 다양한 추론 엔진에서 즉시 활용할 수 있도록 돕는다. 이는 개발된 모델을 실무 환경에 빠르게 통합하고 배포하는 데 매우 유리한 조건이다.

개발 방법
Unsloth는 Hugging Face의 `SFTTrainer`와 완벽하게 호환되며, 모델 로드부터 학습까지의 과정을 비약적으로 단순화한 구조이다.

from unsloth import FastLanguageModel
import torch
from trl import SFTTrainer
from transformers import TrainingArguments

# 1. 모델 및 토크나이저 로드 (4비트 양자화 적용)
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "unsloth/llama-3-8b-bnb-4bit", # 최적화된 프리셋 모델
max_seq_length = 2048,
load_in_4bit = True,
)

# 2. LoRA(Low-Rank Adaptation) 설정
model = FastLanguageModel.get_peft_model(
model,
r = 16, # Rank 설정
target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"],
lora_alpha = 16,
lora_dropout = 0, # Unsloth는 0을 권장 (속도 최적화)
bias = "none",
)

# 3. 학습 인자 및 Trainer 설정
trainer = SFTTrainer(
model = model,
tokenizer = tokenizer,
train_dataset = dataset, # 준비된 데이터셋
dataset_text_field = "text",
max_seq_length = 2048,
args = TrainingArguments(
per_device_train_batch_size = 2,
gradient_accumulation_steps = 4,
warmup_steps = 5,
max_steps = 60, # 테스트용 스텝 수
learning_rate = 2e-4,
fp16 = not torch.cuda.is_bf16_supported(),
bf16 = torch.cuda.is_bf16_supported(),
logging_steps = 1,
output_dir = "outputs",
),
)

# 4. 학습 실행
trainer.train()

보다시피 기존 파인튜닝 코드를 그대로 사용할 수 있다.

Unsloth의 주요 한계점
Unsloth는 매우 강력한 도구이지만, 특정 하드웨어와 소프트웨어 환경에 종속적인 몇 가지 제약 사항이 존재한다.

하드웨어 및 OS 제약
  • NVIDIA GPU 전용:  Unsloth는 OpenAI의 Triton 언어를 기반으로 최적화된 커널을 사용하므로, NVIDIA GPU(Turing, Ampere, Hopper 아키텍처 등)에서만 동작하는 구조이다.
  • Linux 환경 최우선:  기본적으로 리눅스 환경을 위해 설계되었으며, 윈도우 환경에서는 반드시 WSL2(Windows Subsystem for Linux)를 통해서만 안정적인 실행이 가능하다.
모델 및 라이브러리 지원 범위
  • 제한된 모델 아키텍처:  Llama, Mistral, Gemma, Phi, Qwen 등 널리 쓰이는 주요 오픈 소스 모델 위주로 최적화가 진행되어 있으며, 모든 최신 모델을 즉각적으로 지원하는 것은 아니다.
  • 싱글 GPU 최적화 편중:  기본 버전은 단일 GPU에서의 메모리 효율과 속도 극대화에 초점이 맞춰져 있어, 대규모 멀티 GPU 분산 학습(FSDP 등) 설정 시 추가적인 복잡성이 발생할 수 있는 형태이다.
기능적 제약
  • 고정된 최적화 커널:  성능을 위해 특정 연산을 수동으로 튜닝한 커널을 사용하므로, 사용자가 모델 아키텍처를 임의로 크게 수정하거나 특이한 레이어를 추가할 경우 Unsloth의 가속 혜택을 받기 어려운 구조이다.
  • DPO/PPO 학습의 복잡성:  단순한 지도 학습(SFT)은 매우 직관적이지만, 직접적인 인간 피드백 학습(RLHF) 단계인 DPO나 PPO를 적용할 때는 설정이 다소 까다로울 수 있다는 점이 한계이다.

마무리
Unsloth를 이용한 파인튜닝은 자원 소모를 최소화하면서도 모델의 성능을 극대화할 수 있는 최신 개발 방법론이다. 이러한 효율적인 훈련 체계는 개인 개발자나 중소규모 팀에서도 고성능의 특화 모델을 구축할 수 있게 하여, AI 민주화와 기술적 한계 극복에 기여하는 중요한 도구로 평가받는 기술이다.

레퍼런스
작성자: 시간: 댓글 없음:
라벨:

2026년 2월 4일 수요일

스케일AI와 라벨링 작업 뒷이야기

이 글은 스케일AI와 라벨링 작업 뒷이야기를 나눔합니다. 

AI의 이면: 보이지 않는 데이터 라벨러와 착취적 노동 생태계
고품질의 학습 데이터는 성능이 뛰어난 대규모 언어 모델(LLM)을 생성하는 핵심 요소이며, 이는 곧 사람의 손을 거친 라벨링된 데이터셋을 의미한다. LLM의 훈련 과정 중 지도 학습과 인간 피드백 기반 강화 학습(RLHF) 단계에서는 인간의 노동이 필수적이다. 라벨러들은 이미지나 텍스트 같은 원시 데이터에 정답을 붙임으로써 AI가 올바른 판단을 내리고 답변의 품질을 높이도록 돕는다. 하지만 이 거대한 기술의 뒤에는 전 세계에 흩어진 보이지 않는 노동자들의 희생이 존재한다.

글로벌 공급망과 미세 노동의 실체
AI 개발사들은 비용 절감을 위해 케냐, 인도, 필리핀 등 저임금 국가의 노동력을 활용하는 미세 노동(Microwork) 플랫폼에 라벨링 작업을 외주화한다. 이러한 플랫폼들은 노동자와 정식 고용 계약을 맺지 않는 인간 서비스(humans-as-a-service) 모델을 채택하고 있다. 노동자들은 시간당 2달러 미만의 임금을 받으면서도 살인, 학대, 아동 착취 등 트라우마를 유발할 수 있는 유해한 콘텐츠를 수 시간 동안 검토해야 하는 열악한 환경에 처해 있다. 이들은 수십억 달러 가치의 AI 시스템을 구축하는 핵심 동력이지만, 기업의 이익 공유에서는 철저히 배제된다.


투명성 결여와 알고리즘에 의한 관리
데이터 라벨링 생태계는 심각한 불투명성 문제를 안고 있다. Scale AI와 그 자회사인 Remotasks처럼 기업들은 복잡한 지배 구조를 통해 실제 고용주와 의뢰인(OpenAI, MS 등)의 정체를 숨긴다. 노동자들은 자신이 누구를 위해 일하는지도 모른 채 암호화된 프로젝트명 아래에서 단순 작업을 반복한다. 또한, 알고리즘 관리 시스템은 노동자의 일상과 생산성을 철저히 감시한다. 화장실 휴식조차 허용하지 않는 엄격한 타이머가 작동하며, 알고리즘이 산정하는 임금은 수요와 공급에 따라 실시간으로 변동되어 노동자의 소득 예측 가능성을 박탈한다.


불안정한 일자리와 기그 경제의 병폐
노동자들은 일감을 얻기 위해 밤낮없이 화면을 모니터링해야 하며, 알고리즘의 결정에 따라 예고 없이 계정이 정지되거나 일자리를 잃기도 한다. 무급으로 진행되는 장시간의 교육 과정과 테스트 역시 노동자에게 전가되는 부담이다. 최근 기업들이 비용 절감을 위해 더 저렴한 노동 시장으로 거점을 옮기면서 기존 지역 노동자들에게 임금을 지불하지 않고 철수하는 사례도 보고되고 있다. 이는 법적 보호가 취약한 기그 경제(Gig Economy)의 전형적인 착취 구조가 AI 산업에서 재현되고 있음을 보여준다.

자동화의 한계와 규제적 대응의 필요성
데이터 라벨링을 자동화하려는 시도가 이어지고 있으나, 여전히 최종 검수 단계에서는 인간의 판단이 필요하다. 자동화는 기업의 효율성을 높일 뿐 노동자의 권리나 처우 개선으로 이어지지 않는다. 다행히 유럽의 플랫폼 노동 지침(PWD)이나 국제노동기구(ILO)의 논의 등 규제적 노력이 시작되고 있다. AI 공급망의 가장 밑바닥에서 모델을 지탱하고 있는 미세 노동자들의 권리를 보호하고, 거대 기술 기업의 책임을 강화하는 정책적 대응이 시급한 시점이다. 이들이 겪는 착취와 학대를 멈추는 것이야말로 진정한 의미의 신뢰할 수 있는 AI 개발의 시작이다.

작성자: 시간: 댓글 없음:
라벨:

2026년 1월 7일 수요일

랭그래프, 웹 기반 AI Agents 개발 방법 및 디자인패턴

인공지능은 이제 단일 모델 기반의 응답 시스템에서 벗어나, 자율적인 구성 요소들이 추론하고 행동하며 협력하는 에이전트 중심 시스템으로 급격히 이동 중이다. 기존 LLM의 단발성 상호작용과 달리, 에이전트 아키텍처는 컨텍스트가 유지되는 상태 기반 세션, 작업별 특화 로직의 모듈화, 외부 도구 및 워크플로우와의 상호운용성을 지향한다. FastAPI, LangGraph, MCP(Model Context Protocol)를 통해 확장 가능한 플랫폼 구조와 디자인 패턴을 정리한다. 좀 더 상세한 내용은 레퍼런스를 참고한다.

프로젝트 아키텍처 및 주요 구성 요소
시스템은 클라이언트-API-오케스트레이션-도구 실행으로 이어지는 명확한 계층형 구조를 가진다. FastAPI는 고성능 비동기 웹 서비스를 통해 에이전트의 진입점 역할을 수행하며, 실제 추론 로직과 분리되어 있어 유지보수가 용이하다. 서비스 레이어는 API와 오케스트레이션 계층을 연결하며 세션 초기화와 최종 응답 포맷팅을 담당한다. LLM 제공자 추상화를 통해 OpenAI나 Anthropic과 같은 다양한 모델을 유연하게 선택할 수 있는 구조이다.
MCP(Model Context Protocol)는 모델과 도구 간의 상호작용을 표준화하여 시스템의 신뢰성을 높인다. 모든 도구는 @tool 데코레이터를 사용하여 선언적으로 정의되며, 중앙 집중식 레지스트리 패턴을 통해 관리된다. 이러한 방식은 로컬 도구를 외부 MCP 서버로 교체하거나 런타임에 새로운 기능을 동적으로 로드하는 것을 매우 간편하게 만든다. 결과적으로 에이전트는 단순한 텍스트 생성을 넘어 외부 API 호출, 계산, 데이터 분석 등 실질적인 액션을 수행하는 능력을 갖추게 된다.

이제 다이어그램에서 전체 에이전트 패턴 구조와 몇몇 핵심적인 부분을 구현해 본다. 

LangGraph를 이용한 에이전트 오케스트레이션
에이전트 아키텍처의 핵심은 LangGraph를 이용한 상태 관리와 워크플로우 정의이다. StateGraph를 활용하여 에이전트가 언제 도구를 호출하고 언제 작업을 종료할지 명시적으로 제어한다. 이는 에이전트의 행동을 투명하게 만들고 디버깅을 용이하게 하는 핵심적인 설계이다.

다음과 같은 에이전트 구조가 있다고 치자. agent는 사용자 입력을 받아 LLM 추론을 한다. 그 결과 tools 호출(예. 날씨, 온도, IoT 센서, 데이터베이스, 로보틱스 액추에이터 등등)이 필요하면 tools를 호출하고, 그 결과를 다시 LLM에게 전달해 답변을 출력한다. 아니면, END 종료한다. 

이를 랭그래프로 구현하면 다음과 같다.
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

def create_agent_graph():
    llm = get_llm()
    tools = get_tools()
    llm_with_tools = llm.bind_tools(tools)

    workflow = StateGraph(AgentState)

    # LLM 추론 노드 정의
    async def call_model(state: AgentState) -> dict:
        response = await llm_with_tools.ainvoke(state["messages"])
        return {"messages": [response]}

    # 노드 등록 및 경로 설정
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", ToolNode(tools))
    workflow.add_edge(START, "agent")

    # 조건부 엣지: 도구 호출 여부에 따라 경로 결정
    def should_continue(state: AgentState) -> str:
        last_message = state["messages"][-1]
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return "tools"
        return END

    workflow.add_conditional_edges("agent", should_continue, ["tools", END])
    workflow.add_edge("tools", "agent")

    return workflow.compile()

LangGraph 기반 ReAct 패턴 구현 
사용자 질문에 대해 계속 도구를 호출해 반복된 구조로 답을 찾는 에이전트는 ReAct 구조를 사용할 수 있다. 앞의 구조에서 좀 더 실용적인 도구를 사용해 보겠다. 위키피디아 검색 및 계산기 도구를 호출하도록 랭그래프를 다음과 같이 수정한다. 

이 에이전트는 사용자 질문에 대한 특정 도구를 발견하지 못할때까지 반복해 호출할 것이다. 실제 예에서는 토큰 비용 및 성능도 고려해야 하므로, 그래프 LLM 호출 최적화가 필요할 것이다. 

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from dotenv import load_dotenv

load_dotenv()

# 1. 도구(Tool) 정의
from langchain_core.tools import tool

@tool
def search_wikipedia(query: str) -> str:
    """Search Wikipedia for information. Use this when you need factual information."""
    # 실제로는 Wikipedia API 호출
    return f"Wikipedia search results for '{query}': [모의 검색 결과 - 실제로는 API 호출]"

@tool
def calculator(expression: str) -> str:
    """Calculate mathematical expressions. Input should be a valid Python expression."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

tools = [search_wikipedia, calculator]
tool_node = ToolNode(tools)

# 2. State 정의
class AgentState(TypedDict):
    messages: Annotated[list, "The messages in the conversation"]

# 3. LLM 설정 (tool binding)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 4. Agent 노드 - LLM이 판단
def agent_node(state: AgentState) -> dict:
    """LLM이 도구를 사용할지, 최종 답변을 할지 결정 (Thought)"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 5. 라우팅 로직 - 종료 조건 결정
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    """
    ReAct의 핵심 종료 조건:
    - LLM이 tool_calls를 반환하면 → "tools" (Action 필요)
    - tool_calls가 없으면 → "end" (최종 답변 완성)
    """
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 도구 사용 필요
    return "end"  # 최종 답변 완성

# 6. 그래프 구성
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent_node)      # Thought 단계
workflow.add_node("tools", tool_node)        # Action 단계

workflow.add_edge(START, "agent")

# ReAct 사이클: agent → tools → agent → tools → ... → end
workflow.add_conditional_edges(
    "agent",
    should_continue,  # ← 종료 조건 검출
    {
        "tools": "tools",  # 도구 실행 후 다시 agent로
        "end": END         # 답변 완성 시 종료
    }
)
workflow.add_edge("tools", "agent")  # Observation 후 다시 Thought

react_agent = workflow.compile()

# 7. 실행 예시
if __name__ == "__main__":
    # 예시 1: 계산 필요
    result = react_agent.invoke({
        "messages": [HumanMessage(content="What is 342 * 67?")]
    })
    print("답변:", result["messages"][-1].content)
    print("\n" + "="*80 + "\n")
    
    # 예시 2: 검색 + 계산 필요
    result = react_agent.invoke({
        "messages": [HumanMessage(
            content="Search for the population of Seoul, then multiply it by 2"
        )]
    })
    print("답변:", result["messages"][-1].content)

이런 방식으로 필요한 도구들, 목적별 LLM 에이전트(예. 조사, 설계, 아이디어 구현, 요약, 보고서 작성 등)를 추가해 나갈 수 있다. 

결론
이 간단한 에이전트 아키턱쳐 예시는 다중 에이전트 간의 협력 체인 구축, 세션 내 인간 피드백 통합, 워크플로우 관찰 가능성(Observability) 강화 등을 통해 더욱 고도화된 에이전트 시스템으로 확장이 가능하다. 본 글에서는 랭그래프를 사용했다. 요즘에는 랭그래프 라이브러리 안전성이 높아져 이런 멀티 에이전트 구조는 큰 문제 없이 개발이 가능해 졌다. 멀티에이전트 개발 개념은 모두 유사하므로 유스케이스 목적에 따라 다른 프레임웍 라이브러리 사용하는 것도 고려할 수 있을 것이다.

부록: AI 에이전트 디자인 패턴 
앞서 본 것처럼 에이전트 개발에는 여러 디자인 패턴이 있을 수 있다. 이는 유스케이스 목적에 따라 결정되어야 한다. 토큰 사용량 대비 효과적인 답을 낼 수 있도록 구현되어야 한다. 다음은 다양한 에이전트 패턴 구현 방법을 보여준다.

import operator
from typing import Annotated, List, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv

load_dotenv()

# 순차적 에이전트(Sequential Agent)는 여러 단계에 걸쳐 내부 추론을 수행. 스크래치패드(scratchpad)라는 방식으로 진행. 별도의 외부 도구를 전혀 사용하지 않음.
# 순수한 분석과 연역적 추론만으로도 충분한 상황에 주로 사용.

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class State(TypedDict):
    question: str
    steps: Annotated[List[str], operator.add]
    answer: str

def plan_node(state: State) -> dict:
    sys = (
        "당신은 신중한 계획자입니다. 사용자의 질문을 2-4개의 간결한 단계로 나누세요. "
        "문제를 해결하지는 마세요. 번호가 매겨진 단계 목록만 반환하고, 추가 텍스트는 작성하지 마세요."
    )
    messages = [("system", sys), ("user", state["question"])]
    resp = llm.invoke(messages)
    raw = resp.content
    steps = []
    for line in str(raw).splitlines():
        line = line.strip()
        if not line:
            continue
        line = line.lstrip("-• ").split(". ", 1)[-1] if ". " in line[:4] else line.lstrip("-• ")
        steps.append(line)
    return {"steps": steps}

def solve_node(state: State) -> dict:
    """계획된 단계를 사용하여 최종 답변만 도출합니다."""
    sys = (
        "제공된 단계를 사용하여 문제를 해결하세요. "
        "최종 답변만 반환하고, 추론 과정은 포함하지 마세요."
    )
    messages = [
        ("system", sys),
        ("user", f"질문: {state['question']}\\\\n단계: {state['steps']}"),
    ]
    resp = llm.invoke(messages)
    return {"answer": str(resp.content).strip()}

#  Wire up the graph
graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("solve", solve_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "solve")
graph.add_edge("solve", END)

cot_graph = graph.compile()

state = {
    "question": "강의 동영상이 120개이고 하루에 15개를 본다면, 완강하는 데 며칠이 걸릴까요?",
    "steps": [],
    "answer": ""
}
out = cot_graph.invoke(state)
print("최종 답변:", out["answer"])

# 커스텀 에이전트(Custom Agent)는 유연성을 제공. 사용자는 전체적인 로직과 라우팅을 직접 설계할 수 있음. 또한 시스템을 구성하는 개별 노드까지 스스로 정의. 사용자의 요구에 맞춰 자유롭게 맞춤형 제어가 가능.
class CustomState(TypedDict):
    input: str
    task: Literal["math", "capitalize", "count"]
    result: str

def route(state: CustomState) -> str:
    """Deterministic router based on a simple protocol in the input."""
    text = state["input"].strip().lower()
    if text.startswith("math:"):
        return "math"
    if text.startswith("capitalize:"):
        return "capitalize"
    if text.startswith("count:"):
        return "count"
    return "count"

def do_math(state: CustomState) -> dict:
    expr = state["input"].split(":", 1)[-1].strip()
    allowed = set("0123456789+-*/(). ")
    if any(c not in allowed for c in expr):
        return {"result": "Error: unsupported characters in math expression."}
    try:
        res = eval(expr, {"__builtins__": {}})
    except Exception as e:
        res = f"Error: {e}"
    return {"result": str(res)}

def do_capitalize(state: CustomState) -> dict:
    text = state["input"].split(":", 1)[-1].strip()
    return {"result": text.upper()}

def do_count(state: CustomState) -> dict:
    text = state["input"].split(":", 1)[-1].strip()
    tokens = [t for t in text.split() if t]
    return {"result": f"words={len(tokens)} chars={len(text)}"}

graph = StateGraph(CustomState)
graph.add_node("math", do_math)
graph.add_node("capitalize", do_capitalize)
graph.add_node("count", do_count)

graph.add_conditional_edges(
    START,
    route,
    {
        "math": "math",
        "capitalize": "capitalize",
        "count": "count",
    },
)
graph.add_edge("math", END)
graph.add_edge("capitalize", END)
graph.add_edge("count", END)

custom_agent = graph.compile(debug=True)

for user_input in [
    "math: (16 + 3) * 2 + 5",
    "capitalize: hello world from AI agent",
    "count: 여기에 몇 개의 단어가 있나요?",
]:
    out = custom_agent.invoke({"input": user_input, "task": "count", "result": ""})
    print(f"입력: {user_input}\\n결과: {out['result']}\\n---")

# 슈퍼바이저(Supervisor) 패턴은 중앙 제어형 에이전트가 전체 작업을 관리합. 상위 에이전트가 문제를 분석한 뒤 이를 여러 하위 노드에 나누어 배정. 각 하위 노드가 작업을 마치면 그 결과를 다시 취합하고 검토.복잡한 작업을 나누어 병렬로 처리하거나 중앙 통제가 필요할 때 유용.
class SupervisorState(TypedDict):
    """여러 에이전트가 있는 슈퍼바이저 패턴의 상태."""
    topic: str
    messages: Annotated[List[str], operator.add]
    next_agent: str
    final_answer: str


def researcher_agent(state: SupervisorState) -> dict:
    """연구자 에이전트는 주제에 대한 정보를 수집합니다."""
    sys = (
        "당신은 연구자입니다. 주어진 주제에 대한 핵심 사실과 정보를 "
        "수집하는 것이 당신의 임무입니다. 2-3개의 핵심 포인트를 제공하세요. 간결하게 작성하세요."
    )
    messages_for_llm = [
        ("system", sys),
        ("user", f"다음 주제를 조사하세요: {state['topic']}")
    ]
    resp = llm.invoke(messages_for_llm)
    research_msg = f"연구자: {resp.content}"
    return {"messages": [research_msg]}


def expert_agent(state: SupervisorState) -> dict:
    """전문가 에이전트는 연구를 기반으로 분석하고 통찰력을 제공합니다."""
    sys = (
        "당신은 전문 분석가입니다. 제공된 연구를 검토하고 "
        "전문가 분석과 결론을 제공하세요. 구체적이고 통찰력 있게 작성하세요."
    )
    # 이전 메시지에서 컨텍스트 가져오기
    context = "\n".join(state["messages"])
    messages_for_llm = [
        ("system", sys),
        ("user", f"주제: {state['topic']}\n\n이전 조사 내용:\n{context}\n\n전문가 분석을 제공하세요.")
    ]
    resp = llm.invoke(messages_for_llm)
    expert_msg = f"전문가: {resp.content}"
    return {"messages": [expert_msg]}


def supervisor_agent(state: SupervisorState) -> dict:
    """슈퍼바이저는 다음에 어떤 에이전트가 활동할지 또는 토론을 종료할지 결정합니다."""
    sys = (
        "당신은 연구자와 전문가 간의 조사 토론을 관리하는 슈퍼바이저입니다. "
        "지금까지의 대화를 바탕으로 다음에 무엇을 해야 할지 결정하세요:\n"
        "- 초기 조사나 추가 정보가 필요하면 'researcher'를 반환하세요\n"
        "- 조사가 완료되고 전문가 분석이 필요하면 'expert'를 반환하세요\n"
        "- 조사와 전문가 분석이 모두 완료되면 'end'를 반환하세요\n\n"
        "단 하나의 단어만 응답하세요: researcher, expert, 또는 end"
    )

    context = "\n".join(state["messages"]) if state["messages"] else "아직 토론이 없습니다"
    messages_for_llm = [
        ("system", sys),
        ("user", f"주제: {state['topic']}\n\n대화 내용:\n{context}\n\n다음은 무엇인가요?")
    ]
    resp = llm.invoke(messages_for_llm)
    next_step = resp.content.strip().lower()

    # 유효한 응답인지 확인
    if next_step not in ["researcher", "expert", "end"]:
        next_step = "end"

    return {"next_agent": next_step}


def finalize_answer(state: SupervisorState) -> dict:
    """토론에서 최종 답변을 작성합니다."""
    sys = (
        "조사 토론을 명확하고 간결한 최종 답변으로 요약하세요. "
        "핵심 발견 사항과 전문가 통찰력을 포함하세요."
    )
    context = "\n".join(state["messages"])
    messages_for_llm = [
        ("system", sys),
        ("user", f"주제: {state['topic']}\n\n토론 내용:\n{context}\n\n최종 요약을 제공하세요:")
    ]
    resp = llm.invoke(messages_for_llm)
    return {"final_answer": resp.content}

def route_supervisor(state: SupervisorState) -> str:
    """슈퍼바이저의 결정에 따라 라우팅합니다."""
    next_agent = state.get("next_agent", "researcher")
    if next_agent == "end":
        return "finalize"
    return next_agent

supervisor_graph = StateGraph(SupervisorState)

supervisor_graph.add_node("supervisor", supervisor_agent)
supervisor_graph.add_node("researcher", researcher_agent)
supervisor_graph.add_node("expert", expert_agent)
supervisor_graph.add_node("finalize", finalize_answer)

supervisor_graph.add_edge(START, "supervisor")

supervisor_graph.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "researcher": "researcher",
        "expert": "expert",
        "finalize": "finalize"
    }
)

supervisor_graph.add_edge("researcher", "supervisor")
supervisor_graph.add_edge("expert", "supervisor")
supervisor_graph.add_edge("finalize", END)
supervisor_agent_graph = supervisor_graph.compile(debug=True)

topic = "AI 에이전트를 구축하는 데 LangGraph를 사용하는 주요 이점은 무엇인가요?"

initial_state = {
    "topic": topic,
    "messages": [],
    "next_agent": "",
    "final_answer": ""
}

result = supervisor_agent_graph.invoke(initial_state)

print(f"주제: {topic}\n")
print("\n토론 내용:")
for msg in result["messages"]:
    print(f"\n{msg}\n")
print(f"\n최종 답변:\n{result['final_answer']}")

레퍼런스
작성자: 시간: 댓글 없음:
라벨:
피드 구독하기: 글 (Atom)