Skip to content

KAU-gitfit/gitfit-ai

BranchesTags

Repository files navigation

🤖 LangGraph Application Template

Python LangChain LangGraph FastAPI MongoDB Docker

LangGraph 기반 프로덕션 레디 AI 애플리케이션 템플릿

📋 프로젝트 개요

LangGraph와 FastAPI를 기반으로 한 확장 가능한 AI 애플리케이션 템플릿입니다. Supervisor 패턴, 비동기 처리, MongoDB 체크포인터를 활용한 세션 관리 등 프로덕션 환경에 필요한 모든 기능을 제공합니다.

🌟 주요 특징

  • 🏗️ 계층화된 아키텍처: API - Service - Repository 패턴
  • 🔄 비동기 처리: async/await 기반 전체 파이프라인
  • 🗄️ MongoDB 통합: 세션 체크포인터 및 채팅 히스토리 저장
  • 📊 구조화된 로깅: 노드별 실행 시간, 에러 추적, KST 시간대
  • 🤖 멀티 모델 지원: OpenAI, Anthropic, Google Gemini
  • 🎯 BaseNode 패턴: 재사용 가능한 노드 기반 아키텍처
  • 🚀 프로덕션 레디: Docker, 환경변수 관리, 에러 핸들링

📁 프로젝트 구조

langgraph-init/
├── 📁 src/
│   ├── 📁 agents/          # LangGraph 에이전트 구현
│   │   ├── node/              # BaseNode 기반 노드들
│   │   │   ├── base_node.py      # 추상 베이스 노드
│   │   │   ├── example_node.py   # 예시 노드
│   │   │   └── finalize_node.py  # 종료 노드
│   │   ├── state/             # 그래프 상태 정의
│   │   │   └── state.py          # GraphState 정의
│   │   ├── prompts/           # 프롬프트 관리
│   │   │   └── prompt_provider.py
│   │   ├── tools/             # 도구 정의
│   │   │   └── web_search.py
│   │   └── graph_builder.py   # 그래프 빌더
│   ├── 📁 api/             # FastAPI 웹 API 계층
│   │   ├── controller/        # API 컨트롤러
│   │   │   └── chat_controller.py
│   │   ├── service/           # 비즈니스 로직
│   │   │   └── chat_service.py
│   │   ├── repository/        # 데이터 접근
│   │   │   └── chat_history_repository.py
│   │   ├── entity/            # 엔티티 정의
│   │   │   └── chat_history_entity.py
│   │   ├── dto/               # 데이터 전송 객체
│   │   │   ├── request/
│   │   │   │   └── chat_request_dto.py
│   │   │   └── response/
│   │   │       ├── chat_response_dto.py
│   │   │       └── error_response_dto.py
│   │   └── config/            # API 서버 설정
│   │       └── server_config.py
│   ├── 📁 config/          # 환경 설정
│   │   └── settings.py        # 중앙 설정 관리
│   ├── 📁 db/              # 데이터베이스 연결
│   │   └── mongodb_manager.py
│   └── 📁 utils/           # 공통 유틸리티
│       ├── logger.py          # KST 로깅
│       └── model_manager.py   # 모델 관리
├── 🐳 docker-compose.yml   # Docker Compose
├── 🐳 Dockerfile          # Docker 설정
├── 📋 requirements.txt    # 의존성 목록
├── 📋 .env.example        # 환경변수 예시
└── 📋 main.py             # 앱 진입점

🏗️ 아키텍처

계층 구조

┌─────────────────────────────────────────┐
│         FastAPI Controller              │  # HTTP 엔드포인트
├─────────────────────────────────────────┤
│            Service Layer                │  # 비즈니스 로직
├─────────────────────────────────────────┤
│         LangGraph Execution             │  # 그래프 실행
│    ┌───────────┬───────────┬─────────┐  │
│    │   Node    │   Node    │  Node   │  │
│    └───────────┴───────────┴─────────┘  │
├─────────────────────────────────────────┤
│        Repository Layer                 │  # 데이터 접근
├─────────────────────────────────────────┤
│       MongoDB (Checkpoint + History)    │  # 영속성
└─────────────────────────────────────────┘

BaseNode 패턴

모든 노드는 BaseNode를 상속받아 일관된 인터페이스 제공:

  • 자동 로깅 (시작/종료, 소요시간)
  • 예외 처리 (ValueError → 400, Exception → 500)
  • 비동기 실행 (async def execute())
  • 상태 전파 (GraphState 기반)

🚀 빠른 시작

1. 프로젝트 설정

git clone <your-repo-url>
cd langgraph-init

# 가상환경 생성 (Python 3.10 이상 필요)
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 의존성 설치
pip install -r requirements.txt

2. 환경 변수 설정

# .env.example을 복사하여 .env 생성
cp .env.example .env

# .env 파일 편집
# 필수 설정:
# - OPENAI_API_KEY (또는 ANTHROPIC_API_KEY, GOOGLE_API_KEY)
# - MONGODB_URI (MongoDB Atlas 또는 로컬)

.env 예시:

# === 모델 설정 ===
CHAT_MODEL=gpt-4o-mini
EMBEDDING_MODEL=text-embedding-3-small
MODEL_TEMPERATURE=0.7
MODEL_MAX_TOKENS=1000

# === API 키 ===
OPENAI_API_KEY=sk-...
# ANTHROPIC_API_KEY=sk-ant-...
# GOOGLE_API_KEY=...

# === MongoDB ===
MONGODB_URI=mongodb+srv://user:pass@cluster.mongodb.net/?retryWrites=true&w=majority
MONGODB_DB_NAME=langgraph_db

# === API 서버 ===
API_HOST=0.0.0.0
API_PORT=8000
CORS_ORIGINS=http://localhost:3000,http://localhost:8000

# === 로깅 ===
LOG_LEVEL=INFO
TZ=Asia/Seoul

# === LangSmith (선택) ===
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=...
# LANGCHAIN_PROJECT=langgraph-app

3. API 서버 실행

# 개발 모드 (hot reload)
python main.py

# 또는 uvicorn 직접 실행
uvicorn main:app --reload --host 0.0.0.0 --port 8000

4. API 테스트

# 헬스 체크
curl http://localhost:8000/api/v1/health

# 채팅 요청
curl -X POST http://localhost:8000/api/v1/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user123",
    "session_id": "session456",
    "message": "안녕하세요!",
    "context": {"key": "value"},
    "image_url": null
  }'

5. API 문서 확인

서버 실행 후 브라우저에서:

🔧 지원하는 모델

채팅 모델

# OpenAI
CHAT_MODEL=gpt-4o
CHAT_MODEL=gpt-4o-mini
CHAT_MODEL=gpt-4-turbo

# Anthropic
CHAT_MODEL=claude-3-5-sonnet-20241022
CHAT_MODEL=claude-3-opus-20240229
CHAT_MODEL=claude-3-haiku-20240307

# Google Gemini
CHAT_MODEL=gemini-1.5-pro
CHAT_MODEL=gemini-1.5-flash

임베딩 모델

# OpenAI
EMBEDDING_MODEL=text-embedding-3-large
EMBEDDING_MODEL=text-embedding-3-small
EMBEDDING_MODEL=text-embedding-ada-002

🛠️ 새로운 노드 추가하기

# src/agents/node/my_node.py
from src.agents.node.base_node import BaseNode
from src.agents.state.state import GraphState

class MyNode(BaseNode):
    """커스텀 노드 구현"""

    def __init__(self):
        super().__init__("my_node")

    async def execute(self, state: GraphState) -> GraphState:
        """노드 로직 구현"""
        messages = state.get("messages", [])

        # 비즈니스 로직 구현
        # ...

        return {
            "messages": [...],
            "custom_field": "value"
        }

📊 로깅 시스템

from src.utils import get_logger

logger = get_logger("my.module.name")
logger.debug("디버그 메시지")
logger.info("정보 메시지")
logger.warning("경고 메시지")
logger.error("에러 메시지")

로그 출력 예시:

2025-10-29 14:30:25 KST [DEBUG] agents.nodes.example - [example] 노드 실행 시작
2025-10-29 14:30:25 KST [DEBUG] agents.nodes.example - 입력 메시지 수: 1
2025-10-29 14:30:27 KST [INFO] agents.nodes.example - LLM 분석 응답 생성 완료 (234자)
2025-10-29 14:30:27 KST [DEBUG] agents.nodes.example - [example] 노드 실행 완료 (소요 시간: 2.15초)

🗄️ MongoDB 체크포인터

LangGraph의 상태는 MongoDB에 자동으로 저장되어 세션 간 대화 이력을 유지합니다.

# 세션 ID로 대화 이력 자동 복원
config = {
    "configurable": {
        "thread_id": "session123"
    }
}
result = await graph.ainvoke(initial_state, config=config)

🐳 Docker 실행

# Docker Compose로 실행
docker-compose up -d

# 로그 확인
docker-compose logs -f app-server

# 중지
docker-compose down

# 개별 빌드
docker build -t langgraph-app .
docker run -p 8000:8000 --env-file .env langgraph-app

🎯 주요 개발 패턴

1. DTO 기반 API 통신

# Request DTO
class ChatRequestDTO(BaseModel):
    user_id: str
    session_id: str
    message: str
    context: Optional[Dict[str, Any]] = None
    image_url: Optional[str] = None

# Response DTO
class ChatResponseDTO(BaseModel):
    message: str
    metadata: Optional[Dict[str, Any]] = None
    tokens_used: int = 0

2. Service 계층 패턴

class ChatService:
    async def process_chat(self, request: ChatRequestDTO) -> ChatResponseDTO:
        # 1. 그래프 초기화 확인
        # 2. 초기 상태 준비
        # 3. 그래프 실행
        # 4. 응답 추출
        # 5. 히스토리 저장
        pass

3. Repository 패턴

class ChatHistoryRepository:
    async def save_chat(self, chat_history, message_pair):
        # MongoDB upsert 패턴
        pass

    async def get_session_history(self, session_id):
        # 세션별 조회
        pass

💡 개발 팁

모델 교체

# 환경변수만 변경하면 자동으로 다른 모델 사용
export CHAT_MODEL=claude-3-5-sonnet-20241022
export ANTHROPIC_API_KEY=sk-ant-...

새로운 도구 추가

# src/agents/tools/my_tool.py
from langchain_core.tools import tool

@tool
def my_custom_tool(query: str) -> str:
    """커스텀 도구 설명"""
    return f"결과: {query}"

프롬프트 관리

# src/agents/prompts/prompt_provider.py 활용
class PromptProvider:
    def get_custom_prompt(self, **kwargs) -> str:
        """커스텀 프롬프트 생성"""
        return f"시스템: {kwargs.get('context')}\n사용자: {kwargs.get('message')}"

📚 더 알아보기

About

No description, website, or topics provided.

Resources

Readme
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages