SpoonOS Graph System 빠른 시작 가이드
SpoonOS Graph System을 사용해서 2분 만에 LLM 기반 그래프 워크플로우를 만들어보세요. 이 가이드에서는 단계별로 그래프를 구성하고 실행하는 방법을 자세히 설명합니다.
이번 가이드에서 배울 내용:
- 상태(State) 정의하기
- 노드(Node) 추가하기
- 그래프 실행하기
- 체크포인트(Checkpoint) 읽기
대상 독자: 처음 사용하는 분들
예상 소요 시간: 약 2분
시작하기 전에
이 가이드를 따라하기 전에 다음 사항을 확인해주세요:
- 시작하기 / 설치 섹션의 단계를 완료했는지 확인
spoon_ai.graph에서 공개 API를 import 할 수 있는지 확인 (예:from spoon_ai.graph import StateGraph, END)
2분 만에 만드는 Hello World (LLM 없이)
가장 간단한 실행 가능한 그래프부터 시작해봅시다. 하나의 노드, 하나의 엣지, 그리고 thread_id를 사용한 체크포인트 읽기로 구성되어 있습니다.
이 예제는 그래프의 기본 구조를 이해하는 데 도움이 됩니다. 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) # "hello"라는 이름의 노드 추가
graph.set_entry_point("hello") # 진입점 설정
graph.add_edge("hello", END) # hello 노드에서 종료로 연결
app = graph.compile() # 그래프 컴파일
async def main():
# thread_id를 사용한 설정: 같은 thread_id로 실행하면 상태가 유지됨
config = {"configurable": {"thread_id": "hello-demo"}}
# 그래프 실행: 초기 상태를 전달하고 결과를 받음
result = await app.invoke({"name": "Graph", "message": ""}, config=config)
print(result["message"]) # 출력: Hello, Graph!
# 체크포인트 읽기: thread_id가 필요함 (노드 실행 전 상태를 캡처)
snapshot = graph.get_state(config)
print("checkpoint values:", snapshot.values)
if __name__ == "__main__":
asyncio.run(main())실행 방법:
터미널에서 다음 명령어를 실행하세요:
python my_first_graph.py이 코드를 실행하면 "Hello, Graph!"라는 메시지가 출력됩니다. 간단하지만 그래프의 핵심 개념인 상태, 노드, 엣지를 모두 포함하고 있어요.
첫 번째 LLM 그래프 만들기
이제 실제로 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 매니저 초기화: LLM과 통신하기 위한 객체
llm = LLMManager()
# 쿼리 분석 노드: LLM을 사용해서 사용자 질문을 분석
async def analyze_query(state: ChatState) -> dict:
"""LLM을 사용하여 사용자 쿼리를 분석합니다."""
# LLM에게 시스템 프롬프트와 사용자 질문을 전달
response = await llm.chat([
Message(role="system", content="You are a helpful crypto assistant."),
Message(role="user", content=state["user_query"])
])
# LLM의 응답을 상태에 저장
return {"llm_response": response.content}
# 그래프 구성
graph = StateGraph(ChatState) # 채팅 상태로 그래프 생성
graph.add_node("analyze", analyze_query) # 분석 노드 추가
graph.set_entry_point("analyze") # 진입점을 analyze로 설정
graph.add_edge("analyze", END) # analyze에서 종료로 연결
app = graph.compile() # 컴파일
async def main():
# 초기 상태로 그래프 실행
result = await app.invoke({
"messages": [],
"user_query": "What is Bitcoin?",
"llm_response": ""
})
# LLM의 응답 출력
print(result["llm_response"])
if __name__ == "__main__":
asyncio.run(main())실행 방법:
python my_first_graph.py이 코드를 실행하면 "What is Bitcoin?"이라는 질문에 대해 LLM이 생성한 답변이 출력됩니다. LLM이 암호화폐에 대한 설명을 제공할 거예요.
코드 이해하기
이제 각 부분이 어떻게 동작하는지 자세히 살펴봅시다.
1. 상태 스키마 정의하기
상태(State)는 그래프 전체를 흐르는 데이터입니다. 모든 노드가 이 상태를 읽고 쓸 수 있어요.
from typing import Any, Dict, List, TypedDict
# TypedDict를 사용하면 타입 안정성과 IDE 자동완성을 얻을 수 있습니다
class ChatState(TypedDict):
messages: List[Dict[str, Any]] # 대화 기록
user_query: str # 사용자 입력
llm_response: str # LLM 출력상태(State)란?
- 그래프 전체에서 공유되는 데이터 저장소입니다
- 각 노드는 상태를 읽어서 처리하고, 결과를 상태에 다시 쓸 수 있습니다
- 상태는 그래프가 실행되는 동안 계속 유지되고 업데이트됩니다
:::tip TypedDict를 사용하는 이유
- IDE 자동완성: 코드를 작성할 때 상태 필드 이름을 자동으로 제안해줍니다
- 타입 체크: 잘못된 필드 이름이나 타입을 사용하면 미리 오류를 발견할 수 있습니다
- 자기 문서화: 코드만 봐도 어떤 데이터가 필요한지 바로 알 수 있습니다
:::
2. LLM 기반 노드 만들기
노드(Node)는 그래프의 작업 단위입니다. 각 노드는 비동기 함수로 구현되며, 상태를 받아서 처리하고 결과를 반환합니다.
import os
from typing import Any, Dict, List, TypedDict
from spoon_ai.llm import LLMManager
from spoon_ai.schema import Message
# total=False를 사용하면 모든 필드가 필수가 아닙니다
class ChatState(TypedDict, total=False):
messages: List[Dict[str, Any]]
user_query: str
llm_response: str
# LLM 매니저 초기화
llm = LLMManager()
async def analyze_query(state: ChatState) -> dict:
"""LLM을 사용하여 사용자 쿼리를 분석합니다."""
# 문서 스니펫 모드에서는 실제 LLM 호출 없이 스텁 응답 반환
if os.getenv("DOC_SNIPPET_MODE") == "1":
return {"llm_response": f"(stub) analyzed: {state.get('user_query', '')}"}
# 실제 LLM 호출: 시스템 프롬프트와 사용자 메시지 전달
response = await llm.chat([
Message(role="system", content="You are a helpful crypto assistant."),
Message(role="user", content=state["user_query"])
], max_tokens=200) # 최대 토큰 수 제한
# LLM 응답을 상태에 저장
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', '')}"}
# 1단계: 상태 스키마로 그래프 생성
graph = StateGraph(ChatState)
# 2단계: 노드 추가 및 연결
graph.add_node("analyze", analyze_query) # 노드 추가
graph.set_entry_point("analyze") # 진입점 설정
graph.add_edge("analyze", END) # 엣지 추가
# 3단계: 그래프 컴파일 (실행 준비)
app = graph.compile()세 가지 필수 단계:
StateGraph(schema): 상태 스키마를 사용해서 새로운 그래프를 생성합니다.add_node(name, function): 노드를 추가합니다. 이름과 함수를 지정해주세요.set_entry_point()및.compile(): 진입점을 설정하고 그래프를 컴파일합니다
컴파일된 그래프는 app.invoke() 메서드를 사용해서 실행할 수 있습니다.
다단계 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()
# 1단계: 의도 분류 노드
async def classify_intent(state: AnalysisState) -> dict:
"""LLM이 사용자의 의도를 분류합니다."""
if os.getenv("DOC_SNIPPET_MODE") == "1":
return {"intent": "analysis_request"}
# LLM에게 의도 분류 작업을 요청
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()}
# 2단계: 상세 분석 생성 노드
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}
# 3단계: 최종 응답 포맷팅 노드
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") # 시작점: classify
graph.add_edge("classify", "analyze") # classify -> analyze
graph.add_edge("analyze", "format") # analyze -> format
graph.add_edge("format", END) # format -> 종료
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[사용자 질문] --> B[의도 분류<br/>LLM 호출 1]
B --> C[분석 생성<br/>LLM 호출 2]
C --> D[응답 포맷팅<br/>LLM 호출 3]
D --> E[최종 출력]
style A fill:#e1f5fe
style E fill:#c8e6c9
style B fill:#fff3e0
style C fill:#fff3e0
style D fill:#fff3e0실행 순서:
- 사용자 질문 입력: "비트코인 가격 추세에 대해 어떻게 생각하나요?"
- 1차 LLM 호출: 질문의 의도를 분류합니다 (예: "analysis_request")
- 2차 LLM 호출: 분류된 의도에 맞춰 상세한 분석을 생성합니다
- 3차 LLM 호출: 생성된 분석을 사용자 친화적인 형식으로 포맷팅합니다
- 최종 상태: 모든 중간 결과와 최종 응답이 상태에 저장됩니다
왜 여러 단계로 나눌까요?
- 각 단계가 명확한 책임을 가져서 코드가 더 이해하기 쉬워집니다
- 단계별로 결과를 확인하고 디버깅하기 쉽습니다
- 필요에 따라 특정 단계만 수정하거나 교체할 수 있습니다
빠른 참조 가이드
자주 사용하는 API를 한눈에 확인할 수 있는 치트시트입니다:
| 구성 요소 | 용도 | 예제 |
|---|---|---|
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({...}) |
스마트 이코노미를 위한 퍼블릭 블록체인, 네오에 대한 모든것