들어가며

지난 포스트에서는 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를 실제 활용하면서 알게 된 유용한 팁들을 공유해보겠습니다.

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