[toyp] CDocs - 3.에이전트 코어

[에이전트 코어] MCP와 번개 같은 응답 속도 만들기

"MCP 서버를 질문마다 새로 띄우면 너무 오래 걸려요. '캐싱'을 통해 세션을 유지하는 것이 필요합니다."

핵심 코드: src/chain.py (캐싱 로직)
AsyncExitStack을 사용해 한 번 연결된 도구들을 계속 재사용합니다.

import os
import asyncio
import re
import threading
from contextlib import asynccontextmanager
from typing import Any

# [핵심 라이브러리 설명]
# ChatOllama: 로컬에 설치된 Ollama 모델과 대화하기 위한 인터페이스입니다.
# AgentExecutor: 에이전트가 생각하고 도구를 결정하고 실행하는 전체 루프를 관리합니다.
# load_mcp_tools: MCP(Model Context Protocol) 서버로부터 도구를 가져와 LangChain에서 쓸 수 있게 변환합니다.
# AsyncExitStack: 비동기 리소스(예: MCP 서버 연결)를 안전하게 관리하고 프로그램 종료 시 한꺼번에 닫아줍니다.
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableConfig
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.session import ClientSession
from contextlib import AsyncExitStack

# 기본 도구와 로깅 유틸리티를 가져옵니다.
from src.agent.tools import get_base_tools
from src.utils.logger import get_logger

logger = get_logger("AgentProcessor")

# qwen3:1.7b 모델을 사용합니다. 한국어 능력이 뛰어나고 사고 과정(Thinking) 기능이 강력합니다.
DEFAULT_MODEL = "qwen3:1.7b"

# ──────────────────────────────────────────────
# MCP 세션 관리 (성능 최적화용 캐시)
# 질문할 때마다 MCP 서버를 띄우면 5~10초씩 걸립니다.
# 전역 변수를 활용해 세션을 '캐시'해두면 두 번째 질문부터는 응답 속도가 비약적으로 빨라집니다!
# ──────────────────────────────────────────────
_mcp_stack: AsyncExitStack | None = None
_mcp_session: ClientSession | None = None
_cached_executor: AgentExecutor | None = None

# Thinking 모델(qwen3)이 내뱉는 <think>...</think> 블록을 사용자 답변에서 지우기 위한 정규식입니다.
_THINK_RE = re.compile(r'<think>.*?(?:</think>|$)', re.DOTALL | re.IGNORECASE)

def strip_think_blocks(text: str) -> str:
    """
    답변 텍스트에서 AI의 '속마음(<think>)' 부분을 제거합니다.
    실제로 사용자에게 보여줄 '깔끔한 정답'만 남기기 위해 꼭 필요합니다.
    """
    if not text:
        return text
    return _THINK_RE.sub('', text).strip()

# 에이전트의 '인격'과 '행동 규칙'을 정의하는 시스템 프롬프트입니다.
SYSTEM_PROMPT = (
    "당신은 CDocs의 유능한 AI 에이전트입니다.\n"
    "1. 반드시 모든 답변을 한국어로만 작성하십시오.\n"
    "2. 도구 호출 시 한글 텍스트를 그대로 사용하세요 (유니코드 변환 금지).\n"
    "3. 파일 읽기/쓰기 시 반드시 'data/docs' 폴더를 기준으로 상대 경로를 사용하십시오.\n"
    "4. 답변 생성 시 근거가 없는 내용은 절대 지어내지 말고 정직하게 모른다고 하십시오."
)

@asynccontextmanager
async def get_agent_with_mcp(model_name: str = DEFAULT_MODEL):
    """
    MCP 서버 세션과 에이전트를 캐싱하여 반환합니다. 
    최초 호출 시에만 서버를 띄우고 이후엔 재사용하는 고성능 전략입니다.
    """
    global _mcp_stack, _mcp_session, _cached_executor

    # 이미 만들어진 실행기가 있다면 서버를 다시 띄우지 않고 바로 사용합니다.
    if _cached_executor:
        yield _cached_executor
        return

    logger.info(f"에이전트 최초 초기화 시작 (모델: {model_name})...")
    
    # 1. 파일 시스템 경로 설정
    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    docs_path = os.path.join(project_root, "data", "docs")
    os.makedirs(docs_path, exist_ok=True) # 폴더가 없으면 미리 만듭니다.
    
    logger.info(f"MCP Filesystem 설정 경로: {docs_path}")

    # 2. MCP 서버 실행 파라미터 (Node.js의 npx를 사용해 로컬 파일 시스템 도구를 연동합니다)
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", docs_path],
        env=os.environ.copy(),
    )

    # 3. LLM 설정
    llm = ChatOllama(model=model_name, temperature=0, base_url="http://localhost:11434")

    # 4. MCP 연결 및 도구 로드
    if _mcp_stack is None:
        _mcp_stack = AsyncExitStack()

    try:
        # stdio_client: 외부 프로세스와 표준 입출력으로 통신을 시작합니다.
        read, write = await _mcp_stack.enter_async_context(stdio_client(server_params))
        _mcp_session = await _mcp_stack.enter_async_context(ClientSession(read, write))
        await _mcp_session.initialize()

        # mcp_tools: 파일 읽기/목록보기 등 MCP 서버가 제공하는 기능을 LangChain 도구로 변환합니다.
        mcp_tools = await load_mcp_tools(_mcp_session)
        all_tools = get_base_tools() + mcp_tools # 기본 도구와 MCP 도구를 합칩니다.

        # 5. 에이전트 생성
        prompt = ChatPromptTemplate.from_messages([
            ("system", SYSTEM_PROMPT),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        agent = create_tool_calling_agent(llm, all_tools, prompt)
        
        # AgentExecutor: 에이전트 실행의 심장부입니다. verbose=True로 설정하면 사고 과정을 로그로 볼 수 있습니다.
        _cached_executor = AgentExecutor(
            agent=agent,
            tools=all_tools,
            verbose=True,
            handle_parsing_errors=True,
            return_intermediate_steps=True, # 어떤 도구를 썼는지 정보를 얻기 위해 필요합니다.
        )

        yield _cached_executor

    except Exception as e:
        logger.error(f"MCP 초기화 중 치명적 오류 발생: {e}")
        # 오류 시에도 기본 도구만이라도 쓸 수 있도록 최소한의 실행기를 반환합니다.
        yield _cached_executor


async def ask_question_async(question: str, chat_history: list | None = None) -> dict:
    """비동기적으로 에이전트에게 질문하고 답변과 사고 과정을 받아옵니다."""
    async with get_agent_with_mcp() as agent_executor:
        try:
            # ainvoke: 비동기 실행 명령입니다.
            response = await agent_executor.ainvoke(
                {"input": question, "chat_history": chat_history or []}
            )

            # 답변 정제 (Think 블록 제거)
            raw_answer = response.get("output", "")
            clean_answer = strip_think_blocks(raw_answer)

            # 에이전트가 어떤 도구를 썼는지 정리 (사고 과정 시각화용)
            thoughts = []
            for action, observation in response.get("intermediate_steps", []):
                thoughts.append({
                    "tool": action.tool,
                    "thought": getattr(action, "log", ""),
                    "observation": strip_think_blocks(str(observation))[:500]
                })

            return {"answer": clean_answer, "thoughts": thoughts}
        except Exception as e:
            logger.error(f"에이전트 질의 중 오류: {e}")
            return {"answer": f"죄송합니다. 오류가 발생했습니다: {e}", "thoughts": []}

def ask_question(question: str, chat_history: list | None = None) -> dict:
    """Streamlit(동기)에서 비동기 함수를 호출하기 위한 '동기 브릿지' 함수입니다."""
    try:
        import nest_asyncio
        nest_asyncio.apply() # 이미 돌아가는 이벤트 루프에 중첩 실행을 허용합니다.
    except: pass

    try:
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(ask_question_async(question, chat_history))
    except:
        loop = asyncio.new_event_loop()
        return loop.run_until_complete(ask_question_async(question, chat_history))
async def close_mcp():
    """MCP 세션 및 관련 리소스를 안전하게 종료합니다 (프로그램 종료 시 호출 권장)."""
    global _mcp_stack, _mcp_session, _cached_executor
    if _mcp_stack:
        logger.info("MCP 세션 종료 중...")
        await _mcp_stack.aclose()
        _mcp_stack = None
        _mcp_session = None
        _cached_executor = None
        logger.info("MCP 세션 종료 완료.")

async def main_cli():
    """CLI 환경에서 에이전트를 직접 테스트하기 위한 엔트리포인트입니다."""
    try:
        # 테스트용 질문을 던져봅니다.
        res = await ask_question_async("오늘 서울 날씨 어때?")
        print(f"\n[답변]: {res['answer']}")
    finally:
        # CLI 종료 시에는 명시적으로 MCP 세션을 닫아줍니다.
        await close_mcp()

if __name__ == "__main__":
    # asyncio.run()을 사용하여 main_cli를 동기 방식으로 실행합니다.
    asyncio.run(main_cli())

주요 내용이 담긴 파일이라 좀 길지만, 주석을 따라 가다 보면....
실행은 아래와 같습니다.