뭐... 이보다 훨씬 정교하게 만들 수는 있겠지만 구조를 이해하는게 중요하다.
llm 의 개입 없이 그냥 사용자입력과 반환되는 값만으로 컨트롤한다.
각 도구를 실행하는 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)}")
이런 그림 어디서 많이 본거 같은데... spring mvc 구조 보는듯 하다.
이전에 고민했던 llm 의 개입 시기 문제를 적절하게 관리할 수 있고 llm의 응답의 형태에 따라서 업무 Step 을 결정할 수 있다.
아무래도 자유로운 채팅의 형태가 아닌 업무 Flow 를 공급하는 차원으로 도움을 많이 줄 수 있을 것 같다.
끗!