서론

Claude, Gemini, ChatGPT 등을 사용하다 보면 최근 부쩍 자주 등장하는 것이 Deep Research입니다.

주제를 던지면 AI가 알아서 검색하고, 알아서 취사선택하며, 알아서 리포트로 정리해 주는 기능이죠.

편리하긴 하지만, 내부에서 어떤 일이 일어나고 있는지는 상당히 블랙박스(Black Box)입니다.

"검색한다 → 좋고 나쁨을 판단한다 → 안 좋으면 다시 검색한다 → 정리한다"와 같은 자율 루프(Autonomous Loop)를 직접 만들 수 있을까요?

그래서 이 기사에서는 LangGraph를 사용하여 자율형 지식 수집 에이전트를 가볍게 직접 만들어 보겠습니다.

소재로 삼는 것은 실제로 구동 중인 Knowledge Researcher라는 프로젝트입니다.

주제를 전달하면 X / Web / YouTube / Reddit / Zenn을 횡단 검색하여 품질을 자기 평가하고, 노이즈라면 검색을 다시 수행하며, 최종적으로 Markdown 리포트를 출력합니다.

그 설계를 바탕으로 자율 에이전트를 만드는 방법을 정리해 나가겠습니다.

이 기사에서 알 수 있는 것

이 기사에서 다루는 내용은 다음과 같습니다.

대상 독자는 다음과 같습니다.

• LangGraph라는 이름은 들어봤지만, 아직 구조가 완전히 이해되지 않은 사람
• Deep Research 같은 것을 직접 만들어 보고 싶은 사람
• LLM 에이전트를 "단발성 프롬프트"에서 "루프(Loop) 처리"로 진화시키고 싶은 사람

자율형 리서치 에이전트가 해결하는 과제

LLM에게 조사를 시키고 싶을 때, 흔히 있는 요구사항은 다음과 같습니다.

단순하게 "검색해서 정리해 줘"라고 프롬프트 한 번으로 부탁하면, 다음과 같은 상황이 발생하기 쉽습니다.

• 검색 결과가 부실해도 그대로 요약해 버림
• 노이즈 기사를 사실로 받아들여 그럴듯한 거짓말을 작성함
• 하나의 소스만 보기 때문에 관점이 편향됨

에이전트화의 즐거움은 "안 좋으면 다시 한다"라는 판단을 AI 스스로 할 수 있게 만드는 것에 있습니다.

사람이 "이 검색 결과는 별로네, 키워드를 바꿔야겠다"라고 하는 시행착오를 그래프의 루프(Loop)로서組み込む(組み込む, 포함시킨다).

이것이 자율형 리서치 에이전트의 핵심입니다.

시스템의 기본 구조

이 에이전트에는 주로 두 명의 등장인물이 있습니다.

"수집하는 사람"과 "평가하는 사람"을 나누는 것이 포인트입니다.

왜 역할을 나누는 걸까요?

한 명의 에이전트에게 "검색하고, 평가하고, 정리해"를 전부 시키면, 자신이 수집한 결과에 대해 관대한 평가를 내리기 쉽습니다.

사람도 자신이 쓴 문장을 리뷰할 때는 관대해지곤 합니다.

따라서, 수집하는 역할과 그것을 비판적으로 평가하는 역할을 분리합니다.

이것이 멀티 에이전트(Multi-Agent) 구조의 기본적인 발상입니다.

그리고 이 두 사람의 상호작용을 제어하는 것이 LangGraph입니다.

LangGraph의 기본 구조

LangGraph는 LLM 애플리케이션의 처리를 그래프 (상태 머신, State Machine) 로 작성하기 위한 라이브러리입니다.

주요 등장인물은 세 가지입니다.

State

State는 그래프 전체를 흐르는 하나의 상태 객체 (State Object) 입니다.

각 노드(Node)는 이 State를 읽고, 업데이트된 차분(Difference)을 반환합니다.

노드 간에 변수를 직접 주고받는 것이 아니라, 모든 것을 State를 경유합니다.

이를 통해 노드 간의 암묵적인 의존성이 사라집니다.

이 프로젝트에서는 Pydantic의 BaseModel로 State를 정의하고 있습니다.

from pydantic import BaseModel, Field
from typing import Optional
class GraphState(BaseModel):
...

"지금 이 에이전트가 무엇을 가지고 있는가"를 이 State만 보면 전부 알 수 있습니다.

Node

Node는 State를 받아 업데이트된 차분을 반환하는 함수입니다.

I/O(외부 API 호출)가 많으므로, 모두 async def로 작성한다.

async def reviewer(state: GraphState) -> dict:
# state.raw_results를 보고 품질을 판정하여...
decision = await judge_quality(state.raw_results, state.theme)
...

노드는 State 전체를 반환할 필요는 없다.

업데이트하고 싶은 필드만 dict로 반환하면, LangGraph가 State에 머지(merge)해 준다.

Edge

Edge는 노드에서 노드로 향하는 화살표다.

일반적인 화살표(반드시 다음으로 진행)와 조건 분기(상태를 보고 목적지를 변경)가 있다.

from langgraph.graph import StateGraph, START, END
builder = StateGraph(GraphState)
builder.add_node("reviewer", reviewer)
...

이 add_conditional_edges가 자율 루프(autonomous loop)를 만들 때 주인공이 된다.

전체 흐름

이 에이전트의 그래프는 대략 다음과 같은 흐름이다.

사용자 입력
↓
input_parser … 테마 추출
...

포인트는 reviewer에서 뻗어 나오는 4개의 화살표다.

이곳이 "진행할 것인가 또는 다시 시도할 것인가"를 결정하는 분기점이 된다.

자율 루프의 심장부: reviewer

에이전트를 "자율적"으로 만드는 것은 바로 이 reviewer 노드다.

수집한 검색 결과를 보고 3단계로 품질을 판정한다.

이 판정에도 LLM을 사용하지만, 자유 텍스트가 아닌 **구조화된 출력 (Structured Output)**으로 받는다.

from pydantic import BaseModel, Field
from typing import Literal
class ReviewDecision(BaseModel):
...

LLM에게 "PASS인지 NOISE인지 자유롭게 써줘"라고 부탁하면, "대체로 괜찮은 것 같습니다"와 같은 모호한 답변이 돌아온다.

그 대신, Literal["PASS", "NOISE", "INSUFFICIENT"]를 사용하여 취할 수 있는 값을 제한한다.

이렇게 하면 후속 라우팅(routing)을 안전하게 작성할 수 있다.

판정을 받아 목적지를 결정하는 것이 라우팅 함수다.

def route_after_review(state: GraphState) -> str:
if state.review is None:
return "fallback_reporter"
...

query_refiner로 진행하면 재시도 횟수를 늘리고, 검색 쿼리를 query_planner로 되돌린다.

그리고 다시 검색 → 평가 …… 와 같이 루프를 돈다.

무한 루프에 빠지지 않도록 max_retries(기본값 3회)로 중단한다.

중단되면 그 시점의 최선(best effort)을 다해 fallback_reporter가 리포트를 작성한다.

**"다시 시도하되, 포기할 타이밍도 알고 있다"**는 점이 중요한 부분이다.

리서처(Researcher) 측면을 들여다보기

평가 전 단계, 즉 수집 측면도 살펴보자.

query_planner: 검색 쿼리를 생성한다

테마를 각 소스에 최적화된 검색 쿼리로 전개하는 노드다.

X, Web, YouTube, Reddit, Zenn은 각각 효과적인 키워드가 다르다.

따라서 소스별로 쿼리를 나누어 만든다.

이곳 역시 구조화된 출력으로 받는다.

class PlannedQueries(BaseModel):
web_queries: list[str] = Field(default_factory=list)
youtube_queries: list[str] = Field(default_factory=list)
...

재시도 시에는 failed_queries(이전에 실패했던 키워드)를 피하도록 지시한다.

같은 실수를 반복하지 않기 위한 장치다.

검색 노드: 5개의 소스를 병렬로 호출한다

검색 노드는 5개가 있지만 구조는 거의 동일하다.

쿼리를 받아 API를 호출하고, SearchResult 리스트를 반환할 뿐이다.

LangGraph에서는 query_planner에서 5개의 노드로 동시에 에지를 연결하면, **자동으로 병렬 실행 (Parallel Execution)**을 해준다.

# query_planner에서 5개 소스로 fan-out (병렬)
builder.add_edge("query_planner", "search_web")
builder.add_edge("query_planner", "search_x")
...

5개의 소스를 순차적으로 호출하면 느리지만, 병렬로 처리하면 가장 느린 소스의 시간만큼만 소요된다.

참고로, 검색의 실체는 Tavily API에 의존하고 있다.

X도 Reddit도 Zenn도, site: 지정을 통해 도메인을 제한한 웹 검색 (Web Search)으로 취급한다.

X API를 직접 사용하면 비용 부담이 크기 때문에, 검색 API로 대체하고 있다.

aggregator: 수집한 결과를 정리한다

5개 소스의 결과가 raw_results로 합류한다.

여기서 두 가지 전처리를 수행한다.

특히 두 번째 방식이 은근히 효과적이다.

긴 YouTube 자막이나 장문의 기사를 그대로 하류 (Downstream)로 흘려보내면 토큰을 모두 소모해 버린다.

게다가 문맥이 너무 길면 LLM은 중간의 정보를 놓치는 현상(Lost in the Middle)이 발생한다.

따라서 긴 본문만 먼저 1,000자 정도로 압축해 둔다 (Map-Reduce 방식의 전처리).

State 업데이트 시 주의점: reducer

여기서 병렬 노드 특유의 주의할 점이 하나 있다.

search_web과 search_youtube가 동시에 raw_results를 업데이트하려고 하면 어떻게 될까?

일반적인 방식으로 덮어쓰면 한쪽의 결과가 사라진다.

LangGraph에서는 필드에 병합 방법 (reducer)을 지정하여 이 문제를 해결한다.

from typing import Annotated
import operator
class GraphState(BaseModel):
...

operator.add를 지정하면 각 노드가 반환한 리스트가 **연결 (Concatenate)**된다.

병렬로 서로 다른 에러를 반환하더라도 모두 State에 남는다.

raw_results처럼 "검색 중에는 추가, 집계 후에는 교체"와 같은 복잡한 동작이 필요한 경우에는 직접 reducer 함수를 작성한다.

def _replace_or_add(current: list, update: list) -> list:
    if len(update) == 0: # 명시적인 리셋
        return []
    ...

병렬로 State를 다룰 때는 이 reducer 설계가 매우 중요하다.

그래프를 조립한다

노드와 에지가 모두 준비되었다면, 마지막으로 조립하여 컴파일(Compile)한다.

from langgraph.graph import StateGraph, START, END
def build_graph(checkpointer=None):
    builder = StateGraph(GraphState)
    ...

compile()의 반환값은 일반적인 Callable처럼 호출할 수 있다.

graph = build_graph()
result = await graph.ainvoke(GraphState(user_query="AI 개발의 최신 트렌드"))
print(result["final_report"])

이제 주제를 전달하면 자율적으로 조사하여 리포트를 반환하는 에이전트가 동작한다.

checkpointer: 중간부터 재개하기

build_graph(checkpointer=...) 부분에 주목하자.

LangGraph는 각 노드 실행 후의 상태를 저장할 수 있다 (Checkpointer).

도중에 크래시가 발생하더라도 thread_id를 지정하면 중단된 지점부터 재개할 수 있다.

from langgraph.checkpoint.memory import MemorySaver
graph = build_graph(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "run-001"}}
...

에이전트가 오래 실행될수록, 이러한 재개 기능은 매우 유용하다.

현재 구현에서는 MemorySaver (메모리 상에 유지)를 사용하고 있다.

프로세스를 넘어서 영속화(Persistence)하고 싶다면 SQLite 버전으로 교체하는 발전 방향을 생각할 수 있지만, 모든 노드가 async이므로 동기 방식인 SqliteSaver가 아니라 비동기 대응 방식인 AsyncSqliteSaver를 선택하게 된다.

설계의 베스트 프랙티스 (Best Practices)

자율 에이전트(Autonomous Agent)를 만들 때 유효했던 핵심 포인트를 정리해 둔다.

1. 노드는 작게, 단일 책임(Single Responsibility)을 갖도록 한다

1노드 = 1태스크(Task)로 만든다.

"검색하고 평가하고 요약한다"와 같은 거대한 노드를 만들지 않는다.

작게 나누면 다음과 같은 이점이 있다.

2. LLM 출력은 반드시 구조화한다

LLM에게 자유 텍스트(Free text)를 쓰게 하면 후속 처리가 망가진다.

PASS / NOISE와 같이 분기(Branching)에 관여하는 출력은 반드시 Pydantic으로 스키마(Schema)를 강제해야 한다.

# 나쁜 예: 무엇이 반환될지 알 수 없음
"좋은지 나쁜지 알려줘"
# 좋은 예: 취할 수 있는 값을 고정함
...

참고로, 최종 리포트(사람이 읽는 Markdown)만은 예외로 하여 자유 텍스트로 작성한다.

기계가 다음에 사용할 출력은 구조화하고, 사람이 읽는 최종 결과물은 자유 텍스트로 한다는 기준을 세운 것이다.

또한, 이 구조화된 출력은 LLM 프로바이더(Provider)에 의존하지 않는다.

이 프로젝트는 기본적으로 Google Gemini를 사용하지만, LLM_PROVIDER=azure로 전환하면 Azure OpenAI에서도 동일한 노드 코드가 그대로 동작한다.

모델도 노드의 용도에 따라 구분하여 사용하고 있는데, 요약에는 상위 모델을, 단순한 추출이나 압축에는 경량 모델을 사용하는 식으로 배분하고 있다.

3. 외부 I/O는 병렬로 처리한다

검색이나 API 호출은 I/O 대기 시간이 지배적이다.

async def로 작성하고, 독립적인 노드는 병렬로 흘려보낸다.

LangGraph라면 동일한 노드로 에지(Edge)를 여러 개 연결하는 것만으로 팬아웃(Fan-out)을 구현할 수 있다.

4. 실패해도 멈추지 않는다

외부 API는 흔히 다운된다.

하나의 소스가 다운되더라도 나머지 결과로 계속 진행할 수 있는 설계로 만든다.

어디서 실패하더라도 마지막에는 반드시 어떤 형태로든 리포트가 나온다.

이것이 에이전트의 신뢰성(Reliability)으로 이어진다.

5. 토큰을 절약한다

LLM 에이전트는 방심하면 토큰을 낭비한다.

루프(Loop)를 도는 에이전트는 동일한 처리를 여러 번 거치므로, 1회당 비용이 매우 중요하다.

실제로 실행하면 어떻게 되는가

지금까지 구조에 대해 이야기했지만, 실제로 실행하면 어떻게 움직이는지 맛보기로 살펴보자.

로그가 흐른다

실행하면 각 노드가 순차적으로 움직이는 모습이 로그에 나타난다.

[info] url_history: loaded past URLs count=176
[info] input_parser: extracted theme='AI 개발 최신 트렌드' brief_len=199
[info] query_planner: planned x=6 web=8 youtube=4 reddit=4 zenn=3
...

이벤트 이름은 구현된 로그 출력에 대응한다 (값은 실행할 때마다 변한다).

주목해야 할 점은 3가지다.

첫 번째는 query_planner: planned 행이다. 하나의 테마로부터 X, Web, YouTube, Reddit, Zenn용으로 총 25개의 쿼리가 생성되고 있다.

두 번째는 search_youtube: no transcript 행이다. 자막이 없는 영상은 여기서 걸러진다. "가져올 수 없는 것은 버린다"라는 단호함이 멈추지 않는 에이전트에게는 효과적이다.

세 번째는 reviewer: decision quality='PASS' 행이다. 수집한 결과를 스스로 평가하여, 합격이면 다음으로 넘어가고 실패하면 쿼리를 다시 짜서 돌아온다. 이 한 줄이 자율 루프의 분기점이 된다.

출력 리포트 맛보기

최종적으로 나오는 것은 다음과 같은 Markdown 리포트다.

2023년 AI 개발 최신 트렌드 리포트

목차

1. 대규모 언어 모델(LLMs)의 진화
   ...

목차, 개요, 통찰, 참고 링크가 나열된 구조화된 리포트가 주제를 던지는 것만으로 생성된다.

소스는 18개까지 중복 제거되며, 상위 항목은 딥 리서치(Deep Research)를 통해 본문 전체를 읽어 들인 후 요약된다.

어디까지 에이전트화해야 하는가

자율 루프(Autonomous Loop)는 편리하지만, 모든 것을 루프로 만들 필요는 없다.

반대로, 정해진 절차를 흘려보내기만 한다면 단순한 파이프라인(Pipeline)이 더 빠르고 저렴하다.

향후 실용성에 대한 전망

현재 형태에서도 동작하지만, 실용적인 방향으로 나아간다면 아직 발전시킬 여지가 있다. 구현을 위한 기반이 어디까지 마련되어 있는지를 포함하여 몇 가지 사항을 꼽아보겠다.

1. 인간의 개입 포인트 추가 (HIL)

완전 자동은 편하지만, 검색 쿼리(Query)가 어긋나면 25개 전체가 엉뚱한 방향으로 흐를 수 있다.

따라서, query_planner 직후에 한 번 멈추어 인간이 쿼리를 확인하게 하는 메커니즘이 효과적이다.

LangGraph에는 interrupt()라는, 그래프를 중간에 일시 정지하는 기능이 있다.

# query_planner 이후에 인간의 확인을 거침
approved = interrupt({"queries": planned_queries})

이 프로젝트에는 설정 플래그 HIL_ENABLED와 상태 필드 hil_approved까지는 준비되어 있다. 다만 interrupt()를 그래프에 통합하는 작업은 아직 완료되지 않아 기반만 마련된 상태다. 이제 노드에 한 수만 더하면 되는 단계까지는 와 있다.

'전자동'과 '전수동' 사이의 중간을 취하여, 요점만 인간이 키를 잡는 것. 이것이 Human-in-the-Loop (HIL)의 사고방식이다.

2. 정기 실행을 통한 차분(Difference) 추적

리서치는 단 한 번으로 끝나는 것이 아니라, 정점 관측(Fixed-point observation)하고 싶은 경우가 많다.

이 프로젝트에서는 과거에 수집한 URL을 past_urls.csv에 기록하고, 다음 실행 시에는 이미 나온 URL을 제외하는 차분 리포트 메커니즘이 이미 작동하고 있다 (aggregator가 과거 URL을 확인하여 중복을 제거함).

여기에 정기 실행을 연결하면 '지난번 이후의 신착 정보만' 추적할 수 있다. 설정으로는 CRON_SCHEDULE 프레임도 마련되어 있어, 스케줄러 연동은 앞으로의 발전 가능성이 높다.

3. 스트리밍을 통한 중간 과정 표시

지금은 최종 리포트가 나올 때까지 기다리는 구조이지만, 노드의 진행 상황을 순차적으로 흘려보낼 수 있다면 경험이 달라진다.

STREAMING_ENABLED 플래그가 준비되어 있으므로, LangGraph의 astream을 통해 중간 상태를 포착하여 UI에 흘려보내는 확장이 적합하다.

도구들은 대체로 갖춰져 있다. 이제 '어디까지 인간이 관여할 것인가', '얼마나 자주 돌릴 것인가'를 사용하는 상황에 맞춰 조정해 나가는 단계다.

요약

LangGraph로 자율형 리서치 에이전트를 구성해 보았다.

중요한 포인트는 다음과 같다.

"검색한다 → 평가한다 → 안 되면 다시 한다 → 정리한다"라는 인간의 시행착오를 그래프로 구성할 수 있는 것이 LangGraph의 재미다.

처음 만든다면 다음과 같은 작은 루프부터 시작하는 것을 추천한다.

• 1개의 소스만 검색하여 요약하기
• 결과를 자기 평가하고, 안 되면 딱 한 번만 재시도(Retry)하기
• 거기서부터 검색 소스나 판정 로직을 추가하기

작은 노드들을 쌓아 올리다 보면 자신만의 리서치 에이전트가 완성된다.

LangGraph는 LLM에게 '생각하기'뿐만 아니라 '다시 하기'를 부여하기 위한 메커니즘이다.

참고

Discussion