안전장치, 에러 처리, 모니터링 등 프로덕션 환경 배포를 위한 실전 가이드입니다.
이 문서는 Agent를 안전하게 프로덕션 환경에 배포하고 운영하는 방법을 다룹니다. 실전 활용 사례와 함께 성공적인 배포를 위한 필수 요소들을 학습합니다.
핵심 원칙: 프로덕션 Agent는 안전장치, 모니터링, 비용 관리가 필수입니다.
class ProductionAgent:
def __init__(self, max_iterations=10):
self.max_iterations = max_iterations
self.iteration_count = 0
async def execute_with_safeguards(self, task):
self.iteration_count = 0
while self.iteration_count < self.max_iterations:
self.iteration_count += 1
result = await self.execute_iteration(task)
if result.is_complete:
return result
# 진행이 없으면 조기 종료
if not result.made_progress:
raise NoProgressError(
f"No progress after {self.iteration_count} iterations"
)
raise MaxIterationsExceeded(
f"Task incomplete after {self.max_iterations} iterations"
)
class SandboxedAgent:
def __init__(self):
self.allowed_operations = {
"read": True,
"write": False, # 프로덕션에서는 쓰기 금지
"execute": False, # 코드 실행 금지
"network": ["api.company.com"] # 허용된 도메인만
}
async def execute_tool(self, tool_name, params):
# 권한 체크
if not self.is_allowed(tool_name, params):
raise PermissionDeniedError(
f"Tool {tool_name} not allowed in production"
)
# 샌드박스 내에서 실행
return await self.run_in_sandbox(tool_name, params)
def is_allowed(self, tool_name, params):
"""도구 사용 권한 확인"""
if tool_name.startswith("write_") and not self.allowed_operations["write"]:
return False
if tool_name == "execute_code" and not self.allowed_operations["execute"]:
return False
if tool_name.startswith("fetch_") and params.get("url"):
domain = self.extract_domain(params["url"])
return domain in self.allowed_operations["network"]
return True
class CheckpointAgent:
def __init__(self):
self.checkpoints = []
self.requires_approval = {
"destructive_operations": True,
"external_api_calls": True,
"database_modifications": True
}
async def execute_with_checkpoints(self, task):
plan = await self.create_plan(task)
for step in plan.steps:
# Checkpoint: 위험한 작업 전 승인 요청
if self.is_risky_operation(step):
approved = await self.request_human_approval(step)
if not approved:
return self.create_rejection_response(step)
# 실행
result = await self.execute_step(step)
self.checkpoints.append({
"step": step,
"result": result,
"timestamp": datetime.now()
})
return self.finalize_result()
async def request_human_approval(self, step):
"""인간 승인 요청"""
notification = f"""
🚨 Approval Required
Action: {step.action}
Impact: {step.estimated_impact}
Risk: {step.risk_level}
Details:
{step.details}
Approve? (yes/no)
"""
response = await self.send_notification(notification)
return response.lower() == "yes"
| 레벨 | 설명 | 사용 시점 |
|---|---|---|
| Level 1: Read-Only | 읽기만 가능 | 데이터 분석, 리서치 |
| Level 2: Sandboxed Write | 격리된 환경에서 쓰기 | 테스트, 개발 환경 |
| Level 3: Approved Write | 승인 후 쓰기 | PR 생성, 문서 수정 |
| Level 4: Full Access | 전체 권한 | 긴급 상황만 (극히 제한적) |
class RobustAgent:
async def execute_with_fallbacks(self, task, max_retries=3):
"""에러 발생 시 단계적 대응"""
for attempt in range(max_retries):
try:
# 시도 1: 정상 실행
result = await self.execute(task)
return result
except ToolExecutionError as e:
if attempt == max_retries - 1:
# 최종 실패: 간단한 대안으로 전환
logger.warning(f"Tool failed, using fallback: {e}")
return await self.simple_fallback(task)
# 재시도 (Exponential backoff)
await asyncio.sleep(2 ** attempt)
except ContextLengthExceeded:
# 컨텍스트 압축 후 재시도
logger.info("Context exceeded, compressing...")
await self.compress_context()
return await self.execute(task)
except RateLimitError as e:
# Rate limit: 대기 후 재시도
wait_time = e.retry_after or 60
logger.warning(f"Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
except Exception as e:
# 예상치 못한 에러: 로깅 및 안전한 종료
logger.error(f"Unexpected error: {e}", exc_info=True)
return self.create_error_response(task, e)
async def simple_fallback(self, task):
"""복잡한 방법 실패 시 간단한 대안"""
# 예: 복잡한 분석 대신 간단한 요약
return await self.simple_summary(task)
class ErrorHandler:
ERROR_STRATEGIES = {
# 재시도 가능한 에러
"retryable": [
NetworkError,
TemporaryAPIError,
RateLimitError
],
# 컨텍스트 문제
"context_issues": [
ContextLengthExceeded,
TokenLimitError
],
# 도구 실행 실패
"tool_failures": [
ToolExecutionError,
ToolNotFoundError
],
# 복구 불가능
"fatal": [
AuthenticationError,
PermissionDeniedError,
InvalidConfigError
]
}
async def handle_error(self, error, task):
"""에러 유형별 적절한 대응"""
error_type = self.classify_error(error)
if error_type == "retryable":
return await self.retry_with_backoff(task)
elif error_type == "context_issues":
return await self.handle_context_issue(task)
elif error_type == "tool_failures":
return await self.try_alternative_tool(task)
elif error_type == "fatal":
return self.create_fatal_error_response(error)
else:
# 알 수 없는 에러: 보수적으로 처리
logger.error(f"Unknown error type: {error}")
return self.create_safe_failure_response(task)
import logging
from datetime import datetime
import json
class AgentLogger:
def __init__(self, agent_id):
self.agent_id = agent_id
self.logger = logging.getLogger(f"agent.{agent_id}")
# 구조화된 로깅
self.setup_structured_logging()
def log_interaction(self, action, result, metadata):
"""모든 Agent 상호작용 로깅"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"agent_id": self.agent_id,
"action": action,
"result": {
"success": result.success,
"output_summary": result.summary[:200] # 첫 200자만
},
"metadata": {
"tokens_used": metadata.get("tokens"),
"latency_ms": metadata.get("latency"),
"model": metadata.get("model"),
"cost_usd": metadata.get("cost")
}
}
self.logger.info(json.dumps(log_entry))
def log_tool_call(self, tool_name, parameters, result, duration_ms):
"""도구 호출 로깅"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event": "tool_call",
"tool": tool_name,
"params": self.sanitize_params(parameters),
"success": result.success,
"duration_ms": duration_ms
}
self.logger.info(json.dumps(log_entry))
def log_error(self, error, context):
"""에러 상세 로깅"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event": "error",
"error_type": type(error).__name__,
"error_message": str(error),
"context": context,
"stack_trace": traceback.format_exc()
}
self.logger.error(json.dumps(log_entry))
def sanitize_params(self, params):
"""민감 정보 제거"""
sanitized = params.copy()
# 민감 키 마스킹
sensitive_keys = ["password", "api_key", "token", "secret"]
for key in sensitive_keys:
if key in sanitized:
sanitized[key] = "***REDACTED***"
return sanitized
class AgentMetrics:
def __init__(self):
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_tokens": 0,
"total_cost_usd": 0.0,
"avg_latency_ms": 0.0,
"tool_usage": {}
}
def record_request(self, success, tokens, cost, latency_ms):
"""요청 메트릭 기록"""
self.metrics["total_requests"] += 1
if success:
self.metrics["successful_requests"] += 1
else:
self.metrics["failed_requests"] += 1
self.metrics["total_tokens"] += tokens
self.metrics["total_cost_usd"] += cost
# 이동 평균으로 latency 업데이트
n = self.metrics["total_requests"]
current_avg = self.metrics["avg_latency_ms"]
self.metrics["avg_latency_ms"] = (
(current_avg * (n - 1) + latency_ms) / n
)
def record_tool_usage(self, tool_name):
"""도구 사용 통계"""
if tool_name not in self.metrics["tool_usage"]:
self.metrics["tool_usage"][tool_name] = 0
self.metrics["tool_usage"][tool_name] += 1
def get_summary(self):
"""메트릭 요약"""
success_rate = (
self.metrics["successful_requests"] /
self.metrics["total_requests"]
) if self.metrics["total_requests"] > 0 else 0
return {
**self.metrics,
"success_rate": f"{success_rate:.2%}",
"avg_cost_per_request": (
self.metrics["total_cost_usd"] /
self.metrics["total_requests"]
) if self.metrics["total_requests"] > 0 else 0
}
class CostOptimizedAgent:
def select_model(self, task_complexity, budget="medium"):
"""작업 복잡도에 따라 적절한 모델 선택"""
if budget == "low":
# 비용 최우선
return "claude-haiku-4-5-20251001"
if task_complexity == "simple":
# 간단한 작업: Haiku (빠르고 저렴)
return "claude-haiku-4-5-20251001"
elif task_complexity == "medium":
# 중간 복잡도: Sonnet (균형)
return "claude-sonnet-4-5-20250929"
elif task_complexity == "complex":
# 복잡한 작업: Opus (최고 성능)
return "claude-opus-4-6"
else:
# 기본값: Sonnet
return "claude-sonnet-4-5-20250929"
def estimate_complexity(self, task):
"""작업 복잡도 추정"""
factors = {
"requires_reasoning": 2,
"multiple_steps": 1.5,
"large_context": 1.5,
"creative_output": 2
}
score = 1.0
if self.requires_deep_reasoning(task):
score *= factors["requires_reasoning"]
if self.has_multiple_steps(task):
score *= factors["multiple_steps"]
if self.has_large_context(task):
score *= factors["large_context"]
if self.is_creative_task(task):
score *= factors["creative_output"]
# 점수에 따라 복잡도 분류
if score < 2:
return "simple"
elif score < 4:
return "medium"
else:
return "complex"
from functools import lru_cache
import hashlib
class CachedAgent:
def __init__(self, cache_ttl=3600):
self.cache = {}
self.cache_ttl = cache_ttl
def create_cache_key(self, task, context):
"""캐시 키 생성"""
data = f"{task}:{context}"
return hashlib.sha256(data.encode()).hexdigest()
async def execute_with_cache(self, task, context):
"""캐시된 결과 활용"""
cache_key = self.create_cache_key(task, context)
# 캐시 확인
if cache_key in self.cache:
cached_entry = self.cache[cache_key]
# 유효 기간 확인
if time.time() - cached_entry["timestamp"] < self.cache_ttl:
logger.info(f"Cache hit for task: {task[:50]}...")
return cached_entry["result"]
# 캐시 미스: 실행 후 캐싱
result = await self.execute(task, context)
self.cache[cache_key] = {
"result": result,
"timestamp": time.time()
}
return result
def clear_cache(self):
"""캐시 초기화"""
self.cache.clear()
# ❌ 비효율적: 불필요하게 긴 프롬프트
inefficient_prompt = """
Please analyze this code and tell me everything you can find.
Look at the structure, the performance, the security, the style,
the documentation, the tests, the architecture, the patterns,
and anything else that might be relevant. Give me a comprehensive
analysis with all possible details you can think of.
"""
# ✅ 효율적: 구체적이고 간결한 프롬프트
efficient_prompt = """
Analyze this code for:
1. Security vulnerabilities (OWASP Top 10)
2. Performance bottlenecks
3. Code style violations (PEP 8)
Provide: severity, location, fix suggestion for each issue.
"""
import pytest
class TestAgentWorkflow:
@pytest.mark.asyncio
async def test_complete_workflow(self):
"""전체 워크플로우 테스트"""
# Given
agent = CodeRefactoringAgent()
task = "Refactor authentication module for better testability"
expected_files = ["auth.py", "auth_test.py"]
# When
result = await agent.execute(task)
# Then
assert result.success
assert all(f in result.modified_files for f in expected_files)
assert result.tests_passing
assert result.code_coverage > 0.8
# 사람이 읽을 수 있는 출력 검증
assert "refactor" in result.summary.lower()
assert len(result.changes) > 0
@pytest.mark.asyncio
async def test_error_recovery(self):
"""에러 복구 테스트"""
agent = CodeRefactoringAgent()
# 의도적으로 실패하는 작업
task = "Refactor non-existent file"
result = await agent.execute(task)
# Graceful failure 확인
assert not result.success
assert result.error_message is not None
assert "not found" in result.error_message.lower()
from playwright.async_api import async_playwright
class TestAgentUIChanges:
@pytest.mark.asyncio
async def test_ui_changes_work(self):
"""Agent가 수정한 UI가 실제로 동작하는지 테스트"""
# Agent가 UI 수정
agent = FrontendAgent()
await agent.add_login_button()
# 브라우저로 검증
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page()
await page.goto("http://localhost:3000")
# 로그인 버튼 존재 확인
login_button = page.locator("#login-button")
await expect(login_button).to_be_visible()
# 클릭 동작 확인
await login_button.click()
await expect(page).to_have_url("/login")
await browser.close()
class CustomerSupportAgent:
def __init__(self):
self.tools = [
get_customer_data,
search_order_history,
process_refund,
create_support_ticket,
send_email
]
async def handle_query(self, customer_id, query):
# Routing 패턴 사용
intent = await self.classify_intent(query)
if intent == "refund":
return await self.handle_refund(customer_id, query)
elif intent == "order_status":
return await self.handle_order_status(customer_id, query)
elif intent == "complaint":
# 중요: 인간 에스컬레이션
await self.escalate_to_human(customer_id, query)
return "Escalated to human agent"
else:
return await self.handle_general_inquiry(customer_id, query)
async def handle_refund(self, customer_id, query):
# 1. 주문 이력 확인
orders = await get_order_history(customer_id)
# 2. 환불 자격 확인
eligible = await check_refund_eligibility(orders, query)
if not eligible:
return await self.explain_refund_policy(query)
# 3. 환불 처리 (checkpoint 필요)
approval = await self.request_approval(
f"Process refund for customer {customer_id}"
)
if approval:
result = await process_refund(customer_id, orders[0])
await send_email(customer_id, "refund_confirmation", result)
return result
return "Refund request requires manual review"
class SupportMetrics:
metrics = {
"resolution_rate": 0.85, # 85% 자동 해결
"avg_response_time": "2.3s",
"customer_satisfaction": 4.2, # 5점 만점
"escalation_rate": 0.15, # 15%만 인간에게 전달
"cost_per_interaction": "$0.12"
}
class CodingAgent:
async def solve_github_issue(self, issue_url):
# 1. 이슈 이해
issue = await self.fetch_issue(issue_url)
analysis = await self.analyze_issue(issue)
# 2. 솔루션 계획
plan = await self.create_implementation_plan(analysis)
# 3. 반복적 구현 (Evaluator-Optimizer 패턴)
for step in plan.steps:
await self.implement_step(step)
# 테스트 실행
test_result = await self.run_tests()
if not test_result.passed:
# 실패한 테스트 기반 수정
await self.fix_based_on_test_results(test_result)
# 4. PR 생성
pr = await self.create_pull_request(plan, issue)
return pr
async def fix_based_on_test_results(self, test_result):
"""테스트 피드백 기반 수정"""
for failure in test_result.failures:
fix_prompt = f"""
Test failed: {failure.test_name}
Error: {failure.error_message}
Stack trace: {failure.stack_trace}
Analyze the failure and fix the code.
"""
fix = await self.generate_fix(fix_prompt)
await self.apply_fix(fix)
async def security_review_with_voting(code_changes):
# 여러 독립 검토자 (각각 다른 관점)
reviewers = [
SecurityReviewer(focus="OWASP_Top_10"),
SecurityReviewer(focus="injection_attacks"),
SecurityReviewer(focus="auth_vulnerabilities"),
SecurityReviewer(focus="data_exposure"),
SecurityReviewer(focus="crypto_issues")
]
# 병렬 검토
reviews = await asyncio.gather(*[
reviewer.review(code_changes) for reviewer in reviewers
])
# 발견된 취약점 집계
vulnerabilities = aggregate_vulnerabilities(reviews)
# 합의 기반 심각도
for vuln in vulnerabilities:
severities = [
r.get_severity(vuln) for r in reviews
if r.found_vulnerability(vuln)
]
# 2명 이상이 HIGH로 평가하면 HIGH
if severities.count("HIGH") >= 2:
vuln.severity = "HIGH"
else:
vuln.severity = statistics.mode(severities)
return SecurityReport(vulnerabilities)
class PRReviewAgent:
"""GitHub Actions에서 실행되는 안전한 PR 리뷰 Agent"""
def __init__(self):
self.read_only = True # 안전 최우선
async def review_pr(self, pr_number):
# 1. PR 브랜치 체크아웃 (로컬)
await self.checkout_pr_branch(pr_number)
# 2. 변경된 파일 읽기 (파일시스템에서)
changed_files = await self.get_changed_files()
# 3. 프로젝트 규칙 읽기
rules = await self.read_project_rules([
".cursor/rules/",
"ai-prompts-rules/",
"CONTRIBUTING.md"
])
# 4. 리뷰 수행
review = await self.conduct_review(
changed_files,
rules,
template="AI_REVIEW_FORMAT.md"
)
# 5. Impact Score 계산
review.impact_score = self.calculate_impact(review)
# 6. PR 코멘트 작성 (유일한 쓰기 작업)
await self.post_review_comment(pr_number, review)
return review
장점:
언제 사용:
주의사항:
| Framework | 특징 | 적합한 경우 | 단점 |
|---|---|---|---|
| Claude Agent SDK | Claude Code와 동일 엔진, MCP 기반 | 깊은 도구 통합 필요 | 높은 결합도 |
| LangGraph | State Graph 아키텍처, 높은 신뢰성 | B2B 신뢰성 중요 | 복잡한 학습 곡선 |
| 직접 API | 최소 코드, 투명함 | 단순 작업, 학습 | 기능 직접 구현 필요 |
| Rivet/Vellum | 시각적 워크플로우 | 빠른 프로토타이핑 | 프로덕션 제한적 |
def choose_framework(requirements):
"""요구사항에 맞는 프레임워크 선택"""
if requirements.complexity == "simple":
return "Direct API calls"
elif requirements.reliability_critical and requirements.budget == "high":
return "LangGraph"
elif requirements.rapid_prototyping:
return "Rivet or Vellum"
elif requirements.deep_tool_integration:
return "Claude Agent SDK"
else:
return "Start with API, add framework if needed"
## 안전성
- [ ] 최대 반복 횟수 제한 설정
- [ ] 샌드박스 환경 구현
- [ ] 위험한 작업에 인간 승인 필요
- [ ] 롤백 메커니즘 구현
## 에러 처리
- [ ] 모든 예외 타입에 대한 처리
- [ ] Graceful degradation 구현
- [ ] Retry 로직 with exponential backoff
- [ ] Fallback 메커니즘
## 모니터링
- [ ] 구조화된 로깅 설정
- [ ] 메트릭 수집 (성공률, latency, 비용)
- [ ] 알림 시스템 (에러, 임계값 초과)
- [ ] 대시보드 구축
## 비용
- [ ] Model routing 구현
- [ ] 캐싱 전략 적용
- [ ] 프롬프트 최적화
- [ ] 예산 알림 설정
## 테스팅
- [ ] End-to-end 테스트 작성
- [ ] 에러 시나리오 테스트
- [ ] 성능 테스트
- [ ] 보안 테스트
## 문서화
- [ ] API 문서
- [ ] 운영 가이드
- [ ] 트러블슈팅 가이드
- [ ] 사고 대응 플레이북
축하합니다! Claude Agent 완전 가이드를 완료하셨습니다.
문서 버전: 1.0
최종 업데이트: 2026-02-13
기반: Anthropic 공식 가이드 및 2026년 베스트 프랙티스