LangGraph로 AI 에이전트 워크플로우 구축하기

flowchart TD
    A[State 정의] --> B[Node 정의<br/>각 처리 단계]
    B --> C[Edge 연결<br/>조건부 or 직접]
    C --> D[Graph 컴파일]
    D --> E[실행]

    subgraph "State"
        S1[messages]
        S2[next_step]
        S3[results]
    end

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated
import operator

# 1. 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 메시지 누적
    next: str  # 다음 노드

# 2. LLM 설정
llm = ChatAnthropic(model="claude-sonnet-4-6-20251001")

# 3. 도구 정의
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    # 웹에서 정보를 검색합니다
    # 실제 검색 API 호출
    return f"검색 결과: {query}에 대한 정보..."

@tool
def calculate(expression: str) -> str:
    # 수식을 계산합니다
    try:
        result = eval(expression)
        return f"계산 결과: {result}"
    except Exception as e:
        return f"오류: {e}"

tools = [search_web, calculate]
llm_with_tools = llm.bind_tools(tools)

# 4. 노드 함수
def agent_node(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    results = []

    for tool_call in last_message.tool_calls:
        tool = {t.name: t for t in tools}[tool_call["name"]]
        result = tool.invoke(tool_call["args"])
        results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": results}

# 5. 조건부 엣지
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# 6. 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 도구 실행 후 다시 agent로

app = graph.compile()

# 실행
result = app.invoke({
    "messages": [HumanMessage(content="서울 날씨를 검색하고 화씨로 변환해줘")]
})
print(result["messages"][-1].content)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import interrupt_before

# 체크포인트 저장소
memory = MemorySaver()

def dangerous_action_node(state: AgentState):
    # DB 삭제, 이메일 전송 등 위험한 작업
    return {"messages": [AIMessage(content="작업 실행 완료")]}

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("dangerous_action", dangerous_action_node)
graph.set_entry_point("agent")

# 컴파일 시 interrupt 설정
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["dangerous_action"]  # 이 노드 전에 중단
)

# 첫 실행 (중단됨)
config = {"configurable": {"thread_id": "1"}}
result = app.invoke(
    {"messages": [HumanMessage(content="모든 임시 파일을 삭제해줘")]},
    config
)
# → "dangerous_action" 노드 전에 일시정지

print("현재 상태:", app.get_state(config).next)
# → ('dangerous_action',)

# 사용자 승인 후 재개
user_input = input("계속 진행하시겠습니까? (y/n): ")
if user_input == "y":
    result = app.invoke(None, config)  # None으로 재개

# 오케스트레이터 + 전문가 에이전트

class MultiAgentState(TypedDict):
    task: str
    research_result: str
    code_result: str
    final_answer: str

def orchestrator(state: MultiAgentState):
    # 작업을 분석해서 어떤 에이전트가 필요한지 결정
    response = llm.invoke(f'''다음 작업을 분석해서 필요한 에이전트를 결정해줘.
    작업: {state["task"]}
    응답: "research" 또는 "code" 또는 "both"''')
    return {"next": response.content.strip()}

def research_agent(state: MultiAgentState):
    result = research_llm.invoke(f"다음을 조사해줘: {state['task']}")
    return {"research_result": result.content}

def code_agent(state: MultiAgentState):
    context = state.get("research_result", "")
    result = code_llm.invoke(f"컨텍스트: {context}
작업: {state['task']}")
    return {"code_result": result.content}

def should_research_or_code(state):
    return state["next"]  # "research", "code", or "both"