[AI] langchain + mcp 구현 도중 문제해결 과정

1. 대선 후보 토론 요약하다가...

• 전체 코드는 아래와 같다.
• 날짜 + 토론 HumanMessage 가 들어가면 get_discussion 도구를 사용하게 유도한다.
• 날짜가 없을 경우 날짜를 묻는다.
• faiss 에서 토론 벡터 데이터를 읽는다.
• 벡터 데이터를 참고하여 토론 내용을 그대로 출력한다.

from langchain.agents import initialize_agent, AgentType
from llm_util import getLLM
from langchain.memory import ConversationBufferMemory
from langchain.schema import SystemMessage
from langchain.tools import Tool
from pydantic import BaseModel
from typing import Optional
from embedding_util import getEmbeddingHuggingface  
from langchain_community.vectorstores.faiss import FAISS
from langchain.chains import RetrievalQA
from date_util import getNormalizedDate
from prompt_util import getCustomPrompt

# 1. LLM 설정
llm = getLLM()

# 2. 도구 입력 스키마 정의
class DiscussionInput(BaseModel):
    discussion_date: Optional[str] = None

# 3. 도구 함수 정의
def get_discussion(discussion_date: Optional[str] = None):

    date = getNormalizedDate(discussion_date)

    embeddings = getEmbeddingHuggingface()
    faiss_index_path = f"script_{date}"
    vectorstore = FAISS.load_local(
        folder_path=faiss_index_path,
        embeddings=embeddings,
        allow_dangerous_deserialization=True
    )

    retriever = vectorstore.as_retriever()
    chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=retriever,
        chain_type_kwargs={"prompt": getCustomPrompt()}
    )

    result = chain({"query": "토론 내용을 그대로 알려줄래요?"})
    return result['result']

# 4. 도구 정의
# 역할 : 도구의 기능, 입력 요구사항을 설명
# LLM 이 접근하는 시점 : 에이전트가 어떤 도구를 사용할지 판단할 때 참조
# 우선순위 : 가장 중요
discussion_tool = Tool(
    name="get_discussion",
    func=get_discussion,
    description="""
        **특정 날짜의 토론 내용을 그대로 반환하는 도구입니다.**
        날짜(date)가 포함된 입력이 오면 이 도구를 호출하세요.

        예: "20250521", "2025년 5월 21일" 등
    """,
    args_schema=DiscussionInput,
)

# 5. 메모리
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=False)


# 6. 에이전트 프롬프트 설정
agent_kwargs = {
    # 역할 : 에이전트에게 전반적인 행동 방침을 주는 역할
    # LLM이 접근하는 시점 : 실행 전 초기 시스템 설정
    "system_message": SystemMessage(content="""
        ✅ **중요: 도구에서 반환된 내용을 추가적인 가공 없이 사용자에게 그대로 출력하세요.**
        ✅ 답변은 무조건 한국어로 하세요.
    """)
}

# 7. 에이전트 초기화
agent = initialize_agent(
    tools=[discussion_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    memory=memory,
    handle_parsing_errors=True,
    max_iterations=5,
    agent_kwargs=agent_kwargs,
    early_stopping_method="generate",
)

# 8. 콘솔 대화 루프
print("CHAT 에이전트 시작! (종료: 'exit')")

while True:
    user_input = input("🧑‍You: ")
    if user_input.lower() in ["exit", "quit"]:
        break
    try:
        response = agent.run(user_input)
        print("AI:", response)
    except Exception as e:
        print(f"❗ 오류 발생: {str(e)}")

2. 그런데...

2-1. HumanMessage (OK)

2-2. Thought, Action, Action Input (OK)

2-3. Observation (OK?)

2-4. 또다시 Thought(?), Final Answer

3. 의도치 않은 Thought 발생

본인은 get_discussion 도구에서 반환한 내용을 그대로 반환하고 싶었는데 llm 이 다시 판단하여 조금 다른 응답을 내어 주었다.

프로세스를 끝내고 싶을 때에 끝낼 수는 없을까?

4. 대~충 이런듯?

인간 영역 —> Human
Langchain 영역(mcp 도구 선택 영역) —> Action, Action Input
도구 사용 영역(return) —> Observation
Langchain 영역(값을 본 후 판단하는 영역) —> Thought, Final Answer

5. 이럴땐 LangGraph 를 써보자.

% uv pip install langgraph

import re
from typing import TypedDict, Optional
from pydantic import BaseModel

from langchain_community.vectorstores.faiss import FAISS
from langchain.chains import RetrievalQA
from langchain_core.tools import Tool as LangGraphTool
from langgraph.graph import StateGraph, END
from llm_util import getLLM
from embedding_util import getEmbeddingHuggingface
from date_util import getNormalizedDate
from prompt_util import getCustomPrompt
from langgraph.graph import StateGraph
from graphviz import Digraph

# chat history
chat_history = []

# LLM 설정
llm = getLLM()

# 상태 정의
class AgentState(TypedDict):
    input: str
    output: Optional[str]
    chat_history: list

# 도구 입력 스키마
class DiscussionInput(BaseModel):
    discussion_date: Optional[str] = None

# get_discussion 도구 정의
def get_discussion(discussion_date: Optional[str] = None):
    date = getNormalizedDate(discussion_date)
    embeddings = getEmbeddingHuggingface()
    faiss_index_path = f"script_{date}"
    vectorstore = FAISS.load_local(
        folder_path=faiss_index_path,
        embeddings=embeddings,
        allow_dangerous_deserialization=True
    )
    retriever = vectorstore.as_retriever()
    chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=retriever,
        chain_type_kwargs={"prompt": getCustomPrompt()}
    )
    result = chain({"query": "토론 내용을 그대로 알려줄래요?"})
    return result['result']

# LangGraph 용 도구 래핑
discussion_tool = LangGraphTool.from_function(
    name="get_discussion",
    description="특정 날짜의 토론 내용을 그대로 반환하는 도구입니다.",
    func=get_discussion,
    args_schema=DiscussionInput
)

# 도구 실행 노드
def invoke_tool(state: AgentState):
    output = discussion_tool.invoke({"discussion_date": state["input"]})
    return {
        "input": state["input"],
        "output": output,
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", output)],
    }

# 기본 응답 노드 (날짜 없을 때)
def default_reply(state: AgentState):
    output = "❓ 날짜 정보가 없어 토론 내용을 불러올 수 없어요. 예: '20250521', '2025년 5월 21일'"
    return {
        "input": state["input"],
        "output": output,
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", output)],
    }

# 날짜 여부에 따른 라우팅 조건 함수
def route_based_on_date(state: AgentState):
    text = state["input"]
    # "20250521" or "2025년 5월 21일" 같은 날짜가 포함되어 있으면 get_discussion 실행
    if re.search(r"\d{8}|\d{4}년\s*\d{1,2}월\s*\d{1,2}일", text):
        return "get_discussion"
    return "default_reply"

def router(state: AgentState) -> AgentState:
    # 그대로 전달 (조건 분기만 하기 때문에 상태 변경 없음)
    return state

# 1. LangGraph 구성
builder = StateGraph(AgentState)

# 2. 모든 노드 등록
builder.add_node("router", router)  # <- 이거 빠지면 오류 발생
builder.add_node("get_discussion", invoke_tool)
builder.add_node("default_reply", default_reply)

# 3. 라우팅 조건 설정
builder.set_entry_point("router")
builder.add_conditional_edges("router", route_based_on_date, {
    "get_discussion": "get_discussion",
    "default_reply": "default_reply"
})

# 4. 각 종료점 설정
builder.add_edge("get_discussion", END)
builder.add_edge("default_reply", END)

# 5. 컴파일
graph = builder.compile()

# 대화 루프 시작
print("LangGraph MCP Agent 시작! (종료: 'exit')")

while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "quit"]:
        break

    state = {
        "input": user_input,
        "output": None,
        "chat_history": chat_history,
    }

    try:
        result = graph.invoke(state)
        chat_history = result["chat_history"]
        print("AI:", result["output"])
    except Exception as e:
        print(f"❗ 오류 발생: {str(e)}")

5-1. LangGraph 를 구성하는 builder 설명

단계	함수 / 메서드	설명
1️⃣	StateGraph(AgentState)	상태 타입(TypedDict)을 기반으로 상태 기반 그래프 생성
2️⃣	add_node(name, fn)	각 노드를 그래프에 등록 (예: router, get_discussion, default_reply)
3️⃣	set_entry_point("router")	그래프 실행의 시작 지점을 설정
4️⃣	add_conditional_edges("router", route_fn, routes)	조건 분기 함수를 사용해 다음 노드를 선택 (예: 날짜가 포함되어 있으면 get_discussion으로 분기)
5️⃣	add_edge("node", END)	해당 노드 실행 후 종료 지점으로 연결 (흐름 마무리)
6️⃣	compile()	위 정의를 바탕으로 LangGraph를 실행 가능한 상태로 컴파일

6. 일단 결과 확인

바로 응답해줬다! 감동...ㅠㅠ

7. 실행 순서도 출력 가능하다.

uv pip install grandalf

# ✅ LangGraph Flow 출력
print("--- LangGraph Flow ---")
print(graph.get_graph().draw_ascii())
print("----------------------")