[AI] LangGraph 의 conditional_edge 를 좀더 알아보자.

늘 공부하는 괴짜·2025년 5월 27일
0

AI : LangGraph

목록 보기
1/3

1. 간단한 본인인증 프로그램을 만들거다.

뭐... 이보다 훨씬 정교하게 만들 수는 있겠지만 구조를 이해하는게 중요하다.
llm 의 개입 없이 그냥 사용자입력과 반환되는 값만으로 컨트롤한다.

  • "본인인증" 이라고 전송하면 인증절차 시작
  • 이름을 입력. 아무거나 입력해도 되게 함
  • 휴대폰번호 입력. 형식에 맞게 입력해야 함. 틀리면 제자리 뱅뱅
  • 주민등록번호 입력. 6자리-7자리 에 맞게만 입력요구. 틀리면 제자리 뱅뱅

2. 전체 소스

각 도구를 실행하는 exec_tool에서 step 을 반환하는데 그 step 을 route_auth_exec_tool 이 받아서 다음 step 을 결정해 주는 모습이다.

import re
from typing import TypedDict, Optional
from pydantic import BaseModel

from langchain_community.vectorstores.faiss import FAISS
from langchain.chains import RetrievalQA
from langchain_core.tools import Tool as LangGraphTool
from langgraph.graph import StateGraph, END

chat_history = []
step = None

# ✅ 상태 정의
class AgentState(TypedDict):
    input: str
    output: Optional[str]
    chat_history: list
    step: Optional[str]  # ← 인증 단계 상태 추적


# ✅ 도구 입력 스키마
class AuthInput(BaseModel):
    auth: Optional[str] = None

class AuthNameInput(BaseModel):
    name: Optional[str] = None

class AuthPhoneInput(BaseModel):
    phone: Optional[str] = None

class AuthRrnInput(BaseModel):
    rrn: Optional[str] = None


# ✅ 도구 정의
def set_auth(keyword: Optional[str] = None):
    return "본인 인증을 위해 이름을 입력해주세요."

def set_auth_name(name: Optional[str] = None):
    return f"{name}님, 본인 인증을 위해 전화번호를 입력해주세요."

def set_auth_phone(phone: Optional[str] = None):
    if not phone or not re.match(r"^010[-]?\d{4}[-]?\d{4}$", phone):
        return {"result": False, "message": "유효한 전화번호 형식이 아닙니다. 예: 01012345678"}
    return {"result": True, "message": f"{phone}로 전화번호 인증이 완료되었습니다. 주민등록번호를 입력해주세요."}

def set_auth_rrn(rrn: Optional[str] = None):
    if not rrn or not re.match(r"^\d{6}-?\d{7}$", rrn):
        return {"result": False, "message": "⚠️ 주민등록번호 형식이 올바르지 않습니다. 예: 900101-1234567"}
    return {"result": True, "message": "✅ 본인 인증이 완료되었습니다. 감사합니다!"}


# ✅ LangGraph 도구 래핑
auth_tool = LangGraphTool.from_function(
    name="set_auth",
    description="본인인증 명령어를 입력받는 도구입니다.",
    func=set_auth,
    args_schema=AuthInput,
)

auth_name_tool = LangGraphTool.from_function(
    name="set_auth_name",
    description="이름을 받아서 저장하는 도구입니다.",
    func=set_auth_name,
    args_schema=AuthNameInput,
)

auth_phone_tool = LangGraphTool.from_function(
    name="set_auth_phone",
    description="전화번호를 받아서 저장하는 도구입니다.",
    func=set_auth_phone,
    args_schema=AuthPhoneInput,
)

auth_rrn_tool = LangGraphTool.from_function(
    name="set_auth_rrn",
    description="주민등록번호를 받아서 저장하는 도구입니다.",
    func=set_auth_rrn,
    args_schema=AuthRrnInput,
)

# ✅ 도구 실행 노드
def auth_exec_tool(state: AgentState):
    output = auth_tool.invoke({"auth": state["input"]})
    return {
        "input": "",
        "output": output,
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", output)],
        "step": "auth_name",  # 다음 단계로 이동
    }

def auth_name_exec_tool(state: AgentState):
    output = auth_name_tool.invoke({"name": state["input"]})
    return {
        "input": "",
        "output": output,
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", output)],
        "step": "auth_phone",  # 다음 단계로 이동
    }

def auth_phone_exec_tool(state: AgentState):
    result = auth_phone_tool.invoke({"phone": state["input"]})
    return {
        "input": "",
        "output": result["message"],
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", result["message"])],
        "step": "auth_rrn" if result["result"] else "auth_phone",
    }

def auth_rrn_exec_tool(state: AgentState):
    result = auth_rrn_tool.invoke({"rrn": state["input"]})
    return {
        "input": "",
        "output": result["message"],
        "chat_history": state["chat_history"] + [("user", state["input"]), ("ai", result["message"])],
        "step": None if result["result"] else "auth_rrn"
    }


# ✅ 라우팅 함수
def route_auth_exec_tool(state: AgentState):
    if state.get("step") == "auth_name":
        return "auth_name_exec_tool"
    elif state.get("step") == "auth_phone":
        return "auth_phone_exec_tool"
    elif state.get("step") == "auth_rrn":
        return "auth_rrn_exec_tool"
    elif state["input"] == "본인인증":
        return "auth_exec_tool"
    return "end"


# ✅ 라우터 노드
def router(state: AgentState) -> AgentState:
    return state

# ✅ LangGraph 구성
builder = StateGraph(AgentState)

builder.add_node("router", router)
builder.add_node("auth_exec_tool", auth_exec_tool)
builder.add_node("auth_name_exec_tool", auth_name_exec_tool)
builder.add_node("auth_phone_exec_tool", auth_phone_exec_tool)
builder.add_node("auth_rrn_exec_tool", auth_rrn_exec_tool)

builder.set_entry_point("router")

builder.add_conditional_edges("router", route_auth_exec_tool, {
    "auth_exec_tool": "auth_exec_tool",
    "auth_name_exec_tool": "auth_name_exec_tool",
    "auth_phone_exec_tool": "auth_phone_exec_tool",
    "auth_rrn_exec_tool": "auth_rrn_exec_tool",
    "end": END,
})

builder.add_edge("auth_exec_tool", END)
builder.add_edge("auth_name_exec_tool", END)
builder.add_edge("auth_phone_exec_tool", END)
builder.add_edge("auth_rrn_exec_tool", END)

graph = builder.compile()


try:
    png_data = graph.get_graph(xray=True).draw_mermaid_png()
    # 1) 파일 저장 후 출력
    with open("graph.png", "wb") as f:
        f.write(png_data)

    # 또는 2) 파일 없이 바로 출력
    # display(Image(data=png_data))
except Exception as e:
    print(f"그래프 시각화 오류: {e}")


# ✅ 대화 루프
print("🧠 LangGraph MCP Agent 시작! (종료: 'exit')")


while True:
    user_input = input("🧑‍💻 You: ")
    if user_input.lower() in ["exit", "quit"]:
        break

    state = {
        "input": user_input,
        "output": None,
        "chat_history": chat_history,
        "step": step,
    }

    try:
        result = graph.invoke(state)
        chat_history = result["chat_history"]
        step = result.get("step")
        print("🤖 AI:", result["output"])
    except Exception as e:
        print(f"❗ 오류 발생: {str(e)}")

3. graph.png

이런 그림 어디서 많이 본거 같은데... spring mvc 구조 보는듯 하다.

4. Test

Finally

이전에 고민했던 llm 의 개입 시기 문제를 적절하게 관리할 수 있고 llm의 응답의 형태에 따라서 업무 Step 을 결정할 수 있다.

아무래도 자유로운 채팅의 형태가 아닌 업무 Flow 를 공급하는 차원으로 도움을 많이 줄 수 있을 것 같다.

끗!

profile
인공지능이라는 옷을 입었습니다. 뭔가 멋지면서도 잘 맞습니다.

0개의 댓글