[LLM] LangGraph를 써서 간단한 multi AI Agent 시스템을 만들어보자

🦜🕸️ LangGraph?

랭그래프(LangGraph)는 LangChain 생태계를 기반으로 하는 강력한 프레임워크로, 멀티 에이전트 시스템과 AI 워크플로우 개발에 특화되어 있다. LangChain의 기능을 확장하여 LLM을 사용한 상태 유지와 여러 에이전트 간의 협업을 효율적으로 관리할 수 있도록 설계되었다. 즉, LangGraph는 단방향 체인 형태의 LangChain을 확장하여 다수의 에이전트를 상태 기반(Stateful)으로 연결하고, 조건 분기, 루프, 병렬 처리 등을 포함하는 복잡한 AI 워크플로우를 구축할 수 있는 프레임워크이다.

 

🔗 랭체인(LangChain)이 뭐에요? : https://miiinnn23.tistory.com/117

 

LangChain이 A → B → C 순으로 순차 수행된다면, 

LangGraph는 '조건 분기'와 '반복'의 개념이 존재하기 때문에 A ↔ B ↔ C 와 같이 수행될 수 있다.

 

 

이러한 랭그래프는 각 노드가 LLM 에이전트를 나타내고, 엣지가 에이전트 간의 상호작용을 나타내는 그래프 구조를 사용한다. 이를 통해 복잡한 워크플로우를 시각적으로 표현하고 관리할 수 있다. 특히 순환(cycle) 구조를 지원하여 프로세스를 지속적으로 반복하며, 변화하는 조건에 따라 동적으로 작업을 결정할 수 있다.

또한 랭그래프는 상태 관리 시스템을 통해 여러 에이전트가 정보를 공유하고, 전체 시스템의 일관성과 효율성을 유지할 수 있다. 각 노드가 글로벌 상태(STATE)를 참조하면서 동작하여, LLM을 사용한 애플리케이션의 상태를 효과적으로 유지할 수 있다는 장점이 존재한다.

---

용어 정리

✔️ AI 에이전트: 주어진 목표를 달성하기 위해 입력을 받아 reasoning과 행동을 수행하는 독립적인 실행 단위

✔️ 상태(STATE): 에이전트 실행 과정에서 공유되며, 흐름 제어나 의사결정에 활용되는 데이터의 집합

 

AI 에이전트는 각각의 로직/서비스 단위이고, 상태는 로직들이 공유하는 데이터이다.

'프롬프트 build' ➡️ '생성' ➡️ '파일 생성' 의 로직으로 구성된 레포트 생성 application을 예로 들었을 때, 각 단계 별 로직들이 AI 에이전트이자 노드가 되고,

프롬프트 build에서 생성 로직으로 이어질 때 '생성된 프롬프트'와 같이 AI 에이전트들이 생성하여 다음 에이전트로 전해줘야 하는 데이터는 곧 상태가 된다.

 

---

사용 환경

✔️ 파이썬: 3.11.9

✔️ LangChain: 0.3.25

✔️ LangGraph: 0.4.8

 

---

AI agent를 설계해보자

웹 e2e 자동화 테스트 프로젝트를 진행하면서, 테스트 시나리오와 API 정보를 종합하여 각 API별로 적합한 테스트 케이스 데이터를 만들어주는 'TC Generator System'을 설계하게 되었다. 단순히 랭체인을 통해 프롬프트로 TC를 생성할 수 있었지만, 몇 가지 고려사항이 있었기 때문에 교육에서 들었던 LangGraph를 사용한 에이전트 시스템을 구현하기로 했다. 해당 생성 로직을 구현할 때 가장 크게 고려해야 할 점은

 

- 생성된 TC가 시나리오 내에서 사용하는 API의 일정 비율 이상을 커버할 수 있어야 함

- LLM의 hallucination을 방지하기 위해 반드시 자가 검증 로직이 있어야 함

 

이렇게 두 가지가 있었다. 따라서 크게 API coverage 과 자가 검증 로직을 분기점으로 잡고 agent 5개를 가진 작은 생성 시스템을 설계하게 되었다.

 

그래프 정의

실제 랭그래프

✔️ GENERATE_TC: 입력 데이터들로부터 프롬프트를 build하고, 이로부터 LLM 기반 TC 생성.

✔️ CHECK_COVERAGE: API가 임계값 이상 커버 되었는지 룰 기반 검증.

✔️ CLEAR_TC: 생성 오류나 중단되어 임계값 이하의 API를 커버했을 시 재생성 로직 돌입하기 위해 지금까지 생성된 TC 정보 삭제.

✔️ VALIDATE_TC: 생성된 TC가 시나리오의 검증 포인트에 맞게 설계되었는지, 값의 오류는 없는지 LLM을 통한 자가 검증.

✔️ REGENERATE_TC: 검증 로직에서 오류가 있다고 판단된 TC에 한정하여 TC 재생성. 과다 생성을 방지하기 위해 3번 이상 검증 로직이 반복될 시 생성 종료.

 

STATE 정의

TC 생성을 위한 정보와 검증 로직에 필요한 전역 데이터를 고려하여 STATE를 정의하였다. STATE는 일종의 payload 개념으로, 모든 데이터를 포함할 필요는 없지만 각 에이전트가 참조할 수 있는 공통의 컨텍스트로서, 필요한 값을 적절히 정의해주는 것이 중요하다.

class FlowState(BaseModel):
    """
    TC 생성 STATE 정의
    - 시나리오 ID
    - 생성에 필요한 request 객체
    - 생성된 TC 목록
    - TC에 대한 coverage 값(검증률)
    - regenerated 필요한 tc id 및 사유
    - 재시도 횟수
    """
    scenario_id: str
    request: TestcaseGenerationRequest
    tc_list: List[TestcaseData] = []
    coverage: Optional[float] = None
    revalidation_targets: List[Tuple[str, str]] = []  # (tc_list index, reason)
    retry_count: int = 0

 

그래프 설계

노드와 STATE 정의가 끝났다면, 이제부터는 실제 로직을 설계해준다. LangGraph의 실행 흐름은 그래프 빌드 → 컴파일 → 실행(invoke) 순서로 진행되며, 이 과정은 마치 함수를 정의하듯 노드를 배치하고 흐름을 구성하는 과정에 가깝다. 별도의 그래프 설계 함수에서 compile까지 완료된 상태의 그래프를 반환하도록 정의해주었다.

 

먼저 어떤 STATE를 가진 그래프를 설계할 것인지 StateGraph의 builder를 정의해준다.

builder = StateGraph(state_schema=FlowState)

 

이후 차례대로 노드를 추가한다.

# 1. 각 API 매핑에 대해 TC 생성 → 생성된 TC는 상태에 추가
builder.add_node("generate_tc", generate_tc_node)

# 2. coverage 판단 후 분기
builder.add_node("check_coverage", coverage_check_node)

# 2-1. coverage 미달 시 재생성 로직 수행 전 TC 리스트 초기화
builder.add_node("clear_tc", clear_tc_node)

# 3. 검증 진행
builder.add_node("validate_tc", validate_tc_node)

# 4. 검증 후 실패 TC에 대한 재생성 수행
builder.add_node("regenerate_tc", regenerate_tc_node)

 

노드를 추가한 후 어디서 시작하는지 entry point를 설정하고, 각 노드를 연결해준다.

 

# 흐름 정의
builder.set_entry_point("generate_tc")
builder.add_edge("generate_tc", "check_coverage")

# coverage 판단에 따라 분기
builder.add_conditional_edges(
    source="check_coverage",
    path=coverage_decision_fn,
    path_map={
        "pass": "validate_tc",     # coverage 충분한 경우
        "fail": "clear_tc"         # 아닐시 생성 tc 초기화
    }
)

builder.add_edge("clear_tc", "generate_tc")

# 검증 실패 TC 존재 여부에 대해 분기
builder.add_conditional_edges(
    source="validate_tc",
    path=regenerate_decision_fn,
    path_map={
        "pass": END,
        "fail": "regenerate_tc",    # validation 실패 시 재생성 로직 돌입
        "giveup": END               # validation - regenerate 3번 반복 시 루프 탈출
    }
)

builder.add_edge("regenerate_tc", "validate_tc")  # 루프 추가

 

이 때 분기점은 add_conditional_edges를 통해 어떤 이름의 노드(source)로부터 어떤 함수를 기준(path)으로 해당 함수가 어떤 값을 반환하느냐에 따라 어느 노드로 갈지(path_map)를 결정해준다.

 

def regenerate_decision_fn(state: FlowState) -> str:
    """
    revalidation_targets에 재생성 대상이 있는지 판단하여 분기
    - 없으면: "pass"
    - 있으면: "fail"
    - 재검증 3번 이상: "giveup"
    """
    if not state.revalidation_targets:
        return "pass"
    if state.retry_count >= 3:
        return "giveup"  # 3회 실패 시
    return "fail"

 

이 때의 path에 들어갈 함수는 위와 같이 어떤 조건에 따라 특정 값을 반환하게 선언해주면 된다. 해당 단계에서 위와 같이 loop 조건을 설정해줄 수도 있고, 단순히 어느 노드로 갈지 분기점만 연결해주어도 된다.

 

이후 builder를 compile까지 하여 return해주면 바로 invoke하여 사용할 수 있는 그래프 빌드 함수 끝!!

return builder.compile()

 

그래프 실행

그래프는 controller나 다른 단에서 입력받은 request 데이터를 기반으로 state를 정의해주고, 빌드된 그래프에 state를 넣어 invoke만 해주면 끝난다. 정말 간단하죠?

이후 정의된 state를 응답 객체에 맞게 다시 직렬화하여 반환만 해주면 TC 생성은 끝이다. 얏호🙌

state = FlowState(scenario_id=scenario_id, request=request)

# 그래프 빌드
graph = build_testcase_flow()

# 그래프 실행
result: dict = await graph.ainvoke(state)

# 응답 조합
response = build_tc_response_from_state(result)

 

---

실행 결과

LangChain API KEY를 발급받아 실행할 때 env에 넣어주면 트래킹할 수 있다. 이 때 그래프의 어떤 로직을 탔는지, 그 때의 결과값이 어떻게 되었는지, 시간이 얼마나 소요되었는지, 얼마의 토큰이 소모되는지, 심지어는 어떤 에러가 발생했는지 까지도 간편하게 확인할 수 있기 때문에 트래킹 하는 것을 추천한다.(개인 사용 시 Tracing은 무료!)

 

one-shot learning으로 강력하게 프롬프트를 작성했더니 JsonOutputParser를 태워도 문제없는 응답이 반환되는 것을 확인할 수 있다.

coverage 범위는 60%으로 잡았기 때문에, TC 데이터 생성 도중 문제가 생기지 않는다면 웬만해서 검증로직으로 잘 넘어간다.

 

 

근데 가끔 LLM의 반란이 일어나면 저렇게된다😑

 

그치만 설계 의도대로 그래프를 잘 타는 것에 조금이나마 위안을..

ㅜㅜ

 

처음에 설계해줘야 하는 로직이나 STATE 개념이 어려워서 LangGraph와 낯가렸는데, (버전마다 코드도 천차만별이라 GPT 태워서 코드 만들어달라하기도 어려웠음 ㅜㅜ) 동작 원리를 어느정도 이해하고 나니 이거만큼 강력한 프레임워크도 없는거같다. 신기술을 열린 마인드로 배우자.