LangGraph 에이전트 오케스트레이션: StateGraph 기반 멀티에이전트 시스템 설계

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
 
class PipelineState(TypedDict):
    project_id: str
    market: str
    config: dict
    scan_results: list[ScanResult]
    news_items: list[NewsItem]
    position_alerts: list[PositionAlert]
    signals: list[Signal]
    debate_result: dict | None
    final_decision: dict | None
    errors: list[str]
 
graph = StateGraph(PipelineState)

graph = StateGraph(PipelineState)
 
# --- Sense Layer ---
graph.add_node("data_quality_gate", data_quality_gate)
graph.add_node("volume_scanner", volume_scanner)
graph.add_node("news_collector", news_collector)
graph.add_node("position_sensor", position_sensor)
 
# --- Think Layer (Fan-out / Fan-in) ---
graph.add_node("quant_analyst", quant_analyst)
graph.add_node("llm_analyst", llm_analyst)
graph.add_node("sector_rotation", sector_rotation)
graph.add_node("analysis_merge", merge_analysis_results)
graph.add_node("bull_bear_debate", bull_bear_debate)
graph.add_node("ensemble_voter", ensemble_voter)
 
# --- Act Layer ---
graph.add_node("risk_sentinel", risk_sentinel)
graph.add_node("trade_executor", trade_executor)
 
# 엣지 정의
graph.set_entry_point("data_quality_gate")
 
# Data Quality Gate: 조건부 분기
graph.add_conditional_edges(
    "data_quality_gate",
    lambda s: "continue" if not s["errors"] else "end",
    {"continue": "volume_scanner", "end": END},
)
 
# Sense Layer 순차 실행
graph.add_edge("volume_scanner", "news_collector")
graph.add_edge("news_collector", "position_sensor")
 
# Fan-out: 병렬 분석
graph.add_edge("position_sensor", "quant_analyst")
graph.add_edge("position_sensor", "llm_analyst")
graph.add_edge("position_sensor", "sector_rotation")
 
# Fan-in: 결과 합류
graph.add_edge("quant_analyst", "analysis_merge")
graph.add_edge("llm_analyst", "analysis_merge")
graph.add_edge("sector_rotation", "analysis_merge")
 
# Think Layer 후반
graph.add_edge("analysis_merge", "bull_bear_debate")
graph.add_edge("bull_bear_debate", "ensemble_voter")
graph.add_edge("ensemble_voter", "risk_sentinel")
 
# Risk Sentinel: 조건부 분기
graph.add_conditional_edges(
    "risk_sentinel",
    lambda s: "execute" if s["final_decision"].get("approved") else "end",
    {"execute": "trade_executor", "end": END},
)
graph.add_edge("trade_executor", END)

# position_sensor -> [quant, llm, sector] : Fan-out
graph.add_edge("position_sensor", "quant_analyst")
graph.add_edge("position_sensor", "llm_analyst")
graph.add_edge("position_sensor", "sector_rotation")
 
# [quant, llm, sector] -> analysis_merge : Fan-in
graph.add_edge("quant_analyst", "analysis_merge")
graph.add_edge("llm_analyst", "analysis_merge")
graph.add_edge("sector_rotation", "analysis_merge")

from langgraph.checkpoint.postgres import PostgresSaver
 
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
app = graph.compile(checkpointer=checkpointer)
 
# 실행
result = await app.ainvoke(
    initial_state,
    config={"configurable": {"thread_id": "pipeline-run-uuid"}}
)

def risk_sentinel(state: PipelineState) -> PipelineState:
    intent = state["final_decision"]
    portfolio = get_portfolio_snapshot(state["project_id"])
 
    decision = evaluate_risk(intent, portfolio)
 
    if decision.requires_human_approval:
        # LangGraph interrupt: 실행 중단, 사람 승인 대기
        raise NodeInterrupt(
            f"고위험 거래 감지: {decision.reasons}"
        )
 
    state["final_decision"]["approved"] = decision.approved
    return state

graph.add_conditional_edges(
    "data_quality_gate",
    lambda s: "continue" if not s["errors"] else "end",
    {"continue": "volume_scanner", "end": END},
)

graph.add_conditional_edges(
    "risk_sentinel",
    lambda s: "execute" if s["final_decision"].get("approved") else "end",
    {"execute": "trade_executor", "end": END},
)

{
  "event_id": "uuid-v7",
  "ts": "2026-02-21T09:15:00.123+09:00",
  "event_type": "QUANT_DONE",
  "correlation_id": "pipeline-run-uuid",
  "project_id": "project-uuid",
  "agent": "quant_analyst",
  "payload": {
    "symbol": "005930",
    "signal": 0.72,
    "confidence": 0.68,
    "feature_vector": { "rsi_14": 32.5, "macd_signal": 0.003 }
  },
  "schema_version": "v2",
  "idempotency_key": "uuid-for-dedup"
}

Analyst Team  ->  Researcher Team  ->  Trader  ->  Risk Team  ->  Fund Manager
(4명 병렬)        (Bull/Bear 토론)     (종합)      (3관점 토론)    (최종 승인)