안녕하세요, 저는 3년째 HolySheep AI로 AI 에이전트를 개발하고 있는 엔지니어입니다. 오늘은 AI Agent의 핵심 아키텍처인 상태 머신(State Machine)을 처음부터 설명드리겠습니다. 상태 머신은 AI 에이전트가 "어떤 상황에서 무엇을 해야 할지"를 결정하는 논리적 구조입니다.
왜 AI Agent에 상태 머신이 필요한가?
AI 에이전트를 만들 때 가장 흔히 하는 실수는 "입력 → AI 호출 → 출력" 단순 구조입니다. 하지만 실제 서비스에서는:
- 사용자가 여러 번 메시지를 보낼 수 있다
- AI가 도구를 사용해야 할 수도 있다
- 오류가 발생했을 때 재시도해야 한다
- 대화가 종료되었는지 판단해야 한다
이런 복잡한 흐름을 관리하려면 상태 머신이 필수입니다. 상태 머신은 에이전트를 명확한 상태(State)로 나누고, 각 상태에서 가능한 전이를 정의합니다.
기본 상태 머신 구조 이해하기
가장 간단한 AI Agent 상태 머신은 다음과 같습니다:
[대기중] ──사용자 메시지──▶ [처리중] ──완료──▶ [대기중]
│
▼
[오류발생] ──재시도──▶ [처리중]
│
▼
[종료]
4가지 핵심 상태 설명:
- IDLE(대기중): 사용자의 입력을 기다리는 상태
- PROCESSING(처리중): AI가 응답을 생성 중인 상태
- ERROR(오류발생): API 오류나 처리에 실패한 상태
- TERMINATED(종료): 대화가 완전히 종료된 상태
Python으로 구현하는 상태 머신
이제 HolySheep AI API를 사용해서 실제 상태 머신을 구현해보겠습니다.
1단계: 기본 설정과 의존성
# requirements.txt
openai>=1.12.0
python-dotenv>=1.0.0
설치 명령어
pip install openai python-dotenv
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep AI 설정
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY") # HolySheep 대시보드에서 발급
모델별 비용 (per million tokens)
MODEL_COSTS = {
"gpt-4.1": 8.00, # GPT-4.1: $8/MTok
"claude-sonnet-4-20250514": 15.00, # Claude Sonnet 4.5: $15/MTok
"gemini-2.5-flash": 2.50, # Gemini 2.5 Flash: $2.50/MTok
"deepseek-chat-v3.2": 0.42, # DeepSeek V3.2: $0.42/MTok (가장 저렴)
}
환경변수 파일(.env) 생성:
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
2단계: 상태 정의와 열거형
from enum import Enum
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime
class AgentState(Enum):
"""AI Agent의 가능한 모든 상태"""
IDLE = "idle" # 대기 중
PROCESSING = "processing" # AI 응답 생성 중
WAITING_TOOL_RESULT = "waiting_tool_result" # 도구 실행 결과 대기
ERROR = "error" # 오류 발생
TERMINATED = "terminated" # 대화 종료
class AgentEvent(Enum):
"""상태 전이를 유발하는 이벤트"""
USER_MESSAGE = "user_message"
AI_RESPONSE = "ai_response"
TOOL_CALL = "tool_call"
TOOL_RESULT = "tool_result"
ERROR_OCCURRED = "error_occurred"
RETRY = "retry"
MAX_RETRIES_EXCEEDED = "max_retries_exceeded"
USER_END = "user_end"
@dataclass
class Transition:
"""상태 전이 정보"""
from_state: AgentState
event: AgentEvent
to_state: AgentState
action: Optional[callable] = None
@dataclass
class StateContext:
"""현재 상태에 대한 컨텍스트 정보"""
current_state: AgentState = AgentState.IDLE
messages: List[Dict[str, str]] = field(default_factory=list)
tool_calls: List[Dict[str, Any]] = field(default_factory=list)
error_count: int = 0
max_retries: int = 3
last_error: Optional[str] = None
session_id: str = ""
created_at: datetime = field(default_factory=datetime.now)
3단계: 상태 머신 클래스 구현
from openai import OpenAI
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AgentStateMachine:
"""
AI Agent 상태 머신
이 클래스는 HolySheep AI API를 사용하여
상태 기반 AI 에이전트를 구현합니다.
"""
def __init__(self, api_key: str, base_url: str, model: str = "deepseek-chat-v3.2"):
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.model = model
self.context = StateContext()
self._transitions: Dict[tuple, Transition] = {}
self._setup_transitions()
def _setup_transitions(self):
"""유효한 상태 전이 규칙을 정의합니다"""
transitions = [
# (현재상태, 이벤트, 다음상태)
(AgentState.IDLE, AgentEvent.USER_MESSAGE, AgentState.PROCESSING),
(AgentState.PROCESSING, AgentEvent.AI_RESPONSE, AgentState.IDLE),
(AgentState.PROCESSING, AgentEvent.TOOL_CALL, AgentState.WAITING_TOOL_RESULT),
(AgentState.WAITING_TOOL_RESULT, AgentEvent.TOOL_RESULT, AgentState.PROCESSING),
(AgentState.PROCESSING, AgentEvent.ERROR_OCCURRED, AgentState.ERROR),
(AgentState.ERROR, AgentEvent.RETRY, AgentState.PROCESSING),
(AgentState.ERROR, AgentEvent.MAX_RETRIES_EXCEEDED, AgentState.TERMINATED),
(AgentState.IDLE, AgentEvent.USER_END, AgentState.TERMINATED),
]
for from_state, event, to_state in transitions:
self._transitions[(from_state, event)] = Transition(
from_state=from_state,
event=event,
to_state=to_state
)
def can_transition(self, event: AgentEvent) -> bool:
"""현재 상태에서 해당 이벤트로 전이할 수 있는지 확인"""
return (self.context.current_state, event) in self._transitions
def transition(self, event: AgentEvent) -> bool:
"""상태 전이 실행"""
key = (self.context.current_state, event)
if key not in self._transitions:
logger.warning(f"전이 불가: {self.context.current_state.value} + {event.value}")
return False
transition = self._transitions[key]
old_state = self.context.current_state
self.context.current_state = transition.to_state
logger.info(f"상태 전이: {old_state.value} → {transition.to_state.value}")
return True
def add_message(self, role: str, content: str):
"""대화 기록에 메시지 추가"""
self.context.messages.append({
"role": role,
"content": content
})
def process_user_message(self, user_input: str) -> Dict[str, Any]:
"""
사용자 메시지 처리 (메인 엔트리 포인트)
Returns:
Dict containing 'state', 'response', and 'tool_calls' if any
"""
# 1단계: IDLE 상태에서만 처리 가능
if self.context.current_state != AgentState.IDLE:
return {
"success": False,
"state": self.context.current_state.value,
"error": "현재 다른 작업이 진행 중입니다"
}
# 2단계: 사용자 메시지 추가
self.add_message("user", user_input)
# 3단계: PROCESSING 상태로 전이
self.transition(AgentEvent.USER_MESSAGE)
try:
# 4단계: HolySheep AI API 호출
response = self._call_ai()
# 5단계: AI 응답 처리
if response.tool_calls:
# 도구 호출이 있는 경우
self.context.tool_calls = response.tool_calls
self.transition(AgentEvent.TOOL_CALL)
return {
"success": True,
"state": self.context.current_state.value,
"response": response.content,
"tool_calls": response.tool_calls,
"needs_tool_execution": True
}
else:
# 일반 응답인 경우
self.add_message("assistant", response.content)
self.transition(AgentEvent.AI_RESPONSE)
return {
"success": True,
"state": self.context.current_state.value,
"response": response.content
}
except Exception as e:
return self._handle_error(str(e))
def _call_ai(self):
"""HolySheep AI API 호출"""
response = self.client.chat.completions.create(
model=self.model,
messages=self.context.messages,
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "특정 도시의 날씨 확인",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "도시 이름"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "웹 검색 실행",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "검색어"}
},
"required": ["query"]
}
}
}
],
temperature=0.7
)
return response.choices[0]
def execute_tool_and_continue(self, tool_name: str, tool_result: Any) -> Dict[str, Any]:
"""도구 실행 결과 처리 후 AI 응답 계속 생성"""
if self.context.current_state != AgentState.WAITING_TOOL_RESULT:
return {
"success": False,
"error": "도구 결과를 기다리는 상태가 아닙니다"
}
# 도구 결과를 메시지에 추가
self.add_message("tool", f"{tool_name} 결과: {tool_result}")
self.context.tool_calls = []
# PROCESSING 상태로 복귀
self.transition(AgentEvent.TOOL_RESULT)
try:
response = self._call_ai()
self.add_message("assistant", response.content)
self.transition(AgentEvent.AI_RESPONSE)
return {
"success": True,
"state": self.context.current_state.value,
"response": response.content
}
except Exception as e:
return self._handle_error(str(e))
def _handle_error(self, error_message: str) -> Dict[str, Any]:
"""오류 처리 로직"""
self.context.error_count += 1
self.context.last_error = error_message
logger.error(f"오류 발생 ({self.context.error_count}/{self.context.max_retries}): {error_message}")
if self.context.error_count >= self.context.max_retries:
self.transition(AgentEvent.MAX_RETRIES_EXCEEDED)
return {
"success": False,
"state": self.context.current_state.value,
"error": f"최대 재시도 횟수 초과: {error_message}"
}
self.transition(AgentEvent.ERROR_OCCURRED)
return {
"success": False,
"state": self.context.current_state.value,
"error": error_message,
"retry_possible": True,
"remaining_retries": self.context.max_retries - self.context.error_count
}
def retry(self) -> Dict[str, Any]:
"""오류 상태에서 재시도"""
if self.context.current_state != AgentState.ERROR:
return {
"success": False,
"error": "재시도할 수 없는 상태입니다"
}
self.transition(AgentEvent.RETRY)
# 마지막 사용자 메시지를 다시 처리
if self.context.messages:
last_user_msg = None
for msg in reversed(self.context.messages):
if msg["role"] == "user":
last_user_msg = msg["content"]
break
if last_user_msg:
return self.process_user_message(last_user_msg)
return {"success": False, "error": "재시도할 메시지가 없습니다"}
def end_conversation(self):
"""대화 종료"""
self.transition(AgentEvent.USER_END)
logger.info("대화가 종료되었습니다")
4단계: 실제 사용 예제
# main.py
from config import BASE_URL, API_KEY
from agent_state_machine import AgentStateMachine, AgentState
def main():
"""AI Agent 상태 머신 사용 예제"""
# HolySheep AI로 에이전트 초기화
# DeepSeek V3.2 사용 ($0.42/MTok - 가장 비용 효율적)
agent = AgentStateMachine(
api_key=API_KEY,
base_url=BASE_URL,
model="deepseek-chat-v3.2"
)
print(f"초기 상태: {agent.context.current_state.value}")
print("-" * 50)
# 첫 번째 대화
print("\n[사용자] 안녕하세요, 서울 날씨 알려주세요")
result = agent.process_user_message("안녕하세요, 서울 날씨 알려주세요")
print(f"상태: {result['state']}")
if result.get("tool_calls"):
print(f"도구 호출 필요: {[tc['function']['name'] for tc in result['tool_calls']]}")
# 도구 실행 시뮬레이션
tool_result = "서울 날씨: 흐림, 18도, 습도 65%"
tool_name = result["tool_calls"][0]["function"]["name"]
print(f"\n[시스템] {tool_name} 실행 중...")
result = agent.execute_tool_and_continue(tool_name, tool_result)
print(f"\n[AI] {result['response']}")
elif result.get("response"):
print(f"\n[AI] {result['response']}")
# 두 번째 대화
print("\n" + "-" * 50)
print("\n[사용자] 감사합니다, 혹시 내일은 어떨까요?")
result = agent.process_user_message("감사합니다, 혹시 내일은 어떨까요?")
print(f"상태: {result['state']}")
if result.get("response"):
print(f"\n[AI] {result['response']}")
# 대화 종료
print("\n" + "-" * 50)
agent.end_conversation()
print(f"최종 상태: {agent.context.current_state.value}")
print(f"총 메시지 수: {len(agent.context.messages)}")
if __name__ == "__main__":
main()
5단계: 고급 기능 - 상태 머신 시각화와 디버깅
# debug_utils.py
import json
from typing import Dict, Any
class AgentDebugger:
"""상태 머신 디버깅 및 모니터링 유틸리티"""
def __init__(self, agent):
self.agent = agent
def get_state_diagram(self) -> str:
"""현재 상태 다이어그램을 PlantUML 형식으로 반환"""
lines = [
"@startuml",
"skinparam backgroundColor #FEFEFE",
"skinparam state {",
" BackgroundColor #E8F5E9",
" BackgroundColor<> #FFEBEE",
" BackgroundColor<> #FFCDD2",
" BackgroundColor<> #E3F2FD",
"}"
]
# 상태 정의
lines.append("\nstate IDLE")
lines.append("state PROCESSING")
lines.append("state WAITING_TOOL_RESULT")
lines.append("state ERROR")
lines.append("state TERMINATED")
# 현재 상태 하이라이트
lines.append(f"\n[*] --> {self.agent.context.current_state.name.upper()}")
lines.append(f"note right of {self.agent.context.current_state.name.upper()}")
lines.append(f" 현재 상태\n 오류 횟수: {self.agent.context.error_count}")
lines.append("end note")
lines.append("\n@enduml")
return "\n".join(lines)
def get_debug_info(self) -> Dict[str, Any]:
"""디버그 정보 반환"""
return {
"current_state": self.agent.context.current_state.value,
"message_count": len(self.agent.context.messages),
"error_count": self.agent.context.error_count,
"last_error": self.agent.context.last_error,
"session_id": self.agent.context.session_id,
"pending_tool_calls": len(self.agent.context.tool_calls),
"available_transitions": self._get_available_events()
}
def _get_available_events(self) -> list:
"""현재 상태에서 가능한 모든 이벤트 반환"""
from agent_state_machine import AgentEvent
available = []
for event in AgentEvent:
if self.agent.can_transition(event):
available.append(event.value)
return available
def print_status(self):
"""현재 상태를 예쁘게 출력"""
info = self.get_debug_info()
print("┌" + "─" * 40 + "┐")
print(f"│ {'AI Agent 상태 보고서':^38} │")
print("├" + "─" * 40 + "┤")
print(f"│ 상태: {info['current_state']:<30} │")
print(f"│ 메시지: {info['message_count']:<28} │")
print(f"│ 오류: {info['error_count']:<30} │")
print(f"│ 대기중 도구: {info['pending_tool_calls']:<25} │")
print("├" + "─" * 40 + "┤")
print("│ 가능한 이벤트:")
for event in info['available_transitions']:
print(f"│ • {event:<34} │")
print("└" + "─" * 40 + "┘")
if info['last_error']:
print(f"\n⚠️ 마지막 오류: {info['last_error']}")
모델별 비용 비교와 선택 가이드
HolySheep AI에서 제공하는 주요 모델의 비용과 지연 시간을 비교하면:
- DeepSeek V3.2: $0.42/MTok — 가장 저렴, 긴 컨텍스트, 코드에 강점
- Gemini 2.5 Flash: $2.50/MTok — 빠른 응답, 배치 처리에 적합
- GPT-4.1: $8.00/MTok — 범용 성능 최고, 복잡한 추론에 강점
- Claude Sonnet 4.5: $15.00/MTok — 긴 컨텍스트, 분석적 작업에 강점
상태 머신 에이전트에서는 도구 실행 후 결과 재투입이频繁하게 발생하므로, 긴 컨텍스트를 효율적으로 처리하면서 비용이 낮은 DeepSeek V3.2를 기본으로 권장합니다.
자주 발생하는 오류와 해결책
오류 1: "전이 불가" 상태에서 API 호출 시도시
# ❌ 잘못된 코드
result = agent.process_user_message("메시지")
result = agent.process_user_message("또 다른 메시지") # PROCESSING 상태에서 호출 → 실패
✅ 올바른 코드: 상태 확인 후 처리
def safe_process(agent, message):
if agent.context.current_state == AgentState.IDLE:
return agent.process_user_message(message)
else:
print(f"현재 {agent.context.current_state.value} 상태입니다. 완료까지 대기해주세요.")
return {"success": False, "error": "busy"}
오류 2: HolySheep API 키 미설정 시 인증 실패
# ❌ 잘못된 코드
agent = AgentStateMachine(
api_key="YOUR_HOLYSHEEP_API_KEY", # 실제 키가 아님
base_url="https://api.holysheep.ai/v1"
)
✅ 올바른 코드: 환경변수에서 키 로드
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("""
HolySheep API 키가 설정되지 않았습니다.
1. https://www.holysheep.ai/register 에서 가입
2. 대시보드에서 API 키 발급
3. .env 파일에 HOLYSHEEP_API_KEY=발급받은_키 설정
""")
agent = AgentStateMachine(api_key=api_key, base_url=BASE_URL)
오류 3: 도구 결과 재투입 시 메시지 포맷 오류
# ❌ 잘못된 코드: role을 "tool"이 아닌 다른 값으로 설정
agent.add_message("function", f"{tool_name} 결과: {result}")
✅ 올바른 코드: 정확한 포맷으로 도구 결과 추가
agent.context.messages.append({
"role": "tool",
"content": f"{tool_name} 결과: {result}",
"tool_call_id": tool_call_id # 도구 호출 ID 포함
})
또는 간단히 문자열로 추가
agent.add_message("tool", f"{tool_name} 결과: {result}")
오류 4: 최대 재시도 횟수 초과 후 계속 요청 시
# ❌ 잘못된 코드
result = agent.process_user_message("테스트")
if not result["success"]:
agent.retry() # 최대 재시도 초과 시 TERMINATED 상태
✅ 올바른 코드: 재시도 가능 여부 확인
def robust_process(agent, message):
result = agent.process_user_message(message)
if result["state"] == "error" and result.get("retry_possible"):
print(f"재시도 가능: 남은 횟수 {result['remaining_retries']}")
return agent.retry()
elif result["state"] == "terminated":
print("최대 재시도 횟수 초과. 세션을 초기화해주세요.")
# 세션 초기화 로직
agent.context = StateContext()
return {"success": False, "needs_reset": True}
return result
실전 프로젝트 구조
본격적인 프로젝트에서는 다음과 같은 구조를 권장합니다:
my-ai-agent/
├── .env # API 키 관리
├── requirements.txt
├── config.py # 설정 및 모델 비용
├── states/
│ ├── __init__.py
│ ├── base.py # 기본 상태 클래스
│ ├── idle_state.py # 대기 상태
│ ├── processing_state.py # 처리 상태
│ └── error_state.py # 오류 상태
├── agent/
│ ├── __init__.py
│ ├── state_machine.py # 핵심 상태 머신
│ ├── tools.py # 도구 정의
│ └── handlers.py # 이벤트 핸들러
├── debug/
│ ├── __init__.py
│ └── debugger.py # 디버깅 유틸리티
└── main.py
결론
AI Agent 상태 머신은 복잡한 대화 흐름을 관리하는 핵심 아키텍처입니다. 이 튜토리얼에서 다룬内容包括:
- 4가지 기본 상태(IDLE, PROCESSING, WAITING_TOOL_RESULT, ERROR)
- 상태 전이 규칙과 이벤트 기반 아키텍처
- HolySheep AI API를 활용한 실제 구현
- 오류 처리와 재시도 메커니즘
- DeepSeek V3.2($0.42/MTok)를 활용한 비용 최적화
상태 머신을 잘 설계하면 AI Agent의 동작을 예측 가능하고 디버깅하기 쉽게 만들 수 있습니다. HolySheep AI의 다양한 모델을 조합하여 여러분만의 최적의 에이전트를 만들어보세요.
저는日常적으로 상태 머신 패턴을 사용하여客服 봇, 데이터 분석 에이전트, 자동화 워크플로우 등을 구현하고 있습니다. 궁금한 점이 있으면 댓글로 남겨주세요!
👉 HolySheep AI 가입하고 무료 크레딧 받기