도구 설계와 LangGraph — 실제로 움직이는 에이전트 만들기

LangGraph란?

LangGraph는 에이전트의 실행 흐름을 상태(State) 기반 그래프로 표현합니다.

flowchart LR
    subgraph COMPARE["LangChain vs LangGraph"]
        direction TB
        LC["LangChain\nA → B → C\n선형, 단방향"]
        LG["LangGraph\nA ⇄ B → C / D\n분기·루프·조건 가능"]
    end

에이전트는 본질적으로 "결과를 보고 다음 행동을 결정하는 루프"이므로 LangGraph가 적합합니다.

설치

pip install langgraph langchain-openai langchain-community

도구(Tool) 설계

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다. 최신 정보나 사실 확인이 필요할 때 사용합니다."""
    # 실제 구현에서는 Tavily, SerpAPI 등 사용
    return f"'{query}' 검색 결과: [검색 결과 내용]"

@tool
def calculate(expression: str) -> str:
    """수학 계산을 수행합니다. 예: '25 * 4 + 100'"""
    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"계산 오류: {e}"

@tool
def save_file(filename: str, content: str) -> str:
    """결과를 파일로 저장합니다."""
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"파일 저장 완료: {filename}"

tools = [search_web, calculate, save_file]

@tool 데코레이터가 함수의 독스트링을 LLM이 읽는 도구 설명으로 사용합니다. 명확한 설명이 중요합니다.

LangGraph 에이전트 구현

flowchart TB
    START(["시작"]) --> AGENT["에이전트 노드\nLLM이 도구 선택"]
    AGENT -->|"도구 호출"| TOOLS["도구 노드\n실제 함수 실행"]
    TOOLS -->|"결과 반환"| AGENT
    AGENT -->|"완료 판단"| END(["종료"])

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI

# 1. 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 2. LLM + 도구 바인딩
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 3. 에이전트 노드: LLM이 판단
def agent_node(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 4. 그래프 구성
graph_builder = StateGraph(AgentState)

graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", ToolNode(tools))  # 도구 실행 노드

graph_builder.set_entry_point("agent")

# 조건부 엣지: 도구 호출 여부에 따라 분기
graph_builder.add_conditional_edges(
    "agent",
    tools_condition,  # tool_calls 있으면 tools, 없으면 END
)
graph_builder.add_edge("tools", "agent")  # 도구 실행 후 에이전트로 복귀

agent = graph_builder.compile()

에이전트 실행

from langchain_core.messages import HumanMessage

def run_agent(question: str):
    result = agent.invoke({
        "messages": [HumanMessage(content=question)]
    })
    return result["messages"][-1].content

# 테스트
print(run_agent("123 * 456을 계산하고 결과를 result.txt에 저장해줘"))

실행 흐름:

[에이전트] 계산이 필요하다 → calculate("123 * 456") 호출
[도구] 123 * 456 = 56088 반환
[에이전트] 파일 저장이 필요하다 → save_file("result.txt", "56088") 호출
[도구] 파일 저장 완료 반환
[에이전트] 모든 작업 완료 → 최종 답변 생성

스트리밍으로 실행 과정 보기

def run_agent_verbose(question: str):
    for event in agent.stream(
        {"messages": [HumanMessage(content=question)]},
        stream_mode="values"
    ):
        last_message = event["messages"][-1]
        
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            for tc in last_message.tool_calls:
                print(f"🔧 도구 호출: {tc['name']}({tc['args']})")
        elif last_message.type == "tool":
            print(f"📦 결과: {last_message.content[:100]}")
        else:
            print(f"💬 답변: {last_message.content}")

run_agent_verbose("서울 날씨 검색하고 화씨로 변환해줘")

도구 설계 원칙

flowchart TB
    subgraph PRINCIPLES["좋은 도구 설계 원칙"]
        P1["단일 책임\n하나의 도구 = 하나의 기능"]
        P2["명확한 독스트링\n언제 사용하는지 명시"]
        P3["에러 처리\n실패 시 유용한 메시지 반환"]
        P4["타입 힌트\n파라미터 타입 명확히"]
    end

# ❌ 나쁜 도구
@tool
def do_stuff(input: str) -> str:
    """작업을 수행합니다."""
    pass

# ✅ 좋은 도구
@tool
def get_stock_price(symbol: str) -> str:
    """주식 티커 심볼로 현재 주가를 조회합니다.
    
    Args:
        symbol: 주식 티커 (예: 'AAPL', 'MSFT', '005930.KS')
    
    Returns:
        현재 주가와 등락률 정보
    """
    try:
        # 실제 API 호출
        return f"{symbol}: $150.00 (+1.5%)"
    except Exception as e:
        return f"주가 조회 실패 ({symbol}): {str(e)}"

체크포인트: 중간 상태 저장

장시간 실행되는 에이전트는 중간 상태를 저장해야 합니다.

from langgraph.checkpoint.memory import MemorySaver

# 메모리 체크포인터 추가
checkpointer = MemorySaver()
agent = graph_builder.compile(checkpointer=checkpointer)

# thread_id로 이어서 실행 가능
config = {"configurable": {"thread_id": "session-1"}}

result1 = agent.invoke(
    {"messages": [HumanMessage("내 이름은 김철수야")]},
    config=config
)

result2 = agent.invoke(
    {"messages": [HumanMessage("내 이름이 뭐야?")]},
    config=config
)
# → "김철수님이라고 하셨습니다." (이전 대화 기억)

실전 도구 연동: Tavily 웹 검색

from langchain_community.tools import TavilySearchResults

# pip install tavily-python
# TAVILY_API_KEY 환경변수 필요 (무료 티어 있음)
search_tool = TavilySearchResults(max_results=3)

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

정리

개념	내용
`@tool` 데코레이터	함수를 LLM이 호출 가능한 도구로 변환
`AgentState`	에이전트가 유지하는 상태 (메시지 등)
`ToolNode`	도구를 실행하는 LangGraph 노드
`tools_condition`	도구 호출 여부에 따른 조건부 엣지
`MemorySaver`	세션 간 상태 저장 체크포인터

다음 편에서는 메모리와 상태 관리 — 에이전트가 장기적으로 정보를 기억하고 활용하는 방법을 배웁니다.