[LangGraph] 멀티 에이전트(Multi-Agent) 시스템과 structured_output 기반 Supervisor 오케스트레이션 설계

1. 도입부 (Introduction)

단일 에이전트에게 세상 모든 도구(Tools)를 쥐여주고 프롬프트로 "필요할 때마다 알아서 골라 써라"고 지시하는 것은 무모한 설계다. 도구의 개수가 많아질수록 모델의 컨텍스트는 급격히 소모되며, 이로 인해 도구의 인자 값을 오인해 엉뚱하게 실행하는 '도구 오작동(Tool Misuse)' 리스크가 임계치를 넘게 된다.

이 문제를 아키텍처 관점에서 우회하는 해법이 바로 멀티 에이전트(Multi-Agent) 시스템이다. 전체 태스크를 관심사에 따라 쪼개어 독립된 샌드박스형 에이전트(Worker)들에게 전담시키고, 이 워커들의 작업 상태를 종합하여 다음 단계를 통제하는 '수퍼바이저(Supervisor)' 에이전트를 두는 구조다.

특히 최신 LangGraph 아키텍처에서는 기존의 하드코딩된 분기 조건(Conditional Edge) 대신, 노드 내부에서 전역 그래프 흐름을 완전히 동적으로 지배할 수 있는 Command 객체가 도입되어 설계 유연성이 극대화되었다. 본 가이드에서는 금융/주식 분석이라는 정교한 실무 시나리오를 바탕으로, 수퍼바이저 모델과 3대 특화 워커 에이전트가 Command 기반으로 유기적으로 협업하는 멀티 에이전트의 정수를 해부해 보겠다.

 

 

2. 수퍼바이저(Supervisor) 디자인 패턴의 동작 원리

수퍼바이저 패턴은 오케스트라의 지휘자나 대기업의 프로젝트 매니저(PM)와 같은 역할을 설계에 반영한 것이다.

① Structured Output을 활용한 의사결정 정형화

수퍼바이저는 현재 메시지 히스토리를 훑어본 뒤, "다음 작업을 수행하기에 가장 적합한 워커는 누구인가?"라는 단순하면서도 명확한 판단을 내려야 한다. 이를 위해 수퍼바이저 노드에 탑재되는 LLM에는 특정 구조화 규격(Structured Output)이 강제된다. 워커 후보군 배열 리터럴(["market_research", "stock_research", "company_research", "FINISH"]) 중 단 하나를 골라 반환하도록 설계하여, 예측 불가능한 줄글 텍스트 대신 완벽하게 정형화된 JSON 형태의 신호를 출력하도록 보장한다.

 

② Command 객체를 통한 제어권 해방

과거의 LangGraph에서는 한 노드에서 다른 노드로 갈 때, 수퍼바이저의 결정을 해석하는 조건부 엣지(conditional_edges) 함수를 바깥에 주렁주렁 매달아 매핑해야 했다. 하지만 최신 스펙의 Command 객체는 노드 내부에서 직접 상태 변경 지침(update)과 다음 가야 할 노드 지시(goto)를 한 번에 캡슐화하여 런타임 엔진에 전달한다. 이 덕분에 그래프의 외부 결합도(Coupling)는 기하급수적으로 줄어들고 유지보수성은 비약적으로 수직 상승한다.

 

 

3. 멀티 에이전트 협업 워크플로우 설계도

우리가 구현할 주식 투자 분석 멀티 에이전트 시스템이 어떠한 구조적 흐름과 데이터 파이프라인으로 움직이는지 살펴보자.

 

 

4. 실전 코드 구현: 주식 분석 멀티 에이전트 팀

제시된 주식 분석 다이어그램 구조와 연산 도구들을 한데 엮어, 실제로 구동 가능한 완전한 LangGraph 파이프라인을 조립해 보겠다.

from typing import Literal
from typing_extensions import TypedDict
import yfinance as yf
from langchain_core.prompts import PromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools import YahooFinanceNewsTool
from langchain.agents import create_agent # 참고: 내부 에이전트 헬퍼
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.types import Command

# LLM 선언 (분석 및 지휘를 담당할 메인 모델)
llm = ChatOpenAI(model='gpt-4o')

# ==========================================
# 1. 3대 특화 에이전트의 연산 도구(Tools) 정의
# ==========================================

# 도구 A: 주가 변동 및 시세 데이터 수집 도구
@tool
def get_stock_price(ticker: str) -> str:
    """주식 티커 심볼을 전달받아 지난 1개월 동안의 종가 및 거래량 트렌드 데이터를 반환합니다."""
    # yfinance 라이브러리를 통해 과거 1개월 히스토리를 딕셔너리로 마샬링하여 리턴한다
    stock_info = yf.download(ticker, period='1mo').to_dict()
    return str(stock_info)

# 도구 B: 기업 기본 재무제표 및 정보 수집 도구
@tool
def company_research_tool(ticker: str) -> dict:
    """주식 티커 심볼을 전달받아 기초적인 재무 지표와 대차대조표 등 기본적 분석 정보를 반환합니다."""
    company = yf.Ticker(ticker)
    company_info = company.info
    financials = company.financials
    return {
        "company_info": company_info,
        "financials": financials.to_dict()
    }

# ==========================================
# 2. Worker 노드 정의 (Command 기반의 느슨한 전이 구조)
# ==========================================

# (참고) 각 하부 에이전트는 create_agent 등을 통해 적합한 도구를 탑재하여 사전 제작되었다고 가정한다.
# 예: market_research_agent (YahooFinanceNewsTool 소유)
# 예: stock_research_agent (get_stock_price 소유)
# 예: company_research_agent (company_research_tool 소유)

def market_research_node(state: MessagesState) -> Command[Literal["supervisor"]]:
    """시장 뉴스 및 업계 트렌드를 수색하여 리포팅하는 노드입니다."""
    result = market_research_agent.invoke(state)
    # 실행 결과를 전역 메시지 리스트에 누적하고, 다시 수퍼바이저 지휘 본부로 goto 이송한다.
    return Command(
        update={'messages': [result['messages'][-1]]},
        goto='supervisor'
    )

def stock_research_node(state: MessagesState) -> Command[Literal["supervisor"]]:
    """과거 주가 차트 및 가격 변동 패턴을 분석하여 리포팅하는 노드입니다."""
    result = stock_research_agent.invoke(state)
    last_msg = result['messages'][-1]
    
    # 💡 핵심 팁: 다른 노드와의 메시지 충돌을 방지하고 컨텍스트 가독성을 높이기 위해
    # 발신자 이름(name='stock_research')을 명시적으로 바인딩하여 전송한다.
    report_message = HumanMessage(content=last_msg.content, name='stock_research')
    return Command(
        update={'messages': [report_message]},
        goto='supervisor'
    )

def company_research_node(state: MessagesState) -> Command[Literal["supervisor"]]:
    """기업의 순이익, 영업활동 현금흐름 등 기본적 재무제표 정보를 분석하여 리포팅하는 노드입니다."""
    result = company_research_agent.invoke(state)
    last_msg = result['messages'][-1]
    
    report_message = HumanMessage(content=last_msg.content, name='company_research')
    return Command(
        update={'messages': [report_message]},
        goto='supervisor'
    )

# ==========================================
# 3. 최종 의사결정 애널리스트(Analyst) 노드 정의
# ==========================================
analyst_prompt = PromptTemplate.from_template(
    """당신은 월스트리트의 전설적인 투자 전략가입니다.
아래에 여러 하부 연구원들이 수집해 온 정량적/정성적 기업 정보를 기반으로,
이 주식에 대해 '매수(Buy)', '매도(Sell)', '관망(Hold)' 결정을 내리고
그에 대한 구체적 논거와 수치적 출처를 포함한 고품질 투자 분석 리포트를 한국어로 상세히 작성해 주세요.
    
수집된 정보 목록:
{messages}"""
)

analyst_chain = analyst_prompt | llm

def analyst_node(state: MessagesState):
    # 시스템과 유저의 최초 질문을 제외한, Worker들이 올려놓은 핵심 데이터들만 추출하여 LLM에 주입한다.
    result = analyst_chain.invoke({'messages': state['messages'][1:]})
    return {'messages': [HumanMessage(content=result.content, name='analyst')]}

# ==========================================
# 4. Supervisor 오케스트레이터 정의 (Structured Output)
# ==========================================
members = ["market_research", "stock_research", "company_research"]
options = members + ["FINISH"]

# 수퍼바이저 지휘 원칙 수립
supervisor_instruction = (
    "당신은 투자 리서치 팀을 이끄는 전무(Supervisor)입니다."
    f"당신의 임무는 하위 워커들인 {members}의 역량을 적재적소에 활용하여 사용자 질문에 필요한 모든 팩트를 수집하는 것입니다."
    "현재까지 수집된 데이터를 보고, 다음 단계로 어떤 연구원(Worker)을 구동시켜서 데이터를 보강할지 판단하세요."
    "모든 사실 수집(뉴스, 가격 히스토리, 재무 상태 등)이 완료되어 유저의 질문에 최종 투자의견을 낼 수 있을 정도가 된다면,"
    "지체 없이 FINISH를 반환하여 분석가(Analyst) 단계로 이송하세요."
)

# Structured Output을 위한 라우터 스키마 구조체 선언
class Router(TypedDict):
    """다음 작업을 지시할 연구원의 명칭입니다. 더 이상 정보가 필요 없으면 FINISH를 선택하십시오."""
    next: Literal[*options]

def supervisor_node(state: MessagesState) -> Command[Literal[*members, "analyst"]]:
    # 시스템 프롬프트 헤더 생성
    messages = [
        {"role": "system", "content": supervisor_instruction},
    ] + state["messages"]
    
    # 구조화된 출력 규약을 LLM에 강제하여 예기치 않은 예외 발생을 사전에 완전 봉쇄한다
    response = llm.with_structured_output(Router).invoke(messages)
    goto_destination = response["next"]
    
    # 만약 수퍼바이저가 작업을 완료(FINISH)했다면 최종 analyst 노드로 라우팅
    if goto_destination == "FINISH":
        goto_destination = "analyst"
        
    return Command(goto=goto_destination)

# ==========================================
# 5. 전체 그래프 빌드 및 컴파일
# ==========================================
graph_builder = StateGraph(MessagesState)

# 노드 전면 배치
graph_builder.add_node("supervisor", supervisor_node)
graph_builder.add_node("market_research", market_research_node)
graph_builder.add_node("stock_research", stock_research_node)
graph_builder.add_node("company_research", company_research_node)
graph_builder.add_node("analyst", analyst_node)

# 스타트라인 연결: 최초 진입로는 언제나 감독관(Supervisor)이다
graph_builder.add_edge(START, "supervisor")

# 피니시라인 연결: 종합 애널리스트 분석이 완료되면 파이프라인을 최종 종료한다
graph_builder.add_edge("analyst", END)

# 동적 그래프로 완전히 컴파일
graph = graph_builder.compile()

 

 

5. 실전 검증 및 런타임 추적 (Runtime Trace)

컴파일된 주분 분석 멀티 에이전트 시스템에 유저가 "Would you invest in SnowFlake?"(스노우플래크에 투자해야 할까요?)라는 모호하고도 광범위한 질문을 전달했을 때 일어나는 동적 흐름을 트레이싱하여 멀티 에이전틱 루프의 이점을 검증해 보자.

🔍 런타임 단계별 데이터 분석 트레이스

• [Step 1: START ──> supervisor]
  • 입력: [User: "Would you invest in SnowFlake?"]
  • 수퍼바이저 판단: Snowflake의 최신 트렌드를 파악하기 위해 시장 동향 조사가 시급하다고 인식.
  • 출력: Command(goto='market_research')
• [Step 2: market_research ──> supervisor]
  • 작업: YahooFinanceNewsTool을 가동하여 Snowflake 관련 최근 인수합병 소식, 주요 기술 동향 및 경쟁사(Databricks 등) 분석 기사를 파싱.
  • 출력: Command(update={'messages': [...]}, goto='supervisor')
• [Step 3: supervisor]
  • 상태 분석: 최신 뉴스 정보는 확보되었으나, 투자 결정을 유도하기 위해서는 구체적인 차트 데이터와 가격 추세가 부족하다고 판정.
  • 출력: Command(goto='stock_research')
• [Step 4: stock_research ──> supervisor]
  • 작업: yfinance를 가동하여 Snowflake의 1개월간 과거 일일 종가 시세를 대량 크롤링하여 테이블화.
  • 출력: Command(update={'messages': [...]}, goto='supervisor')
• [Step 5: supervisor]
  • 상태 분석: 뉴스 동향과 가격 차트는 수집되었으나, 고평가 여부를 확인하기 위해 기초적인 분기 매출 추이 및 영업이익률 등 재무 정보 조회가 핵심이라 판단.
  • 출력: Command(goto='company_research')
• [Step 6: company_research ──> supervisor]
  • 작업: 기업 조사 툴을 실행하여 Snowflake의 P/E Ratio, P/S Ratio, 당기순이익 트렌드 데이터를 일목요연하게 패킹.
  • 출력: Command(update={'messages': [...]}, goto='supervisor')
• [Step 7: supervisor]
  • 상태 분석: "주식 가치 판단에 필요한 정성적(뉴스), 정량적(차트/재무제표) 사실 정보가 100% 포화 상태에 이르렀다"고 진단. 리서치 작전을 무사히 끝마치기로 결정.
  • 출력: Command(goto='analyst')
• [Step 8: analyst ──> END]
  • 작업: 앞서 수집된 엄청난 양의 팩트 파일들을 넘겨받아 종합적 밸류에이션 평가를 시작. AI의 단순 환각(Hallucination)에 기반한 의견 대신, RAG 및 툴로 엄격히 검증된 숫자를 기반으로 구체적 이유를 곁들인 'BUY(매수)' 또는 'HOLD(관망)' 종합 투자 제안 리포트를 작성한 뒤 파이프라인을 온전하게 종료시킴.

 

 

6. 결론: 느슨한 위임과 정형화된 지휘의 완벽한 조화

단일 LLM 시스템은 아무리 큰 지능을 가졌을지라도, 여러 도구와 무거운 연산이 동시다발적으로 밀려 들어오면 쉽게 탈진(Context Saturation)하고 주의를 빼앗긴다.

멀티 에이전트 구조는 "한 놈은 오직 도구 검색만 수행하고, 다른 놈은 재무제표 팩트만 캐며, 리더는 이들을 엄격히 분기 제어하고 최종 정리한다"는 철저한 분업과 협업(Divide and Conquer)의 철학을 고스란히 정적 그래프 위에 구현해 낸 것이다.

특히 LangGraph의 최신 Command 패턴을 탑재함으로써, 우리는 수퍼바이저 노드의 코드를 단 한 줄도 수정하지 않고도 내일 당장 "슬랙 알림 워커"나 "포트폴리오 백테스팅 워커" 노드를 추가 장착할 수 있는 구조적 고속도로를 확보하게 되었다. AI를 단순한 한 줄의 텍스트 파트너가 아니라, 시스템 수준에서 스스로 계획하고 위임하며 실행을 제어하는 조직의 핵심 개발 파트너로 진화시키는 과정의 끝에 바로 이 LangGraph 멀티 에이전트 설계가 우뚝 서 있다.