손코딩으로 배우는 프로덕션급 AI 에이전트 개발 가이드

게시 2026/01/02 업데이트 2026/01/05

By BLUEBUG

96 분읽는 시간

코드 한 줄 한 줄 직접 타이핑하며 배우는 7계층 아키텍처
Windows 환경에서 처음부터 배포까지 완벽 가이드 (Claude Code는 프로젝트 초기화 사용)

왜 손코딩인가?

AI 시대에 손코딩이라니 시대착오적으로 들릴 수 있습니다. Claude Code 같은 도구가 코드를 자동 생성해주는데 왜 직접 타이핑해야 할까요?

손코딩의 가치

1. 진짜 이해: 코드를 복사-붙여넣기하면 빠르지만, 직접 타이핑하면 각 라인의 의미를 생각하게 됩니다.

2. 근육 기억: 손가락이 기억하는 코드 패턴. 타이핑하며 자연스럽게 문법과 구조가 몸에 배입니다.

3. 문제 해결 능력: 에러가 나면 스스로 해결해야 합니다. 이 과정에서 디버깅 능력이 기릅니다.

4. 자신감: “내가 직접 만들었다”는 자부심. AI 도구 없이도 개발할 수 있다는 확신.

5. 기초 체력: AI 도구를 잘 쓰려면 먼저 기초가 탄탄해야 합니다. 손코딩은 그 기초를 다지는 과정입니다.

이 가이드의 접근법

느리지만 확실하게: 빠른 완성보다 깊은 이해를 우선합니다.

실수를 통한 학습: 에러는 적이 아니라 선생님입니다. 자주 하는 실수와 해결법을 함께 배웁니다.

점진적 구축: 한 번에 모든 것을 만들지 않습니다. Layer 1부터 Layer 7까지 단계적으로 쌓아갑니다.

실전 중심: 이론보다 실습. 코드를 직접 타이핑하며 배웁니다.

누구를 위한 가이드인가

초보 개발자: Python 기초만 알아도 됩니다. 손코딩하며 고급 패턴을 배웁니다.

AI 도구에 의존하던 개발자: 기초를 다시 다지고 싶은 분. 손코딩으로 원리를 이해하면 AI 도구도 더 잘 쓸 수 있습니다.

교육자: 학생들에게 정확한 코딩을 가르치고 싶은 강사. 단계별 커리큘럼으로 활용 가능합니다.

면접 준비자: 코딩 테스트, 기술 면접 대비. 손코딩 없이는 통과하기 어렵습니다.

학습 시간과 노력

총 예상 시간: 40-60시간 (1-2주, 하루 4-6시간 투자 시)

이는 Claude Code 사용 시(6-8시간)보다 5-10배 오래 걸립니다.

하지만:

Claude Code: 빠르지만 이해도 30%
손코딩: 느리지만 이해도 90%

한 달 후 결과:

Claude Code: 비슷한 프로젝트 빠르게 만들 수 있음 (단, 새로운 것은 어려움)
손코딩: 어떤 프로젝트든 스스로 설계하고 구현 가능

학습 후 Claude Code 사용

역설적이지만, 손코딩을 마스터한 후 Claude Code를 쓰면 10배 더 효과적입니다.

왜냐하면:

생성된 코드를 정확히 이해할 수 있음
문제가 있으면 즉시 수정 가능
구체적이고 정확한 프롬프트 작성 가능
아키텍처 결정은 스스로, 구현만 AI에게

이 가이드의 목표: AI 도구를 잘 쓰기 위한 기초 체력 기르기

이 가이드에 대하여

이 문서는 순수 손코딩으로 프로덕션급 AI 에이전트 시스템을 구축하는 완전한 기술 매뉴얼입니다. 코드를 복사-붙여넣기하지 않고, 한 줄 한 줄 직접 타이핑하며 배웁니다.

이 가이드가 제공하는 것

완전성: 환경 설정부터 프로덕션 배포까지 빠진 부분 없이 모든 단계를 다룹니다. Windows 환경에 특화되어 있으며, PowerShell 명령어를 중심으로 설명합니다.

깊이: 각 기술 스택의 선택 이유, 아키텍처 패턴의 의미, 보안 고려사항, 성능 최적화 전략까지 상세히 설명합니다. 표면적인 사용법을 넘어 본질적인 이해를 추구합니다.

실수와 해결: 초보자가 자주 하는 실수를 미리 알려주고, 해결 방법을 제시합니다. 에러 메시지를 보고 당황하지 않도록 돕습니다.

점진적 학습: 한 번에 모든 것을 가르치지 않습니다. Layer 1부터 시작해서 Layer 7까지 천천히, 확실하게 쌓아갑니다.

선행 요구사항

필수 지식:

Python 기초 (변수, 함수, 클래스 작성 가능)
웹 개발 기초 (HTTP, API 개념 이해)
Git 기본 (add, commit, push 가능)
영어 독해 (에러 메시지 읽기)

권장 지식:

비동기 프로그래밍 개념 (async/await)
데이터베이스 기초 (SQL 기본 문법)
Docker 개념 (컨테이너가 무엇인지)

필요 없는 것:

알고리즘/자료구조 전문 지식
DevOps 고급 기술
디자인 패턴 암기
클라우드 경험

타이핑 속도와 시간

타이핑 속도가 중요합니다:

분당 40타 이상: 권장
분당 60타 이상: 이상적
분당 30타 이하: 시간이 2배 걸릴 수 있음

실제 타이핑 분량:

최종 코드: 약 4,000줄
순수 타이핑 시간 (60타 기준): 약 15시간
생각하며 타이핑: 약 30시간
디버깅 + 테스트: 약 15시간
총 60시간

학습 전략

1. 완벽주의 버리기: 처음부터 완벽한 코드를 쓸 필요 없습니다. 작동하게 만든 후 개선하세요.

2. 에러를 두려워하지 말기: 에러는 배움의 기회입니다. 에러 메시지를 주의 깊게 읽으세요.

3. 주석 달며 타이핑: 코드만 타이핑하지 말고, 주석으로 “왜 이렇게 했는지” 메모하세요.

4. 자주 실행하기: 10줄 쓸 때마다 실행해서 작동하는지 확인하세요. 100줄 쓴 후 에러 찾기는 지옥입니다.

5. Git 자주 커밋: 작동하는 상태에서 커밋하세요. 망가뜨렸을 때 되돌릴 수 있습니다.

6. 쉬어가기: 2시간 타이핑하면 15분 쉬세요. 피곤하면 오타가 늘어납니다.

주의사항

복사-붙여넣기 금지: 이 가이드의 코드를 복사하지 마세요. 반드시 직접 타이핑하세요. 그래야 배웁니다.

순서 지키기: Layer 1부터 순서대로 진행하세요. 건너뛰면 나중에 문제가 생깁니다.

타이핑 실수 두려워 말기: 오타는 자연스러운 것입니다. IDE가 대부분 잡아줍니다.

느려도 괜찮습니다: 남들보다 2배 걸려도 됩니다. 중요한 건 완주입니다.

1. 환경 설정

Windows 노트북에서 Claude Code를 사용하여 AI 에이전트를 개발하려면 몇 가지 필수 소프트웨어가 필요합니다. 각 도구가 왜 필요한지, 어떻게 설치하는지, 주의사항은 무엇인지 상세히 설명합니다.

1.1 Python 3.13 설치

Python은 우리 프로젝트의 핵심 언어입니다. 3.13 버전을 사용하는 이유는 최신 타입 힌트 기능, 성능 개선, 그리고 최신 라이브러리 호환성 때문입니다.

설치 과정

Python.org 공식 웹사이트(https://www.python.org/downloads/)에서 Windows 설치 프로그램을 다운로드합니다. 설치 시 반드시 “Add Python to PATH” 옵션을 체크해야 합니다. 이 옵션을 놓치면 PowerShell에서 python 명령어를 인식하지 못합니다.

  
# 설치 후 버전 확인
python --version
# 출력: Python 3.13.x

pip --version
# 출력: pip 24.x from ...

PATH 설정 확인

만약 python 명령어가 인식되지 않는다면, 환경 변수를 수동으로 설정해야 합니다.

  
# 현재 PATH 확인
$env:PATH

# Python 설치 경로 확인 (보통 C:\Users\사용자명\AppData\Local\Programs\Python\Python313)
# 시스템 환경 변수에 추가:
# - C:\Users\사용자명\AppData\Local\Programs\Python\Python313
# - C:\Users\사용자명\AppData\Local\Programs\Python\Python313\Scripts

가상환경 도구 설치

전역 Python 환경을 오염시키지 않기 위해 venv를 사용합니다. Python 3.13에는 기본 포함되어 있습니다.

  
# venv 모듈 확인
python -m venv --help

1.2 Git 설치와 설정

Git은 버전 관리 시스템입니다. Claude Code가 생성한 수백, 수천 줄의 코드를 안전하게 관리하려면 필수입니다.

설치

Git for Windows(https://git-scm.com/download/win)를 다운로드하여 설치합니다. 설치 옵션은 대부분 기본값으로 진행하되, 다음 항목만 확인합니다:

에디터: VSCode 선택 (기본 Vim 대신)
PATH 환경: Git from the command line and also from 3rd-party software
줄바꿈 변환: Checkout Windows-style, commit Unix-style

  
# 설치 확인
git --version
# 출력: git version 2.43.0.windows.1

초기 설정

Git을 처음 사용한다면 사용자 정보를 설정해야 합니다.

  
# 사용자 이름 설정 (커밋에 표시됨)
git config --global user.name "Your Name"

# 이메일 설정
git config --global user.email "your.email@example.com"

# 기본 브랜치 이름을 main으로 설정
git config --global init.defaultBranch main

# 설정 확인
git config --list

.gitignore 기본 설정

Python 프로젝트에서 제외할 파일들을 미리 정의합니다.

# Python
__pycache__/
*.py[cod]
*$py.class
.Python
venv/
ENV/

# 환경 변수 (중요!)
.env
.env.local
.env.production

# IDE
.vscode/
.idea/

# OS
.DS_Store
Thumbs.db

# 로그
*.log
logs/

1.3 Docker Desktop 설치

Docker는 컨테이너 기술입니다. PostgreSQL, Redis, Prometheus 같은 서비스를 쉽게 실행하고 관리할 수 있습니다.

WSL 2 활성화 (Windows 10/11)

Docker Desktop은 WSL 2가 필요합니다.

  
# PowerShell을 관리자 권한으로 실행

# WSL 설치
wsl --install

# 컴퓨터 재시작 후 WSL 버전 확인
wsl --status
wsl --list --verbose

Docker Desktop 설치

Docker Desktop for Windows(https://www.docker.com/products/docker-desktop)를 다운로드하여 설치합니다.

설치 후 Docker Desktop을 실행하고, 설정에서 다음을 확인합니다:

Resources → WSL Integration: WSL 2 배포판과 통합 활성화
Resources → Advanced: 메모리를 최소 4GB 이상 할당

  
# Docker 설치 확인
docker --version
# 출력: Docker version 24.0.x

docker-compose --version
# 출력: Docker Compose version v2.x.x

# 테스트 컨테이너 실행
docker run hello-world
# "Hello from Docker!" 메시지 출력되면 성공

Docker 네트워크 이해

우리 프로젝트는 여러 컨테이너가 통신해야 합니다:

PostgreSQL 컨테이너 (데이터베이스)
FastAPI 컨테이너 (애플리케이션)
Prometheus 컨테이너 (모니터링)
Grafana 컨테이너 (시각화)

Docker Compose는 이들을 하나의 네트워크로 연결하여 서로 통신할 수 있게 합니다.

1.4 VSCode 및 필수 확장 프로그램

VSCode는 Claude Code가 작동하는 환경입니다.

VSCode 설치

Visual Studio Code(https://code.visualstudio.com/)를 다운로드하여 설치합니다.

필수 확장 프로그램

다음 확장을 설치합니다 (Ctrl+Shift+X로 확장 마켓플레이스 열기):

1. Python (Microsoft)

IntelliSense, 린팅, 디버깅 지원
설치 후 Python 인터프리터 선택 (Ctrl+Shift+P → “Python: Select Interpreter”)

2. Claude Code (Anthropic)

바이브 코딩의 핵심 도구
자연어로 코드 생성, 수정, 리팩토링

3. Docker (Microsoft)

docker-compose.yml 편집 지원
컨테이너 관리 UI

4. GitLens (Eric Amodio)

Git 히스토리 시각화
코드 변경 추적

5. Error Lens (Alexander)

에러 메시지를 코드 라인에 직접 표시

6. Markdown All in One

README 및 문서 작성 지원

VSCode 설정 최적화

settings.json에 다음을 추가합니다 (Ctrl+,로 설정 열기 → 우측 상단 아이콘 클릭):

  
{
  "python.linting.enabled": true,
  "python.linting.pylintEnabled": false,
  "python.linting.flake8Enabled": true,
  "python.formatting.provider": "black",
  "editor.formatOnSave": true,
  "editor.codeActionsOnSave": {
    "source.organizeImports": true
  },
  "files.autoSave": "afterDelay",
  "files.autoSaveDelay": 1000
}

1.5 Claude Code API 키 설정

Claude Code를 사용하려면 Anthropic API 키가 필요합니다.

API 키 발급

https://console.anthropic.com/ 접속
계정 생성 또는 로그인
API Keys 메뉴에서 새 키 생성
키를 안전한 곳에 복사 (한 번만 표시됨)

VSCode에서 설정

Ctrl+Shift+P → "Claude Code: Set API Key"
복사한 API 키 붙여넣기

요금제 이해

Claude API는 사용량 기반 과금입니다:

Claude Sonnet 4.5: 입력 $3/1M 토큰, 출력 $15/1M 토큰
Claude Haiku 4.5: 입력 $1/1M 토큰, 출력 $5/1M 토큰

일반적인 프로젝트 개발 시 약 $10-20 정도 소요됩니다.

테스트

VSCode에서 새 파일을 열고:

Ctrl+Shift+P → "Claude Code: New Chat"
프롬프트: "Hello, Claude Code!"

응답이 오면 설정 완료입니다.

2. 기술 스택 이해

프로덕션급 AI 에이전트를 만들려면 여러 기술이 조합됩니다. 각 기술을 왜 선택했는지, 어떤 역할을 하는지 이해하면 나중에 커스터마이징하거나 문제를 해결할 때 도움이 됩니다.

2.1 FastAPI - 웹 프레임워크

FastAPI는 Python 웹 프레임워크입니다. Flask나 Django 대신 FastAPI를 선택한 이유가 있습니다.

FastAPI의 장점

1. 비동기 지원: AI 에이전트는 LLM API 호출에 시간이 걸립니다. 동기 방식이면 한 요청이 처리되는 동안 다른 요청이 대기해야 하지만, async/await를 사용하면 동시에 여러 요청을 처리할 수 있습니다.

  
# 동기 방식 (느림)
def chat(message):
    response = llm_service.call(message)  # 3초 걸림
    return response

# 비동기 방식 (빠름)
async def chat(message):
    response = await llm_service.call(message)  # 다른 요청 처리 가능
    return response

2. 자동 문서 생성: FastAPI는 코드에서 자동으로 OpenAPI 스펙을 생성합니다. /docs에 접속하면 Swagger UI로 모든 엔드포인트를 테스트할 수 있습니다.

3. 타입 힌트 기반 검증: Pydantic을 내장하여 요청/응답 데이터를 자동으로 검증합니다.

  
from pydantic import BaseModel

class ChatRequest(BaseModel):
    message: str
    session_id: str

async def chat(request: ChatRequest):
    # request.message가 str임이 보장됨
    # 잘못된 타입이 오면 자동으로 422 에러 반환
    pass

4. 성능: Starlette 기반으로 매우 빠릅니다. Node.js나 Go와 비슷한 수준의 성능을 냅니다.

대안과의 비교

Flask: 간단하지만 비동기 지원이 약하고, 타입 검증을 수동으로 해야 합니다.

Django: 풀스택 프레임워크로 너무 무겁습니다. ORM, Admin, 템플릿 엔진 등이 포함되어 있지만 API 서버에는 과합니다.

FastAPI: API 서버에 최적화되어 있고, 비동기, 타입 안정성, 자동 문서화를 모두 제공합니다.

2.2 LangGraph - 멀티 에이전트 프레임워크

LangGraph는 복잡한 AI 워크플로우를 설계하는 도구입니다.

왜 직접 LLM을 호출하지 않는가?

간단한 채팅봇은 이렇게 만들 수 있습니다:

  
def simple_chat(message):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": message}]
    )
    return response.choices[0].message.content

하지만 프로덕션급 AI 에이전트는 이것만으로 부족합니다:

도구 사용: 웹 검색, 계산기, 데이터베이스 쿼리 등
멀티 스텝: “먼저 검색하고, 그 결과를 분석하고, 최종 답변 생성”
상태 관리: 대화 히스토리, 사용자 메모리
에러 처리: LLM이 잘못된 답을 하면 재시도
병렬 처리: 여러 도구를 동시에 호출

LangGraph는 이런 복잡한 로직을 그래프 구조로 표현합니다.

StateGraph의 개념

StateGraph는 노드(작업)와 엣지(연결)로 구성됩니다.

  
from langgraph.graph import StateGraph

graph = StateGraph(state_schema)

# 노드 정의
graph.add_node("chat", chat_node)  # LLM 호출
graph.add_node("search", search_node)  # 웹 검색
graph.add_node("calculate", calculate_node)  # 계산

# 조건부 라우팅
graph.add_conditional_edges(
    "chat",
    should_use_tool,  # 도구 사용 여부 판단
    {
        "search": "search",
        "calculate": "calculate",
        "end": END
    }
)

# 진입점
graph.set_entry_point("chat")

이렇게 정의하면:

chat 노드에서 시작
LLM이 도구 사용을 결정하면 해당 노드로 이동
도구 실행 후 다시 chat으로 돌아와 최종 답변
끝

Checkpoint의 중요성

LangGraph는 각 단계의 상태를 저장할 수 있습니다. 대화 중간에 서버가 재시작되어도 이어서 계속할 수 있습니다.

  
from langgraph.checkpoint.postgres import AsyncPostgresSaver

checkpointer = AsyncPostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)

graph.compile(checkpointer=checkpointer)

2.3 Mem0 - 장기 기억 시스템

일반 LLM은 대화 히스토리만 기억합니다. 하지만 진정한 AI 에이전트는 장기 기억이 필요합니다.

대화 히스토리 vs 장기 기억

대화 히스토리: 현재 대화에서 주고받은 메시지들

User: 내 이름은 김철수야
AI: 안녕하세요 김철수님!
User: 내 이름이 뭐였지?
AI: 김철수님이세요. (히스토리에서 찾음)

장기 기억: 대화를 넘어선 중요한 정보

[대화 1 - 1주일 전]
User: 나는 채식주의자야

[대화 2 - 오늘]
User: 저녁 메뉴 추천해줘
AI: 채식 요리를 추천드리겠습니다. (장기 기억에서 가져옴)

Mem0의 작동 원리

Mem0는 대화를 분석하여 중요한 정보를 추출하고 벡터 데이터베이스에 저장합니다.

  
from mem0 import AsyncMemory

memory = AsyncMemory()

# 대화 후 자동으로 메모리 업데이트
await memory.add(
    "User likes vegetarian food",
    user_id="user-123"
)

# 나중에 검색
results = await memory.search(
    "food preference",
    user_id="user-123"
)
# 결과: ["User likes vegetarian food"]

벡터 검색의 마법

일반 데이터베이스는 정확한 일치만 찾습니다:

  
SELECT * FROM memories WHERE content = 'vegetarian food'
-- "vegetarian food"만 찾음, "채식"은 못 찾음

벡터 데이터베이스는 의미적 유사성을 찾습니다:

  
results = memory.search("what food does user like")
# "vegetarian food", "채식", "vegan" 모두 찾음

텍스트를 벡터(숫자 배열)로 변환하여 거리 계산으로 유사도를 측정합니다.

2.4 PostgreSQL + pgvector - 데이터베이스

관계형 데이터베이스와 벡터 검색을 동시에 지원하는 PostgreSQL을 사용합니다.

왜 PostgreSQL인가?

1. 안정성: 30년 역사의 검증된 RDBMS

2. ACID 보장: 트랜잭션이 안전하게 처리됨

3. pgvector 확장: 벡터 검색 기능 추가 가능

4. JSON 지원: JSONB 타입으로 유연한 스키마

pgvector의 역할

pgvector는 PostgreSQL 확장으로, 벡터 데이터를 저장하고 유사도 검색을 지원합니다.

  
-- pgvector 설치
CREATE EXTENSION IF NOT EXISTS vector;

-- 벡터 컬럼이 있는 테이블
CREATE TABLE memories (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536)  -- 1536차원 벡터
);

-- 유사도 검색
SELECT content
FROM memories
ORDER BY embedding <-> query_vector  -- 코사인 유사도
LIMIT 5;

대안과의 비교

MongoDB: NoSQL로 유연하지만 트랜잭션 지원이 약함

MySQL: 벡터 검색 지원 없음

Pinecone/Weaviate: 전용 벡터 DB지만 별도 인프라 필요

PostgreSQL + pgvector: 관계형 데이터와 벡터를 하나의 DB에서 관리

2.5 Docker Compose - 인프라 오케스트레이션

여러 서비스를 쉽게 관리하기 위해 Docker Compose를 사용합니다.

수동 설치의 문제점

PostgreSQL, Redis, Prometheus를 수동으로 설치하면:

Windows에서 각각 설치 프로그램 실행
포트 충돌 해결
서비스 간 연결 설정
재시작 시 수동으로 서비스 시작

Docker Compose의 이점

docker-compose.yml 파일 하나로 모든 서비스를 정의:

  
version: '3.8'

services:
  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: secret
    volumes:
      - db-data:/var/lib/postgresql/data
    
  app:
    build: .
    depends_on:
      - db
    environment:
      DATABASE_URL: postgresql://user:secret@db/mydb

  
# 모든 서비스 시작
docker-compose up -d

# 모든 서비스 종료
docker-compose down

# 로그 확인
docker-compose logs -f

3. 프로젝트 구조 설계

좋은 프로젝트 구조는 유지보수를 쉽게 만듭니다. 우리는 7계층 아키텍처를 사용합니다.

3.1 디렉토리 구조

production-ai-agent/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 앱
│   ├── api/
│   │   └── v1/
│   │       ├── auth.py         # 인증 API
│   │       └── chatbot.py      # 채팅 API
│   ├── core/
│   │   ├── config.py           # 설정 관리
│   │   ├── security/
│   │   │   ├── auth.py         # JWT
│   │   │   ├── limiter.py      # Rate limiting
│   │   │   └── sanitization.py # 입력 검증
│   │   ├── services/
│   │   │   ├── database.py     # DB CRUD
│   │   │   └── llm.py          # LLM 서비스
│   │   ├── prompts/
│   │   │   ├── system.md       # 시스템 프롬프트
│   │   │   └── __init__.py
│   │   └── langgraph/
│   │       ├── graph.py        # LangGraph 에이전트
│   │       └── tools/
│   │           └── search.py   # 검색 도구
│   ├── models/
│   │   └── database.py         # SQLModel 모델
│   └── schemas/
│       └── auth.py             # Pydantic 스키마
├── database/
│   └── schema.sql              # DB 스키마
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── tests/
│   ├── test_auth.py
│   └── test_chatbot.py
├── scripts/
│   └── init_db.py
├── .env.example
├── .gitignore
├── pyproject.toml
└── README.md

3.2 7계층 아키텍처

Layer 1: 설정 및 환경 변수

pyproject.toml: 의존성 관리
.env: 환경 변수
config.py: Pydantic Settings

Layer 2: 데이터 모델

SQLModel 모델 (User, Session, Message)
Pydantic 스키마 (요청/응답)

Layer 3: 보안

JWT 인증
Rate limiting
Input sanitization

Layer 4: 서비스 계층

DatabaseService: CRUD 작업
LLMService: LLM 호출 및 폴백

Layer 5: AI 에이전트

LangGraph 그래프
Mem0 통합
도구 (검색, 계산 등)

Layer 6: API

FastAPI 엔드포인트
미들웨어
에러 핸들링

Layer 7: 배포

Docker 컨테이너
모니터링 (Prometheus, Grafana)
로깅

3.3 의존성 관리

pyproject.toml을 사용하는 이유:

1. 표준화: PEP 518, 621에 따른 Python 프로젝트 표준

2. 통합: 의존성, 빌드, 린팅, 테스트 설정을 한 곳에

3. 도구 호환: Poetry, PDM, Hatch 모두 지원

  
[project]
name = "production-ai-agent"
version = "1.0.0"
requires-python = ">=3.13"
dependencies = [
    "fastapi[all]>=0.104.0",
    "langgraph>=1.0.5",
    "mem0ai>=1.0.0",
    "sqlmodel>=0.0.14",
    # ... 총 25개
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.3",
    "black>=23.11.0",
    "ruff>=0.1.6",
]

4. Layer 1-2: 기반 설정

프로젝트의 기초를 다지는 단계입니다.

4.1 프로젝트 초기화

Claude Code로 프로젝트를 생성합니다.

Ctrl+Shift+P → "Claude Code: New Chat"

프롬프트:
"새로운 Python 프로젝트 'production-ai-agent'를 만들어줘.

디렉토리 구조:
app/
  ├── __init__.py
  ├── main.py
  ├── api/v1/
  ├── core/
  ├── models/
  └── schemas/
database/
tests/
scripts/

pyproject.toml with:
- FastAPI 0.104+
- LangGraph 1.0.5
- Mem0 1.0.0
- SQLModel 0.0.14
- PostgreSQL driver (psycopg[binary])
- 총 25개 의존성

.gitignore for Python project
README.md with basic info"

Claude가 파일을 생성하면 확인합니다:

  
cd production-ai-agent
tree /F  # 디렉토리 구조 확인

4.2 환경 변수 설정

.env.example을 작성하여 필요한 환경 변수를 문서화합니다.

# Application
APP_NAME=Production AI Agent
DEBUG=False
LOG_LEVEL=INFO

# Database
POSTGRES_USER=aiagent
POSTGRES_PASSWORD=change-this-password
POSTGRES_DB=aiagent_db
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
DATABASE_URL=postgresql+psycopg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}

# LLM
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...
DEFAULT_MODEL=claude-sonnet-4-5-20250929
FALLBACK_MODEL=claude-haiku-4-5-20251001

# Security
JWT_SECRET_KEY=generate-a-strong-random-key
JWT_ALGORITHM=HS256
JWT_ACCESS_TOKEN_EXPIRE_DAYS=7

# Rate Limiting
RATE_LIMIT_PER_MINUTE=10

# Mem0
MEM0_POSTGRES_HOST=localhost
MEM0_POSTGRES_PORT=5432

실제 사용을 위한 .env 파일:

  
# .env.example을 복사
cp .env.example .env

# 실제 값으로 수정
notepad .env

중요: .env 파일은 절대 Git에 커밋하면 안 됩니다!

4.3 설정 관리 (config.py)

Pydantic Settings로 타입 안전한 설정 관리:

  
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, SecretStr
from enum import Enum
from pathlib import Path

class Environment(str, Enum):
    DEVELOPMENT = "development"
    PRODUCTION = "production"
    TEST = "test"

class Settings(BaseSettings):
    """애플리케이션 설정"""
    
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )
    
    # Application
    APP_NAME: str = "Production AI Agent"
    DEBUG: bool = False
    ENVIRONMENT: Environment = Environment.DEVELOPMENT
    
    # Database
    POSTGRES_USER: str
    POSTGRES_PASSWORD: SecretStr  # 자동으로 마스킹
    POSTGRES_DB: str
    POSTGRES_HOST: str = "localhost"
    POSTGRES_PORT: int = 5432
    
    @property
    def database_url(self) -> str:
        """데이터베이스 연결 문자열"""
        return (
            f"postgresql+psycopg://{self.POSTGRES_USER}:"
            f"{self.POSTGRES_PASSWORD.get_secret_value()}@"
            f"{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/"
            f"{self.POSTGRES_DB}"
        )
    
    # LLM
    ANTHROPIC_API_KEY: SecretStr
    OPENAI_API_KEY: SecretStr | None = None
    DEFAULT_MODEL: str = "claude-sonnet-4-5-20250929"
    FALLBACK_MODEL: str = "claude-haiku-4-5-20251001"
    
    # Security
    JWT_SECRET_KEY: SecretStr
    JWT_ALGORITHM: str = "HS256"
    JWT_ACCESS_TOKEN_EXPIRE_DAYS: int = 7
    
    # Rate Limiting
    RATE_LIMIT_PER_MINUTE: int = 10
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 프로덕션 환경에서 DEBUG=True는 위험
        if self.ENVIRONMENT == Environment.PRODUCTION and self.DEBUG:
            raise ValueError("Cannot enable DEBUG in production")

# 싱글톤 인스턴스
settings = Settings()

이렇게 하면:

환경 변수가 자동으로 로드됨
타입 검증 자동 수행
비밀번호 등은 SecretStr로 보호
IDE에서 자동완성 지원

4.4 데이터베이스 스키마 설계

database/schema.sql을 작성합니다.

  
-- PostgreSQL 16 + pgvector

-- 확장 설치
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- 사용자 테이블
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- 인덱스
CREATE INDEX idx_users_email ON users(email);

-- 세션 테이블 (대화방)
CREATE TABLE sessions (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    name VARCHAR(255),
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE INDEX idx_sessions_user_id ON sessions(user_id);
CREATE INDEX idx_sessions_created_at ON sessions(created_at DESC);

-- 메시지 테이블
CREATE TABLE messages (
    id SERIAL PRIMARY KEY,
    session_id UUID NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
    role VARCHAR(50) NOT NULL CHECK (role IN ('user', 'assistant', 'system')),
    content TEXT NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE INDEX idx_messages_session_id ON messages(session_id);
CREATE INDEX idx_messages_session_created ON messages(session_id, created_at);

-- 메모리 벡터 테이블 (Mem0용)
CREATE TABLE memory_vectors (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    content TEXT NOT NULL,
    embedding vector(1536),  -- OpenAI embedding 차원
    metadata JSONB,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- 벡터 검색 인덱스 (HNSW - Hierarchical Navigable Small World)
CREATE INDEX idx_memory_vectors_embedding ON memory_vectors 
USING hnsw (embedding vector_cosine_ops);

CREATE INDEX idx_memory_vectors_user_id ON memory_vectors(user_id);

-- 샘플 데이터 (개발용)
INSERT INTO users (email, password_hash) VALUES
('test@example.com', '$2b$12$dummy_hash_for_testing');

INSERT INTO sessions (user_id, name) VALUES
(1, '첫 대화');

INSERT INTO messages (session_id, role, content) VALUES
((SELECT id FROM sessions WHERE name = '첫 대화'), 'user', '안녕하세요'),
((SELECT id FROM sessions WHERE name = '첫 대화'), 'assistant', '안녕하세요! 무엇을 도와드릴까요?');

설계 포인트:

CASCADE DELETE: 사용자 삭제 시 관련 세션과 메시지 자동 삭제
인덱스: 자주 조회하는 컬럼에 인덱스 추가
JSONB: 메타데이터는 유연한 스키마
벡터 인덱스: HNSW 알고리즘으로 빠른 유사도 검색

4.5 데이터 모델 (SQLModel)

app/models/database.py에 Python 모델을 정의합니다.

  
from datetime import UTC, datetime
from typing import Optional
from sqlmodel import Field, Relationship, SQLModel
from uuid import UUID, uuid4

class BaseModel(SQLModel):
    """모든 모델의 기본 클래스"""
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        nullable=False
    )
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        nullable=False,
        sa_column_kwargs={"onupdate": lambda: datetime.now(UTC)}
    )

class User(BaseModel, table=True):
    """사용자 모델"""
    __tablename__ = "users"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, max_length=255)
    password_hash: str = Field(max_length=255)
    
    # 관계
    sessions: list["Session"] = Relationship(
        back_populates="user",
        cascade_delete=True
    )
    
    def verify_password(self, password: str) -> bool:
        """비밀번호 검증"""
        from bcrypt import checkpw
        return checkpw(
            password.encode('utf-8'),
            self.password_hash.encode('utf-8')
        )
    
    @staticmethod
    def hash_password(password: str) -> str:
        """비밀번호 해싱"""
        from bcrypt import hashpw, gensalt
        return hashpw(
            password.encode('utf-8'),
            gensalt()
        ).decode('utf-8')

class Session(BaseModel, table=True):
    """대화 세션 모델"""
    __tablename__ = "sessions"
    
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    user_id: int = Field(foreign_key="users.id")
    name: Optional[str] = Field(default=None, max_length=255)
    
    # 관계
    user: User = Relationship(back_populates="sessions")
    messages: list["Message"] = Relationship(
        back_populates="session",
        cascade_delete=True
    )

class Message(SQLModel, table=True):
    """메시지 모델"""
    __tablename__ = "messages"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    session_id: UUID = Field(foreign_key="sessions.id")
    role: str = Field(max_length=50)  # user, assistant, system
    content: str
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC)
    )
    
    # 관계
    session: Session = Relationship(back_populates="messages")

설계 포인트:

BaseModel: 중복 코드 제거 (created_at, updated_at)
Relationship: SQLModel이 자동으로 JOIN 처리
타입 힌트: IDE 자동완성과 타입 검사
비밀번호 메서드: User 모델에 캡슐화

5. Layer 3-4: 데이터와 서비스

보안과 비즈니스 로직을 구현하는 단계입니다.

5.1 JWT 인증

app/core/security/auth.py에 JWT 기능을 구현합니다.

  
from datetime import UTC, datetime, timedelta
from jose import JWTError, jwt
from app.core.config import settings
from app.schemas.auth import Token

def create_access_token(subject: str) -> Token:
    """
    JWT 액세스 토큰 생성
    
    Args:
        subject: 토큰의 subject (보통 user email)
    
    Returns:
        Token 객체 (access_token, expires_at)
    """
    expire = datetime.now(UTC) + timedelta(
        days=settings.JWT_ACCESS_TOKEN_EXPIRE_DAYS
    )
    
    to_encode = {
        "sub": subject,  # Subject (사용자 식별자)
        "exp": expire,   # Expiration time
        "iat": datetime.now(UTC),  # Issued at
        "jti": f"{subject}-{datetime.now(UTC).timestamp()}"  # JWT ID
    }
    
    encoded_jwt = jwt.encode(
        to_encode,
        settings.JWT_SECRET_KEY.get_secret_value(),
        algorithm=settings.JWT_ALGORITHM
    )
    
    return Token(access_token=encoded_jwt, expires_at=expire)

def verify_token(token: str) -> str:
    """
    JWT 토큰 검증
    
    Args:
        token: JWT 토큰 문자열
    
    Returns:
        subject (user email)
    
    Raises:
        JWTError: 토큰이 유효하지 않을 때
    """
    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET_KEY.get_secret_value(),
            algorithms=[settings.JWT_ALGORITHM]
        )
        subject: str = payload.get("sub")
        if subject is None:
            raise JWTError("Token has no subject")
        return subject
    except JWTError:
        raise

JWT vs Session:

Session 방식:

서버에 세션 저장 (메모리 또는 Redis)
Stateful (서버가 상태 관리)
확장 어려움 (세션 공유 필요)

JWT 방식:

토큰에 모든 정보 포함
Stateless (서버가 상태 관리 안 함)
확장 쉬움 (어느 서버에서나 검증 가능)

5.2 Rate Limiting

app/core/security/limiter.py에 요청 속도 제한을 구현합니다.

  
from slowapi import Limiter
from slowapi.util import get_remote_address
from app.core.config import settings

# SlowAPI는 FastAPI용 rate limiter
limiter = Limiter(
    key_func=get_remote_address,  # IP 기반 제한
    default_limits=[f"{settings.RATE_LIMIT_PER_MINUTE}/minute"]
)

# 사용 예시:
# @app.get("/api/endpoint")
# @limiter.limit("5/minute")  # 엔드포인트별 제한
# async def endpoint():
#     pass

왜 Rate Limiting이 필요한가:

DDoS 방어: 무한 요청 차단
비용 절감: LLM API 호출 비용 제한
공정 사용: 한 사용자가 자원 독점 방지

5.3 Input Sanitization

app/core/security/sanitization.py에 입력 검증을 구현합니다.

  
import html
import re

def sanitize_html(text: str) -> str:
    """HTML 이스케이프"""
    return html.escape(text)

def remove_script_tags(text: str) -> str:
    """<script> 태그 제거"""
    return re.sub(r'<script[^>]*>.*?</script>', '', text, flags=re.DOTALL | re.IGNORECASE)

def sanitize_input(text: str, max_length: int = 10000) -> str:
    """
    사용자 입력 정제
    
    1. 길이 제한
    2. <script> 제거
    3. HTML 이스케이프
    4. NULL 바이트 제거
    """
    if len(text) > max_length:
        text = text[:max_length]
    
    text = remove_script_tags(text)
    text = sanitize_html(text)
    text = text.replace('\x00', '')  # NULL 바이트 제거
    
    return text.strip()

보안 위협:

XSS (Cross-Site Scripting): <script>alert('hack')</script> 같은 악성 코드
SQL Injection: 직접 SQL 작성 시 위험 (SQLModel은 자동 방어)
NULL Byte Injection: \x00으로 문자열 조작

5.4 Database Service

app/core/services/database.py에 CRUD 작업을 캡슐화합니다.

  
from typing import Optional
from uuid import UUID
from sqlmodel import Session, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
from app.models.database import User, Session as ChatSession, Message

class DatabaseService:
    """데이터베이스 작업을 관리하는 싱글톤 서비스"""
    
    _instance: Optional["DatabaseService"] = None
    _engine: Optional[AsyncEngine] = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self):
        if self._engine is None:
            self._engine = create_async_engine(
                settings.database_url.replace('postgresql+psycopg', 'postgresql+asyncpg'),
                echo=settings.DEBUG,
                pool_size=20,  # 연결 풀 크기
                max_overflow=10
            )
    
    async def get_session(self) -> AsyncSession:
        """비동기 데이터베이스 세션 생성"""
        async_session = sessionmaker(
            self._engine,
            class_=AsyncSession,
            expire_on_commit=False
        )
        async with async_session() as session:
            yield session
    
    async def create_user(self, email: str, password: str) -> User:
        """사용자 생성"""
        user = User(
            email=email,
            password_hash=User.hash_password(password)
        )
        
        async with self.get_session() as session:
            session.add(user)
            await session.commit()
            await session.refresh(user)
            return user
    
    async def get_user_by_email(self, email: str) -> Optional[User]:
        """이메일로 사용자 조회"""
        async with self.get_session() as session:
            result = await session.execute(
                select(User).where(User.email == email)
            )
            return result.scalar_one_or_none()
    
    async def create_session(self, user_id: int, name: Optional[str] = None) -> ChatSession:
        """대화 세션 생성"""
        chat_session = ChatSession(user_id=user_id, name=name)
        
        async with self.get_session() as session:
            session.add(chat_session)
            await session.commit()
            await session.refresh(chat_session)
            return chat_session
    
    async def get_user_sessions(self, user_id: int, limit: int = 50) -> list[ChatSession]:
        """사용자의 대화 세션 목록"""
        async with self.get_session() as session:
            result = await session.execute(
                select(ChatSession)
                .where(ChatSession.user_id == user_id)
                .order_by(ChatSession.updated_at.desc())
                .limit(limit)
            )
            return result.scalars().all()
    
    async def create_message(
        self,
        session_id: UUID,
        role: str,
        content: str
    ) -> Message:
        """메시지 생성"""
        message = Message(
            session_id=session_id,
            role=role,
            content=content
        )
        
        async with self.get_session() as session:
            session.add(message)
            await session.commit()
            await session.refresh(message)
            return message
    
    async def get_session_messages(
        self,
        session_id: UUID,
        limit: int = 50
    ) -> list[Message]:
        """세션의 메시지 히스토리"""
        async with self.get_session() as session:
            result = await session.execute(
                select(Message)
                .where(Message.session_id == session_id)
                .order_by(Message.created_at)
                .limit(limit)
            )
            return result.scalars().all()

# 싱글톤 인스턴스
db_service = DatabaseService()

설계 패턴:

싱글톤: 데이터베이스 연결은 하나만
의존성 주입: FastAPI의 Depends로 세션 관리
비동기: 모든 DB 작업은 async
연결 풀: pool_size로 동시 연결 수 관리

5.5 LLM Service (폴백 전략 포함)

app/core/services/llm.py에 LLM 호출 로직을 구현합니다.

  
from typing import Any, Optional
from anthropic import AsyncAnthropic
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
from app.core.config import settings

class LLMRegistry:
    """사용 가능한 LLM 모델 레지스트리"""
    
    MODELS = {
        "claude-sonnet-4-5-20250929": {
            "provider": "anthropic",
            "context_window": 200000,
            "cost_per_1m_input": 3.0,
            "cost_per_1m_output": 15.0,
        },
        "claude-haiku-4-5-20251001": {
            "provider": "anthropic",
            "context_window": 200000,
            "cost_per_1m_input": 1.0,
            "cost_per_1m_output": 5.0,
        },
        "gpt-4o": {
            "provider": "openai",
            "context_window": 128000,
            "cost_per_1m_input": 2.5,
            "cost_per_1m_output": 10.0,
        },
    }
    
    @classmethod
    def get_model_info(cls, model_name: str) -> dict:
        """모델 정보 조회"""
        return cls.MODELS.get(model_name, {})
    
    @classmethod
    def get_fallback_model(cls, current_model: str) -> str:
        """폴백 모델 반환"""
        # Sonnet → Haiku (같은 provider)
        if current_model.startswith("claude-sonnet"):
            return "claude-haiku-4-5-20251001"
        # GPT-4 → GPT-3.5
        elif current_model.startswith("gpt-4"):
            return "gpt-3.5-turbo"
        return current_model


class LLMService:
    """LLM 호출을 관리하는 서비스"""
    
    def __init__(self):
        self.anthropic_client = AsyncAnthropic(
            api_key=settings.ANTHROPIC_API_KEY.get_secret_value()
        )
        
        if settings.OPENAI_API_KEY:
            self.openai_client = AsyncOpenAI(
                api_key=settings.OPENAI_API_KEY.get_secret_value()
            )
        else:
            self.openai_client = None
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def call(
        self,
        messages: list[dict],
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 4000,
        **kwargs
    ) -> str:
        """
        LLM 호출 (자동 폴백 포함)
        
        Args:
            messages: 대화 메시지 리스트
            model: 사용할 모델 (None이면 DEFAULT_MODEL)
            temperature: 창의성 (0-1)
            max_tokens: 최대 토큰 수
        
        Returns:
            LLM 응답 텍스트
        
        Raises:
            Exception: 모든 시도 실패 시
        """
        model = model or settings.DEFAULT_MODEL
        model_info = LLMRegistry.get_model_info(model)
        
        try:
            if model_info.get("provider") == "anthropic":
                return await self._call_anthropic(
                    messages, model, temperature, max_tokens, **kwargs
                )
            elif model_info.get("provider") == "openai":
                return await self._call_openai(
                    messages, model, temperature, max_tokens, **kwargs
                )
            else:
                raise ValueError(f"Unknown model: {model}")
        
        except Exception as e:
            # 폴백 시도
            fallback_model = LLMRegistry.get_fallback_model(model)
            if fallback_model != model:
                print(f"Falling back from {model} to {fallback_model}")
                return await self.call(
                    messages, fallback_model, temperature, max_tokens, **kwargs
                )
            raise
    
    async def _call_anthropic(
        self,
        messages: list[dict],
        model: str,
        temperature: float,
        max_tokens: int,
        **kwargs
    ) -> str:
        """Anthropic Claude 호출"""
        response = await self.anthropic_client.messages.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        )
        return response.content[0].text
    
    async def _call_openai(
        self,
        messages: list[dict],
        model: str,
        temperature: float,
        max_tokens: int,
        **kwargs
    ) -> str:
        """OpenAI GPT 호출"""
        if not self.openai_client:
            raise ValueError("OpenAI API key not configured")
        
        response = await self.openai_client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        )
        return response.choices[0].message.content
    
    async def bind_tools(self, tools: list[dict]) -> dict:
        """도구를 LLM에 바인딩"""
        # LangGraph에서 사용
        return {
            "tools": tools,
            "tool_choice": "auto"
        }

# 싱글톤
llm_service = LLMService()

폴백 전략의 중요성:

비용 절감: Sonnet 실패 시 Haiku로 폴백 (저렴함)
가용성: API 장애 시에도 서비스 유지
성능: Rate limit 초과 시 다른 모델 사용

Tenacity 재시도 로직:

3회까지 재시도
지수 백오프: 2초 → 4초 → 8초 대기
일시적 네트워크 오류 극복

6. Layer 5-6: AI 에이전트

프로젝트의 핵심인 AI 에이전트를 구현합니다.

6.1 시스템 프롬프트 설계

app/core/prompts/system.md에 AI 에이전트의 성격과 행동을 정의합니다.

  
# AI 어시스턴트 시스템 프롬프트

당신은 전문적이고 도움이 되는 AI 어시스턴트입니다.

## 역할
- 사용자의 질문에 정확하고 유용한 답변 제공
- 필요시 도구를 사용하여 최신 정보 검색
- 이전 대화 내용을 기억하고 맥락 유지

## 원칙
1. **정직성**: 모르는 것은 모른다고 말하기
2. **간결성**: 불필요한 설명 피하기
3. **도움**: 사용자 의도 파악 후 최선의 답변
4. **안전**: 해로운 정보 제공 거부

## 사용 가능한 도구
- web_search: 최신 정보 검색
- calculator: 복잡한 계산

## 동적 정보
- 현재 시각: 
- 사용자 장기 기억: 

## 응답 형식
- 명확하고 구조화된 답변
- 필요시 단계별 설명
- 출처 명시 (검색 도구 사용 시)

app/core/prompts/init.py에 프롬프트 로더를 구현합니다.

  
from pathlib import Path
from datetime import datetime, UTC
from typing import Optional

PROMPTS_DIR = Path(__file__).parent

def load_system_prompt(
    current_time: Optional[datetime] = None,
    long_term_memory: Optional[str] = None
) -> str:
    """
    시스템 프롬프트 로드 및 변수 치환
    
    Args:
        current_time: 현재 시각 (None이면 자동)
        long_term_memory: 사용자 장기 기억
    
    Returns:
        완성된 시스템 프롬프트
    """
    with open(PROMPTS_DIR / "system.md", "r", encoding="utf-8") as f:
        prompt = f.read()
    
    # 변수 치환
    if current_time is None:
        current_time = datetime.now(UTC)
    
    prompt = prompt.replace(
        "",
        current_time.strftime("%Y-%m-%d %H:%M:%S UTC")
    )
    
    prompt = prompt.replace(
        "",
        long_term_memory or "없음"
    )
    
    return prompt

6.2 검색 도구 구현

app/core/langgraph/tools/search.py에 웹 검색 도구를 구현합니다.

  
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_core.tools import tool

# DuckDuckGo 검색 도구
ddg_search = DuckDuckGoSearchResults(
    num_results=5,
    backend="api"
)

def web_search(query: str) -> str:
    """
    웹에서 정보를 검색합니다.
    
    Args:
        query: 검색 쿼리
    
    Returns:
        검색 결과 (상위 5개)
    """
    try:
        results = ddg_search.run(query)
        return f"검색 결과:\n{results}"
    except Exception as e:
        return f"검색 실패: {str(e)}"

def calculator(expression: str) -> str:
    """
    수식을 계산합니다.
    
    Args:
        expression: 계산할 수식 (예: "2 + 2 * 3")
    
    Returns:
        계산 결과
    """
    try:
        # 안전한 eval (ast.literal_eval은 수식 지원 안 함)
        # 실제로는 sympy 같은 라이브러리 사용 권장
        result = eval(expression, {"__builtins__": {}})
        return f"결과: {result}"
    except Exception as e:
        return f"계산 실패: {str(e)}"

# 도구 목록
TOOLS = [web_search, calculator]

보안 주의사항:

eval()은 위험! 제한된 환경에서만 사용
실제로는 sympy 또는 numexpr 사용 권장

6.3 LangGraph Agent 구현 (핵심)

app/core/langgraph/graph.py에 AI 에이전트의 두뇌를 구현합니다.

  
from typing import Annotated, Literal, TypedDict
from uuid import UUID
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.prebuilt import ToolNode
from mem0 import AsyncMemory
from app.core.config import settings
from app.core.services.llm import llm_service
from app.core.prompts import load_system_prompt
from app.core.langgraph.tools.search import TOOLS

# 상태 정의
class GraphState(TypedDict):
    """LangGraph 상태"""
    messages: list[BaseMessage]
    long_term_memory: str


class LangGraphAgent:
    """LangGraph 기반 AI 에이전트"""
    
    def __init__(self):
        # LLM 서비스
        self.llm_service = llm_service
        
        # Mem0 초기화 (장기 기억)
        self.memory = AsyncMemory.from_config({
            "vector_store": {
                "provider": "pgvector",
                "config": {
                    "host": settings.MEM0_POSTGRES_HOST,
                    "port": settings.MEM0_POSTGRES_PORT,
                    "user": settings.POSTGRES_USER,
                    "password": settings.POSTGRES_PASSWORD.get_secret_value(),
                    "database": settings.POSTGRES_DB,
                }
            }
        })
        
        # PostgreSQL Checkpointer (대화 상태 저장)
        self.checkpointer = AsyncPostgresSaver.from_conn_string(
            settings.database_url
        )
        
        # 그래프 빌드
        self.graph = self._build_graph()
    
    def _build_graph(self) -> StateGraph:
        """StateGraph 구축"""
        # 그래프 정의
        workflow = StateGraph(GraphState)
        
        # 노드 추가
        workflow.add_node("chat", self._chat_node)
        workflow.add_node("tool_call", ToolNode(TOOLS))
        
        # 진입점
        workflow.set_entry_point("chat")
        
        # 조건부 엣지
        workflow.add_conditional_edges(
            "chat",
            self._should_use_tools,
            {
                "tools": "tool_call",
                "end": END
            }
        )
        
        # 도구 실행 후 다시 chat으로
        workflow.add_edge("tool_call", "chat")
        
        # 컴파일
        return workflow.compile(checkpointer=self.checkpointer)
    
    async def _chat_node(self, state: GraphState) -> GraphState:
        """채팅 노드: LLM 호출"""
        messages = state["messages"]
        long_term_memory = state.get("long_term_memory", "")
        
        # 시스템 프롬프트 생성
        system_prompt = load_system_prompt(
            long_term_memory=long_term_memory
        )
        
        # 메시지 구성
        formatted_messages = [
            {"role": "system", "content": system_prompt}
        ] + [
            {
                "role": "user" if isinstance(msg, HumanMessage) else "assistant",
                "content": msg.content
            }
            for msg in messages
        ]
        
        # LLM 호출 (도구 바인딩)
        response = await self.llm_service.call(
            messages=formatted_messages,
            **await self.llm_service.bind_tools(
                [tool.to_langchain_tool() for tool in TOOLS]
            )
        )
        
        # 응답 추가
        return {
            "messages": messages + [AIMessage(content=response)],
            "long_term_memory": long_term_memory
        }
    
    def _should_use_tools(self, state: GraphState) -> Literal["tools", "end"]:
        """도구 사용 여부 판단"""
        last_message = state["messages"][-1]
        
        # AIMessage에 tool_calls가 있으면 도구 사용
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return "tools"
        
        return "end"
    
    async def get_response(
        self,
        messages: list[dict],
        user_id: int,
        session_id: UUID
    ) -> str:
        """
        에이전트 응답 생성
        
        Args:
            messages: 대화 메시지
            user_id: 사용자 ID
            session_id: 세션 ID
        
        Returns:
            AI 응답
        """
        # Mem0에서 장기 기억 검색
        memory_results = await self.memory.search(
            query=messages[-1]["content"],
            user_id=str(user_id),
            limit=3
        )
        
        long_term_memory = "\n".join([
            f"- {result['memory']}"
            for result in memory_results
        ]) if memory_results else "없음"
        
        # 메시지를 LangChain 형식으로 변환
        lc_messages = [
            HumanMessage(content=msg["content"])
            if msg["role"] == "user"
            else AIMessage(content=msg["content"])
            for msg in messages
        ]
        
        # 그래프 실행
        config = {"configurable": {"thread_id": str(session_id)}}
        
        final_state = await self.graph.ainvoke(
            {
                "messages": lc_messages,
                "long_term_memory": long_term_memory
            },
            config=config
        )
        
        # 응답 추출
        response = final_state["messages"][-1].content
        
        # 백그라운드로 Mem0 업데이트
        asyncio.create_task(
            self._update_memory(messages, response, user_id)
        )
        
        return response
    
    async def _update_memory(
        self,
        messages: list[dict],
        response: str,
        user_id: int
    ):
        """장기 기억 업데이트 (백그라운드)"""
        try:
            # 대화에서 중요한 정보 추출
            conversation = "\n".join([
                f"{msg['role']}: {msg['content']}"
                for msg in messages + [{"role": "assistant", "content": response}]
            ])
            
            await self.memory.add(
                conversation,
                user_id=str(user_id)
            )
        except Exception as e:
            print(f"Memory update failed: {e}")
    
    async def get_stream_response(
        self,
        messages: list[dict],
        user_id: int,
        session_id: UUID
    ):
        """
        스트리밍 응답 생성 (SSE용)
        
        Yields:
            응답 청크
        """
        # 구현 생략 (실제로는 LLM streaming API 사용)
        response = await self.get_response(messages, user_id, session_id)
        
        # 청크로 나눠서 yield
        for chunk in response.split():
            yield f"{chunk} "
            await asyncio.sleep(0.05)  # 스트리밍 효과

# 싱글톤
agent = LangGraphAgent()

설계 포인트:

StateGraph: 복잡한 워크플로우를 명확하게 표현
Checkpointer: 대화 상태를 PostgreSQL에 저장 (재시작해도 이어짐)
Mem0 통합: 장기 기억을 검색하여 컨텍스트 향상
도구 바인딩: LLM이 필요시 도구 사용
백그라운드 업데이트: 응답 지연 없이 메모리 업데이트

6.4 그래프 흐름 상세 설명

우리가 만든 LangGraph는 다음과 같이 작동합니다:

1. get_response() 호출
   ↓
2. Mem0에서 장기 기억 검색
   - "사용자가 채식주의자라고 언급"
   ↓
3. chat 노드 진입
   - 시스템 프롬프트 생성 (장기 기억 포함)
   - LLM 호출
   ↓
4. LLM이 도구 사용 필요 판단
   - 예: "날씨 검색 필요"
   ↓
5. should_use_tools() → "tools"
   ↓
6. tool_call 노드
   - web_search("서울 날씨") 실행
   ↓
7. 다시 chat 노드
   - 검색 결과를 바탕으로 최종 답변
   ↓
8. should_use_tools() → "end"
   ↓
9. 응답 반환
   ↓
10. 백그라운드로 Mem0 업데이트

Checkpoint의 역할:

각 노드 실행 후 상태가 PostgreSQL에 저장됩니다. 서버가 재시작되어도:

  
config = {"configurable": {"thread_id": str(session_id)}}
# 같은 thread_id면 이전 상태에서 이어서 실행

7. Layer 7: API와 배포

사용자가 접근할 수 있는 API를 만들고 배포합니다.

7.1 FastAPI 메인 애플리케이션

app/main.py에 FastAPI 앱을 정의합니다.

  
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.core.config import settings
from app.core.security.limiter import limiter
from app.core.services.database import db_service
from app.core.langgraph.graph import agent
from app.api.v1 import auth, chatbot
import logging

# 로깅 설정
logging.basicConfig(
    level=getattr(logging, settings.LOG_LEVEL),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


async def lifespan(app: FastAPI):
    """앱 시작/종료 시 실행"""
    # 시작
    logger.info("Starting application...")
    
    # 데이터베이스 테이블 생성
    await db_service.create_tables()
    
    # LangGraph 에이전트 초기화
    await agent.initialize()
    
    logger.info("Application started successfully")
    
    yield
    
    # 종료
    logger.info("Shutting down application...")
    await db_service.close()
    logger.info("Application stopped")


# FastAPI 앱 생성
app = FastAPI(
    title=settings.APP_NAME,
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs" if settings.DEBUG else None,  # 프로덕션에서는 문서 비활성화
    redoc_url="/redoc" if settings.DEBUG else None,
)

# CORS 미들웨어
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 프론트엔드 주소
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Rate limiting 미들웨어
app.state.limiter = limiter

# 라우터 등록
app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
app.include_router(chatbot.router, prefix="/api/v1/chatbot", tags=["chatbot"])


# 기본 엔드포인트
async def root():
    """서비스 정보"""
    return {
        "service": settings.APP_NAME,
        "version": "1.0.0",
        "status": "running"
    }


async def health_check():
    """헬스 체크"""
    try:
        # 데이터베이스 연결 확인
        await db_service.health_check()
        
        return {
            "status": "healthy",
            "database": "connected"
        }
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "error": str(e)
            }
        )


# 전역 예외 핸들러
async def global_exception_handler(request, exc):
    """예상치 못한 에러 처리"""
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "detail": str(exc) if settings.DEBUG else "An error occurred"
        }
    )

7.2 인증 API

app/api/v1/auth.py에 인증 관련 엔드포인트를 구현합니다.

  
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from app.core.security.auth import create_access_token, verify_token
from app.core.security.limiter import limiter
from app.core.services.database import db_service
from app.schemas.auth import UserCreate, UserResponse, Token
from app.models.database import User

router = APIRouter()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    """현재 사용자 조회 (JWT 검증)"""
    try:
        email = verify_token(token)
        user = await db_service.get_user_by_email(email)
        
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials"
            )
        
        return user
    
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )


async def register(user_data: UserCreate):
    """회원가입"""
    # 이메일 중복 확인
    existing_user = await db_service.get_user_by_email(user_data.email)
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )
    
    # 사용자 생성
    user = await db_service.create_user(
        email=user_data.email,
        password=user_data.password
    )
    
    return UserResponse(
        id=user.id,
        email=user.email,
        created_at=user.created_at
    )


async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """로그인"""
    # 사용자 조회
    user = await db_service.get_user_by_email(form_data.username)
    
    if user is None or not user.verify_password(form_data.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )
    
    # JWT 토큰 생성
    token = create_access_token(subject=user.email)
    
    return token


async def create_session(
    name: str = "New Chat",
    current_user: User = Depends(get_current_user)
):
    """새 대화 세션 생성"""
    session = await db_service.create_session(
        user_id=current_user.id,
        name=name
    )
    
    return {
        "id": str(session.id),
        "name": session.name,
        "created_at": session.created_at
    }


async def get_sessions(current_user: User = Depends(get_current_user)):
    """사용자의 세션 목록"""
    sessions = await db_service.get_user_sessions(current_user.id)
    
    return {
        "sessions": [
            {
                "id": str(s.id),
                "name": s.name,
                "updated_at": s.updated_at
            }
            for s in sessions
        ]
    }

7.3 채팅 API

app/api/v1/chatbot.py에 채팅 엔드포인트를 구현합니다.

  
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from app.core.security.limiter import limiter
from app.core.services.database import db_service
from app.core.langgraph.graph import agent
from app.api.v1.auth import get_current_user
from app.models.database import User
from app.schemas.chatbot import ChatRequest, ChatResponse

router = APIRouter()


async def chat(
    request: ChatRequest,
    current_user: User = Depends(get_current_user)
):
    """채팅 (전체 응답)"""
    # 세션 존재 확인
    session = await db_service.get_session(request.session_id)
    if session is None or session.user_id != current_user.id:
        raise HTTPException(status_code=404, detail="Session not found")
    
    # 메시지 저장
    await db_service.create_message(
        session_id=request.session_id,
        role="user",
        content=request.message
    )
    
    # 대화 히스토리 가져오기
    messages = await db_service.get_session_messages(request.session_id)
    
    # AI 응답 생성
    response = await agent.get_response(
        messages=[
            {"role": m.role, "content": m.content}
            for m in messages
        ],
        user_id=current_user.id,
        session_id=request.session_id
    )
    
    # 응답 저장
    await db_service.create_message(
        session_id=request.session_id,
        role="assistant",
        content=response
    )
    
    return ChatResponse(
        message=response,
        session_id=request.session_id
    )


async def chat_stream(
    request: ChatRequest,
    current_user: User = Depends(get_current_user)
):
    """채팅 (스트리밍 응답)"""
    # 세션 확인
    session = await db_service.get_session(request.session_id)
    if session is None or session.user_id != current_user.id:
        raise HTTPException(status_code=404, detail="Session not found")
    
    # 메시지 저장
    await db_service.create_message(
        session_id=request.session_id,
        role="user",
        content=request.message
    )
    
    # 대화 히스토리
    messages = await db_service.get_session_messages(request.session_id)
    
    async def generate():
        """SSE 스트림 생성"""
        full_response = ""
        
        async for chunk in agent.get_stream_response(
            messages=[
                {"role": m.role, "content": m.content}
                for m in messages
            ],
            user_id=current_user.id,
            session_id=request.session_id
        ):
            full_response += chunk
            yield f"data: {chunk}\n\n"
        
        # 완전한 응답 저장
        await db_service.create_message(
            session_id=request.session_id,
            role="assistant",
            content=full_response
        )
        
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )


async def get_messages(
    session_id: UUID,
    current_user: User = Depends(get_current_user)
):
    """세션의 메시지 히스토리"""
    session = await db_service.get_session(session_id)
    if session is None or session.user_id != current_user.id:
        raise HTTPException(status_code=404, detail="Session not found")
    
    messages = await db_service.get_session_messages(session_id)
    
    return {
        "messages": [
            {
                "role": m.role,
                "content": m.content,
                "created_at": m.created_at
            }
            for m in messages
        ]
    }

7.4 Docker 설정

docker-compose.yml을 작성합니다.

  
version: '3.8'

services:
  # PostgreSQL + pgvector
  db:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: ${POSTGRES_DB}
    volumes:
      - db-data:/var/lib/postgresql/data
      - ./database/schema.sql:/docker-entrypoint-initdb.d/schema.sql
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
      interval: 10s
      timeout: 5s
      retries: 5

  # FastAPI 앱
  app:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - DATABASE_URL=postgresql+psycopg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    depends_on:
      db:
        condition: service_healthy
    ports:
      - "8000:8000"
    volumes:
      - ./app:/app/app
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

  # Prometheus (모니터링)
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'

  # Grafana (시각화)
  grafana:
    image: grafana/grafana:latest
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana-data:/var/lib/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

volumes:
  db-data:
  prometheus-data:
  grafana-data:

Dockerfile을 작성합니다.

  
FROM python:3.13-slim

WORKDIR /app

# 시스템 패키지 설치
RUN apt-get update && apt-get install -y \
    gcc \
    postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Python 의존성 설치
COPY pyproject.toml ./
RUN pip install --no-cache-dir -e .

# 애플리케이션 코드 복사
COPY app ./app

# 포트 노출
EXPOSE 8000

# 실행
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

.dockerignore를 작성합니다.

__pycache__/
*.py[cod]
.env
.env.*
.git/
.gitignore
.vscode/
tests/
*.md
!README.md

8. 성능 최적화

시스템이 작동하면 이제 빠르게 만들어야 합니다.

8.1 대화 히스토리 요약

대화가 길어지면 토큰 수가 증가하여 비용과 지연이 발생합니다.

문제: 50턴 대화 = 약 10,000 토큰 = LLM 입력에 매번 포함

해결책: 오래된 대화를 요약

  
# app/core/services/summarizer.py

async def summarize_conversation(messages: list[dict], max_turns: int = 10) -> list[dict]:
    """
    대화 히스토리 요약
    
    Args:
        messages: 전체 메시지
        max_turns: 유지할 최신 턴 수
    
    Returns:
        요약된 메시지
    """
    if len(messages) <= max_turns * 2:  # user + assistant = 1턴
        return messages
    
    # 최신 메시지는 그대로 유지
    recent_messages = messages[-(max_turns * 2):]
    
    # 이전 메시지는 요약
    old_messages = messages[:-(max_turns * 2)]
    
    # LLM에게 요약 요청
    summary_prompt = f"""
다음 대화를 3-5문장으로 요약해주세요:

{chr(10).join([f"{m['role']}: {m['content']}" for m in old_messages])}
"""
    
    summary = await llm_service.call(
        messages=[{"role": "user", "content": summary_prompt}],
        model="claude-haiku-4-5-20251001",  # 빠르고 저렴한 모델
        max_tokens=200
    )
    
    # 요약을 시스템 메시지로 추가
    return [
        {"role": "system", "content": f"이전 대화 요약: {summary}"}
    ] + recent_messages

효과:

토큰 수: 10,000 → 2,000 (80% 감소)
비용: $0.03 → $0.006 (80% 절감)
응답 시간: 3초 → 0.8초 (73% 개선)

8.2 캐싱 전략

같은 질문에 대해 매번 LLM을 호출하는 것은 낭비입니다.

  
# app/core/services/cache.py

from functools import lru_cache
from hashlib import sha256
import redis.asyncio as redis
from app.core.config import settings

class CacheService:
    """Redis 캐시 서비스"""
    
    def __init__(self):
        self.redis = redis.from_url(
            f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}"
        )
    
    def _generate_key(self, user_id: int, message: str) -> str:
        """캐시 키 생성"""
        data = f"{user_id}:{message}"
        return f"chat:{sha256(data.encode()).hexdigest()}"
    
    async def get(self, user_id: int, message: str) -> str | None:
        """캐시에서 응답 조회"""
        key = self._generate_key(user_id, message)
        value = await self.redis.get(key)
        return value.decode() if value else None
    
    async def set(
        self,
        user_id: int,
        message: str,
        response: str,
        ttl: int = 300  # 5분
    ):
        """응답을 캐시에 저장"""
        key = self._generate_key(user_id, message)
        await self.redis.setex(key, ttl, response)

cache_service = CacheService()

채팅 API에서 사용:

  
async def chat(request: ChatRequest, current_user: User = Depends(get_current_user)):
    # 캐시 확인
    cached_response = await cache_service.get(current_user.id, request.message)
    if cached_response:
        return ChatResponse(message=cached_response, session_id=request.session_id)
    
    # LLM 호출
    response = await agent.get_response(...)
    
    # 캐시 저장
    await cache_service.set(current_user.id, request.message, response)
    
    return ChatResponse(message=response, session_id=request.session_id)

효과:

캐시 히트 시 응답 시간: 3초 → 0.05초 (60배 빠름)
LLM API 호출 비용 절감

8.3 데이터베이스 최적화

복합 인덱스 추가:

  
-- 자주 함께 조회하는 컬럼
CREATE INDEX idx_messages_session_created 
ON messages(session_id, created_at);

-- WHERE와 ORDER BY에 모두 사용
CREATE INDEX idx_sessions_user_updated 
ON sessions(user_id, updated_at DESC);

연결 풀 튜닝:

  
# app/core/services/database.py

self._engine = create_async_engine(
    settings.database_url,
    pool_size=20,        # 기본 연결 수
    max_overflow=10,     # 추가 연결 수
    pool_pre_ping=True,  # 연결 유효성 체크
    pool_recycle=3600,   # 1시간마다 연결 재생성
)

N+1 쿼리 문제 해결:

  
# 나쁜 예
sessions = await db.get_user_sessions(user_id)
for session in sessions:
    messages = await db.get_session_messages(session.id)  # N번 쿼리

# 좋은 예
sessions_with_messages = await db.execute(
    select(Session)
    .options(selectinload(Session.messages))
    .where(Session.user_id == user_id)
)

8.4 비동기 처리 최적화

병렬 실행:

  
import asyncio

# 순차 실행 (느림)
memory = await memory_service.search(query, user_id)
history = await db_service.get_session_messages(session_id)
# 총 시간: 200ms + 150ms = 350ms

# 병렬 실행 (빠름)
memory, history = await asyncio.gather(
    memory_service.search(query, user_id),
    db_service.get_session_messages(session_id)
)
# 총 시간: max(200ms, 150ms) = 200ms (43% 개선)

9. 보안 강화

프로덕션 환경에서는 보안이 최우선입니다.

9.1 환경 변수 보안

.env 파일 절대 노출 금지:

  
# .gitignore에 추가
.env
.env.*
!.env.example

강력한 비밀 키 생성:

  
import secrets

# JWT_SECRET_KEY 생성
secret_key = secrets.token_urlsafe(32)
print(secret_key)  # 예: "dGhpc19pc19hX3Zlcnlfc2VjdXJlX2tleQ"

환경별 설정 파일:

.env.development  # 로컬 개발
.env.test         # 테스트
.env.production   # 프로덕션

9.2 SQL Injection 방어

SQLModel은 자동으로 방어하지만, 원시 SQL 사용 시 주의:

  
# 절대 이렇게 하지 마세요!
query = f"SELECT * FROM users WHERE email = '{user_input}'"
# user_input = "'; DROP TABLE users; --" 이면 테이블 삭제!

# 안전한 방법
query = select(User).where(User.email == user_input)
# SQLModel이 자동으로 파라미터화

9.3 XSS (Cross-Site Scripting) 방어

사용자 입력을 HTML에 표시할 때:

  
from app.core.security.sanitization import sanitize_input

# 모든 사용자 입력 정제
message = sanitize_input(request.message)

FastAPI는 자동으로 JSON 이스케이프하지만, 프론트엔드에서도 주의:

  
// React에서 안전
<div>{message}</div>

// Vanilla JS에서 위험
element.innerHTML = message;  // XSS 가능!

// 안전한 방법
element.textContent = message;

9.4 Rate Limiting 강화

IP + 사용자 조합:

  
from slowapi import Limiter
from fastapi import Request

def get_rate_limit_key(request: Request) -> str:
    """IP와 사용자 ID 조합"""
    client_ip = request.client.host
    user_id = getattr(request.state, "user_id", "anonymous")
    return f"{client_ip}:{user_id}"

limiter = Limiter(key_func=get_rate_limit_key)

엔드포인트별 다른 제한:

  
async def chat(...):
    pass

async def register(...):
    pass

9.5 HTTPS 강제

프로덕션에서는 반드시 HTTPS:

  
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

if settings.ENVIRONMENT == Environment.PRODUCTION:
    app.add_middleware(HTTPSRedirectMiddleware)

9.6 API 키 노출 방지

응답에 민감한 정보 포함 금지:

  
# 나쁜 예
return {
    "user": user,  # password_hash 포함!
    "api_key": settings.ANTHROPIC_API_KEY  # 절대 안 됨!
}

# 좋은 예
return UserResponse(
    id=user.id,
    email=user.email,
    # password_hash는 스키마에서 제외
)

10. 모니터링과 로깅

운영 환경에서 무슨 일이 일어나는지 알아야 합니다.

10.1 구조화된 로깅

  
import logging
import json
from datetime import datetime, UTC

class JSONFormatter(logging.Formatter):
    """JSON 형식 로깅"""
    
    def format(self, record):
        log_data = {
            "timestamp": datetime.now(UTC).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        
        return json.dumps(log_data)

# 설정
logger = logging.getLogger(__name__)
handler = logging.FileHandler("app.log")
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

10.2 Prometheus 메트릭

  
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

# 메트릭 정의
http_requests_total = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"]
)

http_request_duration_seconds = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"]
)

llm_calls_total = Counter(
    "llm_calls_total",
    "Total LLM API calls",
    ["model", "status"]
)

async def prometheus_middleware(request, call_next):
    """HTTP 메트릭 수집"""
    method = request.method
    path = request.url.path
    
    with http_request_duration_seconds.labels(method, path).time():
        response = await call_next(request)
    
    http_requests_total.labels(method, path, response.status_code).inc()
    
    return response

async def metrics():
    """Prometheus 메트릭 엔드포인트"""
    return Response(
        generate_latest(),
        media_type="text/plain"
    )

10.3 Grafana 대시보드

monitoring/prometheus.yml:

  
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'fastapi'
    static_configs:
      - targets: ['app:8000']

Grafana에서 대시보드 생성:

Prometheus 데이터 소스 추가
패널 추가:
- 요청 수: rate(http_requests_total[5m])
- 응답 시간: http_request_duration_seconds
- LLM 호출 수: llm_calls_total
- 에러율: rate(http_requests_total{status=~"5.."}[5m])

10.4 알람 설정

Prometheus alerting rules:

  
# monitoring/alerts.yml
groups:
  - name: api_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.05
        for: 5m
        annotations:
          summary: "High error rate detected"
      
      - alert: SlowResponse
        expr: http_request_duration_seconds{quantile="0.95"} > 5
        for: 10m
        annotations:
          summary: "95th percentile response time > 5s"

11. 프로덕션 배포

로컬에서 프로덕션으로 전환합니다.

11.1 환경 분리

.env.production:

  
# Production 설정
APP_NAME=Production AI Agent
DEBUG=False
ENVIRONMENT=production
LOG_LEVEL=WARNING

# 실제 데이터베이스
DATABASE_URL=postgresql://...  # AWS RDS 주소

# 강력한 비밀 키 (절대 노출 금지!)
JWT_SECRET_KEY=production_secret_key_very_long_and_random

# Rate limiting 강화
RATE_LIMIT_PER_MINUTE=5

11.2 Docker 프로덕션 빌드

멀티스테이지 Dockerfile:

  
# 빌드 스테이지
FROM python:3.13-slim as builder

WORKDIR /app

RUN apt-get update && apt-get install -y gcc
COPY pyproject.toml ./
RUN pip install --user --no-cache-dir -e .

# 런타임 스테이지
FROM python:3.13-slim

WORKDIR /app

# 빌드에서 패키지만 복사
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

# 애플리케이션 코드
COPY app ./app

# 비root 사용자 생성
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

최적화 효과:

이미지 크기: 1.2GB → 400MB
빌드 시간: 5분 → 2분
보안: root 사용자 미사용

11.3 AWS 배포 (ECS)

1. ECR에 이미지 푸시:

  
# AWS CLI 로그인
aws ecr get-login-password --region ap-northeast-2 | docker login --username AWS --password-stdin {account}.dkr.ecr.ap-northeast-2.amazonaws.com

# 이미지 빌드 및 태그
docker build -t ai-agent .
docker tag ai-agent:latest {account}.dkr.ecr.ap-northeast-2.amazonaws.com/ai-agent:latest

# 푸시
docker push {account}.dkr.ecr.ap-northeast-2.amazonaws.com/ai-agent:latest

2. ECS Task Definition:

  
{
  "family": "ai-agent-task",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "1024",
  "memory": "2048",
  "containerDefinitions": [
    {
      "name": "ai-agent",
      "image": "{account}.dkr.ecr.ap-northeast-2.amazonaws.com/ai-agent:latest",
      "portMappings": [
        {
          "containerPort": 8000,
          "protocol": "tcp"
        }
      ],
      "environment": [
        {
          "name": "ENVIRONMENT",
          "value": "production"
        }
      ],
      "secrets": [
        {
          "name": "ANTHROPIC_API_KEY",
          "valueFrom": "arn:aws:secretsmanager:..."
        }
      ],
      "logConfiguration": {
        "logDriver": "awslogs",
        "options": {
          "awslogs-group": "/ecs/ai-agent",
          "awslogs-region": "ap-northeast-2",
          "awslogs-stream-prefix": "ecs"
        }
      }
    }
  ]
}

3. ECS Service + ALB:

  
# Service 생성
aws ecs create-service \
  --cluster ai-agent-cluster \
  --service-name ai-agent-service \
  --task-definition ai-agent-task \
  --desired-count 2 \
  --launch-type FARGATE \
  --network-configuration "awsvpcConfiguration={subnets=[subnet-xxx],securityGroups=[sg-xxx],assignPublicIp=ENABLED}" \
  --load-balancers "targetGroupArn=arn:aws:elasticloadbalancing:...,containerName=ai-agent,containerPort=8000"

11.4 CI/CD 파이프라인

GitHub Actions (.github/workflows/deploy.yml):

  
name: Deploy to Production

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v2
        with:
          aws-access-key-id: $
          aws-secret-access-key: $
          aws-region: ap-northeast-2
      
      - name: Login to Amazon ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v1
      
      - name: Build and push Docker image
        env:
          ECR_REGISTRY: $
          ECR_REPOSITORY: ai-agent
          IMAGE_TAG: $
        run: |
          docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
      
      - name: Update ECS service
        run: |
          aws ecs update-service \
            --cluster ai-agent-cluster \
            --service ai-agent-service \
            --force-new-deployment

12. 트러블슈팅

실제 운영에서 발생하는 문제와 해결 방법입니다.

12.1 데이터베이스 연결 오류

증상: psycopg.OperationalError: could not connect to server

원인:

PostgreSQL 서비스 미실행
잘못된 연결 문자열
방화벽 차단

해결:

  
# 1. PostgreSQL 실행 확인
docker ps | grep postgres

# 2. 연결 테스트
docker exec -it {container_id} psql -U aiagent -d aiagent_db

# 3. .env 확인
cat .env | grep POSTGRES

12.2 API 키 관련 오류

증상: AuthenticationError: Invalid API key

원인:

API 키 미설정
잘못된 키
만료된 키

해결:

  
# 키 확인
print(settings.ANTHROPIC_API_KEY.get_secret_value()[:10])  # 처음 10자만

# 새 키로 교체
# .env 파일에서 ANTHROPIC_API_KEY 업데이트

12.3 메모리 부족

증상: 컨테이너가 자꾸 재시작됨

원인: Docker 메모리 제한 초과

해결:

  
# docker-compose.yml
services:
  app:
    mem_limit: 2g  # 메모리 제한 증가
    mem_reservation: 1g

12.4 느린 응답 시간

진단:

  
import time

start = time.time()
response = await agent.get_response(...)
print(f"Response time: {time.time() - start:.2f}s")

원인별 해결:

LLM 호출이 느림 → 모델 변경 (Sonnet → Haiku)
DB 쿼리가 느림 → 인덱스 추가, 쿼리 최적화
대화 히스토리가 길음 → 요약 적용

12.5 Docker 컨테이너 간 통신 실패

증상: app이 db에 연결 못 함

원인: 네트워크 설정 오류

해결:

  
# docker-compose.yml
services:
  db:
    networks:
      - app-network
  
  app:
    networks:
      - app-network
    environment:
      # 'localhost' 대신 서비스 이름 사용
      DATABASE_URL: postgresql://user:pass@db:5432/aiagent_db

networks:
  app-network:
    driver: bridge

12.6 pgvector 확장 오류

증상: ERROR: extension "vector" does not exist

해결:

  
-- PostgreSQL 셸에 접속
docker exec -it {db_container} psql -U aiagent -d aiagent_db

-- 확장 설치
CREATE EXTENSION IF NOT EXISTS vector;

-- 확인
\dx

13. 확장과 커스터마이징

이제 시스템을 자신의 요구사항에 맞게 확장합니다.

13.1 새로운 도구 추가

예시: 날씨 조회 도구

  
# app/core/langgraph/tools/weather.py

import httpx
from langchain_core.tools import tool

async def get_weather(city: str) -> str:
    """
    도시의 현재 날씨를 조회합니다.
    
    Args:
        city: 도시 이름
    
    Returns:
        날씨 정보
    """
    api_key = "your_weather_api_key"
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
    
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        data = response.json()
        
        temp = data["main"]["temp"] - 273.15  # Kelvin to Celsius
        desc = data["weather"][0]["description"]
        
        return f"{city}의 현재 날씨: {temp:.1f}°C, {desc}"

그래프에 추가:

  
# app/core/langgraph/graph.py

from app.core.langgraph.tools.weather import get_weather

TOOLS = [web_search, calculator, get_weather]

13.2 다른 LLM 추가

예시: Gemini 추가

  
# app/core/services/llm.py

from google.generativeai import GenerativeModel

class LLMService:
    def __init__(self):
        # 기존 코드...
        
        if settings.GEMINI_API_KEY:
            self.gemini_client = GenerativeModel("gemini-pro")
    
    async def _call_gemini(self, messages, temperature, max_tokens, **kwargs):
        """Google Gemini 호출"""
        # Gemini는 다른 형식 사용
        prompt = "\n".join([f"{m['role']}: {m['content']}" for m in messages])
        
        response = await self.gemini_client.generate_content_async(
            prompt,
            generation_config={
                "temperature": temperature,
                "max_output_tokens": max_tokens,
            }
        )
        
        return response.text

13.3 파일 업로드 기능

  
# app/api/v1/files.py

from fastapi import UploadFile, File
from pathlib import Path

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

async def upload_file(
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_user)
):
    """파일 업로드"""
    # 파일 크기 제한 (10MB)
    if file.size > 10 * 1024 * 1024:
        raise HTTPException(400, "File too large")
    
    # 허용된 확장자
    allowed_extensions = {".pdf", ".txt", ".png", ".jpg"}
    ext = Path(file.filename).suffix.lower()
    if ext not in allowed_extensions:
        raise HTTPException(400, "File type not allowed")
    
    # 저장
    file_path = UPLOAD_DIR / f"{current_user.id}_{file.filename}"
    
    with open(file_path, "wb") as f:
        content = await file.read()
        f.write(content)
    
    return {"filename": file.filename, "path": str(file_path)}

13.4 WebSocket 실시간 채팅

  
# app/api/v1/websocket.py

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, WebSocket] = {}
    
    async def connect(self, websocket: WebSocket, user_id: str):
        await websocket.accept()
        self.active_connections[user_id] = websocket
    
    def disconnect(self, user_id: str):
        self.active_connections.pop(user_id, None)
    
    async def send_message(self, user_id: str, message: str):
        if user_id in self.active_connections:
            await self.active_connections[user_id].send_text(message)

manager = ConnectionManager()

async def websocket_chat(websocket: WebSocket, session_id: str):
    await manager.connect(websocket, session_id)
    
    try:
        while True:
            # 메시지 수신
            data = await websocket.receive_text()
            
            # AI 응답 생성 (스트리밍)
            async for chunk in agent.get_stream_response(...):
                await websocket.send_text(chunk)
    
    except WebSocketDisconnect:
        manager.disconnect(session_id)

부록

A. 유용한 PowerShell 명령어

  
# 프로젝트 전체 검색
Select-String -Path "app/**/*.py" -Pattern "async def"

# 파일 개수 세기
(Get-ChildItem -Recurse -Filter "*.py").Count

# 코드 라인 수
(Get-Content -Path "app/**/*.py" | Measure-Object -Line).Lines

# 포트 사용 확인
netstat -ano | findstr :8000

# 프로세스 종료
taskkill /PID {process_id} /F

B. 참고 자료

공식 문서:

FastAPI: https://fastapi.tiangolo.com
LangGraph: https://langchain-ai.github.io/langgraph
Mem0: https://docs.mem0.ai
SQLModel: https://sqlmodel.tiangolo.com

튜토리얼:

Anthropic Claude: https://docs.anthropic.com
PostgreSQL: https://www.postgresql.org/docs
Docker: https://docs.docker.com

커뮤니티:

Discord: LangChain Discord
GitHub: 이슈 및 토론
Stack Overflow: 태그 검색

C. 체크리스트

개발 완료 전:

모든 환경 변수 설정
데이터베이스 마이그레이션 실행
단위 테스트 작성 및 통과
API 문서 확인 (/docs)
보안 체크리스트 검토
로깅 설정 확인
에러 핸들링 검증

배포 전:

.env.production 준비
Docker 이미지 빌드 성공
프로덕션 DB 백업
롤백 계획 수립
모니터링 대시보드 준비
알람 규칙 설정
부하 테스트 실행

배포 후:

Health check 확인
로그 모니터링
에러율 확인
응답 시간 측정
사용자 피드백 수집

마치며

이 가이드를 완료했다면, 여러분은 이제 프로덕션급 AI 에이전트 시스템을 구축할 수 있는 능력을 갖추었습니다.

배운 것들:

7계층 아키텍처 설계
FastAPI + LangGraph + Mem0 통합
보안과 성능 최적화
Docker를 통한 배포
모니터링과 트러블슈팅

다음 단계:

이 시스템을 실제 프로젝트에 적용
자신만의 기능 추가
성능 벤치마크 측정
프로덕션 배포 경험
커뮤니티에 기여

중요한 원칙:

이해하며 개발하기 (복사-붙여넣기만 하지 말 것)
보안을 타협하지 말 것
테스트를 게을리하지 말 것
문서화를 습관화할 것
커뮤니티와 소통할 것

바이브 코딩은 도구일 뿐, 진짜 개발자는 여러분입니다. Claude Code는 여러분의 아이디어를 빠르게 구현하도록 돕지만, 설계와 의사결정은 여러분의 몫입니다.

행운을 빕니다! 🚀

작성자: AI Agent Development Team
최종 수정: 2026-01-02
버전: 1.0
라이선스: MIT

피드백: 이 가이드에 대한 피드백이나 질문이 있다면 GitHub Issues를 통해 공유해주세요.

작성일: 2026-01-02

함께 읽을 문서: 7계층 아키텍처 상세 분석

AI, Guide

왜 손코딩인가?

손코딩의 가치

이 가이드의 접근법

누구를 위한 가이드인가

학습 시간과 노력

학습 후 Claude Code 사용

이 가이드에 대하여

이 가이드가 제공하는 것

선행 요구사항

타이핑 속도와 시간

학습 전략

주의사항

목차

1부: 준비

2부: 구현

3부: 최적화

4부: 운영

1. 환경 설정

1.1 Python 3.13 설치

설치 과정

PATH 설정 확인

가상환경 도구 설치

1.2 Git 설치와 설정

설치

초기 설정

.gitignore 기본 설정

1.3 Docker Desktop 설치

WSL 2 활성화 (Windows 10/11)

Docker Desktop 설치

Docker 네트워크 이해

1.4 VSCode 및 필수 확장 프로그램

VSCode 설치

필수 확장 프로그램

VSCode 설정 최적화

1.5 Claude Code API 키 설정

API 키 발급

VSCode에서 설정

요금제 이해

테스트

2. 기술 스택 이해

2.1 FastAPI - 웹 프레임워크

FastAPI의 장점

대안과의 비교

2.2 LangGraph - 멀티 에이전트 프레임워크

왜 직접 LLM을 호출하지 않는가?

StateGraph의 개념

Checkpoint의 중요성

2.3 Mem0 - 장기 기억 시스템

대화 히스토리 vs 장기 기억

Mem0의 작동 원리

벡터 검색의 마법

2.4 PostgreSQL + pgvector - 데이터베이스

왜 PostgreSQL인가?

pgvector의 역할

대안과의 비교

2.5 Docker Compose - 인프라 오케스트레이션

수동 설치의 문제점

Docker Compose의 이점

3. 프로젝트 구조 설계

3.1 디렉토리 구조

3.2 7계층 아키텍처

3.3 의존성 관리

4. Layer 1-2: 기반 설정

4.1 프로젝트 초기화

4.2 환경 변수 설정

4.3 설정 관리 (config.py)

4.4 데이터베이스 스키마 설계

4.5 데이터 모델 (SQLModel)

5. Layer 3-4: 데이터와 서비스

5.1 JWT 인증

5.2 Rate Limiting

5.3 Input Sanitization

5.4 Database Service

5.5 LLM Service (폴백 전략 포함)

6. Layer 5-6: AI 에이전트

6.1 시스템 프롬프트 설계

6.2 검색 도구 구현

6.3 LangGraph Agent 구현 (핵심)

6.4 그래프 흐름 상세 설명

7. Layer 7: API와 배포