SpoonOS Graph System으로 시작하기: 2분 완성 가이드

안녕하세요! 오늘은 SpoonOS Graph System을 사용해서 LLM(대규모 언어 모델) 기반 그래프 워크플로우를 만드는 방법을 알아보겠습니다. 복잡해 보이지만 실제로는 정말 간단합니다. 2분이면 충분해요!

이 가이드에서 배울 내용

• ✅ 상태(State) 정의하는 방법
• ✅ 노드(Node) 추가하기
• ✅ 그래프 실행하기
• ✅ 체크포인트 읽기

예상 소요 시간: 약 2분
난이도: 초보자 친화적

시작하기 전에

먼저 다음 준비가 필요합니다:

1. 설치 가이드를 따라 SpoonOS를 설치하세요
2. spoon_ai.graph에서 필요한 API를 import 합니다 (예: StateGraph, END)

준비되셨나요? 그럼 바로 시작해볼까요!

첫 번째 그래프: Hello World (LLM 없이)

가장 간단한 그래프부터 만들어봅시다. 노드 하나, 엣지 하나, 그리고 체크포인트 읽기만 있으면 됩니다.

import asyncio
from typing import TypedDict
from spoon_ai.graph import StateGraph, END

class HelloState(TypedDict):
    name: str
    message: str

async def say_hello(state: HelloState) -> dict:
    return {"message": f"Hello, {state['name']}!"}

graph = StateGraph(HelloState)
graph.add_node("hello", say_hello)
graph.set_entry_point("hello")
graph.add_edge("hello", END)
app = graph.compile()

async def main():
    config = {"configurable": {"thread_id": "hello-demo"}}
    result = await app.invoke({"name": "Graph", "message": ""}, config=config)
    print(result["message"])  # Hello, Graph!

    # Checkpoint read: requires thread_id (captures state before node execution)
    snapshot = graph.get_state(config)
    print("checkpoint values:", snapshot.values)

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

실행 방법:

python my_first_graph.py

이 코드는 정말 간단하죠? say_hello 함수가 이름을 받아서 인사 메시지를 만들어줍니다. 그래프는 이 함수를 노드로 등록하고, 실행하면 자동으로 상태가 전달되고 업데이트됩니다.

첫 번째 LLM 그래프 만들기

이제 진짜 재미있는 부분입니다! LLM을 사용해서 사용자 쿼리를 분석하는 그래프를 만들어봅시다.

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

class ChatState(TypedDict):
    messages: List[dict]
    user_query: str
    llm_response: str

# LLM 초기화
llm = LLMManager()

async def analyze_query(state: ChatState) -> dict:
    """LLM을 사용해서 사용자 쿼리를 분석합니다."""
    response = await llm.chat([
        Message(role="system", content="You are a helpful crypto assistant."),
        Message(role="user", content=state["user_query"])
    ])
    return {"llm_response": response.content}

# 그래프 구성
graph = StateGraph(ChatState)
graph.add_node("analyze", analyze_query)
graph.set_entry_point("analyze")
graph.add_edge("analyze", END)
app = graph.compile()

async def main():
    result = await app.invoke({
        "messages": [],
        "user_query": "What is Bitcoin?",
        "llm_response": ""
    })
    print(result["llm_response"])

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

실행 방법:

python my_first_graph.py

이제 LLM이 사용자의 질문을 받아서 답변을 생성합니다! 정말 신기하지 않나요?

코드 이해하기

1. 상태 스키마 정의하기

그래프의 핵심은 상태(State)입니다. 상태는 그래프 전체를 흐르는 데이터 구조예요.

from typing import Any, Dict, List, TypedDict


class ChatState(TypedDict):
    messages: List[Dict[str, Any]]   # 대화 기록
    user_query: str                  # 사용자 입력
    llm_response: str                # LLM 출력

상태의 특징:

• 모든 노드가 상태를 읽고 쓸 수 있습니다
• 상태는 그래프 전체를 통해 흐릅니다
• 각 노드는 상태의 일부만 업데이트할 수 있습니다

:::tip TypedDict의 장점

• IDE 자동완성으로 실수를 줄일 수 있어요
• 타입 체킹으로 버그를 미리 잡을 수 있습니다
• 코드 자체가 문서가 됩니다
  :::

2. LLM 기반 노드 만들기

노드는 비동기 함수로, 상태를 받아서 처리하고 결과를 반환합니다.

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

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


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


llm = LLMManager()


async def analyze_query(state: ChatState) -> dict:
    """LLM을 사용해서 사용자 쿼리를 분석합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"llm_response": f"(stub) analyzed: {state.get('user_query', '')}"}

    response = await llm.chat([
        Message(role="system", content="You are a helpful crypto assistant."),
        Message(role="user", content=state["user_query"])
    ], max_tokens=200)
    return {"llm_response": response.content}

노드의 역할:

• 입력: 전체 상태 딕셔너리를 받습니다
• 처리: LLM 호출, 도구 사용, 외부 API 호출 등을 수행합니다
• 출력: 변경된 필드만 부분 업데이트로 반환합니다

3. 그래프 구성하고 실행하기

이제 그래프를 만들고 실행해봅시다!

from typing import Any, Dict, List, TypedDict

from spoon_ai.graph import StateGraph, END


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


async def analyze_query(state: ChatState) -> dict:
    # LLM 키 없이도 실행 가능하도록 스텁 버전
    return {"llm_response": f"(stub) analyzed: {state.get('user_query', '')}"}


graph = StateGraph(ChatState)
graph.add_node("analyze", analyze_query)
graph.set_entry_point("analyze")
graph.add_edge("analyze", END)
app = graph.compile()

핵심 3단계:

1. StateGraph 생성: 상태 스키마로 그래프를 만듭니다
2. 노드 추가: .add_node(name, function)으로 노드를 추가합니다
3. 진입점 설정 및 컴파일: 시작 노드를 설정하고 컴파일합니다

멀티 스텝 LLM 워크플로우

실제로는 여러 단계를 거치는 워크플로우가 더 유용하죠. 여러 LLM 호출을 연결해서 더 복잡한 작업을 해봅시다!

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

class AnalysisState(TypedDict):
    user_query: str
    intent: str
    analysis: str
    final_response: str

llm = LLMManager()

async def classify_intent(state: AnalysisState) -> dict:
    """LLM이 사용자의 의도를 분류합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"intent": "analysis_request"}
    response = await llm.chat([
        Message(role="system", content="""사용자 쿼리를 다음 중 하나로 분류하세요:
        - price_query: 가격에 대한 질문
        - analysis_request: 시장 분석 요청
        - general_question: 기타 질문
        카테고리 이름만 답변하세요."""),
        Message(role="user", content=state["user_query"])
    ])
    return {"intent": response.content.strip().lower()}

async def generate_analysis(state: AnalysisState) -> dict:
    """LLM이 상세한 분석을 생성합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"analysis": f"(stub) analysis for: {state['user_query']}"}
    response = await llm.chat([
        Message(role="system", content="You are a crypto analyst. Provide detailed analysis."),
        Message(role="user", content=f"Analyze: {state['user_query']}")
    ])
    return {"analysis": response.content}

async def format_response(state: AnalysisState) -> dict:
    """LLM이 최종 응답을 포맷팅합니다."""
    if os.getenv("DOC_SNIPPET_MODE") == "1":
        return {"final_response": f"(stub) summary: {state.get('analysis', '')[:80]}..."}
    response = await llm.chat([
        Message(role="system", content="Format this analysis into a concise, user-friendly response."),
        Message(role="user", content=f"Intent: {state['intent']}\nAnalysis: {state['analysis']}")
    ])
    return {"final_response": response.content}

# 그래프 구성: classify -> analyze -> format
graph = StateGraph(AnalysisState)
graph.add_node("classify", classify_intent)
graph.add_node("analyze", generate_analysis)
graph.add_node("format", format_response)

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

app = graph.compile()

async def main():
    result = await app.invoke({
        "user_query": "What do you think about Bitcoin's price trend?",
        "intent": "",
        "analysis": "",
        "final_response": ""
    })
    print(f"Intent: {result['intent']}")
    print(f"Response: {result['final_response']}")

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

무슨 일이 일어났나요?

graph LR
    A[User Query] --> B[Classify Intent<br/>LLM Call 1]
    B --> C[Generate Analysis<br/>LLM Call 2]
    C --> D[Format Response<br/>LLM Call 3]
    D --> E[Final Output]

    style A fill:#e1f5fe
    style E fill:#c8e6c9
    style B fill:#fff3e0
    style C fill:#fff3e0
    style D fill:#fff3e0

이 워크플로우는 다음과 같이 동작합니다:

1. 사용자 쿼리 입력: 사용자가 질문을 제공합니다
2. 의도 분류 (첫 번째 LLM 호출): LLM이 질문의 의도를 분류합니다
3. 분석 생성 (두 번째 LLM 호출): 분류된 의도에 따라 상세한 분석을 생성합니다
4. 응답 포맷팅 (세 번째 LLM 호출): 최종 사용자 친화적인 형태로 포맷팅합니다
5. 최종 결과: 모든 중간 결과와 최종 결과가 상태에 포함됩니다

각 단계가 순차적으로 실행되면서 상태가 점진적으로 업데이트되는 모습을 볼 수 있어요!

빠른 참고 가이드

구성 요소	목적	예시
StateGraph(schema)	새 그래프 생성	graph = StateGraph(MyState)
.add_node(name, fn)	LLM 기반 단계 추가	graph.add_node("analyze", llm_fn)
.add_edge(from, to)	노드 연결	graph.add_edge("a", "b")
.set_entry_point(name)	시작 노드 설정	graph.set_entry_point("start")
.compile()	실행 준비	app = graph.compile()
.invoke(state)	그래프 실행	result = await app.invoke({...})

---

마무리

SpoonOS Graph System은 복잡한 LLM 워크플로우를 간단하게 만들 수 있게 해주는 강력한 도구입니다. 처음에는 어려워 보일 수 있지만, 실제로 사용해보면 정말 직관적이고 편리하다는 걸 느끼실 거예요.

궁금한 점이 있으시면 언제든 댓글로 남겨주세요! 😊