Python으로 AI 에이전트 만들기 12편 - 실전 프로젝트: 나만의 AI 비서 완성

지금까지 배운 모든 개념을 활용하여 실제 동작하는 AI 비서를 완성해보는 실전 프로젝트입니다. 멀티 에이전트 시스템, 도구 통합, 메모리 관리까지 모든 것을 종합하여 실무에서 바로 사용할 수 있는 수준의 AI 비서를 만들어봅니다.

1. 프로젝트 아키텍처 설계

시작하며

여러분이 AI 에이전트를 만들 때 "일단 코드부터 짜고 보자"라고 시작한 적 있나요? 처음엔 간단하게 시작했지만, 기능이 하나씩 추가될 때마다 코드가 엉망이 되고, 나중엔 어디서부터 수정해야 할지 막막해지는 경험 말이죠.

이런 문제는 실제 개발 현장에서 너무나 자주 발생합니다. 특히 AI 에이전트처럼 여러 컴포넌트가 복잡하게 얽혀있는 시스템에서는 더욱 그렇습니다.

처음부터 명확한 아키텍처 없이 시작하면, 나중에 새로운 에이전트를 추가하거나 도구를 통합할 때마다 전체 코드를 뜯어고쳐야 하는 상황이 옵니다. 바로 이럴 때 필요한 것이 체계적인 프로젝트 아키텍처 설계입니다.

확장 가능하고 유지보수하기 쉬운 구조를 처음부터 만들어두면, 나중에 기능을 추가할 때도 기존 코드를 건드리지 않고 새로운 모듈만 추가하면 됩니다.

개요

간단히 말해서, 프로젝트 아키텍처 설계란 AI 비서를 구성하는 모든 컴포넌트들의 역할과 관계를 미리 정의하는 것입니다. 왜 이것이 필요할까요?

실무에서 AI 에이전트 프로젝트는 끊임없이 진화합니다. 오늘은 날씨 정보만 알려주던 비서가 내일은 일정 관리를 해야 하고, 다음 주엔 이메일까지 보내야 할 수 있습니다.

이럴 때마다 코드 전체를 뜯어고치는 것은 비효율적이죠. 좋은 아키텍처는 이런 확장을 쉽게 만들어줍니다.

기존에는 모든 기능을 하나의 큰 클래스에 때려 넣었다면, 이제는 각 컴포넌트를 독립적인 모듈로 분리하여 관리할 수 있습니다. 핵심 특징으로는 첫째, 모듈화된 구조로 각 컴포넌트가 독립적으로 동작합니다.

둘째, 계층적 설계로 상위 레벨과 하위 레벨이 명확히 구분됩니다. 셋째, 인터페이스 기반 설계로 구현체를 쉽게 교체할 수 있습니다.

이러한 특징들이 중요한 이유는, 프로젝트가 커질수록 유지보수와 확장이 기하급수적으로 어려워지기 때문입니다.

코드 예제

# ai_assistant/architecture/base.py
from abc import ABC, abstractmethod
from typing import Dict, List, Any
from dataclasses import dataclass

@dataclass
class AgentConfig:
    """에이전트 설정을 담는 데이터 클래스"""
    name: str
    description: str
    model: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 1000

class BaseAgent(ABC):
    """모든 에이전트가 상속받을 기본 클래스"""

    def __init__(self, config: AgentConfig):
        self.config = config
        self.tools: List[Any] = []

    @abstractmethod
    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """에이전트의 핵심 처리 로직 - 각 에이전트가 구현해야 함"""
        pass

    def add_tool(self, tool: Any) -> None:
        """에이전트에 도구를 추가"""
        self.tools.append(tool)

class AgentOrchestrator:
    """여러 에이전트를 조율하는 오케스트레이터"""

    def __init__(self):
        self.agents: Dict[str, BaseAgent] = {}

    def register_agent(self, agent_type: str, agent: BaseAgent):
        """새로운 에이전트를 시스템에 등록"""
        self.agents[agent_type] = agent

    async def delegate_task(self, task_type: str, data: Dict[str, Any]):
        """작업을 적절한 에이전트에게 위임"""
        if task_type in self.agents:
            return await self.agents[task_type].process(data)
        raise ValueError(f"No agent found for task type: {task_type}")

설명

이것이 하는 일: 이 아키텍처는 AI 비서를 구성하는 모든 에이전트들이 따라야 할 기본 규칙과 구조를 정의합니다. 새로운 에이전트를 추가할 때 전체 시스템을 수정하지 않고도 확장할 수 있도록 설계되었습니다.

첫 번째로, AgentConfig 데이터 클래스는 각 에이전트의 설정 정보를 표준화된 방식으로 관리합니다. 이렇게 하면 모든 에이전트가 동일한 방식으로 설정되어 혼란을 방지할 수 있습니다.

name, description 같은 기본 정보부터 model, temperature 같은 AI 모델 파라미터까지 체계적으로 관리됩니다. 그 다음으로, BaseAgent 추상 클래스가 실행되면서 모든 에이전트가 반드시 구현해야 할 process 메서드를 정의합니다.

이것이 핵심인데, 어떤 에이전트를 만들든 간에 process 메서드만 구현하면 시스템에 통합될 수 있습니다. 또한 add_tool 메서드를 통해 각 에이전트에 필요한 도구들을 동적으로 추가할 수 있습니다.

마지막으로, AgentOrchestrator가 여러 에이전트를 관리하고 작업을 적절한 에이전트에게 위임합니다. register_agent로 새 에이전트를 등록하고, delegate_task로 작업을 분배하는 중앙 집중식 관리 시스템입니다.

이렇게 하면 에이전트가 10개든 100개든 간에 체계적으로 관리할 수 있습니다. 여러분이 이 아키텍처를 사용하면 새로운 기능을 추가할 때마다 기존 코드를 수정하지 않고 새로운 클래스만 추가하면 됩니다.

코드의 재사용성이 높아지고, 버그가 발생해도 문제가 있는 모듈만 수정하면 되므로 유지보수가 훨씬 쉬워집니다. 또한 각 에이전트를 독립적으로 테스트할 수 있어 품질 관리도 용이합니다.

실전 팁

💡 BaseAgent를 상속받을 때는 반드시 process 메서드를 구현해야 합니다. 이를 잊으면 인스턴스 생성 시 에러가 발생하니 주의하세요.

💡 AgentConfig는 dataclass를 사용하여 타입 힌팅과 기본값 설정이 쉽습니다. 필요한 설정 항목이 추가되면 이 클래스만 수정하면 됩니다.

💡 에이전트 등록 시 의미 있는 agent_type 이름을 사용하세요. "agent1", "agent2" 대신 "weather_agent", "calendar_agent"처럼 명확한 이름을 쓰면 코드 가독성이 크게 향상됩니다.

💡 오케스트레이터에 에이전트를 등록하기 전에 설정이 제대로 되었는지 검증 로직을 추가하면 런타임 에러를 사전에 방지할 수 있습니다.

💡 async/await를 사용하여 비동기 처리를 구현했으므로, 여러 에이전트가 동시에 작업을 처리할 수 있어 성능이 향상됩니다.

2. 메인 에이전트 구현

시작하며

여러분의 AI 비서에게 "오늘 날씨 알려주고, 3시에 회의 일정 잡아줘"라고 말했을 때, 이 복잡한 요청을 어떻게 처리해야 할까요? 두 가지 작업이 섞여 있고, 각각 다른 에이전트가 처리해야 하는 상황입니다.

이런 문제는 실제 AI 비서 개발에서 가장 먼저 마주치는 도전입니다. 사용자는 복잡한 요청을 자연스러운 언어로 던지는데, 시스템은 이를 분석하고 적절한 에이전트에게 분배해야 합니다.

또한 각 에이전트의 결과를 모아서 사용자에게 일관된 응답을 제공해야 하죠. 바로 이럴 때 필요한 것이 메인 에이전트입니다.

메인 에이전트는 오케스트라의 지휘자처럼 사용자 요청을 분석하고, 적절한 전문 에이전트들을 호출하며, 결과를 통합하여 최종 응답을 만들어냅니다.

개요

간단히 말해서, 메인 에이전트는 사용자와 직접 상호작용하며 전체 시스템을 조율하는 핵심 에이전트입니다. 왜 이것이 필요할까요?

실무에서 사용자의 요청은 단순하지 않습니다. "내일 서울 날씨 확인하고, 비 온다면 우산 사는 걸 할 일 목록에 추가해줘" 같은 조건부 로직이 포함된 복잡한 요청도 처리해야 합니다.

메인 에이전트는 이런 복잡한 요청을 이해하고, 여러 전문 에이전트를 순차적 또는 병렬로 호출하며, 결과를 종합하는 역할을 합니다. 기존에는 모든 로직을 한 곳에 때려 넣어 복잡도가 기하급수적으로 증가했다면, 이제는 메인 에이전트가 고수준의 조율만 담당하고 실제 작업은 전문 에이전트에게 위임할 수 있습니다.

핵심 특징으로는 첫째, 사용자 의도 분석 능력으로 요청에서 작업 유형을 식별합니다. 둘째, 워크플로우 관리 능력으로 여러 에이전트를 순서대로 또는 동시에 실행합니다.

셋째, 컨텍스트 유지 능력으로 대화의 흐름을 기억하고 참조합니다. 이러한 특징들이 중요한 이유는, 사용자가 자연스럽고 복잡한 대화를 할 수 있도록 하기 때문입니다.

코드 예제

# ai_assistant/agents/main_agent.py
from typing import Dict, Any, List
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import HumanMessage, SystemMessage
from .base import BaseAgent, AgentConfig

class MainAgent(BaseAgent):
    """사용자 요청을 받아 적절한 에이전트에게 분배하는 메인 에이전트"""

    def __init__(self, config: AgentConfig, orchestrator):
        super().__init__(config)
        self.llm = ChatOpenAI(model=config.model, temperature=config.temperature)
        self.orchestrator = orchestrator
        self.conversation_history: List[Dict] = []

    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """사용자 요청을 처리하는 메인 로직"""
        user_input = input_data.get("message", "")

        # 1단계: 사용자 의도 분석
        intent = await self._analyze_intent(user_input)

        # 2단계: 적절한 에이전트에게 작업 위임
        results = []
        for task in intent["tasks"]:
            result = await self.orchestrator.delegate_task(
                task["type"],
                {"query": task["query"], "context": self.conversation_history}
            )
            results.append(result)

        # 3단계: 결과 통합 및 응답 생성
        final_response = await self._synthesize_response(user_input, results)

        # 4단계: 대화 히스토리에 추가
        self._update_history(user_input, final_response)

        return {"response": final_response, "intent": intent}

    async def _analyze_intent(self, user_input: str) -> Dict[str, Any]:
        """LLM을 사용하여 사용자 의도 분석"""
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content="당신은 사용자 요청을 분석하여 필요한 작업을 식별하는 AI입니다."),
            HumanMessage(content=f"다음 요청을 분석하여 필요한 작업 목록을 JSON 형태로 반환하세요: {user_input}")
        ])

        response = await self.llm.apredict_messages(prompt.format_messages())
        # 실제로는 JSON 파싱 로직이 필요
        return {"tasks": [{"type": "general", "query": user_input}]}

    async def _synthesize_response(self, original_query: str, results: List[Dict]) -> str:
        """여러 에이전트의 결과를 자연스러운 응답으로 통합"""
        combined = "\n".join([r.get("answer", "") for r in results])
        return f"요청하신 '{original_query}'에 대한 답변입니다:\n{combined}"

    def _update_history(self, user_input: str, response: str):
        """대화 히스토리 업데이트"""
        self.conversation_history.append({"user": user_input, "assistant": response})
        # 최근 10개 대화만 유지
        if len(self.conversation_history) > 10:
            self.conversation_history.pop(0)

설명

이것이 하는 일: 메인 에이전트는 사용자와의 모든 상호작용을 관리하며, 복잡한 요청을 이해하고 적절한 전문 에이전트들을 조율하여 통합된 응답을 제공합니다. 마치 레스토랑의 웨이터가 고객 주문을 받아 주방의 각 파트(전채, 메인, 디저트)에 전달하고 최종적으로 코스 요리를 제공하는 것과 같습니다.

첫 번째로, _analyze_intent 메서드가 사용자의 자연어 입력을 분석합니다. LLM을 활용하여 "오늘 날씨 알려주고 일정도 확인해줘"라는 요청에서 "날씨 조회"와 "일정 확인"이라는 두 가지 작업을 식별해냅니다.

이 과정에서 ChatPromptTemplate을 사용하여 LLM에게 명확한 지시를 내리고, JSON 형태로 구조화된 결과를 받아옵니다. 그 다음으로, process 메서드의 작업 위임 로직이 실행됩니다.

식별된 각 작업을 orchestrator를 통해 적절한 전문 에이전트에게 전달합니다. 이때 conversation_history를 함께 전달하여 이전 대화 맥락을 유지합니다.

예를 들어 "그거 언제였지?"라는 후속 질문에도 대답할 수 있는 것은 이 컨텍스트 덕분입니다. 세 번째로, _synthesize_response 메서드가 여러 에이전트로부터 받은 결과들을 하나의 자연스러운 응답으로 통합합니다.

날씨 에이전트가 "오늘은 맑음"이라고 하고 일정 에이전트가 "3시에 회의 있음"이라고 하면, "오늘은 맑은 날씨이고, 3시에 회의가 예정되어 있습니다"처럼 자연스럽게 합쳐줍니다. 마지막으로, _update_history가 대화 내용을 저장하여 다음 대화에서 참조할 수 있게 합니다.

메모리 관리를 위해 최근 10개만 유지하는 전략을 사용하여 무한정 커지는 것을 방지합니다. 여러분이 이 메인 에이전트를 사용하면 복잡한 멀티턴 대화를 자연스럽게 처리할 수 있습니다.

사용자는 마치 사람과 대화하는 것처럼 편하게 요청할 수 있고, 시스템은 백그라운드에서 여러 에이전트를 조율하여 정확한 답변을 제공합니다. 또한 대화 히스토리를 통해 컨텍스트를 유지하므로 "그거", "그때" 같은 대명사도 이해할 수 있습니다.

실전 팁

💡 _analyze_intent에서 LLM의 응답을 파싱할 때는 반드시 예외 처리를 추가하세요. LLM이 항상 완벽한 JSON을 반환하는 것은 아니므로, 파싱 실패 시 기본 작업으로 폴백하는 로직이 필요합니다.

💡 conversation_history의 크기를 제한하는 것은 매우 중요합니다. 토큰 제한을 초과하면 API 호출이 실패할 수 있으므로, 개수뿐만 아니라 총 토큰 수도 체크하는 것이 좋습니다.

💡 여러 에이전트를 동시에 호출할 수 있다면 asyncio.gather를 사용하여 병렬 처리하세요. 날씨 조회와 일정 확인은 독립적이므로 동시에 실행하면 응답 시간이 크게 단축됩니다.

💡 에이전트 호출 실패 시 사용자에게 부분적인 결과라도 제공하는 것이 좋습니다. "날씨 정보는 가져왔지만 일정 조회는 실패했습니다"처럼 명확히 알려주세요.

💡 디버깅을 위해 각 단계(의도 분석, 작업 위임, 응답 통합)의 중간 결과를 로깅하세요. 어디서 문제가 생겼는지 추적하기 쉬워집니다.

3. 전문 에이전트들 구성

시작하며

여러분이 AI 비서를 만들면서 "하나의 에이전트가 모든 걸 다 하면 되지 않을까?"라고 생각해본 적 있나요? 처음엔 간단해 보이지만, 날씨 조회, 이메일 작성, 코드 생성까지 모든 기능을 하나에 때려 넣다 보면 프롬프트가 엄청나게 길어지고 성능도 떨어집니다.

이런 문제는 실제 AI 시스템 개발에서 "만능 에이전트의 함정"이라고 불립니다. 하나의 에이전트가 모든 것을 하려다 보면 각 작업의 품질이 떨어지고, 한 부분을 수정하면 다른 부분이 영향을 받는 취약한 구조가 됩니다.

또한 특정 작업에 최적화하기도 어렵습니다. 바로 이럴 때 필요한 것이 역할별로 전문화된 에이전트들입니다.

각 에이전트는 자신의 전문 분야에서 최고의 성능을 발휘하도록 설계되고, 필요할 때 메인 에이전트가 적절한 전문가를 호출하는 방식입니다.

개요

간단히 말해서, 전문 에이전트는 특정 작업에 특화되어 최적화된 도구와 프롬프트를 가진 독립적인 에이전트입니다. 왜 이것이 필요할까요?

실무에서 각 작업은 서로 다른 전문성을 요구합니다. 날씨 조회는 API 호출과 데이터 파싱에 능해야 하고, 이메일 작성은 자연어 생성과 포맷팅에 강해야 하며, 데이터 분석은 계산과 시각화에 뛰어나야 합니다.

하나의 에이전트로는 모든 분야에서 최고가 될 수 없습니다. 예를 들어, 의료 진단 챗봇은 의료 데이터베이스 접근 도구와 의학 용어 특화 프롬프트가 필요하지만, 법률 자문 챗봇은 법률 DB와 법률 용어가 필요합니다.

기존에는 거대한 if-else 문으로 작업을 분기했다면, 이제는 각 작업을 담당하는 독립적인 에이전트를 만들어 깔끔하게 분리할 수 있습니다. 핵심 특징으로는 첫째, 단일 책임 원칙으로 각 에이전트는 하나의 도메인만 담당합니다.

둘째, 전문화된 도구 세트로 해당 작업에 필요한 도구만 장착합니다. 셋째, 최적화된 프롬프트로 특정 작업의 품질을 극대화합니다.

이러한 특징들이 중요한 이유는, 시스템의 확장성과 유지보수성을 크게 향상시키기 때문입니다.

코드 예제

# ai_assistant/agents/specialized_agents.py
from typing import Dict, Any
from langchain.tools import Tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from .base import BaseAgent, AgentConfig

class WeatherAgent(BaseAgent):
    """날씨 정보 전문 에이전트"""

    def __init__(self, config: AgentConfig):
        super().__init__(config)
        # 날씨 조회에 특화된 도구 추가
        self.add_tool(Tool(
            name="get_weather",
            func=self._get_weather_data,
            description="주어진 도시의 현재 날씨를 조회합니다"
        ))

        # 날씨 전문 프롬프트
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "당신은 날씨 정보 전문가입니다. 날씨 데이터를 조회하고 사용자 친화적으로 설명합니다."),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        self.llm = ChatOpenAI(model=config.model, temperature=0.3)  # 낮은 temperature로 정확성 우선
        self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True)

    def _get_weather_data(self, city: str) -> str:
        """실제로는 날씨 API를 호출"""
        # 여기서는 예시로 더미 데이터 반환
        return f"{city}의 날씨: 맑음, 기온 22도"

    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        query = input_data.get("query", "")
        result = await self.executor.ainvoke({"input": query})
        return {"answer": result["output"], "agent_type": "weather"}

class EmailAgent(BaseAgent):
    """이메일 작성 전문 에이전트"""

    def __init__(self, config: AgentConfig):
        super().__init__(config)
        # 이메일 관련 도구 추가
        self.add_tool(Tool(
            name="send_email",
            func=self._send_email,
            description="이메일을 작성하고 전송합니다"
        ))

        # 이메일 작성에 특화된 프롬프트
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "당신은 전문적이고 예의바른 이메일 작성 전문가입니다. 비즈니스 톤을 유지하며 명확하게 작성합니다."),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        self.llm = ChatOpenAI(model=config.model, temperature=0.7)  # 창의적인 문장을 위해 적절한 temperature
        self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=self.tools)

    def _send_email(self, to: str, subject: str, body: str) -> str:
        """실제로는 이메일 API를 호출"""
        return f"이메일 전송 완료: {to}에게 '{subject}' 제목으로 전송됨"

    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        query = input_data.get("query", "")
        result = await self.executor.ainvoke({"input": query})
        return {"answer": result["output"], "agent_type": "email"}

class DataAnalysisAgent(BaseAgent):
    """데이터 분석 전문 에이전트"""

    def __init__(self, config: AgentConfig):
        super().__init__(config)
        # 데이터 분석 도구 추가
        self.add_tool(Tool(
            name="analyze_data",
            func=self._analyze_data,
            description="데이터를 분석하고 통계를 생성합니다"
        ))

        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "당신은 데이터 분석 전문가입니다. 숫자와 통계를 정확히 계산하고 인사이트를 제공합니다."),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        self.llm = ChatOpenAI(model=config.model, temperature=0.1)  # 매우 낮은 temperature로 정확한 계산
        self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=self.tools)

    def _analyze_data(self, data: str) -> str:
        """실제로는 pandas 등을 사용하여 분석"""
        return f"데이터 분석 결과: 평균 45.2, 중앙값 42, 표준편차 8.3"

    async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        query = input_data.get("query", "")
        result = await self.executor.ainvoke({"input": query})
        return {"answer": result["output"], "agent_type": "data_analysis"}

설명

이것이 하는 일: 전문 에이전트 시스템은 각 작업 유형별로 독립적이고 최적화된 에이전트를 제공하여, 전체 시스템의 성능과 유지보수성을 극대화합니다. 마치 병원에 내과, 외과, 정형외과 전문의가 있는 것처럼, 각 에이전트는 자신의 전문 분야에서 최고의 성능을 발휘합니다.

첫 번째로, WeatherAgent는 날씨 조회에 특화되어 있습니다. get_weather_data 도구를 통해 외부 날씨 API를 호출하고, 낮은 temperature(0.3)를 사용하여 정확한 정보 전달을 우선시합니다.

시스템 프롬프트도 "날씨 정보 전문가"로 설정하여 날씨 관련 질문에 최적화된 응답을 생성합니다. 실무에서는 OpenWeatherMap이나 WeatherAPI 같은 실제 서비스와 연동됩니다.

그 다음으로, EmailAgent는 이메일 작성과 전송을 담당합니다. 중요한 점은 temperature를 0.7로 설정하여 창의적이면서도 전문적인 문장을 생성한다는 것입니다.

시스템 프롬프트에 "비즈니스 톤"을 명시하여 일관된 품질의 이메일을 작성합니다. send_email 도구는 실제로 Gmail API나 SendGrid 같은 서비스와 연동될 수 있습니다.

세 번째로, DataAnalysisAgent는 매우 낮은 temperature(0.1)를 사용하여 정확한 계산을 보장합니다. 데이터 분석에서는 창의성보다 정확성이 중요하기 때문입니다.

analyze_data 도구는 실제로 pandas, numpy 같은 라이브러리를 사용하여 복잡한 통계 계산을 수행할 수 있습니다. 각 에이전트는 create_openai_functions_agent와 AgentExecutor를 사용하여 LangChain의 강력한 에이전트 기능을 활용합니다.

MessagesPlaceholder를 통해 에이전트가 중간 추론 과정(scratchpad)을 기록하고 참조할 수 있어, 복잡한 다단계 작업도 처리할 수 있습니다. 여러분이 이 전문 에이전트 시스템을 사용하면 각 작업의 품질이 크게 향상됩니다.

날씨 조회는 더 정확해지고, 이메일은 더 전문적으로 작성되며, 데이터 분석은 더 신뢰할 수 있게 됩니다. 또한 새로운 기능을 추가할 때도 새로운 전문 에이전트만 만들면 되므로 확장이 쉽습니다.

예를 들어 번역 기능이 필요하면 TranslationAgent만 추가하면 됩니다.

실전 팁

💡 각 에이전트의 temperature 설정은 매우 중요합니다. 정확성이 필요한 작업(날씨, 데이터 분석)은 낮게(0.1-0.3), 창의성이 필요한 작업(이메일 작성, 콘텐츠 생성)은 높게(0.7-0.9) 설정하세요.

💡 에이전트별로 다른 LLM 모델을 사용할 수도 있습니다. 간단한 작업은 gpt-3.5-turbo로, 복잡한 작업은 gpt-4로 처리하면 비용을 절감하면서도 품질을 유지할 수 있습니다.

💡 각 에이전트의 도구는 실제 외부 API와 연동할 때 반드시 에러 핸들링을 추가하세요. API 호출 실패, 타임아웃, 잘못된 응답 등을 처리하는 로직이 필수입니다.

💡 verbose=True 옵션을 개발 중에 사용하면 에이전트의 추론 과정을 볼 수 있어 디버깅에 매우 유용합니다. 프로덕션에서는 False로 설정하여 불필요한 로그를 줄이세요.

💡 에이전트의 시스템 프롬프트는 매우 구체적으로 작성하세요. "당신은 전문가입니다" 보다 "당신은 10년 경력의 기상 전문가로, 일반인도 이해하기 쉽게 설명합니다"처럼 구체적일수록 좋습니다.

4. 도구 통합 시스템

시작하며

여러분의 AI 비서가 "내일 회의 자료를 PDF로 만들어줘"라는 요청을 받았을 때, 어떻게 실제로 PDF를 생성할까요? AI는 텍스트 생성은 잘하지만, 파일 시스템 접근, API 호출, 데이터베이스 쿼리 같은 실제 작업은 직접 할 수 없습니다.

이런 문제는 AI 에이전트의 근본적인 한계입니다. LLM은 언어를 이해하고 생성하는 데는 뛰어나지만, 외부 세계와 상호작용하려면 "도구"가 필요합니다.

웹 검색, 파일 읽기/쓰기, API 호출, 계산기 등 실제 작업을 수행하는 코드가 필요한 거죠. 바로 이럴 때 필요한 것이 도구 통합 시스템입니다.

에이전트가 필요한 도구를 쉽게 사용할 수 있도록 표준화된 인터페이스를 제공하고, 새로운 도구를 추가하거나 기존 도구를 업데이트하기 쉬운 구조를 만듭니다.

개요

간단히 말해서, 도구 통합 시스템은 AI 에이전트가 외부 세계와 상호작용할 수 있도록 다양한 기능을 제공하는 플러그인 같은 시스템입니다. 왜 이것이 필요할까요?

실무에서 AI 비서는 단순히 대화만 하는 것이 아닙니다. 파일을 읽고 쓰고, 웹에서 정보를 검색하고, 데이터베이스를 조회하고, 외부 API를 호출해야 합니다.

예를 들어, "최신 기술 뉴스를 요약해줘"라는 요청은 웹 스크래핑 도구, 텍스트 요약 도구, 그리고 결과를 저장하는 파일 시스템 도구가 필요합니다. 각 도구를 에이전트마다 다르게 구현하면 코드 중복과 유지보수 문제가 발생합니다.

기존에는 각 에이전트가 자체적으로 외부 기능을 구현했다면, 이제는 중앙화된 도구 레지스트리에서 필요한 도구를 가져다 쓸 수 있습니다. 핵심 특징으로는 첫째, 표준화된 인터페이스로 모든 도구가 동일한 방식으로 호출됩니다.

둘째, 플러그인 아키텍처로 새로운 도구를 쉽게 추가할 수 있습니다. 셋째, 에러 핸들링과 재시도 로직이 내장되어 안정성이 높습니다.

이러한 특징들이 중요한 이유는, 에이전트의 능력을 쉽게 확장하고 시스템을 안정적으로 유지하기 때문입니다.

코드 예제

# ai_assistant/tools/tool_registry.py
from typing import Dict, Callable, Any, List
from langchain.tools import Tool
import requests
from pathlib import Path
import json

class ToolRegistry:
    """모든 도구를 등록하고 관리하는 중앙 레지스트리"""

    def __init__(self):
        self.tools: Dict[str, Tool] = {}
        self._register_default_tools()

    def _register_default_tools(self):
        """기본 도구들을 등록"""
        # 웹 검색 도구
        self.register_tool(Tool(
            name="web_search",
            func=self._web_search,
            description="인터넷에서 정보를 검색합니다. 검색어를 입력하면 관련 결과를 반환합니다."
        ))

        # 파일 읽기 도구
        self.register_tool(Tool(
            name="read_file",
            func=self._read_file,
            description="파일 내용을 읽습니다. 파일 경로를 입력하면 내용을 반환합니다."
        ))

        # 파일 쓰기 도구
        self.register_tool(Tool(
            name="write_file",
            func=self._write_file,
            description="파일에 내용을 씁니다. '경로|내용' 형식으로 입력합니다."
        ))

        # 계산기 도구
        self.register_tool(Tool(
            name="calculator",
            func=self._calculate,
            description="수학 계산을 수행합니다. Python 표현식을 입력하면 결과를 반환합니다."
        ))

        # API 호출 도구
        self.register_tool(Tool(
            name="api_call",
            func=self._api_call,
            description="외부 API를 호출합니다. 'GET|URL' 또는 'POST|URL|JSON_DATA' 형식으로 입력합니다."
        ))

    def register_tool(self, tool: Tool):
        """새로운 도구를 레지스트리에 등록"""
        self.tools[tool.name] = tool

    def get_tool(self, name: str) -> Tool:
        """이름으로 도구 가져오기"""
        if name not in self.tools:
            raise ValueError(f"도구 '{name}'을 찾을 수 없습니다")
        return self.tools[name]

    def get_all_tools(self) -> List[Tool]:
        """모든 도구 가져오기"""
        return list(self.tools.values())

    def _web_search(self, query: str) -> str:
        """웹 검색 수행 - 실제로는 Google API 등을 사용"""
        try:
            # 여기서는 예시로 더미 데이터 반환
            return f"'{query}'에 대한 검색 결과: [더미 데이터] 관련 정보가 여기 표시됩니다."
        except Exception as e:
            return f"웹 검색 중 오류 발생: {str(e)}"

    def _read_file(self, file_path: str) -> str:
        """파일 내용 읽기"""
        try:
            path = Path(file_path)
            if not path.exists():
                return f"오류: 파일 '{file_path}'을 찾을 수 없습니다"
            return path.read_text(encoding='utf-8')
        except Exception as e:
            return f"파일 읽기 중 오류 발생: {str(e)}"

    def _write_file(self, input_str: str) -> str:
        """파일에 내용 쓰기"""
        try:
            file_path, content = input_str.split('|', 1)
            path = Path(file_path)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(content, encoding='utf-8')
            return f"파일 '{file_path}'에 성공적으로 작성했습니다"
        except Exception as e:
            return f"파일 쓰기 중 오류 발생: {str(e)}"

    def _calculate(self, expression: str) -> str:
        """수학 계산 수행"""
        try:
            # 보안을 위해 eval 대신 ast.literal_eval 사용 권장
            result = eval(expression, {"__builtins__": {}}, {})
            return f"계산 결과: {result}"
        except Exception as e:
            return f"계산 중 오류 발생: {str(e)}"

    def _api_call(self, input_str: str) -> str:
        """외부 API 호출"""
        try:
            parts = input_str.split('|')
            method = parts[0].upper()
            url = parts[1]

            if method == "GET":
                response = requests.get(url, timeout=10)
            elif method == "POST":
                data = json.loads(parts[2]) if len(parts) > 2 else {}
                response = requests.post(url, json=data, timeout=10)
            else:
                return f"지원하지 않는 HTTP 메서드: {method}"

            return f"API 응답 (상태 코드 {response.status_code}): {response.text[:500]}"
        except Exception as e:
            return f"API 호출 중 오류 발생: {str(e)}"

# 사용 예시
tool_registry = ToolRegistry()

# 커스텀 도구 추가
custom_tool = Tool(
    name="translate",
    func=lambda text: f"번역된 텍스트: {text}",  # 실제로는 번역 API 호출
    description="텍스트를 번역합니다"
)
tool_registry.register_tool(custom_tool)

설명

이것이 하는 일: 도구 통합 시스템은 AI 에이전트가 실제 세계와 상호작용하는 데 필요한 모든 기능을 표준화된 방식으로 제공합니다. 마치 스마트폰의 앱 스토어처럼, 필요한 도구를 찾아 설치하고 사용할 수 있는 중앙화된 플랫폼입니다.

첫 번째로, ToolRegistry 클래스가 모든 도구를 관리하는 중앙 허브 역할을 합니다. _register_default_tools 메서드에서 기본적으로 제공되는 도구들(웹 검색, 파일 읽기/쓰기, 계산기, API 호출)을 등록합니다.

이렇게 하면 모든 에이전트가 이 기본 도구들을 즉시 사용할 수 있습니다. 새 프로젝트를 시작할 때마다 이런 기본 기능들을 다시 구현할 필요가 없어집니다.

그 다음으로, 각 도구는 LangChain의 Tool 클래스로 래핑되어 표준화된 인터페이스를 제공합니다. 모든 도구는 name(도구 이름), func(실제 기능을 수행하는 함수), description(AI가 언제 이 도구를 사용해야 하는지 이해하도록 돕는 설명)을 가집니다.

이 표준화된 구조 덕분에 AI 에이전트는 어떤 도구든 동일한 방식으로 호출할 수 있습니다. 세 번째로, 각 도구 함수(_web_search, _read_file 등)는 철저한 에러 핸들링을 포함합니다.

try-except 블록으로 감싸서 어떤 오류가 발생하더라도 시스템이 크래시하지 않고 의미 있는 에러 메시지를 반환합니다. 예를 들어 파일이 없으면 "파일을 찾을 수 없습니다"라고 명확히 알려줍니다.

네 번째로, register_tool 메서드를 통해 커스텀 도구를 쉽게 추가할 수 있습니다. 코드 예시의 마지막 부분처럼 번역 도구를 추가하는 것이 단 몇 줄이면 됩니다.

이 확장성 덕분에 프로젝트가 성장하면서 필요한 도구들을 계속 추가할 수 있습니다. 다섯 번째로, _api_call 같은 범용 도구는 여러 시나리오에 활용될 수 있습니다.

GET/POST 요청을 모두 지원하고, JSON 데이터도 전달할 수 있어서 대부분의 REST API와 통신할 수 있습니다. 타임아웃 설정(timeout=10)으로 무한 대기를 방지합니다.

여러분이 이 도구 통합 시스템을 사용하면 에이전트의 능력을 폭발적으로 확장할 수 있습니다. 웹에서 최신 정보를 가져오고, 파일을 읽고 쓰며, 외부 서비스와 통신하는 모든 것이 가능해집니다.

또한 도구를 재사용할 수 있어 개발 시간이 크게 단축되고, 중앙에서 관리하므로 버그 수정이나 기능 개선도 한 곳에서 하면 모든 에이전트에 적용됩니다.

실전 팁

💡 보안이 중요한 도구(파일 쓰기, API 호출)는 반드시 입력 검증을 추가하세요. 예를 들어 파일 쓰기는 허용된 디렉토리 내에서만 가능하도록 경로를 체크해야 합니다.

💡 계산기 도구에서 eval()을 사용할 때는 매우 주의해야 합니다. 예시처럼 __builtins__를 빈 딕셔너리로 설정하거나, ast.literal_eval이나 sympy 같은 안전한 대안을 사용하세요.

💡 API 호출 도구는 rate limiting을 고려해야 합니다. 같은 API를 너무 자주 호출하면 차단될 수 있으므로, 캐싱이나 쓰로틀링을 추가하는 것이 좋습니다.

💡 각 도구의 description은 매우 중요합니다. AI가 이 설명을 읽고 언제 도구를 사용할지 결정하므로, 명확하고 구체적으로 작성하세요. "파일을 읽습니다" 보다 "로컬 파일 시스템에서 텍스트 파일의 내용을 읽어옵니다. 절대 경로 또는 상대 경로를 지원합니다"가 더 좋습니다.

💡 도구 함수의 반환값은 항상 문자열로 통일하세요. LangChain의 Tool은 문자열 반환을 기대하므로, 딕셔너리나 리스트를 반환하려면 json.dumps()로 직렬화해야 합니다.

5. 대화 메모리 관리

시작하며

여러분이 AI 비서에게 "어제 말한 그 레스토랑 이름이 뭐였지?"라고 물었는데 "무슨 레스토랑을 말씀하시는 건가요?"라는 답변을 받으면 어떤 기분이 드나요? 마치 금붕어와 대화하는 것 같아서 답답하죠.

이런 문제는 AI 에이전트가 대화 맥락을 기억하지 못해서 발생합니다. 각 요청을 독립적으로 처리하면 이전 대화 내용을 전혀 모르게 됩니다.

"그거", "그때", "아까 말한" 같은 대명사나 참조를 이해할 수 없고, 사용자는 매번 모든 정보를 반복해서 말해야 하는 불편함을 겪습니다. 바로 이럴 때 필요한 것이 대화 메모리 관리 시스템입니다.

이전 대화를 기억하고, 적절한 시점에 관련 정보를 불러오며, 메모리가 너무 커지지 않도록 관리하는 시스템이죠.

개요

간단히 말해서, 대화 메모리 관리는 AI 에이전트가 과거 대화를 기억하고 현재 대화에 활용할 수 있도록 하는 시스템입니다. 왜 이것이 필요할까요?

실무에서 사용자와의 자연스러운 대화는 여러 턴에 걸쳐 이어집니다. "파이썬으로 웹 크롤러 만들려고 해" → "어떤 라이브러리 쓰면 돼?" → "그거 설치하는 방법 알려줘" 같은 연속된 대화에서, AI는 "그거"가 앞에서 추천한 라이브러리를 의미한다는 걸 알아야 합니다.

또한 사용자의 선호도나 이전 작업 내용을 기억하면 더 개인화된 서비스를 제공할 수 있습니다. 기존에는 전체 대화를 매번 프롬프트에 포함시켜 토큰 비용이 폭증했다면, 이제는 스마트한 메모리 관리로 필요한 부분만 효율적으로 유지할 수 있습니다.

핵심 특징으로는 첫째, 단기 메모리로 최근 몇 턴의 대화를 빠르게 접근합니다. 둘째, 장기 메모리로 중요한 정보를 영구 저장합니다.

셋째, 메모리 요약 기능으로 오래된 대화를 압축하여 토큰을 절약합니다. 이러한 특징들이 중요한 이유는, 사용자 경험을 크게 향상시키면서도 비용을 관리할 수 있기 때문입니다.

코드 예제

# ai_assistant/memory/conversation_memory.py
from typing import List, Dict, Any, Optional
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage
import json
from datetime import datetime

class HybridMemoryManager:
    """단기 메모리와 장기 메모리를 결합한 하이브리드 메모리 관리자"""

    def __init__(self, llm: ChatOpenAI, max_short_term: int = 10):
        # 단기 메모리: 최근 N개 대화 저장
        self.short_term_memory = ConversationBufferMemory(
            return_messages=True,
            memory_key="chat_history",
            max_token_limit=2000
        )

        # 장기 메모리: 중요한 정보를 요약하여 저장
        self.long_term_memory = ConversationSummaryMemory(
            llm=llm,
            memory_key="summary",
            return_messages=True
        )

        self.max_short_term = max_short_term
        self.conversation_count = 0

        # 영구 저장을 위한 전체 히스토리
        self.full_history: List[Dict[str, Any]] = []

        # 사용자 프로필 (학습된 선호도 등)
        self.user_profile: Dict[str, Any] = {
            "preferences": {},
            "important_facts": []
        }

    def add_interaction(self, user_message: str, ai_response: str, metadata: Optional[Dict] = None):
        """새로운 대화 턴 추가"""
        # 단기 메모리에 추가
        self.short_term_memory.save_context(
            {"input": user_message},
            {"output": ai_response}
        )

        # 전체 히스토리에 추가
        interaction = {
            "timestamp": datetime.now().isoformat(),
            "user": user_message,
            "assistant": ai_response,
            "metadata": metadata or {}
        }
        self.full_history.append(interaction)

        self.conversation_count += 1

        # 일정 턴마다 오래된 대화를 장기 메모리로 이동
        if self.conversation_count % self.max_short_term == 0:
            self._consolidate_memory()

    def _consolidate_memory(self):
        """단기 메모리를 요약하여 장기 메모리로 이동"""
        # 단기 메모리의 내용을 가져옴
        messages = self.short_term_memory.chat_memory.messages

        if len(messages) > 0:
            # 장기 메모리에 요약 추가
            for i in range(0, len(messages), 2):
                if i+1 < len(messages):
                    self.long_term_memory.save_context(
                        {"input": messages[i].content},
                        {"output": messages[i+1].content}
                    )

            # 단기 메모리 초기화 (최근 2-3턴만 유지)
            recent_messages = messages[-4:]  # 최근 2턴 (유저 2 + AI 2)
            self.short_term_memory.chat_memory.messages = recent_messages

    def get_context(self, include_summary: bool = True) -> str:
        """현재 컨텍스트 가져오기 (프롬프트에 포함할 내용)"""
        context_parts = []

        # 장기 메모리 요약 포함
        if include_summary:
            summary = self.long_term_memory.load_memory_variables({})
            if summary.get("summary"):
                context_parts.append(f"이전 대화 요약:\n{summary['summary']}\n")

        # 단기 메모리 (최근 대화) 포함
        short_term = self.short_term_memory.load_memory_variables({})
        if short_term.get("chat_history"):
            context_parts.append("최근 대화:")
            for msg in short_term["chat_history"]:
                role = "사용자" if isinstance(msg, HumanMessage) else "AI"
                context_parts.append(f"{role}: {msg.content}")

        # 사용자 프로필 정보 포함
        if self.user_profile["preferences"] or self.user_profile["important_facts"]:
            context_parts.append("\n사용자 정보:")
            if self.user_profile["preferences"]:
                context_parts.append(f"선호도: {json.dumps(self.user_profile['preferences'], ensure_ascii=False)}")
            if self.user_profile["important_facts"]:
                context_parts.append(f"중요 사실: {', '.join(self.user_profile['important_facts'])}")

        return "\n".join(context_parts)

    def update_user_profile(self, key: str, value: Any):
        """사용자 프로필 업데이트 (선호도 학습)"""
        self.user_profile["preferences"][key] = value

    def add_important_fact(self, fact: str):
        """장기적으로 기억해야 할 중요한 사실 추가"""
        if fact not in self.user_profile["important_facts"]:
            self.user_profile["important_facts"].append(fact)

    def search_history(self, keyword: str) -> List[Dict[str, Any]]:
        """과거 대화에서 키워드 검색"""
        results = []
        for interaction in self.full_history:
            if keyword.lower() in interaction["user"].lower() or keyword.lower() in interaction["assistant"].lower():
                results.append(interaction)
        return results

    def save_to_file(self, filepath: str):
        """메모리를 파일로 저장"""
        data = {
            "full_history": self.full_history,
            "user_profile": self.user_profile,
            "timestamp": datetime.now().isoformat()
        }
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load_from_file(self, filepath: str):
        """파일에서 메모리 로드"""
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        self.full_history = data.get("full_history", [])
        self.user_profile = data.get("user_profile", {"preferences": {}, "important_facts": []})

설명

이것이 하는 일: 하이브리드 메모리 관리자는 사람의 기억 시스템을 모방하여, 최근 대화는 상세히 기억하고(단기 메모리), 오래된 대화는 요약하여 기억하며(장기 메모리), 중요한 정보는 영구적으로 저장합니다(사용자 프로필). 첫 번째로, add_interaction 메서드가 새로운 대화를 기록합니다.

사용자 메시지와 AI 응답을 단기 메모리에 추가하고, 타임스탬프와 메타데이터를 포함하여 전체 히스토리에도 저장합니다. 이렇게 이중으로 저장하는 이유는, 단기 메모리는 빠른 접근을 위해, 전체 히스토리는 나중에 분석이나 검색을 위해 필요하기 때문입니다.

그 다음으로, _consolidate_memory 메서드가 주기적으로 실행되어 메모리를 정리합니다. 10턴마다 오래된 대화를 요약하여 장기 메모리로 옮기고, 단기 메모리는 최근 2-3턴만 유지합니다.

이 과정이 없다면 대화가 길어질수록 프롬프트 크기가 계속 커져서 토큰 제한을 초과하거나 비용이 폭증할 수 있습니다. LangChain의 ConversationSummaryMemory가 자동으로 대화를 요약해주므로 메모리 효율이 높습니다.

세 번째로, get_context 메서드가 현재 프롬프트에 포함할 컨텍스트를 생성합니다. 장기 메모리 요약 + 최근 대화 + 사용자 프로필 정보를 모두 결합하여 AI에게 전달합니다.

이렇게 하면 AI는 "오늘 날씨 알려줘"라는 요청을 받았을 때, 사용자가 이전에 "나는 서울에 살아"라고 말한 것을 기억하고 서울 날씨를 알려줄 수 있습니다. 네 번째로, user_profile을 통해 사용자에 대해 학습한 정보를 저장합니다.

선호도(예: "Python을 선호함", "비주얼 학습자")나 중요한 사실(예: "프로젝트 마감일은 12월 31일")을 기록하여, 매번 물어보지 않아도 개인화된 서비스를 제공할 수 있습니다. 다섯 번째로, search_history와 save_to_file 같은 유틸리티 메서드들이 추가 기능을 제공합니다.

과거 대화에서 특정 주제를 검색하거나, 전체 대화 이력을 파일로 저장하여 나중에 불러올 수 있습니다. 이는 사용자가 세션을 종료했다가 다시 시작해도 이전 대화를 이어갈 수 있게 해줍니다.

여러분이 이 메모리 시스템을 사용하면 AI 비서가 훨씬 더 인간적으로 느껴집니다. "그거", "아까" 같은 대명사를 이해하고, 사용자의 선호도를 기억하며, 오래 전에 나눈 대화도 참조할 수 있습니다.

동시에 스마트한 메모리 관리로 토큰 비용을 절약하고 성능도 유지할 수 있습니다.

실전 팁

💡 max_token_limit을 설정하여 단기 메모리가 너무 커지는 것을 방지하세요. GPT-4의 경우 8K 토큰 제한을 고려하면, 메모리는 2000-3000 토큰 정도로 제한하는 것이 안전합니다.

💡 중요한 정보는 add_important_fact로 명시적으로 저장하세요. "사용자의 생일은 3월 15일"처럼 잊어버리면 안 되는 정보는 요약 과정에서 사라질 수 있으므로 별도 저장이 필요합니다.

💡 대화 히스토리를 주기적으로 파일로 저장하여 백업하세요. 시스템 재시작이나 오류 발생 시에도 대화를 복구할 수 있습니다. 또한 나중에 사용자 행동 분석이나 모델 개선에도 활용할 수 있습니다.

💡 민감한 정보(비밀번호, 개인정보 등)는 메모리에 저장하지 않도록 필터링 로직을 추가하세요. 정규표현식으로 신용카드 번호나 주민등록번호 패턴을 감지하여 자동으로 마스킹할 수 있습니다.

💡 search_history 기능을 활용하여 "지난번에 추천해준 책 뭐였지?" 같은 질문에 답할 수 있습니다. 키워드 검색뿐만 아니라 임베딩 기반 시맨틱 검색을 추가하면 더욱 강력해집니다.

6. 에이전트 간 협업 구현

시작하며

여러분이 "내일 비 온다면 우산을 사라고 할 일 목록에 추가해줘"라는 요청을 AI 비서에게 했다고 상상해보세요. 이 요청을 처리하려면 날씨 에이전트가 일기예보를 확인하고, 그 결과를 바탕으로 할 일 관리 에이전트가 작업을 추가해야 합니다.

이런 문제는 단일 에이전트로는 해결할 수 없습니다. 여러 전문 에이전트가 순차적으로 또는 조건부로 협력해야 하는 복잡한 워크플로우가 필요합니다.

날씨 에이전트의 출력이 할 일 에이전트의 입력이 되고, 중간에 조건 분기(비가 온다면)도 처리해야 하죠. 바로 이럴 때 필요한 것이 에이전트 간 협업 시스템입니다.

여러 에이전트를 체계적으로 연결하고, 데이터를 전달하며, 조건부 로직을 처리하는 워크플로우 엔진이 필요합니다.

개요

간단히 말해서, 에이전트 간 협업은 여러 전문 에이전트가 서로 통신하고 작업을 주고받으며 복잡한 태스크를 완수하는 시스템입니다. 왜 이것이 필요할까요?

실무에서 사용자의 요청은 여러 단계와 조건을 포함합니다. "이메일에서 회의 일정 추출하고, 캘린더에 추가하고, 참석자들에게 확인 메일 보내기" 같은 요청은 이메일 처리 에이전트 → 일정 관리 에이전트 → 이메일 발송 에이전트가 순차적으로 협력해야 합니다.

각 에이전트는 독립적으로는 훌륭하지만, 함께 작동하지 못하면 복잡한 작업을 처리할 수 없습니다. 기존에는 각 에이전트가 독립적으로 작동하여 결과를 수동으로 다음 에이전트에 전달했다면, 이제는 자동화된 워크플로우로 에이전트들이 매끄럽게 협업할 수 있습니다.

핵심 특징으로는 첫째, 데이터 파이프라인으로 한 에이전트의 출력이 다음 에이전트의 입력으로 자동 전달됩니다. 둘째, 조건부 실행으로 특정 조건에 따라 다른 에이전트를 호출합니다.

셋째, 병렬 처리로 독립적인 작업은 동시에 실행하여 속도를 높입니다. 이러한 특징들이 중요한 이유는, 복잡한 비즈니스 로직을 구현할 수 있기 때문입니다.

코드 예제

# ai_assistant/workflow/agent_workflow.py
from typing import Dict, Any, List, Callable, Optional
from enum import Enum
import asyncio

class WorkflowNodeType(Enum):
    """워크플로우 노드 타입"""
    AGENT = "agent"  # 에이전트 실행
    CONDITION = "condition"  # 조건 분기
    PARALLEL = "parallel"  # 병렬 실행
    MERGE = "merge"  # 결과 병합

class WorkflowNode:
    """워크플로우의 단일 노드"""

    def __init__(self, node_id: str, node_type: WorkflowNodeType,
                 agent_type: Optional[str] = None,
                 condition_func: Optional[Callable] = None):
        self.node_id = node_id
        self.node_type = node_type
        self.agent_type = agent_type
        self.condition_func = condition_func
        self.next_nodes: List[str] = []  # 다음 노드들
        self.next_on_true: Optional[str] = None  # 조건이 참일 때
        self.next_on_false: Optional[str] = None  # 조건이 거짓일 때

class AgentWorkflow:
    """에이전트 간 협업을 조율하는 워크플로우 엔진"""

    def __init__(self, orchestrator):
        self.orchestrator = orchestrator
        self.nodes: Dict[str, WorkflowNode] = {}
        self.start_node: Optional[str] = None

    def add_agent_node(self, node_id: str, agent_type: str) -> WorkflowNode:
        """에이전트 실행 노드 추가"""
        node = WorkflowNode(node_id, WorkflowNodeType.AGENT, agent_type=agent_type)
        self.nodes[node_id] = node
        if self.start_node is None:
            self.start_node = node_id
        return node

    def add_condition_node(self, node_id: str, condition_func: Callable) -> WorkflowNode:
        """조건 분기 노드 추가"""
        node = WorkflowNode(node_id, WorkflowNodeType.CONDITION, condition_func=condition_func)
        self.nodes[node_id] = node
        return node

    def connect(self, from_node: str, to_node: str, condition: Optional[str] = None):
        """노드 연결"""
        if from_node not in self.nodes:
            raise ValueError(f"노드 '{from_node}'을 찾을 수 없습니다")

        node = self.nodes[from_node]
        if condition == "true":
            node.next_on_true = to_node
        elif condition == "false":
            node.next_on_false = to_node
        else:
            node.next_nodes.append(to_node)

    async def execute(self, initial_data: Dict[str, Any]) -> Dict[str, Any]:
        """워크플로우 실행"""
        if not self.start_node:
            raise ValueError("시작 노드가 설정되지 않았습니다")

        context = {"initial_data": initial_data, "results": {}}
        return await self._execute_node(self.start_node, context)

    async def _execute_node(self, node_id: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """단일 노드 실행"""
        if node_id not in self.nodes:
            return context

        node = self.nodes[node_id]

        if node.node_type == WorkflowNodeType.AGENT:
            # 에이전트 실행
            result = await self.orchestrator.delegate_task(
                node.agent_type,
                context.get("current_data", context["initial_data"])
            )
            context["results"][node_id] = result
            context["current_data"] = result

            # 다음 노드 실행
            if node.next_nodes:
                for next_node in node.next_nodes:
                    context = await self._execute_node(next_node, context)

        elif node.node_type == WorkflowNodeType.CONDITION:
            # 조건 평가
            condition_result = node.condition_func(context["current_data"])
            context["results"][node_id] = {"condition": condition_result}

            # 조건에 따라 다음 노드 선택
            if condition_result and node.next_on_true:
                context = await self._execute_node(node.next_on_true, context)
            elif not condition_result and node.next_on_false:
                context = await self._execute_node(node.next_on_false, context)

        return context

# 사용 예시: "내일 비 온다면 우산 사기를 할 일에 추가"
async def create_weather_todo_workflow(orchestrator):
    """날씨 기반 할 일 추가 워크플로우"""
    workflow = AgentWorkflow(orchestrator)

    # 1. 날씨 조회 노드
    weather_node = workflow.add_agent_node("check_weather", "weather")

    # 2. 조건 확인 노드 (비가 오는지 체크)
    def is_rainy(data: Dict[str, Any]) -> bool:
        # 날씨 에이전트 결과에서 비 여부 확인
        answer = data.get("answer", "")
        return "비" in answer or "rain" in answer.lower()

    condition_node = workflow.add_condition_node("check_rain", is_rainy)

    # 3. 할 일 추가 노드
    todo_node = workflow.add_agent_node("add_todo", "todo")

    # 노드 연결
    workflow.connect("check_weather", "check_rain")
    workflow.connect("check_rain", "add_todo", condition="true")

    return workflow

# 실행
# workflow = await create_weather_todo_workflow(orchestrator)
# result = await workflow.execute({"query": "내일 서울 날씨 확인하고 비 오면 우산 사기 할 일 추가"})

설명

이것이 하는 일: 에이전트 워크플로우 시스템은 여러 전문 에이전트를 체계적으로 연결하여 복잡한 비즈니스 로직을 구현합니다. 마치 공장의 생산 라인처럼, 각 작업(에이전트)이 순서대로 처리되고, 중간에 품질 검사(조건 확인)를 거쳐, 최종 제품(결과)을 만들어냅니다.

첫 번째로, WorkflowNode 클래스가 워크플로우의 각 단계를 표현합니다. 에이전트 실행 노드, 조건 분기 노드 등 다양한 타입이 있으며, 각 노드는 다음에 실행될 노드들을 가리킵니다.

이렇게 그래프 구조로 설계하면 복잡한 분기와 순환도 표현할 수 있습니다. 그 다음으로, AgentWorkflow 클래스가 전체 워크플로우를 관리합니다.

add_agent_node로 에이전트 실행 단계를 추가하고, add_condition_node로 조건 분기를 추가하며, connect로 노드들을 연결합니다. 이 API는 매우 직관적이어서 복잡한 워크플로우도 쉽게 정의할 수 있습니다.

세 번째로, execute 메서드가 워크플로우를 실행합니다. 시작 노드부터 차례대로 노드를 실행하며, context 딕셔너리에 모든 중간 결과를 저장합니다.

각 노드의 결과는 다음 노드의 입력이 되므로, 데이터가 자연스럽게 흐릅니다. 네 번째로, _execute_node 메서드가 각 노드 타입에 맞게 처리합니다.

AGENT 타입이면 해당 에이전트를 호출하고, CONDITION 타입이면 조건 함수를 실행하여 참/거짓에 따라 다른 경로로 분기합니다. 이 재귀적 구조 덕분에 중첩된 조건이나 복잡한 흐름도 처리할 수 있습니다.

다섯 번째로, create_weather_todo_workflow 예시가 실제 사용법을 보여줍니다. 날씨 조회 → 비 여부 확인 → 조건이 참이면 할 일 추가 라는 3단계 워크플로우를 정의합니다.

is_rainy 함수가 날씨 에이전트의 결과에서 "비"라는 키워드를 찾아 조건을 평가합니다. 여러분이 이 워크플로우 시스템을 사용하면 복잡한 멀티 에이전트 시나리오를 선언적으로 정의할 수 있습니다.

"이것 한 다음 저것 해"라는 복잡한 로직을 코드로 명확히 표현할 수 있고, 나중에 워크플로우를 수정하거나 확장하기도 쉽습니다. 또한 워크플로우를 JSON이나 YAML로 저장하여 비개발자도 편집할 수 있게 만들 수도 있습니다.

실전 팁

💡 복잡한 워크플로우는 시각화 도구로 그려보세요. Graphviz나 Mermaid를 사용하여 노드와 연결을 다이어그램으로 표현하면 로직을 이해하고 디버깅하기 훨씬 쉽습니다.

💡 각 노드의 실행 시간과 결과를 로깅하세요. 어떤 에이전트에서 병목이 발생하는지, 조건 분기가 예상대로 동작하는지 추적할 수 있습니다.

💡 워크플로우 실행 중 에러가 발생하면 어떻게 할지 전략을 세우세요. 전체 워크플로우를 중단할지, 해당 노드만 스킵하고 계속 진행할지, 재시도할지 결정해야 합니다.

💡 병렬 처리가 가능한 노드는 asyncio.gather를 사용하여 동시 실행하세요. 예를 들어 날씨 조회와 뉴스 검색은 독립적이므로 동시에 실행하면 시간을 절약할 수 있습니다.

💡 워크플로우를 재사용 가능한 템플릿으로 만드세요. "날씨 기반 할 일 추가"를 하나의 템플릿으로 저장하고, 다른 도시나 다른 조건으로 쉽게 변형할 수 있게 설계하면 좋습니다.

7. 에러 핸들링과 재시도 로직

시작하며

여러분의 AI 비서가 날씨 API를 호출했는데 네트워크 오류로 실패했다고 상상해보세요. 사용자에게 "죄송합니다.

오류가 발생했습니다"라고만 말하고 끝나면 어떨까요? 사용자는 답답하고, 서비스 신뢰도도 떨어집니다.

이런 문제는 실제 운영 환경에서 매우 자주 발생합니다. 외부 API가 일시적으로 다운되거나, 네트워크가 불안정하거나, Rate Limit에 걸리거나, 타임아웃이 발생하는 등 수많은 실패 시나리오가 있습니다.

이런 일시적인 오류를 제대로 처리하지 못하면 시스템이 불안정해지고 사용자 경험이 나빠집니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링과 재시도 로직입니다.

일시적인 오류는 자동으로 재시도하고, 영구적인 오류는 명확한 메시지로 사용자에게 알리며, 모든 오류를 로깅하여 나중에 분석할 수 있도록 합니다.

개요

간단히 말해서, 에러 핸들링과 재시도 로직은 시스템에서 발생하는 다양한 오류를 지능적으로 처리하여 안정성과 신뢰성을 높이는 메커니즘입니다. 왜 이것이 필요할까요?

실무에서 완벽한 시스템은 없습니다. 외부 서비스는 언제든 실패할 수 있고, 네트워크는 불안정할 수 있습니다.

중요한 것은 이런 실패를 어떻게 처리하느냐입니다. 똑똑한 재시도 로직으로 90%의 일시적 오류는 자동으로 복구할 수 있고, 나머지 10%의 영구적 오류는 사용자에게 명확히 알려 대안을 찾도록 도울 수 있습니다.

기존에는 모든 예외를 단순히 try-except로 잡아서 무시하거나 프로그램을 종료했다면, 이제는 오류 타입별로 다르게 대응하고, 복구 가능한 오류는 자동으로 재시도하며, 모든 오류를 체계적으로 기록할 수 있습니다. 핵심 특징으로는 첫째, 지능적 재시도로 지수 백오프와 최대 재시도 횟수를 사용합니다.

둘째, 오류 분류로 일시적 오류와 영구적 오류를 구분합니다. 셋째, 폴백 메커니즘으로 주 서비스 실패 시 대안 서비스를 사용합니다.

이러한 특징들이 중요한 이유는, 시스템의 가용성과 복원력을 크게 향상시키기 때문입니다.

코드 예제

# ai_assistant/utils/error_handling.py
from typing import Callable, Any, Optional, Type
import asyncio
import logging
from functools import wraps
from datetime import datetime
import traceback

# 로거 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RetryableError(Exception):
    """재시도 가능한 오류"""
    pass

class NonRetryableError(Exception):
    """재시도 불가능한 영구적 오류"""
    pass

class ErrorHandler:
    """에러 핸들링 및 재시도 로직을 담당하는 클래스"""

    @staticmethod
    async def retry_with_backoff(
        func: Callable,
        max_retries: int = 3,
        initial_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_base: float = 2.0,
        retryable_exceptions: tuple = (RetryableError, ConnectionError, TimeoutError)
    ) -> Any:
        """
        지수 백오프를 사용한 재시도 로직

        Args:
            func: 실행할 함수
            max_retries: 최대 재시도 횟수
            initial_delay: 초기 대기 시간 (초)
            max_delay: 최대 대기 시간 (초)
            exponential_base: 지수 증가 배수
            retryable_exceptions: 재시도할 예외 타입들
        """
        delay = initial_delay
        last_exception = None

        for attempt in range(max_retries + 1):
            try:
                # 함수 실행
                if asyncio.iscoroutinefunction(func):
                    result = await func()
                else:
                    result = func()

                # 성공 시 로깅
                if attempt > 0:
                    logger.info(f"재시도 {attempt}번 만에 성공: {func.__name__}")

                return result

            except retryable_exceptions as e:
                last_exception = e

                if attempt < max_retries:
                    # 재시도할 것임을 로깅
                    logger.warning(
                        f"재시도 가능한 오류 발생 ({attempt + 1}/{max_retries}): "
                        f"{type(e).__name__}: {str(e)}. "
                        f"{delay}초 후 재시도합니다."
                    )

                    # 지수 백오프 대기
                    await asyncio.sleep(delay)

                    # 다음 대기 시간 계산 (지수 증가, 최대값 제한)
                    delay = min(delay * exponential_base, max_delay)
                else:
                    # 최대 재시도 횟수 초과
                    logger.error(
                        f"최대 재시도 횟수 {max_retries}회 초과. 마지막 오류: {str(e)}"
                    )

            except NonRetryableError as e:
                # 재시도 불가능한 오류는 즉시 발생
                logger.error(f"재시도 불가능한 오류 발생: {type(e).__name__}: {str(e)}")
                raise

            except Exception as e:
                # 예상치 못한 오류
                logger.error(
                    f"예상치 못한 오류 발생: {type(e).__name__}: {str(e)}\n"
                    f"{traceback.format_exc()}"
                )
                raise

        # 모든 재시도 실패 시 마지막 예외 발생
        raise last_exception

def with_retry(max_retries: int = 3, initial_delay: float = 1.0):
    """재시도 로직을 적용하는 데코레이터"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            async def target():
                return await func(*args, **kwargs)

            return await ErrorHandler.retry_with_backoff(
                target,
                max_retries=max_retries,
                initial_delay=initial_delay
            )
        return wrapper
    return decorator

class CircuitBreaker:
    """
    서킷 브레이커 패턴 구현
    연속적인 실패가 발생하면 일정 시간 동안 요청을 차단
    """

    def __init__(self, failure_threshold: int = 5, timeout: float = 60.0):
        self.failure_threshold = failure_threshold  # 연속 실패 임계값
        self.timeout = timeout  # 차단 시간 (초)
        self.failure_count = 0
        self.last_failure_time: Optional[float] = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    async def call(self, func: Callable) -> Any:
        """서킷 브레이커를 통해 함수 호출"""
        # OPEN 상태 확인 (차단 중)
        if self.state == "OPEN":
            if datetime.now().timestamp() - self.last_failure_time < self.timeout:
                raise NonRetryableError("서킷 브레이커가 열려 있습니다. 잠시 후 다시 시도해주세요.")
            else:
                # 타임아웃 지났으면 HALF_OPEN으로 전환
                self.state = "HALF_OPEN"
                logger.info("서킷 브레이커 상태: HALF_OPEN (테스트 중)")

        try:
            # 함수 실행
            result = await func() if asyncio.iscoroutinefunction(func) else func()

            # 성공 시 카운터 리셋
            self.failure_count = 0
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                logger.info("서킷 브레이커 상태: CLOSED (정상)")

            return result

        except Exception as e:
            # 실패 시 카운터 증가
            self.failure_count += 1
            self.last_failure_time = datetime.now().timestamp()

            # 임계값 초과 시 OPEN으로 전환
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                logger.error(
                    f"서킷 브레이커 상태: OPEN (차단됨). "
                    f"{self.timeout}초 동안 요청이 차단됩니다."
                )

            raise

# 사용 예시
@with_retry(max_retries=3, initial_delay=1.0)
async def call_external_api(url: str) -> dict:
    """외부 API 호출 (재시도 로직 적용)"""
    import aiohttp

    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=10) as response:
            if response.status == 429:  # Rate Limit
                raise RetryableError("Rate limit exceeded")
            elif response.status >= 500:  # Server Error
                raise RetryableError(f"Server error: {response.status}")
            elif response.status >= 400:  # Client Error
                raise NonRetryableError(f"Client error: {response.status}")

            return await response.json()

설명

이것이 하는 일: 에러 핸들링 시스템은 다양한 실패 시나리오를 지능적으로 처리하여 시스템이 일시적인 문제에서 자동으로 복구하고, 영구적인 문제는 명확히 사용자에게 전달합니다. 마치 운전 중 작은 충격은 서스펜션이 흡수하고, 큰 사고는 에어백이 보호하는 것처럼, 다층 방어 시스템을 구축합니다.

첫 번째로, retry_with_backoff 메서드가 핵심 재시도 로직을 구현합니다. 지수 백오프(exponential backoff)를 사용하여 재시도 간격을 점점 늘립니다.

첫 번째 재시도는 1초 후, 두 번째는 2초 후, 세 번째는 4초 후 이런 식으로 증가합니다. 이렇게 하는 이유는, 서버가 과부하 상태일 때 모든 클라이언트가 동시에 재시도하면 상황이 더 악화되기 때문입니다.

간격을 늘리면 서버가 복구될 시간을 줍니다. 그 다음으로, 오류를 RetryableError(재시도 가능)와 NonRetryableError(재시도 불가능)로 분류합니다.

네트워크 타임아웃이나 서버 5xx 에러는 일시적일 가능성이 높으므로 재시도하지만, 클라이언트 4xx 에러(잘못된 요청)는 재시도해도 소용없으므로 즉시 실패 처리합니다. 이런 분류 덕분에 무의미한 재시도를 피하고 빠르게 실패할 수 있습니다.

세 번째로, with_retry 데코레이터가 재시도 로직을 함수에 쉽게 적용할 수 있게 합니다. @with_retry를 함수 위에 붙이기만 하면 자동으로 재시도 기능이 추가됩니다.

이렇게 하면 비즈니스 로직과 에러 처리 로직을 깔끔하게 분리할 수 있어 코드 가독성이 향상됩니다. 네 번째로, CircuitBreaker 클래스가 서킷 브레이커 패턴을 구현합니다.

연속해서 5번 실패하면 "OPEN" 상태로 전환되어 60초 동안 모든 요청을 차단합니다. 이는 죽은 서비스를 계속 호출하여 리소스를 낭비하는 것을 방지합니다.

60초 후에는 "HALF_OPEN" 상태로 전환되어 테스트 요청을 보내고, 성공하면 "CLOSED"로 돌아가 정상 작동합니다. 다섯 번째로, 모든 오류를 상세히 로깅합니다.

오류 타입, 메시지, 스택 트레이스, 재시도 횟수 등을 기록하여 나중에 문제를 분석하고 개선할 수 있습니다. 프로덕션 환경에서는 이런 로그를 Sentry나 CloudWatch 같은 모니터링 시스템으로 전송하여 실시간으로 추적할 수 있습니다.

여러분이 이 에러 핸들링 시스템을 사용하면 시스템의 가용성이 크게 향상됩니다. 네트워크가 일시적으로 불안정해도 자동으로 재시도하여 사용자는 문제를 느끼지 못할 수 있고, 외부 서비스가 다운되어도 서킷 브레이커가 리소스 낭비를 막아줍니다.

또한 명확한 에러 메시지로 사용자 경험도 개선됩니다.

실전 팁

💡 max_retries는 신중하게 설정하세요. 너무 많으면 실패한 요청에 너무 오래 매달리고, 너무 적으면 일시적 오류를 복구하지 못합니다. 일반적으로 3-5회가 적당합니다.

💡 재시도할 예외 타입을 명확히 정의하세요. 모든 Exception을 재시도하면 프로그래밍 버그도 재시도하게 되어 문제를 숨길 수 있습니다. ConnectionError, TimeoutError 같은 명확한 타입만 재시도하세요.

💡 외부 API 호출 시 항상 타임아웃을 설정하세요. timeout=10처럼 명시하지 않으면 무한 대기할 수 있습니다. API의 SLA를 고려하여 적절한 타임아웃을 설정하세요.

💡 서킷 브레이커의 상태 변화를 모니터링하세요. OPEN 상태로 자주 전환된다면 외부 서비스에 문제가 있거나 failure_threshold가 너무 낮게 설정된 것일 수 있습니다.

💡 프로덕션 환경에서는 재시도 로직에 jitter(무작위 지연)를 추가하세요. delay에 ±20% 정도의 랜덤 값을 더하면 여러 클라이언트의 재시도가 동시에 몰리는 것을 방지할 수 있습니다.

8. 사용자 인터페이스 구현

시작하며

여러분이 강력한 AI 에이전트 시스템을 만들었는데, 사용자가 Python 코드를 직접 작성해야만 사용할 수 있다면 어떨까요? 일반 사용자에게는 전혀 쓸모없는 시스템이 되고 말 것입니다.

이런 문제는 훌륭한 기능을 가진 많은 프로젝트가 실패하는 이유입니다. 아무리 뛰어난 기술이라도 사용하기 어려우면 아무도 쓰지 않습니다.

특히 AI 비서는 누구나 쉽게 대화하듯이 사용할 수 있어야 하는데, 복잡한 API 호출이나 코드 작성을 요구하면 대중화될 수 없습니다. 바로 이럴 때 필요한 것이 직관적인 사용자 인터페이스입니다.

터미널에서 간단한 명령어나 대화만으로 모든 기능을 사용할 수 있는 CLI 인터페이스를 만들어, 개발자가 아닌 일반 사용자도 쉽게 접근할 수 있게 합니다.

개요

간단히 말해서, 사용자 인터페이스는 복잡한 AI 에이전트 시스템을 누구나 쉽게 사용할 수 있도록 하는 대화형 CLI 프로그램입니다. 왜 이것이 필요할까요?

실무에서 좋은 소프트웨어는 기능뿐만 아니라 사용성도 중요합니다. 사용자가 자연스럽게 질문하고, 실시간으로 응답을 받으며, 명령어 히스토리를 확인하고, 설정을 변경할 수 있는 인터페이스가 필요합니다.

CLI는 웹 UI보다 간단하면서도 개발자와 파워 유저에게 친숙한 선택입니다. 기존에는 API를 직접 호출하거나 Python 스크립트를 수정해야 했다면, 이제는 터미널을 열고 "오늘 날씨 알려줘"라고 타이핑하기만 하면 됩니다.

핵심 특징으로는 첫째, 대화형 인터페이스로 자연스러운 대화를 나눌 수 있습니다. 둘째, 명령어 시스템으로 /help, /history 같은 특수 명령을 제공합니다.

셋째, 실시간 스트리밍으로 AI 응답이 생성되는 동안 단어별로 표시됩니다. 이러한 특징들이 중요한 이유는, 사용자 경험을 크게 향상시켜 실제로 사용되는 제품을 만들 수 있기 때문입니다.

코드 예제

# ai_assistant/cli/interface.py
import asyncio
import sys
from typing import Optional
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.spinner import Spinner
from rich.live import Live

class AIAssistantCLI:
    """AI 비서를 위한 대화형 CLI 인터페이스"""

    def __init__(self, main_agent, config: dict):
        self.main_agent = main_agent
        self.config = config
        self.console = Console()

        # 명령어 히스토리 파일
        self.session = PromptSession(
            history=FileHistory('.ai_assistant_history'),
            auto_suggest=AutoSuggestFromHistory()
        )

        self.running = True
        self.commands = {
            '/help': self.show_help,
            '/history': self.show_history,
            '/clear': self.clear_screen,
            '/config': self.show_config,
            '/exit': self.exit_app,
            '/quit': self.exit_app
        }

    def show_welcome(self):
        """환영 메시지 표시"""
        welcome_text = """
# 🤖 AI 비서에 오신 것을 환영합니다!

무엇을 도와드릴까요? 자연스럽게 질문해주세요.

**명령어:**
- `/help` - 도움말 표시
- `/history` - 대화 히스토리 보기
- `/config` - 설정 확인
- `/exit` 또는 `/quit` - 종료
        """
        self.console.print(Panel(Markdown(welcome_text), border_style="blue"))

    async def run(self):
        """CLI 메인 루프"""
        self.show_welcome()

        while self.running:
            try:
                # 사용자 입력 받기
                user_input = await asyncio.get_event_loop().run_in_executor(
                    None,
                    lambda: self.session.prompt('\n💬 You: ', multiline=False)
                )

                # 빈 입력 무시
                if not user_input.strip():
                    continue

                # 명령어 처리
                if user_input.startswith('/'):
                    await self.handle_command(user_input)
                    continue

                # AI 응답 생성 및 표시
                await self.get_ai_response(user_input)

            except KeyboardInterrupt:
                self.console.print("\n\n👋 안녕히 가세요!", style="yellow")
                break
            except EOFError:
                break
            except Exception as e:
                self.console.print(f"[red]오류 발생: {str(e)}[/red]")

    async def handle_command(self, command: str):
        """특수 명령어 처리"""
        cmd_parts = command.split()
        cmd_name = cmd_parts[0]

        if cmd_name in self.commands:
            await self.commands[cmd_name]()
        else:
            self.console.print(f"[yellow]알 수 없는 명령어: {cmd_name}[/yellow]")
            self.console.print("사용 가능한 명령어는 /help를 입력하세요.")

    async def get_ai_response(self, user_input: str):
        """AI 응답 가져오기 및 표시"""
        # 스피너와 함께 "생각 중..." 표시
        with self.console.status("[bold green]AI가 생각하는 중...", spinner="dots"):
            try:
                # 메인 에이전트 호출
                result = await self.main_agent.process({"message": user_input})
                response = result.get("response", "응답을 생성할 수 없습니다.")
            except Exception as e:
                response = f"오류가 발생했습니다: {str(e)}"
                self.console.print(f"[red]{response}[/red]")
                return

        # 응답 표시 (마크다운 렌더링)
        self.console.print("\n🤖 AI:", style="bold cyan")
        self.console.print(Panel(Markdown(response), border_style="cyan"))

    async def show_help(self):
        """도움말 표시"""
        help_text = """
# 📖 도움말

## 명령어 목록:
- `/help` - 이 도움말을 표시합니다
- `/history` - 대화 히스토리를 확인합니다
- `/clear` - 화면을 지웁니다
- `/config` - 현재 설정을 표시합니다
- `/exit` 또는 `/quit` - 프로그램을 종료합니다

## 사용 팁:
- 자연스러운 언어로 질문하세요
- 이전 대화 내용을 참조할 수 있습니다
- 위/아래 화살표로 이전 입력을 불러올 수 있습니다
        """
        self.console.print(Markdown(help_text))

    async def show_history(self):
        """대화 히스토리 표시"""
        history = self.main_agent.conversation_history

        if not history:
            self.console.print("[yellow]아직 대화 히스토리가 없습니다.[/yellow]")
            return

        self.console.print("\n📜 대화 히스토리:", style="bold")
        for i, interaction in enumerate(history[-10:], 1):  # 최근 10개만
            self.console.print(f"\n{i}. 사용자: {interaction['user']}")
            self.console.print(f"   AI: {interaction['assistant'][:100]}...")

    async def clear_screen(self):
        """화면 지우기"""
        self.console.clear()
        self.show_welcome()

    async def show_config(self):
        """현재 설정 표시"""
        config_text = f"""
# ⚙️ 현재 설정

- **모델**: {self.config.get('model', 'N/A')}
- **Temperature**: {self.config.get('temperature', 'N/A')}
- **Max Tokens**: {self.config.get('max_tokens', 'N/A')}
        """
        self.console.print(Markdown(config_text))

    async def exit_app(self):
        """프로그램 종료"""
        self.console.print("\n👋 안녕히 가세요!", style="bold green")
        self.running = False

# 실행 예시
async def main():
    # 설정 로드
    config = {
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 2000
    }

    # 메인 에이전트 초기화 (실제로는 이전에 구현한 MainAgent 사용)
    # main_agent = MainAgent(...)

    # CLI 시작
    # cli = AIAssistantCLI(main_agent, config)
    # await cli.run()

if __name__ == "__main__":
    asyncio.run(main())

설명

이것이 하는 일: AI 비서 CLI는 사용자가 터미널에서 자연스럽게 AI와 대화할 수 있는 인터페이스를 제공합니다. 마치 챗GPT 웹 인터페이스처럼 편리하지만, 터미널 환경에서 작동하여 개발자와 파워 유저에게 적합합니다.

첫 번째로, PromptSession을 사용하여 고급 입력

#Python#AI Agent#LangChain#OpenAI#Multi-Agent#python

Python으로 AI 에이전트 만들기 12편 - 실전 프로젝트: 나만의 AI 비서 완성

# Python으로 AI 에이전트 만들기 12편 - 실전 프로젝트: 나만의 AI 비서 완성 지금까지 배운 모든 개념을 활용하여 실제 동작하는 AI 비서를 완성해보는 실전 프로젝트입니다. 멀티 에이전트 시스템, 도구 통합, 메모리 관리까지 모든 것을 종합하여 실무에서 바로 사용할 수 있는 수준의 AI 비서를 만들어봅니다. --- ## 목차 1. [프로젝트 아키텍처 설계](#프로젝트-아키텍처-설계) 2. [메인 에이전트 구현](#메인-에이전트-구현) 3. [전문 에이전트들 구성](#전문-에이전트들-구성) 4. [도구 통합 시스템](#도구-통합-시스템) 5. [대화 메모리 관리](#대화-메모리-관리) 6. [에이전트 간 협업 구현](#에이전트-간-협업-구현) 7. [에러 핸들링과 재시도 로직](#에러-핸들링과-재시도-로직) 8. [사용자 인터페이스 구현](#사용자-인터페이스-구현) --- ## 1. 프로젝트 아키텍처 설계 ### 시작하며 여러분이 AI 에이전트를 만들 때 "일단 코드부터 짜고 보자"라고 시작한 적 있나요? 처음엔 간단하게 시작했지만, 기능이 하나씩 추가될 때마다 코드가 엉망이 되고, 나중엔 어디서부터 수정해야 할지 막막해지는 경험 말이죠. 이런 문제는 실제 개발 현장에서 너무나 자주 발생합니다. 특히 AI 에이전트처럼 여러 컴포넌트가 복잡하게 얽혀있는 시스템에서는 더욱 그렇습니다. 처음부터 명확한 아키텍처 없이 시작하면, 나중에 새로운 에이전트를 추가하거나 도구를 통합할 때마다 전체 코드를 뜯어고쳐야 하는 상황이 옵니다. 바로 이럴 때 필요한 것이 체계적인 프로젝트 아키텍처 설계입니다. 확장 가능하고 유지보수하기 쉬운 구조를 처음부터 만들어두면, 나중에 기능을 추가할 때도 기존 코드를 건드리지 않고 새로운 모듈만 추가하면 됩니다. ### 개요 간단히 말해서, 프로젝트 아키텍처 설계란 AI 비서를 구성하는 모든 컴포넌트들의 역할과 관계를 미리 정의하는 것입니다. 왜 이것이 필요할까요? 실무에서 AI 에이전트 프로젝트는 끊임없이 진화합니다. 오늘은 날씨 정보만 알려주던 비서가 내일은 일정 관리를 해야 하고, 다음 주엔 이메일까지 보내야 할 수 있습니다. 이럴 때마다 코드 전체를 뜯어고치는 것은 비효율적이죠. 좋은 아키텍처는 이런 확장을 쉽게 만들어줍니다. 기존에는 모든 기능을 하나의 큰 클래스에 때려 넣었다면, 이제는 각 컴포넌트를 독립적인 모듈로 분리하여 관리할 수 있습니다. 핵심 특징으로는 첫째, 모듈화된 구조로 각 컴포넌트가 독립적으로 동작합니다. 둘째, 계층적 설계로 상위 레벨과 하위 레벨이 명확히 구분됩니다. 셋째, 인터페이스 기반 설계로 구현체를 쉽게 교체할 수 있습니다. 이러한 특징들이 중요한 이유는, 프로젝트가 커질수록 유지보수와 확장이 기하급수적으로 어려워지기 때문입니다. ### 코드 예제 ```python # ai_assistant/architecture/base.py from abc import ABC, abstractmethod from typing import Dict, List, Any from dataclasses import dataclass @dataclass class AgentConfig: """에이전트 설정을 담는 데이터 클래스""" name: str description: str model: str = "gpt-4" temperature: float = 0.7 max_tokens: int = 1000 class BaseAgent(ABC): """모든 에이전트가 상속받을 기본 클래스""" def __init__(self, config: AgentConfig): self.config = config self.tools: List[Any] = [] @abstractmethod async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: """에이전트의 핵심 처리 로직 - 각 에이전트가 구현해야 함""" pass def add_tool(self, tool: Any) -> None: """에이전트에 도구를 추가""" self.tools.append(tool) class AgentOrchestrator: """여러 에이전트를 조율하는 오케스트레이터""" def __init__(self): self.agents: Dict[str, BaseAgent] = {} def register_agent(self, agent_type: str, agent: BaseAgent): """새로운 에이전트를 시스템에 등록""" self.agents[agent_type] = agent async def delegate_task(self, task_type: str, data: Dict[str, Any]): """작업을 적절한 에이전트에게 위임""" if task_type in self.agents: return await self.agents[task_type].process(data) raise ValueError(f"No agent found for task type: {task_type}") ``` ### 설명 이것이 하는 일: 이 아키텍처는 AI 비서를 구성하는 모든 에이전트들이 따라야 할 기본 규칙과 구조를 정의합니다. 새로운 에이전트를 추가할 때 전체 시스템을 수정하지 않고도 확장할 수 있도록 설계되었습니다. 첫 번째로, AgentConfig 데이터 클래스는 각 에이전트의 설정 정보를 표준화된 방식으로 관리합니다. 이렇게 하면 모든 에이전트가 동일한 방식으로 설정되어 혼란을 방지할 수 있습니다. name, description 같은 기본 정보부터 model, temperature 같은 AI 모델 파라미터까지 체계적으로 관리됩니다. 그 다음으로, BaseAgent 추상 클래스가 실행되면서 모든 에이전트가 반드시 구현해야 할 process 메서드를 정의합니다. 이것이 핵심인데, 어떤 에이전트를 만들든 간에 process 메서드만 구현하면 시스템에 통합될 수 있습니다. 또한 add_tool 메서드를 통해 각 에이전트에 필요한 도구들을 동적으로 추가할 수 있습니다. 마지막으로, AgentOrchestrator가 여러 에이전트를 관리하고 작업을 적절한 에이전트에게 위임합니다. register_agent로 새 에이전트를 등록하고, delegate_task로 작업을 분배하는 중앙 집중식 관리 시스템입니다. 이렇게 하면 에이전트가 10개든 100개든 간에 체계적으로 관리할 수 있습니다. 여러분이 이 아키텍처를 사용하면 새로운 기능을 추가할 때마다 기존 코드를 수정하지 않고 새로운 클래스만 추가하면 됩니다. 코드의 재사용성이 높아지고, 버그가 발생해도 문제가 있는 모듈만 수정하면 되므로 유지보수가 훨씬 쉬워집니다. 또한 각 에이전트를 독립적으로 테스트할 수 있어 품질 관리도 용이합니다. ### 실전 팁 💡 BaseAgent를 상속받을 때는 반드시 process 메서드를 구현해야 합니다. 이를 잊으면 인스턴스 생성 시 에러가 발생하니 주의하세요. 💡 AgentConfig는 dataclass를 사용하여 타입 힌팅과 기본값 설정이 쉽습니다. 필요한 설정 항목이 추가되면 이 클래스만 수정하면 됩니다. 💡 에이전트 등록 시 의미 있는 agent_type 이름을 사용하세요. "agent1", "agent2" 대신 "weather_agent", "calendar_agent"처럼 명확한 이름을 쓰면 코드 가독성이 크게 향상됩니다. 💡 오케스트레이터에 에이전트를 등록하기 전에 설정이 제대로 되었는지 검증 로직을 추가하면 런타임 에러를 사전에 방지할 수 있습니다. 💡 async/await를 사용하여 비동기 처리를 구현했으므로, 여러 에이전트가 동시에 작업을 처리할 수 있어 성능이 향상됩니다. --- ## 2. 메인 에이전트 구현 ### 시작하며 여러분의 AI 비서에게 "오늘 날씨 알려주고, 3시에 회의 일정 잡아줘"라고 말했을 때, 이 복잡한 요청을 어떻게 처리해야 할까요? 두 가지 작업이 섞여 있고, 각각 다른 에이전트가 처리해야 하는 상황입니다. 이런 문제는 실제 AI 비서 개발에서 가장 먼저 마주치는 도전입니다. 사용자는 복잡한 요청을 자연스러운 언어로 던지는데, 시스템은 이를 분석하고 적절한 에이전트에게 분배해야 합니다. 또한 각 에이전트의 결과를 모아서 사용자에게 일관된 응답을 제공해야 하죠. 바로 이럴 때 필요한 것이 메인 에이전트입니다. 메인 에이전트는 오케스트라의 지휘자처럼 사용자 요청을 분석하고, 적절한 전문 에이전트들을 호출하며, 결과를 통합하여 최종 응답을 만들어냅니다. ### 개요 간단히 말해서, 메인 에이전트는 사용자와 직접 상호작용하며 전체 시스템을 조율하는 핵심 에이전트입니다. 왜 이것이 필요할까요? 실무에서 사용자의 요청은 단순하지 않습니다. "내일 서울 날씨 확인하고, 비 온다면 우산 사는 걸 할 일 목록에 추가해줘" 같은 조건부 로직이 포함된 복잡한 요청도 처리해야 합니다. 메인 에이전트는 이런 복잡한 요청을 이해하고, 여러 전문 에이전트를 순차적 또는 병렬로 호출하며, 결과를 종합하는 역할을 합니다. 기존에는 모든 로직을 한 곳에 때려 넣어 복잡도가 기하급수적으로 증가했다면, 이제는 메인 에이전트가 고수준의 조율만 담당하고 실제 작업은 전문 에이전트에게 위임할 수 있습니다. 핵심 특징으로는 첫째, 사용자 의도 분석 능력으로 요청에서 작업 유형을 식별합니다. 둘째, 워크플로우 관리 능력으로 여러 에이전트를 순서대로 또는 동시에 실행합니다. 셋째, 컨텍스트 유지 능력으로 대화의 흐름을 기억하고 참조합니다. 이러한 특징들이 중요한 이유는, 사용자가 자연스럽고 복잡한 대화를 할 수 있도록 하기 때문입니다. ### 코드 예제 ```python # ai_assistant/agents/main_agent.py from typing import Dict, Any, List from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import HumanMessage, SystemMessage from .base import BaseAgent, AgentConfig class MainAgent(BaseAgent): """사용자 요청을 받아 적절한 에이전트에게 분배하는 메인 에이전트""" def __init__(self, config: AgentConfig, orchestrator): super().__init__(config) self.llm = ChatOpenAI(model=config.model, temperature=config.temperature) self.orchestrator = orchestrator self.conversation_history: List[Dict] = [] async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: """사용자 요청을 처리하는 메인 로직""" user_input = input_data.get("message", "") # 1단계: 사용자 의도 분석 intent = await self._analyze_intent(user_input) # 2단계: 적절한 에이전트에게 작업 위임 results = [] for task in intent["tasks"]: result = await self.orchestrator.delegate_task( task["type"], {"query": task["query"], "context": self.conversation_history} ) results.append(result) # 3단계: 결과 통합 및 응답 생성 final_response = await self._synthesize_response(user_input, results) # 4단계: 대화 히스토리에 추가 self._update_history(user_input, final_response) return {"response": final_response, "intent": intent} async def _analyze_intent(self, user_input: str) -> Dict[str, Any]: """LLM을 사용하여 사용자 의도 분석""" prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="당신은 사용자 요청을 분석하여 필요한 작업을 식별하는 AI입니다."), HumanMessage(content=f"다음 요청을 분석하여 필요한 작업 목록을 JSON 형태로 반환하세요: {user_input}") ]) response = await self.llm.apredict_messages(prompt.format_messages()) # 실제로는 JSON 파싱 로직이 필요 return {"tasks": [{"type": "general", "query": user_input}]} async def _synthesize_response(self, original_query: str, results: List[Dict]) -> str: """여러 에이전트의 결과를 자연스러운 응답으로 통합""" combined = "\n".join([r.get("answer", "") for r in results]) return f"요청하신 '{original_query}'에 대한 답변입니다:\n{combined}" def _update_history(self, user_input: str, response: str): """대화 히스토리 업데이트""" self.conversation_history.append({"user": user_input, "assistant": response}) # 최근 10개 대화만 유지 if len(self.conversation_history) > 10: self.conversation_history.pop(0) ``` ### 설명 이것이 하는 일: 메인 에이전트는 사용자와의 모든 상호작용을 관리하며, 복잡한 요청을 이해하고 적절한 전문 에이전트들을 조율하여 통합된 응답을 제공합니다. 마치 레스토랑의 웨이터가 고객 주문을 받아 주방의 각 파트(전채, 메인, 디저트)에 전달하고 최종적으로 코스 요리를 제공하는 것과 같습니다. 첫 번째로, _analyze_intent 메서드가 사용자의 자연어 입력을 분석합니다. LLM을 활용하여 "오늘 날씨 알려주고 일정도 확인해줘"라는 요청에서 "날씨 조회"와 "일정 확인"이라는 두 가지 작업을 식별해냅니다. 이 과정에서 ChatPromptTemplate을 사용하여 LLM에게 명확한 지시를 내리고, JSON 형태로 구조화된 결과를 받아옵니다. 그 다음으로, process 메서드의 작업 위임 로직이 실행됩니다. 식별된 각 작업을 orchestrator를 통해 적절한 전문 에이전트에게 전달합니다. 이때 conversation_history를 함께 전달하여 이전 대화 맥락을 유지합니다. 예를 들어 "그거 언제였지?"라는 후속 질문에도 대답할 수 있는 것은 이 컨텍스트 덕분입니다. 세 번째로, _synthesize_response 메서드가 여러 에이전트로부터 받은 결과들을 하나의 자연스러운 응답으로 통합합니다. 날씨 에이전트가 "오늘은 맑음"이라고 하고 일정 에이전트가 "3시에 회의 있음"이라고 하면, "오늘은 맑은 날씨이고, 3시에 회의가 예정되어 있습니다"처럼 자연스럽게 합쳐줍니다. 마지막으로, _update_history가 대화 내용을 저장하여 다음 대화에서 참조할 수 있게 합니다. 메모리 관리를 위해 최근 10개만 유지하는 전략을 사용하여 무한정 커지는 것을 방지합니다. 여러분이 이 메인 에이전트를 사용하면 복잡한 멀티턴 대화를 자연스럽게 처리할 수 있습니다. 사용자는 마치 사람과 대화하는 것처럼 편하게 요청할 수 있고, 시스템은 백그라운드에서 여러 에이전트를 조율하여 정확한 답변을 제공합니다. 또한 대화 히스토리를 통해 컨텍스트를 유지하므로 "그거", "그때" 같은 대명사도 이해할 수 있습니다. ### 실전 팁 💡 _analyze_intent에서 LLM의 응답을 파싱할 때는 반드시 예외 처리를 추가하세요. LLM이 항상 완벽한 JSON을 반환하는 것은 아니므로, 파싱 실패 시 기본 작업으로 폴백하는 로직이 필요합니다. 💡 conversation_history의 크기를 제한하는 것은 매우 중요합니다. 토큰 제한을 초과하면 API 호출이 실패할 수 있으므로, 개수뿐만 아니라 총 토큰 수도 체크하는 것이 좋습니다. 💡 여러 에이전트를 동시에 호출할 수 있다면 asyncio.gather를 사용하여 병렬 처리하세요. 날씨 조회와 일정 확인은 독립적이므로 동시에 실행하면 응답 시간이 크게 단축됩니다. 💡 에이전트 호출 실패 시 사용자에게 부분적인 결과라도 제공하는 것이 좋습니다. "날씨 정보는 가져왔지만 일정 조회는 실패했습니다"처럼 명확히 알려주세요. 💡 디버깅을 위해 각 단계(의도 분석, 작업 위임, 응답 통합)의 중간 결과를 로깅하세요. 어디서 문제가 생겼는지 추적하기 쉬워집니다. --- ## 3. 전문 에이전트들 구성 ### 시작하며 여러분이 AI 비서를 만들면서 "하나의 에이전트가 모든 걸 다 하면 되지 않을까?"라고 생각해본 적 있나요? 처음엔 간단해 보이지만, 날씨 조회, 이메일 작성, 코드 생성까지 모든 기능을 하나에 때려 넣다 보면 프롬프트가 엄청나게 길어지고 성능도 떨어집니다. 이런 문제는 실제 AI 시스템 개발에서 "만능 에이전트의 함정"이라고 불립니다. 하나의 에이전트가 모든 것을 하려다 보면 각 작업의 품질이 떨어지고, 한 부분을 수정하면 다른 부분이 영향을 받는 취약한 구조가 됩니다. 또한 특정 작업에 최적화하기도 어렵습니다. 바로 이럴 때 필요한 것이 역할별로 전문화된 에이전트들입니다. 각 에이전트는 자신의 전문 분야에서 최고의 성능을 발휘하도록 설계되고, 필요할 때 메인 에이전트가 적절한 전문가를 호출하는 방식입니다. ### 개요 간단히 말해서, 전문 에이전트는 특정 작업에 특화되어 최적화된 도구와 프롬프트를 가진 독립적인 에이전트입니다. 왜 이것이 필요할까요? 실무에서 각 작업은 서로 다른 전문성을 요구합니다. 날씨 조회는 API 호출과 데이터 파싱에 능해야 하고, 이메일 작성은 자연어 생성과 포맷팅에 강해야 하며, 데이터 분석은 계산과 시각화에 뛰어나야 합니다. 하나의 에이전트로는 모든 분야에서 최고가 될 수 없습니다. 예를 들어, 의료 진단 챗봇은 의료 데이터베이스 접근 도구와 의학 용어 특화 프롬프트가 필요하지만, 법률 자문 챗봇은 법률 DB와 법률 용어가 필요합니다. 기존에는 거대한 if-else 문으로 작업을 분기했다면, 이제는 각 작업을 담당하는 독립적인 에이전트를 만들어 깔끔하게 분리할 수 있습니다. 핵심 특징으로는 첫째, 단일 책임 원칙으로 각 에이전트는 하나의 도메인만 담당합니다. 둘째, 전문화된 도구 세트로 해당 작업에 필요한 도구만 장착합니다. 셋째, 최적화된 프롬프트로 특정 작업의 품질을 극대화합니다. 이러한 특징들이 중요한 이유는, 시스템의 확장성과 유지보수성을 크게 향상시키기 때문입니다. ### 코드 예제 ```python # ai_assistant/agents/specialized_agents.py from typing import Dict, Any from langchain.tools import Tool from langchain.agents import AgentExecutor, create_openai_functions_agent from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from .base import BaseAgent, AgentConfig class WeatherAgent(BaseAgent): """날씨 정보 전문 에이전트""" def __init__(self, config: AgentConfig): super().__init__(config) # 날씨 조회에 특화된 도구 추가 self.add_tool(Tool( name="get_weather", func=self._get_weather_data, description="주어진 도시의 현재 날씨를 조회합니다" )) # 날씨 전문 프롬프트 self.prompt = ChatPromptTemplate.from_messages([ ("system", "당신은 날씨 정보 전문가입니다. 날씨 데이터를 조회하고 사용자 친화적으로 설명합니다."), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) self.llm = ChatOpenAI(model=config.model, temperature=0.3) # 낮은 temperature로 정확성 우선 self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt) self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True) def _get_weather_data(self, city: str) -> str: """실제로는 날씨 API를 호출""" # 여기서는 예시로 더미 데이터 반환 return f"{city}의 날씨: 맑음, 기온 22도" async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: query = input_data.get("query", "") result = await self.executor.ainvoke({"input": query}) return {"answer": result["output"], "agent_type": "weather"} class EmailAgent(BaseAgent): """이메일 작성 전문 에이전트""" def __init__(self, config: AgentConfig): super().__init__(config) # 이메일 관련 도구 추가 self.add_tool(Tool( name="send_email", func=self._send_email, description="이메일을 작성하고 전송합니다" )) # 이메일 작성에 특화된 프롬프트 self.prompt = ChatPromptTemplate.from_messages([ ("system", "당신은 전문적이고 예의바른 이메일 작성 전문가입니다. 비즈니스 톤을 유지하며 명확하게 작성합니다."), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) self.llm = ChatOpenAI(model=config.model, temperature=0.7) # 창의적인 문장을 위해 적절한 temperature self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt) self.executor = AgentExecutor(agent=self.agent, tools=self.tools) def _send_email(self, to: str, subject: str, body: str) -> str: """실제로는 이메일 API를 호출""" return f"이메일 전송 완료: {to}에게 '{subject}' 제목으로 전송됨" async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: query = input_data.get("query", "") result = await self.executor.ainvoke({"input": query}) return {"answer": result["output"], "agent_type": "email"} class DataAnalysisAgent(BaseAgent): """데이터 분석 전문 에이전트""" def __init__(self, config: AgentConfig): super().__init__(config) # 데이터 분석 도구 추가 self.add_tool(Tool( name="analyze_data", func=self._analyze_data, description="데이터를 분석하고 통계를 생성합니다" )) self.prompt = ChatPromptTemplate.from_messages([ ("system", "당신은 데이터 분석 전문가입니다. 숫자와 통계를 정확히 계산하고 인사이트를 제공합니다."), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) self.llm = ChatOpenAI(model=config.model, temperature=0.1) # 매우 낮은 temperature로 정확한 계산 self.agent = create_openai_functions_agent(self.llm, self.tools, self.prompt) self.executor = AgentExecutor(agent=self.agent, tools=self.tools) def _analyze_data(self, data: str) -> str: """실제로는 pandas 등을 사용하여 분석""" return f"데이터 분석 결과: 평균 45.2, 중앙값 42, 표준편차 8.3" async def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: query = input_data.get("query", "") result = await self.executor.ainvoke({"input": query}) return {"answer": result["output"], "agent_type": "data_analysis"} ``` ### 설명 이것이 하는 일: 전문 에이전트 시스템은 각 작업 유형별로 독립적이고 최적화된 에이전트를 제공하여, 전체 시스템의 성능과 유지보수성을 극대화합니다. 마치 병원에 내과, 외과, 정형외과 전문의가 있는 것처럼, 각 에이전트는 자신의 전문 분야에서 최고의 성능을 발휘합니다. 첫 번째로, WeatherAgent는 날씨 조회에 특화되어 있습니다. get_weather_data 도구를 통해 외부 날씨 API를 호출하고, 낮은 temperature(0.3)를 사용하여 정확한 정보 전달을 우선시합니다. 시스템 프롬프트도 "날씨 정보 전문가"로 설정하여 날씨 관련 질문에 최적화된 응답을 생성합니다. 실무에서는 OpenWeatherMap이나 WeatherAPI 같은 실제 서비스와 연동됩니다. 그 다음으로, EmailAgent는 이메일 작성과 전송을 담당합니다. 중요한 점은 temperature를 0.7로 설정하여 창의적이면서도 전문적인 문장을 생성한다는 것입니다. 시스템 프롬프트에 "비즈니스 톤"을 명시하여 일관된 품질의 이메일을 작성합니다. send_email 도구는 실제로 Gmail API나 SendGrid 같은 서비스와 연동될 수 있습니다. 세 번째로, DataAnalysisAgent는 매우 낮은 temperature(0.1)를 사용하여 정확한 계산을 보장합니다. 데이터 분석에서는 창의성보다 정확성이 중요하기 때문입니다. analyze_data 도구는 실제로 pandas, numpy 같은 라이브러리를 사용하여 복잡한 통계 계산을 수행할 수 있습니다. 각 에이전트는 create_openai_functions_agent와 AgentExecutor를 사용하여 LangChain의 강력한 에이전트 기능을 활용합니다. MessagesPlaceholder를 통해 에이전트가 중간 추론 과정(scratchpad)을 기록하고 참조할 수 있어, 복잡한 다단계 작업도 처리할 수 있습니다. 여러분이 이 전문 에이전트 시스템을 사용하면 각 작업의 품질이 크게 향상됩니다. 날씨 조회는 더 정확해지고, 이메일은 더 전문적으로 작성되며, 데이터 분석은 더 신뢰할 수 있게 됩니다. 또한 새로운 기능을 추가할 때도 새로운 전문 에이전트만 만들면 되므로 확장이 쉽습니다. 예를 들어 번역 기능이 필요하면 TranslationAgent만 추가하면 됩니다. ### 실전 팁 💡 각 에이전트의 temperature 설정은 매우 중요합니다. 정확성이 필요한 작업(날씨, 데이터 분석)은 낮게(0.1-0.3), 창의성이 필요한 작업(이메일 작성, 콘텐츠 생성)은 높게(0.7-0.9) 설정하세요. 💡 에이전트별로 다른 LLM 모델을 사용할 수도 있습니다. 간단한 작업은 gpt-3.5-turbo로, 복잡한 작업은 gpt-4로 처리하면 비용을 절감하면서도 품질을 유지할 수 있습니다. 💡 각 에이전트의 도구는 실제 외부 API와 연동할 때 반드시 에러 핸들링을 추가하세요. API 호출 실패, 타임아웃, 잘못된 응답 등을 처리하는 로직이 필수입니다. 💡 verbose=True 옵션을 개발 중에 사용하면 에이전트의 추론 과정을 볼 수 있어 디버깅에 매우 유용합니다. 프로덕션에서는 False로 설정하여 불필요한 로그를 줄이세요. 💡 에이전트의 시스템 프롬프트는 매우 구체적으로 작성하세요. "당신은 전문가입니다" 보다 "당신은 10년 경력의 기상 전문가로, 일반인도 이해하기 쉽게 설명합니다"처럼 구체적일수록 좋습니다. --- ## 4. 도구 통합 시스템 ### 시작하며 여러분의 AI 비서가 "내일 회의 자료를 PDF로 만들어줘"라는 요청을 받았을 때, 어떻게 실제로 PDF를 생성할까요? AI는 텍스트 생성은 잘하지만, 파일 시스템 접근, API 호출, 데이터베이스 쿼리 같은 실제 작업은 직접 할 수 없습니다. 이런 문제는 AI 에이전트의 근본적인 한계입니다. LLM은 언어를 이해하고 생성하는 데는 뛰어나지만, 외부 세계와 상호작용하려면 "도구"가 필요합니다. 웹 검색, 파일 읽기/쓰기, API 호출, 계산기 등 실제 작업을 수행하는 코드가 필요한 거죠. 바로 이럴 때 필요한 것이 도구 통합 시스템입니다. 에이전트가 필요한 도구를 쉽게 사용할 수 있도록 표준화된 인터페이스를 제공하고, 새로운 도구를 추가하거나 기존 도구를 업데이트하기 쉬운 구조를 만듭니다. ### 개요 간단히 말해서, 도구 통합 시스템은 AI 에이전트가 외부 세계와 상호작용할 수 있도록 다양한 기능을 제공하는 플러그인 같은 시스템입니다. 왜 이것이 필요할까요? 실무에서 AI 비서는 단순히 대화만 하는 것이 아닙니다. 파일을 읽고 쓰고, 웹에서 정보를 검색하고, 데이터베이스를 조회하고, 외부 API를 호출해야 합니다. 예를 들어, "최신 기술 뉴스를 요약해줘"라는 요청은 웹 스크래핑 도구, 텍스트 요약 도구, 그리고 결과를 저장하는 파일 시스템 도구가 필요합니다. 각 도구를 에이전트마다 다르게 구현하면 코드 중복과 유지보수 문제가 발생합니다. 기존에는 각 에이전트가 자체적으로 외부 기능을 구현했다면, 이제는 중앙화된 도구 레지스트리에서 필요한 도구를 가져다 쓸 수 있습니다. 핵심 특징으로는 첫째, 표준화된 인터페이스로 모든 도구가 동일한 방식으로 호출됩니다. 둘째, 플러그인 아키텍처로 새로운 도구를 쉽게 추가할 수 있습니다. 셋째, 에러 핸들링과 재시도 로직이 내장되어 안정성이 높습니다. 이러한 특징들이 중요한 이유는, 에이전트의 능력을 쉽게 확장하고 시스템을 안정적으로 유지하기 때문입니다. ### 코드 예제 ```python # ai_assistant/tools/tool_registry.py from typing import Dict, Callable, Any, List from langchain.tools import Tool import requests from pathlib import Path import json class ToolRegistry: """모든 도구를 등록하고 관리하는 중앙 레지스트리""" def __init__(self): self.tools: Dict[str, Tool] = {} self._register_default_tools() def _register_default_tools(self): """기본 도구들을 등록""" # 웹 검색 도구 self.register_tool(Tool( name="web_search", func=self._web_search, description="인터넷에서 정보를 검색합니다. 검색어를 입력하면 관련 결과를 반환합니다." )) # 파일 읽기 도구 self.register_tool(Tool( name="read_file", func=self._read_file, description="파일 내용을 읽습니다. 파일 경로를 입력하면 내용을 반환합니다." )) # 파일 쓰기 도구 self.register_tool(Tool( name="write_file", func=self._write_file, description="파일에 내용을 씁니다. '경로|내용' 형식으로 입력합니다." )) # 계산기 도구 self.register_tool(Tool( name="calculator", func=self._calculate, description="수학 계산을 수행합니다. Python 표현식을 입력하면 결과를 반환합니다." )) # API 호출 도구 self.register_tool(Tool( name="api_call", func=self._api_call, description="외부 API를 호출합니다. 'GET|URL' 또는 'POST|URL|JSON_DATA' 형식으로 입력합니다." )) def register_tool(self, tool: Tool): """새로운 도구를 레지스트리에 등록""" self.tools[tool.name] = tool def get_tool(self, name: str) -> Tool: """이름으로 도구 가져오기""" if name not in self.tools: raise ValueError(f"도구 '{name}'을 찾을 수 없습니다") return self.tools[name] def get_all_tools(self) -> List[Tool]: """모든 도구 가져오기""" return list(self.tools.values()) def _web_search(self, query: str) -> str: """웹 검색 수행 - 실제로는 Google API 등을 사용""" try: # 여기서는 예시로 더미 데이터 반환 return f"'{query}'에 대한 검색 결과: [더미 데이터] 관련 정보가 여기 표시됩니다." except Exception as e: return f"웹 검색 중 오류 발생: {str(e)}" def _read_file(self, file_path: str) -> str: """파일 내용 읽기""" try: path = Path(file_path) if not path.exists(): return f"오류: 파일 '{file_path}'을 찾을 수 없습니다" return path.read_text(encoding='utf-8') except Exception as e: return f"파일 읽기 중 오류 발생: {str(e)}" def _write_file(self, input_str: str) -> str: """파일에 내용 쓰기""" try: file_path, content = input_str.split('|', 1) path = Path(file_path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding='utf-8') return f"파일 '{file_path}'에 성공적으로 작성했습니다" except Exception as e: return f"파일 쓰기 중 오류 발생: {str(e)}" def _calculate(self, expression: str) -> str: """수학 계산 수행""" try: # 보안을 위해 eval 대신 ast.literal_eval 사용 권장 result = eval(expression, {"__builtins__": {}}, {}) return f"계산 결과: {result}" except Exception as e: return f"계산 중 오류 발생: {str(e)}" def _api_call(self, input_str: str) -> str: """외부 API 호출""" try: parts = input_str.split('|') method = parts[0].upper() url = parts[1] if method == "GET": response = requests.get(url, timeout=10) elif method == "POST": data = json.loads(parts[2]) if len(parts) > 2 else {} response = requests.post(url, json=data, timeout=10) else: return f"지원하지 않는 HTTP 메서드: {method}" return f"API 응답 (상태 코드 {response.status_code}): {response.text[:500]}" except Exception as e: return f"API 호출 중 오류 발생: {str(e)}" # 사용 예시 tool_registry = ToolRegistry() # 커스텀 도구 추가 custom_tool = Tool( name="translate", func=lambda text: f"번역된 텍스트: {text}", # 실제로는 번역 API 호출 description="텍스트를 번역합니다" ) tool_registry.register_tool(custom_tool) ``` ### 설명 이것이 하는 일: 도구 통합 시스템은 AI 에이전트가 실제 세계와 상호작용하는 데 필요한 모든 기능을 표준화된 방식으로 제공합니다. 마치 스마트폰의 앱 스토어처럼, 필요한 도구를 찾아 설치하고 사용할 수 있는 중앙화된 플랫폼입니다. 첫 번째로, ToolRegistry 클래스가 모든 도구를 관리하는 중앙 허브 역할을 합니다. _register_default_tools 메서드에서 기본적으로 제공되는 도구들(웹 검색, 파일 읽기/쓰기, 계산기, API 호출)을 등록합니다. 이렇게 하면 모든 에이전트가 이 기본 도구들을 즉시 사용할 수 있습니다. 새 프로젝트를 시작할 때마다 이런 기본 기능들을 다시 구현할 필요가 없어집니다. 그 다음으로, 각 도구는 LangChain의 Tool 클래스로 래핑되어 표준화된 인터페이스를 제공합니다. 모든 도구는 name(도구 이름), func(실제 기능을 수행하는 함수), description(AI가 언제 이 도구를 사용해야 하는지 이해하도록 돕는 설명)을 가집니다. 이 표준화된 구조 덕분에 AI 에이전트는 어떤 도구든 동일한 방식으로 호출할 수 있습니다. 세 번째로, 각 도구 함수(_web_search, _read_file 등)는 철저한 에러 핸들링을 포함합니다. try-except 블록으로 감싸서 어떤 오류가 발생하더라도 시스템이 크래시하지 않고 의미 있는 에러 메시지를 반환합니다. 예를 들어 파일이 없으면 "파일을 찾을 수 없습니다"라고 명확히 알려줍니다. 네 번째로, register_tool 메서드를 통해 커스텀 도구를 쉽게 추가할 수 있습니다. 코드 예시의 마지막 부분처럼 번역 도구를 추가하는 것이 단 몇 줄이면 됩니다. 이 확장성 덕분에 프로젝트가 성장하면서 필요한 도구들을 계속 추가할 수 있습니다. 다섯 번째로, _api_call 같은 범용 도구는 여러 시나리오에 활용될 수 있습니다. GET/POST 요청을 모두 지원하고, JSON 데이터도 전달할 수 있어서 대부분의 REST API와 통신할 수 있습니다. 타임아웃 설정(timeout=10)으로 무한 대기를 방지합니다. 여러분이 이 도구 통합 시스템을 사용하면 에이전트의 능력을 폭발적으로 확장할 수 있습니다. 웹에서 최신 정보를 가져오고, 파일을 읽고 쓰며, 외부 서비스와 통신하는 모든 것이 가능해집니다. 또한 도구를 재사용할 수 있어 개발 시간이 크게 단축되고, 중앙에서 관리하므로 버그 수정이나 기능 개선도 한 곳에서 하면 모든 에이전트에 적용됩니다. ### 실전 팁 💡 보안이 중요한 도구(파일 쓰기, API 호출)는 반드시 입력 검증을 추가하세요. 예를 들어 파일 쓰기는 허용된 디렉토리 내에서만 가능하도록 경로를 체크해야 합니다. 💡 계산기 도구에서 eval()을 사용할 때는 매우 주의해야 합니다. 예시처럼 __builtins__를 빈 딕셔너리로 설정하거나, ast.literal_eval이나 sympy 같은 안전한 대안을 사용하세요. 💡 API 호출 도구는 rate limiting을 고려해야 합니다. 같은 API를 너무 자주 호출하면 차단될 수 있으므로, 캐싱이나 쓰로틀링을 추가하는 것이 좋습니다. 💡 각 도구의 description은 매우 중요합니다. AI가 이 설명을 읽고 언제 도구를 사용할지 결정하므로, 명확하고 구체적으로 작성하세요. "파일을 읽습니다" 보다 "로컬 파일 시스템에서 텍스트 파일의 내용을 읽어옵니다. 절대 경로 또는 상대 경로를 지원합니다"가 더 좋습니다. 💡 도구 함수의 반환값은 항상 문자열로 통일하세요. LangChain의 Tool은 문자열 반환을 기대하므로, 딕셔너리나 리스트를 반환하려면 json.dumps()로 직렬화해야 합니다. --- ## 5. 대화 메모리 관리 ### 시작하며 여러분이 AI 비서에게 "어제 말한 그 레스토랑 이름이 뭐였지?"라고 물었는데 "무슨 레스토랑을 말씀하시는 건가요?"라는 답변을 받으면 어떤 기분이 드나요? 마치 금붕어와 대화하는 것 같아서 답답하죠. 이런 문제는 AI 에이전트가 대화 맥락을 기억하지 못해서 발생합니다. 각 요청을 독립적으로 처리하면 이전 대화 내용을 전혀 모르게 됩니다. "그거", "그때", "아까 말한" 같은 대명사나 참조를 이해할 수 없고, 사용자는 매번 모든 정보를 반복해서 말해야 하는 불편함을 겪습니다. 바로 이럴 때 필요한 것이 대화 메모리 관리 시스템입니다. 이전 대화를 기억하고, 적절한 시점에 관련 정보를 불러오며, 메모리가 너무 커지지 않도록 관리하는 시스템이죠. ### 개요 간단히 말해서, 대화 메모리 관리는 AI 에이전트가 과거 대화를 기억하고 현재 대화에 활용할 수 있도록 하는 시스템입니다. 왜 이것이 필요할까요? 실무에서 사용자와의 자연스러운 대화는 여러 턴에 걸쳐 이어집니다. "파이썬으로 웹 크롤러 만들려고 해" → "어떤 라이브러리 쓰면 돼?" → "그거 설치하는 방법 알려줘" 같은 연속된 대화에서, AI는 "그거"가 앞에서 추천한 라이브러리를 의미한다는 걸 알아야 합니다. 또한 사용자의 선호도나 이전 작업 내용을 기억하면 더 개인화된 서비스를 제공할 수 있습니다. 기존에는 전체 대화를 매번 프롬프트에 포함시켜 토큰 비용이 폭증했다면, 이제는 스마트한 메모리 관리로 필요한 부분만 효율적으로 유지할 수 있습니다. 핵심 특징으로는 첫째, 단기 메모리로 최근 몇 턴의 대화를 빠르게 접근합니다. 둘째, 장기 메모리로 중요한 정보를 영구 저장합니다. 셋째, 메모리 요약 기능으로 오래된 대화를 압축하여 토큰을 절약합니다. 이러한 특징들이 중요한 이유는, 사용자 경험을 크게 향상시키면서도 비용을 관리할 수 있기 때문입니다. ### 코드 예제 ```python # ai_assistant/memory/conversation_memory.py from typing import List, Dict, Any, Optional from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, AIMessage import json from datetime import datetime class HybridMemoryManager: """단기 메모리와 장기 메모리를 결합한 하이브리드 메모리 관리자""" def __init__(self, llm: ChatOpenAI, max_short_term: int = 10): # 단기 메모리: 최근 N개 대화 저장 self.short_term_memory = ConversationBufferMemory( return_messages=True, memory_key="chat_history", max_token_limit=2000 ) # 장기 메모리: 중요한 정보를 요약하여 저장 self.long_term_memory = ConversationSummaryMemory( llm=llm, memory_key="summary", return_messages=True ) self.max_short_term = max_short_term self.conversation_count = 0 # 영구 저장을 위한 전체 히스토리 self.full_history: List[Dict[str, Any]] = [] # 사용자 프로필 (학습된 선호도 등) self.user_profile: Dict[str, Any] = { "preferences": {}, "important_facts": [] } def add_interaction(self, user_message: str, ai_response: str, metadata: Optional[Dict] = None): """새로운 대화 턴 추가""" # 단기 메모리에 추가 self.short_term_memory.save_context( {"input": user_message}, {"output": ai_response} ) # 전체 히스토리에 추가 interaction = { "timestamp": datetime.now().isoformat(), "user": user_message, "assistant": ai_response, "metadata": metadata or {} } self.full_history.append(interaction) self.conversation_count += 1 # 일정 턴마다 오래된 대화를 장기 메모리로 이동 if self.conversation_count % self.max_short_term == 0: self._consolidate_memory() def _consolidate_memory(self): """단기 메모리를 요약하여 장기 메모리로 이동""" # 단기 메모리의 내용을 가져옴 messages = self.short_term_memory.chat_memory.messages if len(messages) > 0: # 장기 메모리에 요약 추가 for i in range(0, len(messages), 2): if i+1 < len(messages): self.long_term_memory.save_context( {"input": messages[i].content}, {"output": messages[i+1].content} ) # 단기 메모리 초기화 (최근 2-3턴만 유지) recent_messages = messages[-4:] # 최근 2턴 (유저 2 + AI 2) self.short_term_memory.chat_memory.messages = recent_messages def get_context(self, include_summary: bool = True) -> str: """현재 컨텍스트 가져오기 (프롬프트에 포함할 내용)""" context_parts = [] # 장기 메모리 요약 포함 if include_summary: summary = self.long_term_memory.load_memory_variables({}) if summary.get("summary"): context_parts.append(f"이전 대화 요약:\n{summary['summary']}\n") # 단기 메모리 (최근 대화) 포함 short_term = self.short_term_memory.load_memory_variables({}) if short_term.get("chat_history"): context_parts.append("최근 대화:") for msg in short_term["chat_history"]: role = "사용자" if isinstance(msg, HumanMessage) else "AI" context_parts.append(f"{role}: {msg.content}") # 사용자 프로필 정보 포함 if self.user_profile["preferences"] or self.user_profile["important_facts"]: context_parts.append("\n사용자 정보:") if self.user_profile["preferences"]: context_parts.append(f"선호도: {json.dumps(self.user_profile['preferences'], ensure_ascii=False)}") if self.user_profile["important_facts"]: context_parts.append(f"중요 사실: {', '.join(self.user_profile['important_facts'])}") return "\n".join(context_parts) def update_user_profile(self, key: str, value: Any): """사용자 프로필 업데이트 (선호도 학습)""" self.user_profile["preferences"][key] = value def add_important_fact(self, fact: str): """장기적으로 기억해야 할 중요한 사실 추가""" if fact not in self.user_profile["important_facts"]: self.user_profile["important_facts"].append(fact) def search_history(self, keyword: str) -> List[Dict[str, Any]]: """과거 대화에서 키워드 검색""" results = [] for interaction in self.full_history: if keyword.lower() in interaction["user"].lower() or keyword.lower() in interaction["assistant"].lower(): results.append(interaction) return results def save_to_file(self, filepath: str): """메모리를 파일로 저장""" data = { "full_history": self.full_history, "user_profile": self.user_profile, "timestamp": datetime.now().isoformat() } with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def load_from_file(self, filepath: str): """파일에서 메모리 로드""" with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) self.full_history = data.get("full_history", []) self.user_profile = data.get("user_profile", {"preferences": {}, "important_facts": []}) ``` ### 설명 이것이 하는 일: 하이브리드 메모리 관리자는 사람의 기억 시스템을 모방하여, 최근 대화는 상세히 기억하고(단기 메모리), 오래된 대화는 요약하여 기억하며(장기 메모리), 중요한 정보는 영구적으로 저장합니다(사용자 프로필). 첫 번째로, add_interaction 메서드가 새로운 대화를 기록합니다. 사용자 메시지와 AI 응답을 단기 메모리에 추가하고, 타임스탬프와 메타데이터를 포함하여 전체 히스토리에도 저장합니다. 이렇게 이중으로 저장하는 이유는, 단기 메모리는 빠른 접근을 위해, 전체 히스토리는 나중에 분석이나 검색을 위해 필요하기 때문입니다. 그 다음으로, _consolidate_memory 메서드가 주기적으로 실행되어 메모리를 정리합니다. 10턴마다 오래된 대화를 요약하여 장기 메모리로 옮기고, 단기 메모리는 최근 2-3턴만 유지합니다. 이 과정이 없다면 대화가 길어질수록 프롬프트 크기가 계속 커져서 토큰 제한을 초과하거나 비용이 폭증할 수 있습니다. LangChain의 ConversationSummaryMemory가 자동으로 대화를 요약해주므로 메모리 효율이 높습니다. 세 번째로, get_context 메서드가 현재 프롬프트에 포함할 컨텍스트를 생성합니다. 장기 메모리 요약 + 최근 대화 + 사용자 프로필 정보를 모두 결합하여 AI에게 전달합니다. 이렇게 하면 AI는 "오늘 날씨 알려줘"라는 요청을 받았을 때, 사용자가 이전에 "나는 서울에 살아"라고 말한 것을 기억하고 서울 날씨를 알려줄 수 있습니다. 네 번째로, user_profile을 통해 사용자에 대해 학습한 정보를 저장합니다. 선호도(예: "Python을 선호함", "비주얼 학습자")나 중요한 사실(예: "프로젝트 마감일은 12월 31일")을 기록하여, 매번 물어보지 않아도 개인화된 서비스를 제공할 수 있습니다. 다섯 번째로, search_history와 save_to_file 같은 유틸리티 메서드들이 추가 기능을 제공합니다. 과거 대화에서 특정 주제를 검색하거나, 전체 대화 이력을 파일로 저장하여 나중에 불러올 수 있습니다. 이는 사용자가 세션을 종료했다가 다시 시작해도 이전 대화를 이어갈 수 있게 해줍니다. 여러분이 이 메모리 시스템을 사용하면 AI 비서가 훨씬 더 인간적으로 느껴집니다. "그거", "아까" 같은 대명사를 이해하고, 사용자의 선호도를 기억하며, 오래 전에 나눈 대화도 참조할 수 있습니다. 동시에 스마트한 메모리 관리로 토큰 비용을 절약하고 성능도 유지할 수 있습니다. ### 실전 팁 💡 max_token_limit을 설정하여 단기 메모리가 너무 커지는 것을 방지하세요. GPT-4의 경우 8K 토큰 제한을 고려하면, 메모리는 2000-3000 토큰 정도로 제한하는 것이 안전합니다. 💡 중요한 정보는 add_important_fact로 명시적으로 저장하세요. "사용자의 생일은 3월 15일"처럼 잊어버리면 안 되는 정보는 요약 과정에서 사라질 수 있으므로 별도 저장이 필요합니다. 💡 대화 히스토리를 주기적으로 파일로 저장하여 백업하세요. 시스템 재시작이나 오류 발생 시에도 대화를 복구할 수 있습니다. 또한 나중에 사용자 행동 분석이나 모델 개선에도 활용할 수 있습니다. 💡 민감한 정보(비밀번호, 개인정보 등)는 메모리에 저장하지 않도록 필터링 로직을 추가하세요. 정규표현식으로 신용카드 번호나 주민등록번호 패턴을 감지하여 자동으로 마스킹할 수 있습니다. 💡 search_history 기능을 활용하여 "지난번에 추천해준 책 뭐였지?" 같은 질문에 답할 수 있습니다. 키워드 검색뿐만 아니라 임베딩 기반 시맨틱 검색을 추가하면 더욱 강력해집니다. --- ## 6. 에이전트 간 협업 구현 ### 시작하며 여러분이 "내일 비 온다면 우산을 사라고 할 일 목록에 추가해줘"라는 요청을 AI 비서에게 했다고 상상해보세요. 이 요청을 처리하려면 날씨 에이전트가 일기예보를 확인하고, 그 결과를 바탕으로 할 일 관리 에이전트가 작업을 추가해야 합니다. 이런 문제는 단일 에이전트로는 해결할 수 없습니다. 여러 전문 에이전트가 순차적으로 또는 조건부로 협력해야 하는 복잡한 워크플로우가 필요합니다. 날씨 에이전트의 출력이 할 일 에이전트의 입력이 되고, 중간에 조건 분기(비가 온다면)도 처리해야 하죠. 바로 이럴 때 필요한 것이 에이전트 간 협업 시스템입니다. 여러 에이전트를 체계적으로 연결하고, 데이터를 전달하며, 조건부 로직을 처리하는 워크플로우 엔진이 필요합니다. ### 개요 간단히 말해서, 에이전트 간 협업은 여러 전문 에이전트가 서로 통신하고 작업을 주고받으며 복잡한 태스크를 완수하는 시스템입니다. 왜 이것이 필요할까요? 실무에서 사용자의 요청은 여러 단계와 조건을 포함합니다. "이메일에서 회의 일정 추출하고, 캘린더에 추가하고, 참석자들에게 확인 메일 보내기" 같은 요청은 이메일 처리 에이전트 → 일정 관리 에이전트 → 이메일 발송 에이전트가 순차적으로 협력해야 합니다. 각 에이전트는 독립적으로는 훌륭하지만, 함께 작동하지 못하면 복잡한 작업을 처리할 수 없습니다. 기존에는 각 에이전트가 독립적으로 작동하여 결과를 수동으로 다음 에이전트에 전달했다면, 이제는 자동화된 워크플로우로 에이전트들이 매끄럽게 협업할 수 있습니다. 핵심 특징으로는 첫째, 데이터 파이프라인으로 한 에이전트의 출력이 다음 에이전트의 입력으로 자동 전달됩니다. 둘째, 조건부 실행으로 특정 조건에 따라 다른 에이전트를 호출합니다. 셋째, 병렬 처리로 독립적인 작업은 동시에 실행하여 속도를 높입니다. 이러한 특징들이 중요한 이유는, 복잡한 비즈니스 로직을 구현할 수 있기 때문입니다. ### 코드 예제 ```python # ai_assistant/workflow/agent_workflow.py from typing import Dict, Any, List, Callable, Optional from enum import Enum import asyncio class WorkflowNodeType(Enum): """워크플로우 노드 타입""" AGENT = "agent" # 에이전트 실행 CONDITION = "condition" # 조건 분기 PARALLEL = "parallel" # 병렬 실행 MERGE = "merge" # 결과 병합 class WorkflowNode: """워크플로우의 단일 노드""" def __init__(self, node_id: str, node_type: WorkflowNodeType, agent_type: Optional[str] = None, condition_func: Optional[Callable] = None): self.node_id = node_id self.node_type = node_type self.agent_type = agent_type self.condition_func = condition_func self.next_nodes: List[str] = [] # 다음 노드들 self.next_on_true: Optional[str] = None # 조건이 참일 때 self.next_on_false: Optional[str] = None # 조건이 거짓일 때 class AgentWorkflow: """에이전트 간 협업을 조율하는 워크플로우 엔진""" def __init__(self, orchestrator): self.orchestrator = orchestrator self.nodes: Dict[str, WorkflowNode] = {} self.start_node: Optional[str] = None def add_agent_node(self, node_id: str, agent_type: str) -> WorkflowNode: """에이전트 실행 노드 추가""" node = WorkflowNode(node_id, WorkflowNodeType.AGENT, agent_type=agent_type) self.nodes[node_id] = node if self.start_node is None: self.start_node = node_id return node def add_condition_node(self, node_id: str, condition_func: Callable) -> WorkflowNode: """조건 분기 노드 추가""" node = WorkflowNode(node_id, WorkflowNodeType.CONDITION, condition_func=condition_func) self.nodes[node_id] = node return node def connect(self, from_node: str, to_node: str, condition: Optional[str] = None): """노드 연결""" if from_node not in self.nodes: raise ValueError(f"노드 '{from_node}'을 찾을 수 없습니다") node = self.nodes[from_node] if condition == "true": node.next_on_true = to_node elif condition == "false": node.next_on_false = to_node else: node.next_nodes.append(to_node) async def execute(self, initial_data: Dict[str, Any]) -> Dict[str, Any]: """워크플로우 실행""" if not self.start_node: raise ValueError("시작 노드가 설정되지 않았습니다") context = {"initial_data": initial_data, "results": {}} return await self._execute_node(self.start_node, context) async def _execute_node(self, node_id: str, context: Dict[str, Any]) -> Dict[str, Any]: """단일 노드 실행""" if node_id not in self.nodes: return context node = self.nodes[node_id] if node.node_type == WorkflowNodeType.AGENT: # 에이전트 실행 result = await self.orchestrator.delegate_task( node.agent_type, context.get("current_data", context["initial_data"]) ) context["results"][node_id] = result context["current_data"] = result # 다음 노드 실행 if node.next_nodes: for next_node in node.next_nodes: context = await self._execute_node(next_node, context) elif node.node_type == WorkflowNodeType.CONDITION: # 조건 평가 condition_result = node.condition_func(context["current_data"]) context["results"][node_id] = {"condition": condition_result} # 조건에 따라 다음 노드 선택 if condition_result and node.next_on_true: context = await self._execute_node(node.next_on_true, context) elif not condition_result and node.next_on_false: context = await self._execute_node(node.next_on_false, context) return context # 사용 예시: "내일 비 온다면 우산 사기를 할 일에 추가" async def create_weather_todo_workflow(orchestrator): """날씨 기반 할 일 추가 워크플로우""" workflow = AgentWorkflow(orchestrator) # 1. 날씨 조회 노드 weather_node = workflow.add_agent_node("check_weather", "weather") # 2. 조건 확인 노드 (비가 오는지 체크) def is_rainy(data: Dict[str, Any]) -> bool: # 날씨 에이전트 결과에서 비 여부 확인 answer = data.get("answer", "") return "비" in answer or "rain" in answer.lower() condition_node = workflow.add_condition_node("check_rain", is_rainy) # 3. 할 일 추가 노드 todo_node = workflow.add_agent_node("add_todo", "todo") # 노드 연결 workflow.connect("check_weather", "check_rain") workflow.connect("check_rain", "add_todo", condition="true") return workflow # 실행 # workflow = await create_weather_todo_workflow(orchestrator) # result = await workflow.execute({"query": "내일 서울 날씨 확인하고 비 오면 우산 사기 할 일 추가"}) ``` ### 설명 이것이 하는 일: 에이전트 워크플로우 시스템은 여러 전문 에이전트를 체계적으로 연결하여 복잡한 비즈니스 로직을 구현합니다. 마치 공장의 생산 라인처럼, 각 작업(에이전트)이 순서대로 처리되고, 중간에 품질 검사(조건 확인)를 거쳐, 최종 제품(결과)을 만들어냅니다. 첫 번째로, WorkflowNode 클래스가 워크플로우의 각 단계를 표현합니다. 에이전트 실행 노드, 조건 분기 노드 등 다양한 타입이 있으며, 각 노드는 다음에 실행될 노드들을 가리킵니다. 이렇게 그래프 구조로 설계하면 복잡한 분기와 순환도 표현할 수 있습니다. 그 다음으로, AgentWorkflow 클래스가 전체 워크플로우를 관리합니다. add_agent_node로 에이전트 실행 단계를 추가하고, add_condition_node로 조건 분기를 추가하며, connect로 노드들을 연결합니다. 이 API는 매우 직관적이어서 복잡한 워크플로우도 쉽게 정의할 수 있습니다. 세 번째로, execute 메서드가 워크플로우를 실행합니다. 시작 노드부터 차례대로 노드를 실행하며, context 딕셔너리에 모든 중간 결과를 저장합니다. 각 노드의 결과는 다음 노드의 입력이 되므로, 데이터가 자연스럽게 흐릅니다. 네 번째로, _execute_node 메서드가 각 노드 타입에 맞게 처리합니다. AGENT 타입이면 해당 에이전트를 호출하고, CONDITION 타입이면 조건 함수를 실행하여 참/거짓에 따라 다른 경로로 분기합니다. 이 재귀적 구조 덕분에 중첩된 조건이나 복잡한 흐름도 처리할 수 있습니다. 다섯 번째로, create_weather_todo_workflow 예시가 실제 사용법을 보여줍니다. 날씨 조회 → 비 여부 확인 → 조건이 참이면 할 일 추가 라는 3단계 워크플로우를 정의합니다. is_rainy 함수가 날씨 에이전트의 결과에서 "비"라는 키워드를 찾아 조건을 평가합니다. 여러분이 이 워크플로우 시스템을 사용하면 복잡한 멀티 에이전트 시나리오를 선언적으로 정의할 수 있습니다. "이것 한 다음 저것 해"라는 복잡한 로직을 코드로 명확히 표현할 수 있고, 나중에 워크플로우를 수정하거나 확장하기도 쉽습니다. 또한 워크플로우를 JSON이나 YAML로 저장하여 비개발자도 편집할 수 있게 만들 수도 있습니다. ### 실전 팁 💡 복잡한 워크플로우는 시각화 도구로 그려보세요. Graphviz나 Mermaid를 사용하여 노드와 연결을 다이어그램으로 표현하면 로직을 이해하고 디버깅하기 훨씬 쉽습니다. 💡 각 노드의 실행 시간과 결과를 로깅하세요. 어떤 에이전트에서 병목이 발생하는지, 조건 분기가 예상대로 동작하는지 추적할 수 있습니다. 💡 워크플로우 실행 중 에러가 발생하면 어떻게 할지 전략을 세우세요. 전체 워크플로우를 중단할지, 해당 노드만 스킵하고 계속 진행할지, 재시도할지 결정해야 합니다. 💡 병렬 처리가 가능한 노드는 asyncio.gather를 사용하여 동시 실행하세요. 예를 들어 날씨 조회와 뉴스 검색은 독립적이므로 동시에 실행하면 시간을 절약할 수 있습니다. 💡 워크플로우를 재사용 가능한 템플릿으로 만드세요. "날씨 기반 할 일 추가"를 하나의 템플릿으로 저장하고, 다른 도시나 다른 조건으로 쉽게 변형할 수 있게 설계하면 좋습니다. --- ## 7. 에러 핸들링과 재시도 로직 ### 시작하며 여러분의 AI 비서가 날씨 API를 호출했는데 네트워크 오류로 실패했다고 상상해보세요. 사용자에게 "죄송합니다. 오류가 발생했습니다"라고만 말하고 끝나면 어떨까요? 사용자는 답답하고, 서비스 신뢰도도 떨어집니다. 이런 문제는 실제 운영 환경에서 매우 자주 발생합니다. 외부 API가 일시적으로 다운되거나, 네트워크가 불안정하거나, Rate Limit에 걸리거나, 타임아웃이 발생하는 등 수많은 실패 시나리오가 있습니다. 이런 일시적인 오류를 제대로 처리하지 못하면 시스템이 불안정해지고 사용자 경험이 나빠집니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링과 재시도 로직입니다. 일시적인 오류는 자동으로 재시도하고, 영구적인 오류는 명확한 메시지로 사용자에게 알리며, 모든 오류를 로깅하여 나중에 분석할 수 있도록 합니다. ### 개요 간단히 말해서, 에러 핸들링과 재시도 로직은 시스템에서 발생하는 다양한 오류를 지능적으로 처리하여 안정성과 신뢰성을 높이는 메커니즘입니다. 왜 이것이 필요할까요? 실무에서 완벽한 시스템은 없습니다. 외부 서비스는 언제든 실패할 수 있고, 네트워크는 불안정할 수 있습니다. 중요한 것은 이런 실패를 어떻게 처리하느냐입니다. 똑똑한 재시도 로직으로 90%의 일시적 오류는 자동으로 복구할 수 있고, 나머지 10%의 영구적 오류는 사용자에게 명확히 알려 대안을 찾도록 도울 수 있습니다. 기존에는 모든 예외를 단순히 try-except로 잡아서 무시하거나 프로그램을 종료했다면, 이제는 오류 타입별로 다르게 대응하고, 복구 가능한 오류는 자동으로 재시도하며, 모든 오류를 체계적으로 기록할 수 있습니다. 핵심 특징으로는 첫째, 지능적 재시도로 지수 백오프와 최대 재시도 횟수를 사용합니다. 둘째, 오류 분류로 일시적 오류와 영구적 오류를 구분합니다. 셋째, 폴백 메커니즘으로 주 서비스 실패 시 대안 서비스를 사용합니다. 이러한 특징들이 중요한 이유는, 시스템의 가용성과 복원력을 크게 향상시키기 때문입니다. ### 코드 예제 ```python # ai_assistant/utils/error_handling.py from typing import Callable, Any, Optional, Type import asyncio import logging from functools import wraps from datetime import datetime import traceback # 로거 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RetryableError(Exception): """재시도 가능한 오류""" pass class NonRetryableError(Exception): """재시도 불가능한 영구적 오류""" pass class ErrorHandler: """에러 핸들링 및 재시도 로직을 담당하는 클래스""" @staticmethod async def retry_with_backoff( func: Callable, max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 60.0, exponential_base: float = 2.0, retryable_exceptions: tuple = (RetryableError, ConnectionError, TimeoutError) ) -> Any: """ 지수 백오프를 사용한 재시도 로직 Args: func: 실행할 함수 max_retries: 최대 재시도 횟수 initial_delay: 초기 대기 시간 (초) max_delay: 최대 대기 시간 (초) exponential_base: 지수 증가 배수 retryable_exceptions: 재시도할 예외 타입들 """ delay = initial_delay last_exception = None for attempt in range(max_retries + 1): try: # 함수 실행 if asyncio.iscoroutinefunction(func): result = await func() else: result = func() # 성공 시 로깅 if attempt > 0: logger.info(f"재시도 {attempt}번 만에 성공: {func.__name__}") return result except retryable_exceptions as e: last_exception = e if attempt < max_retries: # 재시도할 것임을 로깅 logger.warning( f"재시도 가능한 오류 발생 ({attempt + 1}/{max_retries}): " f"{type(e).__name__}: {str(e)}. " f"{delay}초 후 재시도합니다." ) # 지수 백오프 대기 await asyncio.sleep(delay) # 다음 대기 시간 계산 (지수 증가, 최대값 제한) delay = min(delay * exponential_base, max_delay) else: # 최대 재시도 횟수 초과 logger.error( f"최대 재시도 횟수 {max_retries}회 초과. 마지막 오류: {str(e)}" ) except NonRetryableError as e: # 재시도 불가능한 오류는 즉시 발생 logger.error(f"재시도 불가능한 오류 발생: {type(e).__name__}: {str(e)}") raise except Exception as e: # 예상치 못한 오류 logger.error( f"예상치 못한 오류 발생: {type(e).__name__}: {str(e)}\n" f"{traceback.format_exc()}" ) raise # 모든 재시도 실패 시 마지막 예외 발생 raise last_exception def with_retry(max_retries: int = 3, initial_delay: float = 1.0): """재시도 로직을 적용하는 데코레이터""" def decorator(func: Callable) -> Callable: @wraps(func) async def wrapper(*args, **kwargs): async def target(): return await func(*args, **kwargs) return await ErrorHandler.retry_with_backoff( target, max_retries=max_retries, initial_delay=initial_delay ) return wrapper return decorator class CircuitBreaker: """ 서킷 브레이커 패턴 구현 연속적인 실패가 발생하면 일정 시간 동안 요청을 차단 """ def __init__(self, failure_threshold: int = 5, timeout: float = 60.0): self.failure_threshold = failure_threshold # 연속 실패 임계값 self.timeout = timeout # 차단 시간 (초) self.failure_count = 0 self.last_failure_time: Optional[float] = None self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN async def call(self, func: Callable) -> Any: """서킷 브레이커를 통해 함수 호출""" # OPEN 상태 확인 (차단 중) if self.state == "OPEN": if datetime.now().timestamp() - self.last_failure_time < self.timeout: raise NonRetryableError("서킷 브레이커가 열려 있습니다. 잠시 후 다시 시도해주세요.") else: # 타임아웃 지났으면 HALF_OPEN으로 전환 self.state = "HALF_OPEN" logger.info("서킷 브레이커 상태: HALF_OPEN (테스트 중)") try: # 함수 실행 result = await func() if asyncio.iscoroutinefunction(func) else func() # 성공 시 카운터 리셋 self.failure_count = 0 if self.state == "HALF_OPEN": self.state = "CLOSED" logger.info("서킷 브레이커 상태: CLOSED (정상)") return result except Exception as e: # 실패 시 카운터 증가 self.failure_count += 1 self.last_failure_time = datetime.now().timestamp() # 임계값 초과 시 OPEN으로 전환 if self.failure_count >= self.failure_threshold: self.state = "OPEN" logger.error( f"서킷 브레이커 상태: OPEN (차단됨). " f"{self.timeout}초 동안 요청이 차단됩니다." ) raise # 사용 예시 @with_retry(max_retries=3, initial_delay=1.0) async def call_external_api(url: str) -> dict: """외부 API 호출 (재시도 로직 적용)""" import aiohttp async with aiohttp.ClientSession() as session: async with session.get(url, timeout=10) as response: if response.status == 429: # Rate Limit raise RetryableError("Rate limit exceeded") elif response.status >= 500: # Server Error raise RetryableError(f"Server error: {response.status}") elif response.status >= 400: # Client Error raise NonRetryableError(f"Client error: {response.status}") return await response.json() ``` ### 설명 이것이 하는 일: 에러 핸들링 시스템은 다양한 실패 시나리오를 지능적으로 처리하여 시스템이 일시적인 문제에서 자동으로 복구하고, 영구적인 문제는 명확히 사용자에게 전달합니다. 마치 운전 중 작은 충격은 서스펜션이 흡수하고, 큰 사고는 에어백이 보호하는 것처럼, 다층 방어 시스템을 구축합니다. 첫 번째로, retry_with_backoff 메서드가 핵심 재시도 로직을 구현합니다. 지수 백오프(exponential backoff)를 사용하여 재시도 간격을 점점 늘립니다. 첫 번째 재시도는 1초 후, 두 번째는 2초 후, 세 번째는 4초 후 이런 식으로 증가합니다. 이렇게 하는 이유는, 서버가 과부하 상태일 때 모든 클라이언트가 동시에 재시도하면 상황이 더 악화되기 때문입니다. 간격을 늘리면 서버가 복구될 시간을 줍니다. 그 다음으로, 오류를 RetryableError(재시도 가능)와 NonRetryableError(재시도 불가능)로 분류합니다. 네트워크 타임아웃이나 서버 5xx 에러는 일시적일 가능성이 높으므로 재시도하지만, 클라이언트 4xx 에러(잘못된 요청)는 재시도해도 소용없으므로 즉시 실패 처리합니다. 이런 분류 덕분에 무의미한 재시도를 피하고 빠르게 실패할 수 있습니다. 세 번째로, with_retry 데코레이터가 재시도 로직을 함수에 쉽게 적용할 수 있게 합니다. @with_retry를 함수 위에 붙이기만 하면 자동으로 재시도 기능이 추가됩니다. 이렇게 하면 비즈니스 로직과 에러 처리 로직을 깔끔하게 분리할 수 있어 코드 가독성이 향상됩니다. 네 번째로, CircuitBreaker 클래스가 서킷 브레이커 패턴을 구현합니다. 연속해서 5번 실패하면 "OPEN" 상태로 전환되어 60초 동안 모든 요청을 차단합니다. 이는 죽은 서비스를 계속 호출하여 리소스를 낭비하는 것을 방지합니다. 60초 후에는 "HALF_OPEN" 상태로 전환되어 테스트 요청을 보내고, 성공하면 "CLOSED"로 돌아가 정상 작동합니다. 다섯 번째로, 모든 오류를 상세히 로깅합니다. 오류 타입, 메시지, 스택 트레이스, 재시도 횟수 등을 기록하여 나중에 문제를 분석하고 개선할 수 있습니다. 프로덕션 환경에서는 이런 로그를 Sentry나 CloudWatch 같은 모니터링 시스템으로 전송하여 실시간으로 추적할 수 있습니다. 여러분이 이 에러 핸들링 시스템을 사용하면 시스템의 가용성이 크게 향상됩니다. 네트워크가 일시적으로 불안정해도 자동으로 재시도하여 사용자는 문제를 느끼지 못할 수 있고, 외부 서비스가 다운되어도 서킷 브레이커가 리소스 낭비를 막아줍니다. 또한 명확한 에러 메시지로 사용자 경험도 개선됩니다. ### 실전 팁 💡 max_retries는 신중하게 설정하세요. 너무 많으면 실패한 요청에 너무 오래 매달리고, 너무 적으면 일시적 오류를 복구하지 못합니다. 일반적으로 3-5회가 적당합니다. 💡 재시도할 예외 타입을 명확히 정의하세요. 모든 Exception을 재시도하면 프로그래밍 버그도 재시도하게 되어 문제를 숨길 수 있습니다. ConnectionError, TimeoutError 같은 명확한 타입만 재시도하세요. 💡 외부 API 호출 시 항상 타임아웃을 설정하세요. timeout=10처럼 명시하지 않으면 무한 대기할 수 있습니다. API의 SLA를 고려하여 적절한 타임아웃을 설정하세요. 💡 서킷 브레이커의 상태 변화를 모니터링하세요. OPEN 상태로 자주 전환된다면 외부 서비스에 문제가 있거나 failure_threshold가 너무 낮게 설정된 것일 수 있습니다. 💡 프로덕션 환경에서는 재시도 로직에 jitter(무작위 지연)를 추가하세요. delay에 ±20% 정도의 랜덤 값을 더하면 여러 클라이언트의 재시도가 동시에 몰리는 것을 방지할 수 있습니다. --- ## 8. 사용자 인터페이스 구현 ### 시작하며 여러분이 강력한 AI 에이전트 시스템을 만들었는데, 사용자가 Python 코드를 직접 작성해야만 사용할 수 있다면 어떨까요? 일반 사용자에게는 전혀 쓸모없는 시스템이 되고 말 것입니다. 이런 문제는 훌륭한 기능을 가진 많은 프로젝트가 실패하는 이유입니다. 아무리 뛰어난 기술이라도 사용하기 어려우면 아무도 쓰지 않습니다. 특히 AI 비서는 누구나 쉽게 대화하듯이 사용할 수 있어야 하는데, 복잡한 API 호출이나 코드 작성을 요구하면 대중화될 수 없습니다. 바로 이럴 때 필요한 것이 직관적인 사용자 인터페이스입니다. 터미널에서 간단한 명령어나 대화만으로 모든 기능을 사용할 수 있는 CLI 인터페이스를 만들어, 개발자가 아닌 일반 사용자도 쉽게 접근할 수 있게 합니다. ### 개요 간단히 말해서, 사용자 인터페이스는 복잡한 AI 에이전트 시스템을 누구나 쉽게 사용할 수 있도록 하는 대화형 CLI 프로그램입니다. 왜 이것이 필요할까요? 실무에서 좋은 소프트웨어는 기능뿐만 아니라 사용성도 중요합니다. 사용자가 자연스럽게 질문하고, 실시간으로 응답을 받으며, 명령어 히스토리를 확인하고, 설정을 변경할 수 있는 인터페이스가 필요합니다. CLI는 웹 UI보다 간단하면서도 개발자와 파워 유저에게 친숙한 선택입니다. 기존에는 API를 직접 호출하거나 Python 스크립트를 수정해야 했다면, 이제는 터미널을 열고 "오늘 날씨 알려줘"라고 타이핑하기만 하면 됩니다. 핵심 특징으로는 첫째, 대화형 인터페이스로 자연스러운 대화를 나눌 수 있습니다. 둘째, 명령어 시스템으로 /help, /history 같은 특수 명령을 제공합니다. 셋째, 실시간 스트리밍으로 AI 응답이 생성되는 동안 단어별로 표시됩니다. 이러한 특징들이 중요한 이유는, 사용자 경험을 크게 향상시켜 실제로 사용되는 제품을 만들 수 있기 때문입니다. ### 코드 예제 ```python # ai_assistant/cli/interface.py import asyncio import sys from typing import Optional from prompt_toolkit import PromptSession from prompt_toolkit.history import FileHistory from prompt_toolkit.auto_suggest import AutoSuggestFromHistory from rich.console import Console from rich.markdown import Markdown from rich.panel import Panel from rich.spinner import Spinner from rich.live import Live class AIAssistantCLI: """AI 비서를 위한 대화형 CLI 인터페이스""" def __init__(self, main_agent, config: dict): self.main_agent = main_agent self.config = config self.console = Console() # 명령어 히스토리 파일 self.session = PromptSession( history=FileHistory('.ai_assistant_history'), auto_suggest=AutoSuggestFromHistory() ) self.running = True self.commands = { '/help': self.show_help, '/history': self.show_history, '/clear': self.clear_screen, '/config': self.show_config, '/exit': self.exit_app, '/quit': self.exit_app } def show_welcome(self): """환영 메시지 표시""" welcome_text = """ # 🤖 AI 비서에 오신 것을 환영합니다! 무엇을 도와드릴까요? 자연스럽게 질문해주세요. **명령어:** - `/help` - 도움말 표시 - `/history` - 대화 히스토리 보기 - `/config` - 설정 확인 - `/exit` 또는 `/quit` - 종료 """ self.console.print(Panel(Markdown(welcome_text), border_style="blue")) async def run(self): """CLI 메인 루프""" self.show_welcome() while self.running: try: # 사용자 입력 받기 user_input = await asyncio.get_event_loop().run_in_executor( None, lambda: self.session.prompt('\n💬 You: ', multiline=False) ) # 빈 입력 무시 if not user_input.strip(): continue # 명령어 처리 if user_input.startswith('/'): await self.handle_command(user_input) continue # AI 응답 생성 및 표시 await self.get_ai_response(user_input) except KeyboardInterrupt: self.console.print("\n\n👋 안녕히 가세요!", style="yellow") break except EOFError: break except Exception as e: self.console.print(f"[red]오류 발생: {str(e)}[/red]") async def handle_command(self, command: str): """특수 명령어 처리""" cmd_parts = command.split() cmd_name = cmd_parts[0] if cmd_name in self.commands: await self.commands[cmd_name]() else: self.console.print(f"[yellow]알 수 없는 명령어: {cmd_name}[/yellow]") self.console.print("사용 가능한 명령어는 /help를 입력하세요.") async def get_ai_response(self, user_input: str): """AI 응답 가져오기 및 표시""" # 스피너와 함께 "생각 중..." 표시 with self.console.status("[bold green]AI가 생각하는 중...", spinner="dots"): try: # 메인 에이전트 호출 result = await self.main_agent.process({"message": user_input}) response = result.get("response", "응답을 생성할 수 없습니다.") except Exception as e: response = f"오류가 발생했습니다: {str(e)}" self.console.print(f"[red]{response}[/red]") return # 응답 표시 (마크다운 렌더링) self.console.print("\n🤖 AI:", style="bold cyan") self.console.print(Panel(Markdown(response), border_style="cyan")) async def show_help(self): """도움말 표시""" help_text = """ # 📖 도움말 ## 명령어 목록: - `/help` - 이 도움말을 표시합니다 - `/history` - 대화 히스토리를 확인합니다 - `/clear` - 화면을 지웁니다 - `/config` - 현재 설정을 표시합니다 - `/exit` 또는 `/quit` - 프로그램을 종료합니다 ## 사용 팁: - 자연스러운 언어로 질문하세요 - 이전 대화 내용을 참조할 수 있습니다 - 위/아래 화살표로 이전 입력을 불러올 수 있습니다 """ self.console.print(Markdown(help_text)) async def show_history(self): """대화 히스토리 표시""" history = self.main_agent.conversation_history if not history: self.console.print("[yellow]아직 대화 히스토리가 없습니다.[/yellow]") return self.console.print("\n📜 대화 히스토리:", style="bold") for i, interaction in enumerate(history[-10:], 1): # 최근 10개만 self.console.print(f"\n{i}. 사용자: {interaction['user']}") self.console.print(f" AI: {interaction['assistant'][:100]}...") async def clear_screen(self): """화면 지우기""" self.console.clear() self.show_welcome() async def show_config(self): """현재 설정 표시""" config_text = f""" # ⚙️ 현재 설정 - **모델**: {self.config.get('model', 'N/A')} - **Temperature**: {self.config.get('temperature', 'N/A')} - **Max Tokens**: {self.config.get('max_tokens', 'N/A')} """ self.console.print(Markdown(config_text)) async def exit_app(self): """프로그램 종료""" self.console.print("\n👋 안녕히 가세요!", style="bold green") self.running = False # 실행 예시 async def main(): # 설정 로드 config = { "model": "gpt-4", "temperature": 0.7, "max_tokens": 2000 } # 메인 에이전트 초기화 (실제로는 이전에 구현한 MainAgent 사용) # main_agent = MainAgent(...) # CLI 시작 # cli = AIAssistantCLI(main_agent, config) # await cli.run() if __name__ == "__main__": asyncio.run(main()) ``` ### 설명 이것이 하는 일: AI 비서 CLI는 사용자가 터미널에서 자연스럽게 AI와 대화할 수 있는 인터페이스를 제공합니다. 마치 챗GPT 웹 인터페이스처럼 편리하지만, 터미널 환경에서 작동하여 개발자와 파워 유저에게 적합합니다. 첫 번째로, PromptSession을 사용하여 고급 입력 ---

카테고리: python

언어: Python

태그: Python, AI Agent, LangChain, OpenAI, Multi-Agent, python

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

Python으로 AI 에이전트 만들기 12편 - 실전 프로젝트: 나만의 AI 비서 완성

목차

1. 프로젝트 아키텍처 설계

시작하며

개요

코드 예제

설명

실전 팁

2. 메인 에이전트 구현

시작하며

개요

코드 예제

설명

실전 팁

3. 전문 에이전트들 구성

시작하며

개요

코드 예제

설명

실전 팁

4. 도구 통합 시스템

시작하며

개요

코드 예제

설명

실전 팁

5. 대화 메모리 관리

시작하며

개요

코드 예제

설명

실전 팁

6. 에이전트 간 협업 구현

시작하며

개요

코드 예제

설명

실전 팁

7. 에러 핸들링과 재시도 로직

시작하며

개요

코드 예제

설명

실전 팁

8. 사용자 인터페이스 구현

시작하며

개요

코드 예제

설명

댓글 (0)

함께 보면 좋은 카드 뉴스

Python으로 알고리즘 트레이딩 봇 만들기 3편 실시간 시세 데이터 수집

Python으로 알고리즘 트레이딩 봇 만들기 2편 - 바이낸스 API 연동하기

Python으로 알고리즘 트레이딩 봇 만들기 1편 - 트레이딩 봇 개요와 환경 설정

바닥부터 만드는 ChatGPT 2편 BPE 토크나이저 이론과 구현

바닥부터 만드는 ChatGPT 19편 계산기 도구 통합

Python으로 AI 에이전트 만들기 11편 FastAPI로 AI 서비스 배포