SpoonOS 그래프 시스템 핵심 개념 완벽 가이드

안녕하세요! 오늘은 SpoonOS 그래프 시스템의 핵심 개념들을 깊이 있게 살펴보겠습니다. 이 개념들만 제대로 이해하면 어떤 LLM 기반 워크플로우든 구축할 수 있습니다.

이 글에서 배울 내용: State, Node, Edge, Checkpointing, 그리고 병합(Merge) 동작 방식
대상 독자: Quick Start를 마친 입문자
예상 소요 시간: 약 5~8분

---

📌 전체 구조 미리보기

개념	설명	핵심 포인트
State	모든 노드가 공유하는 타입이 지정된 딕셔너리	각 노드는 상태를 읽고, 업데이트를 반환
Node	비동기 함수 (주로 LLM 호출)	단일 책임, 부분 업데이트 반환
Edge	노드 간 연결	정적, 조건부, 또는 LLM 기반
Checkpoint	각 노드 실행 전 상태 스냅샷	복구 및 Human-in-the-loop 지원

---

🗂️ State (상태)

State는 그래프 전체를 흐르는 TypedDict입니다. 모든 노드는 현재 상태를 받아서 다시 병합할 업데이트를 반환합니다.

LLM 워크플로우를 위한 State 정의하기

from typing import TypedDict, List, Dict, Any, Optional

class ConversationState(TypedDict):
    # 입력 필드
    user_query: str
    user_id: str

    # LLM 관련 필드
    messages: List[dict]           # 컨텍스트를 위한 대화 기록
    intent: str                    # 분류된 의도
    extracted_params: Dict[str, Any]  # LLM이 추출한 파라미터

    # 처리 필드
    llm_analysis: str              # LLM 분석 결과
    tool_results: Dict[str, Any]   # 도구 호출 결과

    # 출력 필드
    final_response: str            # 사용자에게 보낼 최종 응답
    confidence: float              # LLM의 신뢰도 점수

    # 시스템 필드
    execution_log: List[str]

State 병합 동작 방식

노드가 업데이트를 반환하면, SpoonOS는 이를 기존 상태에 병합합니다:

필드 타입	병합 전략	예시
스칼라 (str, int, float, bool)	교체	"old" → "new"
dict	깊은 병합	{a: 1} + {b: 2} → {a: 1, b: 2}
list	추가 (최대 100개)	[1, 2] + [3] → [1, 2, 3]
None	변경 없음	이전 값 유지

import asyncio
import os
from typing import Any, Dict, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class ConversationState(TypedDict, total=False):
    user_query: str
    intent: str
    extracted_params: Dict[str, Any]
    confidence: float


llm = LLMManager()


async def analyze_with_llm(state: ConversationState) -> dict:
    """예시: LLM 노드가 부분 업데이트를 반환합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {
            "intent": "price_query",
            "extracted_params": {"symbol": "BTC"},
            "confidence": 0.92,
        }

    await llm.chat(
        [
            Message(role="system", content="사용자 의도를 분석하고 파라미터를 추출하세요."),
            Message(role="user", content=state["user_query"]),
        ],
        max_tokens=80,
    )

    # 변경된 필드만 반환
    return {
        "intent": "price_query",
        "extracted_params": {"symbol": "BTC"},
        "confidence": 0.92,
    }


async def main() -> None:
    print(await analyze_with_llm({"user_query": "BTC 가격?"}))


if __name__ == "__main__":
    asyncio.run(main())

State 베스트 프랙티스

💡 가이드라인
1. TypedDict 사용 - IDE 자동완성과 타입 체킹 지원
2. messages 필드 포함 - 멀티턴 LLM 대화를 위해 필수
3. 신뢰도 추적 - LLM 출력에는 항상 신뢰도 점수 포함
4. JSON 직렬화 가능하게 유지 - 체크포인팅에 필수
5. 모든 필드 초기화 - invoke 시점에 기본값 제공

---

🔧 Node (노드)

Node는 작업을 수행하는 비동기 함수입니다. 주로 LLM 호출, 도구 실행, 데이터 처리 등을 담당합니다.

노드 계약(Contract)

import asyncio
import os
from typing import Any, Dict, List, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class MyState(TypedDict, total=False):
    user_query: str
    messages: List[Dict[str, Any]]
    llm_response: str


llm = LLMManager()


async def my_llm_node(state: MyState) -> dict:
    """
    노드 함수 시그니처:

    Args:
        state: 현재 그래프 상태 (읽기 전용 뷰)

    Returns:
        dict: 업데이트할 필드들 (상태에 병합됨)
    """
    # 상태에서 읽기
    query = state.get("user_query", "")
    messages = state.get("messages", [])  # 직렬화를 위한 딕셔너리 리스트

    if os.getenv("DOC_SNIPPET_MODE") == "1":
        response_text = f"(스텁) 응답: {query}"
    else:
        # 기록을 Message 객체로 변환하고 LLM 호출
        history = [Message(role=m["role"], content=m["content"]) for m in messages]
        response = await llm.chat(history + [Message(role="user", content=query)], max_tokens=120)
        response_text = response.content

    # 업데이트 반환 (부분만, 전체 상태 아님)
    # JSON 직렬화를 위해 메시지를 딕셔너리로 저장
    return {
        "llm_response": response_text,
        "messages": messages + [
            {"role": "user", "content": query},
            {"role": "assistant", "content": response_text}
        ]
    }


async def main() -> None:
    result = await my_llm_node({"user_query": "안녕하세요", "messages": []})
    print(result["llm_response"])


if __name__ == "__main__":
    asyncio.run(main())

LLM을 위한 노드 패턴

패턴 1: 의도 분류 (Intent Classification)

import asyncio
import os
from typing import Any, Dict, List, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class ConversationState(TypedDict, total=False):
    user_query: str
    intent: str
    confidence: float
    messages: List[Dict[str, Any]]


llm = LLMManager()


async def classify_intent_node(state: ConversationState) -> dict:
    """LLM을 사용하여 사용자 의도를 분류합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"intent": "general_question", "confidence": 0.9}

    response = await llm.chat(
        [
            Message(
                role="system",
                content=(
                    'JSON만 응답하세요: {"intent": "price_query|analysis_request|trade_command|general_question", '
                    '"confidence": 0.0-1.0}'
                ),
            ),
            Message(role="user", content=state["user_query"]),
        ],
        max_tokens=80,
    )

    import json

    result = json.loads(response.content)
    return {"intent": result["intent"], "confidence": result["confidence"]}


async def main() -> None:
    print(await classify_intent_node({"user_query": "BTC 가격이 얼마야?"}))


if __name__ == "__main__":
    asyncio.run(main())

패턴 2: 파라미터 추출 (Parameter Extraction)

import asyncio
import os
from typing import Any, Dict, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class ConversationState(TypedDict, total=False):
    user_query: str
    extracted_params: Dict[str, Any]


llm = LLMManager()


async def extract_params_node(state: ConversationState) -> dict:
    """LLM을 사용하여 자연어에서 파라미터를 추출합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"extracted_params": {"symbol": "BTC", "action": "buy", "amount": 0.1, "price_type": "market"}}

    response = await llm.chat(
        [
            Message(
                role="system",
                content=(
                    '트레이딩 파라미터를 JSON으로만 추출하세요. 예시: '
                    '{"symbol": "BTC", "action": "buy", "amount": 0.1, "price_type": "market"}'
                ),
            ),
            Message(role="user", content=state["user_query"]),
        ],
        max_tokens=120,
    )

    import json

    params = json.loads(response.content)
    return {"extracted_params": params}


async def main() -> None:
    print(await extract_params_node({"user_query": "시장가로 BTC 0.1개 매수"}))


if __name__ == "__main__":
    asyncio.run(main())

패턴 3: 컨텍스트 기반 분석 (Analysis with Context)

import asyncio
import os
from typing import Any, Dict, List, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class ConversationState(TypedDict, total=False):
    user_query: str
    intent: str
    tool_results: Dict[str, Any]
    messages: List[Dict[str, Any]]
    llm_analysis: str


llm = LLMManager()


async def analyze_with_context_node(state: ConversationState) -> dict:
    """축적된 컨텍스트를 활용한 LLM 분석."""
    context = f"""
사용자 질문: {state.get('user_query')}
의도: {state.get('intent')}
시장 데이터: {state.get('tool_results', {}).get('market_data', 'N/A')}
최근 메시지: {state.get('messages', [])[-3:]}
"""

    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"llm_analysis": f"(스텁) 분석: {state.get('user_query', '')}"}

    response = await llm.chat(
        [
            Message(role="system", content="당신은 전문 암호화폐 분석가입니다. 상세한 분석을 제공하세요."),
            Message(role="user", content=context),
        ],
        max_tokens=200,
    )

    return {"llm_analysis": response.content}


async def main() -> None:
    result = await analyze_with_context_node(
        {"user_query": "BTC 분석해줘", "intent": "analysis_request", "tool_results": {}, "messages": []}
    )
    print(result["llm_analysis"])


if __name__ == "__main__":
    asyncio.run(main())

패턴 4: 응답 생성 (Response Generation)

import asyncio
import os
from typing import Any, Dict, List, TypedDict

from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message


class ConversationState(TypedDict, total=False):
    user_query: str
    llm_analysis: str
    tool_results: Dict[str, Any]
    messages: List[Dict[str, Any]]
    final_response: str


llm = LLMManager()


async def generate_response_node(state: ConversationState) -> dict:
    """최종 사용자 응답을 생성합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        response_text = f"(스텁) 응답: {state.get('user_query', '')}"
    else:
        response = await llm.chat(
            [
                Message(
                    role="system",
                    content="유용하고 간결한 응답을 생성하세요. 명확하고 실행 가능하게.",
                ),
                Message(
                    role="user",
                    content=(
                        f"원래 질문: {state.get('user_query')}\n"
                        f"분석: {state.get('llm_analysis')}\n"
                        f"데이터: {state.get('tool_results', {})}\n"
                    ),
                ),
            ],
            max_tokens=200,
        )
        response_text = response.content

    return {
        "final_response": response_text,
        "messages": state.get("messages", []) + [{"role": "assistant", "content": response_text}],
    }


async def main() -> None:
    result = await generate_response_node(
        {"user_query": "안녕", "llm_analysis": "", "tool_results": {}, "messages": []}
    )
    print(result["final_response"])


if __name__ == "__main__":
    asyncio.run(main())

---

🔀 Edge (엣지)

Edge는 노드 간의 제어 흐름을 정의합니다. 그래프 시스템은 여러 유형의 엣지를 지원합니다.

1. 정적 엣지 (Static Edges)

항상 소스에서 타겟으로 전이합니다:

import asyncio
from typing import Any, Dict, List, TypedDict

from spoon_ai.graph import StateGraph, END


class ConversationState(TypedDict, total=False):
    user_query: str
    intent: str
    llm_analysis: str
    final_response: str
    messages: List[Dict[str, Any]]


async def classify_intent_node(state: ConversationState) -> dict:
    return {"intent": "general_question"}


async def analyze_with_context_node(state: ConversationState) -> dict:
    return {"llm_analysis": f"(스텁) 분석: {state.get('user_query', '')}"}


async def generate_response_node(state: ConversationState) -> dict:
    return {"final_response": f"(스텁) 응답: {state.get('llm_analysis', '')}"}


graph = StateGraph(ConversationState)
graph.add_node("classify", classify_intent_node)
graph.add_node("analyze", analyze_with_context_node)
graph.add_node("respond", generate_response_node)

graph.set_entry_point("classify")
graph.add_edge("classify", "analyze")
graph.add_edge("analyze", "respond")
graph.add_edge("respond", END)

app = graph.compile()


async def main() -> None:
    result = await app.invoke({"user_query": "비트코인에 대해 설명해줘"})
    print(result["final_response"])


if __name__ == "__main__":
    asyncio.run(main())

2. 조건부 엣지 (LLM 기반 라우팅)

LLM 분류에 따라 라우팅합니다:

import asyncio
from typing import TypedDict

from spoon_ai.graph import StateGraph, END


class ConversationState(TypedDict, total=False):
    user_query: str
    intent: str
    confidence: float
    output: str


def route_by_intent(state: ConversationState) -> str:
    """LLM이 분류한 의도에 따라 라우팅합니다."""
    intent = state.get("intent", "general")
    confidence = state.get("confidence", 0)

    # 신뢰도가 낮으면 → 명확화 요청
    if confidence < 0.7:
        return "clarify"

    return intent


async def classify(state: ConversationState) -> dict:
    q = (state.get("user_query") or "").lower()
    if "가격" in q or "price" in q:
        return {"intent": "price_query", "confidence": 0.95}
    if "분석" in q or "analy" in q:
        return {"intent": "analysis_request", "confidence": 0.9}
    if "매수" in q or "매도" in q or "buy" in q or "sell" in q:
        return {"intent": "trade_command", "confidence": 0.9}
    return {"intent": "general_question", "confidence": 0.9}


async def fetch_price(state: ConversationState) -> dict:
    return {"output": "가격 조회 핸들러"}


async def deep_analysis(state: ConversationState) -> dict:
    return {"output": "분석 핸들러"}


async def confirm_trade(state: ConversationState) -> dict:
    return {"output": "거래 확인 핸들러"}


async def general_response(state: ConversationState) -> dict:
    return {"output": "일반 응답 핸들러"}


async def ask_clarification(state: ConversationState) -> dict:
    return {"output": "명확화 요청 핸들러"}


graph = StateGraph(ConversationState)
graph.add_node("classify", classify)
graph.add_node("fetch_price", fetch_price)
graph.add_node("deep_analysis", deep_analysis)
graph.add_node("confirm_trade", confirm_trade)
graph.add_node("general_response", general_response)
graph.add_node("ask_clarification", ask_clarification)
graph.set_entry_point("classify")

graph.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "price_query": "fetch_price",
        "analysis_request": "deep_analysis",
        "trade_command": "confirm_trade",
        "general_question": "general_response",
        "clarify": "ask_clarification",
    },
)

for node in ["fetch_price", "deep_analysis", "confirm_trade", "general_response", "ask_clarification"]:
    graph.add_edge(node, END)

app = graph.compile()


async def main() -> None:
    result = await app.invoke({"user_query": "BTC 가격이 얼마야?"})
    print(result["output"])


if __name__ == "__main__":
    asyncio.run(main())

3. LLM 기반 동적 라우팅

LLM이 직접 다음 단계를 결정하도록 합니다. 가장 간단한 방법은 내장 LLM 라우터를 활성화하고 다음 노드 이름을 선택하게 하는 것입니다.

import asyncio
from typing import TypedDict

from spoon_ai.graph import StateGraph, END
from spoon_ai.graph.config import GraphConfig, RouterConfig


class ConversationState(TypedDict, total=False):
    user_query: str
    result: str


async def route(state: ConversationState) -> dict:
    # No-op 진입 노드. LLM 라우팅은 이 노드 후에 실행됩니다.
    return state


async def web_search(state: ConversationState) -> dict:
    return {"result": f"(스텁) 웹 검색: {state['user_query']}"}


async def respond(state: ConversationState) -> dict:
    return {"result": f"(스텁) 응답: {state['user_query']}"}


graph = StateGraph(ConversationState)
graph.add_node("route", route)
graph.add_node("web_search", web_search)
graph.add_node("respond", respond)
graph.set_entry_point("route")

# 핸들러 실행 후 그래프 종료
graph.add_edge("web_search", END)
graph.add_edge("respond", END)

# LLM 라우팅 활성화 및 대상 제한
graph.config = GraphConfig(
    router=RouterConfig(
        allow_llm=True,
        allowed_targets=["web_search", "respond"],
        default_target="respond",
    )
)
graph.enable_llm_routing(config={"model": "gpt-4", "temperature": 0.1, "max_tokens": 50})

app = graph.compile()


async def main():
    result = await app.invoke({"user_query": "오늘 BTC 뉴스 있어?", "result": ""})
    print(result["result"])


if __name__ == "__main__":
    asyncio.run(main())

완전한 LLM 라우팅 예제

import asyncio
from typing import TypedDict
from spoon_ai.graph import StateGraph, END
from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message

class RouterState(TypedDict):
    query: str
    intent: str
    confidence: float
    result: str

llm = LLMManager()

async def classify_intent(state: RouterState) -> dict:
    """LLM이 신뢰도와 함께 의도를 분류합니다."""
    response = await llm.chat([
        Message(role="system", content="""분류 후 JSON으로 응답:
        {"intent": "price|news|analysis|general", "confidence": 0.0-1.0}"""),
        Message(role="user", content=state["query"])
    ])

    import json
    result = json.loads(response.content)
    return {"intent": result["intent"], "confidence": result["confidence"]}

async def handle_price(state: RouterState) -> dict:
    response = await llm.chat([
        Message(role="system", content="암호화폐 가격 정보를 제공하세요."),
        Message(role="user", content=state["query"])
    ])
    return {"result": response.content}

async def handle_news(state: RouterState) -> dict:
    response = await llm.chat([
        Message(role="system", content="관련 암호화폐 뉴스를 요약하세요."),
        Message(role="user", content=state["query"])
    ])
    return {"result": response.content}

async def handle_analysis(state: RouterState) -> dict:
    response = await llm.chat([
        Message(role="system", content="상세한 시장 분석을 제공하세요."),
        Message(role="user", content=state["query"])
    ])
    return {"result": response.content}

async def handle_general(state: RouterState) -> dict:
    response = await llm.chat([
        Message(role="system", content="당신은 유용한 암호화폐 어시스턴트입니다."),
        Message(role="user", content=state["query"])
    ])
    return {"result": response.content}

def route_by_intent(state: RouterState) -> str:
    return state.get("intent", "general")

# 그래프 구축
graph = StateGraph(RouterState)

graph.add_node("classify", classify_intent)
graph.add_node("price_handler", handle_price)
graph.add_node("news_handler", handle_news)
graph.add_node("analysis_handler", handle_analysis)
graph.add_node("general_handler", handle_general)

graph.set_entry_point("classify")

graph.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "price": "price_handler",
        "news": "news_handler",
        "analysis": "analysis_handler",
        "general": "general_handler"
    }
)

graph.add_edge("price_handler", END)
graph.add_edge("news_handler", END)
graph.add_edge("analysis_handler", END)
graph.add_edge("general_handler", END)

app = graph.compile()


async def main():
    result = await app.invoke(
        {"query": "비트코인 현재 가격이 얼마야?", "intent": "", "confidence": 0.0, "result": ""}
    )
    print(result["result"])


if __name__ == "__main__":
    asyncio.run(main())

---

💾 Checkpointing (체크포인팅)

Checkpointing은 각 노드 실행 전에 상태 스냅샷을 자동으로 저장합니다. 이를 통해 다음이 가능합니다:

• 복구: 실패 후 마지막 성공 노드부터 재개
• 멀티턴 대화: 세션 간 LLM 컨텍스트 유지
• Human-in-the-loop: 사용자 입력을 위해 일시 중지, 새 데이터로 재개

Checkpointing 설정하기

from typing import Any, Dict, List, TypedDict

from spoon_ai.graph import StateGraph, InMemoryCheckpointer

checkpointer = InMemoryCheckpointer(
    max_checkpoints_per_thread=100
)

class ConversationState(TypedDict, total=False):
    user_query: str
    messages: List[Dict[str, Any]]
    llm_analysis: str

graph = StateGraph(
    ConversationState,
    checkpointer=checkpointer
)
print("checkpointer 설정됨:", graph.checkpointer is checkpointer)

멀티턴 LLM 대화

import asyncio

from typing import Any, Dict, List, TypedDict

from spoon_ai.graph import StateGraph, END, InMemoryCheckpointer


class ConversationState(TypedDict, total=False):
    user_query: str
    messages: List[Dict[str, Any]]
    llm_response: str


checkpointer = InMemoryCheckpointer(max_checkpoints_per_thread=100)


async def respond(state: ConversationState) -> dict:
    # 문서용 최소한의 결정론적 "LLM"
    user_query = state.get("user_query", "")
    response_text = f"(스텁) 답변: {user_query}"
    messages = state.get("messages", [])
    return {
        "llm_response": response_text,
        "messages": messages
        + [{"role": "user", "content": user_query}, {"role": "assistant", "content": response_text}],
    }


graph = StateGraph(ConversationState, checkpointer=checkpointer)
graph.add_node("respond", respond)
graph.set_entry_point("respond")
graph.add_edge("respond", END)
app = graph.compile()


async def main() -> None:
    # 첫 번째 턴
    result = await app.invoke(
        {"user_query": "비트코인이 뭐야?", "messages": []},
        config={"configurable": {"thread_id": "user_123_session"}},
    )

    # 두 번째 턴 - LLM이 첫 번째 턴의 컨텍스트를 가지고 있음
    result = await app.invoke(
        {"user_query": "가격 추세는 어때?", "messages": result["messages"]},
        config={"configurable": {"thread_id": "user_123_session"}},
    )

    # LLM은 대화 기록을 통해 "그것"이 비트코인을 가리킨다는 것을 알 수 있음
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

실패로부터 복구

import asyncio
from typing import TypedDict

from spoon_ai.graph import StateGraph, END, InMemoryCheckpointer


class ConversationState(TypedDict, total=False):
    user_query: str
    llm_analysis: str
    should_fail: bool


checkpointer = InMemoryCheckpointer(max_checkpoints_per_thread=100)


async def maybe_fail(state: ConversationState) -> dict:
    if state.get("should_fail"):
        raise RuntimeError("시뮬레이션된 실패")
    return {"llm_analysis": f"(스텁) 분석: {state.get('user_query', '')}"}


graph = StateGraph(ConversationState, checkpointer=checkpointer)
graph.add_node("maybe_fail", maybe_fail)
graph.set_entry_point("maybe_fail")
graph.add_edge("maybe_fail", END)
app = graph.compile()


async def main() -> None:
    config = {"configurable": {"thread_id": "analysis_session"}}
    try:
        initial_state = {"user_query": "BTC 분석해줘", "llm_analysis": "", "should_fail": True}
        result = await app.invoke(initial_state, config=config)
        print(result)
    except Exception as e:
        print(f"실패: {e}")

        # 마지막 성공 상태 가져오기
        last_state = graph.get_state(config)

        if last_state:
            print(f"마지막 노드: {last_state.metadata.get('node')}")
            print(f"체크포인트 값: {last_state.values}")


if __name__ == "__main__":
    asyncio.run(main())

---

🎯 핵심 정리

1. State는 LLM 컨텍스트를 전달 - 메시지, 추출된 파라미터, 분석 결과가 흐름
2. Node는 LLM 호출을 캡슐화 - 각 노드는 하나의 LLM 작업을 잘 수행
3. Edge는 LLM 출력에 따라 라우팅 - 의도 분류가 워크플로우를 주도
4. Checkpoint는 대화를 가능하게 함 - 멀티턴 컨텍스트가 호출 간에 보존

---