Claude Code를 이용하여 4개의 개성을 가진 AI 채팅 구현하기

# app/firebase.py
import firebase_admin
from firebase_admin import auth, credentials
from app.config import get_settings
 
settings = get_settings()
_firebase_app = None
 
def get_firebase_app():
    global _firebase_app
    if _firebase_app is None:
        cred = credentials.Certificate(settings.firebase_credentials_path)
        _firebase_app = firebase_admin.initialize_app(cred)
    return _firebase_app
 
def verify_firebase_token(id_token: str) -> dict:
    """Firebase ID 토큰을 검증하고 사용자 정보를 반환합니다."""
    get_firebase_app()
    decoded_token = auth.verify_id_token(id_token)
    return decoded_token

# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from app.firebase import verify_firebase_token
 
security = HTTPBearer()
 
class FirebaseUser(BaseModel):
    uid: str
    email: str | None = None
    name: str | None = None
 
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> FirebaseUser:
    """Firebase 토큰을 검증하고 현재 사용자를 반환합니다."""
    token = credentials.credentials
 
    try:
        decoded_token = verify_firebase_token(token)
        return FirebaseUser(
            uid=decoded_token["uid"],
            email=decoded_token.get("email"),
            name=decoded_token.get("name")
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid authentication: {str(e)}",
            headers={"WWW-Authenticate": "Bearer"},
        )

# app/schemas/ai.py
from pydantic import BaseModel, Field
 
class ChatRequest(BaseModel):
    prompt: str = Field(
        ...,
        min_length=1,
        max_length=10000,
        description="AI에게 보낼 메시지"
    )
    timeout_seconds: int | None = Field(
        default=None,
        ge=10,
        le=300,
        description="응답 대기 시간 (초)"
    )
 
class ChatResponse(BaseModel):
    response: str = Field(..., description="AI 응답")
    elapsed_time_ms: int = Field(..., description="처리 시간 (밀리초)")
    truncated: bool = Field(default=False, description="응답 잘림 여부")
    persona: str | None = Field(
        default=None,
        description="응답한 AI 캐릭터 (말랑이/루팡/푸딩/마이콜)"
    )

# app/routers/ai.py
from fastapi import APIRouter, Depends, HTTPException, status
from app.dependencies import get_current_user, FirebaseUser
from app.schemas.ai import ChatRequest, ChatResponse
from app.services.claude_service import ClaudeService, get_claude_service
 
router = APIRouter(prefix="/ai", tags=["ai"])
 
@router.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    user: FirebaseUser = Depends(get_current_user),
    claude_service: ClaudeService = Depends(get_claude_service),
):
    """AI와 대화합니다. Firebase 인증이 필요합니다."""
    try:
        result = await claude_service.chat(
            prompt=request.prompt,
            timeout_seconds=request.timeout_seconds
        )
        return ChatResponse(
            response=result.output,
            elapsed_time_ms=result.elapsed_ms,
            truncated=result.truncated,
            persona=result.persona_name  # AI 캐릭터 이름 반환
        )
    except ValueError as e:
        # AI 트리거가 감지되지 않은 경우
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )
    except TimeoutError:
        raise HTTPException(
            status_code=status.HTTP_408_REQUEST_TIMEOUT,
            detail="AI response timed out"
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"AI service error: {str(e)}"
        )

# app/services/personas.py
from enum import Enum
from dataclasses import dataclass
 
class PersonaType(str, Enum):
    MALLANGI = "mallangi"   # 말랑이
    LUPIN = "lupin"         # 루팡
    PUDDING = "pudding"     # 푸딩
    MICHAEL = "michael"     # 마이콜
 
@dataclass
class Persona:
    type: PersonaType
    name: str               # 한글 이름
    triggers: list[str]     # 호출 트리거
    system_prompt: str      # 캐릭터 프롬프트
 
# 페르소나 정의
PERSONAS = {
    PersonaType.MALLANGI: Persona(
        type=PersonaType.MALLANGI,
        name="말랑이",
        triggers=["말랑아", "말랑이야"],
        system_prompt="""너는 '말랑이'야. 다정하고 따뜻한 친구야.
        - 항상 공감하고 칭찬해줘
        - 부드럽고 친근한 말투를 사용해
        - "~해줄게", "~하자" 같은 표현을 사용해
        - 이모지를 적절히 사용해 (💕, 🤗, ✨)"""
    ),
    PersonaType.LUPIN: Persona(
        type=PersonaType.LUPIN,
        name="루팡",
        triggers=["루팡아", "루팡이야"],
        system_prompt="""너는 '루팡'이야. 자신감 넘치는 친구야.
        - 약간 건방지지만 실력이 있어
        - "흥, 그 정도는 쉽지~" 같은 말투
        - 도움을 줄 때도 쿨하게
        - 가끔 유머도 섞어서"""
    ),
    PersonaType.PUDDING: Persona(
        type=PersonaType.PUDDING,
        name="푸딩",
        triggers=["푸딩아", "푸딩이야"],
        system_prompt="""너는 '푸딩'이야. 귀여운 애완동물이야.
        - 의성어를 많이 사용해 (멍멍, 왈왈, 낑낑)
        - 단순하고 순수하게 반응해
        - 꼬리 흔들며 기뻐하는 느낌으로
        - 짧고 귀엽게 대답해"""
    ),
    PersonaType.MICHAEL: Persona(
        type=PersonaType.MICHAEL,
        name="마이콜",
        triggers=["마이콜아", "마이콜이야"],
        system_prompt="""너는 '마이콜'이야. 친절한 영어 선생님이야.
        - 한국어와 영어를 섞어서 대답해
        - 영어 표현을 가르쳐줄 때 발음도 알려줘
        - "In English, we say..." 같은 표현 사용
        - 재미있게 영어를 알려줘"""
    ),
}

def detect_persona(message: str) -> tuple[PersonaType, str]:
    """메시지에서 페르소나 트리거를 감지합니다.
 
    Returns:
        tuple[PersonaType, str]: (페르소나 타입, 트리거 제거된 실제 질문)
 
    Raises:
        ValueError: 트리거가 감지되지 않은 경우
    """
    message_lower = message.strip().lower()
 
    for persona_type, persona in PERSONAS.items():
        for trigger in persona.triggers:
            if message_lower.startswith(trigger.lower()):
                # 트리거 제거하고 실제 질문만 추출
                actual_prompt = message[len(trigger):].strip()
                return persona_type, actual_prompt
 
    raise ValueError("no_trigger: AI 캐릭터 호출이 감지되지 않았습니다")

BASE_INSTRUCTIONS = """
다음 규칙을 반드시 따르세요:
- 파일을 절대 수정하지 않습니다
- 100단어 이내로 간결하게 대답합니다
- 최신 정보가 필요하면 웹검색을 활용합니다
- 아이들에게 적합한 언어를 사용합니다
"""
 
def get_persona_prompt(persona_type: PersonaType, user_message: str) -> str:
    """캐릭터별 시스템 프롬프트와 사용자 메시지를 조합합니다."""
    persona = PERSONAS[persona_type]
    return f"""{persona.system_prompt}
 
{BASE_INSTRUCTIONS}
 
사용자 메시지: {user_message}
"""

# app/services/claude_service.py
import asyncio
import time
from dataclasses import dataclass
from app.config import get_settings
from app.services.personas import detect_persona, get_persona_prompt, PERSONAS
 
settings = get_settings()
 
@dataclass
class ClaudeResponse:
    output: str
    elapsed_ms: int
    truncated: bool = False
    persona_name: str | None = None  # AI 캐릭터 이름
 
class ClaudeService:
    def __init__(self):
        self.cli_path = settings.claude_cli_path
        self.default_timeout = settings.claude_timeout_seconds
        self.max_timeout = settings.claude_max_timeout_seconds
 
    async def chat(
        self,
        prompt: str,
        timeout_seconds: int | None = None
    ) -> ClaudeResponse:
        # 1. 페르소나 감지 (없으면 ValueError 발생)
        persona_type, actual_prompt = detect_persona(prompt)
        persona = PERSONAS[persona_type]
 
        # 2. 캐릭터별 프롬프트 생성
        full_prompt = get_persona_prompt(persona_type, actual_prompt)
 
        timeout = min(
            timeout_seconds or self.default_timeout,
            self.max_timeout
        )
 
        cmd = [
            self.cli_path,
            "--dangerously-skip-permissions",
            "-p",
            full_prompt
        ]
 
        start_time = time.time()
 
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.DEVNULL,  # systemd 환경에서 필수!
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
 
        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            process.kill()
            raise TimeoutError(f"Claude CLI timed out after {timeout}s")
 
        elapsed_ms = int((time.time() - start_time) * 1000)
 
        if process.returncode != 0:
            error_msg = stderr.decode() if stderr else "Unknown error"
            raise RuntimeError(f"Claude CLI failed: {error_msg}")
 
        output = stdout.decode().strip()
 
        # 응답 길이 제한 (안전장치)
        MAX_LENGTH = 5000
        truncated = len(output) > MAX_LENGTH
        if truncated:
            output = output[:MAX_LENGTH] + "..."
 
        return ClaudeResponse(
            output=output,
            elapsed_ms=elapsed_ms,
            truncated=truncated,
            persona_name=persona.name  # 말랑이, 루팡, 푸딩, 마이콜
        )
 
# 의존성 주입용
def get_claude_service() -> ClaudeService:
    return ClaudeService()

# app/config.py
class Settings(BaseSettings):
    # ... 기존 설정 ...
 
    # Claude CLI
    claude_cli_path: str = "claude"
    claude_timeout_seconds: int = 120
    claude_max_timeout_seconds: int = 300

which claude && claude --version
# /home/funq/.local/bin/claude
# 2.0.75 (Claude Code)

# /etc/systemd/system/backend-api.service
Environment="PATH=/home/funq/dev/backend-api/venv/bin"

echo 'CLAUDE_CLI_PATH=/home/funq/.local/bin/claude' >> ~/backend-api/.env
sudo systemctl restart backend-api

class Settings(BaseSettings):
    claude_cli_path: str = "claude"  # .env의 CLAUDE_CLI_PATH로 오버라이드됨

timeout 30 claude -p "안녕" --dangerously-skip-permissions
# 성공! "안녕하세요! FastAPI 백엔드 프로젝트에서 무엇을 도와드릴까요?"

sudo cat /etc/systemd/system/backend-api.service

[Service]
User=funq
Group=funq
WorkingDirectory=/home/funq/dev/backend-api
Environment="PATH=/home/funq/dev/backend-api/venv/bin"
ExecStart=/home/funq/dev/backend-api/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000

timeout 30 setsid claude -p "안녕" --dangerously-skip-permissions </dev/null 2>&1
# 성공!

# app/services/claude_service.py
process = await asyncio.create_subprocess_exec(
    *cmd,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    # stdin이 없음!
)

[Service]
User=funq
Group=funq
WorkingDirectory=/home/funq/dev/backend-api
Environment="PATH=/home/funq/dev/backend-api/venv/bin:/home/funq/.local/bin"
Environment="HOME=/home/funq"
ExecStart=/home/funq/dev/backend-api/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000
Restart=always
RestartSec=3

sudo systemctl daemon-reload
sudo systemctl restart backend-api

# app/services/claude_service.py
process = await asyncio.create_subprocess_exec(
    *cmd,
    stdin=asyncio.subprocess.DEVNULL,  # 추가!
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
)