64일차. Modular RAG - LangGraph / MCP

 

64일 차 회고.

 

 오늘 수업은 따라가기 어렵기도 했고, 단위 프로젝트 때문에 정신이 없기도 해서 제대로 참여하지 못했다.

 

 

 

 

1. Modular RAG

 

 

1-1. LangGraph

 

LangGraph

• LLM을 사용한 애플리케이션에서 상태를 유지하고, 여러 에이전트 간의 협업을 설계할 수 있도록 도와주는 프레임워크
• LangChain의 기능을 확장한 형태로, 복잡한 에이전트 런타임을 구현할 때 핵심적인 구조인 순환 그래프를 만들고 관리할 수 있다.

 

LangGraph 구성요소

• Node
  • 각각의 작업 단위
  • 수행하고자 하는 로직을 포함한다.
• Edge
  • 노드와 노드를 연결하는 경로
  • 각 노드의 실행 결과나 상태에 따라 다음 노드를 결정하며, 이 과정에서 노드 간 커뮤니케이션이 이루어진다.
• State
  • 그래프 실행 도중 생성되는 데이터의 상태를 나타낸다.
  • 각 노드가 수행한 작업의 결과를 저장하고, 이후 노드들이 해당 정보를 활용할 수 있도록 한다.
• Graph
  • 노드와 엣지의 모음으로 구성된 전체 워크플로우 구조

 

Simple LangGraph

• State

from typing_extensions import TypedDict

class State(TypedDict):
    input:str
    result:int

• Node

def len_str(state:State) -> State:
    input = state["input"]
    result = len(input)
    return {
        "result": result
    }

def add_one(state:State) -> State:
    input = state["result"]
    result = input + 1
    return {
        "result": result
    }

• Graph

# Create
from langgraph.graph import StateGraph

simple_graph = StateGraph(State)

# Add Nodes
simple_graph.add_node(
    node="len_str_node",
    action=len_str
)

simple_graph.add_node(
    node="add_one_node",
    action=add_one
)

# Add Edge
from langgraph.graph import START, END

simple_graph.add_edge(
    start_key=START,
    end_key="len_str_node"
)

simple_graph.add_edge(
    start_key="len_str_node",
    end_key="add_one_node"
)

simple_graph.add_edge(
    start_key="add_one_node",
    end_key=END
)

# Compile Graph
graph = simple_graph.compile()

• Display Graph

from IPython.display import display, Image

try:
    display(
        Image(
            graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass

• Run

result = graph.invoke(
    input={
        "input": "Hello World"
    }
)
result
# {'input': 'Hello World', 'result': 12}

 

LangGraph with LLM

• State

from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

• LLM Model

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

• Node

def chatbot(state:State) -> State:
    return {
        "messages": [
            llm.invoke(state["messages"])
        ]
    }

• Graph

# Create
llm_graph = StateGraph(State)

# Add Nodes
llm_graph.add_node(
    node="chatbot_node",
    action=chatbot
)

# Add Edge
llm_graph.set_entry_point("chatbot_node")
llm_graph.set_finish_point("chatbot_node")

# Compile Graph
graph = llm_graph.compile()

• Display Graph

from IPython.display import display, Image

try:
    display(
        Image(
            graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass

• Run

from langchain_core.messages import HumanMessage

result = graph.invoke(
    input={
        "messsages": [
            HumanMessage(
                content="한국의 수도는?"
            )
        ]
    }
)
result["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================

한국의 수도는 서울입니다.
"""

 

LangGraph with Conditional Edge

• State

from typing_extensions import TypedDict, Optional

class State(TypedDict):
    input: Optional[str]
    node_output: Optional[int]
    is_stop: Optional[bool]

• Node

def len_str(state:State) -> State:
    input = state["input"]
    result = len(input)
    return {
        "node_output": result,
        "is_stop": False
    }

def add_one(state:State) -> State:
    input = state["node_output"]
    is_stop = state["is_stop"]
    result = input + 1
    
    if result > 10:
        is_stop = True
    
    return {
        **state,
        "node_output": result,
        "is_stop": False
    }

def add_two(state:State) -> State:
    input = state["node_output"]
    result = input + 2
    return {
        **state,
        "node_output": result
    }

• Graph

# Create
condition_graph = StateGraph(State)

# Add Nodes
condition_graph.add_node(
    node="len_str_node",
    action=len_str
)

condition_graph.add_node(
    node="add_one_node",
    action=add_one
)

condition_graph.add_node(
    node="add_two_node",
    action=add_two
)

# Add Edge
condition_graph.add_edge(
    start_key=START,
    end_key="len_str_node"
)

condition_graph.add_edge(
    start_key="len_str_node",
    end_key="add_one_node"
)

condition_graph.add_edge(
    start_key="add_two_node",
    end_key="add_one_node"
)

# Add Conditional Edge
def is_stop(state:State) -> str:
    is_stop = state["is_stop"]
    
    if is_stop:
        return "go_stop"
    else:
        return "go_to_add_two_fnc"

condition_graph.add_conditional_edges(
    source="add_one_node",
    path=is_stop,
    path_map={
        "go_stop": END,
        "go_to_add_two_fnc": "add_two_node"
    }
)

# Compile
graph = condition_graph.compile()

• Display Graph

from IPython.display import display, Image

try:
    display(
        Image(
            graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass

• Run

for chunk in graph.stream(
    input={
        "input": "Hello"
    }
):
    print(chunk)
"""
{'len_str_node': {'node_output': 5, 'is_stop': False}}
{'add_one_node': {'input': 'Hello', 'node_output': 6, 'is_stop': False}}
{'add_two_node': {'input': 'Hello', 'node_output': 8, 'is_stop': False}}
{'add_one_node': {'input': 'Hello', 'node_output': 9, 'is_stop': False}}
{'add_two_node': {'input': 'Hello', 'node_output': 11, 'is_stop': False}}
{'add_one_node': {'input': 'Hello', 'node_output': 12, 'is_stop': True}}
"""

 

 

 

2. MCP

 

 

2-1. MCP

 

MCP(Model Context Protocol)

• LLM 기반 애플리케이션이 다양한 컨텍스트(데이터, 코드, 문서 등)를 보다 효과적으로 사용할 수 있도록 설계된 개방형 표준 프로토콜
• 여러 도구나 서비스와 연동하며 동작할 수 있다.

 

MCP 작동 방식

• MCP Host
  • LLM을 활용하는 애플리케이션
    • Claude, IDE, AI 도구 등
  • 컨텍스트가 필요한 작업을 수행할 때 MCP를 통해 외부 리소스에 접근한다.
• MCP Clienet
  • MCP Host와 MCP Server 사이에서 통신을 중계하는 클라이언트
  • MCP Server와 1:1 연결을 유지하며 요청을 전달한다.
• MCP Server
  
  • MCP 표준에 따라 기능을 노출하는 경량 서버
  • 다양한 데이터 소스(Local/Remote)에 접근할 수 있도록 처리한다.
• Local Data Source
  • MCP Server가 설치된 컴퓨터의 파일, 데이터베이스, 서비스 등
  • Local 시스템에서 직접 접근이 가능한 데이터 자원
• Remote Service
  • 외부 API나 클라우드 서비스처럼 인터넷을 통해 접근 가능한 시스템

 

MCP Server 생성

# Terminal
uv init .					# 프로젝트 생성

uv venv .venv -p 3.13				# 가상환경 생성
.\.venv\Scripts\activate

uv add "mcp[cli]"				# 관련 라이브러리 추가

# main.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MCPTestServer")

@mcp.tool()
def add(a: int, b: int) -> int:
    """
    Add two numbers
    """
    return a + b

# Terminal
mcp install main.py				# MCP Server 적용

# claude_desktop_config.json
{
  "mcpServers": {
    "MCPTestServer": {
      "command": "uv",
      "args": [
        "run",
        "--with",
        "mcp[cli]",
        "mcp",
        "run",
        "C:\\dev\\MCP\\2. mcp server\\main.py"
      ]
    }
  }
}