지능형 에이전트의 이해와 직접 구현하기
기존의 대화형 AI 서비스는 주로 일회성 질의응답에 최적화되어 있습니다. 사용자의 복잡한 의도를 파악하여 여러 단계를 거쳐 스스로 행동할 수 있는 시스템은 아직 보편화되지 않았습니다. 이러한 한계를 극복하고, 환경 변화를 감지하며 목표를 달성하는 주체(Agent) 를 만들기 위해서는 단순한 프롬프트 엔지니어링을 넘어선 아키텍처 설계가 필요합니다. 본 가이드에서는 대형 언어 모델 (LLM) 을 기반으로 하되, 도구 활용과 기억 장치를 갖춘 완전히 자율적인 에이전트를 개발하는 과정을 다룹니다.
1. 기존 LLM 과 AI 에이전트의 차별성
일반 사용자가 접하는 챗봇이나 생성형 AI 는 입력된 텍스트에 대해 확률적으로 다음 토큰을 예측하는 엔진입니다. 이는 능동적이지 않으며, 외부 세계와의 상호작용도 제한적입니다. 반면, AI 에이전트는 다음과 같은 핵심 역량을 추가로 갖추어야 합니다.
- 지각 (Perception): 텍스트뿐만 아니라 API 데이터, 파일, 실시간 정보 등 다양한 입력원을 해석하는 능력.
- 기억 (Memory): 과거 상호작용 내역을 보존하고 필요 시 참조할 수 있는 구조화된 저장소.
- 계획 (Planning): 최종 목표를 세부 실행 단계로 분해하고 우선순위를 정하는 논리.
- 실행 (Action): 실제로 외부 시스템을 조작하거나 데이터를 생성하는 동작 수행 기능.
이러한 요소를 결합할 때만 비로소 LLM 은 보조 도구를 넘어서서 임무를 완수하는 파트너가 될 수 있습니다.
2. 시스템 아키텍처 설계
효율적인 에이전트를 구성하기 위해 모듈화를 진행합니다. 전체 흐름은 '입력 처리 → 컨텍스트 관리 → 의사결정 → 작업 실행'의 순환 구조를 따릅니다. 이를 구현하기 위한 주요 클래스 구성은 다음과 같습니다.
Orchestrator: 전체 프로세스를 제어하는 메인 컨트롤러ToolRegistry: 사용 가능한 도구의 목록과 메타데이터를 관리StateManager: 작업 현황 및 장기 메모리를 유지Reasoner: LLM 을 활용한 계획 수립 및 도구 선택 담당
3. 코어 컴포넌트 구현
구현 단계에서는 Python 라이브러리를 활용하여 각 모듈을 정의합니다. 여기서는 날씨 조회, 할일 관리, 이메일 발송이라는 세 가지 기본 기능을 지원합니다.
3.1 공통 인터페이스 및 도구 정의
모든 도구는 일관된 인터페이스를 따라야 하며, 이 정보를 바탕으로 LLM 이 올바른 파라미터를 추론할 수 있어야 합니다.
from abc import ABC, abstractmethod
import json
import requests
import smtplib
from email.mime.text import MIMEText
from typing import Dict, Any, List
import os
from datetime import datetime
class ExecutableModule(ABC):
"""모든 액션 모듈의 베이스 클래스"""
@property
@abstractmethod
def identifier(self) -> str:
"""LLM 이 식별할 수 있는 고유 이름"""
pass
@property
@abstractmethod
def definition(self) -> Dict[str, Any]:
"""LLM 이 이해할 수 있는 스키마 정의"""
pass
@abstractmethod
def execute(self, args: Dict[str, Any]) -> str:
"""실제 비즈니스 로직 수행"""
pass
class WeatherService(ExecutableModule):
identifier = "query_climate"
def definition(self) -> Dict[str, Any]:
return {
"name": self.identifier,
"description": "특정 도시의 일별 기상 정보를 가져옵니다.",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "검색할 도시명"},
"target_date": {"type": "string", "description": "날짜 (YYYY-MM-DD)"}
},
"required": ["location", "target_date"]
}
}
def execute(self, args: Dict[str, Any]) -> str:
location = args["location"]
date = args["target_date"]
# 좌표 변환 API 호출 생략 및 모의 데이터 로직 (실제 구현시에는 Open-Meteo 등 사용)
try:
coord_url = f"https://geocoding-api.open-meteo.com/v1/search?name={location}"
r_coord = requests.get(coord_url)
coords = r_coord.json()["results"][0]
lat, lon = coords["latitude"], coords["longitude"]
forecast_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&daily=temperature_2m_max,precipitation_probability_max&start_date={date}"
r_weather = requests.get(forecast_url)
data = r_weather.json()["daily"]
return f"{location}의 {date} 예보: 최고 {data['temperature_2m_max'][0]}°C, 강수확률 {data['precipitation_probability_max'][0]}%"
except Exception as e:
return f"기상 정보 조회 중 오류 발생: {str(e)}"
class TaskBoard(ExecutableModule):
identifier = "manage_tasks"
def __init__(self):
self.storage = []
def definition(self) -> Dict[str, Any]:
return {
"name": self.identifier,
"description": "개인 업무 항목을 생성, 조회 또는 제거합니다.",
"parameters": {
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["create", "view", "remove"]},
"message": {"type": "string"},
"index": {"type": "integer"}
},
"required": ["operation"]
}
}
def execute(self, args: Dict[str, Any]) -> str:
op = args["operation"]
if op == "create":
task = {"id": len(self.storage) + 1, "msg": args["message"], "time": datetime.now().isoformat()}
self.storage.append(task)
return f"새 항목 등록 완료 [ID:{task['id']}]"
elif op == "view":
return json.dumps(self.storage, ensure_ascii=False) if self.storage else "등록된 업무 없음"
elif op == "remove":
idx = args.get("index")
if idx and 0 <= idx < len(self.storage):
self.storage.pop(idx)
return "항목 삭제됨"
return "삭제 실패"
return "알 수 없는 명령"
class MailSender(ExecutableModule):
identifier = "dispatch_email"
def __init__(self, smtp_host: str, port: int, auth_user: str, auth_pass: str):
self.smtp_host = smtp_host
self.port = port
self.auth_user = auth_user
self.auth_pass = auth_pass
def definition(self) -> Dict[str, Any]:
return {
"name": self.identifier,
"description": "지정된 수신자에게 텍스트 형식의 메일을 전송합니다.",
"parameters": {
"type": "object",
"properties": {
"recipient": {"type": "string"},
"subject_line": {"type": "string"},
"body_text": {"type": "string"}
},
"required": ["recipient", "subject_line", "body_text"]
}
}
def execute(self, args: Dict[str, Any]) -> str:
try:
msg = MIMEText(args["body_text"])
msg["Subject"] = args["subject_line"]
msg["From"] = self.auth_user
msg["To"] = args["recipient"]
server = smtplib.SMTP_SSL(self.smtp_host, self.port)
server.login(self.auth_user, self.auth_pass)
server.sendmail(self.auth_user, args["recipient"], msg.as_string())
server.quit()
return "전송 성공"
except Exception as err:
return f"전송 장애: {err}"3.2 상태 관리 및 추론 엔진
에이전트의 뇌 역할을 하는 부분으로, 현재 상황을 분석하고 다음 행동을 결정합니다.
class ContextManager:
def __init__(self):
self.short_term_history: List[Dict] = []
self.permanent_storage: Dict = {}
def push(self, role: str, text: str):
self.short_term_history.append({"role": role, "content": text})
if len(self.short_term_history) > 10:
self.short_term_history = self.short_term_history[-10:]
def retrieve_history(self) -> List[Dict]:
return self.short_term_history.copy()
class DecisionEngine:
def __init__(self, llm_client, tools: List[ExecutableModule]):
self.client = llm_client
self.available_tools = tools
self.tool_manifest = [t.definition() for t in tools]
def resolve_next_step(self, user_query: str, ctx: ContextManager) -> Dict:
system_prompt = """당신은 효율적인 작업 조율자입니다.
다음 JSON 형태로만 답변하세요: {"tool": "사용할 도구명", "input": {...}}
또는 직접 답변해야 하면 {"tool": "none", "output": "내용"}."""
messages = [{"role": "system", "content": system_prompt + "\n" + json.dumps(self.tool_manifest)}]
messages.extend(ctx.retrieve_history())
messages.append({"role": "user", "content": user_query})
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=0
)
raw_content = response.choices[0].message.content
try:
# 마크다운 코드 블록 처리
if "```json" in raw_content:
content = raw_content.split("```json")[1].split("```")[0]
else:
content = raw_content
return json.loads(content)
except:
return {"tool": "none", "output": "이해하지 못했습니다."}3.3 통합 제어기
모든 요소를 연결하여 사용자 요청을 처리하는 최상위 클래스입니다.
class AutonomousController:
def __init__(self, api_key: str, email_settings: Dict = None):
os.environ["OPENAI_API_KEY"] = api_key
import openai
self.llm = openai.OpenAI()
self.state = ContextManager()
self.tools_map = {}
tool_list = [WeatherService(), TaskBoard()]
if email_settings:
tool_list.append(MailSender(**email_settings))
for t in tool_list:
self.tools_map[t.identifier] = t
self.planner = DecisionEngine(self.llm, tool_list)
def process_request(self, request: str, limit_steps: int = 10) -> str:
step_count = 0
outcome = ""
while step_count < limit_steps:
decision = self.planner.resolve_next_step(request, self.state)
tool_name = decision.get("tool", "none")
if tool_name == "none":
outcome = decision.get("output", "")
break
if tool_name in self.tools_map:
tool_result = self.tools_map[tool_name].execute(decision.get("input", {}))
self.state.push("tool_response", tool_result)
print(f"[Step {step_count}] {tool_name} 결과: {tool_result}")
else:
error = f"유효하지 않은 도구명: {tool_name}"
self.state.push("error", error)
step_count += 1
if not outcome and step_count >= limit_steps:
outcome = "최대 재시도 횟수를 초과했습니다."
return outcome4. 실제 운영 시 고려사항
코드만 작성한다고 해서 바로 안정적인 서비스가 되는 것은 아닙니다. 실전에서 겪을 수 있는 문제들을 해결하기 위한 전략이 필요합니다.
4.1 안정성과 오류 처리
LLM 의 출력은 불확실성이 존재합니다. 특히 JSON 파싱 오류나 잘못된 도구 호출은 빈번히 발생합니다. 이를 방지하기 위해:
- 매 단계마다 검증 로직을 배치하여 타입 에러를 줄입니다.
- 재귀적 사고 과정보다는 명확한 단계별 지시를 통해 사이클릭 루프 (무한 반복) 를 막습니다.
- 외부 API 가 응답하지 않을 경우를 대비한 타임아웃 설정이 필수적입니다.
4.2 비용 최적화
단순한 사실 확인이나 계산 작업까지 거대한 모델을 사용하면 비용이 급증합니다.
- 작업 난이도에 따라 소규모 모델 (Small Language Model) 과 대형 모델을 적절히 섞어 사용합니다 (Routed inference).
- 반복되는 쿼리에 대해서는 캐싱 전략을 도입하여 LLM 호출 횟수를 줄입니다.
4.3 보안 및 프라이버시
에이전트가 외부 서비스를 제어하므로 권한 관리가 중요합니다.
- API 키나 비밀번호는 절대 하드코딩하지 않고 환경 변수 (Environment Variable) 로 관리해야 합니다.
- 민감한 데이터를 LLM 컨텍스트에 그대로 주입하지 않도록 마스킹 필터를 통과시킵니다.
- 이메일 발송과 같이 외부 영향을 주는 작업은 승인 절차를 거치거나 로그를 남기는 것이 바람직합니다.
5. 확장 방향성
초반 단계에서 구축한 기초적인 에이전트를 실제 업무 환경에 맞게 발전시킬 수 있는 경로는 다양합니다.
- 다중 에이전트 협업: 하나의 시스템 대신 기획자, 개발자, 테스트담당자 역할의 에이전트들이 서로 소통하며 작업을 수행하는 구조입니다.
- 비디오 및 오디오 지원: 시각적 입력과 음성 출력을 추가하여 UI 와의 상호작용 범위를 넓힙니다.
- 벡터 데이터베이스 연동: 대용량 문서나 역사적 기록을 검색 가능하게 하여 RAG(Retrieval-Augmented Generation) 패턴을 결합합니다.
- 자동 학습: 실패한 실행 내역을 분석하여 향후 계획 수립 능력을 자동으로 개선하는 피드백 루프를 만듭니다.