도메인에 관계없이 멀티 에이전트 아키텍처에서 재사용할 수 있는 공통 Tool 및 MCP(Model Context Protocol) 구현체와, 이를 조합하여 실제 사용 사례를 보여주는 예시 시나리오로 구성된 프레임워크입니다.
Agentic AI의 핵심 구성 요소 — LLM + Tools + Memory + Evaluation + Loop.
이 프로젝트는 위 개념을 LangGraph + MCP 기반으로 실제 구현한 것입니다.
| Layer | 역할 | 구성 요소 |
|---|---|---|
| Multi-Agent Graph | LLM 에이전트들이 협력하여 작업 수행 | Planner → Executor → Reviewer |
| Tool Layer | LLM이 호출 가능한 LangChain @tool 래퍼 |
34개 도메인-독립 tool (33 in ALL_TOOLS + flight_search) |
| MCP Layer | 실제 백엔드 시스템과 연결되는 구현체 | 8개 MCP 클래스 (플러그인 백엔드) |
START
│
▼
┌─────────────┐ ┌──────────────┐ ┌──────────────┐
│ Planner │──────▶│ Executor │──────▶│ Reviewer │
│ (계획 수립) │ │ (도구 실행) │◀──────│ (품질 검증) │
└─────────────┘ │ ↕ │ rev. └──────┬───────┘
│ ToolNode │ │ approved
└──────────────┘ ▼
END
- Planner: 사용자 요청을 분석하여 번호 매긴 실행 계획 생성 — Memory · Retrieval · Crawl · HTTP · Scheduler · Notification · Auth · Logging · Flight 9개 tool 카테고리를 인지하고 각 단계에서 사용할 tool을 명시
- Executor: 계획 순서대로 tool을 호출하여 작업 실행 (max 10 iterations) — ALL_TOOLS 33개(Memory · Retrieval · Crawl · HTTP · Scheduler · Notification · Auth · Logging)를 LLM에 바인딩, tool 스키마를 자동 인지
- Reviewer: 실행 결과를 검토하여 APPROVED / REVISION NEEDED / FAILED 판정
- ToolNode: Executor의
tool_calls를 자동으로 실행하고 결과를 메시지로 반환
MemoryMCP — SQLite 기반 영속 KV 스토어 (네임스페이스 + TTL 지원)
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
memory_get |
저장된 값 조회 | key, namespace="default" |
memory_set |
값 저장 (TTL 선택) | key, value, namespace, ttl=0 |
memory_delete |
키 삭제 | key, namespace |
memory_list_keys |
네임스페이스 내 전체 키 목록 | namespace="default" |
RetrievalMCP — 플러그인 가능한 문서 검색 엔진
RETRIEVAL_BACKEND 환경 변수로 백엔드를 선택합니다.
| 백엔드 | 값 | 특징 | 추가 설치 |
|---|---|---|---|
| BM25 + SQLite FTS5 | bm25_sqlite |
BM25 랭킹, 추가 의존성 없음, 수백만 doc 이상 처리 가능 | — |
| Vector (ChromaDB) | vector (기본값) |
임베딩 시맨틱 검색, 동의어/패러프레이즈 처리 | chromadb |
| PostgreSQL | postgres |
tsvector 전문 검색, 대규모 코퍼스 |
psycopg2-binary |
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
retrieval_index |
문서를 인덱스에 추가/업데이트 (청킹 선택) | doc_id, content, metadata="{}", chunk_size=0, chunk_overlap=50 |
retrieval_delete_chunks |
소스 문서의 청크 전체 삭제 | source_doc_id |
retrieval_delete |
특정 문서(또는 청크) 삭제 | doc_id |
retrieval_search |
자연어 쿼리로 문서 검색 (메타데이터 필터 지원) | query, top_k=5, filter="{}" |
retrieval_build_context |
검색 결과를 LLM 프롬프트용 컨텍스트 문자열로 조립 | query, top_k=5, max_chars=3000, filter="{}" |
RAG 흐름:
1. retrieval_index(chunk_size=500) ← 문서를 청크 단위로 분할·인덱싱
2. retrieval_build_context(query) ← 관련 청크를 검색·조립하여 LLM 컨텍스트 반환
↳ "[Source: doc-id | Score: 0.85]\n청크 내용...\n\n---\n\n[Source: ...]"
filter 파라미터로 메타데이터 필터링:
filter='{"category": "billing"}' # 단일 필터
filter='{"_source_id": "faq-001"}' # 특정 소스 문서의 청크만 검색
filter='{"category": "docs", "lang": "ko"}' # 다중 필터 (AND 조건)
crawl_tools — 웹 페이지를 fetch·정제·청킹하여 Retrieval 인덱스에 자동 적재하는 RAG 수집 파이프라인
HTML 파싱 전략: BeautifulSoup 설치 시 고품질 텍스트 추출, 미설치 시 정규식 fallback
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
crawl_and_index |
단일 URL을 fetch하여 청킹·인덱싱 | url, doc_id="", chunk_size=500, chunk_overlap=50, metadata="{}", css_selector="" |
crawl_and_index_urls |
JSON 배열로 받은 여러 URL을 일괄 수집 | urls_json, chunk_size=500, chunk_overlap=50, metadata="{}", css_selector="", request_delay=1.0 |
crawl_sitemap |
sitemap.xml을 파싱하여 전체 사이트 크롤링 | sitemap_url, max_pages=50, chunk_size=500, chunk_overlap=50, metadata="{}", css_selector="", request_delay=1.0 |
crawl_recursive |
시작 URL에서 링크를 BFS로 따라가며 재귀 크롤링 | start_url, max_pages=20, same_domain_only=True, chunk_size=500, chunk_overlap=50, metadata="{}", css_selector="", request_delay=1.0 |
크롤링 워크플로우:
crawl_and_index(url)
↓
fetch URL → BeautifulSoup 정제 → delete_chunks(기존 청크 제거)
↓
TextChunker(chunk_size=500, chunk_overlap=50) → retrieval_index x N
↓
"indexed 'url': 12 chunks"
beautifulsoup4설치를 권장합니다.crawl_recursive는 필수입니다.
HttpMCP — 자동 재시도·백오프·타임아웃이 적용된 HTTP 클라이언트
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
http_get |
HTTP GET 요청 | url, headers="{}", params="{}" |
http_post |
HTTP POST (JSON body) | url, json_body="{}", headers="{}" |
응답: {"status_code": int, "body": str, "ok": bool, "headers": dict}
SchedulerMCP — 플러그인 가능한 백그라운드 작업 스케줄러
SCHEDULER_BACKEND 환경 변수로 백엔드를 선택합니다.
| 백엔드 | 값 | 특징 | 추가 설치 |
|---|---|---|---|
| APScheduler | apscheduler (기본값) |
인-프로세스, SQLite 영속 | — |
| Celery | celery |
분산 실행, Redis/RabbitMQ 브로커 | celery[redis] |
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
schedule_create |
새 작업 스케줄 등록 | job_id, func_name, trigger, trigger_args, kwargs="{}" |
schedule_list |
활성 스케줄 목록 조회 | — |
schedule_remove |
스케줄 취소 및 삭제 | job_id |
trigger 값: "interval" · "cron" · "date"
trigger_args 예시:
interval → '{"seconds": 30}' or '{"minutes": 5}'
cron → '{"hour": "*/2", "minute": "0"}'
date → '{"run_date": "2026-05-01 09:00:00"}'
NotificationMCP — 멀티 채널 알림 서비스 (dry-run 지원)
Retrieval/Scheduler와 달리 채널은 단일 선택이 아닌 동시 구성입니다. 에이전트가 상황에 따라 적합한 채널 tool을 직접 선택합니다. 환경 변수가 설정된 채널만 실제 발송되며, 미설정 채널은 자동으로 콘솔 출력으로 폴백됩니다.
| 채널 | 용도 | 환경 변수 |
|---|---|---|
| SMTP 이메일 | 보고서·요약 전달 | SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD |
| Slack | 팀 전체 알림 | SLACK_WEBHOOK_URL |
| Discord | 개발팀·커뮤니티 알림 | DISCORD_WEBHOOK_URL |
| Telegram | 온콜 담당자 모바일 푸시 | TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID |
| MS Teams | 기업 내부 채널 알림 | TEAMS_WEBHOOK_URL |
| Console | 항상 활성, dry-run 폴백 | — |
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
notify_email |
이메일 발송 | to, subject, body |
notify_slack |
Slack 채널 메시지 | channel, message |
notify_discord |
Discord 채널 메시지 | message |
notify_telegram |
Telegram Bot 메시지 | message |
notify_teams |
MS Teams 채널 메시지 (Adaptive Card) | message |
notify_console |
구조화된 콘솔 로그 | level, message |
NOTIFICATION_DRY_RUN=true설정 시 모든 채널이 실제 발송 없이 콘솔에 출력합니다.
AuthMCP — Fernet 대칭 암호화 기반 API 키 볼트 (SQLite 저장)
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
auth_store_key |
API 키 암호화 저장 | service, key |
auth_get_key |
저장된 키 복호화 조회 | service |
auth_validate |
키 존재 여부 확인 (평문 노출 없음) | service |
auth_list_services |
저장된 전체 서비스 목록 조회 (평문 노출 없음) | — |
auth_revoke |
저장된 키 영구 삭제 | service |
LoggingMCP — 구조화 로그 기록·조회·삭제. 에이전트가 실행 이력을 남기고 추후 분석할 수 있습니다.
LOGGING_BACKEND 환경 변수로 백엔드를 선택합니다.
| 백엔드 | 값 | write | query/tail | clear | 추가 설치 |
|---|---|---|---|---|---|
| SQLite | sqlite (기본값) |
✅ | ✅ | ✅ | — |
| File (Rotating) | file |
✅ | ✅ (현재 파일) | ✅ | — |
| Grafana Loki | loki |
✅ | ✅ (LogQL) | ❌ immutable | — |
| Elasticsearch / OpenSearch | elasticsearch |
✅ | ✅ | ✅ | — |
| Datadog | datadog |
✅ | ✅ (App key 필요) | ❌ immutable | — |
| PostgreSQL | postgres |
✅ | ✅ | ✅ | psycopg2-binary |
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
log_write |
구조화 로그 엔트리 기록 | level, message, source="", metadata="{}" |
log_query |
조건(레벨·소스·기간)으로 로그 검색 | level="", source="", since="", until="", limit=50 |
log_tail |
최신 N개 엔트리 조회 | n=20, source="" |
log_clear |
오래된 로그 삭제 | before="" (ISO 8601), source="" |
각 로그 엔트리는 다음 구조로 반환됩니다:
{
"id": 42,
"timestamp": "2026-04-30T14:30:00.123456+00:00",
"level": "INFO",
"source": "search_agent",
"message": "항공편 검색 완료",
"metadata": {"check": 3, "cheapest_price": 198.45}
}백엔드별 설정:
# SQLite (기본값 — 설정 불필요)
LOGGING_BACKEND=sqlite
LOGGING_DB_PATH=data/agent_logs.db
# File — 10 MB 회전, 백업 5개
LOGGING_BACKEND=file
LOGGING_FILE_PATH=data/agent.log
LOGGING_FILE_MAX_BYTES=10485760
LOGGING_FILE_BACKUP_COUNT=5
# Grafana Loki
LOGGING_BACKEND=loki
LOGGING_LOKI_URL=http://localhost:3100
LOGGING_LOKI_LABELS={"app": "agentic-ai", "env": "dev"}
# Elasticsearch / OpenSearch
LOGGING_BACKEND=elasticsearch
LOGGING_ES_URL=http://localhost:9200
LOGGING_ES_INDEX=agentic-ai-logs
LOGGING_ES_API_KEY= # 선택사항
# Datadog
LOGGING_BACKEND=datadog
LOGGING_DATADOG_API_KEY=<dd-api-key>
LOGGING_DATADOG_APP_KEY=<dd-app-key> # query/tail에 필요
LOGGING_DATADOG_SITE=datadoghq.com # 또는 datadoghq.eu / us3 / us5 / ap1
LOGGING_DATADOG_SERVICE=agentic-ai
# PostgreSQL (metadata 컬럼이 JSONB — 필드 직접 쿼리 가능)
LOGGING_BACKEND=postgres
LOGGING_POSTGRES_DSN=postgresql://user:password@localhost:5432/dbname
LOGGING_POSTGRES_TABLE=agent_logsLoki와 Datadog는 로그가 불변(immutable) 이므로
log_clear를 지원하지 않습니다.
Loki는 retention 정책, Datadog는 콘솔의 Archive 설정으로 데이터를 관리합니다.
flight_tools — 항공편 검색 tool. Mock(기본값) 또는 SerpAPI(Google Flights 실시간) 백엔드로 동작합니다.
flight_search는ALL_TOOLS에 포함되지 않습니다.configure_flight_client()로 백엔드를 초기화한 후 에이전트별로 직접 추가해야 합니다.
| Tool | 설명 | 주요 파라미터 |
|---|---|---|
flight_search |
두 공항 간 항공편을 검색하고 최저가·임계값 비교 결과 반환 | origin, destination, date, max_price, check_number=0, adults=1, return_date="", depart_after="", depart_before="" |
반환값 (JSON):
{
"check_number": 1,
"mode": "mock",
"origin": "ICN",
"destination": "NRT",
"date": "2026-08-01",
"max_price_threshold": 250000,
"cheapest_price": 198000,
"below_threshold": true,
"total_flights_found": 8,
"top5_flights": [{"flight_id": "KE703", "airline": "Korean Air", "price": 198000, ...}],
"searched_at": "2026-05-23T10:00:00"
}백엔드 설정:
# Mock (기본값 — API 키 불필요)
FLIGHT_API_MODE=mock
# SerpAPI — Google Flights 실시간 데이터 (무료 100회/월)
FLIGHT_API_MODE=serpapi
SERP_API_KEY=<your-serpapi-key>
FLIGHT_CURRENCY=KRWcore/ — LangChain tool이 아닌 에이전트 내부에서 직접 임포트하는 공통 Python API.
| 함수 | 설명 | 주요 파라미터 |
|---|---|---|
make_llm(tools, structured_output, thinking) |
활성 provider에 맞는 LLM 반환. tools/structured_output 지정 시 자동 바인딩 |
tools=None, structured_output=None, thinking=None |
make_thinking_llm(structured_output) |
추론 우선 LLM 반환. local은 TwoStepLLM, 나머지는 native structured output |
structured_output (Pydantic 클래스) |
invoke_with_retry(llm, messages, n) |
None 반환·예외 발생 시 최대 n회 재시도 | llm, messages, n=3 |
LLM_PROVIDER 별 동작:
| 값 | 사용 LLM | 필요 환경변수 |
|---|---|---|
anthropic (기본값) |
ChatAnthropic |
ANTHROPIC_API_KEY, LLM_MODEL |
openai |
ChatOpenAI |
OPENAI_API_KEY, OPENAI_MODEL |
gemini |
ChatGoogleGenerativeAI |
GEMINI_API_KEY, GEMINI_MODEL |
local |
ChatOpenAI (OpenAI-compat) |
LOCAL_LLM_BASE_URL, LOCAL_LLM_MODEL |
thinking 파라미터는 local provider에서만 의미 있습니다 (Ollama think 모드). openai·gemini·anthropic에서는 무시됩니다.
| 함수 | 설명 | 주요 파라미터 |
|---|---|---|
retrieve_context(query, top_k, max_chars, min_score, metadata_filter, section_header) |
Retrieval MCP를 호출해 관련 청크를 검색하고 LLM 프롬프트용 문자열로 조립. 실패 시 "" 반환 |
query, top_k=3, max_chars=2000, min_score=0.05, metadata_filter=None, section_header="Retrieved Knowledge" |
retrieval_build_context tool과의 차이:
retrieve_context (core/rag.py) |
retrieval_build_context (tool) |
|
|---|---|---|
| 사용 주체 | 에이전트 Python 코드 내부 | LLM이 tool call로 호출 |
min_score 필터 |
✅ 지원 | ❌ 없음 |
metadata_filter |
dict 직접 전달 | JSON 문자열 파싱 |
| 실패 시 | "" 반환 (안전) |
오류 메시지 반환 |
네 가지 예시는 각각 다른 tool 조합을 사용하여 실제 시나리오를 처리합니다.
사용 Tool:
retrieval_index·retrieval_search·memory_set·notify_slack·notify_console
Agent 구성 (단일 Executor 에이전트 — 범용 Planner→Executor→Reviewer 워크플로우 사용):
| 단계 | 역할 | 사용 Tool |
|---|---|---|
| FAQ 적재 | 6개 FAQ 문서를 검색 인덱스에 등록 (중복 실행 시 덮어쓰기) | retrieval_index |
| 질문 검색 | 사용자 질문과 코사인 유사도 비교, top-5 결과 반환 | retrieval_search |
| 세션 기록 | 질문을 support 네임스페이스에 영속 저장 |
memory_set |
| 답변/에스컬레이션 | score ≥ 0.1 이면 답변, 미달이면 Slack 에스컬레이션 | notify_slack, notify_console |
워크플로우:
사용자 질문
↓
retrieval_index ← FAQ 문서 6개를 인덱스에 적재 (idempotent)
↓
retrieval_search ← 질문과 유사도 높은 FAQ 검색 (top_k=5)
↓
memory_set ← 질문을 'support' 네임스페이스에 저장
↓
[score ≥ 0.1?]
YES → 답변 반환
NO → notify_slack → #support-escalation 채널 에스컬레이션
주요 설계 포인트:
| 포인트 | 설명 |
|---|---|
| Idempotent 인덱싱 | FAQ 문서를 매 실행 시 재적재해도 동일 doc_id는 덮어쓰기 처리 — 중복 없이 항상 최신 상태 유지 |
| 유사도 검색 | retrieval_search가 질문과 FAQ 문서 간 유사도를 계산, score 필드로 신뢰도 수치 반환 |
| 임계값 기반 에스컬레이션 | score < 0.1이면 LLM이 답변을 생성하지 않고 즉시 Slack 에스컬레이션 — 오답 방지 |
| 네임스페이스 격리 | memory_set(namespace="support")로 다른 예시 데이터와 메모리 충돌 없이 독립 저장 |
실행:
python main.py --example customer_support "환불 정책이 어떻게 되나요?"
python -m examples.customer_support "비밀번호를 잊어버렸어요"핵심 조합 원리: Retrieval(지식 조회) + Memory(세션 추적) + Notification(에스컬레이션)을 연결하면 도메인 지식 없이도 FAQ 기반 지원 시스템을 구성할 수 있습니다.
사용 Tool:
http_get·retrieval_index·retrieval_search·memory_set·notify_email·notify_console
Agent 구성 (단일 Executor 에이전트 — 범용 Planner→Executor→Reviewer 워크플로우 사용):
| 단계 | 역할 | 사용 Tool |
|---|---|---|
| 콘텐츠 수집 | 지정된 URL에서 HTML/JSON 페이지 fetch (retry + timeout 내장) | http_get |
| 동적 인덱싱 | 수집 페이지를 URL을 doc_id로 검색 인덱스에 추가 |
retrieval_index |
| 주제 검색 | 연구 주제와 관련 문서 검색, 유사도 점수 기반 top-5 선별 | retrieval_search |
| 결과 보존 | 생성된 요약을 research 네임스페이스에 영속 저장 |
memory_set |
| 보고서 발송 | 요약을 이메일로 전송 (dry-run 시 콘솔 출력) | notify_email, notify_console |
워크플로우:
연구 주제 + URL 목록 입력
↓
http_get ← 각 URL의 HTML/JSON 콘텐츠 수집 (자동 retry)
↓
retrieval_index ← 수집된 페이지를 doc_id=URL로 인덱싱
↓
retrieval_search ← 주제 관련 문서 검색 (top_k=5, 유사도 점수 포함)
↓
요약 생성 (3~5문장, 출처 명시)
↓
memory_set ← 요약을 'research' 네임스페이스에 저장
↓
notify_email ← 연구 보고서 이메일 발송 (dry-run 가능)
주요 설계 포인트:
| 포인트 | 설명 |
|---|---|
| 동적 인덱스 구축 | 실행 시마다 URL을 fetch하여 인덱싱 — 사전 정의된 문서가 아닌 실시간 수집 데이터를 검색 |
| 응답 본문 10,000자 제한 | HttpMCP가 응답을 자동 truncate — LLM 컨텍스트 오버플로 방지 |
| URL을 doc_id로 활용 | 동일 URL 재수집 시 자동 덮어쓰기, 인덱스 중복 방지 |
| Dry-run 이메일 | NOTIFICATION_DRY_RUN=true로 SMTP 설정 없이도 전체 파이프라인 테스트 가능 |
실행:
python main.py --example research "Python async patterns"
EMAIL_RECIPIENT=you@email.com python -m examples.research_agent핵심 조합 원리: HTTP(데이터 수집) + Retrieval(동적 인덱싱) + Memory(결과 보존) + Notification(보고서 전달)로 완전한 research-to-report 파이프라인을 구성합니다.
사용 Tool:
http_get·memory_set·memory_list_keys·notify_slack·notify_console·schedule_create
Agent 구성 (단일 Executor 에이전트 — 범용 Planner→Executor→Reviewer 워크플로우 사용):
| 단계 | 역할 | 사용 Tool |
|---|---|---|
| 헬스체크 | 각 대상 URL에 HTTP GET 요청, 응답 상태 수집 | http_get |
| 결과 저장 | URL별 status_code, ok, 타임스탬프를 monitoring 네임스페이스에 저장 |
memory_set |
| 장애 알림 | status_code ≠ 200인 타겟마다 즉시 Slack #alerts 채널 발송 |
notify_slack |
| 요약 리포트 | 전체 타겟 중 정상/비정상 집계 후 #monitoring 채널 요약 발송 |
notify_slack |
| 완료 로그 | 실행 완료 구조화 로그 출력 | notify_console |
| 반복 스케줄 | APScheduler에 주기적 헬스체크 job 등록 (선택) | schedule_create |
워크플로우:
대상 URL 목록 (예: 3개 엔드포인트)
↓
http_get ← 각 URL 헬스체크 (HTTP status 확인, retry 내장)
↓
memory_set ← 결과를 'monitoring' 네임스페이스에 URL별 저장
↓
[status_code != 200?]
YES → notify_slack → #alerts 채널 장애 알림 (타겟별)
↓
notify_slack ← #monitoring 채널 헬스체크 요약 리포트 (X/Y 정상)
↓
notify_console ← 실행 완료 로그
주요 설계 포인트:
| 포인트 | 설명 |
|---|---|
| URL slug 키 전략 | health_https___api_example_com_health 형태로 URL을 정규화하여 memory key로 사용 — 특수문자 충돌 없이 URL 기반 조회 가능 |
| per-target 알림 | 각 실패 URL마다 개별 Slack 메시지 발송 — 일괄 집계가 아닌 즉각 감지 |
| APScheduler 통합 | schedule_create(func_name="health_check_all", trigger="interval")로 에이전트가 스스로 반복 실행을 예약 |
| 함수 사전 등록 | SchedulerMCP.register("health_check_all", fn)으로 실행 함수를 화이트리스트에 등록 — 임의 코드 실행 방지 |
| MONITORING_TARGETS 환경변수 | 쉼표 구분 URL 목록으로 대상 변경, 코드 수정 불필요 |
실행:
python main.py --example monitoring
MONITORING_TARGETS=https://api.example.com/health python -m examples.monitoring_agent핵심 조합 원리: HTTP(상태 확인) + Memory(이력 저장) + Notification(이상 감지 알림)을 연결하면 어떤 인프라에도 적용 가능한 모니터링 에이전트가 됩니다. schedule_create를 추가하면 주기적 자동 실행으로 확장됩니다.
사용 Tool:
flight_search·memory_get·memory_set·notify_email·notify_console·log_write
자연어 목표를 자동으로 검색 계획으로 분해하고, 가격 조건 충족 시 이메일 알림을 발송하는 멀티 에이전트 시스템입니다. SerpAPI(Google Flights 실시간 데이터) 또는 Mock 모드로 동작합니다.
에이전트 구성 (Principle of Least Privilege — 각 에이전트는 필요한 tool만 접근):
| 에이전트 | 역할 | 전용 Tool / 출력 |
|---|---|---|
PlannerAgent |
자연어 목표 → 구체적 노선·날짜 목록 분해 | Pydantic SearchPlan (tool 없음) |
SearchAgent |
항공편 검색, 결과를 memory에 저장 | flight_search, memory_set, notify_console, log_write |
PriceAnalysisAgent |
가격 vs 임계값 비교 + RAG 지식·히스토리 참조 | Pydantic _PriceDecision (tool 없음) |
NotificationAgent |
조건 충족 시 이메일 알림 발송, 미충족 시 로그만 기록 | notify_email, memory_set, log_write |
ReflectionAgent |
체크 결과 회고, 다음 사이클 전략 메모 저장 | Pydantic _ReflectionResult (tool 없음) |
ProactivityAgent |
과거 패턴·캘린더·현재가 분석 → 자율 행동 결정 | Pydantic ProactivityAnalysis (tool 없음) |
워크플로우:
[Goal Decomposition 모드]
목표 자연어 입력
↓
PlannerAgent → SearchPlan (노선·날짜 목록, 목표 범위 전체 커버)
↓ (각 타겟별 가격 스캔 → 목표가 이하 옵션 수집)
[단일 체크 사이클]
START
↓
SearchAgent ──[flight_search]──→ memory_set(latest_search) → notify_console
↓
PriceAnalysisAgent ──[memory_get + RAG + 가격 히스토리]──→ should_book 결정
↓
[should_book?]
YES ──→ NotificationAgent ──[notify_email]──→ 예약 링크 이메일 발송
NO ──→ NotificationAgent ──[log_write]──→ 스킵 로그
↓
ReflectionAgent (thinking) ──→ 트렌드 분석 + 전략 메모 memory 저장
↓
END (cycle) → sleep → next cycle
실행:
# Mock 모드 (기본값 — 설정 없이 즉시 실행)
python -m examples.flight_monitor.run
# Mock — 딜 시뮬레이션 (3번째, 7번째 체크에서 임계값 이하 가격 발생)
python -m examples.flight_monitor.run \
--origin ICN --dest BKK --date 2026-08-01 \
--max-price 400000 --cheap-on 3 7
# SerpAPI — Goal Decomposition 모드 (자연어 목표 자동 분해)
python -m examples.flight_monitor.run \
--mode serpapi \
--goal "ICN에서 동남아 30만원 이하 최저가를 2026년 7월~2027년 1월 사이에서 찾아줘" \
--max-price 300000 --currency KRW --user <이름>
# SerpAPI — 단일 노선 모드
python -m examples.flight_monitor.run \
--mode serpapi --origin ICN --dest NRT --date 2026-07-15 --max-price 250000출력 예시 (Goal Decomposition):
🧠 PlannerAgent: 목표 분석 및 검색 계획 수립 중...
전략: 후쿠오카·오사카·도쿄·상하이 × 6~7월 날짜 조합으로 300,000 KRW 이하 탐색
탐색 대상: 5개 날짜/노선
[1/5] ICN → FUK 2026-06-09 최저가: 102,600 KRW ✅
[2/5] ICN → KIX 2026-06-16 최저가: 117,600 KRW ✅
[3/5] ICN → NRT 2026-07-07 최저가: 138,300 KRW ✅
[4/5] ICN → PVG 2026-06-23 최저가: 286,000 KRW ✅
[5/5] ICN → FUK 2026-06-30 최저가: 102,600 KRW ✅
목표가 이하 옵션 5개 → 이메일에 모두 포함
🎉 최적 선택: 2026-06-09 102,600 KRW (T'Way Air TW 207)
이메일 제목: 🎉 목표가 이하 항공권 5개 발견! 최저 102,600 KRW
→ (수신자 이메일) (5개 옵션 카드 + 노선별 예약 딥링크 포함)
주요 설계 포인트:
| 포인트 | 설명 |
|---|---|
| Goal Decomposition | PlannerAgent가 자연어 목표를 목표 범위를 완전히 커버하는 구체적 노선·날짜 목록으로 분해 — 날짜 지정 없이도 최적 탐색 가능 |
| 플러그인 Flight Client | mcp/flight.py의 BaseFlightClient ABC — Mock / SerpAPI(Google Flights) 교체 가능 |
| RAG + 장기 메모리 | PriceAnalysisAgent가 ChromaDB 지식베이스(노선 가격 전략)와 SQLite 가격 히스토리를 함께 참조하여 결정 |
| Reflection 루프 | ReflectionAgent가 매 체크 후 가격 트렌드를 분석하고 전략 메모를 Memory에 저장 → 다음 사이클의 PriceAnalysisAgent가 읽어 의사결정에 반영 |
단일 ToolNode + active_phase 라우팅 |
하나의 공유 ToolNode가 모든 에이전트의 tool call을 처리하고 FlightState.active_phase로 복귀 에이전트 결정 |
| 에이전트별 thinking 전략 | Reflection/Proactivity는 TwoStepLLM(core/llm.py, think=True 추론 → think=False 포맷), Planner는 make_llm(thinking=True)(로컬 LLM structured output 안정성), Search/Notification은 tool call 정밀도 우선 |
| 목표가 이하 통합 이메일 | Goal Decomposition 모드에서 예산 이하 옵션을 최대 20개 수집, 옵션 카드를 반복하는 한 통의 이메일로 발송 — 노선별 예약 딥링크(Google Flights/네이버/인터파크/스카이스캐너) 포함 |
| 멀티 LLM 지원 | LLM_PROVIDER 한 줄로 Anthropic / OpenAI / Gemini / 로컬(Ollama·vLLM) 전환 — 코드 변경 없음 → LOCAL_LLM_SETUP.md 참고 |
| 재현 가능한 Mock 시뮬레이션 | --cheap-on 3 7으로 딜 발생 체크 번호 지정 → API 없이 전체 플로우 결정론적 테스트 가능 |
핵심 조합 원리:
PlannerAgent+SearchAgent+PriceAnalysisAgent로 자율 탐색 파이프라인 구성ReflectionAgent가 과거 체크를 회고하여 다음 사이클 전략을 자동 생성- Notification은 목표가 이하 옵션 최대 20개를 한 통의 이메일에 카드 형식으로 발송 — 노선별 예약 딥링크(Google Flights/네이버/인터파크/스카이스캐너) 포함
LLM_PROVIDER한 줄로 Anthropic / OpenAI / Gemini / 로컬 GPU LLM 무중단 전환
agentic_ai_project/
│
├── config.py # 환경 변수 로드 및 전역 설정
├── main.py # CLI 진입점
├── requirements.txt
├── .env.example # 환경 변수 템플릿
│
├── core/
│ ├── base_mcp.py # BaseMCP 추상 클래스 + MCPResult 데이터클래스
│ ├── llm.py # LLM 팩토리 (make_llm, make_thinking_llm, TwoStepLLM, invoke_with_retry)
│ └── rag.py # RAG 유틸리티 (retrieve_context — Retrieval MCP 검색 + 프롬프트 포맷팅)
│
├── mcp/ # MCP 퍼사드 — 백엔드에 위임
│ ├── memory.py # KV 스토어 (SQLite)
│ ├── retrieval.py # 문서 검색 (백엔드 선택: RETRIEVAL_BACKEND)
│ ├── http.py # HTTP 클라이언트 (retry + backoff)
│ ├── scheduler.py # 잡 스케줄러 (백엔드 선택: SCHEDULER_BACKEND)
│ ├── notification.py # SMTP / Slack / Discord / Telegram / Teams / console
│ ├── auth.py # Fernet 암호화 키 볼트 (SQLite)
│ ├── logging_mcp.py # 구조화 로그 (백엔드 선택: LOGGING_BACKEND)
│ ├── flight.py # 항공 검색/예약 클라이언트 (Mock / SerpAPI)
│ └── backends/ # 플러그인 백엔드 구현체
│ ├── memory/
│ │ ├── base.py # BaseMemoryBackend ABC
│ │ └── sqlite.py # SQLite
│ ├── retrieval/
│ │ ├── base.py # BaseRetrievalBackend ABC
│ │ ├── chunker.py # TextChunker + clean_html_text 유틸리티
│ │ ├── bm25_sqlite.py # BM25 + SQLite FTS5 (추가 의존성 없음)
│ │ ├── vector.py # ChromaDB 임베딩 검색 (기본값)
│ │ └── postgres.py # PostgreSQL tsvector 전문 검색
│ ├── scheduler/
│ │ ├── base.py # BaseSchedulerBackend ABC
│ │ ├── apscheduler.py # APScheduler + SQLite (기본값)
│ │ └── celery.py # Celery + Redis/RabbitMQ
│ └── logging/
│ ├── base.py # BaseLoggingBackend ABC
│ ├── sqlite.py # SQLite (기본값)
│ ├── file.py # Rotating JSON Lines 파일
│ ├── loki.py # Grafana Loki HTTP Push API
│ ├── elasticsearch.py# Elasticsearch / OpenSearch
│ ├── datadog.py # Datadog Logs API
│ └── postgres.py # PostgreSQL (JSONB metadata)
│
├── tools/ # LangChain @tool 래퍼 (35개 tool)
│ ├── memory_tools.py # 4 tools
│ ├── retrieval_tools.py # 5 tools (chunking + RAG 컨텍스트 조립)
│ ├── crawl_tools.py # 4 tools (RAG 데이터 수집 파이프라인)
│ ├── http_tools.py # 2 tools
│ ├── scheduler_tools.py # 3 tools
│ ├── notification_tools.py # 6 tools
│ ├── auth_tools.py # 5 tools
│ ├── logging_tools.py # 4 tools (write / query / tail / clear)
│ └── flight_tools.py # 2 tools (configure_flight_client() 필요)
│
├── agents/
│ ├── planner.py # 실행 계획 생성 에이전트
│ ├── executor.py # 도구 호출 실행 에이전트
│ └── reviewer.py # 결과 품질 검토 에이전트
│
├── graph/
│ ├── state.py # AgentState TypedDict
│ └── workflow.py # LangGraph StateGraph 정의
│
├── examples/
│ ├── customer_support.py # FAQ 검색 + 에스컬레이션
│ ├── research_agent.py # HTTP 수집 + 문서 인덱싱 + 이메일 보고
│ ├── monitoring_agent.py # 헬스체크 + Slack 알림
│ └── flight_monitor/ # ✈️ 항공권 자동 모니터링 (SerpAPI / Mock)
│ ├── state.py # FlightState TypedDict
│ ├── agents.py # Search/PriceAnalysis/Notification/Reflection 에이전트
│ ├── planner.py # PlannerAgent — 자연어 목표 → SearchPlan 분해
│ ├── proactivity.py # ProactivityAgent — 과거 패턴·캘린더 분석 자율 행동
│ ├── workflow.py # LangGraph StateGraph
│ ├── knowledge_loader.py # RAG 지식베이스 로드 (pricing_strategy, route_pricing_guide)
│ ├── run.py # 모니터링 루프 진입점 (Goal / 단일 노선 / Proactive 모드)
│ └── LOCAL_LLM_SETUP.md # 로컬 GPU LLM(Ollama) 설정 가이드
│
├── assets/
│ ├── architecture_overview.svg
│ ├── workflow_graph.svg
│ ├── mcp_tools_map.svg
│ └── flight_monitor_workflow.svg
│
└── data/ # 런타임 생성
├── memory.db # MemoryMCP (SQLite)
├── retrieval_bm25.db # RETRIEVAL_BACKEND=bm25_sqlite
├── scheduler.db # SCHEDULER_BACKEND=apscheduler
├── auth.db
└── vector_retrieval/ # RETRIEVAL_BACKEND=vector (ChromaDB)
pip install -r requirements.txtcp .env.example .env.env에서 LLM 제공자를 선택합니다:
# Anthropic Claude
LLM_PROVIDER=anthropic
ANTHROPIC_API_KEY=sk-ant-...
LLM_MODEL=claude-sonnet-4-6
# OpenAI GPT
LLM_PROVIDER=openai
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o
# Google Gemini — requires: pip install langchain-google-genai
LLM_PROVIDER=gemini
GEMINI_API_KEY=AIza...
GEMINI_MODEL=gemini-2.5-flash
# 로컬 LLM (Ollama / vLLM) — 추론 비용 없음
LLM_PROVIDER=local
LOCAL_LLM_BASE_URL=http://localhost:11434/v1
LOCAL_LLM_MODEL=qwen3:30b-a3b-q4_K_M
LOCAL_LLM_API_KEY=ollama로컬 GPU LLM 설정 상세: examples/flight_monitor/LOCAL_LLM_SETUP.md
# PostgreSQL 백엔드 (Retrieval)
pip install psycopg2-binary
# Vector 백엔드 — ChromaDB (Retrieval)
pip install chromadb
# Celery 백엔드 (Scheduler)
pip install celery[redis]
# HTML 파싱 품질 향상 (crawl_tools 사용 시 권장, crawl_recursive 필수)
pip install beautifulsoup4.env에서 백엔드를 선택합니다:
# Retrieval 백엔드
RETRIEVAL_BACKEND=vector # 기본값 — ChromaDB 임베딩 시맨틱 검색
RETRIEVAL_BACKEND=bm25_sqlite # BM25 + SQLite FTS5 (추가 설치 불필요, 대용량 권장)
RETRIEVAL_BACKEND=postgres # PostgreSQL tsvector 전문 검색
# Scheduler 백엔드
SCHEDULER_BACKEND=apscheduler # 기본값 — 인-프로세스
SCHEDULER_BACKEND=celery # 분산 실행; SCHEDULER_CELERY_BROKER 필요python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
# 출력값을 .env의 AUTH_FERNET_KEY에 설정# 커스텀 작업
python main.py "내 이름을 기억하고 Slack으로 알려줘"
# 예시 시나리오
python main.py --example customer_support
python main.py --example research "LangGraph multi-agent"
python main.py --example monitoring도메인 독립성: 모든 MCP와 Tool은 특정 도메인 로직 없이 구현되어 있습니다. 고객 지원, 연구, 모니터링, 금융, 의료 등 어떤 도메인에서도 동일한 Tool을 재사용할 수 있습니다.
플러그인 백엔드: Retrieval, Scheduler, Logging은 환경 변수 하나로 백엔드를 교체합니다. BaseRetrievalBackend / BaseSchedulerBackend / BaseLoggingBackend ABC를 구현하면 새로운 백엔드를 추가할 수 있습니다. 에이전트 코드와 Tool 코드는 변경 없이 그대로 사용합니다.
단방향 의존성: Tool → MCP → Backend → DB/Service 방향으로만 의존합니다. 에이전트는 Tool만 알고, MCP 구현과 백엔드는 교체 가능합니다.
Singleton MCP: 각 MCP는 프로세스당 하나의 인스턴스를 유지합니다. 백엔드 연결을 공유하여 리소스를 절약하고, Tool에서 get_xxx_mcp() 팩토리로 접근합니다.
RAG 파이프라인 내장: crawl_tools → retrieval_index(chunk_size>0) → retrieval_build_context 조합으로 완전한 RAG 수집·검색·컨텍스트 조립 파이프라인이 구성됩니다. TextChunker는 문단 → 문장 → 문자 단계적 분할 전략을 사용하며, 재크롤링 시 기존 청크를 자동 교체합니다.
Dry-run 우선: NOTIFICATION_DRY_RUN=true로 실제 이메일/Slack 발송 없이 전체 워크플로우를 안전하게 테스트할 수 있습니다.