ReAct Agent 패턴详解与 Python 实现

저는 이 패턴을 처음 구현할 때 단순히 프롬프트를 보내고 답변을 받는 구조로 시작했지만, 복잡한 다단계 작업을 수행해야 하는 순간이 왔습니다. 예를 들어 웹 검색을 통해 최신 정보를 확인하고, 그 결과를 바탕으로 계산を行い, 최종 답변을 구성하는 작업이 필요했죠. 이때 단순한 API 호출 방식으로는 한계가 있었습니다. 이 글에서는 ReAct (Reasoning + Acting) 에이전트 패턴을 HolySheep AI와 함께 구현하는 방법과, 실제로 경험한 오류들을 해결하는 방법을 상세히 설명드리겠습니다.

ReAct Agent 패턴이란?

ReAct 에이전트 패턴은 추론(Reasoning)과 행동(Acting)을 교대로 수행하여 복잡한 문제를 해결하는 구조입니다. 일반 API 호출과 달리:

• Reasoning: 현재 상황을 분석하고 다음 행동을 결정
• Acting: 도구(tool)를 사용하여 실제 행동 수행
• Observation: 행동 결과를 관찰하고 피드백 확보

이 패턴의 핵심은 반복 루프입니다. 단일 응답이 아닌 Thought → Action → Observation 단계를 반복하며 점진적으로 정답에 근접합니다. HolySheep AI의 게이트웨이 구조는 이러한 다단계 API 호출에 최적화되어 있습니다.

HolySheep AI Gateway 설정

import os
from openai import OpenAI

HolySheep AI Gateway configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

OpenAI 호환 클라이언트 초기화
client = OpenAI(
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL,
    timeout=120.0,  # 다단계 에이전트 실행을 위한 타임아웃 설정
    max_retries=3
)

모델 설정 (비용 최적화를 위한 모델 선택)
MODEL_CONFIG = {
    "fast": "gpt-4.1-mini",      # $0.30/MTok - 빠른 응답
    "standard": "gpt-4.1",       # $8/MTok - 균형형
    "powerful": "claude-sonnet-4", # $4.50/MTok - 고품질
    "cheap": "deepseek-chat"     # $0.42/MTok -低成本
}

print("✅ HolySheep AI Gateway initialized successfully")
print(f"📍 Base URL: {HOLYSHEEP_BASE_URL}")
print(f"📦 Available Models: {list(MODEL_CONFIG.keys())}")

도구(Tool) 시스템 구현

from typing import TypedDict, Literal, List, Union
from dataclasses import dataclass
from enum import Enum
import json
import time

class ToolType(Enum):
    """지원되는 도구 유형"""
    WEB_SEARCH = "web_search"
    CALCULATOR = "calculator"
    CODE_EXEC = "code_executor"
    FILE_READ = "file_read"
    API_CALL = "api_call"

@dataclass
class ToolResult:
    """도구 실행 결과"""
    tool_name: str
    success: bool
    result: any
    error: str = None
    execution_time_ms: float = 0.0

class ToolRegistry:
    """도구 등록 및 관리 시스템"""
    
    def __init__(self):
        self.tools = {}
        self.execution_log = []
    
    def register(self, name: str, func: callable, description: str):
        """도구 등록"""
        self.tools[name] = {
            "function": func,
            "description": description
        }
        print(f"🔧 Registered tool: {name}")
    
    def execute(self, tool_name: str, **kwargs) -> ToolResult:
        """도구 실행"""
        start_time = time.time()
        
        if tool_name not in self.tools:
            return ToolResult(
                tool_name=tool_name,
                success=False,
                result=None,
                error=f"Tool '{tool_name}' not found"
            )
        
        try:
            result = self.tools[tool_name]["function"](**kwargs)
            execution_time = (time.time() - start_time) * 1000
            
            tool_result = ToolResult(
                tool_name=tool_name,
                success=True,
                result=result,
                execution_time_ms=execution_time
            )
            self.execution_log.append(tool_result)
            return tool_result
            
        except Exception as e:
            return ToolResult(
                tool_name=tool_name,
                success=False,
                result=None,
                error=str(e),
                execution_time_ms=(time.time() - start_time) * 1000
            )
    
    def get_available_tools(self) -> List[str]:
        return list(self.tools.keys())

도구 구현 예시
def web_search_impl(query: str) -> str:
    """웹 검색 도구 구현"""
    # 실제 구현에서는 Google Search API, SerpAPI 등 사용
    return f"Search results for '{query}': [Result 1] {query} is a complex topic..."

def calculator_impl(expression: str) -> str:
    """계산기 도구 구현"""
    try:
        # 안전한 eval을 위한 간단한 파서
        allowed_chars = set('0123456789+-*/.() ')
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return str(result)
        else:
            raise ValueError("Invalid characters in expression")
    except Exception as e:
        raise ValueError(f"Calculation error: {e}")

def code_executor_impl(code: str, language: str = "python") -> str:
    """코드 실행 도구 구현"""
    if language == "python":
        # 실제 환경에서는 sandboxed execution 사용
        local_vars = {}
        exec(code, {}, local_vars)
        return str(local_vars)
    return "Unsupported language"

도구 레지스트리 생성 및 도구 등록
tool_registry = ToolRegistry()
tool_registry.register("web_search", web_search_impl, "Search the web for information")
tool_registry.register("calculator", calculator_impl, "Calculate mathematical expressions")
tool_registry.register("code_executor", code_executor_impl, "Execute code safely")

print(f"📋 Available tools: {tool_registry.get_available_tools()}")

ReAct 에이전트 코어 구현

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum

class ReActStepType(Enum):
    """ReAct 단계 유형"""
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    FINAL = "final"

@dataclass
class ReActStep:
    """ReAct 실행 단계"""
    step_type: ReActStepType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[Dict] = None
    tool_result: Optional[Any] = None

@dataclass
class ReActAgent:
    """ReAct 에이전트 메인 클래스"""
    client: OpenAI
    model: str
    max_iterations: int = 10
    tool_registry: ToolRegistry = field(default_factory=ToolRegistry)
    
    def __post_init__(self):
        self.conversation_history: List[Dict] = []
        self.execution_trace: List[ReActStep] = []
    
    def build_system_prompt(self) -> str:
        """시스템 프롬프트 구성"""
        available_tools = self.tool_registry.get_available_tools()
        tools_description = "\n".join([
            f"- {name}: {self.tool_registry.tools[name]['description']}"
            for name in available_tools
        ])
        
        return f"""You are a ReAct (Reasoning + Acting) agent.
You solve problems by iterating through these steps:
1. THOUGHT: Analyze the current situation and determine what to do next
2. ACTION: Use a tool to gather information or perform an action
3. OBSERVATION: Review the result and continue reasoning

Available tools:
{tools_description}

Output format for actions:
{{"action": "tool_name", "input": {{"param": "value"}}}}


Output format for final answer:
{{"final_answer": "Your complete answer here"}}


Rules:
- If you need external information, use appropriate tools
- For calculations, always use the calculator tool
- If you've gathered enough information, provide your final answer
- Maximum {self.max_iterations} iterations allowed
"""
    
    def execute_action(self, action_str: str) -> Any:
        """액션 문자열 파싱 및 실행"""
        try:
            import re
            # JSON 형식 파싱
            json_match = re.search(r'\{[^{}]*\}', action_str, re.DOTALL)
            if json_match:
                action_data = json.loads(json_match.group())
                
                if "action" in action_data:
                    tool_name = action_data["action"]
                    tool_input = action_data.get("input", {})
                    return self.tool_registry.execute(tool_name, **tool_input)
                
                if "final_answer" in action_data:
                    return action_data["final_answer"]
            
            return None
        except json.JSONDecodeError as e:
            return f"JSON Parse Error: {e}"
        except Exception as e:
            return f"Execution Error: {e}"
    
    def run(self, user_query: str) -> Dict[str, Any]:
        """ReAct 에이전트 실행"""
        self.conversation_history = [
            {"role": "system", "content": self.build_system_prompt()},
            {"role": "user", "content": user_query}
        ]
        self.execution_trace = []
        
        for iteration in range(self.max_iterations):
            print(f"\n{'='*50}")
            print(f"🔄 Iteration {iteration + 1}/{self.max_iterations}")
            
            try:
                # HolySheep AI API 호출
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=self.conversation_history,
                    temperature=0.7,
                    max_tokens=2000
                )
                
                assistant_message = response.choices[0].message.content
                self.conversation_history.append({
                    "role": "assistant",
                    "content": assistant_message
                })
                
                print(f"🤖 Assistant: {assistant_message[:200]}...")
                
                # 액션 실행
                result = self.execute_action(assistant_message)
                
                if isinstance(result, str) and result:
                    # 최종 답변인 경우
                    if "final_answer" not in assistant_message.lower():
                        observation = f"Tool result: {result}"
                        self.conversation_history.append({
                            "role": "user",
                            "content": f"Observation: {observation}"
                        })
                        self.execution_trace.append(ReActStep(
                            step_type=ReActStepType.OBSERVATION,
                            content=observation,
                            tool_result=result
                        ))
                    else:
                        return {
                            "answer": result,
                            "iterations": iteration + 1,
                            "trace": self.execution_trace
                        }
                else:
                    # 빈 결과
                    self.conversation_history.append({
                        "role": "user",
                        "content": "Observation: No result returned. Please continue."
                    })
                    
            except Exception as e:
                return {
                    "error": str(e),
                    "iterations": iteration + 1,
                    "trace": self.execution_trace
                }
        
        return {
            "error": "Max iterations exceeded",
            "iterations": self.max_iterations,
            "trace": self.execution_trace
        }

에이전트 인스턴스 생성
agent = ReActAgent(
    client=client,
    model=MODEL_CONFIG["standard"],  # gpt-4.1 - $8/MTok
    max_iterations=5,
    tool_registry=tool_registry
)
print("✅ ReAct Agent initialized")

실전 활용 예제

# 예제 1: 복잡한 수학 문제 풀기
query_1 = """
Question: A company has 1,250 products. They sold 45% in Q1, then 30% of remaining in Q2.
After Q3, they had 180 products left. How many products were sold in Q3?
"""

print("📊 Example 1: Complex Math Problem")
print(f"Question: {query_1.strip()}")
print("-" * 50)

result_1 = agent.run(query_1)
print(f"\n✅ Final Answer: {result_1}")

예제 2: 웹 검색 후 정보 분석
query_2 = """
Question: What is the current price of Bitcoin and what is 10% of that value?
You need to first search for the current Bitcoin price.
"""

print("\n\n📊 Example 2: Search and Calculate")
print(f"Question: {query_2.strip()}")
print("-" * 50)

result_2 = agent.run(query_2)
print(f"\n✅ Final Answer: {result_2}")

HolySheep AI 비용 분석
print("\n\n💰 Cost Analysis (HolySheep AI)")
print("-" * 50)
gpt-4.1: $8/MTok, 평균 1회 요청 약 500 토큰 가정
tokens_per_request = 500
cost_per_request = (tokens_per_request / 1_000_000) * 8
print(f"Model: GPT-4.1 ($8/MTok)")
print(f"Avg tokens/request: {tokens_per_request}")
print(f"Cost/request: ${cost_per_request:.4f}")
print(f"5 iterations cost: ${cost_per_request * 5:.4f}")

비용 최적화 전략

HolySheep AI를 활용하면 다양한 모델을 단일 API 키로 접근할 수 있어, 작업 유형에 따라 비용을 최적화할 수 있습니다:

• gpt-4.1-mini ($0.30/MTok): 단순 계산, 도구 호출 파싱
• gpt-4.1 ($8/MTok): 복잡한 추론, 다단계 계획
• deepseek-chat ($0.42/MTok): 배치 처리, 반복 작업
• claude-sonnet-4 ($4.50/MTok): 고품질 텍스트 분석

자주 발생하는 오류와 해결

1. 401 Unauthorized Error

# ❌ 잘못된 접근
client = OpenAI(api_key="invalid_key")  # 인증 실패

✅ 해결 방법
client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # HolySheep에서 받은 키
    base_url="https://api.holysheep.ai/v1"  # 정확한 엔드포인트
)

키 유효성 검증
import os
def validate_api_key(api_key: str) -> bool:
    if not api_key or len(api_key) < 10:
        return False
    if api_key.startswith("sk-"):
        return True
    return False

HolySheep AI 키 형식 확인
assert HOLYSHEEP_API_KEY.startswith("sk-"), "Invalid API key format"

2. 429 Rate Limit Error

# ❌ 무한 재시도 (계속 실패)
for i in range(100):
    response = client.chat.completions.create(...)

✅ 해결 방법: 지수 백오프 + 레이트 리밋
import time
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class RateLimitHandler:
    def __init__(self, max_retries=5, base_delay=1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.request_count = 0
        self.last_reset = time.time()
    
    async def execute_with_rate_limit(self, func, *args, **kwargs):
        """레이트 리밋이 적용된 실행"""
        # HolySheep AI 권장: 분당 60회 요청
        while self.request_count >= 60:
            wait_time = 60 - (time.time() - self.last_reset)
            if wait_time > 0:
                await asyncio.sleep(wait_time)
                self.request_count = 0
                self.last_reset = time.time()
        
        for attempt in range(self.max_retries):
            try:
                self.request_count += 1
                return await func(*args, **kwargs)
            except RateLimitError:
                delay = self.base_delay * (2 ** attempt)
                print(f"⏳ Rate limited, waiting {delay}s...")
                await asyncio.sleep(delay)
        
        raise Exception("Max retries exceeded")

모델별 레이트 리밋 설정
RATE_LIMITS = {
    "gpt-4.1": {"rpm": 500, "tpm": 150000},
    "gpt-4.1-mini": {"rpm": 1500, "tpm": 150000},
    "deepseek-chat": {"rpm": 1000, "tpm": 2000000},
    "claude-sonnet-4": {"rpm": 100, "tpm": 20000}
}

3. JSONDecodeError / Invalid Response

# ❌ 잘못된 응답 처리
response = client.chat.completions.create(...)
result = json.loads(response)  # 직접 파싱 실패

✅ 해결 방법: 다양한 응답 형식 처리
import re

def extract_action_from_response(response_text: str) -> Optional[Dict]:
    """다양한 형식의 응답에서 액션 추출"""
    
    # 형식 1: ```json 블록
    json_match = re.search(r'``(?:json)?\s*(\{.*?\})\s*``', response_text, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group(1))
        except json.JSONDecodeError:
            pass
    
    # 형식 2: 일반 JSON
    json_match = re.search(r'\{[^{}]*\}', response_text, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
    
    # 형식 3: 자연어에서 키워드 추출
    if "final_answer" in response_text.lower():
        return {"final_answer": response_text.split("final_answer")[-1].strip()}
    
    return None

def safe_json_parse(text: str, default=None):
    """안전한 JSON 파싱"""
    try:
        return json.loads(text)
    except (json.JSONDecodeError, TypeError):
        return default

응답 검증
class ResponseValidator:
    REQUIRED_FIELDS = ["action"]  # 또는 "final_answer"
    
    @staticmethod
    def validate(response: Dict) -> tuple[bool, str]:
        if not response:
            return False, "Empty response"
        
        if "final_answer" in response:
            return True, "Final answer"
        
        if "action" in response:
            if "web_search" not in str(response) and \
               "calculator" not in str(response) and \
               "code_executor" not in str(response):
                return False, f"Unknown action: {response.get('action')}"
            return True, "Valid action"
        
        return False, "Missing required fields"

4. TimeoutError / ConnectionError

# ❌ 기본 타임아웃 설정
client = OpenAI(api_key="key", base_url="url")  # 30초 기본

✅ 해결 방법: 적절한 타임아웃 + 재시도 로직
from openai import APIError, Timeout
import httpx

재시도 정책이 적용된 클라이언트
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

client = OpenAI(
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL,
    timeout=httpx.Timeout(120.0, connect=30.0),  # 읽기 120초, 연결 30초
    max_retries=3
)

@retry(
    retry=retry_if_exception_type((Timeout, APIError, ConnectionError)),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def resilient_api_call(query: str) -> str:
    """복원력 있는 API 호출"""
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": query}],
        timeout=120.0
    )
    return response.choices[0].message.content

다중 모델 폴백
def multi_model_fallback(query: str) -> str:
    """모델 장애
관련 리소스
📚 AI API 기술 문서
💰 요금제 보기
📖 개발자 문서
🚀 무료 가입
관련 문서
AI API 보안 취약점 진단: HolySheep AI 환경의 침투 테스트 완벽 가이드
일본 개발자를 위한 AI API 완벽 가이드: 해외 신용카드 없이 AI 모델 활용하기
AI API 에러 로그 분석: ELK Stack 통합 튜토리얼