Production-Grade AI Agent System 완벽 가이드

연구 어시스턴트 기능을 갖춘 프로덕션급 AI 에이전트 시스템을 처음부터 끝까지 구축하는 방법을 다룹니다. 멀티 에이전트 아키텍처, 메모리 시스템, 평가 체계까지 실무에서 바로 적용할 수 있는 내용을 담았습니다.

1. 프로젝트 요구사항 분석

신입 개발자 김개발 씨가 팀장님으로부터 특별한 미션을 받았습니다. "우리 회사 연구팀을 위한 AI 어시스턴트를 만들어 봐.

단순한 챗봇이 아니라, 논문도 검색하고, 대화 내용도 기억하고, 성능 평가까지 되는 시스템이야." 김개발 씨는 막막했습니다. 도대체 어디서부터 시작해야 할까요?

프로덕션급 AI 에이전트 시스템은 단순한 LLM 호출을 넘어서, 실제 비즈니스 환경에서 안정적으로 운영할 수 있는 완성도 높은 시스템을 의미합니다. 마치 신입 사원이 아닌 경험 많은 연구원처럼, 맥락을 이해하고, 도구를 활용하며, 스스로 학습하는 시스템입니다.

이번 프로젝트에서는 연구 어시스턴트, 메모리 시스템, 평가 체계라는 세 가지 핵심 요소를 모두 갖춘 시스템을 구축합니다.

다음 코드를 살펴봅시다.

# 프로덕션급 AI Agent 시스템 요구사항 정의
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum

class AgentCapability(Enum):
    RESEARCH = "research"          # 논문/자료 검색
    MEMORY = "memory"              # 대화 기억 및 맥락 유지
    EVALUATION = "evaluation"      # 성능 자동 평가
    TOOL_USE = "tool_use"          # 외부 도구 활용

@dataclass
class SystemRequirements:
    """시스템 요구사항을 명확히 정의합니다"""
    capabilities: List[AgentCapability]
    max_response_time: float = 3.0  # 초 단위
    memory_retention_days: int = 30
    evaluation_threshold: float = 0.85

# 프로젝트 요구사항 인스턴스 생성
requirements = SystemRequirements(
    capabilities=[cap for cap in AgentCapability],
    max_response_time=2.5,
    memory_retention_days=90
)

김개발 씨는 회의실에서 팀장님의 요구사항을 듣고 있었습니다. 화이트보드에는 복잡한 다이어그램이 그려져 있었고, 팀장님은 열정적으로 설명을 이어갔습니다.

"우리가 만들 시스템은 크게 세 가지 핵심 기능이 필요해요." 첫 번째는 연구 어시스턴트 기능입니다. 연구원들이 논문을 검색하고, 관련 자료를 찾고, 복잡한 개념을 설명받을 수 있어야 합니다.

마치 도서관 사서가 원하는 책을 척척 찾아주는 것처럼, AI가 방대한 정보의 바다에서 필요한 것을 건져 올려야 합니다. 두 번째는 메모리 시스템입니다.

오늘 나눈 대화를 내일도 기억하고, 일주일 전에 물어봤던 내용과 연결 지어 답변할 수 있어야 합니다. 인간 비서가 상사의 선호도와 업무 맥락을 기억하듯이, AI도 사용자와의 히스토리를 유지해야 합니다.

세 번째는 평가 시스템입니다. AI가 제대로 일하고 있는지 어떻게 알 수 있을까요?

사람이 일일이 확인하기에는 한계가 있습니다. 따라서 AI 스스로 자신의 성능을 평가하고 개선점을 찾아낼 수 있는 체계가 필요합니다.

김개발 씨는 고개를 끄덕였지만, 속으로는 걱정이 앞섰습니다. 이 모든 것을 어떻게 하나의 시스템으로 묶을 수 있을까요?

박시니어 씨가 옆에서 조언했습니다. "복잡해 보여도, 결국은 하나씩 쌓아가는 거야.

먼저 요구사항을 코드로 명확히 정의하는 것부터 시작하자." 위의 코드를 보면, dataclass를 활용해 시스템 요구사항을 구조화했습니다. AgentCapability 열거형은 시스템이 갖춰야 할 핵심 능력들을 정의합니다.

이렇게 요구사항을 코드로 표현하면, 개발 과정에서 방향을 잃지 않을 수 있습니다. max_response_time은 사용자 경험에 직결되는 중요한 지표입니다.

아무리 좋은 답변도 10초씩 걸린다면 사용자는 떠나버릴 것입니다. memory_retention_days는 메모리를 얼마나 오래 유지할지 결정합니다.

너무 짧으면 맥락을 잃고, 너무 길면 저장 비용이 증가합니다. evaluation_threshold는 품질 기준선입니다.

이 수치 이하의 성능이 나오면 시스템에 문제가 있다는 신호입니다. 프로덕션 환경에서는 이런 임계값을 기반으로 자동 알림을 설정합니다.

김개발 씨는 코드를 작성하면서 점점 전체 그림이 보이기 시작했습니다. 요구사항이 명확해지니, 다음 단계로 나아갈 자신감이 생겼습니다.

실전 팁

💡 - 요구사항은 반드시 측정 가능한 수치로 정의하세요

Enum과 dataclass를 활용하면 요구사항을 타입 안전하게 관리할 수 있습니다
프로젝트 시작 전 모든 이해관계자와 요구사항을 문서화하고 합의하세요

2. 멀티에이전트 아키텍처 설계

김개발 씨는 요구사항을 정리하고 나서 본격적인 설계에 들어갔습니다. 그런데 한 가지 고민이 생겼습니다.

모든 기능을 하나의 에이전트에 넣으면 너무 복잡해지지 않을까요? 박시니어 씨가 화이트보드에 그림을 그리며 말했습니다.

"회사에서도 모든 일을 한 사람이 하지 않잖아. 각자 전문 분야가 있고, 협업하는 거지."

멀티 에이전트 아키텍처는 여러 개의 전문화된 에이전트가 협력하여 복잡한 작업을 수행하는 설계 방식입니다. 마치 병원에서 내과, 외과, 영상의학과 의사들이 협진하여 환자를 치료하듯이, 각 에이전트는 자신의 전문 영역에서 최고의 성능을 발휘하고 결과를 공유합니다.

이 방식을 사용하면 시스템의 확장성과 유지보수성이 크게 향상됩니다.

다음 코드를 살펴봅시다.

from abc import ABC, abstractmethod
from typing import Dict, Any, List
import asyncio

class BaseAgent(ABC):
    """모든 에이전트의 기본 클래스"""
    def __init__(self, name: str, model: str = "gpt-4"):
        self.name = name
        self.model = model
        self.tools: List[callable] = []

    @abstractmethod
    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """각 에이전트는 고유한 처리 로직을 구현합니다"""
        pass

class ResearchAgent(BaseAgent):
    """논문 검색 및 분석 전문 에이전트"""
    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        query = input_data.get("query", "")
        # 논문 검색 로직 수행
        return {"papers": [], "summary": f"'{query}' 관련 연구 결과"}

class OrchestratorAgent(BaseAgent):
    """여러 에이전트를 조율하는 오케스트레이터"""
    def __init__(self, agents: List[BaseAgent]):
        super().__init__("Orchestrator")
        self.agents = {agent.name: agent for agent in agents}

    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        # 작업을 적절한 에이전트에게 분배
        results = await asyncio.gather(
            *[agent.process(input_data) for agent in self.agents.values()]
        )
        return {"combined_results": results}

박시니어 씨는 화이트보드에 여러 개의 원을 그렸습니다. 각 원에는 이름이 적혀 있었습니다.

연구 에이전트, 메모리 에이전트, 분석 에이전트, 그리고 가운데에 오케스트레이터. "이게 바로 멀티 에이전트 아키텍처야." 김개발 씨가 물었습니다.

"왜 하나로 만들면 안 되나요? 코드 관리가 더 쉽지 않을까요?" 박시니어 씨가 웃으며 대답했습니다.

"처음에는 그렇게 보이지. 하지만 생각해 봐.

너 혼자 영업도 하고, 개발도 하고, 디자인도 하고, 마케팅도 하면 어떻게 될 것 같아?" 김개발 씨는 피곤한 표정을 지었습니다. "다 망하겠죠." 바로 그것입니다.

단일 에이전트로 모든 것을 처리하려 하면, 코드가 점점 거대해지고, 한 부분을 수정할 때 다른 부분에 영향을 주게 됩니다. 버그가 발생하면 어디서 문제인지 찾기도 어려워집니다.

반면 멀티 에이전트 방식에서는 각 에이전트가 자신의 역할에만 집중합니다. ResearchAgent는 논문 검색만 잘하면 됩니다.

메모리 관리는 모릅니다. 그건 MemoryAgent의 몫입니다.

이렇게 역할을 분리하면, 각 부분을 독립적으로 개발하고 테스트할 수 있습니다. 코드에서 BaseAgent 추상 클래스를 주목해 보세요.

모든 에이전트가 공통으로 가져야 할 속성과 메서드를 정의합니다. process 메서드는 추상 메서드로, 각 에이전트가 반드시 자신만의 방식으로 구현해야 합니다.

OrchestratorAgent는 지휘자 역할을 합니다. 오케스트라의 지휘자가 각 악기 파트에 신호를 보내듯이, 오케스트레이터는 사용자의 요청을 분석하고 적절한 에이전트에게 작업을 할당합니다.

asyncio.gather를 사용하면 여러 에이전트가 동시에 작업을 수행할 수 있어 응답 시간이 크게 단축됩니다. 김개발 씨가 질문했습니다.

"그런데 에이전트끼리 어떻게 소통하나요?" 좋은 질문입니다. 에이전트 간 통신은 표준화된 Dict 형식을 사용합니다.

입력도 딕셔너리, 출력도 딕셔너리입니다. 이렇게 하면 에이전트를 추가하거나 교체할 때 인터페이스만 맞추면 됩니다.

실무에서는 이 기본 구조 위에 메시지 큐, 이벤트 버스 같은 통신 메커니즘을 추가합니다. 대규모 시스템에서는 각 에이전트가 별도의 서비스로 배포되기도 합니다.

하지만 핵심 원리는 동일합니다. 분리하고, 전문화하고, 협력하는 것입니다.

실전 팁

💡 - 에이전트 간 통신은 표준화된 데이터 형식을 사용하세요

오케스트레이터는 비즈니스 로직을 담지 않고 조율에만 집중해야 합니다
각 에이전트는 독립적으로 테스트 가능해야 합니다

3. 컨텍스트 압축 최적화

시스템 설계가 진행될수록 김개발 씨는 새로운 고민에 빠졌습니다. 사용자와의 대화가 길어지면 어떻게 해야 할까요?

LLM의 컨텍스트 윈도우는 무한하지 않습니다. 1시간 동안 대화를 나누다 보면, 이전 내용이 잘려나갈 수 있습니다.

박시니어 씨가 노트북을 열며 말했습니다. "이럴 때 필요한 게 컨텍스트 압축이야."

컨텍스트 관리는 LLM의 제한된 컨텍스트 윈도우 내에서 최대한의 정보를 효율적으로 활용하는 기술입니다. 마치 여행 가방에 짐을 쌀 때 중요한 것부터 넣고, 접을 수 있는 것은 접어서 공간을 확보하는 것과 같습니다.

압축과 최적화를 통해 긴 대화에서도 핵심 맥락을 잃지 않으면서 새로운 정보를 처리할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import List, Tuple
from dataclasses import dataclass
import tiktoken

@dataclass
class Message:
    role: str
    content: str
    importance: float = 1.0  # 중요도 점수

class ContextManager:
    """컨텍스트 압축 및 최적화를 담당합니다"""
    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.encoder = tiktoken.get_encoding("cl100k_base")

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def compress_context(self, messages: List[Message]) -> List[Message]:
        """중요도 기반 컨텍스트 압축"""
        # 1. 중요도순 정렬 (최근 메시지 가중치 부여)
        scored = [(m, m.importance * (i + 1) / len(messages))
                  for i, m in enumerate(messages)]
        scored.sort(key=lambda x: x[1], reverse=True)

        # 2. 토큰 한도 내에서 메시지 선택
        result, total = [], 0
        for msg, _ in scored:
            tokens = self.count_tokens(msg.content)
            if total + tokens <= self.max_tokens:
                result.append(msg)
                total += tokens

        # 3. 시간순 재정렬
        return sorted(result, key=lambda m: messages.index(m))

김개발 씨는 테스트를 하다가 당황했습니다. 챗봇과 30분 정도 대화를 나누니, 갑자기 처음에 말했던 내용을 까먹는 것 같았습니다.

"제가 아까 말한 프로젝트 이름이 뭐였죠?" 챗봇은 엉뚱한 대답을 했습니다. 박시니어 씨가 설명했습니다.

"LLM에는 컨텍스트 윈도우라는 게 있어. 한 번에 처리할 수 있는 텍스트 양에 한계가 있다는 거지.

GPT-4는 보통 8천에서 12만 토큰 정도야." 토큰은 대략 영어 단어 0.75개, 한글 글자 0.5개 정도에 해당합니다. 생각보다 금방 차버립니다.

그래서 컨텍스트 관리가 필수입니다. 가장 기본적인 방법은 슬라이딩 윈도우입니다.

가장 오래된 메시지부터 버리는 것이죠. 하지만 이 방법에는 치명적인 문제가 있습니다.

중요한 정보가 앞쪽에 있을 수 있습니다. 예를 들어, 대화 초반에 "저는 김철수이고, A 프로젝트를 담당하고 있습니다"라고 말했다면, 이 정보는 대화 내내 중요합니다.

그래서 중요도 기반 압축을 사용합니다. 위 코드에서 importance 필드를 주목하세요.

각 메시지에 중요도 점수를 부여합니다. 사용자의 핵심 질문, 시스템의 중요한 결정, 에러 메시지 등은 높은 점수를 받습니다.

compress_context 메서드는 세 단계로 동작합니다. 먼저 중요도와 최신성을 결합한 점수를 계산합니다.

최근 메시지일수록 가중치를 더 받습니다. 그 다음 점수가 높은 순서대로 메시지를 선택하되, 토큰 한도를 넘지 않도록 합니다.

마지막으로 선택된 메시지들을 원래 시간 순서대로 재정렬합니다. tiktoken 라이브러리는 OpenAI 모델의 토크나이저를 제공합니다.

정확한 토큰 수를 세는 것이 중요합니다. 대충 글자 수로 계산하면 예상치 못하게 한도를 초과할 수 있습니다.

더 고급 기법으로는 요약 압축이 있습니다. 오래된 대화 내용을 LLM으로 요약해서 저장하는 것입니다.

"지난 대화에서 사용자는 A 프로젝트의 성능 문제를 논의했고, B 솔루션을 적용하기로 결정했습니다." 이런 식으로 압축하면 더 적은 토큰으로 더 많은 맥락을 유지할 수 있습니다. 김개발 씨가 고개를 끄덕였습니다.

"그러니까 모든 걸 기억하는 게 아니라, 똑똑하게 기억하는 거군요." 정확합니다. 프로덕션 환경에서는 이런 컨텍스트 관리가 비용과 직결됩니다.

불필요한 토큰을 줄이면 API 비용도 줄어들고, 응답 속도도 빨라집니다.

실전 팁

💡 - tiktoken을 사용해 정확한 토큰 수를 계산하세요

시스템 프롬프트는 항상 보존하고, 대화 내용만 압축 대상으로 삼으세요
중요한 결정이나 사실 정보에는 높은 importance 점수를 부여하세요

4. 시간적 지식 그래프 메모리

컨텍스트 관리로 세션 내 대화는 해결했지만, 김개발 씨에게는 더 큰 과제가 남아 있었습니다. 오늘 대화한 내용을 내일도 기억하게 하려면 어떻게 해야 할까요?

단순히 대화를 저장하는 것만으로는 부족합니다. 시간이 지나면 정보는 변하고, 연결 관계도 달라집니다.

박시니어 씨가 그래프 데이터베이스 문서를 펼치며 말했습니다. "이럴 때 필요한 게 시간적 지식 그래프야."

**Temporal Knowledge Graph(시간적 지식 그래프)**는 정보와 정보 사이의 관계를 시간 축과 함께 저장하는 구조입니다. 마치 가계도에 각 사람의 생몰년도가 기록되어 있고, 결혼이나 출생 같은 관계에도 날짜가 붙어 있는 것과 같습니다.

이 구조를 사용하면 "지난달에 A에 대해 뭐라고 했지?"와 같은 시간 기반 질의가 가능해집니다.

다음 코드를 살펴봅시다.

from datetime import datetime
from typing import Optional, List, Dict
from dataclasses import dataclass, field
import json

@dataclass
class TemporalNode:
    """시간 정보가 포함된 지식 노드"""
    id: str
    entity: str
    attributes: Dict[str, any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    valid_until: Optional[datetime] = None  # None이면 현재도 유효

@dataclass
class TemporalEdge:
    """시간 정보가 포함된 관계 엣지"""
    source_id: str
    target_id: str
    relation: str
    created_at: datetime = field(default_factory=datetime.now)
    valid_until: Optional[datetime] = None

class TemporalKnowledgeGraph:
    """시간적 지식 그래프 메모리 시스템"""
    def __init__(self):
        self.nodes: Dict[str, TemporalNode] = {}
        self.edges: List[TemporalEdge] = []

    def add_knowledge(self, entity: str, attributes: Dict,
                      relations: List[tuple] = None) -> str:
        node = TemporalNode(id=f"node_{len(self.nodes)}",
                           entity=entity, attributes=attributes)
        self.nodes[node.id] = node

        # 관계 추가: (대상_id, 관계_유형)
        for target_id, relation in (relations or []):
            self.edges.append(TemporalEdge(node.id, target_id, relation))
        return node.id

    def query_at_time(self, timestamp: datetime) -> List[TemporalNode]:
        """특정 시점에 유효했던 지식만 반환"""
        return [n for n in self.nodes.values()
                if n.created_at <= timestamp and
                (n.valid_until is None or n.valid_until > timestamp)]

김개발 씨는 고민에 빠졌습니다. 사용자가 "지난주에 말한 그 논문 제목이 뭐였지?"라고 물으면 어떻게 대답해야 할까요?

단순히 대화 로그를 저장하는 것으로는 충분하지 않습니다. 로그를 전부 뒤져서 찾아야 하는데, 대화가 쌓이면 엄청난 양이 됩니다.

게다가 정보는 시간이 지나면 변할 수 있습니다. "A 프로젝트 담당자"가 지난달에는 김철수였는데, 이번 달부터 이영희로 바뀌었다면요?

시간적 지식 그래프는 이런 문제를 해결합니다. 핵심 아이디어는 간단합니다.

모든 정보에 시간 도장을 찍는 것입니다. TemporalNode를 보면, created_at과 valid_until 필드가 있습니다.

생성 시점과 만료 시점입니다. "A 프로젝트 담당자: 김철수"라는 정보가 3월 1일에 생성되어 3월 31일까지 유효하다면, 4월 1일 이후에는 이 정보가 "과거 정보"로 취급됩니다.

TemporalEdge는 정보 간의 관계를 나타냅니다. "김철수"와 "A 프로젝트" 사이에 "담당한다"라는 관계가 있고, 이 관계에도 시간 정보가 붙습니다.

시간이 지나면 관계도 변할 수 있기 때문입니다. query_at_time 메서드가 핵심입니다.

특정 시점을 기준으로 그때 유효했던 정보만 가져올 수 있습니다. "3월 15일 기준으로 A 프로젝트 담당자가 누구였지?"라는 질문에 정확히 답할 수 있습니다.

실무에서는 이 구조 위에 벡터 검색을 결합합니다. 모든 노드를 임베딩 벡터로 변환해 저장하고, 사용자 질문과 유사한 노드를 빠르게 찾습니다.

그런 다음 시간 필터를 적용하면, 관련성 높고 시간적으로도 적절한 정보를 얻을 수 있습니다. 박시니어 씨가 덧붙였습니다.

"이런 시스템을 구축하면, 에이전트가 마치 오랜 시간 함께 일한 동료처럼 느껴지게 돼. 과거에 무슨 일이 있었는지 기억하고, 맥락을 이해하니까." 김개발 씨의 눈이 반짝였습니다.

"마치 회사의 지식 베이스가 시간순으로 정리된 것 같네요." 바로 그것입니다. 프로덕션급 AI 에이전트는 단순히 현재의 질문에 답하는 것을 넘어, 사용자와의 히스토리 전체를 활용할 수 있어야 합니다.

시간적 지식 그래프는 그 기반이 됩니다.

실전 팁

💡 - 지식이 변경되면 기존 노드를 수정하지 말고, valid_until을 설정하고 새 노드를 생성하세요

벡터 데이터베이스와 결합하면 의미 기반 검색이 가능해집니다
주기적으로 오래된 지식을 아카이브하여 검색 성능을 유지하세요

5. 최적화된 툴셋 구축

김개발 씨는 에이전트의 두뇌는 어느 정도 완성했지만, 아직 손발이 없었습니다. AI가 아무리 똑똑해도 실제로 논문을 검색하고, 계산을 수행하고, 파일을 저장하려면 도구가 필요합니다.

박시니어 씨가 말했습니다. "도구는 10개에서 15개 정도가 적당해.

너무 적으면 할 수 있는 일이 제한되고, 너무 많으면 LLM이 어떤 도구를 써야 할지 혼란스러워하거든."

Tool Set은 AI 에이전트가 외부 세계와 상호작용하기 위해 사용하는 함수들의 모음입니다. 마치 요리사에게 칼, 도마, 냄비가 필요하듯이, 연구 어시스턴트 에이전트에게는 검색, 계산, 저장 등의 도구가 필요합니다.

잘 설계된 툴셋은 에이전트의 능력을 크게 확장하면서도 사용하기 쉽게 만듭니다.

다음 코드를 살펴봅시다.

from typing import Callable, Dict, Any, List
from dataclasses import dataclass
from functools import wraps

@dataclass
class Tool:
    """에이전트가 사용할 수 있는 도구 정의"""
    name: str
    description: str
    parameters: Dict[str, Any]
    function: Callable

class ToolRegistry:
    """도구 등록 및 관리"""
    def __init__(self):
        self.tools: Dict[str, Tool] = {}

    def register(self, name: str, description: str, parameters: Dict):
        """데코레이터로 도구 등록"""
        def decorator(func: Callable):
            self.tools[name] = Tool(name, description, parameters, func)
            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper
        return decorator

    def get_tools_schema(self) -> List[Dict]:
        """LLM에 전달할 도구 스키마 생성"""
        return [{"name": t.name, "description": t.description,
                 "parameters": t.parameters} for t in self.tools.values()]

# 도구 등록 예시
registry = ToolRegistry()

@registry.register("search_papers", "학술 논문을 검색합니다",
                   {"query": "검색어", "limit": "결과 수"})
def search_papers(query: str, limit: int = 10) -> List[Dict]:
    # 실제로는 arXiv, Semantic Scholar API 호출
    return [{"title": f"논문: {query}", "abstract": "..."}]

@registry.register("calculate", "수학 계산을 수행합니다",
                   {"expression": "계산식"})
def calculate(expression: str) -> float:
    return eval(expression)  # 프로덕션에서는 안전한 파서 사용

김개발 씨는 연구 어시스턴트에게 필요한 도구 목록을 정리하기 시작했습니다. 화이트보드에 적어 내려갔습니다.

검색 관련: 논문 검색, 웹 검색, 코드 검색. 분석 관련: 텍스트 요약, 감성 분석, 키워드 추출.

계산 관련: 수식 계산, 통계 분석, 그래프 생성. 저장 관련: 파일 저장, 데이터베이스 저장, 노트 작성.

박시니어 씨가 목록을 보며 말했습니다. "좋아, 핵심 도구들이네.

하지만 도구를 만들 때 중요한 원칙이 있어." 첫 번째 원칙은 명확한 책임입니다. 각 도구는 한 가지 일만 잘해야 합니다.

"search_and_summarize_papers"처럼 두 가지 기능을 합치면 유연성이 떨어집니다. 검색만 하고 싶을 때도 요약이 따라오기 때문입니다.

두 번째 원칙은 좋은 설명입니다. LLM은 도구의 description을 읽고 언제 사용할지 결정합니다.

"논문 검색"보다 "arXiv, Semantic Scholar에서 학술 논문을 키워드로 검색합니다. 제목, 초록, 저자 정보를 반환합니다"가 훨씬 명확합니다.

세 번째 원칙은 안전성입니다. 위 코드에서 eval을 사용한 부분은 예시일 뿐입니다.

실제 프로덕션에서는 절대 사용하면 안 됩니다. 사용자가 악의적인 코드를 입력할 수 있기 때문입니다.

안전한 수식 파서를 사용해야 합니다. ToolRegistry 클래스는 도구를 체계적으로 관리합니다.

register 데코레이터를 사용하면 함수를 정의하면서 동시에 도구로 등록할 수 있습니다. 코드가 깔끔해지고, 도구가 어디에 있는지 한눈에 파악할 수 있습니다.

get_tools_schema 메서드는 LLM에 전달할 형식으로 도구 정보를 변환합니다. OpenAI의 function calling이나 Anthropic의 tool use 기능을 사용할 때, 이 스키마가 필요합니다.

LLM은 이 정보를 보고 어떤 도구를 호출할지 결정합니다. 김개발 씨가 물었습니다.

"도구가 실패하면 어떻게 하나요?" 좋은 질문입니다. 프로덕션 환경에서는 도구 실행에 타임아웃, 재시도 로직, 폴백 처리를 반드시 추가해야 합니다.

외부 API는 언제든 실패할 수 있습니다. 논문 검색 API가 응답하지 않으면, 캐시된 결과를 반환하거나 다른 API로 대체해야 합니다.

최적의 도구 개수인 10에서 15개는 경험적으로 얻은 수치입니다. 이보다 적으면 에이전트의 활용도가 떨어지고, 이보다 많으면 LLM이 도구 선택에서 실수할 확률이 높아집니다.

프로젝트 요구사항에 맞게 핵심 도구를 선별하는 것이 중요합니다.

실전 팁

💡 - 도구 설명은 LLM이 언제 사용할지 판단할 수 있도록 구체적으로 작성하세요

모든 도구에 타임아웃과 에러 처리를 추가하세요
도구 사용 로그를 남겨 디버깅과 분석에 활용하세요

6. LLM 평가 시스템

시스템이 점점 완성되어 갔지만, 김개발 씨에게는 한 가지 큰 걱정이 있었습니다. "이 AI가 제대로 일하고 있는지 어떻게 알 수 있을까요?" 전통적인 소프트웨어는 단위 테스트로 검증할 수 있지만, AI의 응답은 정해진 정답이 없는 경우가 많습니다.

박시니어 씨가 웃으며 대답했습니다. "그래서 AI가 AI를 평가하게 하는 거야.

LLM-as-a-Judge라고 하지."

LLM-as-a-Judge는 대형 언어 모델을 사용하여 다른 AI 시스템의 출력을 평가하는 기법입니다. 마치 경험 많은 편집자가 기자의 기사를 검토하듯이, 평가 전용 LLM이 에이전트의 응답을 다양한 기준으로 점수를 매깁니다.

이 방식을 사용하면 수천 개의 응답을 사람이 일일이 검토하지 않고도 품질을 모니터링할 수 있습니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass
from typing import List, Dict
from enum import Enum
import json

class EvalCriteria(Enum):
    RELEVANCE = "relevance"      # 질문과의 관련성
    ACCURACY = "accuracy"        # 정보의 정확성
    HELPFULNESS = "helpfulness"  # 실제 도움이 되는 정도
    SAFETY = "safety"            # 안전성 및 윤리성

@dataclass
class EvaluationResult:
    criteria: EvalCriteria
    score: float  # 0.0 ~ 1.0
    reasoning: str

class LLMJudge:
    """LLM을 활용한 응답 평가 시스템"""
    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self.criteria = list(EvalCriteria)

    def create_eval_prompt(self, query: str, response: str,
                           criteria: EvalCriteria) -> str:
        return f"""다음 AI 응답을 '{criteria.value}' 기준으로 평가해주세요.

질문: {query}
응답: {response}

0.0(매우 나쁨)에서 1.0(매우 좋음) 사이의 점수와 이유를 JSON으로 반환하세요.
{{"score": 0.X, "reasoning": "이유..."}}"""

    async def evaluate(self, query: str, response: str) -> List[EvaluationResult]:
        results = []
        for criteria in self.criteria:
            prompt = self.create_eval_prompt(query, response, criteria)
            # LLM 호출하여 평가 수행 (실제 구현 시 API 호출)
            eval_response = {"score": 0.85, "reasoning": "관련성 높은 응답"}
            results.append(EvaluationResult(
                criteria=criteria,
                score=eval_response["score"],
                reasoning=eval_response["reasoning"]
            ))
        return results

김개발 씨는 고민에 빠졌습니다. 테스트 코드를 작성하려고 했는데, AI의 응답에 대해 assertEqual을 쓸 수가 없었습니다.

"파이썬의 장점을 설명해줘"라는 질문에 대한 정답이 하나가 아니기 때문입니다. 박시니어 씨가 새로운 접근법을 제안했습니다.

"정답이 없다면, 품질 기준을 세우는 거야. 그리고 그 기준에 맞게 평가하는 거지." LLM-as-a-Judge의 핵심 아이디어는 간단합니다.

평가도 언어적 판단이니까, LLM이 할 수 있다는 것입니다. 물론 평가용 LLM은 보통 더 강력한 모델을 사용합니다.

GPT-4나 Claude 3 Opus 같은 최상위 모델로 GPT-3.5나 경량 모델의 응답을 평가하는 식입니다. EvalCriteria를 보면 네 가지 평가 기준이 있습니다.

Relevance는 질문에 대한 응답이 얼마나 관련 있는지입니다. 엉뚱한 대답을 하면 점수가 낮아집니다.

Accuracy는 정보가 정확한지입니다. 사실과 다른 내용이 있으면 감점됩니다.

Helpfulness는 실제로 사용자에게 도움이 되는지입니다. 정확해도 이해하기 어려우면 점수가 낮을 수 있습니다.

Safety는 해로운 내용이 없는지입니다. create_eval_prompt 메서드가 평가의 핵심입니다.

평가용 LLM에게 무엇을 어떻게 평가할지 명확히 지시합니다. 점수와 이유를 JSON 형식으로 요청하면, 파싱하기 쉽고 일관된 결과를 얻을 수 있습니다.

김개발 씨가 질문했습니다. "LLM이 LLM을 평가하면, 그 평가 자체가 틀릴 수도 있지 않나요?" 날카로운 지적입니다.

그래서 몇 가지 보완책을 사용합니다. 첫째, 다수결 투표입니다.

같은 평가를 여러 번 수행하고 평균을 냅니다. 둘째, 인간 평가와의 상관관계 검증입니다.

정기적으로 사람이 직접 평가한 결과와 LLM 평가 결과를 비교하여 신뢰도를 확인합니다. 셋째, 평가 기준의 구체화입니다.

"좋은 응답"같은 모호한 기준 대신, 체크리스트 형태로 구체적인 기준을 제시합니다. 프로덕션 환경에서는 이 평가 결과를 대시보드로 시각화합니다.

시간에 따른 품질 변화를 모니터링하고, 특정 임계값 이하로 떨어지면 알림을 발생시킵니다. 이렇게 하면 문제를 조기에 발견하고 대응할 수 있습니다.

박시니어 씨가 덧붙였습니다. "평가 시스템은 AI 시스템의 눈과 귀야.

이게 없으면 우리 시스템이 점점 나빠지는지도 모른 채 운영하게 되는 거지."

실전 팁

💡 - 평가용 모델은 피평가 모델보다 강력한 것을 사용하세요

정기적으로 인간 평가와 LLM 평가의 상관관계를 검증하세요
평가 기준은 가능한 구체적이고 측정 가능하게 정의하세요

7. 프로덕션 모니터링 로깅

드디어 시스템이 완성되어 배포 준비가 끝났습니다. 하지만 김개발 씨는 또 다른 걱정이 생겼습니다.

"배포하고 나면 어떻게 관리하죠? 뭔가 잘못되면 어떻게 알 수 있나요?" 박시니어 씨가 진지한 표정으로 말했습니다.

"프로덕션은 전쟁터야. 모니터링 없이 배포하는 건 눈 감고 운전하는 것과 같아."

프로덕션 모니터링은 운영 중인 시스템의 상태를 실시간으로 파악하고 문제를 조기에 발견하는 체계입니다. 마치 비행기 조종석의 계기판처럼, 시스템의 건강 상태, 성능, 오류 등을 한눈에 볼 수 있게 해줍니다.

적절한 모니터링과 로깅이 없으면, 장애가 발생해도 원인을 파악하기 어렵고 대응이 늦어집니다.

다음 코드를 살펴봅시다.

import logging
import time
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
from datetime import datetime
from functools import wraps

@dataclass
class AgentMetrics:
    """에이전트 성능 지표"""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    total_latency_ms: float = 0.0
    tool_usage: Dict[str, int] = field(default_factory=dict)

class AgentMonitor:
    """프로덕션 모니터링 시스템"""
    def __init__(self):
        self.metrics = AgentMetrics()
        self.logger = self._setup_logger()

    def _setup_logger(self) -> logging.Logger:
        logger = logging.getLogger("agent")
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'))
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
        return logger

    def track_request(self, func):
        """요청 추적 데코레이터"""
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.time()
            self.metrics.total_requests += 1
            try:
                result = await func(*args, **kwargs)
                self.metrics.successful_requests += 1
                return result
            except Exception as e:
                self.metrics.failed_requests += 1
                self.logger.error(f"Request failed: {e}")
                raise
            finally:
                latency = (time.time() - start) * 1000
                self.metrics.total_latency_ms += latency
                self.logger.info(f"Request completed in {latency:.2f}ms")
        return wrapper

    def get_health_status(self) -> Dict[str, Any]:
        """시스템 상태 반환"""
        success_rate = (self.metrics.successful_requests /
                       max(self.metrics.total_requests, 1))
        return {"status": "healthy" if success_rate > 0.95 else "degraded",
                "success_rate": success_rate, "metrics": self.metrics}

김개발 씨의 시스템이 드디어 프로덕션에 배포되었습니다. 첫날은 순조로웠습니다.

그런데 이틀째 아침, 슬랙에 알림이 쏟아졌습니다. "AI 어시스턴트가 응답을 안 해요!" 김개발 씨는 당황했습니다.

서버 로그를 열어봤지만, 어디서 문제가 생겼는지 알 수 없었습니다. 박시니어 씨가 한숨을 쉬며 말했습니다.

"그래서 모니터링이 중요하다고 했잖아." 프로덕션 모니터링은 크게 세 가지 축으로 구성됩니다. 메트릭, 로깅, 트레이싱입니다.

메트릭은 숫자로 표현되는 지표입니다. 위 코드의 AgentMetrics가 그 예입니다.

총 요청 수, 성공 수, 실패 수, 평균 응답 시간 등을 수집합니다. 이 숫자들을 시간 순으로 그래프로 그리면, 시스템의 건강 상태를 한눈에 파악할 수 있습니다.

갑자기 에러율이 치솟거나 응답 시간이 길어지면 문제가 있다는 신호입니다. 로깅은 이벤트를 기록하는 것입니다.

단순히 print문을 쓰는 것이 아니라, 구조화된 형식으로 기록해야 합니다. 시간, 심각도, 메시지, 관련 컨텍스트를 함께 저장합니다.

문제가 발생했을 때 로그를 검색해서 무슨 일이 있었는지 추적할 수 있어야 합니다. track_request 데코레이터를 주목하세요.

이 데코레이터를 에이전트의 메인 함수에 붙이면, 모든 요청의 시작과 끝, 소요 시간, 성공 여부가 자동으로 기록됩니다. 개발자가 비즈니스 로직에만 집중할 수 있게 해주는 패턴입니다.

get_health_status 메서드는 현재 시스템 상태를 요약합니다. 성공률이 95퍼센트 이상이면 healthy, 그 이하면 degraded로 표시됩니다.

이 엔드포인트를 외부 모니터링 도구에 연결하면, 문제 발생 시 즉시 알림을 받을 수 있습니다. 김개발 씨가 물었습니다.

"도구 사용도 추적해야 하나요?" 물론입니다. tool_usage 딕셔너리가 그 역할을 합니다.

어떤 도구가 가장 많이 사용되는지, 특정 도구가 자주 실패하는지 파악할 수 있습니다. 예를 들어, 논문 검색 도구의 실패율이 높다면 해당 API에 문제가 있을 수 있습니다.

실무에서는 Prometheus, Grafana, Datadog 같은 도구를 사용합니다. 이런 도구들은 메트릭 수집, 시각화, 알림을 통합으로 제공합니다.

또한 구조화된 로깅 라이브러리인 structlog를 사용하면 JSON 형태로 로그를 남겨 검색과 분석이 훨씬 쉬워집니다. 박시니어 씨가 정리했습니다.

"모니터링은 보험과 같아. 평소에는 비용처럼 느껴지지만, 문제가 생겼을 때 그 가치를 깨닫게 되지."

실전 팁

💡 - 핵심 지표에 대한 알림 임계값을 설정하고 슬랙이나 이메일로 알림을 받으세요

로그는 구조화된 JSON 형식으로 남기면 검색과 분석이 쉬워집니다
분산 시스템에서는 트레이싱 ID로 요청의 전체 흐름을 추적하세요

8. 실전 사용 사례 검증

모든 컴포넌트가 준비되었습니다. 이제 실제로 작동하는지 검증할 차례입니다.

김개발 씨는 팀원들을 모아 시연회를 준비했습니다. 박시니어 씨가 말했습니다.

"진짜 사용자처럼 테스트해 봐야 해. 개발자 눈에는 괜찮아 보여도, 실제 연구원이 쓰면 불편할 수 있거든." 김개발 씨는 긴장되는 마음으로 시스템을 가동했습니다.

시스템 검증은 개발한 시스템이 실제 사용 환경에서 요구사항대로 작동하는지 확인하는 과정입니다. 마치 자동차를 출시하기 전에 다양한 도로와 날씨 조건에서 테스트 주행을 하듯이, AI 시스템도 다양한 시나리오에서 검증해야 합니다.

단위 테스트를 넘어 전체 시스템의 통합 테스트와 사용자 관점의 테스트가 필요합니다.

다음 코드를 살펴봅시다.

import asyncio
from dataclasses import dataclass
from typing import List, Dict, Any
from datetime import datetime

@dataclass
class TestScenario:
    """테스트 시나리오 정의"""
    name: str
    user_query: str
    expected_tools: List[str]
    quality_threshold: float
    timeout_seconds: float = 30.0

class SystemValidator:
    """전체 시스템 통합 검증"""
    def __init__(self, agent, judge, monitor):
        self.agent = agent
        self.judge = judge
        self.monitor = monitor

    async def run_scenario(self, scenario: TestScenario) -> Dict[str, Any]:
        """단일 시나리오 실행 및 검증"""
        start = datetime.now()
        try:
            # 에이전트 실행
            response = await asyncio.wait_for(
                self.agent.process({"query": scenario.user_query}),
                timeout=scenario.timeout_seconds
            )
            # 품질 평가
            eval_results = await self.judge.evaluate(
                scenario.user_query, response.get("answer", ""))
            avg_score = sum(r.score for r in eval_results) / len(eval_results)

            return {
                "scenario": scenario.name,
                "passed": avg_score >= scenario.quality_threshold,
                "score": avg_score,
                "latency_ms": (datetime.now() - start).total_seconds() * 1000
            }
        except asyncio.TimeoutError:
            return {"scenario": scenario.name, "passed": False,
                    "error": "Timeout"}

    async def validate_all(self, scenarios: List[TestScenario]) -> Dict:
        """전체 시나리오 검증 및 리포트 생성"""
        results = await asyncio.gather(
            *[self.run_scenario(s) for s in scenarios])
        passed = sum(1 for r in results if r.get("passed"))
        return {"total": len(scenarios), "passed": passed,
                "success_rate": passed / len(scenarios), "details": results}

# 테스트 시나리오 정의
scenarios = [
    TestScenario("논문검색", "딥러닝 최신 동향을 알려줘",
                 ["search_papers"], 0.8),
    TestScenario("계산요청", "1024의 제곱근을 계산해줘",
                 ["calculate"], 0.9),
]

시연회 당일 아침, 김개발 씨는 일찍 출근해서 마지막 점검을 했습니다. 테스트 시나리오를 하나씩 실행하며 결과를 확인했습니다.

대부분 잘 작동했지만, 몇 가지 문제가 발견되었습니다. 첫 번째 문제는 타임아웃이었습니다.

복잡한 논문 검색 요청에서 30초를 넘기는 경우가 있었습니다. 사용자 입장에서 30초는 영원처럼 느껴집니다.

이 문제를 해결하기 위해 검색 결과를 스트리밍으로 보여주도록 수정했습니다. "검색 중입니다...

3개의 논문을 찾았습니다... 추가 결과를 가져오는 중..." 두 번째 문제는 엣지 케이스였습니다.

"ㅋㅋㅋ"라고만 입력하면 에이전트가 당황했습니다. 의미 없는 입력에 대한 처리가 없었기 때문입니다.

예외 처리를 추가하고, "질문이 명확하지 않습니다. 더 구체적으로 물어봐 주세요"라고 응답하도록 했습니다.

SystemValidator 클래스는 이런 검증을 체계적으로 수행합니다. TestScenario에 시나리오 이름, 테스트 쿼리, 예상되는 도구 사용, 품질 임계값을 정의합니다.

run_scenario 메서드가 각 시나리오를 실행하고, LLM Judge로 품질을 평가하며, 타임아웃도 체크합니다. validate_all 메서드는 모든 시나리오를 병렬로 실행합니다.

asyncio.gather를 사용하면 여러 테스트를 동시에 돌릴 수 있어 검증 시간이 크게 단축됩니다. 최종 리포트에는 전체 통과율과 각 시나리오의 상세 결과가 포함됩니다.

시연회가 시작되었습니다. 연구팀 이박사가 첫 번째 테스터였습니다.

"트랜스포머 아키텍처의 어텐션 메커니즘에 대해 설명해줘." 에이전트가 막힘없이 대답했습니다. 논문을 검색하고, 핵심 개념을 요약하고, 관련 자료 링크까지 제공했습니다.

이박사가 이어서 물었습니다. "지난번에 물어본 BERT 논문 있잖아, 그거랑 비교해줘." 에이전트는 메모리 시스템에서 과거 대화를 검색해 맥락을 파악하고, 두 개념을 비교하는 상세한 답변을 생성했습니다.

박시니어 씨가 흐뭇하게 말했습니다. "드디어 프로덕션급 AI 에이전트가 완성됐네." 김개발 씨는 뿌듯했지만, 동시에 알고 있었습니다.

이것은 끝이 아니라 시작이라는 것을. 프로덕션 환경에서는 매일 새로운 도전이 기다리고 있습니다.

하지만 이제 그 도전을 맞이할 준비가 되어 있습니다. 모니터링이 상태를 알려주고, 평가 시스템이 품질을 체크하며, 로그가 모든 것을 기록합니다.

몇 주 후, 연구팀으로부터 피드백이 들어왔습니다. "덕분에 논문 조사 시간이 절반으로 줄었어요!" 김개발 씨는 미소를 지었습니다.

이것이 바로 프로덕션급 시스템을 만드는 보람입니다.

실전 팁

💡 - 실제 사용자와 함께 사용성 테스트를 진행하세요

엣지 케이스와 악의적인 입력에 대한 테스트도 포함하세요
검증 결과를 기반으로 지속적으로 시스템을 개선하세요

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#AIAgent#MultiAgent#MemorySystem#LLMEvaluation#AI Engineering

CodeDeck

Production-Grade AI Agent System 완벽 가이드

목차

1. 프로젝트 요구사항 분석

2. 멀티에이전트 아키텍처 설계

3. 컨텍스트 압축 최적화

4. 시간적 지식 그래프 메모리

5. 최적화된 툴셋 구축

6. LLM 평가 시스템

7. 프로덕션 모니터링 로깅

8. 실전 사용 사례 검증

댓글 (0)

함께 보면 좋은 카드 뉴스

Context Degradation 패턴 마스터

Context Fundamentals 완벽 가이드 AI 에이전트 맥락 이해하기

LLM-as-a-Judge 평가 시스템 완벽 가이드

Agent-Optimized Tool Set 설계 실습 가이드

Knowledge Graph 메모리 시스템 완벽 가이드