LangGraph 멀티에이전트 챗봇 설계와 프로덕션 운영 가이드

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str
    user_intent: str
    tool_results: dict
    escalation_needed: bool

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# 전문 에이전트 정의
order_agent = create_react_agent(
    model=model,
    tools=[search_order, cancel_order, track_shipping],
    name="order_expert",
    prompt="주문 관련 문의만 처리합니다. 주문번호가 없으면 반드시 먼저 확인하세요."
)

tech_agent = create_react_agent(
    model=model,
    tools=[search_docs, run_diagnostic, create_ticket],
    name="tech_expert",
    prompt="기술 지원 문의를 처리합니다. 문제 재현 단계를 먼저 파악하세요."
)

refund_agent = create_react_agent(
    model=model,
    tools=[check_refund_policy, process_refund, send_confirmation],
    name="refund_expert",
    prompt="환불/반품 처리 전담. 반드시 환불 정책을 먼저 확인하세요."
)

def supervisor_router(state: AgentState) -> str:
    """Supervisor가 의도를 분류하여 적절한 에이전트로 라우팅"""
    intent = state.get("user_intent", "")
    if "주문" in intent or "배송" in intent:
        return "order_expert"
    elif "오류" in intent or "버그" in intent or "설정" in intent:
        return "tech_expert"
    elif "환불" in intent or "반품" in intent:
        return "refund_expert"
    return "fallback"

# 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("order_expert", order_agent)
graph.add_node("tech_expert", tech_agent)
graph.add_node("refund_expert", refund_agent)
graph.add_node("fallback", fallback_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", supervisor_router)
graph.add_edge("order_expert", END)
graph.add_edge("tech_expert", END)
graph.add_edge("refund_expert", END)
graph.add_edge("fallback", END)

app = graph.compile()

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL 기반 체크포인터 설정
DB_URI = "postgresql://user:password@localhost:5432/langgraph_checkpoints"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 테이블 자동 생성

    app = graph.compile(checkpointer=checkpointer)

    # thread_id로 대화 세션 구분
    config = {"configurable": {"thread_id": "user-session-abc123"}}

    # 실행: 체크포인트가 각 노드 실행 후 자동 저장
    result = app.invoke(
        {"messages": [{"role": "user", "content": "주문 12345 배송 상태 알려줘"}]},
        config=config
    )

    # 장애 후 복구: 동일 thread_id로 자동 재개
    resumed = app.invoke(
        {"messages": [{"role": "user", "content": "환불도 가능해?"}]},
        config=config  # 이전 대화 컨텍스트 자동 로드
    )

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class OrderSearchInput(BaseModel):
    order_id: str = Field(description="주문번호 (예: ORD-2026-12345)")

@tool(args_schema=OrderSearchInput)
def search_order(order_id: str) -> dict:
    """주문 정보를 조회합니다. 주문번호 형식: ORD-YYYY-NNNNN"""
    if not order_id.startswith("ORD-"):
        return {"error": "유효하지 않은 주문번호 형식입니다"}
    # DB 조회 로직
    return db.orders.find_one({"order_id": order_id})

@tool
def process_refund(order_id: str, reason: str, amount: float) -> dict:
    """환불을 처리합니다. 환불 정책 확인 후에만 호출하세요."""
    if amount > 1_000_000:
        return {"error": "100만원 초과 환불은 수동 승인이 필요합니다", "escalate": True}
    # 환불 처리 로직
    return payment_gateway.refund(order_id, amount, reason)

# 에이전트별 도구 할당 (격리)
ORDER_TOOLS = [search_order, track_shipping, cancel_order]
REFUND_TOOLS = [check_refund_policy, process_refund, send_confirmation]
TECH_TOOLS = [search_docs, run_diagnostic, create_ticket]
# 주문 에이전트는 환불 도구에 접근 불가, 환불 에이전트는 주문 취소 도구에 접근 불가

from langchain_core.messages import HumanMessage

def input_guardrail_node(state: AgentState) -> AgentState:
    """입력 가드레일: prompt injection, PII, 부적절한 콘텐츠 탐지"""
    last_message = state["messages"][-1]

    # 1. 입력 길이 제한
    if len(last_message.content) > 4000:
        return {
            **state,
            "messages": state["messages"] + [
                {"role": "assistant", "content": "메시지가 너무 깁니다. 4000자 이내로 줄여주세요."}
            ],
            "escalation_needed": False
        }

    # 2. Prompt injection 탐지
    injection_patterns = [
        "시스템 프롬프트를 무시",
        "ignore previous instructions",
        "you are now",
        "new instructions:",
    ]
    content_lower = last_message.content.lower()
    for pattern in injection_patterns:
        if pattern.lower() in content_lower:
            return {
                **state,
                "messages": state["messages"] + [
                    {"role": "assistant", "content": "요청을 처리할 수 없습니다."}
                ],
                "escalation_needed": True
            }

    # 3. PII 마스킹 (주민등록번호, 카드번호 등)
    import re
    masked_content = re.sub(
        r'\d{6}-\d{7}', '[주민번호 마스킹]', last_message.content
    )
    masked_content = re.sub(
        r'\d{4}-\d{4}-\d{4}-\d{4}', '[카드번호 마스킹]', masked_content
    )

    if masked_content != last_message.content:
        state["messages"][-1] = HumanMessage(content=masked_content)

    return state

# 그래프에 가드레일 노드 추가
graph.add_node("input_guardrail", input_guardrail_node)
graph.add_edge(START, "input_guardrail")
graph.add_conditional_edges("input_guardrail", supervisor_router)

from langgraph.types import interrupt

def refund_approval_node(state: AgentState) -> AgentState:
    """환불 금액이 50만원 이상이면 관리자 승인을 요청"""
    refund_amount = state["tool_results"].get("refund_amount", 0)

    if refund_amount >= 500_000:
        # 그래프 실행을 여기서 중단 - 사람의 승인을 기다림
        approval = interrupt({
            "type": "refund_approval",
            "amount": refund_amount,
            "order_id": state["tool_results"]["order_id"],
            "message": f"환불 금액 {refund_amount:,}원. 승인하시겠습니까?"
        })

        if approval.get("approved"):
            return {**state, "escalation_needed": False}
        else:
            return {
                **state,
                "messages": state["messages"] + [
                    {"role": "assistant",
                     "content": "환불 요청이 관리자에 의해 거절되었습니다. 고객센터로 문의해주세요."}
                ]
            }

    return state

# 최대 25번의 노드 실행 후 강제 종료
app = graph.compile(checkpointer=checkpointer)
result = app.invoke(
    {"messages": [{"role": "user", "content": "문의 내용"}]},
    config={"configurable": {"thread_id": "session-1"}, "recursion_limit": 25}
)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError))
)
def call_external_api(endpoint: str, payload: dict) -> dict:
    """외부 API 호출 with exponential backoff"""
    response = httpx.post(
        endpoint,
        json=payload,
        timeout=10.0
    )
    response.raise_for_status()
    return response.json()

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.postgres import PostgresSaver
import uvicorn

app_api = FastAPI(title="Multi-Agent Chatbot API")

# 그래프 컴파일 (앱 시작 시 1회)
DB_URI = "postgresql://user:password@db:5432/langgraph"
checkpointer = PostgresSaver.from_conn_string(DB_URI)
checkpointer.setup()
chatbot_graph = build_multi_agent_graph()  # 위에서 정의한 그래프
compiled_app = chatbot_graph.compile(checkpointer=checkpointer)

class ChatRequest(BaseModel):
    session_id: str
    message: str

class ChatResponse(BaseModel):
    session_id: str
    reply: str
    agent_used: str

@app_api.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        config = {
            "configurable": {"thread_id": request.session_id},
            "recursion_limit": 25
        }
        result = compiled_app.invoke(
            {"messages": [{"role": "user", "content": request.message}]},
            config=config
        )
        last_ai_message = result["messages"][-1]
        return ChatResponse(
            session_id=request.session_id,
            reply=last_ai_message.content,
            agent_used=result.get("current_agent", "unknown")
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    uvicorn.run(app_api, host="0.0.0.0", port=8000)