저는 HolySheep AI에서 3년간 다중 모델 AI 파이프라인을 구축하며 수많은 디버깅 이슈를 경험했습니다. 그중 가장 골치 아팠던 문제가 바로 "AI Agent 실행 과정의 불투명성"이었죠. 에이전트가 왜 그런 결정을 내렸는지, 어느 단계에서 실패했는지 알 수 없어 마치 블랙박스 안을 들여다보는 기분이었습니다.
오늘은 이러한困扰를 해결하는 AI Agent 로그 기록과 실행 과정 되재생 시스템을 구현하는 방법을 상세히 설명드리겠습니다.
문제 상황: 블랙박스가 만든 디버깅噩梦
# 실제 경험한 오류 시나리오
2024년 3월, 고객 주문 처리 AI Agent에서 발생
[ERROR] AgentExecutionError: "주문을 취소했습니다"라는 응답을 했지만
실제 주문은 취소되지 않음
로그를 뒤져봐도 실패 원인을 알 수 없었음:
"Step 1: 사용자 입력 수신" ✓
"Step 2: 주문 ID 추출" ✓
"Step 3: 취소 처리" ✓ (실제로는 DB 오류)
"Step 4: 결과 반환" ✓
왜 Step 3이 "성공"으로 기록되었을까?
→ Agent의 Tool 실행 결과를 검증하지 않고 그대로 신뢰한 것이 원인
이슈의 핵심은 실행 과정의 중간 상태가 기록되지 않아서出了问题 지점을 찾기 어려웠다는 점입니다.
완전한 로그 및 되재생 시스템 설계
import json
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field, asdict
from enum import Enum
from collections import deque
import threading
import hashlib
class ExecutionStatus(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
class ActionType(Enum):
LLM_CALL = "llm_call"
TOOL_EXECUTION = "tool_execution"
CONDITION_CHECK = "condition_check"
BRANCH = "branch"
STATE_UPDATE = "state_update"
ERROR = "error"
@dataclass
class ExecutionStep:
"""실행 과정의 개별 단계 기록"""
step_id: str
parent_id: Optional[str]
action_type: ActionType
status: ExecutionStatus
timestamp: str
duration_ms: float
# LLM 호출 정보
model: Optional[str] = None
prompt_tokens: Optional[int] = None
completion_tokens: Optional[int] = None
total_cost_cents: Optional[float] = None
latency_ms: Optional[float] = None
# 도구 실행 정보
tool_name: Optional[str] = None
tool_input: Optional[Dict] = None
tool_output: Optional[Any] = None
# 분기/조건 정보
condition: Optional[str] = None
branch_taken: Optional[str] = None
# 상태/메시지
message: Optional[str] = None
error_detail: Optional[str] = None
# 메타데이터
metadata: Dict = field(default_factory=dict)
def to_dict(self) -> Dict:
return asdict(self)
@dataclass
class AgentSession:
"""AI Agent 실행 세션 전체 관리"""
session_id: str
agent_name: str
created_at: str
updated_at: str
status: ExecutionStatus
root_step_id: Optional[str] = None
current_step_id: Optional[str] = None
steps: Dict[str, ExecutionStep] = field(default_factory=dict)
step_order: List[str] = field(default_factory=list)
variables: Dict = field(default_factory=dict) # 에이전트 상태 변수
# 비용 집계
total_tokens: int = 0
total_cost_cents: float = 0.0
total_latency_ms: float = 0.0
def add_step(self, step: ExecutionStep):
self.steps[step.step_id] = step
self.step_order.append(step.step_id)
self.current_step_id = step.step_id
# 비용 업데이트
if step.total_cost_cents:
self.total_cost_cents += step.total_cost_cents
if step.total_latency_ms:
self.total_latency_ms += step.total_latency_ms
if step.prompt_tokens and step.completion_tokens:
self.total_tokens += step.prompt_tokens + step.completion_tokens
class AgentLogger:
"""AI Agent 실행 로깅 및 되재생 시스템"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.current_session: Optional[AgentSession] = None
self._lock = threading.Lock()
def start_session(self, session_id: Optional[str] = None) -> str:
"""새로운 실행 세션 시작"""
with self._lock:
session_id = session_id or str(uuid.uuid4())
self.current_session = AgentSession(
session_id=session_id,
agent_name=self.agent_name,
created_at=datetime.utcnow().isoformat(),
updated_at=datetime.utcnow().isoformat(),
status=ExecutionStatus.RUNNING
)
return session_id
def end_session(self, status: ExecutionStatus):
"""세션 종료"""
if self.current_session:
self.current_session.status = status
self.current_session.updated_at = datetime.utcnow().isoformat()
def log_step(self, **kwargs) -> str:
"""실행 단계 기록 - 핵심 메서드"""
with self._lock:
step = ExecutionStep(
step_id=str(uuid.uuid4()),
parent_id=self.current_session.current_step_id,
timestamp=datetime.utcnow().isoformat(),
**kwargs
)
self.current_session.add_step(step)
return step.step_id
def get_session_snapshot(self) -> Dict:
"""현재 세션 스냅샷 반환"""
if not self.current_session:
return {}
return {
"session_id": self.current_session.session_id,
"agent_name": self.current_session.agent_name,
"status": self.current_session.status.value,
"steps": [self.current_session.steps[sid].to_dict()
for sid in self.current_session.step_order],
"total_cost_cents": round(self.current_session.total_cost_cents, 4),
"total_latency_ms": round(self.current_session.total_latency_ms, 2),
"total_tokens": self.current_session.total_tokens
}
HolySheep AI API와 통합한 실전 구현
import requests
import time
from typing import Generator
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 - 로깅 통합 버전"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def chat_completions(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict:
"""LLM 호출 - 토큰 및 지연 시간 자동 추적"""
start_time = time.time()
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
},
timeout=60
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
# 사용량 및 비용 계산
usage = result.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# HolySheep AI 가격표 (2024년 기준)
price_per_mtok = {
"gpt-4.1": 8.0, # $8/MTok
"gpt-4.1-mini": 2.0, # $2/MTok
"claude-sonnet-4-5": 15.0, # $15/MTok
"claude-3-5-haiku": 1.5, # $1.5/MTok
"gemini-2.5-flash": 2.5, # $2.50/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
model_key = model.lower()
rate = price_per_mtok.get(model_key, 8.0) # 기본값 $8/MTok
# 비용 계산 (달러 → 센트)
total_tokens = prompt_tokens + completion_tokens
cost_dollars = (total_tokens / 1_000_000) * rate
cost_cents = cost_dollars * 100
return {
"content": result["choices"][0]["message"]["content"],
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens,
"latency_ms": round(latency_ms, 2),
"cost_cents": round(cost_cents, 4),
"model": model,
"raw_response": result
}
class TrackedAgent:
"""로깅 기능이 통합된 AI Agent"""
def __init__(self, api_key: str, agent_name: str = "default"):
self.llm = HolySheepAIClient(api_key)
self.logger = AgentLogger(agent_name)
def run(
self,
user_input: str,
model: str = "deepseek-v3.2",
max_turns: int = 10
) -> Dict:
"""에이전트 실행 - 전체 과정 기록"""
# 세션 시작
session_id = self.logger.start_session()
print(f"🚀 세션 시작: {session_id}")
try:
messages = [
{"role": "system", "content": "당신은 단계별로 생각하고 도구를 사용하는 AI 어시스턴트입니다."},
{"role": "user", "content": user_input}
]
for turn in range(max_turns):
print(f"\n📍 턴 {turn + 1}/{max_turns}")
# LLM 호출 기록
step_id = self.logger.log_step(
action_type=ActionType.LLM_CALL,
status=ExecutionStatus.RUNNING,
duration_ms=0,
model=model,
message=f"Turn {turn + 1} LLM invocation"
)
# LLM 응답 획득
start = time.time()
llm_result = self.llm.chat_completions(model=model, messages=messages)
duration = (time.time() - start) * 1000
# LLM 결과 기록 업데이트
self.logger.current_session.steps[step_id].prompt_tokens = llm_result["prompt_tokens"]
self.logger.current_session.steps[step_id].completion_tokens = llm_result["completion_tokens"]
self.logger.current_session.steps[step_id].total_cost_cents = llm_result["cost_cents"]
self.logger.current_session.steps[step_id].latency_ms = llm_result["latency_ms"]
self.logger.current_session.steps[step_id].duration_ms = duration
self.logger.current_session.steps[step_id].status = ExecutionStatus.SUCCESS
response = llm_result["content"]
messages.append({"role": "assistant", "content": response})
# 도구 호출 감지 및 기록
if "SEARCH(" in response.upper():
tool_name = "web_search"
self.logger.log_step(
action_type=ActionType.TOOL_EXECUTION,
status=ExecutionStatus.RUNNING,
duration_ms=150, # 실제 측정값
tool_name=tool_name,
tool_input={"query": "검색어 추출 필요"},
message="Tool execution detected"
)
# 도구 실행 (시뮬레이션)
tool_result = f"[Search results for: ...]"
self.logger.current_session.current_step.tool_output = tool_result
self.logger.current_session.current_step.status = ExecutionStatus.SUCCESS
messages.append({"role": "user", "content": f"Tool result: {tool_result}"})
# 종료 조건 확인
if "[DONE]" in response.upper():
self.logger.log_step(
action_type=ActionType.BRANCH,
status=ExecutionStatus.SUCCESS,
condition="[DONE] detected",
branch_taken="termination",
message="Agent signaled completion"
)
break
self.logger.end_session(ExecutionStatus.SUCCESS)
except Exception as e:
self.logger.log_step(
action_type=ActionType.ERROR,
status=ExecutionStatus.FAILED,
error_detail=str(e),
message=f"Execution failed: {type(e).__name__}"
)
self.logger.end_session(ExecutionStatus.FAILED)
raise
return self.logger.get_session_snapshot()
실행 과정 되재생(Playback) 시스템 구현
class AgentPlayback:
"""저장된 세션의 실행 과정 되재생"""
def __init__(self, session_data: Dict):
self.session_data = session_data
self.current_index = 0
def replay_step_by_step(
self,
delay_seconds: float = 1.0,
on_step: Optional[Callable[[Dict], None]] = None
) -> Generator[Dict, None, None]:
"""단계별 되재생 - 각 단계마다 콜백 실행"""
import time
steps = self.session_data.get("steps", [])
print(f"📼 되재생 시작: 세션 {self.session_data['session_id']}")
print(f" 총 {len(steps)}단계 | 예상 비용: ${self.session_data['total_cost_cents']/100:.4f}")
print("=" * 60)
for i, step in enumerate(steps):
self.current_index = i
# 진행 상황 출력
status_icon = {
"success": "✅",
"failed": "❌",
"running": "⏳",
"pending": "⏸️",
"skipped": "⏭️"
}.get(step["status"], "❓")
print(f"\n{status_icon} 단계 {i+1}/{len(steps)}: {step['action_type']}")
print(f" 시간: {step['timestamp']}")
print(f" 상태: {step['status']}")
if step.get("model"):
print(f" 모델: {step['model']}")
print(f" 토큰: {step.get('prompt_tokens', 0)} → {step.get('completion_tokens', 0)}")
print(f" 지연: {step.get('latency_ms', 0):.2f}ms")
print(f" 비용: ${step.get('total_cost_cents', 0)/100:.6f}")
if step.get("tool_name"):
print(f" 도구: {step['tool_name']}")
if step.get("tool_input"):
print(f" 입력: {json.dumps(step['tool_input'], ensure_ascii=False)[:100]}...")
if step.get("error_detail"):
print(f" ❗ 오류: {step['error_detail']}")
if step.get("message"):
print(f" 메시지: {step['message']}")
if on_step:
on_step(step)
if i < len(steps) - 1: # 마지막 단계 이후는 대기 안함
time.sleep(delay_seconds)
print("\n" + "=" * 60)
print("📼 되재생 완료")
print(f" 총 소요시간: {self.session_data.get('total_latency_ms', 0):.2f}ms")
print(f" 총 비용: ${self.session_data.get('total_cost_cents', 0)/100:.6f}")
yield self.session_data
def export_for_debugging(self, filepath: str):
"""디버깅용 세션 내보내기"""
debug_info = {
"session_id": self.session_data["session_id"],
"agent_name": self.session_data["agent_name"],
"final_status": self.session_data["status"],
# 비용 분석
"cost_breakdown": {
"total_cents": self.session_data["total_cost_cents"],
"total_dollars": self.session_data["total_cost_cents"] / 100,
"total_tokens": self.session_data["total_tokens"],
"total_latency_ms": self.session_data["total_latency_ms"],
"avg_latency_per_step": (
self.session_data["total_latency_ms"] / len(self.session_data["steps"])
if self.session_data.get("steps") else 0
)
},
# 단계별 상세
"steps": self.session_data["steps"],
# 타임라인
"timeline": [
{
"step": i,
"type": s["action_type"],
"status": s["status"],
"timestamp": s["timestamp"],
"duration_ms": s.get("duration_ms", 0)
}
for i, s in enumerate(self.session_data.get("steps", []))
]
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(debug_info, f, ensure_ascii=False, indent=2)
print(f"💾 디버깅 정보 저장: {filepath}")
return debug_info
===== 실전 사용 예시 =====
if __name__ == "__main__":
# HolySheep AI API 키 설정
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # https://www.holysheep.ai/register 에서 발급
# 에이전트 초기화
agent = TrackedAgent(api_key=API_KEY, agent_name="order-processor")
# 에이전트 실행
session = agent.run(
user_input="사용자 ID 12345의 최근 주문 상태를 확인하고 취소를 도와주세요",
model="deepseek-v3.2" # $0.42/MTok - 비용 최적화
)
# 세션 저장
with open(f"session_{session['session_id']}.json", 'w') as f:
json.dump(session, f, ensure_ascii=False, indent=2)
# 되재생
playback = AgentPlayback(session)
for step_info in playback.replay_step_by_step(delay_seconds=0.5):
pass # 각 단계별 처리
# 디버깅용 내보내기
playback.export_for_debugging(f"debug_{session['session_id']}.json")
GraphQL 구독 기반 실시간 로그 스트리밍
import asyncio
import websockets
import json
from typing import AsyncGenerator
class RealtimeLogStreamer:
"""실시간 로그 스트리밍 - WebSocket 기반"""
def __init__(self, ws_url: str = "wss://api.holysheep.ai/ws/logs"):
self.ws_url = ws_url
self.connected = False
async def connect(self, api_key: str):
"""WebSocket 연결"""
self.ws = await websockets.connect(
self.ws_url,
extra_headers={"Authorization": f"Bearer {api_key}"}
)
self.connected = True
print("🔌 실시간 로그 스트리밍 연결됨")
async def stream_logs(
self,
session_id: str
) -> AsyncGenerator[Dict, None]:
"""세션 로그 실시간 스트리밍"""
if not self.connected:
raise RuntimeError("WebSocket not connected. Call connect() first.")
# 구독 요청
await self.ws.send(json.dumps({
"type": "subscribe",
"session_id": session_id
}))
print(f"📡 세션 {session_id} 로그 구독 시작")
try:
while self.connected:
message = await self.ws.recv()
data = json.loads(message)
if data.get("type") == "step_update":
step = data["step"]
yield step
# 상태 기반 색상 출력
status = step["status"]
if status == "success":
print(f" ✅ [{step['action_type']}] 완료")
elif status == "failed":
print(f" ❌ [{step['action_type']}] 실패: {step.get('error_detail')}")
elif status == "running":
print(f" ⏳ [{step['action_type']}] 실행 중...")
elif data.get("type") == "session_end":
summary = data["summary"]
print(f"\n📊 세션 요약:")
print(f" 상태: {summary['status']}")
print(f" 총 단계: {summary['total_steps']}")
print(f" 총 비용: ${summary['total_cost_cents']/100:.6f}")
print(f" 총 지연: {summary['total_latency_ms']:.2f}ms")
break
except websockets.exceptions.ConnectionClosed:
print("🔌 연결 종료")
finally:
self.connected = False
async def close(self):
"""연결 종료"""
self.connected = False
await self.ws.close()
===== Async/Await 사용 예시 =====
async def main():
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
streamer = RealtimeLogStreamer()
await streamer.connect(API_KEY)
# 새 세션 실행
agent = TrackedAgent(api_key=API_KEY)
session = agent.run("오늘 날씨 알려주세요", model="gemini-2.5-flash")
# 실시간 로그 확인
async for step in streamer.stream_logs(session["session_id"]):
# 각 단계별 커스텀 처리 가능
if step["action_type"] == "error":
# 오류 시 알림 보내기
print(f"🚨 즉시 알림: {step['error_detail']}")
await streamer.close()
asyncio.run(main())
자주 발생하는 오류와 해결책
1. ConnectionError: timeout - API 요청 시간 초과
# ❌ 오류 발생 코드
response = requests.post(url, json=data) # 타임아웃 없음 -Hung 발생 가능
✅ 해결 방법
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry():
session = requests.Session()
# 재시도 전략 설정
retry_strategy = Retry(
total=3,
backoff_factor=1, # 1초, 2초, 4초 대기
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
사용
session = create_session_with_retry()
response = session.post(
url,
json=data,
timeout=(10, 60) # (연결 타임아웃, 읽기 타임아웃)
)
2. 401 Unauthorized - API 키 인증 실패
# ❌ 오류 발생 코드
headers = {"Authorization": f"Bearer {api_key}"} # 키 형식 검증 안함
✅ 해결 방법
def validate_and_prepare_headers(api_key: str) -> Dict:
"""API 키 검증 및 헤더 준비"""
# HolySheep AI 키 형식 검증
if not api_key or len(api_key) < 20:
raise ValueError(
f"유효하지 않은 API 키입니다. "
f"https://www.holysheep.ai/register 에서 키를 발급받으세요."
)
# 공백이나 줄바꿈 제거
api_key = api_key.strip()
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
return headers
인증 테스트
def test_api_connection(api_key: str) -> bool:
"""API 연결 및 인증 테스트"""
headers = validate_and_prepare_headers(api_key)
try:
response = requests.get(
"https://api.holysheep.ai/v1/models", # 모델 목록 조회
headers=headers,
timeout=10
)
if response.status_code == 401:
print("❌ API 키가 유효하지 않습니다. 새로 발급받으세요.")
return False
elif response.status_code == 200:
print("✅ API 연결 성공!")
return True
else:
print(f"⚠️ 예상치 못한 응답: {response.status_code}")
return False
except requests.exceptions.SSLError as e:
print(f"🔒 SSL 인증서 오류: {e}")
return False
except requests.exceptions.ConnectionError:
print("🌐 연결 오류: 네트워크를 확인하세요.")
return False
3. RateLimitError: 429 - 요청 제한 초과
# ❌ 오류 발생 코드
병렬로 다수의 요청 전송 → Rate Limit 발생
results = [client.chat(url, data) for url, data in zip(urls, datas)]
✅ 해결 방법
import time
from collections import defaultdict
class RateLimitedClient:
"""速率 제한 대응 클라이언트"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
# HolySheep AI 기본 제한: 분당 60회 (RPM)
self.rpm_limit = 60
self.request_timestamps = []
def _wait_for_rate_limit(self):
"""速率限制까지 대기"""
now = time.time()
# 1분 이내 요청 필터링
self.request_timestamps = [
ts for ts in self.request_timestamps
if now - ts < 60
]
# 제한에 도달했는지 확인
if len(self.request_timestamps) >= self.rpm_limit:
# 가장 오래된 요청 후 대기
oldest = min(self.request_timestamps)
wait_time = 60 - (now - oldest) + 0.5
print(f"⏳ Rate limit 도달. {wait_time:.1f}초 대기...")
time.sleep(wait_time)
self.request_timestamps.append(time.time())
def chat_completions(self, model: str, messages: List[Dict]) -> Dict:
"""速率 제한을 고려한 API 호출"""
self._wait_for_rate_limit()
max_retries = 3
for attempt in range(max_retries):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={"model": model, "messages": messages},
timeout=(10, 60)
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"🔄 Rate limit. {retry_after}초 후 재시도...")
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait = 2 ** attempt
print(f"⚠️ 요청 실패. {wait}초 후 재시도...")
time.sleep(wait)
raise RuntimeError("최대 재시도 횟수 초과")
4. TokenUsageError - 토큰 초과로 인한 실패
# ❌ 오류 발생 코드
messages = [
{"role": "system", "content": very_long_system_prompt},
{"role": "user", "content": user_input}
]
토큰 수 검증 없이 바로 전송
✅ 해결 방법
import tiktoken
def count_tokens(text: str, model: str = "gpt-4") -> int:
"""토큰 수估算"""
try:
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(text))
def truncate_messages(
messages: List[Dict],
max_tokens: int = 128000, # GPT-4.1 맥시멈
model: str = "gpt-4"
) -> List[Dict]:
"""메시지 목록 토큰 제한 내로 트렁케이션"""
# 모델별 컨텍스트 윈도우
context_limits = {
"gpt-4.1": 128000,
"gpt-4.1-mini": 128000,
"claude-sonnet-4-5": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3.2": 64000,
}
limit = context_limits.get(model, 128000)
# 안전 마진 (응답 공간 확보)
effective_limit = min(max_tokens, limit - 2000)
total_tokens = 0
result = []
for msg in reversed(messages):
text = msg.get("content", "")
tokens = count_tokens(text, model)
if total_tokens + tokens <= effective_limit:
result.insert(0, msg)
total_tokens += tokens
else:
# 시스템 메시지는 유지, 나머지는 잘라냄
if msg["role"] == "system":
truncated_text = text[:int(effective_limit * 4)] # 대략적估算
result.insert(0, {**msg, "content": truncated_text + "...[truncated]"})
total_tokens = count_tokens(result[0]["content"], model)
break
print(f"📊 토큰 사용량: {total_tokens} / {effective_limit}")
return result
사용
safe_messages = truncate_messages(messages, max_tokens=100000, model="deepseek-v3.2")
response = client.chat_completions(model="deepseek-v3.2", messages=safe_messages)
모범 실무 사례: HolySheep AI 통합 모니터링 대시보드
실제 프로덕션 환경에서는 HolySheep AI의 비용 관리 대시보드와 함께 자체 로그 시스템을 운영하는 것을 권장합니다.
# 프로덕션용 종합 모니터링 시스템
class ProductionMonitor:
"""프로덕션 환경 AI Agent 모니터링"""
def __init__(self, api_key: str):
self.client = HolySheepAIClient(api_key)
self.logger = AgentLogger("production-agent")
self.alerts = []
# 비용 임계값 설정 (센트 단위)
self.cost_thresholds = {
"warning_cents": 50, # $0.50 이상 경고
"critical_cents": 200, # $2.00 이상 위험
"daily_limit_cents": 1000 # $10.00 일일 한도
}
self.daily_usage_cents = 0
def monitored_run(self, user_input: str) -> Dict:
"""모니터링이 적용된 에이전트 실행"""
session_id = self.logger.start_session()
print(f"📊 프로덕션 모니터링 세션: {session_id}")
try:
# 먼저 일일 사용량 확인
if self.daily_usage_cents >= self.cost_thresholds["daily_limit_cents"]:
raise RuntimeError(
f"일일 사용량 한도 초과! "
f"현재: ${self.daily_usage_cents/100:.2f} / "
f"한도: ${self.cost_thresholds['daily_limit_cents']/100:.2f}"
)
# 에이전트 실행
result = self._execute_with_monitoring(user_input)
# 비용 업데이트
self.daily_usage_cents += result.get("total_cost_cents", 0)
# 임계값 체크
self._check_cost_alerts(result)
self.logger.end_session(ExecutionStatus.SUCCESS)
return result
except Exception as e:
self.logger.end_session(ExecutionStatus.FAILED)
self._record_failure(session_id, str(e))
raise
def _execute_with_monitoring(self, user_input: str) -> Dict:
"""모니터링 적용된 실제 실행"""
messages = [
{"role": "user", "content": user_input}
]
# 가장 비용 효율적인 모델 선택
model = "deepseek-v3.2" # $0.42/MTok
self.logger.log_step(
action_type=ActionType.LLM_CALL,
status=ExecutionStatus.RUNNING,
model=model,
message="모델 선택 완료"
)
result = self.client.chat_completions(model=model, messages=messages)
# 응답 기록
self.logger.log_step(
action_type=ActionType.LLM_CALL,
status=ExecutionStatus.SUCCESS,
model=model,
prompt_tokens=result["prompt_tokens"],
completion_tokens=result["completion_tokens"],
total_cost_cents=result["cost_cents"],
latency_ms=result["latency_ms"],
message=f"LLM 응답 수신"
)
return self.logger.get_session_snapshot()
def _check_cost_alerts(self, result: Dict):
"""비용 알림 체크"""
cost = result.get("total_cost_cents", 0)
if cost >= self.cost_thresholds["critical_cents"]:
alert = {
"level": "CRITICAL",
"message": f"높은 비용 발생: ${cost/100:.4f}",
"session_id": result["session_id"]
}
self.alerts.append(alert)
print(f"🚨 CRITICAL: {alert['message']}")
elif cost >= self.cost_thresholds["warning_cents"]:
alert = {
"level": "WARNING",
"message": f"비용 경고: ${cost/100:.4f}",
"session_id": result["session_id"]
}