이미지 로딩 중...
AI Generated
2025. 11. 16. · 9 Views
ReAct Prompting 완벽 가이드
AI가 생각하고 행동하는 과정을 투명하게 만드는 ReAct Prompting 기법을 배워봅니다. 추론(Reason)과 행동(Act)을 결합하여 더 신뢰할 수 있고 정확한 AI 시스템을 만드는 방법을 실무 예제와 함께 알아봅니다.
목차
- ReAct Prompting 핵심 개념 - 생각과 행동의 결합
- Thought 단계 설계하기 - AI의 사고 과정 구조화
- 행동 계획: weather_api 도구를 사용해 최신 날씨 정보 검색
- Action 단계 구현하기 - 도구 호출과 실행
- Observation 단계 처리하기 - 결과를 다음 사고로 연결
- 실전 예제 - 복잡한 질문 해결하기
- 에러 처리와 재시도 전략 - 견고한 에이전트 만들기
- 프롬프트 엔지니어링 - ReAct 패턴 최적화
- Observation이 에러면 다른 접근을 시도하세요
- 멀티 에이전트 협업 - ReAct 에이전트 조율하기
- 성능 측정과 최적화 - ReAct 에이전트 개선하기
- 실무 배포 가이드 - 프로덕션 환경에서의 ReAct
1. ReAct Prompting 핵심 개념 - 생각과 행동의 결합
시작하며
여러분이 AI 챗봇을 만들 때 이런 상황을 겪어본 적 있나요? 사용자가 "오늘 서울 날씨가 어때?"라고 물었는데, AI가 며칠 전 데이터를 기반으로 엉뚱한 답을 내놓거나, 왜 그런 답을 했는지 알 수 없어서 디버깅이 어려웠던 경험 말이죠.
이런 문제는 실제 개발 현장에서 자주 발생합니다. 기존 프롬프팅 방식은 AI가 어떤 사고 과정을 거쳐 답을 도출했는지 알 수 없고, 외부 도구나 API를 활용하는 과정이 블랙박스처럼 숨겨져 있어서 신뢰성이 떨어집니다.
바로 이럴 때 필요한 것이 ReAct Prompting입니다. AI가 "생각(Thought)"을 명시적으로 표현하고, 그에 따라 "행동(Action)"을 수행하며, 그 "결과(Observation)"를 다시 생각에 반영하는 순환 구조로 문제를 해결합니다.
개요
간단히 말해서, ReAct는 Reasoning(추론) + Acting(행동)의 합성어로, AI가 문제를 해결하는 과정을 투명하게 만드는 프롬프팅 기법입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, AI 에이전트가 복잡한 작업을 수행할 때 각 단계의 의사결정 과정을 추적할 수 있고, 오류가 발생했을 때 어느 단계에서 잘못되었는지 쉽게 파악할 수 있기 때문입니다.
예를 들어, 사용자의 주문을 처리하는 AI 에이전트가 재고를 확인하고, 가격을 계산하고, 결제를 진행하는 과정에서 각 단계의 논리를 명확히 볼 수 있다면 훨씬 안정적인 시스템을 만들 수 있습니다. 기존에는 AI에게 "이 문제를 해결해줘"라고만 요청했다면, 이제는 "먼저 무엇을 생각하고 있는지 말해줘, 그 다음 어떤 행동을 할 건지 설명해줘, 그 결과를 보고 다시 생각해줘"라는 구조화된 방식으로 작업을 진행할 수 있습니다.
ReAct의 핵심 특징은 세 가지입니다. 첫째, Thought(생각)를 통해 현재 상황을 분석하고 다음 단계를 계획합니다.
둘째, Action(행동)을 통해 실제로 도구를 사용하거나 정보를 검색합니다. 셋째, Observation(관찰)을 통해 행동의 결과를 받아 다음 생각에 반영합니다.
이러한 특징들이 AI의 의사결정을 투명하게 만들고, 복잡한 멀티스텝 문제를 단계적으로 해결할 수 있게 해줍니다.
코드 예제
# ReAct 패턴을 사용한 간단한 AI 에이전트 예제
def react_agent(question):
"""ReAct 패턴으로 질문에 답하는 에이전트"""
# Thought: 현재 상황 분석
print(f"Thought: {question}에 답하기 위해 먼저 최신 정보가 필요함")
# Action: 검색 도구 사용 결정
print("Action: search['서울 날씨 2025-11-16']")
# Observation: 도구 실행 결과
weather_data = search_weather("서울", "2025-11-16")
print(f"Observation: 서울 날씨는 {weather_data['temp']}도, {weather_data['condition']}")
# Thought: 결과를 바탕으로 다시 생각
print(f"Thought: 날씨 정보를 확인했으니 이제 사용자 친화적으로 답변 가능")
# Action: 최종 답변 생성
print(f"Action: answer['오늘 서울은 {weather_data['temp']}도이며 {weather_data['condition']}입니다']")
return f"오늘 서울은 {weather_data['temp']}도이며 {weather_data['condition']}입니다"
설명
이것이 하는 일: ReAct 패턴은 AI가 복잡한 문제를 해결할 때 사람처럼 "생각하고, 행동하고, 결과를 보고, 다시 생각하는" 순환 과정을 명시적으로 표현하도록 만듭니다. 코드를 단계별로 나누어 설명하면, 첫 번째 단계에서 Thought를 출력합니다.
"서울 날씨가 어때?"라는 질문을 받으면, AI는 "최신 정보가 필요하다"는 판단을 먼저 내립니다. 이 생각 과정을 명시적으로 출력함으로써 개발자는 AI가 올바른 방향으로 사고하고 있는지 확인할 수 있습니다.
두 번째 단계에서 Action이 실행됩니다. AI는 자신의 생각을 바탕으로 "검색 도구를 사용하겠다"고 선언하고, 실제로 search_weather() 함수를 호출합니다.
이 과정에서 어떤 도구를 왜 사용하는지가 명확하게 드러나며, 여러 도구 중 적절한 것을 선택하는 로직을 추적할 수 있습니다. 세 번째 단계에서 Observation을 통해 행동의 결과를 받아옵니다.
검색 결과 "15도, 맑음"이라는 데이터를 받으면, 이를 다시 Thought 단계로 전달합니다. AI는 "이제 충분한 정보가 있으니 답변할 수 있다"고 판단하고, 최종 Action으로 사용자 친화적인 답변을 생성합니다.
여러분이 이 코드를 사용하면 AI의 의사결정 과정이 완전히 투명해져서 디버깅이 쉬워지고, 각 단계에서 로깅이나 에러 처리를 추가할 수 있으며, 필요시 중간에 인간의 개입이나 검증을 넣을 수 있습니다. 또한 AI가 잘못된 도구를 선택했을 때 어느 시점에서 문제가 생겼는지 정확히 파악할 수 있어, 프롬프트를 개선하거나 도구를 추가하는 등의 최적화가 가능합니다.
실전 팁
💡 ReAct 패턴을 사용할 때는 각 단계(Thought, Action, Observation)를 명확히 구분하는 포맷을 정의하세요. 예를 들어 "Thought:"로 시작하는 줄은 항상 생각을, "Action:"으로 시작하는 줄은 행동을 나타내도록 일관성을 유지하면 파싱과 로깅이 훨씬 쉬워집니다.
💡 초보자가 흔히 하는 실수는 Thought 없이 바로 Action을 실행하는 것입니다. 반드시 "왜 이 행동을 하는가?"에 대한 추론을 먼저 출력하도록 프롬프트를 설계해야 AI의 판단 근거를 추적할 수 있습니다.
💡 성능 최적화를 위해 불필요한 Thought-Action 반복을 줄이세요. 간단한 질문에는 1-2회 순환으로 충분하지만, 복잡한 문제는 5-6회까지 반복할 수 있습니다. max_iterations 파라미터로 무한 루프를 방지하세요.
💡 디버깅할 때는 각 Thought, Action, Observation을 별도 로그 파일에 기록하세요. 나중에 AI가 어떤 경로로 답에 도달했는지 시각화하면 프롬프트 개선 포인트를 쉽게 찾을 수 있습니다.
💡 더 발전된 사용법으로는 여러 도구를 조합할 때 ReAct를 적용하는 것입니다. 예를 들어 "검색 → 계산 → 데이터베이스 조회 → 최종 답변" 같은 복잡한 워크플로우에서 각 단계의 논리를 명확히 할 수 있습니다.
2. Thought 단계 설계하기 - AI의 사고 과정 구조화
시작하며
여러분이 AI 에이전트를 만들 때 이런 경험 있으셨나요? AI가 갑자기 엉뚱한 도구를 선택하거나, 왜 특정 결정을 내렸는지 이해할 수 없어서 답답했던 순간 말이죠.
이런 문제는 AI가 내부적으로 어떤 추론을 하는지 명시적으로 표현하지 않기 때문에 발생합니다. 개발자는 입력과 출력만 볼 수 있고, 중간의 사고 과정은 블랙박스로 남아있어 신뢰성을 확보하기 어렵습니다.
바로 이럴 때 필요한 것이 체계적인 Thought 단계 설계입니다. AI가 현재 상황을 어떻게 이해하고, 어떤 정보가 부족하며, 다음에 무엇을 해야 하는지를 명확한 언어로 표현하도록 유도하면 훨씬 예측 가능하고 제어 가능한 시스템을 만들 수 있습니다.
개요
간단히 말해서, Thought 단계는 AI가 현재 상태를 분석하고 다음 행동을 계획하는 추론 과정을 텍스트로 명시적으로 출력하는 단계입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 복잡한 비즈니스 로직을 AI에게 맡길 때 각 의사결정의 근거를 추적할 수 있어야 감사(audit)가 가능하고, 규제 요구사항을 충족할 수 있기 때문입니다.
예를 들어, 금융 서비스에서 대출 승인 AI가 "이 고객의 신용 점수가 낮아서 추가 정보가 필요하다"라고 생각 과정을 기록하면, 나중에 왜 그런 결정이 내려졌는지 설명할 수 있습니다. 기존에는 AI에게 질문을 던지고 바로 답만 받았다면, 이제는 AI가 "먼저 이 질문의 의도를 파악하고, 내가 가진 정보로 충분한지 판단하고, 부족하다면 어떤 도구를 사용해야 할지 계획"하는 과정을 단계별로 볼 수 있습니다.
Thought 단계의 핵심 특징은 세 가지입니다. 첫째, 현재 상황 분석(Situation Analysis) - "사용자가 무엇을 원하는가?", "내가 지금 알고 있는 것은 무엇인가?".
둘째, 갭 식별(Gap Identification) - "답변하기 위해 부족한 정보는 무엇인가?". 셋째, 계획 수립(Action Planning) - "다음에 어떤 행동을 해야 하는가?".
이러한 구조화된 사고가 AI의 추론을 일관되고 추적 가능하게 만듭니다.
코드 예제
# Thought 단계를 체계적으로 설계한 프롬프트 예제
THOUGHT_TEMPLATE = """
당신은 단계적으로 생각하는 AI 에이전트입니다.
질문: {question}
다음 형식으로 생각을 정리하세요:
Thought:
4. 행동 계획: weather_api 도구를 사용해 최신 날씨 정보 검색
설명
이것이 하는 일: Thought 단계는 AI가 무작위로 행동하지 않고, 논리적인 순서로 문제를 분해하고 해결 전략을 세우도록 강제합니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 THOUGHT_TEMPLATE이 AI에게 "어떻게 생각해야 하는지"의 구조를 제공합니다.
이 템플릿은 프롬프트에 포함되어 AI가 단순히 "날씨를 검색해야겠다"가 아니라 "왜 검색이 필요한가? 내가 모르는 정보가 무엇인가?"까지 깊이 있게 사고하도록 유도합니다.
그 다음으로, structured_thought 함수가 실행되면서 네 가지 하위 함수를 호출합니다. analyze_question()은 사용자 의도를 파악하고, extract_known_info()는 이미 알고 있는 정보를 정리하며, identify_missing_info()는 무엇이 부족한지 명확히 하고, plan_next_action()은 구체적인 다음 단계를 결정합니다.
각 함수는 독립적으로 테스트 가능하며, 프롬프트 엔지니어링으로 구현할 수도 있고 별도의 분류 모델로 구현할 수도 있습니다. 마지막으로, format_thought()가 이 모든 정보를 일관된 텍스트 형식으로 변환하여 로그에 기록하고 다음 단계로 전달합니다.
이렇게 구조화된 출력은 사람이 읽기도 쉽고, 프로그램이 파싱하기도 쉬워서 자동화된 모니터링이나 이상 탐지에 활용할 수 있습니다. 여러분이 이 패턴을 사용하면 AI의 추론 품질이 크게 향상됩니다.
무작위로 도구를 선택하는 대신, 논리적인 근거를 가지고 선택하게 되며, 중간에 잘못된 가정이 있으면 Thought 단계를 보고 즉시 발견할 수 있습니다. 또한 여러 AI 에이전트를 협업시킬 때, 각 에이전트의 Thought를 공유하면 서로의 의도를 이해하고 조율할 수 있어 멀티 에이전트 시스템 구축이 훨씬 쉬워집니다.
실전 팁
💡 Thought 단계에 너무 많은 정보를 넣으면 토큰 비용이 증가하고 응답 속도가 느려집니다. 핵심만 간결하게 정리하도록 프롬프트를 조정하세요. "한 문장으로 요약하라"는 지시가 효과적입니다.
💡 흔한 실수는 Thought를 AI가 자유롭게 쓰도록 두는 것입니다. 반드시 템플릿이나 예시를 제공해 일관된 형식을 유지하도록 하세요. 그래야 나중에 정규표현식이나 파서로 자동 분석할 수 있습니다.
💡 보안 측면에서 Thought 단계에 민감한 정보(API 키, 개인정보 등)가 노출되지 않도록 주의하세요. 로깅 시 자동으로 마스킹하는 필터를 추가하는 것이 좋습니다.
💡 디버깅할 때는 Thought의 각 항목(상황 분석, 정보 갭 등)을 별도 필드로 저장하세요. 나중에 "어떤 상황에서 AI가 잘못 판단했는가?"를 분석할 때 구조화된 데이터가 훨씬 유용합니다.
💡 더 발전된 사용법으로 Chain-of-Thought와 ReAct를 결합할 수 있습니다. Thought 단계 내에서도 "단계 1: ..., 단계 2: ..." 형식으로 다단계 추론을 유도하면 더 복잡한 문제도 해결 가능합니다.
3. Action 단계 구현하기 - 도구 호출과 실행
시작하며
여러분이 AI 에이전트에 여러 도구(검색, 계산기, 데이터베이스 등)를 연결했을 때 이런 문제를 겪어본 적 있나요? AI가 어떤 도구를 언제 사용해야 할지 헷갈려하거나, 도구를 잘못된 방식으로 호출해서 에러가 발생하는 상황 말이죠.
이런 문제는 Action 단계가 명확하게 정의되지 않았을 때 발생합니다. AI는 "검색해야겠다"고 생각할 수는 있지만, 그 생각을 실제 API 호출로 변환하는 과정이 애매하면 잘못된 파라미터를 넘기거나 아예 실행에 실패합니다.
바로 이럴 때 필요한 것이 체계적인 Action 단계 구현입니다. Thought에서 계획한 행동을 정확한 함수 호출로 변환하고, 각 도구의 인터페이스를 명확히 정의하며, 실행 결과를 다시 AI에게 전달하는 파이프라인을 만들어야 합니다.
개요
간단히 말해서, Action 단계는 AI의 생각을 실제 함수 호출이나 API 요청으로 변환하여 실행하고, 그 결과를 받아오는 단계입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, AI 에이전트의 진짜 가치는 "생각"에만 있는 것이 아니라 "실제로 작업을 수행"할 수 있다는 점에 있습니다.
예를 들어, 고객 서비스 봇이 "환불 처리가 필요하다"고 생각만 하고 실제로 환불 API를 호출하지 않는다면 아무 의미가 없습니다. Action 단계가 Thought를 현실 세계의 변화로 연결해줍니다.
기존에는 AI가 "이메일을 보내야 한다"는 텍스트만 생성했다면, 이제는 "send_email(to='user@example.com', subject='주문 확인', body='...')" 같은 구조화된 함수 호출을 생성하고, 실제로 그 함수를 실행한 뒤 성공/실패 결과를 받아올 수 있습니다. Action 단계의 핵심 특징은 세 가지입니다.
첫째, 명확한 도구 정의 - 각 도구의 이름, 파라미터, 반환값을 AI가 이해할 수 있는 형식으로 문서화. 둘째, 구조화된 출력 - AI가 "Action: tool_name[param1, param2]" 같은 파싱 가능한 형식으로 출력.
셋째, 안전한 실행 - 샌드박스나 권한 검증을 통해 AI가 위험한 작업을 수행하지 못하도록 제어. 이러한 특징들이 AI를 단순한 대화 모델에서 실제 작업을 수행하는 에이전트로 진화시킵니다.
코드 예제
# Action 단계를 구현한 도구 실행 시스템
class ToolExecutor:
"""AI가 선택한 도구를 안전하게 실행하는 클래스"""
def __init__(self):
# 사용 가능한 도구들을 등록
self.tools = {
"search": self.search_web,
"calculate": self.calculate,
"database_query": self.query_database
}
def parse_action(self, action_text):
"""AI 출력에서 도구 이름과 파라미터 추출"""
# 예: "Action: search['OpenAI GPT-4']" -> ("search", ["OpenAI GPT-4"])
import re
match = re.match(r"Action: (\w+)\[(.*?)\]", action_text)
if match:
tool_name = match.group(1)
params = match.group(2).strip("'\"").split(",")
return tool_name, params
return None, None
def execute(self, action_text):
"""Action을 실제로 실행하고 Observation 반환"""
tool_name, params = self.parse_action(action_text)
if tool_name not in self.tools:
return f"Error: 도구 '{tool_name}'를 찾을 수 없습니다"
try:
result = self.tools[tool_name](*params)
return f"Observation: {result}"
except Exception as e:
return f"Observation: 실행 중 오류 발생 - {str(e)}"
def search_web(self, query):
"""웹 검색 도구 (예시)"""
return f"'{query}'에 대한 검색 결과: [관련 정보들...]"
def calculate(self, expression):
"""계산기 도구"""
return eval(expression) # 실제로는 안전한 파서 사용 필요
def query_database(self, sql):
"""데이터베이스 조회 도구"""
return "쿼리 실행 결과: [데이터...]"
설명
이것이 하는 일: Action 단계는 AI의 텍스트 출력을 실제 코드 실행으로 연결하는 브릿지 역할을 하며, 외부 시스템과의 통합을 가능하게 합니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 ToolExecutor 클래스가 사용 가능한 모든 도구를 딕셔너리로 관리합니다.
각 도구는 Python 함수로 구현되어 있으며, AI는 이 함수들의 이름과 시그니처를 알고 있어야 합니다. 이를 위해 보통 프롬프트에 "사용 가능한 도구: search(query: str), calculate(expression: str)" 같은 문서를 포함시킵니다.
그 다음으로, parse_action 메서드가 AI의 출력 텍스트를 파싱합니다. AI가 "Action: search['ReAct prompting']"라고 출력하면, 정규표현식을 사용해 도구 이름 "search"와 파라미터 "ReAct prompting"을 추출합니다.
이 파싱 로직은 AI 출력 형식에 맞춰 커스터마이즈해야 하며, 더 복잡한 경우에는 JSON 형식을 사용할 수도 있습니다. 마지막으로, execute 메서드가 추출된 도구와 파라미터로 실제 함수를 호출합니다.
try-except 블록으로 감싸서 도구 실행 중 발생하는 에러를 안전하게 처리하고, 결과를 "Observation: ..." 형식으로 반환하여 AI가 다음 Thought 단계에서 활용할 수 있도록 합니다. 성공하든 실패하든 항상 명확한 피드백을 제공하는 것이 중요합니다.
여러분이 이 패턴을 사용하면 AI 에이전트에 무한히 많은 기능을 추가할 수 있습니다. 새로운 도구를 만들려면 그냥 self.tools 딕셔너리에 함수를 추가하고 프롬프트를 업데이트하면 됩니다.
또한 각 도구 실행을 로깅하면 AI가 실제로 어떤 작업을 수행했는지 감사 추적이 가능하고, 권한 시스템을 추가하면 특정 사용자는 위험한 도구(예: 데이터 삭제)를 사용하지 못하도록 제한할 수 있습니다.
실전 팁
💡 도구 실행 시 타임아웃을 반드시 설정하세요. AI가 무한 루프에 빠지거나 응답 없는 API를 호출할 경우 전체 시스템이 멈출 수 있습니다. 각 도구는 5-10초 내에 결과를 반환하도록 제한하는 것이 좋습니다.
💡 흔한 실수는 AI 출력을 그대로 eval()로 실행하는 것입니다. 이는 심각한 보안 취약점이므로, 반드시 허용된 도구 목록(whitelist)과 파라미터 검증을 거쳐야 합니다. 예시 코드의 calculate도 실제로는 ast.literal_eval 같은 안전한 방법을 써야 합니다.
💡 성능 최적화를 위해 자주 사용되는 도구의 결과를 캐싱하세요. 예를 들어 "서울 날씨"는 5분마다만 새로 조회하고, 그 사이에는 캐시된 결과를 반환하면 API 비용과 응답 시간을 크게 줄일 수 있습니다.
💡 디버깅할 때는 각 Action 실행 전후에 상태를 로깅하세요. "도구 X 호출 시작 → 파라미터: {...} → 실행 시간: 1.2초 → 결과: ..." 같은 상세 로그가 있으면 어느 도구가 병목인지, 어떤 파라미터가 문제를 일으키는지 쉽게 파악됩니다.
💡 더 발전된 사용법으로 도구 간 의존성을 관리할 수 있습니다. 예를 들어 "결제" 도구는 반드시 "재고 확인" 도구가 먼저 실행된 후에만 사용 가능하도록 제약을 걸면, AI가 잘못된 순서로 작업을 수행하는 것을 방지할 수 있습니다.
4. Observation 단계 처리하기 - 결과를 다음 사고로 연결
시작하며
여러분이 AI 에이전트를 만들 때 이런 상황을 겪어본 적 있나요? AI가 도구를 실행한 후 그 결과를 제대로 해석하지 못하고, 마치 도구를 실행하지 않은 것처럼 엉뚱한 답을 내놓는 경우 말이죠.
이런 문제는 Action의 결과인 Observation을 다음 Thought 단계로 제대로 피드백하지 않아서 발생합니다. AI가 도구를 실행했지만 그 결과를 "읽고 이해하고 활용"하는 메커니즘이 없으면, 결국 도구를 사용한 의미가 사라집니다.
바로 이럴 때 필요한 것이 체계적인 Observation 단계 처리입니다. 도구 실행 결과를 AI가 이해하기 쉬운 형식으로 변환하고, 이를 대화 컨텍스트에 추가하여 다음 추론에 반영되도록 만들어야 합니다.
개요
간단히 말해서, Observation 단계는 Action의 실행 결과를 받아서 AI가 다음 Thought에서 활용할 수 있도록 포맷팅하고 컨텍스트에 주입하는 단계입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, AI 에이전트가 멀티스텝 태스크를 수행할 때 각 단계의 결과가 누적되어야 최종 목표를 달성할 수 있기 때문입니다.
예를 들어, "항공권 예약" 태스크에서 1) 날짜 검색 → 2) 가격 비교 → 3) 예약 처리 단계를 거칠 때, 1단계의 Observation(사용 가능한 항공편 목록)이 2단계의 Thought(어떤 항공편을 비교할지)에 반영되어야 합니다. 기존에는 도구 실행 결과를 단순히 출력만 했다면, 이제는 그 결과를 "Observation: ..." 형식으로 명확히 라벨링하고, AI의 대화 히스토리에 추가하여 다음 생성 시 참고하도록 만듭니다.
Observation 단계의 핵심 특징은 세 가지입니다. 첫째, 결과 포맷팅 - 도구의 원시 출력(JSON, HTML 등)을 AI가 이해하기 쉬운 자연어나 구조화된 텍스트로 변환.
둘째, 컨텍스트 주입 - Observation을 대화 히스토리에 추가하여 다음 프롬프트에 포함. 셋째, 에러 핸들링 - 도구 실행이 실패했을 때도 명확한 에러 메시지를 Observation으로 반환하여 AI가 대응 전략을 세울 수 있도록 함.
이러한 특징들이 AI 에이전트가 실패를 학습하고 재시도할 수 있는 회복력을 제공합니다.
코드 예제
# Observation 단계를 처리하는 ReAct 루프
class ReActAgent:
"""완전한 ReAct 루프를 구현한 에이전트"""
def __init__(self, llm, tools):
self.llm = llm # 언어 모델 (예: OpenAI API)
self.tools = tools # ToolExecutor 인스턴스
self.conversation_history = []
def run(self, question, max_iterations=5):
"""ReAct 루프를 실행하여 질문에 답변"""
self.conversation_history.append(f"Question: {question}")
for i in range(max_iterations):
# Step 1: Thought 생성
prompt = self._build_prompt()
response = self.llm.generate(prompt)
# Step 2: Thought를 히스토리에 추가
self.conversation_history.append(response)
# Step 3: Action이 포함되어 있는지 확인
if "Action:" in response:
action = self._extract_action(response)
# Step 4: Action 실행하여 Observation 얻기
observation = self.tools.execute(action)
# Step 5: Observation을 히스토리에 추가 (핵심!)
self.conversation_history.append(observation)
elif "Answer:" in response:
# 최종 답변에 도달
return self._extract_answer(response)
return "최대 반복 횟수에 도달했습니다"
def _build_prompt(self):
"""현재까지의 대화 히스토리를 프롬프트로 구성"""
history = "\n".join(self.conversation_history)
return f"{history}\n\n다음 Thought를 생성하세요:"
설명
이것이 하는 일: Observation 단계는 AI의 행동과 다음 생각을 연결하는 피드백 루프를 형성하여, 에이전트가 이전 결과로부터 학습하고 적응할 수 있게 합니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 ReActAgent 클래스가 conversation_history 리스트를 유지합니다.
이 리스트는 Question, Thought, Action, Observation이 순차적으로 쌓이는 대화의 전체 맥락을 담고 있습니다. 매 반복마다 이 히스토리가 프롬프트로 변환되어 AI에게 "지금까지 무슨 일이 있었는지" 알려줍니다.
그 다음으로, run 메서드의 핵심 루프에서 AI가 Thought와 Action을 생성한 후, tools.execute()로 실제 도구를 실행합니다. 여기서 중요한 것은 5번 줄의 self.conversation_history.append(observation)입니다.
이 한 줄이 도구 실행 결과를 대화 컨텍스트에 주입하여, 다음 반복에서 AI가 _build_prompt()를 통해 이 정보를 보고 새로운 판단을 내릴 수 있게 만듭니다. 마지막으로, "Answer:"가 나타나면 최종 답변에 도달한 것으로 판단하고 루프를 종료합니다.
만약 max_iterations에 도달할 때까지 답을 찾지 못하면 "최대 반복 횟수 도달"을 반환하여 무한 루프를 방지합니다. 실무에서는 이 경우 사용자에게 "더 구체적인 질문이 필요합니다" 같은 안내를 할 수 있습니다.
여러분이 이 패턴을 사용하면 AI 에이전트가 진짜로 "학습하는" 것처럼 동작합니다. 첫 번째 검색 결과가 충분하지 않으면, Observation을 보고 "추가 정보가 필요하다"고 판단하여 두 번째 검색을 수행할 수 있습니다.
또한 도구 실행이 실패했을 때 "Observation: 에러 발생 - 권한 부족" 같은 피드백을 받으면, AI는 다음 Thought에서 "다른 도구를 사용해야겠다"고 전략을 변경할 수 있습니다. 이런 적응력이 견고한 프로덕션 에이전트의 핵심입니다.
실전 팁
💡 Observation이 너무 길면 컨텍스트 윈도우를 초과할 수 있습니다. 도구 결과가 방대할 때는 요약 함수를 통해 핵심만 추출하세요. 예를 들어 100개 검색 결과 중 상위 3개만 포함하는 식입니다.
💡 흔한 실수는 Observation을 그냥 문자열로만 저장하는 것입니다. 구조화된 형식(예: {"type": "observation", "tool": "search", "result": "..."})으로 저장하면, 나중에 "어떤 도구의 결과가 최종 답변에 영향을 미쳤는가?"를 분석할 때 훨씬 유용합니다.
💡 성능 최적화를 위해 이전 Observation이 현재 Thought와 무관하다면 히스토리에서 제거할 수 있습니다. 예를 들어 10단계 전의 검색 결과는 더 이상 필요 없을 수 있으므로, 최근 3-5개의 Observation만 유지하는 슬라이딩 윈도우 방식을 고려하세요.
💡 디버깅할 때는 각 반복마다 전체 conversation_history를 파일로 저장하세요. AI가 어느 시점에서 잘못된 결론을 내렸는지, Observation이 제대로 반영되었는지를 시각적으로 추적할 수 있습니다.
💡 더 발전된 사용법으로 Observation에 메타데이터를 추가할 수 있습니다. "Observation (신뢰도: 85%, 출처: Wikipedia): ..." 같은 형식으로 정보의 품질을 표시하면, AI가 더 신뢰할 만한 정보에 가중치를 두고 판단할 수 있습니다.
5. 실전 예제 - 복잡한 질문 해결하기
시작하며
여러분이 AI 에이전트를 실제 서비스에 배포했을 때 이런 복잡한 요청을 받은 적 있나요? "내일 서울에서 부산으로 가는 가장 저렴한 교통편을 찾아주고, 부산 날씨를 고려해 짐싸기 추천도 해줘" 같은 멀티스텝 태스크 말이죠.
이런 문제는 단일 도구로는 해결할 수 없고, 여러 도구를 순차적으로 사용하면서 중간 결과를 계속 반영해야 합니다. 또한 각 단계에서 실패 가능성도 있어서, 에러 핸들링과 대안 전략도 필요합니다.
바로 이럴 때 진가를 발휘하는 것이 ReAct Prompting입니다. 복잡한 목표를 여러 하위 작업으로 분해하고, 각 작업마다 Thought-Action-Observation 사이클을 돌려가며 단계적으로 접근하면, 사람이 문제를 해결하는 것처럼 자연스럽고 견고한 시스템을 만들 수 있습니다.
개요
간단히 말해서, 이 실전 예제는 여행 계획이라는 복잡한 도메인에서 ReAct 패턴을 적용하여 실제로 작동하는 AI 에이전트를 구현합니다. 왜 이 예제가 중요한지 실무 관점에서 설명하면, 대부분의 비즈니스 유스케이스는 단순한 Q&A가 아니라 이런 복합적인 태스크이기 때문입니다.
예를 들어, 전자상거래에서 "30만원 예산으로 노트북 추천, 재고 확인, 할인 쿠폰 적용, 배송일 계산"을 모두 해야 하는 경우, 각 단계가 이전 결과에 의존하며 순차적으로 실행되어야 합니다. 기존의 단순 프롬프팅에서는 이런 복잡한 태스크를 하나의 거대한 프롬프트로 처리하려다 실패하거나 부정확한 결과를 냈다면, ReAct를 사용하면 각 하위 작업을 명확히 분리하고 중간 검증을 거쳐 최종 목표에 도달할 수 있습니다.
이 예제의 핵심 특징은 세 가지입니다. 첫째, 실제 외부 API 연동(교통편 검색, 날씨 API) 시뮬레이션.
둘째, 에러가 발생했을 때의 재시도 및 대안 전략. 셋째, 여러 도구의 결과를 종합하여 최종 추천을 생성하는 통합 로직.
이러한 특징들이 토이 프로젝트가 아닌 실제 프로덕션 레벨의 에이전트를 만들 때 필요한 실무 스킬을 보여줍니다.
코드 예제
# 복잡한 여행 계획 질문을 ReAct로 해결하는 예제
class TravelAgent:
"""여행 계획을 돕는 ReAct 에이전트"""
def __init__(self):
self.history = []
self.tools = {
"search_transport": self.search_transport,
"get_weather": self.get_weather,
"recommend_packing": self.recommend_packing
}
def solve(self, query):
"""복잡한 여행 질문을 단계적으로 해결"""
self.history.append(f"Query: {query}")
# Iteration 1: 교통편 검색
self.history.append("Thought: 먼저 서울-부산 교통편을 찾아야 함")
self.history.append("Action: search_transport['서울', '부산', '2025-11-17']")
result1 = self.search_transport("서울", "부산", "2025-11-17")
self.history.append(f"Observation: {result1}")
# Iteration 2: 날씨 확인
self.history.append("Thought: 교통편을 찾았으니 이제 부산 날씨를 확인해야 짐싸기 추천 가능")
self.history.append("Action: get_weather['부산', '2025-11-17']")
result2 = self.get_weather("부산", "2025-11-17")
self.history.append(f"Observation: {result2}")
# Iteration 3: 짐싸기 추천
self.history.append(f"Thought: 날씨가 {result2}이므로 그에 맞는 짐 추천")
self.history.append(f"Action: recommend_packing['{result2}']")
result3 = self.recommend_packing(result2)
self.history.append(f"Observation: {result3}")
# Final Answer
answer = f"가장 저렴한 교통편: {result1}, 날씨: {result2}, 짐싸기: {result3}"
self.history.append(f"Answer: {answer}")
return answer
def search_transport(self, origin, dest, date):
"""교통편 검색 API (시뮬레이션)"""
return "KTX 05:00 출발, 35,000원"
def get_weather(self, city, date):
"""날씨 API (시뮬레이션)"""
return "맑음, 18도"
def recommend_packing(self, weather):
"""날씨 기반 짐싸기 추천"""
if "맑음" in weather:
return "가벼운 외투, 선글라스 추천"
return "우산, 방수 재킷 추천"
설명
이것이 하는 일: 이 예제는 실제 서비스에서 발생할 법한 복합적인 사용자 요청을 ReAct 패턴으로 단계적으로 처리하는 전체 플로우를 보여줍니다. 코드를 단계별로 나누어 설명하면, 첫 번째 반복(Iteration 1)에서 AI는 "교통편을 먼저 찾아야 한다"고 생각하고, search_transport 도구를 호출합니다.
이 도구는 실제 환경에서는 네이버나 카카오 같은 교통 API를 호출할 것이며, 여기서는 "KTX 35,000원"이라는 결과를 반환합니다. 이 Observation이 히스토리에 추가됩니다.
그 다음으로, 두 번째 반복에서 AI는 첫 번째 Observation을 보고 "이제 날씨를 확인해야 짐 추천이 가능하다"고 판단합니다. 이것이 ReAct의 핵심입니다 - 이전 결과를 보고 다음 행동을 결정하는 적응적 사고.
get_weather를 호출해 "맑음, 18도"라는 Observation을 얻습니다. 세 번째 반복에서 AI는 날씨 정보를 활용하여 recommend_packing을 호출합니다.
"맑음"이라는 키워드를 감지한 함수는 "가벼운 외투, 선글라스"를 추천합니다. 만약 날씨가 "비"였다면 다른 추천을 했을 것입니다 - 이처럼 중간 결과에 따라 동적으로 분기하는 것이 ReAct의 강점입니다.
마지막으로, 모든 Observation을 종합하여 최종 Answer를 생성합니다. 이 답변은 세 가지 도구의 결과를 모두 포함하며, 사용자가 요청한 "교통편 + 날씨 + 짐싸기"를 완벽히 충족합니다.
여러분이 이 패턴을 사용하면 훨씬 복잡한 시나리오도 처리할 수 있습니다. 예를 들어 중간에 "교통편이 매진되었다"는 Observation이 오면, 다음 Thought에서 "대안 교통편을 찾아야겠다"고 판단하여 다른 도구를 호출할 수 있습니다.
또한 각 단계의 히스토리를 저장하므로, 나중에 "이 사용자는 보통 어떤 교통편을 선호하는가?" 같은 분석도 가능합니다. 이런 유연성과 추적 가능성이 ReAct를 실무에서 강력하게 만듭니다.
실전 팁
💡 실전에서는 각 도구 호출에 타임아웃과 재시도 로직을 추가하세요. 외부 API가 느리거나 실패할 수 있으므로, 3초 내에 응답 없으면 다음 대안으로 넘어가는 식의 견고함이 필요합니다.
💡 흔한 실수는 모든 도구를 순차적으로만 호출하는 것입니다. 경우에 따라 병렬 호출이 가능합니다. 예를 들어 날씨와 교통편 검색은 서로 독립적이므로 동시에 실행하고 결과를 합치면 응답 시간을 절반으로 줄일 수 있습니다.
💡 보안을 위해 사용자 입력을 도구 파라미터로 넘기기 전에 검증하세요. SQL Injection처럼 악의적인 입력이 "부산'; DROP TABLE--" 같은 형태로 들어올 수 있으므로, 화이트리스트 검증이나 파라미터 이스케이프가 필수입니다.
💡 디버깅할 때는 self.history를 JSON으로 저장하여 시각화 도구로 분석하세요. "어느 단계에서 시간이 가장 많이 걸렸는가?", "어떤 도구가 가장 자주 실패하는가?" 같은 인사이트를 얻을 수 있습니다.
💡 더 발전된 사용법으로 사용자 피드백을 ReAct 루프에 통합할 수 있습니다. 예를 들어 중간 단계에서 "이 교통편은 너무 비싸요"라는 피드백이 오면, 새로운 Thought "더 저렴한 옵션 찾기"를 추가하여 재탐색하는 인터랙티브 에이전트를 만들 수 있습니다.
6. 에러 처리와 재시도 전략 - 견고한 에이전트 만들기
시작하며
여러분이 ReAct 에이전트를 만들어서 테스트할 때 이런 상황을 겪어본 적 있나요? 외부 API가 갑자기 타임아웃되거나, 예상치 못한 형식의 데이터가 들어와서 전체 에이전트가 멈춰버리는 경우 말이죠.
이런 문제는 실제 프로덕션 환경에서 필연적으로 발생합니다. 네트워크 불안정, API 장애, 잘못된 사용자 입력 등 수많은 예외 상황이 있고, 이를 처리하지 않으면 사용자 경험이 크게 떨어집니다.
바로 이럴 때 필요한 것이 체계적인 에러 처리와 재시도 전략입니다. 각 도구 실행마다 실패 가능성을 고려하고, 실패 시 자동으로 대안을 찾거나 재시도하며, 최악의 경우에도 사용자에게 명확한 에러 메시지를 제공하는 회복력 있는 시스템을 만들어야 합니다.
개요
간단히 말해서, 에러 처리와 재시도 전략은 ReAct 에이전트가 예외 상황에서도 안정적으로 작동하도록 만드는 방어 메커니즘입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 사용자는 AI 에이전트가 항상 완벽하게 작동하기를 기대하지만, 현실에서는 의존하는 외부 시스템들이 99.9% 이상의 가용성을 보장하지 못합니다.
예를 들어, 결제 처리 에이전트가 결제 게이트웨이 장애로 멈춰버리면 사용자는 돈이 빠져나갔는지도 모르는 최악의 상황이 됩니다. 적절한 에러 처리로 "결제 실패, 재시도 중..."이라고 알리고 자동으로 재시도하면 훨씬 나은 경험을 제공할 수 있습니다.
기존에는 try-except만 사용해서 에러를 단순히 로깅하고 종료했다면, 이제는 ReAct의 Thought-Action-Observation 구조 안에서 에러도 하나의 Observation으로 처리하여, AI가 "에러가 발생했으니 다른 접근을 시도해야겠다"고 판단하도록 만들 수 있습니다. 에러 처리 전략의 핵심 특징은 세 가지입니다.
첫째, 지수 백오프(Exponential Backoff) 재시도 - 1초, 2초, 4초 간격으로 재시도하여 일시적 장애 극복. 둘째, 폴백(Fallback) 도구 - 주 도구가 실패하면 자동으로 대체 도구 사용.
셋째, 우아한 실패(Graceful Degradation) - 일부 기능이 실패해도 나머지는 계속 작동하도록 격리. 이러한 특징들이 사용자에게 신뢰할 수 있는 서비스를 제공하는 핵심입니다.
코드 예제
# 에러 처리와 재시도를 포함한 견고한 ReAct 에이전트
import time
from typing import Optional
class RobustReActAgent:
"""에러 처리와 재시도 기능이 있는 ReAct 에이전트"""
def __init__(self, max_retries=3):
self.max_retries = max_retries
self.history = []
def execute_tool_with_retry(self, tool_name, *args):
"""지수 백오프 재시도를 사용한 도구 실행"""
for attempt in range(self.max_retries):
try:
result = self._call_tool(tool_name, *args)
return f"Observation: 성공 - {result}"
except Exception as e:
wait_time = 2 ** attempt # 1초, 2초, 4초...
self.history.append(
f"Observation: 시도 {attempt+1} 실패 - {str(e)}, "
f"{wait_time}초 후 재시도"
)
if attempt < self.max_retries - 1:
time.sleep(wait_time)
else:
# 최종 실패 시 폴백 전략
return self._fallback_strategy(tool_name, e)
return "Observation: 모든 재시도 실패"
def _fallback_strategy(self, tool_name, error):
"""도구 실패 시 대안 찾기"""
fallback_map = {
"search_api": "search_backup_api",
"payment_gateway": "alternative_payment"
}
if tool_name in fallback_map:
backup_tool = fallback_map[tool_name]
self.history.append(
f"Thought: {tool_name} 실패, {backup_tool}로 전환"
)
try:
result = self._call_tool(backup_tool)
return f"Observation: 폴백 성공 - {result}"
except Exception as e:
return f"Observation: 폴백도 실패 - {str(e)}"
return f"Observation: 사용 가능한 대안 없음 - {str(error)}"
def _call_tool(self, tool_name, *args):
"""실제 도구 호출 (시뮬레이션)"""
# 실전에서는 여기서 실제 API 호출
if tool_name == "unreliable_api":
import random
if random.random() < 0.7: # 70% 확률로 실패
raise Exception("API 타임아웃")
return "도구 실행 결과"
설명
이것이 하는 일: 이 코드는 ReAct 에이전트가 외부 의존성의 실패로부터 회복할 수 있도록 다층 방어 메커니즘을 제공합니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 execute_tool_with_retry 메서드가 최대 3회까지 도구 실행을 시도합니다.
첫 시도가 실패하면 1초를 기다리고, 두 번째 실패하면 2초, 세 번째는 4초를 기다립니다. 이 "지수 백오프" 패턴은 일시적인 네트워크 혼잡이나 서버 부하가 해소될 시간을 주어 성공률을 크게 높입니다.
그 다음으로, 모든 재시도가 실패하면 _fallback_strategy가 호출됩니다. 이 메서드는 실패한 도구의 대체 도구를 찾아서 자동으로 전환합니다.
예를 들어 메인 검색 API가 다운되면 백업 검색 API로 전환하여 사용자는 서비스 중단을 느끼지 못하게 합니다. fallback_map을 설정 파일로 분리하면 운영 중에도 유연하게 대체 전략을 변경할 수 있습니다.
세 번째로, 각 실패와 재시도 과정이 모두 history에 기록됩니다. 이는 단순한 로깅이 아니라 ReAct의 Observation으로 작용하여, AI가 "이 도구는 불안정하니 다음부터는 다른 도구를 먼저 시도해야겠다"는 학습을 할 수 있게 합니다.
프로덕션 환경에서는 이 히스토리를 분석해 어떤 도구가 가장 자주 실패하는지, 어느 시간대에 문제가 많은지 등을 파악할 수 있습니다. 여러분이 이 패턴을 사용하면 에이전트의 신뢰성이 극적으로 향상됩니다.
단일 장애 지점(Single Point of Failure)을 제거하여 한 도구의 문제가 전체 시스템을 멈추지 않도록 하고, 사용자에게는 항상 "최선을 다해 작업을 완수하려는" 인상을 줍니다. 또한 재시도와 폴백 과정을 투명하게 보여줌으로써 디버깅이 쉬워지고, 시스템의 회복력을 계속 개선할 수 있는 데이터를 얻게 됩니다.
실전 팁
💡 재시도 횟수는 도구의 특성에 따라 다르게 설정하세요. 빠른 API는 5회까지, 느린 API는 2회만 재시도하는 식으로 조정하면 불필요한 대기 시간을 줄일 수 있습니다.
💡 흔한 실수는 모든 에러를 동일하게 처리하는 것입니다. 400 Bad Request는 재시도해도 소용없지만, 503 Service Unavailable은 재시도할 가치가 있습니다. 에러 코드에 따라 전략을 달리하세요.
💡 성능 최적화를 위해 Circuit Breaker 패턴을 추가하세요. 특정 도구가 계속 실패하면 일정 시간 동안 호출을 차단하고 바로 폴백으로 넘어가서, 실패가 확실한 도구에 시간을 낭비하지 않도록 합니다.
💡 디버깅할 때는 각 재시도와 폴백을 별도의 로그 레벨로 기록하세요. INFO: 정상 재시도, WARNING: 폴백 사용, ERROR: 완전 실패 같은 식으로 구분하면 모니터링 대시보드에서 시스템 건강도를 한눈에 파악할 수 있습니다.
💡 더 발전된 사용법으로 머신러닝을 활용한 동적 재시도 전략이 있습니다. 과거 데이터를 학습하여 "이 시간대에는 이 API가 자주 실패하니 재시도 간격을 늘리자" 같은 적응적 로직을 만들 수 있습니다.
7. 프롬프트 엔지니어링 - ReAct 패턴 최적화
시작하며
여러분이 ReAct 에이전트를 만들었는데 AI가 엉뚱한 도구를 선택하거나, Thought 단계를 건너뛰고 바로 Action을 실행하는 등 의도대로 작동하지 않는 경험 있으셨나요? 이런 문제는 ReAct 구조는 잘 설계했지만, AI에게 "어떻게 행동해야 하는지"를 명확히 알려주는 프롬프트가 부족해서 발생합니다.
AI는 여러분이 원하는 ReAct 패턴을 자동으로 따르지 않으며, 명시적인 지시와 예시가 필요합니다. 바로 이럴 때 필요한 것이 ReAct 전용 프롬프트 엔지니어링입니다.
AI가 일관되게 Thought, Action, Observation 형식을 따르도록 템플릿을 제공하고, Few-shot 예시로 학습시키며, 경계 조건을 명확히 하는 프롬프트를 작성해야 합니다.
개요
간단히 말해서, ReAct 프롬프트 엔지니어링은 AI가 ReAct 패턴을 정확히 이해하고 실행하도록 유도하는 지시문과 예시를 설계하는 기술입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 같은 ReAct 코드라도 프롬프트 품질에 따라 성능이 50%에서 95%까지 크게 달라지기 때문입니다.
예를 들어, "도구를 사용해서 답해줘"라는 모호한 지시 대신 "먼저 Thought로 생각을 정리하고, Action으로 도구를 호출하며, Observation을 보고 다시 생각하라"고 구체적으로 지시하면 AI의 정확도가 몇 배 향상됩니다. 기존에는 프롬프트를 즉흥적으로 작성했다면, 이제는 ReAct 논문과 베스트 프랙티스를 참고하여 체계적으로 설계합니다.
특히 Few-shot Learning(몇 가지 예시 제공)이 ReAct에서 매우 효과적입니다. 프롬프트 최적화의 핵심 특징은 세 가지입니다.
첫째, 명확한 형식 지정 - "Thought:", "Action:", "Observation:"을 정확히 어떻게 쓸지 예시로 보여주기. 둘째, Few-shot 예시 제공 - 실제 성공 케이스 2-3개를 프롬프트에 포함하여 패턴 학습.
셋째, 제약 조건 명시 - "Action 없이 Answer를 내면 안 됨", "최대 5회 반복" 같은 규칙 설정. 이러한 특징들이 AI를 "추측하는 모델"에서 "정확히 따르는 에이전트"로 변화시킵니다.
코드 예제
# ReAct 패턴을 위한 최적화된 프롬프트 템플릿
REACT_PROMPT_TEMPLATE = """
당신은 ReAct (Reasoning + Acting) 패턴을 따르는 AI 에이전트입니다.
**반드시 다음 형식을 따르세요:**
Question: [사용자 질문]
Thought: [현재 상황 분석, 다음에 할 일 계획]
Action: tool_name[param1, param2, ...]
Observation: [도구 실행 결과]
Thought: [결과 분석, 추가 행동 필요 여부 판단]
Action: ... (필요시 반복)
Thought: [충분한 정보를 모았으므로 답변 가능]
Answer: [최종 답변]
**사용 가능한 도구:**
- search[query]: 웹에서 정보 검색
- calculate[expression]: 수식 계산
- weather[city, date]: 날씨 정보 조회
**중요 규칙:**
4. Observation이 에러면 다른 접근을 시도하세요
설명
이것이 하는 일: 이 프롬프트 템플릿은 AI에게 "무엇을 해야 하는지"뿐 아니라 "어떻게 해야 하는지"까지 구체적으로 가르칩니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 프롬프트 맨 위에서 AI의 역할을 명확히 정의합니다.
"당신은 ReAct 패턴을 따르는 에이전트"라고 선언함으로써 AI가 일반적인 대화가 아닌 구조화된 추론을 해야 함을 인지하게 합니다. 그 다음으로, "반드시 다음 형식을 따르세요" 섹션에서 Thought, Action, Observation, Answer의 정확한 형식을 보여줍니다.
단순히 "이렇게 써라"가 아니라 각 단계가 어떤 정보를 담아야 하는지([현재 상황 분석, 다음에 할 일 계획] 등) 구체적으로 안내합니다. 이렇게 하면 AI가 "Thought: 검색해야겠다" 같은 너무 짧은 출력 대신, "Thought: 사용자가 X를 물었는데 내 지식으로는 모르므로 Y 도구를 사용해야 함"처럼 풍부한 추론을 생성합니다.
세 번째로, "사용 가능한 도구" 섹션에서 각 도구의 시그니처를 명시합니다. 이는 AI가 "Action: search[...]"를 쓸 때 정확한 파라미터 형식을 사용하도록 돕습니다.
실전에서는 각 도구의 입력/출력 예시도 추가하면 더욱 효과적입니다. 네 번째로, "중요 규칙" 섹션이 경계 조건을 설정합니다.
"Action 없이 바로 Answer 하지 마세요"는 AI가 추측으로 답하는 것을 방지하고 반드시 도구를 사용하도록 강제합니다. "최대 5회 반복"은 무한 루프를 막습니다.
마지막으로, Few-shot 예시가 가장 강력한 학습 도구입니다. 실제 질문-답변 시나리오를 보여줌으로써 AI는 패턴을 내재화합니다.
2-3개의 예시만으로도 정확도가 크게 향상되며, 예시는 실제 사용 케이스와 유사할수록 좋습니다. 여러분이 이 템플릿을 사용하면 AI의 일관성과 정확도가 크게 향상됩니다.
프롬프트를 매번 새로 작성하는 대신 검증된 템플릿을 재사용하여 개발 속도가 빨라지고, 팀원들도 동일한 프롬프트를 사용해 결과의 재현성이 보장됩니다. 또한 프롬프트를 버전 관리하면 어떤 변경이 성능을 개선했는지 A/B 테스트로 측정할 수 있습니다.
실전 팁
💡 Few-shot 예시는 단순한 것부터 복잡한 것 순으로 배치하세요. 첫 예시는 1-2회 반복, 두 번째는 3-4회 반복으로 점진적으로 난이도를 높이면 AI가 패턴을 더 잘 학습합니다.
💡 흔한 실수는 예시에 성공 케이스만 포함하는 것입니다. 실패 케이스와 회복 과정도 보여주세요. "Observation: 에러 발생 → Thought: 다른 도구 시도" 같은 예시가 견고성을 높입니다.
💡 토큰 비용 절감을 위해 프롬프트를 압축할 수 있지만, 핵심 구조는 유지하세요. "반드시 다음 형식을 따르세요" 부분은 절대 생략하면 안 됩니다. 대신 예시를 2개에서 1개로 줄이는 등 다른 부분을 최적화하세요.
💡 디버깅할 때는 AI 출력이 템플릿을 얼마나 잘 따르는지 자동으로 검증하는 스크립트를 만드세요. 정규표현식으로 "Thought:", "Action:" 등이 올바른 순서로 나타나는지 체크하여 프롬프트 품질을 측정할 수 있습니다.
💡 더 발전된 사용법으로 도메인별 프롬프트 변형을 만들 수 있습니다. 의료, 금융, 법률 등 분야마다 다른 예시와 도구를 포함한 특화된 템플릿을 관리하면, 각 도메인에서 더 높은 정확도를 얻을 수 있습니다.
8. 멀티 에이전트 협업 - ReAct 에이전트 조율하기
시작하며
여러분이 하나의 복잡한 프로젝트를 AI에게 맡기려 할 때 이런 문제를 겪어본 적 있나요? 단일 에이전트가 너무 많은 책임을 지다 보니 성능이 떨어지거나, 각 단계에서 전문성이 부족해 결과 품질이 낮아지는 상황 말이죠.
이런 문제는 모든 작업을 하나의 에이전트에 몰아주려 할 때 발생합니다. 실제 조직처럼 각 전문가가 자기 영역에 집중하고 협업하는 구조가 더 효율적입니다.
바로 이럴 때 필요한 것이 멀티 에이전트 ReAct 시스템입니다. 여러 전문화된 에이전트(검색 전문가, 계산 전문가, 요약 전문가 등)를 만들고, 각 에이전트가 ReAct 패턴으로 작업한 결과를 서로 공유하며 협업하도록 조율하면 훨씬 강력한 시스템을 만들 수 있습니다.
개요
간단히 말해서, 멀티 에이전트 ReAct는 여러 전문화된 에이전트가 각자의 ReAct 루프를 실행하면서 결과를 공유하고 조율하는 협업 시스템입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 복잡한 비즈니스 문제는 여러 전문 영역을 동시에 다뤄야 하기 때문입니다.
예를 들어, "투자 추천" 태스크는 시장 데이터 분석, 리스크 평가, 법률 검토를 모두 필요로 하며, 각각을 전문 에이전트에게 맡기고 결과를 종합하는 것이 효율적입니다. 기존에는 하나의 거대한 프롬프트로 모든 것을 처리하려 했다면, 이제는 작업을 분해하여 각 에이전트에게 위임하고, 에이전트 간 통신 프로토콜을 정의하며, 중앙 조율자가 전체 워크플로우를 관리하도록 만듭니다.
멀티 에이전트 시스템의 핵심 특징은 세 가지입니다. 첫째, 전문화(Specialization) - 각 에이전트는 특정 도메인의 도구와 지식만 가짐.
둘째, 통신 프로토콜 - 에이전트 간 결과를 주고받는 표준 형식 정의. 셋째, 조율 메커니즘 - 어느 에이전트를 언제 호출할지 결정하는 오케스트레이터.
이러한 특징들이 확장 가능하고 유지보수하기 쉬운 AI 시스템을 만듭니다.
코드 예제
# 멀티 에이전트 ReAct 시스템 구현
class SpecializedAgent:
"""특정 도메인에 전문화된 에이전트"""
def __init__(self, name, tools):
self.name = name
self.tools = tools
self.history = []
def solve(self, task):
"""ReAct 패턴으로 태스크 해결"""
self.history.append(f"Thought: {self.name}으로서 {task}를 처리")
# 실제로는 여기서 ReAct 루프 실행
result = f"{self.name}의 결과: {task} 완료"
self.history.append(f"Answer: {result}")
return result
class MultiAgentOrchestrator:
"""여러 에이전트를 조율하는 오케스트레이터"""
def __init__(self):
# 전문화된 에이전트들 생성
self.agents = {
"researcher": SpecializedAgent("연구 에이전트", ["search", "fetch"]),
"analyst": SpecializedAgent("분석 에이전트", ["calculate", "visualize"]),
"writer": SpecializedAgent("작성 에이전트", ["summarize", "format"])
}
self.shared_memory = {} # 에이전트 간 공유 메모리
def solve_complex_task(self, task):
"""복잡한 태스크를 여러 에이전트에게 분배"""
results = []
# Step 1: 연구 에이전트가 정보 수집
print("Thought: 먼저 정보를 수집해야 함")
research_result = self.agents["researcher"].solve(f"{task} 정보 수집")
self.shared_memory["research"] = research_result
results.append(research_result)
# Step 2: 분석 에이전트가 연구 결과 분석
print("Thought: 수집된 정보를 분석해야 함")
analysis_input = f"다음 정보를 분석: {self.shared_memory['research']}"
analysis_result = self.agents["analyst"].solve(analysis_input)
self.shared_memory["analysis"] = analysis_result
results.append(analysis_result)
# Step 3: 작성 에이전트가 최종 보고서 작성
print("Thought: 분석 결과를 보고서로 작성해야 함")
write_input = f"다음을 요약: {self.shared_memory['analysis']}"
final_report = self.agents["writer"].solve(write_input)
results.append(final_report)
return {
"final_answer": final_report,
"intermediate_results": results,
"agent_histories": {name: agent.history for name, agent in self.agents.items()}
}
설명
이것이 하는 일: 멀티 에이전트 시스템은 복잡한 문제를 전문가 팀처럼 분업하여 처리하는 구조를 만듭니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 SpecializedAgent 클래스는 각 에이전트의 전문성을 정의합니다.
"연구 에이전트"는 검색과 정보 수집 도구만 가지고, "분석 에이전트"는 계산과 시각화 도구만 가집니다. 이렇게 책임을 분리하면 각 에이전트의 프롬프트를 최적화하기 쉽고, 한 영역의 변경이 다른 영역에 영향을 주지 않습니다.
그 다음으로, MultiAgentOrchestrator가 전체 워크플로우를 관리합니다. solve_complex_task 메서드는 "누구에게 무엇을 언제 시킬지"를 결정하는 마스터 ReAct 루프라고 볼 수 있습니다.
오케스트레이터 자신도 Thought(어떤 에이전트가 필요한가?)를 하고, Action(에이전트 호출)을 하며, Observation(에이전트 결과)을 받아 다음 단계를 결정합니다. 세 번째로, shared_memory 딕셔너리가 에이전트 간 통신을 가능하게 합니다.
연구 에이전트가 찾은 정보를 여기에 저장하면, 분석 에이전트가 이를 읽어서 작업을 이어갑니다. 실전에서는 Redis나 데이터베이스를 사용해 더 견고한 공유 메모리를 구현할 수 있으며, 메시지 큐를 사용해 비동기 협업도 가능합니다.
마지막으로, 모든 에이전트의 히스토리를 수집하여 반환합니다. 이는 디버깅과 감사에 매우 유용하며, "왜 이런 결론이 나왔는가?"를 추적할 때 각 에이전트의 추론 과정을 모두 볼 수 있습니다.
여러분이 이 패턴을 사용하면 시스템을 수평적으로 확장할 수 있습니다. 새로운 전문 영역이 필요하면 그냥 새 에이전트를 추가하면 되고, 기존 에이전트를 건드릴 필요가 없습니다.
또한 각 에이전트를 병렬로 실행할 수 있어, 예를 들어 "연구"와 "법률 검토"를 동시에 진행하고 결과를 합치면 전체 응답 시간을 크게 줄일 수 있습니다. 이런 확장성과 성능이 프로덕션 AI 시스템에서 멀티 에이전트 아키텍처가 인기 있는 이유입니다.
실전 팁
💡 에이전트 간 의존성을 명확히 문서화하세요. "분석 에이전트는 반드시 연구 에이전트 후에 실행"같은 DAG(방향성 비순환 그래프)를 그려서 데드락을 방지하세요.
💡 흔한 실수는 모든 에이전트에게 모든 정보를 공유하는 것입니다. 필요한 정보만 선택적으로 전달하면 프롬프트 길이와 비용을 줄이고, 각 에이전트가 자기 일에 집중할 수 있습니다.
💡 성능 최적화를 위해 에이전트 풀을 만들어 재사용하세요. 매번 새 에이전트를 생성하는 대신, 미리 초기화된 에이전트를 풀에서 꺼내 쓰고 반환하면 초기화 오버헤드를 제거할 수 있습니다.
💡 디버깅할 때는 각 에이전트의 입출력을 시각화하는 도구를 만드세요. Mermaid 다이어그램으로 "연구 → 분석 → 작성" 흐름을 그리고, 각 화살표에 전달된 데이터를 표시하면 문제를 쉽게 찾을 수 있습니다.
💡 더 발전된 사용법으로 동적 에이전트 선택이 있습니다. 고정된 워크플로우 대신, 오케스트레이터가 태스크 특성을 분석해 "이 문제는 법률 에이전트가 필요 없다"고 판단하여 불필요한 에이전트를 건너뛰는 적응적 시스템을 만들 수 있습니다.
9. 성능 측정과 최적화 - ReAct 에이전트 개선하기
시작하며
여러분이 ReAct 에이전트를 배포했는데 사용자들이 "응답이 너무 느려요" 또는 "가끔 이상한 답을 해요"라는 피드백을 준다면 어떻게 개선하시겠어요? 이런 문제는 시스템의 성능을 객관적으로 측정하고 병목 지점을 찾지 않으면 해결하기 어렵습니다.
추측으로 "이 부분이 느릴 것 같다"고 최적화하면 실제로는 효과가 없거나 오히려 다른 문제를 일으킬 수 있습니다. 바로 이럴 때 필요한 것이 체계적인 성능 측정과 데이터 기반 최적화입니다.
각 단계의 실행 시간, 도구 호출 횟수, 정확도, 비용 등을 추적하고, 이를 분석하여 실제 병목을 찾아 개선해야 합니다.
개요
간단히 말해서, ReAct 에이전트의 성능 측정은 응답 시간, 정확도, 비용 등의 메트릭을 수집하고 분석하여 최적화 포인트를 찾는 과정입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 프로덕션 환경에서는 사용자 경험과 운영 비용이 직결되기 때문입니다.
예를 들어, 응답 시간이 10초에서 3초로 줄어들면 사용자 만족도가 크게 올라가고, 불필요한 도구 호출을 줄이면 API 비용이 절반으로 떨어질 수 있습니다. 기존에는 "대충 빨라 보이네" 같은 주관적 판단으로 최적화했다면, 이제는 각 Thought-Action-Observation 사이클의 시간을 밀리초 단위로 측정하고, A/B 테스트로 프롬프트 변경의 효과를 정량화하며, 사용자 피드백을 메트릭으로 전환하여 지속적으로 개선합니다.
성능 최적화의 핵심 특징은 세 가지입니다. 첫째, 포괄적 메트릭 수집 - 시간, 토큰 사용량, 도구 호출 횟수, 정확도 등을 모든 요청마다 기록.
둘째, 병목 분석 - 전체 시간의 80%를 차지하는 20% 작업 찾기(파레토 법칙). 셋째, 점진적 개선 - 한 번에 하나씩 변경하고 효과를 측정하여 되돌릴 수 있도록 함.
이러한 체계적 접근이 지속 가능한 성능 향상을 가능하게 합니다.
코드 예제
# ReAct 에이전트의 성능을 측정하고 최적화하는 시스템
import time
from typing import Dict, List
from dataclasses import dataclass
from statistics import mean, median
@dataclass
class PerformanceMetrics:
"""성능 메트릭을 저장하는 데이터 클래스"""
total_time: float
thought_time: float
action_time: float
observation_time: float
num_iterations: int
num_tool_calls: int
tokens_used: int
success: bool
class InstrumentedReActAgent:
"""성능 측정 기능이 내장된 ReAct 에이전트"""
def __init__(self):
self.metrics_history: List[PerformanceMetrics] = []
def solve_with_metrics(self, question):
"""성능을 측정하면서 질문 해결"""
start_time = time.time()
thought_time = 0
action_time = 0
observation_time = 0
iterations = 0
tool_calls = 0
for i in range(5): # 최대 5회 반복
iterations += 1
# Thought 단계 측정
t_start = time.time()
thought = self._generate_thought(question)
thought_time += time.time() - t_start
if "Answer:" in thought:
break
# Action 단계 측정
a_start = time.time()
action_result = self._execute_action(thought)
action_time += time.time() - a_start
tool_calls += 1
# Observation 단계 측정
o_start = time.time()
observation = self._process_observation(action_result)
observation_time += time.time() - o_start
total_time = time.time() - start_time
# 메트릭 저장
metrics = PerformanceMetrics(
total_time=total_time,
thought_time=thought_time,
action_time=action_time,
observation_time=observation_time,
num_iterations=iterations,
num_tool_calls=tool_calls,
tokens_used=self._estimate_tokens(),
success=True
)
self.metrics_history.append(metrics)
return thought, metrics
def get_performance_report(self) -> Dict:
"""성능 리포트 생성"""
if not self.metrics_history:
return {"error": "메트릭 없음"}
return {
"평균_응답_시간": mean([m.total_time for m in self.metrics_history]),
"중앙값_응답_시간": median([m.total_time for m in self.metrics_history]),
"평균_반복_횟수": mean([m.num_iterations for m in self.metrics_history]),
"평균_도구_호출": mean([m.num_tool_calls for m in self.metrics_history]),
"성공률": sum([m.success for m in self.metrics_history]) / len(self.metrics_history),
"병목_단계": self._identify_bottleneck()
}
def _identify_bottleneck(self) -> str:
"""가장 시간을 많이 쓰는 단계 식별"""
avg_thought = mean([m.thought_time for m in self.metrics_history])
avg_action = mean([m.action_time for m in self.metrics_history])
avg_obs = mean([m.observation_time for m in self.metrics_history])
bottleneck_times = {
"Thought": avg_thought,
"Action": avg_action,
"Observation": avg_obs
}
return max(bottleneck_times, key=bottleneck_times.get)
def _generate_thought(self, question):
"""Thought 생성 (시뮬레이션)"""
time.sleep(0.1) # LLM 호출 시뮬레이션
return "Thought: 검색 필요"
def _execute_action(self, thought):
"""Action 실행 (시뮬레이션)"""
time.sleep(0.5) # API 호출 시뮬레이션
return "결과 데이터"
def _process_observation(self, result):
"""Observation 처리 (시뮬레이션)"""
time.sleep(0.05)
return f"Observation: {result}"
def _estimate_tokens(self):
"""토큰 사용량 추정"""
return 500 # 실제로는 tiktoken 등으로 계산
설명
이것이 하는 일: 이 코드는 ReAct 에이전트의 모든 작업을 계측(instrumentation)하여 성능 데이터를 수집하고 분석 가능하게 만듭니다. 코드를 단계별로 나누어 설명하면, 첫 번째로 solve_with_metrics 메서드가 각 단계 전후에 시간을 측정합니다.
Python의 time.time()을 사용해 밀리초 단위로 정확한 시간을 기록하며, Thought, Action, Observation 각각에 소요된 시간을 분리하여 추적합니다. 이렇게 세분화된 측정이 "어느 단계가 느린가?"를 정확히 알려줍니다.
그 다음으로, PerformanceMetrics 데이터클래스에 모든 메트릭을 구조화하여 저장합니다. 시간뿐 아니라 반복 횟수, 도구 호출 횟수, 토큰 사용량, 성공 여부까지 기록하여 다각도로 성능을 분석할 수 있습니다.
실전에서는 여기에 사용자 만족도(thumbs up/down), 오류 유형 등 비즈니스 메트릭도 추가합니다. 세 번째로, get_performance_report 메서드가 수집된 메트릭을 집계합니다.
평균뿐 아니라 중앙값도 계산하여 이상치(outlier)의 영향을 줄이고, _identify_bottleneck으로 가장 시간을 많이 쓰는 단계를 자동으로 찾아냅니다. 예를 들어 "Action 단계가 전체 시간의 70%를 차지"하면, 도구 실행 최적화에 집중해야 함을 알 수 있습니다.
마지막으로, 이 메트릭 히스토리를 사용해 트렌드를 분석할 수 있습니다. 시간에 따라 성능이 나빠지는가(성능 회귀)?
특정 유형의 질문이 느린가? 프롬프트 변경 전후로 정확도가 변했는가?
등을 데이터로 답할 수 있습니다. 여러분이 이 패턴을 사용하면 "느낌"이 아닌 "데이터"로 최적화 결정을 내릴 수 있습니다.
예를 들어 캐싱을 추가했을 때 실제로 응답 시간이 30% 감소했는지 측정하고, 효과가 없으면 되돌릴 수 있습니다. 또한 성능 리포트를 대시보드에 연결하면 실시간으로 시스템 건강도를 모니터링하고, 성능 저하 시 즉시 알림을 받을 수 있습니다.
이런 관측 가능성(Observability)이 안정적인 프로덕션 서비스의 핵심입니다.
실전 팁
💡 측정 오버헤드를 최소화하세요. time.time() 호출 자체도 시간이 걸리므로, 개발 환경에서는 상세 측정, 프로덕션에서는 샘플링(예: 10% 요청만 측정)하는 식으로 조절하세요.
💡 흔한 실수는 평균만 보는 것입니다. P50(중앙값), P95, P99 백분위수를 함께 보세요. 평균 2초여도 P99가 30초라면 일부 사용자는 매우 나쁜 경험을 하고 있다는 뜻입니다.
💡 성능 최적화 시 "조기 최적화는 악의 근원"을 기억하세요. 병목이 확인된 부분만 최적화하고, 추측으로 코드를 복잡하게 만들지 마세요. 측정 → 병목 식별 → 최적화 → 재측정 사이클을 따르세요.
💡 디버깅할 때는 느린 요청의 트레이스를 저장하세요. "왜 이 특정 요청이 10초나 걸렸는가?"를 재현하려면 그 요청의 모든 단계를 저장해야 합니다. OpenTelemetry 같은 분산 추적 도구를 사용하면 더욱 강력합니다.
💡 더 발전된 사용법으로 자동 최적화 시스템을 만들 수 있습니다. 예를 들어 "Action 시간이 3초 이상이면 자동으로 타임아웃 줄이기" 같은 규칙 기반 또는 ML 기반 자동 튜닝을 적용하여 사람 개입 없이 성능을 유지할 수 있습니다.
10. 실무 배포 가이드 - 프로덕션 환경에서의 ReAct
시작하며
여러분이 ReAct 에이전트를 완성하고 실제 서비스에 배포하려 할 때, 개발 환경에서는 잘 작동하던 것이 프로덕션에서 문제를 일으키는 경험 있으셨나요? 이런 문제는 개발과 프로덕션 환경의 차이 때문에 발생합니다.
트래픽 규모, 보안 요구사항, 가용성 기대치, 비용 제약 등이 모두 다르며, 이를 고려하지 않으면 배포 후 심각한 장애를 겪을 수 있습니다. 바로 이럴 때 필요한 것이 프로덕션 배포 체크리스트와 베스트 프랙티스입니다.
보안, 확장성, 모니터링, 비용 관리 등 실무에서 반드시 고려해야 할 요소들을 체계적으로 준비하면 안정적인 서비스를 제공할 수 있습니다.
개요
간단히 말해서, ReAct 에이전트의 프로덕션 배포는 개발 환경에서 작동하는 에이전트를 실제 사용자를 대상으로 안정적이고 안전하게 서비스하기 위한 전반적인 준비 과정입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하면, 프로덕션 장애는 단순히 기술적 문제를 넘어 비즈니스 신뢰와 수익에 직접 영향을 미치기 때문입니다.
예를 들어, 고객 서비스 챗봇이 개인정보를 유출하거나, 갑작스러운 트래픽 증가로 다운되면 고객 이탈과 법적 문제가 발생할 수 있습니다. 기존에는 "일단 배포하고 문제 생기면 고치자"는 접근이었다면, 이제는 배포 전에 보안 감사, 부하 테스트, 롤백 계획, 모니터링 설정 등을 완료하고, 단계적 배포(카나리, 블루-그린)로 리스크를 최소화하며, 사고 대응 플레이북을 준비하는 성숙한 엔지니어링 프랙티스를 따릅니다.
프로덕션 배포의 핵심 요소는 다섯 가지입니다. 첫째, 보안 - API 키 관리, 입력 검증, 출력 필터링.
둘째, 확장성 - 로드 밸런싱, 오토스케일링, 속도 제한. 셋째, 관측성 - 로깅, 메트릭, 알림.
넷째, 비용 관리 - 토큰 사용량 모니터링, 캐싱, 배치 처리. 다섯째, 복구력 - 백업, 롤백, 재해 복구.
이러한 요소들이 24/7 안정적인 서비스를 가능하게 합니다.
코드 예제
# 프로덕션용 ReAct 에이전트 배포 예제
import os
import logging
from functools import lru_cache
from typing import Optional
# 1. 환경 변수로 민감 정보 관리
class ProductionReActAgent:
"""프로덕션 환경에 최적화된 ReAct 에이전트"""
def __init__(self):
# 보안: 환경 변수에서 API 키 로드 (코드에 하드코딩 금지)
self.api_key = os.getenv("OPENAI_API_KEY")
if not self.api_key:
raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않음")
# 로깅 설정 (민감 정보 자동 마스킹)
self.logger = self._setup_logger()
# 속도 제한 (DDoS 방지)
self.rate_limiter = RateLimiter(max_requests=100, window=60)
# 캐싱 (비용 절감)
self.cache_enabled = True
def _setup_logger(self):
"""프로덕션용 로거 설정"""
logger = logging.getLogger("ReActAgent")
logger.setLevel(logging.INFO)
# 파일과 콘솔 모두에 로그
handler = logging.FileHandler("/var/log/react-agent.log")
handler.setFormatter(
logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
)
logger.addHandler(handler)
return logger
@lru_cache(maxsize=1000)
def _cached_tool_call(self, tool_name, params_tuple):
"""자주 사용되는 도구 결과 캐싱"""
# params_tuple: 튜플로 변환한 파라미터 (해시 가능하도록)
return self._execute_tool(tool_name, *params_tuple)
def solve(self, question: str, user_id: Optional[str] = None) -> dict:
"""안전하고 관측 가능한 방식으로 질문 해결"""
# 속도 제한 확인
if not self.rate_limiter.allow(user_id):
self.logger.warning(f"속도 제한 초과: user_id={user_id}")
return {"error": "너무 많은 요청, 잠시 후 다시 시도하세요"}
# 입력 검증 및 새니타이징
if not self._validate_input(question):
self.logger.error(f"부적절한 입력 감지: {question[:50]}")
return {"error": "입력이 유효하지 않습니다"}
try:
# 실제 ReAct 로직 실행
self.logger.info(f"처리 시작: user_id={user_id}, question={question[:100]}")
result = self._internal_solve(question)
# 출력 필터링 (민감 정보 제거)
filtered_result = self._filter_output(result)
self.logger.info(f"처리 완료: user_id={user_id}")
return {"answer": filtered_result, "success": True}
except Exception as e:
# 에러 로깅 및 안전한 에러 메시지 반환
self.logger.error(f"처리 실패: {str(e)}", exc_info=True)
return {"error": "처리 중 오류가 발생했습니다", "success": False}
def _validate_input(self, question: str) -> bool:
"""입력 검증 (SQL Injection, XSS 등 방지)"""
# 실전에서는 더 정교한 검증 필요
if len(question) > 5000:
return False
dangerous_patterns = ["<script>", "DROP TABLE", "'; --"]
return not any(pattern in question for pattern in dangerous_patterns)
def _filter_output(self, result: str) -> str:
"""출력에서 민감 정보 제거"""
# 이메일, 전화번호 등 마스킹
import re
result = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'[EMAIL]', result)
result = re.sub(r'\b\d{3}-\d{4}-\d{4}\b', '[PHONE]', result)
return result
def _internal_solve(self, question):
"""실제 ReAct 로직 (간소화)"""
return f"{question}에 대한 답변"
def _execute_tool(self, tool_name, *args):
"""도구 실행"""
return f"{tool_name} 실행 결과"
class RateLimiter:
"""간단한 속도 제한 구현"""
def __init__(self, max_requests, window):
self.max_requests = max_requests
self.window = window
self.requests = {}
def allow(self, user_id):
# 실전에서는 Redis 등 사용
return True
설명
이것이 하는 일: 이 코드는 개발 환경의 ReAct 에이전트를 프로덕션 등급으로 강화