LangGraph 에이전트 워크플로우 실전 가이드: 멀티에이전트 오케스트레이션부터 프로덕션 배포까지

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

# 1) 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, "add_messages"]  # 메시지 누적
    next_action: str
    result: str

# 2) 노드 함수 정의
def classify_intent(state: AgentState) -> dict:
    """사용자 의도 분류"""
    last_msg = state["messages"][-1].content
    # LLM으로 의도 분류
    intent = llm.invoke(f"Classify intent: {last_msg}")
    return {"next_action": intent.content}

def handle_question(state: AgentState) -> dict:
    """질문 처리"""
    answer = llm.invoke(state["messages"])
    return {"result": answer.content}

def handle_task(state: AgentState) -> dict:
    """작업 실행"""
    result = agent_executor.invoke(state["messages"])
    return {"result": result}

# 3) 그래프 구축
graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("question", handle_question)
graph.add_node("task", handle_task)

# 4) 엣지 연결
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    lambda state: state["next_action"],
    {
        "question": "question",
        "task": "task",
    }
)
graph.add_edge("question", END)
graph.add_edge("task", END)

# 5) 컴파일 & 실행
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("What is Kubernetes?")]})

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_docs(query: str) -> str:
    """Search internal documentation."""
    # 벡터 DB 검색
    results = vectorstore.similarity_search(query, k=3)
    return "\n".join([r.page_content for r in results])

@tool
def run_query(sql: str) -> str:
    """Execute a read-only SQL query."""
    return db.execute(sql).fetchall()

@tool
def create_ticket(title: str, description: str) -> str:
    """Create a Jira ticket."""
    return jira.create_issue(title=title, description=description)

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

# ReAct 에이전트 자동 생성
agent = create_react_agent(
    llm,
    tools=[search_docs, run_query, create_ticket],
    state_modifier="You are a helpful DevOps assistant."
)

# 실행
result = agent.invoke({
    "messages": [HumanMessage("Check if order-service has errors and create a ticket")]
})

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

class SupervisorState(TypedDict):
    messages: Annotated[list, "add_messages"]
    next_agent: str
    final_answer: str

# 전문 에이전트들
researcher = create_react_agent(llm, tools=[search_web, search_docs])
coder = create_react_agent(llm, tools=[run_code, read_file])
reviewer = create_react_agent(llm, tools=[analyze_code, lint_code])

def supervisor(state: SupervisorState) -> dict:
    """어떤 에이전트에게 위임할지 결정"""
    response = llm.invoke([
        SystemMessage("You are a supervisor. Route to: researcher, coder, reviewer, or FINISH"),
        *state["messages"]
    ])
    return {"next_agent": response.content.strip()}

def run_researcher(state):
    result = researcher.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_coder(state):
    result = coder.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_reviewer(state):
    result = reviewer.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

# 그래프 구성
workflow = StateGraph(SupervisorState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", run_researcher)
workflow.add_node("coder", run_coder)
workflow.add_node("reviewer", run_reviewer)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    lambda s: s["next_agent"],
    {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "FINISH": END,
    }
)
# 각 에이전트 실행 후 supervisor로 복귀
for agent_name in ["researcher", "coder", "reviewer"]:
    workflow.add_edge(agent_name, "supervisor")

app = workflow.compile()

from langgraph.types import interrupt, Command

def sensitive_action(state):
    """민감한 작업 전 인간 승인 요청"""
    action = state["pending_action"]

    # 실행을 중단하고 인간 승인 대기
    approval = interrupt({
        "question": f"Approve this action?\n{action}",
        "options": ["approve", "reject", "modify"]
    })

    if approval == "approve":
        return execute_action(action)
    elif approval == "reject":
        return {"result": "Action cancelled by user"}
    else:
        return {"result": f"Action modified: {approval}"}

# 체크포인터와 함께 컴파일
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 실행 (interrupt에서 멈춤)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)
# result = {"__interrupt__": {...}}

# 인간이 승인 후 재개
app.invoke(Command(resume="approve"), config)

from langgraph.graph import StateGraph

class ParallelState(TypedDict):
    query: str
    web_results: str
    db_results: str
    combined: str

def search_web_node(state):
    return {"web_results": search_web(state["query"])}

def search_db_node(state):
    return {"db_results": search_db(state["query"])}

def combine_results(state):
    combined = f"Web: {state['web_results']}\nDB: {state['db_results']}"
    return {"combined": combined}

graph = StateGraph(ParallelState)
graph.add_node("web", search_web_node)
graph.add_node("db", search_db_node)
graph.add_node("combine", combine_results)

# 병렬 실행: START → [web, db] → combine
graph.add_edge(START, "web")
graph.add_edge(START, "db")
graph.add_edge("web", "combine")
graph.add_edge("db", "combine")
graph.add_edge("combine", END)

app = graph.compile()

# 토큰 단위 스트리밍
async for event in app.astream_events(
    {"messages": [HumanMessage("Explain CQRS")]},
    version="v2"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

# 노드 단위 스트리밍
for chunk in app.stream({"messages": [HumanMessage("...")]}):
    for node_name, output in chunk.items():
        print(f"[{node_name}]: {output}")

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL 체크포인터 (프로덕션)
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

app = workflow.compile(checkpointer=checkpointer)

# 대화 지속 (thread_id로 구분)
config = {"configurable": {"thread_id": "user-456"}}

# 첫 번째 메시지
app.invoke({"messages": [HumanMessage("Hi")]}, config)

# 두 번째 메시지 (이전 대화 컨텍스트 유지)
app.invoke({"messages": [HumanMessage("What did I just say?")]}, config)

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:app"
  },
  "env": ".env"
}

# LangGraph CLI 설치
pip install langgraph-cli

# 로컬 테스트
langgraph dev

# Docker 빌드
langgraph build -t my-agent:latest

# LangGraph Cloud 배포
langgraph deploy --app my-agent