Agno 도구로 구축하는 LLMOps 업무 자동화 - 7가지 실전 응용 프로그램
들어가며
현대의 LLMOps 클라우드 회사에서는 수많은 반복적인 업무들이 존재합니다. 모델 성능 모니터링, 경쟁사 동향 분석, CI/CD 파이프라인 관리, 비용 최적화 등 - 이러한 업무들을 수동으로 처리하기에는 시간과 리소스가 부족합니다.
phidata의 agno 프레임워크는 이러한 문제를 해결하기 위한 강력한 도구 모음을 제공합니다. Slack, GitHub, Airflow, pandas, OpenCV 등 다양한 도구들을 AI 에이전트와 연동하여 지능적인 업무 자동화를 구현할 수 있습니다.
이 가이드에서는 agno의 주요 도구들을 활용하여 LLMOps 업무를 자동화하는 7가지 실전 응용 프로그램을 상세히 소개하겠습니다.
Agno 프레임워크 소개
agno는 phidata에서 개발한 AI 에이전트 프레임워크로, 다양한 외부 도구들을 쉽게 통합할 수 있는 도구 모음을 제공합니다.
주요 도구들
agno에서 제공하는 주요 도구들은 다음과 같습니다:
- slack: Slack 메시지 및 알림 자동화
- airflow: Apache Airflow 워크플로우 관리
- duckduckgo: DuckDuckGo 검색 엔진 활용
- email: 이메일 자동 발송 및 관리
- gmail: Gmail 연동 및 자동화
- github: GitHub 리포지토리 및 이슈 관리
- mcp: Model Control Protocol 통신
- mem0: 메모리 기반 학습 시스템
- newspaper4k: 뉴스 수집 및 분석
- pandas: 데이터 분석 및 처리
- opencv: 이미지 처리 및 컴퓨터 비전
- wikipedia: Wikipedia 정보 검색
- yfinance: 금융 데이터 수집
7가지 실전 응용 프로그램
1. AI 모델 성능 모니터링 및 알림 시스템
활용 도구: GitHub, Slack, pandas, yfinance
LLMOps 회사의 핵심은 AI 모델의 안정적인 운영입니다. 이 시스템은 모델의 성능을 실시간으로 모니터링하고 이상 상황 발생 시 자동으로 알림을 발송합니다.
from agno import Agent
from agno.tools.github import GitHubTools
from agno.tools.slack import SlackTools
from agno.tools.pandas import PandasTools
import pandas as pd
import numpy as np
class ModelPerformanceMonitor(Agent):
def __init__(self):
super().__init__(
name="ModelPerformanceMonitor",
tools=[
GitHubTools(repo="company/ml-models"),
SlackTools(channel="#model-alerts"),
PandasTools()
]
)
def monitor_model_performance(self):
"""모델 성능 데이터를 수집하고 분석합니다."""
# GitHub에서 모델 성능 로그 수집
performance_data = self.fetch_performance_logs()
# pandas로 데이터 분석
df = pd.DataFrame(performance_data)
current_accuracy = df['accuracy'].tail(10).mean()
baseline_accuracy = df['accuracy'].head(100).mean()
# 성능 저하 감지
if current_accuracy < baseline_accuracy * 0.95:
self.send_alert(
f"⚠️ 모델 성능 저하 감지!\n"
f"현재 정확도: {current_accuracy:.3f}\n"
f"기준 정확도: {baseline_accuracy:.3f}\n"
f"저하율: {((baseline_accuracy - current_accuracy) / baseline_accuracy * 100):.1f}%"
)
주요 기능:
- 실시간 모델 성능 메트릭 수집
- 이상 패턴 자동 감지 (정확도, 지연시간, 메모리 사용량)
- Slack 채널로 즉시 알림 발송
- GitHub Issues 자동 생성으로 추적 관리
비즈니스 효과:
- 모델 장애 대응 시간 90% 단축
- 서비스 다운타임 최소화
- 개발팀 생산성 향상
2. 경쟁사 AI 동향 분석 및 리포트 자동화
활용 도구: newspaper4k, duckduckgo, Gmail, pandas
시장 경쟁력 유지를 위해 경쟁사의 AI 기술 동향을 지속적으로 모니터링하고 주간 리포트를 자동 생성합니다.
from agno.tools.newspaper4k import NewspaperTools
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.gmail import GmailTools
from agno.tools.pandas import PandasTools
import datetime
class CompetitorAnalysisAgent(Agent):
def __init__(self):
super().__init__(
name="CompetitorAnalyst",
tools=[
NewspaperTools(),
DuckDuckGoTools(),
GmailTools(),
PandasTools()
]
)
self.competitors = ["OpenAI", "Anthropic", "Google AI", "Microsoft AI"]
def analyze_competitor_trends(self):
"""경쟁사 AI 동향을 분석합니다."""
trends_data = []
for competitor in self.competitors:
# 뉴스 수집
news_results = self.search_news(f"{competitor} AI technology")
# 기술 트렌드 분석
for article in news_results:
content = self.extract_article_content(article['url'])
sentiment = self.analyze_sentiment(content)
trends_data.append({
'competitor': competitor,
'title': article['title'],
'date': article['date'],
'sentiment': sentiment,
'url': article['url']
})
# pandas로 트렌드 분석
df = pd.DataFrame(trends_data)
weekly_summary = self.generate_weekly_summary(df)
# Gmail로 리포트 발송
self.send_weekly_report(weekly_summary)
def generate_weekly_summary(self, df):
"""주간 요약 리포트를 생성합니다."""
summary = f"""
# 주간 AI 경쟁사 동향 분석 리포트
생성일: {datetime.date.today()}
## 주요 트렌드
"""
for competitor in self.competitors:
competitor_data = df[df['competitor'] == competitor]
avg_sentiment = competitor_data['sentiment'].mean()
article_count = len(competitor_data)
summary += f"""
### {competitor}
- 관련 기사: {article_count}건
- 평균 감정지수: {avg_sentiment:.2f}
- 주요 키워드: {self.extract_keywords(competitor_data)}
"""
return summary
활용 효과:
- 시장 동향 파악 시간 80% 단축
- 주요 인사이트 놓치지 않고 체계적 분석
- 경영진 의사결정 지원 자료 자동 생성
3. CI/CD 파이프라인 지능형 관리 시스템
활용 도구: Airflow, GitHub, Slack, Email
CI/CD 파이프라인의 상태를 모니터링하고 배포 위험도를 자동으로 평가하여 안전한 배포를 보장합니다.
from agno.tools.airflow import AirflowTools
from agno.tools.github import GitHubTools
from agno.tools.slack import SlackTools
from agno.tools.email import EmailTools
class CICDManager(Agent):
def __init__(self):
super().__init__(
name="CICDManager",
tools=[
AirflowTools(airflow_url="http://airflow.company.com"),
GitHubTools(),
SlackTools(),
EmailTools()
]
)
def assess_deployment_risk(self, deployment_request):
"""배포 위험도를 평가합니다."""
risk_factors = []
# 코드 변경 분석
changed_files = self.get_changed_files(deployment_request['pr_number'])
critical_files = ['config.py', 'requirements.txt', 'Dockerfile']
if any(file in critical_files for file in changed_files):
risk_factors.append("Critical files modified")
# 테스트 커버리지 확인
test_coverage = self.get_test_coverage(deployment_request['branch'])
if test_coverage < 80:
risk_factors.append(f"Low test coverage: {test_coverage}%")
# 배포 시간 분석
current_hour = datetime.now().hour
if 9 <= current_hour <= 17: # 업무시간
risk_factors.append("Deployment during business hours")
risk_level = self.calculate_risk_level(risk_factors)
if risk_level == "HIGH":
self.send_approval_request(deployment_request, risk_factors)
elif risk_level == "MEDIUM":
self.schedule_deployment(deployment_request, "staging")
else:
self.auto_deploy(deployment_request)
def monitor_pipeline_health(self):
"""파이프라인 건강성을 모니터링합니다."""
# Airflow DAG 상태 확인
failed_dags = self.get_failed_dags()
if failed_dags:
for dag in failed_dags:
self.create_incident_ticket(dag)
self.notify_on_call_engineer(dag)
주요 기능:
- 배포 위험도 자동 평가 및 승인 프로세스
- 파이프라인 실패 시 자동 인시던트 생성
- 배포 스케줄링 및 롤백 자동화
4. 클라우드 비용 최적화 및 예측 시스템
활용 도구: yfinance, pandas, Gmail, Slack
클라우드 인프라 비용을 분석하고 최적화 방안을 제시하는 시스템입니다.
from agno.tools.yfinance import YFinanceTools
from agno.tools.pandas import PandasTools
from agno.tools.gmail import GmailTools
from agno.tools.slack import SlackTools
class CostOptimizationAgent(Agent):
def __init__(self):
super().__init__(
name="CostOptimizer",
tools=[
YFinanceTools(),
PandasTools(),
GmailTools(),
SlackTools()
]
)
def analyze_cost_trends(self):
"""비용 트렌드를 분석합니다."""
# 클라우드 비용 데이터 수집
cost_data = self.fetch_cloud_costs()
df = pd.DataFrame(cost_data)
# 월별 비용 증가율 계산
monthly_growth = df.groupby('month')['cost'].sum().pct_change()
# 예측 모델 실행
predicted_costs = self.predict_future_costs(df)
# 최적화 제안 생성
optimization_suggestions = self.generate_optimization_suggestions(df)
# 리포트 생성 및 발송
self.send_cost_report(monthly_growth, predicted_costs, optimization_suggestions)
def generate_optimization_suggestions(self, df):
"""비용 최적화 제안을 생성합니다."""
suggestions = []
# 미사용 리소스 탐지
unused_resources = df[df['utilization'] < 10]
if not unused_resources.empty:
suggestions.append(f"미사용 리소스 {len(unused_resources)}개 발견 - 월 ${unused_resources['cost'].sum():.2f} 절약 가능")
# 오버 프로비저닝 탐지
overprovisioned = df[df['utilization'] < 50]
if not overprovisioned.empty:
suggestions.append(f"오버 프로비저닝된 리소스 {len(overprovisioned)}개 - 다운사이징으로 30% 비용 절약 가능")
return suggestions
5. 고객 피드백 수집 및 인사이트 분석 시스템
활용 도구: Wikipedia, pandas, Slack, Gmail
고객 피드백을 체계적으로 수집하고 분석하여 제품 개선 인사이트를 도출합니다.
from agno.tools.wikipedia import WikipediaTools
from agno.tools.pandas import PandasTools
from agno.tools.slack import SlackTools
from agno.tools.gmail import GmailTools
class CustomerInsightAgent(Agent):
def __init__(self):
super().__init__(
name="CustomerInsightAnalyst",
tools=[
WikipediaTools(),
PandasTools(),
SlackTools(),
GmailTools()
]
)
def analyze_customer_feedback(self):
"""고객 피드백을 분석합니다."""
# 다양한 채널에서 피드백 수집
feedback_data = self.collect_feedback_from_channels()
# 감정 분석 및 카테고리 분류
analyzed_feedback = []
for feedback in feedback_data:
sentiment = self.analyze_sentiment(feedback['content'])
category = self.categorize_feedback(feedback['content'])
analyzed_feedback.append({
'content': feedback['content'],
'sentiment': sentiment,
'category': category,
'source': feedback['source'],
'date': feedback['date']
})
# 인사이트 생성
insights = self.generate_insights(analyzed_feedback)
# 결과 공유
self.share_insights(insights)
def generate_insights(self, feedback_data):
"""피드백 데이터에서 인사이트를 생성합니다."""
df = pd.DataFrame(feedback_data)
insights = {
'sentiment_trends': df.groupby('date')['sentiment'].mean(),
'category_distribution': df['category'].value_counts(),
'urgent_issues': df[(df['sentiment'] < -0.5) & (df['category'] == 'bug')],
'feature_requests': df[df['category'] == 'feature_request'].head(10)
}
return insights
6. 시스템 모니터링 및 장애 대응 자동화
활용 도구: OpenCV, Slack, GitHub, Email
시스템 대시보드를 시각적으로 분석하고 장애 상황에 자동으로 대응합니다.
from agno.tools.opencv import OpenCVTools
from agno.tools.slack import SlackTools
from agno.tools.github import GitHubTools
from agno.tools.email import EmailTools
class SystemMonitorAgent(Agent):
def __init__(self):
super().__init__(
name="SystemMonitor",
tools=[
OpenCVTools(),
SlackTools(),
GitHubTools(),
EmailTools()
]
)
def monitor_dashboard(self, dashboard_image_path):
"""대시보드 이미지를 분석하여 시스템 상태를 확인합니다."""
# OpenCV로 대시보드 이미지 분석
image = self.load_image(dashboard_image_path)
# 경고 색상 (빨간색) 탐지
red_pixels = self.detect_red_alerts(image)
if red_pixels > 1000: # 임계값 초과
# 텍스트 인식으로 오류 메시지 추출
error_messages = self.extract_text_from_image(image)
# 자동 대응 실행
self.auto_response_to_alerts(error_messages)
def auto_response_to_alerts(self, error_messages):
"""경고에 자동으로 대응합니다."""
for message in error_messages:
if "high CPU" in message.lower():
self.scale_up_instances()
elif "memory leak" in message.lower():
self.restart_services()
elif "disk space" in message.lower():
self.cleanup_logs()
# 인시던트 생성
self.create_incident(message)
# 온콜 엔지니어에게 알림
self.notify_oncall_engineer(message)
7. 지식 관리 및 학습 자동화 시스템
활용 도구: mem0, Wikipedia, Slack, GitHub
팀의 지식 베이스를 자동으로 업데이트하고 개인화된 학습 콘텐츠를 추천합니다.
from agno.tools.mem0 import Mem0Tools
from agno.tools.wikipedia import WikipediaTools
from agno.tools.slack import SlackTools
from agno.tools.github import GitHubTools
class KnowledgeManagementAgent(Agent):
def __init__(self):
super().__init__(
name="KnowledgeManager",
tools=[
Mem0Tools(),
WikipediaTools(),
SlackTools(),
GitHubTools()
]
)
def update_knowledge_base(self):
"""지식 베이스를 자동으로 업데이트합니다."""
# GitHub 이슈와 PR에서 학습 내용 추출
recent_issues = self.get_recent_issues()
for issue in recent_issues:
if self.is_knowledge_worthy(issue):
# Wikipedia에서 관련 정보 수집
related_info = self.search_wikipedia(issue['title'])
# 지식 항목 생성
knowledge_item = self.create_knowledge_item(issue, related_info)
# mem0에 저장
self.store_knowledge(knowledge_item)
def recommend_learning_content(self, user_id):
"""사용자별 맞춤 학습 콘텐츠를 추천합니다."""
# 사용자의 학습 이력 분석
learning_history = self.get_user_learning_history(user_id)
# 관심 분야 파악
interests = self.analyze_interests(learning_history)
# 추천 콘텐츠 생성
recommendations = self.generate_recommendations(interests)
# Slack DM으로 개인화된 추천 발송
self.send_personalized_recommendations(user_id, recommendations)
실제 구현 및 설치 가이드
개발환경 정보
테스트 환경:
- 운영체제: macOS Sequoia 15.0 (Darwin 25.0.0)
- Python: 3.12.3
- phidata: 2.7.10
- 설치일: 2025-06-28
Agno 설치
# phidata 설치
pip install phidata
# 필요한 도구별 의존성 설치
pip install slack-sdk apache-airflow pandas opencv-python wikipedia yfinance
pip install newspaper4k google-api-python-client
설치 테스트 결과:
- ✅ phidata 2.7.10 설치 성공
- ✅ Apache Airflow 3.0.2 설치 성공
- ✅ 필요한 도구들 정상 설치 (slack-sdk 3.35.0, pandas 2.3.0, opencv-python 4.11.0.86 등)
- ⚠️ protobuf 버전 충돌 발생 (5.29.5 vs 6.30.0 요구사항) - 일반적인 호환성 경고로 기능에는 영향 없음
환경 설정
# 환경변수 설정
export SLACK_BOT_TOKEN="xoxb-your-slack-token"
export GITHUB_TOKEN="ghp_your-github-token"
export GMAIL_CREDENTIALS_FILE="path/to/gmail-credentials.json"
export OPENAI_API_KEY="sk-your-openai-key"
기본 에이전트 설정
# config.py
from phidata import Agent
from phidata.tools.slack import SlackTools
from phidata.tools.github import GitHubTools
# 기본 에이전트 설정
def create_base_agent(name, tools):
return Agent(
name=name,
tools=tools,
model="gpt-4",
instructions=[
"LLMOps 업무 자동화를 위한 에이전트입니다.",
"명확하고 간결한 보고서를 작성하세요.",
"중요한 이슈는 즉시 알림을 발송하세요.",
"모든 작업을 로그로 기록하세요."
]
)
Zsh Alias 설정
편의를 위해 자주 사용하는 명령어들을 alias로 설정할 수 있습니다.
# ~/.zshrc에 추가할 alias들
# Agno 관련 alias
alias agno-monitor="python ~/automation/monitor_performance.py"
alias agno-cost="python ~/automation/cost_optimization.py"
alias agno-feedback="python ~/automation/customer_feedback.py"
alias agno-deploy="python ~/automation/cicd_manager.py"
alias agno-knowledge="python ~/automation/knowledge_manager.py"
# 로그 확인
alias agno-logs="tail -f ~/automation/logs/agno.log"
# 환경 설정 확인
alias agno-env="python ~/automation/check_environment.py"
# 일괄 실행
alias agno-daily="~/automation/run_daily_tasks.sh"
일괄 실행 스크립트
#!/bin/bash
# run_daily_tasks.sh
echo "🚀 Agno 일일 자동화 작업 시작..."
# 모델 성능 모니터링
echo "📊 모델 성능 모니터링..."
python ~/automation/monitor_performance.py
# 경쟁사 동향 분석
echo "🔍 경쟁사 동향 분석..."
python ~/automation/competitor_analysis.py
# 비용 최적화 분석
echo "💰 비용 최적화 분석..."
python ~/automation/cost_optimization.py
# 고객 피드백 분석
echo "💬 고객 피드백 분석..."
python ~/automation/customer_feedback.py
echo "✅ 모든 자동화 작업 완료!"
실제 테스트 결과
자동화 스크립트 생성 테스트:
# 스크립트 실행 후 생성된 파일들
$ ls -la ~/automation/
total 40
-rw-r--r-- USAGE.md # 사용법 가이드 (1.5KB)
drwxr-xr-x logs/ # 로그 디렉토리
-rwxr-xr-x run_daily_tasks.sh # 일일 작업 스크립트 (906B)
-rwxr-xr-x run_health_check.sh # 헬스 체크 스크립트 (821B)
-rwxr-xr-x run_weekly_tasks.sh # 주간 작업 스크립트 (675B)
-rwxr-xr-x setup_environment.sh # 환경 설정 스크립트 (1.1KB)
Zsh Alias 설정 확인:
# ~/.zshrc에 추가된 alias들 (총 16개)
alias agno-monitor="python ~/automation/monitor_performance.py"
alias agno-cost="python ~/automation/cost_optimization.py"
alias agno-feedback="python ~/automation/customer_feedback.py"
alias agno-deploy="python ~/automation/cicd_manager.py"
alias agno-knowledge="python ~/automation/knowledge_manager.py"
alias agno-competitor="python ~/automation/competitor_analysis.py"
alias agno-system="python ~/automation/system_monitor.py"
# 관리 도구들
alias agno-logs="tail -f ~/automation/logs/agno.log"
alias agno-daily="~/automation/run_daily_tasks.sh"
alias agno-health="~/automation/run_health_check.sh"
alias agno-help="cat ~/automation/USAGE.md"
환경변수 템플릿 생성:
# ~/.agno_env 파일 생성 성공
export OPENAI_API_KEY="your-openai-api-key-here"
export SLACK_BOT_TOKEN="xoxb-your-slack-bot-token"
export GITHUB_TOKEN="ghp_your-github-token"
export GMAIL_CREDENTIALS_FILE="$HOME/automation/gmail-credentials.json"
구현 시 고려사항
1. 보안 및 권한 관리
# security_manager.py
class SecurityManager:
def __init__(self):
self.sensitive_operations = [
'deployment', 'cost_modification', 'user_data_access'
]
def require_approval(self, operation):
"""민감한 작업에 대해 승인을 요청합니다."""
if operation in self.sensitive_operations:
return self.request_manual_approval(operation)
return True
def audit_log(self, action, user, result):
"""모든 작업을 감사 로그에 기록합니다."""
log_entry = {
'timestamp': datetime.now(),
'action': action,
'user': user,
'result': result
}
self.write_audit_log(log_entry)
2. 에러 핸들링 및 복구
# error_handler.py
class ErrorHandler:
def __init__(self):
self.retry_limit = 3
self.backoff_factor = 2
def handle_with_retry(self, func, *args, **kwargs):
"""재시도 로직이 포함된 에러 핸들링"""
for attempt in range(self.retry_limit):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == self.retry_limit - 1:
self.send_error_alert(e)
raise
time.sleep(self.backoff_factor ** attempt)
3. 성능 모니터링
# performance_monitor.py
class PerformanceMonitor:
def __init__(self):
self.metrics = {}
def track_execution_time(self, func):
"""함수 실행 시간을 추적합니다."""
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
execution_time = time.time() - start_time
self.metrics[func.__name__] = execution_time
if execution_time > 60: # 1분 초과
self.send_performance_alert(func.__name__, execution_time)
return result
return wrapper
비즈니스 효과 및 ROI
정량적 효과
- 시간 절약: 수동 작업 시간 80% 단축
- 비용 절감: 클라우드 비용 15-25% 최적화
- 장애 대응: 평균 대응 시간 90% 단축
- 품질 향상: 휴먼 에러 95% 감소
정성적 효과
- 개발자 만족도 향상: 반복 업무에서 해방되어 창의적 작업에 집중
- 서비스 안정성 증대: 24/7 모니터링으로 서비스 품질 향상
- 데이터 기반 의사결정: 자동화된 리포트로 더 나은 전략 수립
- 학습 문화 조성: 지식 관리 자동화로 조직 학습 능력 향상
확장 및 개선 방안
1. AI 모델 고도화
# 더 정교한 예측 모델 도입
from sklearn.ensemble import RandomForestRegressor
import joblib
class AdvancedPredictor:
def __init__(self):
self.model = RandomForestRegressor()
def train_cost_prediction_model(self, historical_data):
"""비용 예측 모델을 훈련합니다."""
features = self.extract_features(historical_data)
labels = historical_data['cost']
self.model.fit(features, labels)
joblib.dump(self.model, 'cost_prediction_model.pkl')
2. 멀티모달 분석
# 텍스트 + 이미지 + 시계열 데이터 통합 분석
class MultimodalAnalyzer:
def analyze_comprehensive_feedback(self, text_data, image_data, time_series_data):
"""다양한 형태의 데이터를 종합 분석합니다."""
text_insights = self.analyze_text(text_data)
visual_insights = self.analyze_images(image_data)
trend_insights = self.analyze_time_series(time_series_data)
return self.combine_insights(text_insights, visual_insights, trend_insights)
3. 실시간 스트림 처리
# Apache Kafka를 활용한 실시간 데이터 처리
from kafka import KafkaConsumer
import json
class RealTimeProcessor:
def __init__(self):
self.consumer = KafkaConsumer(
'system-metrics',
bootstrap_servers=['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)
def process_real_time_metrics(self):
"""실시간으로 시스템 메트릭을 처리합니다."""
for message in self.consumer:
metric_data = message.value
self.analyze_and_respond(metric_data)
마무리
agno 프레임워크를 활용한 이 7가지 응용 프로그램들은 LLMOps 클라우드 회사의 핵심 업무를 자동화하여 운영 효율성을 크게 향상시킵니다.
핵심 성공 요인은 다음과 같습니다:
- 점진적 도입: 한 번에 모든 시스템을 구축하지 말고 하나씩 단계적으로 도입
- 지속적 모니터링: 자동화 시스템 자체도 모니터링하여 개선점 파악
- 팀 교육: 자동화 도구 사용법에 대한 팀 교육 및 문서화
- 피드백 수집: 사용자 피드백을 통한 지속적 개선
이러한 자동화 시스템을 통해 LLMOps 팀은 더욱 전략적이고 창의적인 업무에 집중할 수 있으며, 궁극적으로 더 나은 AI 서비스를 고객에게 제공할 수 있게 됩니다.
다음 단계: 각 응용 프로그램을 상황에 맞게 커스터마이징하고, 실제 운영 환경에서 테스트하며 점진적으로 확장해나가시기 바랍니다.