Content
# Agentic AI Project
> English version: [README.en.md](README.en.md)
도메인에 관계없이 멀티 에이전트 아키텍처에서 재사용할 수 있는 **공통 Tool 및 MCP(Model Context Protocol)** 구현체와, 이를 조합하여 실제 사용 사례를 보여주는 예시 시나리오로 구성된 프레임워크입니다.
<img src="assets/ChatGPT Image 2026년 4월 29일 오전 01_39_24.png" alt="Agentic AI 아키텍처 한눈에 보기" width="100%"/>
> Agentic AI의 핵심 구성 요소 — LLM + Tools + Memory + Evaluation + Loop.
> 이 프로젝트는 위 개념을 **LangGraph + MCP** 기반으로 실제 구현한 것입니다.
---
## Architecture Overview
### Layered Design
| Layer | 역할 | 구성 요소 |
|-------|------|-----------|
| **Multi-Agent Graph** | LLM 에이전트들이 협력하여 작업 수행 | Planner → Executor → Reviewer |
| **Tool Layer (2가지 경로)** | LLM이 호출 가능한 tool 노출 | ① in-process LangChain `@tool` 래퍼(`tools/`, 동기) ② **실제 MCP 서버**(`mcp_servers/`, FastMCP stdio — 도구명·docstring 동일) |
| **Service Layer** | 실제 백엔드 시스템과 연결되는 퍼사드 | 8개 서비스 클래스 (`services/`, 플러그인 백엔드) |
두 tool 경로는 병행 운영됩니다: `main.py`의 기본 워크플로우는 **MCP 서버 경유**(stdio 서브프로세스 + `ainvoke`), `examples/` 3종과 flight_monitor는 기존 in-process 경로(동기 `invoke`)를 사용합니다. 어느 경로든 동일한 `services/` 퍼사드로 수렴하므로 동작은 같습니다.
> **Breaking changes (2026-07):**
> 1. 과거 최상위 패키지 `mcp/` 는 공식 `mcp` PyPI SDK와 이름이 충돌하여 **`services/` 로 개명**되었습니다. 외부 프로젝트에서 `from mcp.xxx import ...` 로 퍼사드를 직접 import했다면 `from services.xxx import ...` 로 바꿔야 합니다. `from tools import ...` / `ALL_TOOLS` API는 변경 없습니다.
> 2. 의존성이 **langchain/langgraph 1.x 라인**으로 올라갔습니다. langgraph 0.2.x 시절 API(`MemorySaver`, `interrupt/Command` 등)를 쓰는 소비 프로젝트는 1.x로 마이그레이션하거나 별도 가상환경을 쓰세요. 또한 MCP 경유 도구 결과가 문자열이 아닌 **content block 리스트**(`[{"type": "text", "text": ...}]`)로 반환됩니다 — LLM 경유(ToolNode)는 영향 없고, 결과를 직접 문자열 비교하는 코드만 텍스트 추출이 필요합니다.
---
## LangGraph Workflow
```
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`를 자동으로 실행하고 결과를 메시지로 반환
---
## Tool & MCP Reference
> 아래 도구들은 in-process `@tool` 래퍼(`tools/`)와 MCP 서버(`mcp_servers/`) 어느 경로로 호출해도 **이름·파라미터·반환 형식(`'ERROR: ...'` 규약 포함)이 동일**합니다.
### Memory Tools
`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"` |
---
### Retrieval Tools
`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 (RAG 데이터 수집)
`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`는 필수입니다.
---
### HTTP Tools
`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}`
---
### Scheduler Tools
`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"}'
```
---
### Notification Tools
`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` 설정 시 모든 채널이 실제 발송 없이 콘솔에 출력합니다.
---
### Auth Tools
`AuthMCP` — Fernet 대칭 암호화 기반 API 키 볼트 (SQLite 저장)
| Tool | 설명 | 주요 파라미터 |
|------|------|--------------|
| `auth_store_key` | API 키 암호화 저장 | `service`, `key` |
| `auth_get_key` | 저장된 키 복호화 조회 | `service` |
| `auth_validate` | 키 존재 여부 확인 (평문 노출 없음) | `service` |
| `auth_list_services` | 저장된 전체 서비스 목록 조회 (평문 노출 없음) | — |
| `auth_revoke` | 저장된 키 영구 삭제 | `service` |
---
### Logging Tools
`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=""` |
각 로그 엔트리는 다음 구조로 반환됩니다:
```json
{
"id": 42,
"timestamp": "2026-04-30T14:30:00.123456+00:00",
"level": "INFO",
"source": "search_agent",
"message": "항공편 검색 완료",
"metadata": {"check": 3, "cheapest_price": 198.45}
}
```
**백엔드별 설정:**
```bash
# 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_logs
```
> Loki와 Datadog는 로그가 **불변(immutable)** 이므로 `log_clear`를 지원하지 않습니다.
> Loki는 retention 정책, Datadog는 콘솔의 Archive 설정으로 데이터를 관리합니다.
---
### Flight Tools
`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):
```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"
}
```
**백엔드 설정:**
```bash
# SerpAPI (기본값 — Google Flights 실시간)
FLIGHT_API_MODE=serpapi
SERP_API_KEY=<your-serpapi-key>
FLIGHT_CURRENCY=KRW
# Mock — API 키 불필요, 로컬 테스트용
FLIGHT_API_MODE=mock
```
---
### Core Utilities
`core/` — LangChain tool이 아닌 에이전트 내부에서 직접 임포트하는 공통 Python API.
#### `core/llm.py` — Provider-agnostic LLM Factory
| 함수 | 설명 | 주요 파라미터 |
|------|------|--------------|
| `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, config)` | None 반환·예외 발생 시 최대 n회 재시도 | `llm`, `messages`, `n=3`, `config=None` |
`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`에서는 무시됩니다.
#### `core/rag.py` — RAG 컨텍스트 빌더
| 함수 | 설명 | 주요 파라미터 |
|------|------|--------------|
| `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 문자열 파싱 |
| 실패 시 | `""` 반환 (안전) | 오류 메시지 반환 |
---
## Multi-Agent Examples
네 가지 예시는 각각 다른 tool 조합을 사용하여 실제 시나리오를 처리합니다.
---
### Example 1: Customer Support Agent
> **사용 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")`로 다른 예시 데이터와 메모리 충돌 없이 독립 저장 |
**실행:**
```bash
python main.py --example customer_support "환불 정책이 어떻게 되나요?"
python -m examples.customer_support "비밀번호를 잊어버렸어요"
```
**핵심 조합 원리:** Retrieval(지식 조회) + Memory(세션 추적) + Notification(에스컬레이션)을 연결하면 도메인 지식 없이도 FAQ 기반 지원 시스템을 구성할 수 있습니다.
---
### Example 2: Research Agent
> **사용 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 설정 없이도 전체 파이프라인 테스트 가능 |
**실행:**
```bash
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 파이프라인을 구성합니다.
---
### Example 3: Monitoring Agent
> **사용 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 목록으로 대상 변경, 코드 수정 불필요 |
**실행:**
```bash
python main.py --example monitoring
MONITORING_TARGETS=https://api.example.com/health python -m examples.monitoring_agent
```
**핵심 조합 원리:** HTTP(상태 확인) + Memory(이력 저장) + Notification(이상 감지 알림)을 연결하면 어떤 인프라에도 적용 가능한 모니터링 에이전트가 됩니다. `schedule_create`를 추가하면 주기적 자동 실행으로 확장됩니다.
---
### Example 4: Flight Monitor Agent ✈️
> **사용 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 (예산 초과 대안 있음, Goal 모드) ──→ NotificationAgent ──→ 참고 메일 발송
NO (대안 없음) ──→ NotificationAgent 건너뜀
↓
ReflectionAgent (thinking) ──→ 트렌드 분석 + 전략 메모 memory 저장
↓
END (cycle) → sleep → next cycle
```
**실행:**
```bash
# 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 모드 — 프롬프트 파일(.json/.txt)에서 목표·옵션을 읽어와 실행
examples/flight_monitor/test/run_flight_monitor_goal.sh
```
**출력 예시 (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** | `services/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](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 **무중단 전환**
---
## Project Structure
```
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 검색 + 프롬프트 포맷팅)
│
├── services/ # 서비스 퍼사드 — 백엔드에 위임 (구 mcp/ — SDK 이름 충돌로 개명)
│ ├── 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)
│
├── mcp_servers/ # 실제 MCP 서버 (FastMCP stdio) — 도메인당 1서버
│ ├── _common.py # guard_stdout() — stdio JSON-RPC 스트림 보호
│ ├── memory.py # 4 tools (python -m mcp_servers.memory)
│ ├── retrieval.py # 9 tools (검색 5 + crawl 4, 기동 시 임베딩 모델 로드)
│ ├── http.py # 2 tools
│ ├── scheduler.py # 3 tools (stdio 수명 = 세션 수명 주의)
│ ├── notification.py # 6 tools
│ ├── auth.py # 5 tools
│ ├── logging_server.py # 4 tools
│ └── flight.py # 1 tool (FLIGHT_API_MODE=mock|serpapi, 기본 serpapi)
│
├── mcp_client/
│ └── loader.py # open_mcp_tools() — 서버 spawn + 지속 세션 + tool 로딩
│
├── 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)
│ ├── run.py # 모니터링 루프 진입점 (Goal / 단일 노선 / Proactive 모드)
│ ├── agents/ # 에이전트 노드 — Search/PriceAnalysis/Notification/Reflection
│ │ # + Planner(목표 분해)·Proactivity(자율 행동), 파일 1개 = 에이전트 1개
│ ├── graph/ # FlightState TypedDict + LangGraph StateGraph
│ ├── services/ # 이메일 포맷팅, RAG 지식 로더, 노선 지식 자동 갱신, Mock API
│ ├── test/ # Goal 모드 프롬프트 파일 실행 스크립트 (my_goal.json)
│ ├── knowledge/ # RAG 지식 문서 (노선 시세·가격 전략, 스캔 누적 자동 갱신)
│ └── LOCAL_LLM_SETUP.md # 로컬 GPU LLM(Ollama) 설정 가이드
│
├── scripts/
│ └── mcp_smoke.py # MCP 서버 스모크 테스트 (기동·tool 패리티·라운드트립)
│
├── 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)
```
---
## Running as MCP Servers
`mcp_servers/`의 각 모듈은 공식 `mcp` SDK의 FastMCP로 구현된 **실제 MCP 서버**(stdio transport)입니다.
```bash
# 개별 서버 직접 실행 (Claude Desktop 등 외부 MCP 클라이언트 연결용)
python -m mcp_servers.memory
python -m mcp_servers.flight # env로 백엔드 선택 (아래 참고)
# 스모크 테스트 — 7개 서버 기동 + tool 패리티(33개) + 라운드트립 검증
python scripts/mcp_smoke.py
python scripts/mcp_smoke.py --include-flight
```
`main.py`의 기본 워크플로우는 `mcp_client.open_mcp_tools()`가 서버들을 서브프로세스로 spawn하고
**그래프 실행 동안 세션을 유지**한 채 tool을 로딩합니다 (세션 없이 호출하면 tool 호출마다
서버가 새로 뜨므로 주의 — 특히 retrieval은 기동 시 임베딩 모델을 로드해 수 초 걸립니다).
| 환경 변수 | 대상 | 설명 |
|-----------|------|------|
| `FLIGHT_API_MODE` | flight 서버 | `mock` \| `serpapi`. **서버 기본값은 `serpapi`** (config.py의 기본값 `mock`과 다름 — flight_monitor 내부 로직용 기본값과 분리) |
| `FLIGHT_MOCK_BASE_URL` | flight 서버 | mock 모드에서 사용할 MockFlightAPI 주소 (기본 `http://127.0.0.1:18990`, 별도 기동 필요) |
주의사항:
- **scheduler 서버 수명**: stdio 서버는 클라이언트 세션과 함께 종료됩니다. APScheduler 잡도 프로세스와 함께 죽습니다(기존 in-process 동작과 동일). 상시 스케줄링이 필요하면 추후 HTTP transport로 기동해야 합니다.
- **stdout 보호**: stdio 서버에서 stdout은 JSON-RPC 스트림입니다. 퍼사드의 콘솔 출력(알림 미리보기 등)은 `guard_stdout()`에 의해 stderr로 우회됩니다.
- **HTTP 전환**: FastMCP 특성상 `mcp.run(transport=...)` 인자와 `mcp_client/loader.py`의 connection dict만 바꾸면 됩니다 — 서버 코드는 동일.
---
## Setup
### 1. Install dependencies
```bash
pip install -r requirements.txt
```
### 2. Configure environment
```bash
cp .env.example .env
```
`.env`에서 LLM 제공자를 선택합니다:
```bash
# 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](examples/flight_monitor/LOCAL_LLM_SETUP.md)
### 3. (선택) 다른 백엔드 설치
```bash
# 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`에서 백엔드를 선택합니다:
```bash
# 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 필요
```
### 4. Generate Auth encryption key (선택)
```bash
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
# 출력값을 .env의 AUTH_FERNET_KEY에 설정
```
### 5. Run
```bash
# 커스텀 작업
python main.py "내 이름을 기억하고 Slack으로 알려줘"
# 예시 시나리오
python main.py --example customer_support
python main.py --example research "LangGraph multi-agent"
python main.py --example monitoring
```
---
## Key Design Principles
**도메인 독립성**: 모든 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 발송 없이 전체 워크플로우를 안전하게 테스트할 수 있습니다.
