실제 코드로 만들어보기

from typing import TypedDict

class SimpleRAGState(TypedDict):
    query: str
    results: list
    answer: str

def search_node(state: SimpleRAGState):
    # 사용자의 질문을 받음
    query = state["query"]

    # 실제 검색 수행 (여기서는 더미)
    results = ["문서1", "문서2", "문서3"]

    # State 수정
    state["results"] = results
    return state

def answer_node(state: SimpleRAGState):
    # 검색 결과를 받음
    results = state["results"]

    # LLM으로 답변 생성
    answer = "검색 결과를 바탕으로 만든 답변입니다"

    # State 수정
    state["answer"] = answer
    return state

from langgraph.graph import StateGraph

# Graph 생성
graph = StateGraph(SimpleRAGState)

# Node 추가 (이름, 함수)
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)

# Edge 추가 (어디에서 어디로)
graph.add_edge("search", "answer")

# 시작점 설정
graph.set_entry_point("search")

# 끝점 설정
graph.set_finish_point("answer")

# ============================================================
# 17차시 | LangGraph 입문
# 교안: 17차시_LangGraph입문.pptx
# 역할: 검색 노드와 답변 노드를 직선으로 잇는 가장 작은 그래프.
# 입력: State 키 query, results, answer
# 출력: results 와 answer 가 채워진 같은 State
#
#   State  = 노드들이 함께 쓰는 공용 딕셔너리. 공유 노트.
#            노드는 노트 전체를 다시 쓰지 않고, 고친 키만 return 한다.
#   Node   = State를 인자로 받는 함수.
#   Edge   = 다음에 실행할 노드. add_edge(지금, 다음).
#   START  = 시작 지점. END = 더 이상 갈 노드가 없음.
#   compile = 노드와 엣지를 묶어 실행 앱(app)으로 확정.
#   invoke  = 초기 State를 넣고 END까지 한 번에 실행. 최종 State를 돌려준다.
#
# 흐름
#   START
#     |
#     v
#   search      읽기 query / 쓰기 results
#     |
#     v
#   answer      읽기 results / 쓰기 answer
#     |
#     v
#    END
# ============================================================

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SimpleRAGState(TypedDict):
    query: str      # 사용자의 질문
    results: list   # 검색 결과 목록
    answer: str     # 최종 답변

# 노드 search
#   읽기: query
#   쓰기: results
def search_node(state: SimpleRAGState):
    # 현재 State에서 질문을 읽습니다.
    print("[검색 노드] 질문:", state["query"])

    # 실제 검색 대신 고정된 문서 목록을 준비합니다.
    results = [
        "문서1: 제품 가격은 10,000원입니다.",
        "문서2: 배송비는 무료입니다.",
        "문서3: 현재 할인 행사가 진행 중입니다."
    ]

    # LangGraph가 results 항목을 갱신하도록 반환합니다.
    return {"results": results}

# 노드 answer
#   읽기: results
#   쓰기: answer
def answer_node(state: SimpleRAGState):
    # 앞의 검색 노드가 반환한 결과를 읽습니다.
    print("[답변 노드] 검색 결과:", state["results"])

    # 실제 LLM 대신 고정된 답변을 준비합니다.
    answer = "검색 결과를 바탕으로 만든 답변입니다."

    # LangGraph가 answer 항목을 갱신하도록 반환합니다.
    return {"answer": answer}

# StateGraph(상태 타입) 로 빈 그래프를 만든다.
graph = StateGraph(SimpleRAGState)

# 노드 이름과 함수를 등록한다. 엣지는 이름이 아니라 이 이름을 가리킨다.
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)

# 엣지: START 다음 search, search 다음 answer, answer 다음 END.
graph.add_edge(START, "search")
graph.add_edge("search", "answer")
graph.add_edge("answer", END)

# compile: 위 연결을 실행 가능한 앱으로 확정한다.
app = graph.compile()

initial_state = {
    "query": "가격이 얼마야?",
    "results": [],
    "answer": ""
}

# invoke: 초기 State를 넣고 그래프를 끝까지 실행한다.
result = app.invoke(initial_state)

print("\n===== 최종 결과 =====")
print("질문:", result["query"])
print("검색 결과:", result["results"])
print("답변:", result["answer"])

# ============================================================
# 17차시 | LangGraph 입문
# 교안: 17차시_LangGraph입문.pptx
# 역할: 검색 결과가 있으면 답변으로, 없으면 검색으로 돌아가는 갈림길.
# 입력: State 키 query, results, answer
# 출력: 결과가 있으면 answer 가 채워진 State.
#       이 예제는 질문이 '가격'을 포함하지 않아 결과가 비고, 제한에서 멈춘다.
#
#   State  = 공유 노트(딕셔너리). 노드는 고친 키만 return 한다.
#   Node   = State를 읽고 쓰는 함수. 여기의 노드는 search, answer.
#   Edge   = 다음 노드가 정해진 연결.
#   조건부 엣지 = 함수가 State를 보고 다음에 갈 노드 이름을 고른다.
#            add_conditional_edges(출발노드, 고르는함수, {반환값: 다음노드})
#   START / END = 시작과 끝.
#   compile = 연결 확정. invoke = 초기 State로 끝까지 실행.
#   recursion_limit = 같은 구간을 너무 많이 돌면 멈추는 상한.
#
# 흐름
#   START
#     |
#     v
#   search ---- bad (results 가 빔) ----+
#     |                                  |
#     | good (results 가 있음)           |
#     v                                  |
#   answer                               |
#     |                                  v
#     v                               search (다시)
#    END
# ============================================================

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.errors import GraphRecursionError

class SimpleRAGState(TypedDict):
    query: str      # 사용자의 질문
    results: list   # 검색 결과 목록
    answer: str     # 최종 답변

# 노드 search
#   읽기: query
#   쓰기: results
def search_node(state: SimpleRAGState):
    query = state["query"]
    # 현재 State에서 질문을 읽습니다.
    print("[검색 노드] 질문:", state["query"])

    # 실제 검색 대신 고정된 문서 목록을 준비합니다.
    if "가격" in query:
        results = [
            "문서1: 제품 가격은 10,000원입니다.",
            "문서2: 배송비는 무료입니다.",
            "문서3: 현재 할인 행사가 진행 중입니다."
        ]
    else:
        results = []

    print("[검색] 결과:", results)

    # LangGraph가 results 항목을 갱신하도록 반환합니다.
    return {"results": results}

# 노드 answer
#   읽기: results
#   쓰기: answer
def answer_node(state: SimpleRAGState):
    # 앞의 검색 노드가 반환한 결과를 읽습니다.
    print("[답변 노드] 검색 결과:", state["results"])

    # 실제 LLM 대신 고정된 답변을 준비합니다.
    answer = "검색 결과를 바탕으로 만든 답변입니다."

    # LangGraph가 answer 항목을 갱신하도록 반환합니다.
    return {"answer": answer}

# 조건부 엣지 함수. 노드가 아니다. State를 고치지 않고 다음 이름만 반환한다.
#   읽기: results
#   쓰기: 없음
#   "good" → answer, "bad" → search
def check_search_quality(state: SimpleRAGState):
    # 검색 결과가 하나 이상이면 답변 경로를 선택합니다.
    if len(state["results"]) > 0:
        print("[판단] 검색 성공 → 답변")
        return "good"

    # 검색 결과가 없으면 재검색 경로를 선택합니다.
    print("[판단] 검색 실패 → 다시 검색")
    return "bad"

graph = StateGraph(SimpleRAGState)

graph.add_node("search", search_node)
graph.add_node("answer", answer_node)

graph.add_edge(START, "search")

# 조건부 엣지: search 가 끝난 뒤 check_search_quality 의 반환값으로 다음 노드를 고른다.
graph.add_conditional_edges(
    "search",
    check_search_quality,
    {
        "good": "answer",  # 성공하면 답변 노드로 이동
        "bad": "search"    # 실패하면 검색 노드로 돌아감
    }
)

graph.add_edge("answer", END)

# compile: 갈림길까지 포함해 실행 앱으로 확정한다.
app = graph.compile()

initial_state = {
    "query": "환불은 되니?",
    "results": [],
    "answer": ""
}

try:
    # invoke 두 번째 인자는 실행 설정. recursion_limit 은 반복 상한이다.
    result = app.invoke(initial_state, {"recursion_limit" : 10})

    print("\n===== 최종 결과 =====")
    print("질문:", result["query"])
    print("검색 결과:", result["results"])
    print("답변:", result["answer"])

except GraphRecursionError:
    print("검색이 반복되어 실행 제한에 도달했습니다.")

# Graph 컴파일 (준비하기)
runnable = graph.compile()

# Graph 실행
input_state = {"query": "가격이 얼마야?"}
output_state = runnable.invoke(input_state)

# 결과 보기
print("질문:", output_state["query"])
print("검색결과:", output_state["results"])
print("답변:", output_state["answer"])