[LangChain] Tool Calling과 Agent 구현하기 — @tool, ReAct, Memory, LangGraph
by 우와한개발자
1. Tool Calling
1) 개념
- Tool Calling은 LLM이 외부 함수나 API를 호출할 수 있도록 하는 메커니즘이다.
- LLM은 단순히 답변을 생성하는 것뿐만 아니라 계산기, 검색 API, DB 조회, 날씨 API 같은 외부 도구를 사용할 수 있다.
- 이를 통해 최신 정보 조회, 계산, 파일 처리, 외부 시스템 연동처럼 실제 실행이 필요한 작업을 처리할 수 있다.
2) Tool Calling 동작 흐름
사용자 요청 → 도구 필요 여부 판단 → 도구 선택 → 인자 생성 → 도구 실행 → 결과 반영 → 최종 답변
(1) 함수 정의 전달
- 개발자는 LLM에게 사용 가능한 도구 목록을 스키마 형태로 제공한다.
- 도구 스키마에는 도구 이름, 설명, 파라미터, 반환 타입 등이 포함된다.
(2) 도구 선택 및 실행
- LLM은 사용자 요청을 분석하여 도구가 필요한지 판단한다.
- 필요한 경우 적절한 도구를 선택하고, 실행에 필요한 인자를 생성한다.
- 실제 함수가 실행되면 그 결과가 다시 LLM에게 전달된다.
(3) 최종 응답 생성
- LLM은 도구 실행 결과를 바탕으로 사용자에게 최종 답변을 생성한다.
2. Tool 정의 — @tool 데코레이터
1) 개념
- LangChain에서는 @tool 데코레이터를 사용하여 Python 함수를 Agent가 사용할 수 있는 도구로 등록할 수 있다.
- LLM은 함수 이름, 독스트링, 파라미터 정보를 참고하여 도구의 용도와 사용 방법을 판단한다.
2) 작성 시 주의점
- 독스트링에는 도구의 목적을 명확하게 작성한다.
- 파라미터가 무엇을 의미하는지도 함께 설명하는 것이 좋다.
- 파라미터 타입과 return 타입을 명시한다.
- 함수 이름은 도구의 목적이 드러나도록 작성한다.
from langchain.tools import tool
@tool
def calculator(expr: str) -> float:
"""수학 표현식을 계산합니다.
Args:
expr: 계산할 수학 표현식입니다. 예: '2 + 3 * 4'
Returns:
계산 결과를 float 타입으로 반환합니다.
"""
return eval(expr)
3. ReAct Framework
- ReAct는 Reasoning과 Acting을 결합한 Agent 실행 방식이다.
- LLM이 추론하고, 필요한 도구를 실행하고, 실행 결과를 관찰하면서 문제를 해결한다.
- 즉, Thought → Action → Observation 과정을 반복하며 최종 답변에 도달한다.
사용자: 서울의 현재 날씨를 알려줘.
Thought: 현재 날씨는 실시간 정보이므로 날씨 API가 필요하다.
Action: weather_tool(location="서울")
Observation: 서울의 현재 기온은 27도이고 흐림 상태이다.
Final Answer: 현재 서울은 흐리고 기온은 약 27도입니다.
4. Agent 실행 구조
1) Agent Executor
- LLM과 도구를 연결하고, Agent의 실행 루프를 관리하는 엔진
- 사용자 입력 → LLM 판단 → 도구 실행 → 결과 반영 → 최종 답변 흐름을 자동으로 처리한다
2) Agent 생성 흐름
Tool 정의 → Tool 등록 → LLM 연결 → Agent 생성 → 사용자 입력 → 도구 실행 → 최종 답변
5. Memory
1) 개념
- Agent가 이전 대화 내용을 기억하고, 이를 바탕으로 일관된 응답을 생성할 수 있도록 하는 기능
- 이전 질문, 답변, 도구 실행 결과 등을 저장하여 다음 응답에 활용한다.
2) 종류
(1) 단기 기억
- 현재 세션의 대화 이력을 유지하는 방식이다.
- 일반적으로 Message History를 사용한다.
- 사용자가 이전에 말한 내용을 바탕으로 후속 질문을 이해할 수 있다.
(2) 장기 기억
- 여러 세션에 걸쳐 유지되는 정보를 저장하는 방식이다.
- 벡터 데이터베이스나 일반 DB 등을 활용할 수 있다.
- 사용자 선호, 과거 작업 결과, 프로젝트 정보 등을 저장할 때 사용한다.
3) Message History
- 사용자와 Agent 간의 메시지 교환 기록
- 사용자 메시지, AI 응답, Tool 호출 결과 등을 포함할 수 있다.
- LLM은 context window에 제한이 있어 대화가 길어지면 모든 메시지를 전달하기 어렵기 때문에 오래된 메시지를 제거하거나 요약하는 등 Message History 관리 전략이 필요하다
6. LangGraph에서의 Memory 관리
1) InMemorySaver
- LangGraph에서 Agent의 상태를 메모리에 저장하는 Checkpointer이다
- 서버가 재시작되면 저장된 상태는 사라지므로 운영 환경에서는 DB 기반 저장소를 사용하는 것이 좋다
2) thread_id
- 대화 세션을 구분하기 위한 식별자이다
- 같은 thread_id를 사용하면 이전 대화 상태를 이어갈 수 있다
config = {
"configurable": {
"thread_id": "user-1"
}
}
result = agent.invoke(
{"messages": [("user", user_input)]},
config=config
)
7. LangChain LangGraph Agent 구현 예시
- Streamlit으로 챗봇 UI 구성
- @tool로 multiply, weather 도구 정의
- create_agent로 LLM과 Tool 연결
- InMemorySaver로 대화 상태 저장
- thread_id 기준으로 대화 흐름 유지
import os
import requests
import streamlit as st
from dotenv import load_dotenv
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
load_dotenv()
@tool
def multiply(a: int, b: int) -> int:
"""두 수를 곱하여 계산한다.
Args:
a: 첫 번째 숫자
b: 두 번째 숫자
Returns:
두 수를 곱한 결과
"""
print(f"multiply({a}, {b}) tool 호출됨")
return a * b
@tool
def weather(city: str) -> str:
"""
도시의 현재 날씨를 조회한다.
Args:
city: 영문 도시명. 예: Seoul, Busan, Tokyo, London
Returns:
현재 날씨, 기온, 체감온도, 습도 정보
"""
api_key = os.getenv("WEATHER_API_KEY")
url = "<https://api.openweathermap.org/data/2.5/weather>"
params = {
"q": city,
"appid": api_key,
"units": "metric",
"lang": "kr"
}
response = requests.get(url, params=params, timeout=10)
data = response.json()
temp = data["main"]["temp"]
weather_desc = data["weather"][0]["description"]
feels_like = data["main"]["feels_like"]
humidity = data["main"]["humidity"]
return (
f"{city}의 현재 날씨는 {weather_desc}, "
f"기온은 {temp}도, 체감온도는 {feels_like}도, "
f"습도는 {humidity}%입니다."
)
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0
)
memory = InMemorySaver()
agent = create_agent(
model=llm,
tools=[multiply, weather],
checkpointer=memory
)
config = {
"configurable": {
"thread_id": "user1"
}
}
st.title("날씨 챗봇")
# 화면 출력용 메시지
if "ui_messages" not in st.session_state:
st.session_state.ui_messages = []
# 이전 채팅 다시 그리기
for message in st.session_state.ui_messages:
with st.chat_message(message["role"]):
st.write(message["content"])
user_input = st.chat_input("질문을 입력하세요")
if user_input:
# UI에 사용자 메시지 저장
st.session_state.ui_messages.append({
"role": "user",
"content": user_input
})
with st.chat_message("user"):
st.write(user_input)
# Agent에게는 이번 입력만 전달한다.
# 이전 대화 상태는 InMemorySaver와 thread_id를 통해 관리된다.
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": user_input
}
]
},
config=config
)
answer = result["messages"][-1].content
# UI에 assistant 답변 저장
st.session_state.ui_messages.append({
"role": "assistant",
"content": answer
})
with st.chat_message("assistant"):
st.write(answer)
'AI > LangChain' 카테고리의 다른 글
블로그의 정보
우와한개발자 님의 블로그
우와한개발자