LangGraph 5 - 멀티 에이전트 협업

1. 이번 단계에서 달라지는 것

지금까지는 에이전트 하나가 도구·기억·RAG를 갖춰왔다. 이번에는 역할이 다른 두 에이전트가 협업하며 결과물을 스스로 고쳐 나가는 구조를 만들었음.

1~4편의 에이전트는 "혼자 일하는 한 명"이었다. 이번에는 **작성자(Coder)**와 **검수자(Reviewer)**로 역할을 나누고, 검수를 통과할 때까지 작성→리뷰→수정을 반복하는 자기수정(self-refine) 루프를 구성하였음. 예제 주제는 "코드 작성과 코드 리뷰"

• Coder 에이전트 — 요청받은 기능을 구현하거나, 리뷰 피드백을 반영해 코드를 수정
• Reviewer 에이전트 — 보안·효율·스타일을 점검해 PASS / FAIL을 판정하고, FAIL이면 수정 지시를 남김.
• 자기수정 순환 — Reviewer가 PASS를 줄 때까지 Coder가 반복 수정함. 단, 무한 루프와 비용 폭탄을 막기 위해 반복 횟수 상한을 두었음.

2~4편이 "한 에이전트의 진화"였다면, 이번엔 "여러 에이전트의 분업"이다. 같은 일을 먼저 LangChain 선형 체인으로 짜본 뒤, 같은 흐름을 LangGraph 순환 그래프로 다시 짜며 둘의 차이를 체감해본다.

---

2. 먼저 LangChain으로 — 선형 협업

세 개의 체인(작성→리뷰→수정)을 직렬로 이어 한 번씩 흘려보내는 가장 단순한 협업 형태부터 만들었음.

2.1 역할별 프롬프트

developer_prompt = ChatPromptTemplate.from_messages([
    ('system', '당신은 열정적인 "신입 파이썬 개발자"입니다. 요청받은 기능을 구현하는 코드를 작성하세요. 설명은 최소화하고 코드 위주로 작성하세요.'),
    ('user'  , '{request}'),
])

reviewer_prompt = ChatPromptTemplate.from_messages([
    ('system', '당신은 까다로운 "전문 개발자"입니다. 신입 개발자가 작성한 코드를 리뷰하세요.\n'
               '보안 취약점, 비효율적인 부분, 스타일 가이드를 점검하고 수정 제안을 하세요.\n'
               '코드가 완벽하다면 "PASS"라고만 답하세요.'),
    ('user'  , '다음 코드를 리뷰해주세요:\n\n{code}'),
])

refiner_prompt = ChatPromptTemplate.from_messages([
    ('system', '당신은 열정적인 "신입 파이썬 개발자"입니다. 전문개발자의 리뷰를 보고 코드를 수정하여 다시 제출하세요.'),
    ('user'  , '이전 코드:\n{original_code}\n\n리뷰 내용:\n{feedback}\n\n위 내용을 반영하여 개선된 전체 코드를 다시 작성하세요'),
])

페르소나(system)만 다르고, ④편까지 써온 ChatPromptTemplate 구조 그대로임. 에이전트의 "역할"은 결국 system 프롬프트로 갈린다.

2.2 체인 연결과 실행

developer_agent = developer_prompt | llm | StrOutputParser()
reviewer_agent  = reviewer_prompt  | llm | StrOutputParser()
refiner_agent   = refiner_prompt   | llm | StrOutputParser()

def run_agent_collaboration(topic):
    draft_code = developer_agent.invoke({"request": topic})      # 1. 초안 작성
    feedback   = reviewer_agent.invoke({"code": draft_code})     # 2. 리뷰

    if feedback == 'PASS':
        print(draft_code)
    else:
        final_code = refiner_agent.invoke({                      # 3. 1회 수정
            "original_code": draft_code,
            "feedback": feedback
        })
        print(final_code)

②편 109글의 `prompt | llm | StrOutputParser()` 패턴을 세 번 쓴 것뿐임. 동작은 하지만, 구조적 한계가 분명함.

• 반복이 안 됨 — 수정은 코드에 박힌 1회뿐. "PASS 나올 때까지" 같은 순환을 표현하려면 while을 손으로 짜야 함.
• 상태가 흩어짐 — draft_code, feedback을 변수로 일일이 들고 다님. 대화 맥락이 누적되지 않음.
• 분기가 어색함 — if feedback == 'PASS'처럼 흐름 제어를 비즈니스 코드에 섞어야 함.

선형 체인은 "한 방향으로 한 번"에 최적화돼 있다. 작성→리뷰→체크→재작성→리뷰…처럼 도는 흐름은 체인으로는 손이 많이 간다. 이 지점이 정확히 LangGraph가 필요한 이유다.

---

3. LangGraph로 재구성 — 상태 정의

흩어진 변수 대신, 누적되는 대화와 반복 횟수를 하나의 공유 상태로 묶음.

import operator
from typing import Annotated, List, TypedDict
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    # operator.add -> 덮어쓰지 말고 기존 메시지에 누적하라
    messages   : Annotated[List[BaseMessage], operator.add]
    iterations : int   # 수정 반복 횟수 (비용 상한 비교용)

①편 110글은 상태를 그냥 덮어썼지만, 여기선 Annotated[..., operator.add]로 "누적"을 명시함. ③편 112글의 단기기억(체크포인터)이 외부에서 상태를 보관했다면, 여기선 상태 정의 자체에 누적 규칙을 박아 넣은 셈임.

• messages — 작성·리뷰가 오갈수록 대화가 쌓임. 리듀서 operator.add가 리스트를 이어붙임.
• iterations — Coder가 한 번 돌 때마다 +1. 이후 조건부 엣지에서 상한 비교에 씀.

---

4. 노드 — 두 에이전트를 함수로

2편에서 만든 세 체인 중 작성/리뷰를 각각 노드 함수로 옮김. 수정(refine)은 별도 노드가 아니라 "리뷰를 본 Coder의 재실행"으로 흡수됨.

4.1 coder_node — 작성과 수정 겸용

def coder_node(state: AgentState):
    msg = state['messages']
    coder_prompt = ChatPromptTemplate.from_messages([
        ('system',      '당신은 "초보 파이썬 개발자"입니다. 요청받은 기능을 구현하세요. '
                        '리뷰어의 피드백이 있다면 반영하여 코드를 수정하시오'),
        ('placeholder', '{messages}'),   # 메시지 종류(Human/AI)를 알아서 채움
    ])
    chain = coder_prompt | llm
    draft_code = chain.invoke({"messages": msg})
    return {
        "messages"   : [draft_code],
        "iterations" : state.get('iterations', 0) + 1
    }

포인트는 `placeholder`임. 초안 작성이든 피드백 반영이든, 누적된 messages를 통째로 프롬프트에 흘려넣으므로 같은 노드 하나가 "작성"과 "수정"을 모두 담당함. 2편의 developer_agent와 refiner_agent가 하나로 합쳐진 것.

4.2 reviewer_node — 판정

def reviewer_node(state: AgentState):
    last_msg = state['messages'][-1]   # Coder가 방금 쓴 코드
    reviewer_prompt = ChatPromptTemplate.from_messages([
        ('system', "당신은 까다로운 '전문 개발자'입니다. 코드를 엄격하게 리뷰하세요.\n"
                   "보안 취약점, 비효율, 스타일을 점검하세요.\n"
                   "완벽하고 보안 문제가 없다면 반드시 'PASS'라고만 답하세요.\n"
                   "문제가 있다면 'FAIL'이라 적고 구체적인 수정 지시를 남기세요."),
        ('user'  , '다음 코드를 리뷰해주세요:\n\n{code}'),
    ])
    chain  = reviewer_prompt | llm
    review = chain.invoke({"code": last_msg.content})
    return {'messages': [review]}   # iterations는 올리지 않음

리뷰 결과도 messages에 누적됨. 다음 차례에 Coder가 이 리뷰를 placeholder로 받아 수정에 반영함.

---

5. 조건부 엣지 — 순환을 끊는 규칙

"계속 수정할지 / 끝낼지"를 직접 판정하는 분기 함수를 작성함.

def is_continue(state: AgentState):
    last_msg   = state['messages'][-1].content
    iterations = state['iterations']

    # 1. 안전장치 — 무한 루프 = 토큰 = 비용 폭탄 방지
    if iterations >= 3:
        print('-- [System] 최대 반복 도달. 종료. --')
        return 'my_end'
    # 2. 리뷰 통과
    if "PASS" in last_msg:
        print('-- [System] 리뷰 통과. 종료. --')
        return 'my_end'
    # 3. 거절 -> 다시 작성자에게
    print('-- [System] 리뷰 거절(FAIL). 재작성. --')
    return 'gogo'

②편 111글은 라이브러리 제공 `tools_condition`에, ④편 113글은 도구 호출 여부를 보는 커스텀 분기에 맡겼음. 이번 분기 기준은 "PASS 여부 + 반복 횟수"라는 도메인 규칙이라 직접 작성함. iterations 상한이 곧 ③편 단기기억(누적 상태)을 흐름 제어에 활용하는 지점.

• iterations >= 3 — 비용·시간 안전장치. 품질이 안 나와도 무조건 멈춤.
• PASS 포함 — 완전 일치(== 'PASS')가 아니라 부분 포함(in)으로 둬, 모델이 군더더기를 붙여도 통과 판정이 됨.
• 그 외 — gogo 반환 → Coder로 되돌아가 순환.

---

6. 그래프 구성과 실행

Coder를 시작점으로, reviewer 뒤에 조건부 엣지를 걸어 순환/종료를 매핑함.

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)
workflow.add_node('coder',    coder_node)
workflow.add_node('reviewer', reviewer_node)

workflow.set_entry_point('coder')
workflow.add_edge('coder', 'reviewer')        # 작성 -> 무조건 리뷰

workflow.add_conditional_edges('reviewer', is_continue, {
    'my_end': END,       # 통과/상한 -> 종료
    'gogo'  : 'coder'    # 거절 -> 다시 작성(순환)
})

app = workflow.compile()

`{'my_end': END, 'gogo': 'coder'}` 매핑이 핵심임. is_continue가 돌려준 문자열이 실제 다음 행선지로 치환됨. 'gogo' -> 'coder' 연결이 곧 자기수정 순환(reviewer → coder)을 만든다.

흐름 시나리오는 다음과 같음.

시나리오	흐름
한 번에 통과	coder → reviewer → (PASS) → END
1회 수정	coder → reviewer → (FAIL) → coder → reviewer → (PASS) → END
상한 종료	coder → reviewer → FAIL → … → (iterations=3) → END

실행은 일부러 "비효율적으로 짜달라"고 주문해 순환을 유도함.

if __name__ == '__main__':
    inputs = {
        'messages'  : [HumanMessage(content='리스트에서 중복을 제거하고 정렬하는 함수를 만들어줘. 단, 좀 비효율적으로 작성해줘')],
        'iterations': 0
    }
    for output in app.stream(inputs):   # stream -> 노드별 진행 관찰
        print(output)

Coder가 일부러 비효율 코드를 내면 Reviewer가 FAIL을 주고, 수정본이 다시 리뷰로 돌아감. 몇 바퀴 안에 PASS가 나거나 상한에서 멈춤.

---

요약

역할이 다른 두 에이전트를 노드로 두고, 조건부 엣지로 "통과까지 반복"하는 자기수정 협업 그래프를 완성했음.

항목	LangChain 선형 협업	LangGraph 멀티 에이전트
흐름	작성→리뷰→1회 수정(고정)	작성↔리뷰 순환(통과까지)
상태	변수로 흩어짐	messages 누적 + iterations
반복 제어	while 직접 구현 필요	조건부 엣지 + 상한값
분기	if문을 비즈니스 코드에 혼합	is_continue 분기 함수로 분리
수정 역할	refiner 별도 체인	coder 노드가 placeholder로 흡수

1 ~ 4편이 한 에이전트에 도구·기억·RAG를 더해 "능력"을 키웠다면, 5편은 에이전트를 둘로 나눠 "협업과 자기검증"을 더했다. 같은 워크플로우를 선형 체인 → 순환 그래프로 옮겨 보면, LangGraph가 왜 "분기·순환·상태"에 강한지가 코드로 드러난다. 다음 확장 지점은 에이전트 N명(작성·리뷰·테스트·문서화) 협업과 MCP 연계다.