"고객 메일을 받으면 문의 종류를 분류하고 지식 창고에서 근거를 찾아 답변 초안을 쓰고 근거가 없으면 사람에게 넘긴다." 이런 요구사항 문장을 받으면 어디서부터 그래프를 그려야 할지 막막합니다. 방법은 단순합니다. 문장에서 동사에 밑줄을 긋고 명사에는 따로 밑줄을 긋습니다.
동사는 일을 하는 노드가 되고 명사는 노드끼리 주고받는 State의 칸이 됩니다. 아래에서 단계를 눌러 문장이 그래프로 바뀌는 모습을 보세요.
이 글 끝의 미니 프로젝트(이메일 문의 답변 초안)를 설계하는 과정입니다. 예시 요구사항이고 이름은 직접 정했습니다.
칸을 정했으면 표를 하나 그려 봅니다. 칸마다 그 칸을 쓰는 노드와 읽는 노드를 찾는 표입니다. 쓰는 곳이 없는 칸은 입력으로 받아야 하는 값이고 읽는 곳이 없는 칸은 지워도 됩니다.
모든 정보를 State에 넣고 싶어지지만 칸이 늘수록 어느 노드가 무엇을 바꾸는지 따라가기 어려워집니다. 노드끼리 주고받는 값만 칸으로 만들고 한 노드 안에서만 쓰는 값은 지역 변수로 둡니다. 칸을 하나 더 만들어 표가 어떻게 달라지는지 보세요.
검색 서버가 잠깐 응답하지 않는 경우, 검색했더니 근거가 없는 경우, 코드가 잘못된 경우는 서로 다른 실패입니다. 앞의 둘은 정상 운영에서도 일어날 수 있는 일이고 마지막은 고쳐야 하는 오류입니다. 다루는 방법도 다릅니다.
잠깐의 오류는 같은 일을 다시 해 보면 낫습니다. 노드에 RetryPolicy를 붙이면 지정한 오류에 한해 간격을 늘려 가며 다시 시도합니다. 예상한 실패는 다른 길로 보냅니다. 실패 종류를 눌러 보세요.
policy = RetryPolicy(max_attempts=3, initial_interval=0.1,
backoff_factor=2.0, retry_on=ConnectionError) # 1 규칙
b.add_node("search", search, retry_policy=policy) # 2 노드에 붙임
# 3 예상한 실패는 오류가 아니라 길로 푼다
def has_docs(s: State) -> Literal["draft", "to_human"]:
return "draft" if s["docs"] else "to_human"출력은 예시입니다. 시도 횟수는 langgraph 1.2로 직접 실행해 확인했습니다.
max_attempts=3은 처음 시도를 포함해 모두 3번입니다. 간격은 initial_interval에서 시작해 backoff_factor배씩 늘어납니다. 기본으로 간격이 조금씩 흔들려서 여러 요청이 한꺼번에 몰리지 않습니다.retry_on에 적은 오류만 다시 합니다. 지정하지 않으면 기본 규칙이 적용되는데 ValueError 같은 코드 실수는 다시 하지 않았습니다. 끝내 안 되면 마지막 오류가 그대로 올라옵니다.docs가 비어 있는지 보고 길을 가르면 됩니다. 앞에서 배운 조건부 엣지입니다.그래프가 이상하게 동작하면 마지막 결과만 봐서는 어디서 틀어졌는지 알 수 없습니다. invoke 대신 stream을 쓰면 노드가 끝날 때마다 중간 결과를 받습니다. 받는 방식은 stream_mode로 고릅니다.
updates는 방금 끝난 노드가 바꾼 칸만 주고 values는 그 시점의 State 전체를 줍니다. 어느 노드가 무엇을 바꿨는지 볼 때는 앞의 것이 편하고 전체 모습이 필요할 때는 뒤의 것이 편합니다.
for ev in graph.stream(inputs, stream_mode="updates"): # 1 바뀐 칸만
print("update", ev)
for st in graph.stream(inputs, stream_mode="values"): # 2 State 전체
print("values", list(st.keys()))출력은 예시입니다. 노드 이름과 칸 이름은 이 글의 미니 프로젝트 기준입니다.
{노드 이름: 그 노드가 바꾼 칸들} 모양입니다. 어느 노드가 어떤 값을 냈는지 한눈에 보입니다.지금까지 만든 조각을 모아 이메일 문의를 분류하고 근거를 찾아 답변 초안을 쓰는 그래프를 만들었습니다. 분류에는 구조화 출력을, 검색에는 재시도를 썼고 근거가 없을 때는 사람에게 넘기는 길로 보냅니다. 첫 검색 호출이 일부러 한 번 실패하게 해 두었는데 재시도가 그 실패를 받아 냈습니다.
class State(TypedDict): # 1 칸은 다섯 개
mail: str; category: str; docs: list[str]; draft: str; handoff: bool
b = StateGraph(State)
b.add_node("classify", classify) # 구조화 출력 분류
b.add_node("search", search, retry_policy=policy) # 2 재시도
b.add_node("draft", draft); b.add_node("to_human", to_human)
b.add_edge(START, "classify"); b.add_edge("classify", "search")
b.add_conditional_edges("search", has_docs) # 3 길 가르기
b.add_edge("draft", END); b.add_edge("to_human", END)
graph = b.compile()출력은 예시입니다. 분류와 답변은 모의 서버로 확인했고 실제 모델의 결과는 달라질 수 있습니다. 코드는 langgraph 1.2 기준이며 노드 함수는 일부 생략했습니다.
docs가 비었고 사람에게 넘기는 길로 갔습니다. 억지로 답을 지어내지 않습니다.다음 글부터는 이 그래프를 여러 Agent로 나눌 필요가 정말 있는지부터 따져 봅니다.