LangGraph: 서브그래프와 서브에이전트

LangGraph부터 시작해서 LangChain, DeepAgent 등 여러개를 배우다보니, 개념이 자주 헷갈리게 된다. (LangChain부터 하지 않아서 그럴수도..) 그래서 공부한 내용들을 정리해보려고 한다.

LangGraph의 역할

LangGraph는 실행 흐름을 제어하는 프레임워크고, 각 노드가 LLM을 호출하든, 뭘 하든 관심이 없다.
LangGraph는 단지 state를 받아서 state를 반환하고 업데이트하는 것에 관심이 있다.

 

서브에이전트는 개념적 용어이고, LangGraph에서는 이를 서브그래프나 툴 호출 형태로 구현할 수 있다.

• LangChain의 create_agent()로 생성한 에이전트도 단순히 말하자면 그냥 툴노드 - 메인노드를 루핑하는 그래프다. (물론 단순화하면 그렇단거고, 실제로는 미들웨어 등 여러 요소가 들어갈 수 있다.)

Subgraph

서브그래프는 에이전트일 수도 있고, 아래처럼 단순 워크플로우일 수도 있다.

class SubState(TypedDict):
    value: int

def double_node(state: SubState):
    return {"value": state["value"] * 2}

sub_workflow = StateGraph(SubState)
sub_workflow.add_node("double", double_node)
sub_workflow.add_edge(START, "double")
sub_workflow.add_edge("double", END)

sub_graph = sub_workflow.compile()

서브그래프 연결 방식

LangGraph 공식 문서 기준으로 두 가지 방식이 있다.

Subgraphs - Docs by LangChain

 

1. 노드 안에서 서브그래프 호출

When to use: 부모와 서브그래프의 state schema가 다를 때, 또는 둘 사이에서 state를 변환해야할 때

class ParentState(TypedDict):
    query: str
    result: str

class SubState(TypedDict):
    messages: Annotated[List, operator.add]

def call_sub_node(state: ParentState):
    # 부모 state → 서브그래프 input 변환
    sub_input = {"messages": [HumanMessage(content=state["query"])]}

    # 서브그래프 직접 invoke
    sub_result = sub_graph.invoke(sub_input)

    # 서브그래프 output → 부모 state 변환
    return {"result": sub_result["messages"][-1].content}

parent_workflow = StateGraph(ParentState)
parent_workflow.add_node("call_sub", call_sub_node)

2. 서브그래프를 노드로 연결

When to use: 부모와 서브그래프가 shared key/channel을 가질 때(서브그래프만의 별도의 key를 가질수도있음)

class SharedState(TypedDict):
    messages: Annotated[List, operator.add]  # 부모/서브 모두 동일한 key

# 컴파일된 서브그래프를 그대로 노드로 추가
parent_workflow = StateGraph(SharedState)
parent_workflow.add_node("sub_agent", sub_graph)  # 변환 없이 바로 추가
parent_workflow.add_edge(START, "sub_agent")
parent_workflow.add_edge("sub_agent", END)

서브에이전트

서브그래프 안에 LLM 호출과 툴 루프가 있으면 서브에이전트라고 볼 수 있을 것 같다.

from typing import Annotated, List, TypedDict
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
import operator

@tool
def search_book(title: str) -> str:
    """책 제목으로 간단한 도서 정보를 검색합니다."""
    book_data = {
        "데미안": "헤르만 헤세의 성장소설입니다.",
        "1984": "조지 오웰의 디스토피아 소설입니다.",
        "어린 왕자": "생텍쥐페리의 철학적 동화입니다."
    }
    return book_data.get(title, "도서 정보를 찾을 수 없습니다.")

class SubAgentState(TypedDict):
    messages: Annotated[List, operator.add]

def book_node(state: SubAgentState):
    llm_with_tools = llm.bind_tools([search_book])
    system_msg = SystemMessage(content="당신은 도서 추천 및 책 정보 전문가입니다.")
    messages = [system_msg] + state["messages"]
    return {"messages": [llm_with_tools.invoke(messages)]}

sub_workflow = StateGraph(SubAgentState)

sub_workflow.add_node("book_specialist", book_node)
sub_workflow.add_node("tools", ToolNode([search_book]))

sub_workflow.add_edge(START, "book_specialist")

sub_workflow.add_conditional_edges(
    "book_specialist",
    lambda s: "tools" if s["messages"][-1].tool_calls else END
)

sub_workflow.add_edge("tools", "book_specialist")

book_graph = sub_workflow.compile()

 

위 컴파일된 그래프를 다른 그래프에서 book_graph.invoke()로 호출하거나, 아예 노드로 추가하여(스키마가 동일한경우) 서브에이전트처럼 활용이 가능해진다.

book_graph.invoke()로 호출하여 사용한다면,

result = book_graph.invoke(state)

final_response = result["messages"][-1]

return {
    "messages": [final_response]
}

이렇게 마지막 메시지를 추출해 사용하게 될 것이다.

서브에이전트를 Tool처럼 호출하기

서브그래프를 Tool로 만들면 메인 에이전트가 직접 호출할 수 있다.

# 서브그래프를 툴로 감싸기  
@tool  
def book_expert(description: str) -> str:  
"""도서 관련 작업을 전문 에이전트에게 위임합니다."""  
result = book_graph.invoke({  
"messages": [HumanMessage(content=description)]  
})  
return result["messages"][-1].content

# 메인 에이전트 state
class MainState(TypedDict):  
messages: Annotated[List, operator.add]

# 메인 노드
def main_node(state: MainState):  
llm_with_tools = llm.bind_tools([book_expert])  
return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 그래프 연결
main_workflow = StateGraph(MainState)  

main_workflow.add_node("agent", main_node)  
main_workflow.add_node("tools", ToolNode([book_expert]))  

main_workflow.add_edge(START, "agent")  

main_workflow.add_conditional_edges(  
"agent",  
lambda s: "tools" if s["messages"][-1].tool_calls else END  
)  

main_workflow.add_edge("tools", "agent")  

agent = main_workflow.compile()

SubAgentMiddleware로 서브에이전트 등록

위처럼 직접 Tool로 등록할 필요 없이, DeepAgent에서 제공하는 Middleware를 사용할 수도 있다.

# 서브그래프를 CompiledSubAgent로 래핑 (TypedDict)
book_subagent: CompiledSubAgent = {
    "name": "book_expert",
    "description": "책 제목으로 도서 정보를 조회하고 책에 대한 설명이나 추천을 제공하는 전문 에이전트입니다.",
    "runnable": book_graph,
}

agent = create_agent(
    model=llm,
    system_prompt="당신은 도서 큐레이터입니다.",
    middleware=[
        SubAgentMiddleware(
            backend=StateBackend,
            subagents=[book_subagent]
        )
    ],
)

agent.invoke()를 하고 LangSmith에서 보면 다음과 같이 task라는 툴로 등록된 것이 보인다.
(다른 예제 실행 결과라 서브에이전트는 약간 다르다.)

 

SubAgentMiddleware 내부 동작

1. task 툴을 자동으로 에이전트에 주입
   └─ tools = [task_tool] 형태로 미들웨어가 주입

2. LLM이 서브에이전트 호출 판단:
   AIMessage(tool_calls=[{"name": "task", "args": {
       "subagent_type": "weather_expert",
       "description": "서울 날씨 조회해줘"
   }}])

3. subagent_graphs[subagent_type].invoke(subagent_state) 실행
   └─ subagent_state = {부모 state에서 _EXCLUDED_STATE_KEYS 제거}
   └─ subagent_state["messages"] = [HumanMessage(content=description)]

4. result["messages"][-1].text 추출 → ToolMessage로 감싸 Command로 반환
   └─ 단순 ToolMessage가 아니라 Command(update={...}) 형태
   └─ 부모 state에 subagent state 일부도 함께 merge됨 (_EXCLUDED_STATE_KEYS 제외)

툴 방식 vs SubAgentMiddleware 비교

	툴로 직접 감싸기	SubAgentMiddleware
의존성	LangChain/LangGraph	deepagents 추가 필요
보일러플레이트	많음	적음
동작 가시성	명확	추상화됨
툴 이름	직접 지정	task 고정
반환값 가공	툴 함수 안에서 자유롭게	messages[-1] 고정
서브에이전트 여러 개	툴 여러 개 등록	subagents=[...] 리스트로 관리

 

외에도, langgraph의 supervisor 패턴이나 handoff/swarm 패턴 등이 있다고 하는데 나중에 더 알아봐야겠다.