LLM 개발 입문 (10) - 3

루나·2026년 3월 8일
AIAI에이전트GPTLLM랭그래프

LLM STUDY

목록 보기
28/31
post-thumbnail

본 포스팅은 "Do it! LLM을 활용한 AI 에이전트 개발 입문"을 독학하며 쓴 글입니다.
내돈내산 포스팅임을 참고해주시면 감사하겠습니다.
2026년 3월 7일 기준으로 작성되었습니다.

Chapter 10

랭그래프와 친해지기

본 포스팅에서 인터넷 검색 후 기사를 작성하는 챗봇을 만들어보려고 합니다

1. 신문기사 챗봇 만들기

6장에서는 오픈AI의 GPT API를 활용할 때 펑션 콜링을 사용하고
7장에서는 랭체인에서 도구를 호출하는 방법에 대해서 배웠다

랭그래프는 랭체인에 기반하느로 GPT와 같은 언어 모델이 미리 정의한 도구를 사용할 수 있게 구현할 수 있다
따라서 이번 포스팅에서는 사용자가 주제를 제시하면 인터넷을 검색하여 최신 이슈를 기반으로 세부 주제를 선정하고
그 주제를 검색해서 최종 기사를 작성하는 신문기자 챗봇을 만들어보려고한다

먼저 기사를 어떻게 작성할지 일의 순서를 설명해보자
여기서 {about}에 주제만 입력하면 챗봇이 알아서 기사를 쓰도록 만들어본다
나중엔 이 문구를 프롬프트로 지정해서 언어 모델에 적용하면 맞춰서 기사를 작성해줄 것이다

너는 신문기자이다
최근 {about}에 대해 비판하는 심층 분석 기사를 쓰려고 한다

- 최근 어떤 이슈가 있는지 검색하고 사람들이 제일 관심있어 할만한 주제를 선정하고 왜 선정했는지 말해줘
- 그 내용으로 원고를 작성하기 위한 목차를 만들고 목차 내용을 채우기 위해 추가로 검색할 내용을 리스트로 정리해
- 검색할 리스트를 토대로 재검색해
- 목차에 있는 내용을 작성하기 위해 더 검색이 필요한 정보가 있는지 확인하고 있다면 추가로 검색해
- 검색된 결과에서 원하는 정보를 찾지 못했다면 다른 검색어로 재검색해도 좋아

더 이상 검색할 내용이 없다면 조선일보 신문 기사 형식으로 최종 기사를 작성한다
제목, 부제, 리드문, 본문의 구성으로 작성한다
본문 내용은 심층 분석 기사에 맞게 구체적이고 깊이 있게 작성해야 한다

그리고 현재 시간을 알려주는 함수와 웹 검색을 하는 함수를 랭체인의 도구로 등록하고 랭그래프로 구현해보자
이를 위해 langgraph_tools.ipynb 파일을 생성하고 언어모델로 GPT를 설정한다

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
model.invoke("안녕하세요!")

이제 랭그래프의 상태를 선언하고 그 안에 필요한 정보들을 담는 코드를 작성한다
이 코드는 이전 포스팅에서 사용한 코드와 동일하다

from typing import Annotated    # annotated는 타입 힌트를 사용할 때 사용하는 함수
from typing_extensions import TypedDict   # TypedDict는 딕셔너리 타입을 정의할 때 사용하는 클래스

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
  """
  State 클래스는 TypedDict를 상속받습니다

  속성: 
    messages (Annotated[list[str], add_messages]) : 메세지들은 "list" 타입을 갖비니다
    'add_messages' 함수는 이 상태 키가 어떻게 업데이트 되어야 하는지를 정의합니다
    (이 경우, 메세지를 덮어쓰는 대신 리스트에 추가합니다)
  """

  messages: Annotated[list[str], add_messages]

# StateGraph 클래스를 사용하여 State 타입의 그래프 생성
graph_builder = StateGraph(State)

랭그래프에서도 랭체인과 마찬가지로 도구를 정의하고 활용할 수 있다
예전에 랭체인 도구를 다룰 때 사용한 시간을 알려주는 get_current_time 함수와
웹 검색을 해주는 get_web_search 함수를 활용해 코드를 작성해보자

streamlit_with_web_search.py 파일 코드를 참고해서 작성하였다

import pytz
from langchain_core.tools import tool
from datetime import datetime

# 웹 검색을 위한 라이브러리
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 도구 함수 정의
@tool
def get_current_time(timezone : str, location : str ) -> str:
  """현재 시각을 반환하는 함수"""
  try:
    tz = pytz.timezone(timezone)
    now = datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")

    result = f'{timezone} ({location}) 현재 시각 {now}'
    print(result)
    return result
  except pytz.UnknownTimeZoneError:
     return f"알수 없는 타임 존 : {timezone}"

@tool
def get_web_search(query : str, search_period: str ='m') -> str:
    """
    웹 검색을 수행하는 함수

    Args:
        query (str) : 검색어
        search_period (str) : 검색 기간 (e.g., "w" for past week, "m" for past month, "y" for past year, "d" for past day )

    Returns:
        str : 검색 결과
    """

    wrapper = DuckDuckGoSearchAPIWrapper(
      #  region = "kr-kr", 
      time= search_period
    )

    print('-----------------WEB SEARCH-----------------')
    print(query)
    print(search_period)

    search = DuckDuckGoSearchResults(
       api_wrapper = wrapper,
       # source = "news",
       results_separator = ';\n'
    )

    searched = search.invoke(query)

    for i, result in enumerate(searched.split(';\n')):
      print(f'{i+1}. {result}')
      
    return searched


# 도구 바인딩
tools = [get_current_time, get_web_search]

tools에 바인딩한 도구들이 랭체인과 랭그래프 방식으로 잘 작동하는지 확인해보자
tools[0]은 시간을 알려주는 get_current_time 함수를 의미한다
여기에 필요한 매개변수는 딕셔너리로 입력해보자

tools[0].invoke({"timezone" : "Asia/Seoul" , "location" : "서울"})

tools[1]은 웹 검색을 하는 get_web_search 함수를 의미한다
'파이썬'을 질의어로 하고 검색기간을 의미하는 search_periond를 'm'으로 설정해서 최근 한 달 간의 문서를 검색해보자

tools[1].invoke({"query" : "파이썬" , "search_perios" : "m"})

도구들이 잘 등록되었다면 각각의 셀을 실행했을 때,
시간을 정상적으로 알려주고
파이썬으로 검색한 결과를 보여줄 것이다

이제 tools가 어떻게 저장되었는지 확인해보기 위해 for문을 이용해 하나씩 출력해보자

for tool in tools:
  print(tool.name, tool)

사진처럼 도구 이름과 도구 설명 내용이 함께 정리되어 있다

이제 챗봇이 응답메세지를 생성하는 generate 노드를 만들어보자
그리고 model로 선언한 언어 모델에 .bind_tools로 붙인 model_with_tools로 도구를 사용한다
이렇게 만든 generate 함수를 graph_builder에 "generate"라는 이름의 노드로 붙이자

model_with_tools = model.bind_tools(tools) # GPT 언어 모델에 도구 연결

def generate(state: State):
  """
  주어진 상태를 기반으로 챗봇의 응답 메세지를 생성한다

  매개변수 : 
  state (State) : 현재 대화 상태를 나타내는 객체로, 이전 메세지들이 포함되어 있다

  반환값:
  dict :  모델이 생성한 응답 메세지를 포함하는 딕셔너리
          형식은 {"messages" : [응답메세지]} 입니다.
  """

  return {"messages" : [model_with_tools.invoke(state["messages"])]}

graph_builder.add_node("generate", generate)

랭그래프에서 도구를 편하게 관리하기 위해 BasicToolNode 클래스를 만들어보려고 한다
지금 만들고 있는 랭그래프의 generate 노드에서 챗봇의 답변을 생성할 때
get_web_search나 get_current_time 같은 도구를 사용해야 한다면 따로 독립된 노드에서 실행하게 된다

BasicToolNode 클래스는 AIMessage에서 도구 요청이 있을 때 이를 실행시키는 역할을 한다
(해당 코드는 랭그래프 공식 문서를 참고해서 작성했습니다)

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
  """
  도구를 실핸하는 노드 클래스입니다. 마지막 AIMessage에서 요청된 도구를 실행합니다

  Attributes:
    tool_by_name (dict) : 도구 이름을 키로 하고 도구 객체를 값으로 가지는 사전

  Methods:
    __init__(tools : list) : 도구 객체들의 리스트를 받아서 초기화합니다
    __call__(inputs : dict) : 입력 메세지를 받아서 도구를 실행하고 결과 메세지를 반환한다
  """
  """A node that rutns the tools requested in the last AIMessage"""

  def __init__(self, tools:list) -> None:
    self.tools_by_name = {tool.name: tool for tool in tools}

  def __call__(self, inputs:dict):
    if messages := inputs.get("messages", []):
      # inputs에 messages가 있으면 messages를 가져오고 없으면 빈 리스트 가져오기
      message = messages[-1]

    else:
      raise ValueError("No message fount in input")
    
    outputs = []
    for tool_call in message.tool_calls:
      tool_result = self.tools_by_name[tool_call["name"]].invoke(
        tool_call["args"]
      )

      outputs.append(
        ToolMessage(
          content = json.dumps(tool_result),
          name = tool_call["name"],
          tool_call_id = tool_call["id"]
        )
      )

      return {"messages" : messages + outputs}
    
tool_node = BasicToolNode(tools = tools)
graph_builder.add_node("tools", tool_node)
  • def __init__(self, tools:list) -> None:
    클래스의 초기화 메소드(__init__) 에 있는 tools_by_name은 딕셔너리 형태로 도구의 이름과 해당 도구 자체를 저장한다
  • def __call__(self, inputs:dict):
    __call__ 메소드는 입력 메세지인 input을 딕셔너리 형태로 받는다
    이때 input은 랭그래프에서 상태를 관리하는 state가 딕셔너리로 전달된다
    state에서는 messages가 포함되어있다
  • for tool_call in message.tool_calls:
    message에 tool_calls가 포함되어 있으면 도구를 사용해야 하므로 반복문을 이용해 도구를 실행시킨다

2. 라우터 설정하기

이전 포스팅에서 만들었던 랭그래프의 흐름은 갈림길 없이 진행되었다
이번 실습에서는 흐름이 조금 더 다양해진다
예를 들어 사용자가 질문하면 generate 노드에서 답변을 생성하고 그 후 바로 END 노드로 가서 작업을 끝낼 수도 있다

하지만 만약 인터넷 검색이나 시간 확인이 필요하다고 판단되면 tools 노드에서 그 기능을 수행한 수 결과를 바탕으로
다시 generate 노드에 돌아와 답변을 생성할 수 있다
그리고 검색이나 시간 확인이 더 필요하다고 판단된다면 tools 노드에서 작업을 반복할 수도 있다

이번 실습에서는 AI 에이전트가 상황에 맞게 다음에 해야 할 일을 알아서 결정하도록 만들어보자
AI 에이전트가 랭그래프 내에서 스스로 다음 경로를 선택해야 할 때 라우터(Router)를 활용한다
상황에 따라 방향을 결정하는 것을 라우팅(Routing)이라고 하고
이때 조건에 따라 활성화되고 비활성화 되는 조건부 엣지를 사용한다

START와 generate 까지는 이전과 같이 순차로 진행하고 generate 노드에서 언어 모델이 판단한 결과에 따라 경로가 달라지도록 설정한다
이를 위해 route_tools 함수를 만들고 state를 받아 messages 리스트에서 마지막 메세지를 확인한다
만약 마지막 메세지에 tool_calls가 포함되어 있으면 tools 노드로 이동하여 필요한 도구를 실행하고
없다면 END 노드로 이동해서 작업을 종료한다
tools 노드에서 작업을 마치면 다시 generate로 돌아온다

def route_tools(state: State):
  """
  마지막 메세지에 도구 호출이 있는 경우 ToolNode로 라우팅하고
  그렇지 않은 경우 끝으로 라우팅하기 위해 contidional_edge에서 사용한다
  """

  if isinstance(state, list):
    ai_message = state[-1]
  elif messages := state.get("messages", []):
    ai_message = messages[-1]
  else:
    raise ValueError(f"tool_edge 입력 상태에서 메세지를 찾을 수 없습니다 : {state}")
  
  if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
    return "tools"

  return END

graph_builder.add_edge(START, "generate")

graph_builder.add_conditional_edges(
  "generate",
  route_tools,
  {"tools" : "tools", END : END}
)

# 도구가 호출된 때마다 다음 단계를 결정하기 위해 챗봇으로 돌아감
graph_builder.add_edge("tools", "generate")
graph = graph_builder.compile()

지금까지 만든 랭그래프를 그래프로 그려서 확인해보자
저번 포스팅에서 사용한 코드와 동일하다

from IPython.display import Image, display

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

3. 도구 테스트하고 기사 작성하기

챗봇이 여러 도구 중에서 필요한 도구를 잘 찾아서 실행시키는지 테스트해보자
이어서 웹 검색 후 기사 작성 기능을 구현해보자

일단 messages에 HumanMessage로 '지금 서울 몇시야?' 라는 질문을 넣어 graph.stream으로 실행해보자
스트림 출력을 위해 graph.invoke가 아닌 graph.stream으로 설정하고 stream_mode는 'messages'로 설정한다
결과가 AiMesageChunk로 나오자마자 for 문을 통해 즉각 출력된다

AIMessageChunk는 조각난 상태로 조금씩 넘어오므로 gathered라는 변수에 조각들을 계속 붙이고
이렇게 완성한 최종 결과 gathered를 출력한다

from langchain_core.messages import AIMessageChunk, HumanMessage

inputs = [HumanMessage(content = "지금 서울 몇 시야?")]

gathered = None

for msg, metadata in graph.stream({"messages" : inputs}, stream_mode = "mesages"):
  if isinstance(msg, AIMessageChunk):
    print(msg.content, end = '')

    if gathered is None:
      gathered = msg
    else:
      gathered = gathered + msg
      
gathered

정상적으로 도구를 사용해서 현재 시간을 불러왔다

앞에서는 inputs에 HumanMessages를 사용했지만 이번에는 언어 모델에 역할을 지시하기 위해 SystemMessage를 사용한다
그리고 f-string에 기사 주제를 사용해서 {about}을 받아오는 형태로 프롬프트를 입력한다
여기에서는 최근 발생한 '미국,이란 - 이스라엘 전쟁' 이라는 주제를 설정했다

from langchain_core.messages import AIMessageChunk, SystemMessage

about = "미국,이란 - 이스라엘 전쟁"

inputs = [SystemMessage(content = f"""
          너는 신문기자이다
          최근 {about}에 대해 비판하는 심층 분석 기사를 쓰려고 한다

          - 최근 어떤 이슈가 있는지 검색하고 사람들이 제일 관심있어 할만한 주제를 선정하고 왜 선정했는지 말해줘
          - 그 내용으로 원고를 작성하기 위한 목차를 만들고 목차 내용을 채우기 위해 추가로 검색할 내용을 리스트로 정리해
          - 검색할 리스트를 토대로 재검색해
          - 목차에 있는 내용을 작성하기 위해 더 검색이 필요한 정보가 있는지 확인하고 있다면 추가로 검색해
          - 검색된 결과에서 원하는 정보를 찾지 못했다면 다른 검색어로 재검색해도 좋아

          더 이상 검색할 내용이 없다면 조선일보 신문 기사 형식으로 최종 기사를 작성한다
          제목, 부제, 리드문, 본문의 구성으로 작성한다
          본문 내용은 심층 분석 기사에 맞게 구체적이고 깊이 있게 작성해야 한다
        """)]


for msg, metadata in graph.stream({"messages" : inputs}, stream_mode = "messages"):
  if isinstance(msg, AIMessageChunk):
    print(msg.content, end='')

실행 결과는 다음과 같다
웹 검색으로 주제에 대한 최근 소식을 검색해서 기사의 방향을 결정하고 목차를 생성한 후, 관련 내용을 다시 검색한다
그리고 수집한 내용을 기반으로 기사를 작성했다

-----------------WEB SEARCH-----------------
질문 : 2026 미국 이란 이스라엘 전쟁 최근 이슈
검색 기간 : m
(생략)
4. **이스라엘-팔레스타인 갈등 원인**
   - 이란 전쟁과 이스라엘의 군사적 행동은 이스라엘-팔레스타인 간의 갈등에도 영향을 미치며, 이에 대한 공적 여론이 더욱 격화되고 있습니다. 해당 갈등의 주요 원인은 역사적 배경과 지리적 요인이 복합적으로 작용하는 것입니다.
   - [출처](https://en.wikipedia.org/wiki/Israeli%E2%80%93Palestinian_conflict)

### 최종 기사 작성

#### 제목: 중동의 불길, 2026 미국-이스라엘-이란 전쟁의 심층 분석

#### 부제: 미국과 이스라엘의 군사적 개입, 이란의 저항 그리고 국제 사회의 반응이 뒤얽힌 복잡한 양상

#### 리드문: 2026년 2월 28일, 이스라엘과 미국의 합동 공군 공격이 이란에서 벌어진 후 전 세계가 새로운 불안에 휘말리게 되었다. 이란의 반격과 이어지는 국제 사회의 다양한 반응은 이 전쟁이 가져올 여파를 더욱 심각하게 만들고 있다. 본 기사는 이 전쟁의 배경, 주요 원인 그리고 그 사회적 영향을 분석하고자 한다.

#### 본문

1. **전쟁의 배경과 현재 상황**
   - 이란의 핵 개발과 이란이 지원하는 무장 단체는 이스라엘의 보안 우려를 증대시켰으며, 이러한 우려는 미국의 군사 개입으로 이어졌다. 전쟁 발발 시점에서 이란은 미국 측의 군사적 공격에 대해 강력한 반격을 예고하였고, 이는 중동의 긴장을 더욱 심화시켰다.

2. **갈등의 주요 원인**
   - 이란의 핵 프로그램은 갈등의 중심에 있으며, 이란은 국가 안보의 관점에서 스스로를 방어하고자 하였다. 이란과 이스라엘의 대립은 단순한 군사적 충돌을 넘어선 지리적, 정치적 대결로 발전하고 있다.

3. **미국의 개입**
   - 미국은 이란에 대한 공군 공격을 지원하며 이스라엘을 군사적으로 지원하고 있다. 이는 제재와 외교적 압박을 넘어선 군사적 해법을 선택한 것으로, 보다 복잡한 국제 관계를 형성하고 있다.

4. **국제 사회의 반응**
   - 러시아와 중국은 미국-이스라엘의 공격을 비난하며 이란에 지원할 의향이 없음을 선언했다. 이러한 입장은 결국 중동 지역의 긴장이 더욱 고조되는 원인이 되고 있다.

5. **사회적 영향**
   - 전투로 인해 1,332명이 사망하고, 수천 명이 부상을 입었으며, 인도적 위기가 발생하고 있다. 이란의 민간인과 이스라엘의 민간인 모두 고통을 받고 있다.

6. **경제적 영향**
   - 유가 급등과 주식 시장의 불안정성이 나타나고 있으며, 이러한 경제적 손실은 전 세계적으로 확산되고 있다.

### 결론
중동 지역에서의 갈등은 단순한 군사적 대결로 끝나지 않을 것이며, 그 여파는 국제 사회의 정치적, 경제적 지형에 큰 변화를 가져올 것으로 예상된다. 전 세계는 이란-이스라엘 전쟁이 어떻게 전개될 것인지 주목해야 한다.

4. 마무리

이렇게 구조가 단순하지만 랭그래프를 이용해 기사를 작성하는 챗봇을 만들어보았다
이렇게 하니까 기본적인 랭체인으로 만들었을 때보다 조금 더 간단하게 만들 수 있었던것 같다
또한, 이렇게 노드와 엣지를 만들어서 관리하니까 추후에 유지보수에도 쉬울 것이라 생각된다

다음 챕터에서는 랭그래프를 활용한 멀티에이전트 RAG를 만들어보려고 한다

profile
루나
Per ardua ad astra
이전 포스트

LLM 개발 입문 (10) - 2

다음 포스트

LLM 개발 입문 (11) - 1

0개의 댓글