LangChain과 LangGraph를 활용한 AI Agent 개발 가이드

🧠 LangChain과 LangGraph를 활용한 AI 에이전트 구현 가이드

🧩 시나리오

"고객 데이터에서 VIP 고객을 선별하고, 각 고객에게 맞춤 이메일을 생성한 뒤, 이를 검토하고 발송 대기 리스트에 올려라."

✅ 전체 구성 흐름

[1] Goal 설정
 ↓
[2] LLM Reasoning → 어떤 고객이 VIP인지 판단
 ↓
[3] Tool 사용 → Google Sheets에서 고객 데이터 가져오기
 ↓
[4] Tool 사용 → OpenAI로 이메일 작성
 ↓
[5] Tool 사용 → 이메일 품질 검토 (Self-critique)
 ↓
[6] 저장 또는 전송 준비

🛠️ 사용 기술

• LangChain (에이전트 프레임워크)
• OpenAI GPT-4 API
• Google Sheets API (또는 임시 CSV)
• @tool 데코레이터를 통한 함수 등록
• Python

🧑‍💻 예제 코드 (LangChain Agent)

from langchain.agents import initialize_agent, Tool
from langchain.agents.agent_types import AgentType
from langchain.chat_models import ChatOpenAI

# Step 1: 도구 정의
def get_vip_customers():
    # 예제: 100만 원 이상 구매 고객만 추출
    return ["홍길동", "김철수"]

def write_email(name: str) -> str:
    return f"{name}님 안녕하세요. VIP 고객님께 감사 인사 드립니다!"

def review_email(content: str) -> str:
    # 간단한 예시: 너무 짧으면 수정 권장
    if len(content) < 50:
        return content + " 다시 한번 감사드립니다."
    return content

tools = [
    Tool(name="Get VIP Customers", func=get_vip_customers, description="VIP 고객 리스트를 가져온다."),
    Tool(name="Write Email", func=write_email, description="고객 이름으로 이메일을 생성한다."),
    Tool(name="Review Email", func=review_email, description="이메일 내용을 검토하고 개선한다."),
]

# Step 2: 에이전트 초기화
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent_executor = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)

# Step 3: 목표 지시
agent_executor.run("VIP 고객에게 보낼 감사 이메일을 작성하고 검토해줘.")

🔍 설명 요약

컴포넌트	설명
Tool	LLM이 사용할 수 있도록 등록된 기능 (ex. DB 조회, 이메일 작성)
initialize_agent	툴과 LLM을 연결하는 구성 함수
ZERO_SHOT_REACT_DESCRIPTION	툴 설명만으로 행동을 결정하는 에이전트 타입
agent_executor.run()	실제 실행 – 목표(Goal)를 LLM이 이해하고 툴을 선택 & 실행

---

🧠 LangGraph 기반 FSM AI 에이전트 구성 예제

💡 시나리오

"고객 데이터에서 VIP 고객을 선별하고,
각 고객에게 맞춤형 이메일을 작성한 뒤,
그 이메일을 AI가 리뷰하고,
최종 승인된 이메일만 발송 대기 리스트에 저장한다."

도구가 많아지고, 단계별로 결정 흐름이 분기되므로 FSM 구조가 적합합니다.

📌 구성 개요

LangGraph를 사용하면 FSM처럼 다음과 같은 형태로 구성됩니다:

[Start] → [GetVIPs] → [GenerateEmail] → [ReviewEmail]
      ↘︎        ↘︎             ↘︎            ↓
     [Error]   [Retry]       [Done] ← [RewriteEmail]

🛠 기술 스택

• LangGraph – 상태 그래프 정의
• ChatOpenAI – Reasoning 담당
• @tool 또는 LangChain Tool – 외부 기능 호출
• Python – FSM 정의 및 실행

🧑‍💻 예제 코드: LangGraph FSM 구성

from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatOpenAI
from typing import TypedDict, List

# 상태 정의
class AgentState(TypedDict):
    customers: List[str]
    email: str
    approved: bool

# LLM 초기화
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 상태 처리 함수들
def fetch_vips(state):
    # 예시: DB나 API에서 VIP 고객 추출
    return {"customers": ["홍길동", "김영희"]}

def generate_email(state):
    name = state["customers"][0]
    email = f"{name}님 안녕하세요. VIP 감사드립니다!"
    return {"email": email}

def review_email(state):
    email = state["email"]
    if "감사" not in email:
        return {"approved": False}
    return {"approved": True}

def rewrite_email(state):
    email = state["email"]
    return {"email": email + " 다시 한번 감사드립니다!"}

# 상태 그래프 정의
workflow = StateGraph(AgentState)

workflow.add_node("GetVIPs", fetch_vips)
workflow.add_node("GenerateEmail", generate_email)
workflow.add_node("ReviewEmail", review_email)
workflow.add_node("RewriteEmail", rewrite_email)

# 상태 전이 정의
workflow.set_entry_point("GetVIPs")
workflow.add_edge("GetVIPs", "GenerateEmail")
workflow.add_edge("GenerateEmail", "ReviewEmail")
workflow.add_conditional_edges(
    "ReviewEmail",
    {
        "approved": END,
        "default": "RewriteEmail"
    }
)
workflow.add_edge("RewriteEmail", "ReviewEmail")

# 실행
app = workflow.compile()
result = app.invoke({})
print(result)

🧩 구조 설명

구성 요소	역할
StateGraph()	전체 에이전트 흐름 정의
add_node()	각각의 함수(노드)를 FSM에 추가
add_edge()	다음 상태로의 전이 정의
add_conditional_edges()	상태 조건에 따라 분기
invoke()	그래프 실행 (state 입력 가능)

✅ 사용 장점

• 단계별 전이/분기/재시도 로직이 명확하게 보임
• 에이전트의 각 행동 상태가 추적 가능
• 추후에 Tool 사용 수 증가 시 확장성 우수