LLM 개발 입문 (12) - 1

루나·2026년 3월 19일

LLM STUDY

목록 보기
31/31
post-thumbnail

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


Chapter 12

랭그래프로 목차를 작성하는 멀티에이전트 만들기

본 포스팅에서는 랭그래프로 목차를 작성하는 멀티에이전트를 만들어봅시다

1. 멀티에이전트

먼저 멀티에이전트에 대해 알아보고 사용자의 질문에 대답하는 챗봇을 만들어보자
작은 단위의 인공지능 프로그램들이 서로 협업해서 작업을 수핵하고 단순한 요구사항부터 복잡한 업무까지 어떻게 하는지 알아보자


이 사진은 책에서 사용하고 있는 멀티에이전트의 구조이다
'XX에 대한 보고서를 써야하니까 인터넷에서 자료를 조사하고 목차를 작성해줘' 라는 문장을 수행하기 위해서는
위와 같은 그림처럼 조직된 AI 멀티에이전트 시스템이 필요하다

각각의 노드(에이전트)는 고유한 임무를 수행하고 자신의 작업 결과를 다른 AI 에이전트와 공유하면서 작업을 완수한다

이번 챕터에서는 AI 에이전트들이 개별 역할을 수행하면서 복잡한 업무를 처리하는 과정을 직접 구현해보려고 한다
이를 통해 인공지능을 이용한 과제 진행의 기본 원리를 이해할 수 있다!!

2. 사용자와 의사소통하는 커뮤니케이터 에이전트

사용자와 단순한 대화를 할 수 있는 커뮤니케이터 에이전트 communicator를 만들어보자
이어지는 실습에서 여러가지 기능을 계속 추가할 예정이다
사용자의 메세지를 받으면 커뮤니케이터 에이전트가 답변을 생성하고 결과를 내놓는 가장 기본적이 ㄴ구조이다.

book_writer.py 파일을 새로 만들고 다음과 같이 코드를 작성하자
대부분은 이전에 작성한 코드를 그대로 가져온 것이다

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import AnyMessages, SystemMessage, HumanMessage
from langchain_core.prompts import PromptTemplate
from typing_extensions import TypedDict
from typing import List

from utils import save_state
from datetime import datetime
import os

# 현재 폴더 경로 찾기
# 랭그래피ㅡ 이미지로 저장 및 추후 작업 결과 파일 저장 경로로 활용
filename = os.path.basename(__file__)           # 현재 파일명 반환
absolute_path = os.path.basename(__file__)      # 현재 파일의 절대 경로 반환
current_path = os.path.dirname(absolute_path)   # 현재 .py 파일이 있는 폴더 경로

# 모델 초기화
llm = ChatOpenAI(model = "gpt-4o")
  • from utils import save_state
    save_state 함수를 가져올 utils 파일을 import 한다
    utils 파일은 나중에 작성할 예정!!

이제 상태를 정의해보자
실습을 진행하면서 더 많은 상태가 필요하지만 현재는 대화 기록을 담아 둘 수 있는 messages 라는 변수에 리스트 자료형을 사용한다
이 리스트에 들어갈 수 있는 자료형은 AnyMessages 또는 문자열(str)이다
꼭 랭체인 메세지가 아니라 문자열로 들어와도 GPT 에서 처리할 수 있으므로 str로도 담을 수 있게 처리해야 한다

# 상태 정의
class State(TypedDict):
  messages : List[AnyMessages | str]
  • AnyMessage란?
    랭체인에서 여러 종류의 메세지 타입을 하나로 통합해서 표현하기 위해 사용하는 타입 별칭이다
    즉, AnyMessage는 여러가지 메세지 클래스를 포함하는 유니온타입으로 정의된다
    AnyMessage를 사용하면 다양한 메세지 타입을 모두 받아들일 수 있어서 코드의 유연성과 가독성을 높일 수 있다
    예를 들어, 채팅 기록이나 메세지 리스트를 처리할 때 AnyMessage 타입으로 선언하면 앞에 언급한 모든 메세지 객체를 하나의 리스트로 다룰 수 있다
타입설명
AIMessageAI(모델)가 생성한 응답 메세지
HumanMessage사용자(인간)가 입력한 메세지
SystemMessage시스템에서 모델의 행동이나 대화의 맥락을 지정하는 메세지
ToolMessage도구(tool)의 호출 결과를 나타내는 메세지

이제 사용자와 대화하는 에이전트 communicator를 만들어보자
이 AI 에이전트는 목차를 작성하는 AI팀의 일원으로 기존 대화 내용을 바탕으로 사용자와 상호작용하며 대화를 진행한다

# 사용자와 대화할 노드(agent) : communicator
def communicator(state: State):
  print('\n\n==============COMMUNICATOR==============')

  coummunicator_system_prompt = PromptTemplate.from_template(
    """
    너는 책을 쓰는 AI 팀의 커뮤니케이터로서, 
    AI 팀의 진행 상황을 사용자에게 보고하고, 사용자의 의견을 파악하기 위해 대화를 나눈다.

    messages : {messages}
    """
  )

  system_chain = coummunicator_system_prompt | llm

  # 상태에서 메세지 가져오기
  messages = state["messages"]

  # 입력값 정의
  inputs = {"messages" : messages}

  gathered = None
  print('\nAI\t:', end = '')
  for chunk in system_chain.stream(inputs):
    print(chunk.content, end = "")

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

  messages.append(gathered)

  return {"messages" : messages}
  • coummunicator_system_prompt = PromptTemplate.from_templates()
    communicator 노드의 프롬프트 내용을 PromptTemplate을 이용해서 정의한다
  • for chunk in system_chain.stream(inputs):
    gathered로 messages에 append 하고 출력할 때는 stream 방식으로 출력하는 코드이다
    이전 포스팅에서 꽤나 자주 사용한 코드 중 하나이다

이제 StateGrap로 그래프를 만들어보자
이 그래프는 Start -> communicator -> END 의 단순한 구조로 되어있다

방금 만든 communicator 노드만 graph_builder에 추가하고, 이 노드를 START와 END에 연결한 뒤 그래프를 컴파일한다

# 상태 그래프 정의
graph_builder = StateGraph(State)

# Nodes
graph_builder.add_node("communicator", communicator)

# Edges
graph_builder.add_edge("START", "communicator")
graph_builder.add_edge("communicator", "END")
graph = graph_builder.compile()

그리고 랭그래프가 올바르게 구성되었는지 확인하기 위해서 그림으로 구조를 확인해보자

# 그래프 도식화
graph.get_graph().draw_mermaid_png(output_file_path = absolute_path.replace('.py', '.png'))

아직은 utils 파일을 생성하지 않아서 코드를 실행할 수 없지만 코드를 실행하면 그래프의 구조를 시각적으로 확인할 수 있다

이제 워크플로의 시스템 메세지를 작성해보자
아직은 AI 에이전트가 communicator 하나지만 나중에 여러 AI 에이전트가 추가될 것을 고려해서 작성해보자
그리고 프롬프트에 현재 시각 정보를 포함하도록 하여 나중에 책이나 보고서를 만들 때 언어 모델이 만들어진 시점을 기준으로 판단하는 오류를 방지하도록 한다

# 상태 초기화
state = State(
  messages = [
    SystemMessage(
      f"""
      너희 AI 들은 사용자의 요구에 맞는 책을 쓰는 작가팀이다.
      사용자가 사용하는 언어로 대화하라

      현재 시각은 {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
      """
    )
  ]
)

아래 코드는 터미널 창에서 사용자의 입력을 받고 graph를 실행하는 코드이다
사용자가 입력한 값을 user_input 변수에 담아 워크플로를 실행한다

실제로 AI가 생성한 답변은 communicator 에이전트에서 print로 출력되며 그 후 메세지 수를 파악하기 위해 print 문을 추가한다
그리고 현재 상태를 save_state 함수로 저장합니다 이 함수는 utils.py에 저장한다

# 터미널 창에서 사용자의 입력을 받고 graph 실행하는 부분
while True:
  user_input = input("\nUser\t: ").strip()

  if user_input.lower() in ['exit', 'quit', 'q']:
    print("Goodbye!")
    break

  state["messages"].append(HumanMessage(user_input))
  state = graph.invoke(state)

  print('\n------------------------------MESSAGE COUNT\t', len(state["messages"]))

  save_state(current_path, state)       # 현재 state 내용 저장

3. utils 파일에 save_state 함수 만들기

이제 save_state 함수를 만들어보자. 파일 유지보수를 위해 utils.py에 따로 만들어보자
current_path와 state를 매개변수로 받아 current_path/data 폴더에 state를 JSON 파일로 저장한다
현재 statedp은 사용자와 AI 간의 대화 내용이 담긴 messages 만 존재하므로
[(m.__class__.__name__, m.content)] 형태로 자료형과 대화 내용을 튜플로 변환해서 저장한다

이때 ensure_ascii = False와 endocing = 'utf-8'을 지정해서 한글이 깨지지 않도록 한다

import os
import json

def save_state(current_path, state):
  if not os.path.exists(f"{current_path}/data"):
    os.makedirs(f"{current_path}/data")

  state_dict = {}

  messages = [(m.__class__.__name__, m.content) for m in state["messages"]]
  state_dict["messages"] = messages

  with open(f"{current_path}/data/state.json", "w", encoding = 'utf-8') as f:
    json.dump(state_dict, f, indent = 4, ensure_ascii = False)

이 코드를 실행하면 다음처럼 터미널 창에서 대화할 수 있다
아직은 communicator 노드만 있기 때문에 단순한 대화만 가능하다

이렇게 실제로 행동은 하지 않지만 질문에 대한 답변은 생성해준다

4. 책의 목차를 작성하는 콘텐츠 전략가 에이전트

보고서나 책을 쓰기 위한 방법은 많지만 목차를 먼저 만들고 파트별로 내용을 구성하는 방식을 사용해보자
이번에는 목차 작성을 전문으로 하는 콘텐츠 전략가 에이전트 content_strategist를 만들어보자

하나의 에이전트에게 하나의 임무만 맡기면 엉뚱하게 답변할 확률을 줄일 수 있다
따라서 content_strategist가 우선 질문 내용을 받고 생성 결과를 communicator로 전달받는 구조를 만들어보자

# AIMessage 추가
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage
from langchain_core.output_parsers.string import StrOutputParser
from utils import save_state, get_outline, save_outline

def content_strategist(state: State):
    print("\n\n============ CONTENT STRATEGIST ============")

    # 시스템 프롬프트 정의
    content_strategist_system_prompt = PromptTemplate.from_template(
        """
        너는 책을 쓰는 AI팀의 콘텐츠 전략가(Content Strategist)로서,
        이전 대화 내용을 바탕으로 사용자의 요구사항을 분석하고, AI팀이 쓸 책의 세부 목차를 결정한다.

        지난 목차가 있다면 그 버전을 사용자의 요구에 맞게 수정하고, 없다면 새로운 목차를 제안한다.

        --------------------------------
        - 지난 목차: {outline}
        --------------------------------
        - 이전 대화 내용: {messages}
        """
    )

    # 시스템 프롬프트와 모델을 연결
    content_strategist_chain = content_strategist_system_prompt | llm | StrOutputParser()

    messages = state["messages"]        # 상태에서 메시지를 가져옴
    outline = get_outline(current_path) # 저장된 목차를 가져옴

    # 입력값 정의
    inputs = {
        "messages": messages,
        "outline": outline
    }

    # 목차 작성
    gathered = ''
    for chunk in content_strategist_chain.stream(inputs):
        gathered += chunk
        print(chunk, end='')

    print()

    save_outline(current_path, gathered) # 목차 저장

    # 메시지 추가    
    content_strategist_message = f"[Content Strategist] 목차 작성 완료"
    print(content_strategist_message)
    messages.append(AIMessage(content_strategist_message))

    return {"messages": messages} # 메시지 업데이트
    
    (생략)
# Nodes
graph_builder.add_node("communicator", communicator)
graph_builder.add_node("content_strategist", content_strategist)

# Edges
graph_builder.add_edge(START, "content_strategist")
graph_builder.add_edge("content_strategist", "communicator")
graph_builder.add_edge("communicator", END)
graph = graph_builder.compile()

그리고 이제 get_outling과 save_outline 함수를 utils.py에 추가하자

def get_outline(current_path):
  outline = '아직 작성된 목차가 업습니다.'

  if os.path.exists(f"{current_path}/data/outline.md"):
    with open(f"{current_path}/data/outline.md", "r", encoding= 'utf-8') as f:
      outline = f.read()
  return outline

def save_outline(current_path, outline):
  if not os.path.exists(f"{current_path}/data"):
    os.makedirs("f{current_path}/data")

  with open(f"{current_path}/data/outline.md", "w", encoding = 'utf-8') as f:
    f.write(outline)
  return outline

이제 대화를 통해 목차를 작성할 수 있다
코드를 실행해서 이전 질문과 똑같이 'JYP와 HYBE의 경영 전략과 기업 문화를 비교하는 책을 쓰고 싶어' 라고 해보자

우리가 의도한대로 목차를 작성해주는 것을 볼 수 있다

5. 마무리

이렇게 해서 다양한 AI 에이전트를 사용하는 실습 첫 단계를 진행했다
AI 별로 프롬프트를 직접 작성해야 한다는 점에서 생각보다 쉽지는 않은 것 같다
또한 utils.py 에 필요한 함수를 지정하는 것도 간단한 일은 아닌 것 같다

profile
Per ardua ad astra

0개의 댓글