LangChain 고급 컴포넌트 (Agents, Tools, LangGraph) 활용하기 🚀 - 나만의 AI 비서 만들기 #2

들어가며

지난 포스트에서는 LangChain의 기본 개념과 Models, Prompts, OutputParser, Chains와 같은 핵심 컴포넌트들을 살펴보았습니다. 이번 포스트에서는 Agents, Tools 그리고 LangGraph에 대해 자세히 알아보겠습니다.

이 컴포넌트들을 활용해서 단순한 질문, 답변을 넘어서서 보다 복잡한 기능을 수행할 수 있는 AI 비서를 만들 수 있습니다.

LangChain 설치

랭체인을 사용하기 위해서는 라이브러리 설치가 필요합니다.

pip install langchain

Agents and Tools: 자율적인 AI 비서 만들기 🤖

Agents

LangChain의 Agent는 AI 시스템이 자율적으로 문제를 해결하는 핵심 컴포넌트입니다. 다음과 같은 특징을 가지고 있습니다:

• 자율적 의사결정: Agent는 사용자의 요청을 분석하고, 어떤 도구(Tools)를 사용할지 스스로 결정합니다.
• 문제 해결 과정: 단순히 답변만 하는 것이 아니라, 문제 해결을 위한 단계적 접근 방식을 취합니다.
• 도구 활용 능력: 다양한 도구(Tools)를 상황에 맞게 선택하고 활용할 수 있습니다.
• 반복적 실행: 필요한 경우 여러 도구를 순차적으로 사용하여 복잡한 작업을 완료합니다.

Tools

Tools는 AI 모델이 외부 세계와 상호작용할 수 있게 해주는 기능입니다. 쉽게 말해, AI에게 "능력"을 부여하는 것이라고 생각하면 됩니다.

Tools를 통해 AI는 다음과 같은 일들을 할 수 있게 됩니다:

• 웹 검색으로 최신 정보 찾기
• 계산기 사용하기
• 데이터베이스에서 정보 조회하기
• 외부 API 호출하기 (날씨, 주식, 뉴스 등)
• 파일 시스템 접근하기
• 이메일 보내기

즉, Tools는 AI의 한계를 극복하고 실제 세계의 작업을 수행할 수 있게 해주는 확장 기능이라고 볼 수 있습니다.

Agents와 Tools 사용 예시

다음은 기본적인 계산기 도구를 사용하는 예시입니다:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate


# 계산기 도구 정의
@tool
def calculator(expression: str) -> str:
    """수학 표현식을 계산합니다."""
    try:
        # 여기서는 예제를 위해 eval을 사용했으나, 보안상 지양해야 합니다.
        return str(eval(expression))
    except Exception as e:
        return f"계산 오류: {str(e)}"

# 프롬프트 템플릿 생성
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 수학 문제를 푸는 도우미입니다. 제공된 도구를 활용하여 문제를 해결하세요.",
        ),
        ("user", "{input}"),
        ("assistant", "{agent_scratchpad}"),
    ],
)

llm = ChatOpenAI(model="gpt-4o-mini")
agent = create_openai_tools_agent(llm, [calculator], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[calculator], verbose=True)

result = agent_executor.invoke({"input": "125 * 32를 계산해주세요",})
print(result["output"])

실행결과

위의 코드는 계산기 툴을 사용하여, 주어진 표현식을 계산하는 코드입니다.

1. @tool 데코레이터를 사용해 calculator 함수를 LangChain 도구로 등록합니다. 이 도구는 문자열로 된 수학 표현식을 받아 계산 결과를 반환합니다.
2. 프롬프트 템플릿에서 {agent_scratchpad}를 포함시켜, Agent가 사고 과정을 기록할 공간을 제공합니다.
3. create_openai_tools_agent 함수를 사용해 언어 모델, 도구, 프롬프트를 결합하여 에이전트를 생성합니다.
4. AgentExecutor를 통해 에이전트와 도구를 연결합니다.

💡 tool로 등록할 함수의 docstring과 타입힌트를 잘 명시해줄수록 AI가 알맞은 툴을 찾을 확률이 높아집니다.

서드파티 도구 연동하기

LangChain의 강력한 장점 중 하나는 다양한 서드파티 서비스와 쉽게 연동할 수 있다는 점입니다.

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.tools.serpapi import SerpAPIWrapper
from langchain_community.tools.openweathermap import OpenWeatherMapQueryRun

# Tavily 검색 도구
tavily_search = TavilySearchResults(api_key="your-tavily-api-key")

# SerpAPI 검색 도구
serpapi_search = SerpAPIWrapper(serpapi_api_key="your-serpapi-key")

# OpenWeatherMap 날씨 도구
weather_tool = OpenWeatherMapQueryRun(api_key="your-openweathermap-api-key")

# 도구 목록에 추가
tools = [tavily_search, serpapi_search, weather_tool]

# Agent 생성 및 실행
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 실행
result = agent_executor.invoke({"input": "서울의 오늘 날씨와 주요 뉴스를 알려주세요"})
print(result["output"])

이처럼 LangChain을 통해 다양한 외부 서비스(Notion, Slack, GitHub, Google Calendar 등)와 연동하여 AI의 기능을 확장할 수 있습니다.

LangGraph: 복잡한 AI 워크플로우 구축하기 📊

LangGraph는 LangChain의 확장 라이브러리로, 복잡한 AI 작업 과정을 단계별로 구성할 수 있게 해주는 도구입니다.

예를 들어:

1. 사용자 질문 이해하기
2. 필요한 정보 검색하기
3. 답변 작성하기
4. 필요하면 2단계로 돌아가 더 찾아보기

이렇게 AI가 따라야 할 작업 흐름을 명확하게 설계할 수 있어, 더 체계적이고 정확한 결과를 얻을 수 있습니다.

LangGraph 설치

pip install langgraph

랭그래프 예제 살펴보기

LangGraph는 복잡한 AI 워크플로우를 구축하는 데 특화된 도구입니다. 아래 예제는 질문-답변 시스템을 구현한 것으로, 사용자 질문 분석, 웹 검색, 명확화 요청, 답변 생성 등의 단계를 체계적으로 관리합니다.

이 예제를 통해 LangGraph가 어떻게 여러 단계의 AI 작업을 상태 기반 그래프로 연결하고, 각 단계에서 필요한 결정을 내리며 작업을 진행하는지 살펴보겠습니다.

특히, create_rag_system과 create_graph 코드를 주의깊게 살펴봐주세요. 나머지 코드들은 구현과 관련된 사항이니 가볍게 넘기셔도 좋습니다.

코드가 복잡하다보니, 주석을 통해 코드가 하는 일을 명시하였습니다.

from typing import TypedDict, List, Dict, Any, Literal
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph import StateGraph, END
from pydantic import BaseModel, Field


# 1. State 클래스: 워크플로우의 각 단계에서 공유되는 상태 정보를 정의
# 이 상태 객체는 그래프의 각 노드 간에 전달되며 데이터 흐름을 관리합니다
class State(TypedDict):
    question: str                # 사용자의 원래 질문
    search_query: str            # 검색에 사용될 최적화된 쿼리
    search_results: List[Dict[str, Any]]  # 검색 결과 저장
    need_web_search: bool        # 추가 웹 검색 필요 여부
    need_clarification: bool     # 질문 명확화 필요 여부
    clarification_question: str  # 사용자에게 물어볼 명확화 질문
    user_clarification: str      # 사용자의 명확화 응답
    answer: str                  # 최종 답변
    reasoning: str               # 답변 도출 과정
    confidence: float            # 답변 신뢰도
    sources: List[str]           # 참고 출처 목록

# 2. 출력 형식 정의: Pydantic 모델을 사용하여 LLM 출력의 구조화된 형식 지정
class SearchQueryOutput(BaseModel):
    search_query: str = Field(description="검색에 사용할 최적화된 쿼리")
    need_clarification: bool = Field(
        description="사용자 질문이 모호하여 명확화가 필요한지 여부"
    )
    clarification_question: str = Field(
        description="사용자에게 물어볼 명확화 질문 (필요한 경우)"
    )

class AnswerOutput(BaseModel):
    answer: str = Field(description="사용자 질문에 대한 최종 답변")
    reasoning: str = Field(description="답변에 도달한 추론 과정")
    confidence: float = Field(description="답변의 신뢰도 (0.0-1.0)")
    need_web_search: bool = Field(description="추가 웹 검색이 필요한지 여부")
    sources: List[str] = Field(description="답변에 사용된 출처 URL 목록")


# 3. 전체 RAG 시스템 실행 함수: 워크플로우 초기화 및 실행
def run_rag_system(question: str):
    """RAG 시스템 실행"""
    app = create_graph()
    # 초기 상태 설정
    initial_state = {
        "question": question,
        "search_query": "",
        "search_results": [],
        "need_web_search": False,
        "need_clarification": False,
        "clarification_question": "",
        "user_clarification": "",
        "answer": "",
        "reasoning": "",
        "confidence": 0.0,
        "sources": [],
    }
    # 그래프 실행
    result = app.invoke(initial_state)

    print("\n" + "=" * 50)
    print(f"질문: {question}")
    print("=" * 50)
    print(f"\n답변: {result['answer']}")
    print("\n추론 과정:")
    print(result["reasoning"])
    print(f"\n신뢰도: {result['confidence']:.2f}")
    print("\n출처:")

    for source in result["sources"]:
        print(f"- {source}")

    return result

# 4. 그래프 생성 함수: LangGraph의 핵심 - 워크플로우 구조 정의
def create_graph():
    # StateGraph 객체 생성 - State 타입을 기반으로 함
    workflow = StateGraph(State)

    # 5. 노드 추가: 각 노드는 워크플로우의 한 단계를 담당하는 함수
    workflow.add_node("analyze", analyze_question)  # 질문 분석 노드
    workflow.add_node("clarify", ask_clarification)  # 질문 명확화 노드
    workflow.add_node("search", web_search)  # 웹 검색 노드
    workflow.add_node("generate", generate_answer)  # 답변 생성 노드

    # 6. 시작점 설정: 워크플로우의 첫 단계 지정
    workflow.set_entry_point("analyze")

    # 7. 조건부 엣지 추가: 상태에 따라 다른 경로로 진행
    # analyze 노드 이후 should_clarify 함수의 반환값에 따라 다음 노드 결정
    workflow.add_conditional_edges(
        "analyze", should_clarify, {"clarify": "clarify", "search": "search"}
    )

    # 8. 일반 엣지 추가: 항상 같은 경로로 진행
    workflow.add_edge("clarify", "search")  # 명확화 후 항상 검색으로
    workflow.add_edge("search", "generate")  # 검색 후 항상 답변 생성으로

    # 9. 종료 조건 설정: generate 노드 이후 추가 검색 필요 여부에 따라 종료 또는 재검색
    workflow.add_conditional_edges(
        "generate", should_search_again, {"search": "search", "end": END}
    )

    # 10. 그래프 컴파일: 실행 가능한 형태로 변환
    return workflow.compile()

# 11. 질문 분석 노드: 사용자 질문을 분석하여 검색 쿼리 생성 또는 명확화 필요 여부 결정
def analyze_question(state: State) -> Dict:
    """사용자 질문을 분석하여 검색 쿼리를 생성하거나 명확화가 필요한지 결정"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    # 출력 파서 설정 - LLM 응답을 구조화된 형식으로 변환
    parser = PydanticOutputParser(pydantic_object=SearchQueryOutput)
    prompt = ChatPromptTemplate.from_template(
        """당신은 사용자 질문을 분석하여 웹 검색에 최적화된 쿼리를 생성하거나,
질문이 모호한 경우 명확화를 요청하는 AI 어시스턴트입니다.

사용자 질문: {question}

다음 형식으로 응답하세요:
{format_instructions}"""
    ).partial(format_instructions=parser.get_format_instructions())

    # 체인 구성: 프롬프트 -> LLM -> 파서
    chain = prompt | llm | parser

    # 체인 실행 및 결과 반환
    result = chain.invoke({"question": state["question"]})

    # 12. 상태 업데이트: 노드는 상태의 일부만 업데이트하고 반환
    # 반환된 딕셔너리는 기존 상태와 병합됨
    return {
        "search_query": result.search_query,
        "need_clarification": result.need_clarification,
        "clarification_question": (
            result.clarification_question if result.need_clarification else ""
        ),
    }

# 13. 질문 명확화 노드: 사용자에게 추가 정보 요청
def ask_clarification(state: State) -> Dict:
    """사용자에게 명확화 질문을 하고 응답을 받는 시뮬레이션"""
    # 실제 구현에서는 사용자와 상호작용하는 코드가 필요합니다
    # 여기서는 시뮬레이션을 위해 가상의 응답을 생성합니다
    print(f"명확화 질문: {state['clarification_question']}")

    # 시뮬레이션된 사용자 응답 (실제 구현에서는 사용자 입력을 받아야 함)
    simulated_response = "최신 인공지능 기술 동향에 대해 알고 싶습니다."
    print(f"사용자 응답: {simulated_response}")

    return {"user_clarification": simulated_response, "need_clarification": False}

# 14. 웹 검색 노드: 외부 도구를 사용한 정보 검색
def web_search(state: State) -> Dict:
    """웹 검색 수행"""
    search_tool = TavilySearchResults(k=5)

    # 명확화 응답이 있으면 그것을 포함하여 검색
    query = (
        state["user_clarification"]
        if state.get("user_clarification")
        else state["search_query"]
    )

    results = search_tool.invoke({"query": query})

    return {"search_results": results, "need_web_search": False}

# 15. 답변 생성 노드: 검색 결과를 바탕으로 최종 답변 생성
def generate_answer(state: State) -> Dict:
    """검색 결과를 바탕으로 답변 생성"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

    # 검색 결과 포맷팅
    formatted_results = "\n\n".join(
        [
            f"내용: {result['content']}\n출처: {result['url']}"
            for result in state["search_results"]
        ]
    )

    prompt = ChatPromptTemplate.from_template(
        """당신은 웹 검색 결과를 바탕으로 정확하고 유용한 답변을 제공하는 AI 어시스턴트입니다.

사용자 질문: {question}

검색 결과:
{search_results}

위 정보를 바탕으로 사용자 질문에 답변하세요. 정보가 부족하다면 추가 검색이 필요하다고 표시하세요.

다음 형식으로 응답하세요:
{format_instructions}"""
    )

    parser = PydanticOutputParser(pydantic_object=AnswerOutput)
    chain = (
        prompt.partial(format_instructions=parser.get_format_instructions())
        | llm
        | parser
    )
    result = chain.invoke(
        {"question": state["question"], "search_results": formatted_results}
    )
    sources = [result["url"] for result in state["search_results"]]

    return {
        "answer": result.answer,
        "reasoning": result.reasoning,
        "confidence": result.confidence,
        "need_web_search": result.need_web_search,
        "sources": result.sources if result.sources else sources[:3],
    }

# 16. 라우터 함수: 조건부 엣지에서 다음 노드를 결정하는 함수
def should_clarify(state: State) -> Literal["clarify", "search"]:
    """명확화가 필요한지 결정하는 라우터"""
    if state["need_clarification"]:
        return "clarify"  # 명확화 필요시 clarify 노드로
    else:
        return "search"   # 그렇지 않으면 search 노드로

# 17. 종료 결정 함수: 워크플로우 종료 또는 계속 여부 결정
def should_search_again(state: State) -> Literal["search", "end"]:
    """추가 검색이 필요한지 결정하는 라우터"""
    if state["need_web_search"]:
        return "search"  # 추가 검색 필요시 search 노드로 돌아감
    else:
        return "end"     # 그렇지 않으면 워크플로우 종료

우선 위의 LangGraph를 시각화하여 살펴보겠습니다. 아래의 코드를 작성하면 mermaid 코드를 통해 시각화하여 볼 수 있습니다.

app = create_graph()
png_data = app.get_graph().draw_mermaid()
print(png_data)

랭그래프 시각화

위 LangGraph 워크플로우는 다음과 같은 단계로 진행됨을 확인할 수 있습니다.

1. 질문 분석 (analyze): 사용자의 질문을 입력받아 분석합니다.
2. 명확화 필요 여부 결정:
   • 질문이 모호하면 → 명확화 요청 단계로 이동 (clarify)
   • 질문이 명확하면 → 웹 검색 단계로 이동 (search)
3. 명확화 요청 (clarify): 사용자에게 추가 정보를 요청합니다.
4. 웹 검색 (search): 검색 도구를 사용해 관련 정보를 수집합니다.
5. 답변 생성 (generate): 검색 결과를 바탕으로 답변을 작성합니다.
6. 추가 검색 필요 여부 결정:
   • 추가 정보가 필요하면 → 웹 검색 단계로 돌아감 (serach)
   • 충분한 정보가 있으면 → 워크플로우 종료 (end)

그럼 해당 코드를 실행해보겠습니다.

run_rag_system("양자 컴퓨팅이 머신러닝에 미치는 영향은?")

실행결과

run_rag_system("AI")

실행결과

마치며... 📢

이번 포스트에서는 LangChain의 고급 컴포넌트인 Agents, Tools, LangGraph에 대해 살펴보았습니다. 처음에는 다소 복잡하게 느껴질 수 있지만, 하나씩 차근차근 기능을 추가해 나가다 보면 나의 생산성을 높여주는 나만의 AI 비서를 만들 수 있으리라 믿습니다.

다음 포스트에서는 제가 LangChain과 LangGraph를 실제 활용하면서 알게 된 유용한 팁들을 공유해보겠습니다.

긴 글 읽어주셔서 감사합니다. 😊