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

pip install langgraph langchain-openai langchain-community

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]

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("서울 날씨 검색하고 화씨로 변환해줘")

# ❌ 나쁜 도구
@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
)
# → "김철수님이라고 하셨습니다." (이전 대화 기억)

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)