이미지 로딩 중...
AI Generated
2025. 11. 8. · 2 Views
LangChain Agent 완벽 가이드 - AI 에이전트 활용법
LangChain을 활용한 AI 에이전트 구축의 핵심 개념을 실무 중심으로 배워봅니다. Tool, Agent Executor, ReAct 패턴 등 실제 프로젝트에 바로 적용 가능한 내용을 다룹니다.
목차
- LangChain Agent 개요
- Tool 정의하기
- Agent Executor
- ReAct 패턴
- Tool Calling
- Memory와 State
- Custom Tool 구현
- Error Handling
1. LangChain Agent 개요
시작하며
여러분이 챗봇을 만들다가 "사용자가 날씨를 물어보면 API를 호출하고, 계산이 필요하면 계산기를 사용하고, 검색이 필요하면 구글을 검색하는" 복잡한 로직을 구현해야 했던 적 있나요? 모든 경우의 수를 if-else로 처리하려다 코드가 스파게티처럼 엉키는 경험 말이죠.
이런 문제는 실제 AI 애플리케이션 개발에서 가장 흔하게 발생합니다. 사용자의 의도를 파악하고, 적절한 도구를 선택하고, 결과를 조합하는 과정을 수동으로 구현하면 유지보수가 거의 불가능해집니다.
바로 이럴 때 필요한 것이 LangChain Agent입니다. Agent는 LLM이 스스로 상황을 판단하고, 필요한 도구를 선택해서 실행하고, 결과를 종합하는 자율적인 시스템을 구축할 수 있게 해줍니다.
개요
간단히 말해서, LangChain Agent는 LLM에게 "뇌"뿐만 아니라 "손과 발"까지 제공하는 프레임워크입니다. LLM은 사고만 할 수 있지만, Agent는 실제 행동(API 호출, 데이터베이스 조회, 계산 등)까지 수행할 수 있습니다.
이것이 필요한 이유는 명확합니다. 실무에서는 단순히 텍스트를 생성하는 것을 넘어서, 실제 시스템과 상호작용하고, 데이터를 조회하고, 업무를 처리해야 합니다.
예를 들어, 고객 문의에 답변하면서 동시에 재고를 확인하고 주문을 생성하는 CS 봇 같은 경우에 매우 유용합니다. 기존에는 개발자가 모든 시나리오를 예측하고 코드로 구현했다면, 이제는 Agent가 상황에 맞게 스스로 판단하고 행동할 수 있습니다.
Agent의 핵심 특징은 세 가지입니다: 1) 자율성(Autonomy) - LLM이 스스로 다음 행동을 결정 2) 도구 사용(Tool Use) - 다양한 외부 도구를 활용 3) 반복적 추론(Iterative Reasoning) - 목표 달성까지 여러 단계를 거침. 이러한 특징들이 복잡한 업무를 자동화하고 AI를 실제 비즈니스에 통합하는 데 핵심적입니다.
코드 예제
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub
# LLM 모델 초기화
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 간단한 도구 정의
def get_weather(location: str) -> str:
# 실제로는 날씨 API 호출
return f"{location}의 현재 날씨는 맑음, 22도입니다."
# Tool 객체 생성
tools = [
Tool(
name="WeatherTool",
func=get_weather,
description="특정 지역의 현재 날씨를 조회합니다. 입력: 지역명"
)
]
# ReAct 프롬프트 가져오기
prompt = hub.pull("hwchase17/react")
# Agent 생성 및 실행
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 에이전트 실행
result = agent_executor.invoke({"input": "서울의 날씨를 알려주세요"})
print(result["output"])
설명
이것이 하는 일: 위 코드는 사용자의 질문을 받아 LLM이 스스로 판단해서 필요한 도구(여기서는 날씨 조회)를 선택하고 실행한 뒤 결과를 자연어로 답변하는 완전한 Agent 시스템을 구축합니다. 첫 번째로, ChatOpenAI와 Tool을 설정하는 부분은 Agent의 "뇌"와 "도구함"을 준비하는 과정입니다.
temperature=0으로 설정해서 일관된 판단을 하도록 하고, Tool 객체에는 도구의 이름, 실행할 함수, 그리고 가장 중요한 description을 명시합니다. 이 description이 LLM이 도구를 선택할 때 참고하는 유일한 정보이기 때문에 명확하게 작성하는 것이 핵심입니다.
그 다음으로, create_react_agent가 실행되면서 LLM, 도구, 프롬프트를 조합하여 실제 Agent 인스턴스를 생성합니다. 내부적으로는 ReAct(Reasoning + Acting) 패턴을 사용하는데, 이는 "생각(Thought) → 행동(Action) → 관찰(Observation)"을 반복하는 방식입니다.
AgentExecutor로 감싸는 이유는 이러한 반복 과정을 관리하고, 오류를 처리하고, 최대 반복 횟수를 제한하기 위함입니다. 마지막으로, invoke() 메서드가 호출되면 실제 마법이 일어납니다.
사용자 입력 "서울의 날씨를 알려주세요"가 들어오면, Agent는 1) 이 질문에 답하려면 날씨 정보가 필요하다고 판단 2) 사용 가능한 도구 중 WeatherTool이 적합하다고 선택 3) "서울"을 인자로 도구 실행 4) 결과를 받아서 자연스러운 문장으로 재구성하여 답변을 생성합니다. 여러분이 이 코드를 사용하면 복잡한 분기 로직 없이도 다양한 사용자 요청을 처리할 수 있습니다.
도구를 추가하면 자동으로 Agent가 그것을 활용할 수 있고, 새로운 시나리오에 대해 코드를 수정할 필요가 없습니다. verbose=True로 설정하면 Agent의 사고 과정을 실시간으로 볼 수 있어 디버깅과 최적화에도 유용합니다.
실무에서는 이를 기반으로 고객 서비스 자동화, 데이터 분석 어시스턴트, 개발 도구 통합 등 다양한 애플리케이션을 구축할 수 있습니다.
실전 팁
💡 verbose=True는 개발 단계에서 필수입니다. Agent가 어떤 도구를 왜 선택했는지 보면 프롬프트나 도구 설명을 개선할 포인트를 찾을 수 있습니다.
💡 Tool의 description은 최대한 구체적으로 작성하세요. "날씨 조회"보다 "특정 지역의 현재 날씨를 조회합니다. 입력: 지역명(한글 또는 영문)"이 훨씬 효과적입니다.
💡 temperature는 0 또는 낮은 값을 사용하세요. Agent는 창의성보다 일관성과 정확성이 중요하므로 동일한 입력에 동일한 도구를 선택해야 합니다.
💡 max_iterations를 설정해서 무한 루프를 방지하세요. AgentExecutor(agent=agent, tools=tools, max_iterations=5)처럼 제한을 두는 것이 안전합니다.
💡 실제 프로덕션에서는 Agent의 비용을 모니터링하세요. 여러 번의 LLM 호출이 발생하므로 캐싱이나 결과 재사용 전략이 필요할 수 있습니다.
2. Tool 정의하기
시작하며
여러분이 AI 비서를 만들면서 "데이터베이스 조회", "이메일 발송", "파일 업로드" 같은 실제 기능을 연결하고 싶었던 적 있나요? LLM은 똑똑하지만, 실제 시스템과 상호작용하는 방법을 모릅니다.
이런 문제는 LangChain을 처음 사용할 때 가장 먼저 부딪히는 벽입니다. LLM의 능력을 실제 비즈니스 로직과 연결하지 못하면 결국 단순한 챗봇에 머무르게 됩니다.
바로 이럴 때 필요한 것이 Tool 정의입니다. Tool은 Agent가 사용할 수 있는 "기능 카탈로그"를 만드는 것으로, Python 함수를 LLM이 이해하고 호출할 수 있는 형태로 변환해줍니다.
개요
간단히 말해서, Tool은 일반 Python 함수를 Agent가 사용할 수 있는 인터페이스로 감싸는 래퍼입니다. 함수에 이름, 설명, 입력 스키마를 추가하면 LLM이 그것을 "도구"로 인식합니다.
Tool이 필요한 이유는 LLM과 실제 코드 사이의 간극을 메우기 위함입니다. LLM은 텍스트만 이해하므로, 우리의 Python 함수가 무엇을 하는지, 어떤 인자가 필요한지를 자연어로 설명해야 합니다.
예를 들어, 데이터베이스에서 사용자 정보를 조회하는 함수가 있다면, LLM이 "이 도구는 사용자 ID로 정보를 조회한다"는 것을 알아야 적절한 상황에 사용할 수 있습니다. 기존에는 함수를 직접 호출하는 하드코딩된 로직을 작성했다면, 이제는 Tool로 등록하면 Agent가 상황에 맞게 자동으로 선택하고 호출할 수 있습니다.
Tool의 핵심 구성 요소는 세 가지입니다: 1) name - 도구의 고유 식별자 2) description - LLM이 도구를 이해하는 유일한 정보 3) func - 실제 실행될 Python 함수. 여기에 선택적으로 args_schema를 추가하면 입력 검증과 더 정확한 파라미터 전달이 가능합니다.
코드 예제
from langchain.tools import Tool, StructuredTool
from langchain.pydantic_v1 import BaseModel, Field
from typing import Optional
# 방법 1: 간단한 Tool 정의 (단일 문자열 입력)
def calculate_age(birth_year: str) -> str:
"""출생년도로 나이를 계산합니다."""
current_year = 2024
age = current_year - int(birth_year)
return f"현재 나이는 {age}세입니다."
simple_tool = Tool(
name="AgeCalculator",
func=calculate_age,
description="출생년도(4자리 숫자)를 입력받아 현재 나이를 계산합니다."
)
# 방법 2: 구조화된 Tool 정의 (복수 파라미터)
class UserQueryInput(BaseModel):
user_id: int = Field(description="조회할 사용자의 ID")
include_history: Optional[bool] = Field(default=False, description="구매 이력 포함 여부")
def query_user(user_id: int, include_history: bool = False) -> str:
"""사용자 정보를 조회합니다."""
user_info = f"사용자 {user_id}의 이름: 홍길동, 등급: VIP"
if include_history:
user_info += ", 최근 구매: 노트북"
return user_info
structured_tool = StructuredTool(
name="UserQuery",
func=query_user,
description="사용자 ID로 회원 정보를 조회합니다. 선택적으로 구매 이력도 포함할 수 있습니다.",
args_schema=UserQueryInput
)
설명
이것이 하는 일: 위 코드는 두 가지 방식으로 Tool을 정의하는 방법을 보여줍니다. 간단한 경우는 Tool 클래스를, 복잡한 파라미터가 필요한 경우는 StructuredTool을 사용합니다.
첫 번째 방법인 Tool 클래스는 단일 문자열 입력만 받는 간단한 함수에 적합합니다. calculate_age 함수는 출생년도 하나만 받아서 나이를 계산하는데, 이런 경우 Tool로 감싸면 됩니다.
중요한 점은 description이 매우 명확해야 한다는 것입니다. "나이를 계산한다"보다 "출생년도(4자리 숫자)를 입력받아 현재 나이를 계산합니다"가 LLM이 올바른 입력을 제공하는 데 훨씬 도움이 됩니다.
두 번째 방법인 StructuredTool은 여러 파라미터가 필요하거나 타입 검증이 중요한 경우 사용합니다. UserQueryInput이라는 Pydantic 모델로 입력 스키마를 정의하면, LLM이 각 파라미터의 의미와 타입을 정확히 이해할 수 있습니다.
Field의 description도 각 파라미터마다 설명을 추가할 수 있어서 더 정밀한 제어가 가능합니다. 예를 들어, include_history는 선택적 파라미터인데, LLM이 이를 이해하고 필요한 경우에만 True로 설정할 수 있습니다.
내부적으로는 Agent가 도구를 호출할 때, Tool 객체가 입력을 검증하고 func에 정의된 실제 Python 함수를 실행한 뒤 결과를 문자열로 반환합니다. 이 과정에서 args_schema가 있으면 자동으로 타입 변환과 검증이 이루어집니다.
여러분이 이 패턴을 사용하면 기존 비즈니스 로직을 거의 수정 없이 Agent에 통합할 수 있습니다. 데이터베이스 조회, API 호출, 파일 처리 등 어떤 Python 코드든 Tool로 만들 수 있고, description만 잘 작성하면 Agent가 알아서 적절한 상황에 사용합니다.
실무에서는 StructuredTool을 사용하는 것을 추천합니다. 타입 안정성이 높고, 복잡한 입력을 처리하기 쉬우며, IDE의 자동완성도 잘 작동합니다.
또한 Pydantic의 검증 기능으로 잘못된 입력을 사전에 차단할 수 있어 안정적입니다.
실전 팁
💡 description은 "무엇을 하는가"뿐만 아니라 "언제 사용해야 하는가"도 포함하세요. "재고가 부족한지 확인할 때 사용"처럼 상황을 명시하면 더 정확합니다.
💡 Tool의 반환값은 항상 문자열이어야 합니다. 객체나 리스트를 반환하면 LLM이 이해하지 못하므로 JSON 문자열이나 자연어로 변환하세요.
💡 args_schema를 사용하면 자동으로 입력 검증이 됩니다. user_id: int로 정의했는데 문자열이 들어오면 자동으로 에러를 발생시켜 잘못된 호출을 방지합니다.
💡 도구 이름은 명확하고 일관성 있게 작성하세요. "tool1", "func_a" 같은 이름보다 "UserQuery", "InventoryCheck"처럼 의미가 명확한 이름이 좋습니다.
💡 복잡한 로직은 도구 내부에서 처리하지 말고 별도 함수로 분리하세요. Tool은 얇은 래퍼로 유지하고 실제 비즈니스 로직은 다른 모듈에 두는 것이 테스트와 유지보수에 유리합니다.
3. Agent Executor
시작하며
여러분이 Agent를 만들었는데 "도구를 무한 반복 호출한다", "중간에 에러가 나면 멈춘다", "실행 과정을 추적할 수 없다" 같은 문제를 겪어본 적 있나요? Agent 자체는 판단만 할 뿐, 실제 실행을 관리하는 것은 별개의 문제입니다.
이런 문제는 Agent를 프로덕션에 배포할 때 반드시 마주치는 도전 과제입니다. Agent의 자율성은 강력하지만, 동시에 예측 불가능한 동작을 일으킬 수 있어서 안전장치가 필요합니다.
바로 이럴 때 필요한 것이 AgentExecutor입니다. Agent와 도구를 실행하고, 반복을 제어하고, 에러를 처리하고, 로그를 남기는 실행 엔진 역할을 합니다.
개요
간단히 말해서, AgentExecutor는 Agent의 "관리자"입니다. Agent가 무엇을 해야 할지 결정하면, AgentExecutor가 실제로 그것을 안전하게 실행하고 결과를 다시 Agent에게 전달하는 루프를 관리합니다.
이것이 필요한 이유는 실제 운영 환경에서 안정성과 제어가 필수적이기 때문입니다. Agent는 목표 달성까지 여러 단계를 거칠 수 있는데, 각 단계에서 오류가 발생할 수 있고, 무한 루프에 빠질 수 있으며, 예상치 못한 비용이 발생할 수 있습니다.
예를 들어, 고객 문의 처리 Agent가 무한히 같은 도구를 호출하면서 API 비용이 폭증하는 상황을 방지해야 합니다. 기존에는 Agent를 직접 호출하고 결과를 수동으로 처리했다면, 이제는 AgentExecutor가 전체 생명주기를 관리하면서 안전하고 예측 가능한 실행을 보장합니다.
AgentExecutor의 핵심 기능은 다섯 가지입니다: 1) 반복 관리 - max_iterations로 무한 루프 방지 2) 시간 제한 - max_execution_time으로 긴 실행 차단 3) 에러 처리 - handle_parsing_errors로 복구 가능한 오류 처리 4) 로깅 - verbose로 전체 과정 추적 5) 조기 종료 - early_stopping_method로 적절한 시점에 중단. 이러한 기능들이 Agent를 안전하게 프로덕션에 배포할 수 있게 만듭니다.
코드 예제
from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub
# 도구 정의
def search_db(query: str) -> str:
return f"'{query}'에 대한 검색 결과: 3건 발견"
tools = [Tool(name="DatabaseSearch", func=search_db,
description="데이터베이스에서 정보를 검색합니다.")]
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
# AgentExecutor 설정 - 안전장치 포함
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 실행 과정 로깅
max_iterations=5, # 최대 5번만 반복
max_execution_time=30, # 30초 제한
handle_parsing_errors=True, # 파싱 에러 자동 복구
early_stopping_method="generate" # 조기 종료 시 답변 생성
)
# 실행 및 결과 확인
try:
result = agent_executor.invoke({
"input": "사용자 정보를 검색하고 요약해주세요"
})
print(f"최종 답변: {result['output']}")
print(f"실행 단계: {result.get('intermediate_steps', [])}")
except Exception as e:
print(f"실행 중 오류 발생: {e}")
설명
이것이 하는 일: 위 코드는 Agent를 안전하게 실행할 수 있는 AgentExecutor를 구성하고, 다양한 안전장치를 설정하여 프로덕션 환경에 적합한 실행 환경을 만듭니다. 첫 번째로, AgentExecutor를 생성할 때 설정하는 파라미터들이 핵심입니다.
max_iterations=5는 Agent가 최대 5번의 "생각 → 행동 → 관찰" 사이클만 수행하도록 제한합니다. 만약 Agent가 목표를 달성하지 못하고 계속 반복하려 하면, 5번째에서 강제로 종료됩니다.
max_execution_time=30은 전체 실행 시간을 30초로 제한해서, 네트워크 지연이나 복잡한 도구 호출로 인해 무한정 대기하는 상황을 방지합니다. 그 다음으로, handle_parsing_errors=True는 매우 실용적인 기능입니다.
때때로 LLM이 도구 호출 형식을 잘못 생성할 수 있는데, 이 옵션을 켜면 AgentExecutor가 자동으로 에러를 감지하고 LLM에게 "형식이 잘못되었으니 다시 시도하세요"라고 피드백을 줍니다. 이를 통해 일시적인 파싱 오류로 전체 실행이 실패하는 것을 막을 수 있습니다.
early_stopping_method="generate"는 조기 종료 전략을 정의합니다. 만약 max_iterations에 도달했는데 아직 목표를 달성하지 못했다면, 그냥 멈추는 대신("force") Agent가 현재까지 수집한 정보로 최선의 답변을 생성하도록("generate") 합니다.
사용자 입장에서는 "반복 횟수 초과" 같은 에러보다 "확실하지 않지만 이런 정보를 찾았습니다" 같은 답변이 훨씬 낫습니다. invoke() 메서드를 호출하면 전체 실행 루프가 시작됩니다.
AgentExecutor는 1) Agent에게 다음 행동 물어보기 2) 도구 실행 3) 결과를 Agent에게 전달 4) 완료되었는지 확인을 반복합니다. verbose=True 덕분에 각 단계가 콘솔에 출력되어 디버깅이 쉽습니다.
여러분이 이 설정을 사용하면 Agent가 폭주하거나 예측 불가능하게 동작하는 것을 방지할 수 있습니다. 실무에서는 이러한 제한이 필수적인데, 특히 API 비용이 토큰 수에 비례하는 OpenAI 같은 서비스를 사용할 때 더욱 중요합니다.
반환되는 result 딕셔너리에는 output(최종 답변)뿐만 아니라 intermediate_steps(중간 단계 정보)도 포함되어, 나중에 분석하거나 로깅할 수 있습니다. 이를 통해 Agent가 어떤 도구를 얼마나 사용했는지, 어디서 시간이 많이 걸렸는지 등을 파악할 수 있습니다.
실전 팁
💡 개발 중에는 verbose=True를 켜서 Agent의 사고 과정을 확인하세요. 어떤 도구를 왜 선택했는지 보면 프롬프트 개선 포인트를 찾을 수 있습니다.
💡 max_iterations는 도구 개수 + 2 정도로 설정하는 것이 적절합니다. 너무 적으면 복잡한 작업을 못하고, 너무 많으면 비용이 증가합니다.
💡 프로덕션에서는 반드시 max_execution_time을 설정하세요. 외부 API 호출이 타임아웃되면 전체 Agent가 멈출 수 있습니다.
💡 intermediate_steps를 로깅하면 나중에 Agent의 행동 패턴을 분석할 수 있습니다. 어떤 도구가 자주 사용되는지, 어떤 순서로 호출되는지 파악하면 최적화 기회를 찾을 수 있습니다.
💡 return_intermediate_steps=True를 설정하면 결과에 모든 중간 단계가 포함됩니다. 디버깅이나 감사(audit) 목적으로 유용합니다.
4. ReAct 패턴
시작하며
여러분이 복잡한 문제를 해결할 때 "일단 생각하고 → 행동하고 → 결과를 보고 → 다시 생각하는" 과정을 반복하죠? AI Agent도 마찬가지로 한 번에 답을 내는 것보다 단계적으로 추론하고 행동하는 것이 더 효과적입니다.
이런 접근은 전통적인 프롬프트 엔지니어링의 한계를 극복합니다. "한 번의 LLM 호출로 모든 것을 해결하려는" 방식은 복잡한 작업에서 자주 실패합니다.
중간에 새로운 정보가 필요하거나, 이전 단계의 결과에 따라 다음 행동이 달라져야 하는 상황을 처리하지 못하기 때문입니다. 바로 이럴 때 필요한 것이 ReAct 패턴입니다.
Reasoning(추론)과 Acting(행동)을 결합하여, Agent가 생각하고, 행동하고, 관찰하는 과정을 반복하면서 점진적으로 목표에 도달하도록 합니다.
개요
간단히 말해서, ReAct는 "생각(Thought) → 행동(Action) → 관찰(Observation)" 사이클을 반복하는 프레임워크입니다. 단순히 답을 생성하는 대신, 각 단계에서 무엇을 할지 명시적으로 추론하고 그 결과를 다음 단계에 활용합니다.
ReAct가 필요한 이유는 복잡한 작업을 해결하는 데 있어 투명성과 제어 가능성을 제공하기 때문입니다. 전통적인 Chain of Thought는 생각만 연결하지만, ReAct는 실제 행동(도구 사용)과 결합합니다.
예를 들어, "지난 분기 매출 중 가장 많이 팔린 제품의 재고를 확인하라"는 요청은 여러 단계의 추론과 데이터 조회가 필요한데, ReAct는 이를 체계적으로 처리합니다. 기존의 단일 프롬프트 방식이 "블랙박스"처럼 작동했다면, ReAct는 각 단계를 명시적으로 드러내어 디버깅과 최적화가 가능합니다.
ReAct 패턴의 핵심 단계는 다섯 가지입니다: 1) Question - 사용자의 질문 이해 2) Thought - 다음에 무엇을 해야 할지 추론 3) Action - 구체적인 도구와 입력 결정 4) Observation - 도구 실행 결과 확인 5) 반복 또는 Final Answer - 충분한 정보를 얻을 때까지 2-4 반복, 그 후 최종 답변. 이러한 구조가 복잡한 다단계 작업을 체계적으로 처리할 수 있게 합니다.
코드 예제
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
# 다양한 도구 정의
def get_revenue(quarter: str) -> str:
revenues = {"Q1": "1200만원", "Q2": "1500만원", "Q3": "1800만원"}
return f"{quarter} 매출: {revenues.get(quarter, '데이터 없음')}"
def get_top_product(quarter: str) -> str:
products = {"Q1": "노트북", "Q2": "모니터", "Q3": "키보드"}
return f"{quarter} 최다 판매 제품: {products.get(quarter, '데이터 없음')}"
def check_inventory(product: str) -> str:
inventory = {"노트북": "15개", "모니터": "23개", "키보드": "47개"}
return f"{product} 재고: {inventory.get(product, '재고 없음')}"
tools = [
Tool(name="RevenueQuery", func=get_revenue, description="분기별 매출 조회"),
Tool(name="TopProductQuery", func=get_top_product, description="분기별 최다 판매 제품 조회"),
Tool(name="InventoryCheck", func=check_inventory, description="제품별 재고 확인")
]
# ReAct 프롬프트는 이미 최적화되어 있음
llm = ChatOpenAI(model="gpt-4", temperature=0)
react_prompt = hub.pull("hwchase17/react") # 표준 ReAct 프롬프트
agent = create_react_agent(llm, tools, react_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)
# 복잡한 다단계 질문
result = executor.invoke({
"input": "Q3 분기에 가장 많이 팔린 제품의 현재 재고를 알려주세요"
})
print(result["output"])
설명
이것이 하는 일: 위 코드는 ReAct 패턴을 사용하여 복잡한 다단계 질문을 해결하는 과정을 보여줍니다. 사용자가 "Q3에 가장 많이 팔린 제품의 재고"를 물어보면, Agent가 자동으로 1) Q3 최다 판매 제품 조회 2) 그 제품의 재고 확인이라는 두 단계를 수행합니다.
첫 번째로, hub.pull("hwchase17/react")로 가져오는 프롬프트가 ReAct의 핵심입니다. 이 프롬프트는 LLM에게 "각 단계마다 Thought(생각)를 명시적으로 작성하고, Action(도구 선택)과 Action Input(입력)을 구조화된 형식으로 제공하라"고 지시합니다.
예를 들어, LLM의 첫 번째 출력은 "Thought: Q3의 최다 판매 제품을 먼저 찾아야 한다. Action: TopProductQuery, Action Input: Q3" 형태가 됩니다.
그 다음으로, AgentExecutor가 이 출력을 파싱하여 TopProductQuery 도구를 "Q3"라는 입력으로 실행합니다. 결과는 "Q3 최다 판매 제품: 키보드"가 되고, 이것이 Observation으로 LLM에게 다시 전달됩니다.
이제 LLM은 이 정보를 바탕으로 두 번째 추론을 시작합니다: "Thought: 이제 키보드의 재고를 확인해야 한다. Action: InventoryCheck, Action Input: 키보드".
이 과정이 반복되면서 Agent는 점진적으로 목표에 접근합니다. 중요한 점은 각 단계가 이전 단계의 결과에 의존한다는 것입니다.
만약 한 번의 LLM 호출로 처리하려 했다면, LLM은 Q3의 최다 판매 제품이 무엇인지 모르는 상태에서 재고를 확인해야 하므로 불가능했을 것입니다. verbose=True 덕분에 콘솔에서 전체 사고 과정을 볼 수 있습니다.
실제 출력은 대략 다음과 같습니다: ``` > Entering new AgentExecutor chain... Thought: Q3의 최다 판매 제품을 먼저 알아야 한다.
Action: TopProductQuery Action Input: Q3 Observation: Q3 최다 판매 제품: 키보드 Thought: 이제 키보드의 재고를 확인하면 된다. Action: InventoryCheck Action Input: 키보드 Observation: 키보드 재고: 47개 Thought: 충분한 정보를 얻었다.
Final Answer: Q3 분기에 가장 많이 팔린 제품은 키보드이며, 현재 재고는 47개입니다. ``` 여러분이 이 패턴을 사용하면 복잡한 업무 흐름을 자동화할 수 있습니다.
"데이터 조회 → 분석 → 추가 조회 → 결론"처럼 여러 단계가 필요한 작업에서 각 단계를 명시적으로 추적하고 제어할 수 있습니다. 실무에서는 ReAct 패턴이 디버깅과 최적화에 큰 도움이 됩니다.
Agent가 잘못된 도구를 선택했다면 Thought를 보고 왜 그런 판단을 했는지 알 수 있고, 프롬프트나 도구 설명을 개선할 수 있습니다.
실전 팁
💡 verbose=True는 ReAct 패턴에서 특히 중요합니다. 각 Thought를 보면 Agent의 추론 과정을 이해하고 문제를 진단할 수 있습니다.
💡 도구가 많을수록 max_iterations를 늘려야 합니다. 5개 도구라면 최소 7-8번의 반복이 필요할 수 있습니다(각 도구 호출 + 최종 답변).
💡 ReAct 프롬프트를 커스터마이징할 수 있습니다. 특정 도메인에 맞게 "Thought" 부분에 추가 지침을 넣으면 더 정확한 추론이 가능합니다.
💡 도구 실행 결과는 가능한 한 구조화하세요. "성공" 같은 모호한 응답보다 "키보드 재고: 47개, 위치: 창고 A"처럼 구체적인 정보를 제공하면 다음 단계 추론이 정확해집니다.
💡 중간 단계에서 오류가 발생하면 Agent가 복구할 수 있도록 에러 메시지를 명확히 작성하세요. "에러"보다 "해당 분기 데이터가 없습니다. Q1, Q2, Q3만 사용 가능합니다"가 훨씬 유용합니다.
5. Tool Calling
시작하며
여러분이 LangChain Agent를 사용하면서 "LLM이 도구 호출 형식을 자꾸 틀린다", "파싱 에러가 빈번하다", "신뢰성이 떨어진다"는 문제를 겪어본 적 있나요? 초기 Agent 구현에서는 LLM이 자유 형식 텍스트로 도구를 호출했기 때문에 이런 문제가 흔했습니다.
이런 문제는 프로덕션 환경에서 치명적입니다. 10번 중 8번은 성공하지만 2번은 파싱 에러로 실패한다면, 사용자 경험이 크게 나빠지고 신뢰도가 떨어집니다.
바로 이럴 때 필요한 것이 Tool Calling(또는 Function Calling)입니다. OpenAI, Anthropic 등의 최신 LLM들이 제공하는 네이티브 도구 호출 기능으로, JSON 스키마를 사용해 구조화된 방식으로 도구를 호출합니다.
개요
간단히 말해서, Tool Calling은 LLM이 "자유 텍스트"가 아닌 "구조화된 JSON"으로 도구를 호출하는 방식입니다. LLM 자체가 도구 정의를 이해하고 올바른 형식으로 호출 정보를 생성합니다.
Tool Calling이 필요한 이유는 신뢰성과 정확성 때문입니다. 기존의 텍스트 파싱 방식은 "Action: CalculateTool\nAction Input: 123"처럼 자유 형식이었는데, LLM이 줄바꿈을 빼먹거나 오타를 내면 파싱이 실패했습니다.
하지만 Tool Calling은 LLM이 {"name": "CalculateTool", "arguments": {"value": 123}}처럼 JSON을 직접 생성하므로 훨씬 안정적입니다. 기존의 텍스트 기반 ReAct 패턴이 "사람이 읽기 좋은" 형식이었다면, Tool Calling은 "기계가 파싱하기 좋은" 형식으로 진화한 것입니다.
Tool Calling의 핵심 장점은 네 가지입니다: 1) 높은 신뢰성 - JSON 스키마로 형식 보장 2) 타입 안정성 - 파라미터 타입이 명확함 3) 다중 도구 호출 - 한 번에 여러 도구를 병렬 호출 가능 4) 네이티브 지원 - LLM 자체 기능이므로 빠르고 정확. 이러한 장점들이 현대적인 Agent 시스템의 표준이 되고 있습니다.
코드 예제
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from langchain_core.prompts import ChatPromptTemplate
# 도구 정의 (이전과 동일)
def multiply(a: str) -> str:
"""두 수를 곱합니다. 입력은 'x,y' 형식입니다."""
x, y = map(float, a.split(','))
return str(x * y)
def add(a: str) -> str:
"""두 수를 더합니다. 입력은 'x,y' 형식입니다."""
x, y = map(float, a.split(','))
return str(x + y)
tools = [
Tool(name="Multiply", func=multiply, description="두 수를 곱합니다. 입력: 'x,y'"),
Tool(name="Add", func=add, description="두 수를 더합니다. 입력: 'x,y'")
]
# Tool Calling 지원 모델 (gpt-4, gpt-3.5-turbo-1106 이상)
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Tool Calling용 프롬프트 (더 간단함)
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 수학 문제를 해결하는 도우미입니다."),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"), # Agent의 중간 과정
])
# Tool Calling Agent 생성 (create_react_agent 대신)
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 실행
result = executor.invoke({"input": "3과 4를 곱한 후, 그 결과에 5를 더하세요"})
print(result["output"])
설명
이것이 하는 일: 위 코드는 OpenAI의 Tool Calling 기능을 사용하는 Agent를 구축합니다. 기존의 ReAct 패턴과 달리 텍스트 파싱이 필요 없고, LLM이 직접 구조화된 함수 호출을 생성합니다.
첫 번째로, create_tool_calling_agent를 사용하는 것이 핵심 차이입니다. 이 함수는 create_react_agent와 달리 LLM의 네이티브 Tool Calling API를 활용합니다.
내부적으로 도구 정의를 OpenAI의 functions 파라미터로 변환하여 전달합니다. 예를 들어, Multiply 도구는 {"name": "Multiply", "description": "두 수를 곱합니다", "parameters": {...}} 형태의 JSON 스키마로 변환됩니다.
그 다음으로, LLM이 도구를 호출할 때의 차이가 중요합니다. ReAct 패턴에서는 "Action: Multiply\nAction Input: 3,4"처럼 자유 텍스트로 생성했지만, Tool Calling에서는 {"name": "Multiply", "arguments": {"a": "3,4"}}처럼 JSON 객체를 생성합니다.
이 JSON은 LLM의 특수 출력 형식으로 제공되어 파싱 에러가 거의 발생하지 않습니다. 프롬프트 구조도 더 간단해집니다.
ReAct 프롬프트는 "Thought", "Action", "Observation" 형식을 명시적으로 가르쳐야 했지만, Tool Calling은 LLM이 이미 알고 있으므로 간단한 시스템 메시지만 있으면 됩니다. agent_scratchpad 플레이스홀더는 이전 도구 호출 결과를 누적하는 역할을 하지만, 이것도 자동으로 관리됩니다.
실제 실행 과정을 보면, "3과 4를 곱한 후 5를 더하라"는 입력에 대해 Agent는 1) Multiply 도구를 {"a": "3,4"}로 호출 → 결과 "12" 2) Add 도구를 {"a": "12,5"}로 호출 → 결과 "17" 3) 최종 답변 "17입니다"를 생성합니다. 각 단계마다 JSON 형식이 보장되어 파싱 실패가 없습니다.
여러분이 Tool Calling을 사용하면 프로덕션 안정성이 크게 향상됩니다. 실제로 OpenAI와 Anthropic은 Agent 구축 시 Tool Calling 사용을 강력히 권장하며, LangChain도 새 프로젝트에서는 create_tool_calling_agent를 기본으로 사용하도록 변경했습니다.
또한 Tool Calling은 병렬 도구 호출도 지원합니다. 예를 들어, "날씨도 확인하고 뉴스도 가져와"라는 요청에 대해 두 도구를 동시에 호출할 수 있어 속도가 빠릅니다.
이는 텍스트 기반 ReAct에서는 불가능했던 기능입니다.
실전 팁
💡 Tool Calling은 gpt-4, gpt-3.5-turbo-1106 이상, claude-3 계열 모델에서만 지원됩니다. 오래된 모델을 사용하면 에러가 발생하니 모델 버전을 확인하세요.
💡 새 프로젝트는 항상 create_tool_calling_agent를 사용하세요. create_react_agent는 레거시이며 신뢰성이 낮습니다.
💡 Tool의 description이 여전히 중요합니다. Tool Calling도 LLM이 도구를 "선택"할 때는 description을 읽기 때문에 명확하게 작성해야 합니다.
💡 Pydantic 모델로 args_schema를 정의하면 Tool Calling에서 더 정확한 타입 정보를 제공할 수 있습니다. OpenAI API는 이를 JSON Schema로 변환합니다.
💡 병렬 도구 호출을 활용하려면 도구들이 독립적이어야 합니다. 한 도구의 결과가 다른 도구의 입력이 되는 경우는 순차적으로 실행됩니다.
6. Memory와 State
시작하며
여러분이 Agent를 만들었는데 "이전 대화를 기억하지 못한다", "같은 질문을 반복한다", "문맥을 잃어버린다"는 문제를 겪어본 적 있나요? 기본적으로 Agent는 각 호출마다 독립적이어서 대화 기록이 유지되지 않습니다.
이런 문제는 실제 챗봇이나 대화형 애플리케이션에서 치명적입니다. 사용자가 "그것에 대해 더 알려줘"라고 했을 때 "그것"이 무엇인지 모르면 대화가 성립하지 않습니다.
바로 이럴 때 필요한 것이 Memory와 State 관리입니다. Agent가 이전 대화를 기억하고, 문맥을 유지하며, 세션 전체에 걸쳐 일관된 행동을 할 수 있게 합니다.
개요
간단히 말해서, Memory는 Agent가 이전 대화 내용을 저장하고 참조하는 메커니즘입니다. 단순히 모든 메시지를 누적하는 방식부터 요약해서 저장하는 방식까지 다양합니다.
Memory가 필요한 이유는 대화의 연속성과 개인화 때문입니다. 실무에서는 단발성 질문보다 여러 번의 상호작용을 통해 문제를 해결하는 경우가 많습니다.
예를 들어, 사용자가 "내 주문을 취소해줘" → "아니 잠깐, 배송지만 변경할게" → "역시 취소할래"처럼 여러 번 의견을 바꾸는 상황에서 Memory가 없으면 각 요청을 독립적으로 처리하게 되어 혼란이 생깁니다. 기존의 stateless Agent가 "금붕어 기억력"이었다면, Memory를 추가하면 "정상적인 대화 상대"가 됩니다.
Memory의 핵심 유형은 네 가지입니다: 1) ConversationBufferMemory - 모든 메시지를 저장 (간단하지만 토큰 소비 증가) 2) ConversationSummaryMemory - 대화를 주기적으로 요약 (긴 대화에 적합) 3) ConversationBufferWindowMemory - 최근 N개 메시지만 유지 (적절한 균형) 4) ConversationEntityMemory - 주요 엔티티(사람, 장소 등)를 추적 (복잡한 대화에 유용). 사용 사례에 따라 적절한 타입을 선택하는 것이 중요합니다.
코드 예제
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
# 간단한 도구
def get_user_info(user_id: str) -> str:
users = {"123": "이름: 김철수, 등급: VIP", "456": "이름: 박영희, 등급: 일반"}
return users.get(user_id, "사용자 없음")
tools = [Tool(name="UserInfo", func=get_user_info, description="사용자 ID로 정보 조회")]
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Memory 추가된 프롬프트
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 고객 서비스 담당자입니다. 이전 대화를 참고하여 답변하세요."),
MessagesPlaceholder(variable_name="chat_history"), # 메모리가 들어갈 자리
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# Memory 설정
memory = ConversationBufferMemory(
memory_key="chat_history", # 프롬프트의 variable_name과 일치
return_messages=True # ChatPromptTemplate에는 메시지 객체 필요
)
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)
# 대화 진행 - Memory가 누적됨
print(executor.invoke({"input": "사용자 123의 정보를 알려줘"})["output"])
print(executor.invoke({"input": "그 사람의 등급이 뭐였지?"})["output"]) # "123"을 기억함
print(executor.invoke({"input": "VIP 고객에게 할인을 적용해줘"})["output"]) # 문맥 유지
설명
이것이 하는 일: 위 코드는 대화 히스토리를 유지하는 Agent를 구축하여, 사용자가 "그 사람"처럼 이전 문맥을 참조해도 이해하고 응답할 수 있게 합니다. 첫 번째로, MessagesPlaceholder를 프롬프트에 추가하는 것이 핵심입니다.
chat_history라는 변수명으로 플레이스홀더를 만들면, 실행 시점에 여기에 이전 대화 메시지들이 주입됩니다. 시스템 메시지와 사용자 입력 사이에 배치하여, Agent가 대화 흐름을 자연스럽게 이해할 수 있습니다.
그 다음으로, ConversationBufferMemory를 생성하면서 memory_key="chat_history"로 설정합니다. 이것이 프롬프트의 MessagesPlaceholder와 연결되는 지점입니다.
return_messages=True는 중요한데, ChatPromptTemplate은 문자열이 아닌 메시지 객체 리스트를 기대하기 때문입니다. 만약 일반 PromptTemplate을 사용한다면 return_messages=False로 설정해야 합니다.
AgentExecutor에 memory를 전달하면, 실행 후 자동으로 입력과 출력이 메모리에 저장됩니다. 첫 번째 호출 "사용자 123의 정보를 알려줘"가 끝나면, memory.chat_history에 [HumanMessage("사용자 123..."), AIMessage("김철수님은 VIP 등급입니다")]가 저장됩니다.
두 번째 호출 "그 사람의 등급이 뭐였지?"가 들어오면, Agent는 chat_history를 보고 "그 사람"이 "사용자 123" 즉 "김철수"를 가리킨다는 것을 이해합니다. 도구를 다시 호출할 필요 없이 메모리에서 정보를 찾아 "VIP 등급입니다"라고 답변할 수 있습니다.
세 번째 호출에서는 더 복잡한 문맥 유지가 일어납니다. "VIP 고객에게 할인을 적용해줘"는 1) 현재 고객이 VIP라는 정보 2) 그 고객이 김철수(사용자 123)라는 정보를 모두 기억해야 합니다.
Memory 덕분에 이 모든 문맥이 유지됩니다. 여러분이 이 패턴을 사용하면 진짜 대화처럼 느껴지는 Agent를 만들 수 있습니다.
사용자가 대명사("그것", "그 사람"), 생략("등급은?"), 암묵적 참조를 사용해도 이해하고 응답합니다. 실무에서는 ConversationBufferMemory는 짧은 세션에만 적합합니다.
대화가 길어지면 토큰이 폭증하므로, ConversationBufferWindowMemory(최근 5개 메시지만)나 ConversationSummaryMemory(요약 사용)를 고려하세요.
실전 팁
💡 �� memory_key와 MessagesPlaceholder의 variable_name은 반드시 일치해야 합니다. 다르면 메모리가 주입되지 않습니다.
💡 return_messages=True/False는 프롬프트 타입에 따라 결정됩니다. ChatPromptTemplate은 True, 일반 PromptTemplate은 False입니다.
💡 긴 대화에서는 ConversationBufferWindowMemory(k=5)로 최근 5개만 유지하세요. 토큰 비용과 성능의 균형점을 찾는 것이 중요합니다.
💡 프로덕션에서는 메모리를 외부 저장소(Redis, DB)에 저장하세요. 기본 메모리는 메모리에만 있어서 재시작하면 사라집니다.
💡 memory.clear()로 대화를 리셋할 수 있습니다. 사용자가 "새로운 대화 시작"을 요청하거나 세션이 만료되면 호출하세요.
7. Custom Tool 구현
시작하며
여러분이 Agent에 "우리 회사만의 특별한 비즈니스 로직", "레거시 시스템 연동", "복잡한 데이터 처리 파이프라인"을 연결하고 싶었던 적 있나요? LangChain이 제공하는 기본 도구로는 한계가 있습니다.
이런 문제는 실제 엔터프라이즈 환경에서 필수적으로 마주치는 과제입니다. 모든 회사는 고유한 시스템과 프로세스를 가지고 있으며, Agent가 진정으로 유용하려면 이것들과 통합되어야 합니다.
바로 이럴 때 필요한 것이 Custom Tool 구현입니다. BaseTool 클래스를 상속받아 완전히 커스터마이징된 도구를 만들고, 복잡한 로직, 에러 처리, 비동기 작업까지 모두 통합할 수 있습니다.
개요
간단히 말해서, Custom Tool은 BaseTool 클래스를 상속하여 _run() 메서드를 구현하는 방식으로 만듭니다. 이를 통해 간단한 함수 래퍼를 넘어서 복잡한 비즈니스 로직을 캡슐화할 수 있습니다.
Custom Tool이 필요한 이유는 실무의 복잡성 때문입니다. 단순한 계산이나 API 호출을 넘어서, 데이터 검증, 권한 확인, 트랜잭션 처리, 로깅, 에러 복구 등 다양한 요구사항을 처리해야 합니다.
예를 들어, 주문 처리 도구는 재고 확인, 결제 검증, 배송 시스템 연동, 이메일 발송 등 여러 단계를 거쳐야 하는데, 이런 복잡한 로직을 체계적으로 관리하려면 클래스 기반 접근이 필요합니다. 기존의 Tool() 래퍼가 "간단한 함수용"이라면, BaseTool은 "엔터프라이즈급 기능용"입니다.
Custom Tool의 핵심 구성 요소는 다섯 가지입니다: 1) name - 도구 이름 (속성) 2) description - 도구 설명 (속성) 3) args_schema - 입력 스키마 (Pydantic 모델) 4) _run() - 동기 실행 메서드 5) _arun() - 비동기 실행 메서드 (선택). 이러한 구조가 복잡한 도구를 체계적으로 관리할 수 있게 합니다.
코드 예제
from langchain.tools import BaseTool
from langchain.pydantic_v1 import BaseModel, Field
from typing import Optional, Type
import logging
# 입력 스키마 정의
class OrderProcessInput(BaseModel):
order_id: str = Field(description="처리할 주문 ID")
action: str = Field(description="수행할 작업: 'cancel', 'ship', 'refund'")
reason: Optional[str] = Field(default=None, description="작업 사유 (선택)")
# Custom Tool 구현
class OrderProcessTool(BaseTool):
name: str = "OrderProcessor"
description: str = """
주문을 처리합니다. 취소, 배송, 환불 등의 작업을 수행할 수 있습니다.
주문 관련 요청이 있을 때 사용하세요.
"""
args_schema: Type[BaseModel] = OrderProcessInput
# 내부 설정 (Agent에 노출되지 않음)
def __init__(self):
super().__init__()
self.logger = logging.getLogger(__name__)
def _run(self, order_id: str, action: str, reason: Optional[str] = None) -> str:
"""동기 실행 메서드 - 실제 비즈니스 로직"""
try:
# 1. 주문 존재 확인
if not self._validate_order(order_id):
return f"오류: 주문 {order_id}를 찾을 수 없습니다."
# 2. 권한 확인 (실제로는 사용자 컨텍스트 필요)
if not self._check_permission(action):
return f"오류: {action} 작업 권한이 없습니다."
# 3. 작업 수행
if action == "cancel":
result = self._cancel_order(order_id, reason)
elif action == "ship":
result = self._ship_order(order_id)
elif action == "refund":
result = self._refund_order(order_id, reason)
else:
return f"오류: 알 수 없는 작업 '{action}'"
# 4. 로깅
self.logger.info(f"Order {order_id}: {action} - {result}")
return result
except Exception as e:
self.logger.error(f"Order processing error: {e}")
return f"처리 중 오류 발생: {str(e)}"
async def _arun(self, order_id: str, action: str, reason: Optional[str] = None) -> str:
"""비동기 실행 메서드 (선택)"""
# 비동기가 필요하면 구현, 아니면 NotImplementedError
raise NotImplementedError("비동기 실행은 지원하지 않습니다.")
# 헬퍼 메서드들
def _validate_order(self, order_id: str) -> bool:
# 실제로는 DB 조회
return order_id.startswith("ORD")
def _check_permission(self, action: str) -> bool:
# 실제로는 권한 시스템 확인
return True
def _cancel_order(self, order_id: str, reason: Optional[str]) -> str:
# 실제 취소 로직
return f"주문 {order_id}가 취소되었습니다. 사유: {reason or '없음'}"
def _ship_order(self, order_id: str) -> str:
return f"주문 {order_id}가 배송 처리되었습니다."
def _refund_order(self, order_id: str, reason: Optional[str]) -> str:
return f"주문 {order_id}가 환불 처리되었습니다. 사유: {reason or '없음'}"
# 사용 예시
custom_tool = OrderProcessTool()
result = custom_tool._run("ORD12345", "cancel", "고객 요청")
print(result)
설명
이것이 하는 일: 위 코드는 주문 처리라는 복잡한 비즈니스 로직을 Agent가 사용할 수 있는 도구로 만듭니다. 단순한 함수 호출을 넘어서 검증, 권한 확인, 다양한 작업 분기, 에러 처리, 로깅까지 모두 포함합니다.
첫 번째로, OrderProcessInput Pydantic 모델이 입력 스키마를 정의합니다. order_id, action, reason 세 개의 필드가 있으며, 각각 description이 있어서 LLM이 무엇을 입력해야 할지 정확히 알 수 있습니다.
Optional[str]로 정의된 reason은 선택적 파라미터로, 없어도 되지만 있으면 더 나은 처리가 가능합니다. 그 다음으로, OrderProcessTool 클래스의 속성들이 도구의 메타데이터를 정의합니다.
name과 description은 Agent가 도구를 선택할 때 참고하는 정보이고, args_schema는 LLM이 올바른 형식으로 입력을 생성하도록 가이드합니다. description에 "주문 관련 요청이 있을 때 사용하세요"처럼 사용 시점을 명시하면 더 정확한 선택이 가능합니다.
_run() 메서드는 실제 실행 로직입니다. 여기서 중요한 점은 단계별로 검증과 처리를 분리했다는 것입니다: 1) _validate_order로 주문 존재 확인 2) _check_permission으로 권한 확인 3) action에 따라 적절한 헬퍼 메서드 호출 4) 로깅과 에러 처리.
이러한 구조 덕분에 각 단계를 독립적으로 테스트하고 유지보수할 수 있습니다. init 메서드에서 logger를 설정하는 것처럼, 클래스 기반 접근은 내부 상태를 관리할 수 있습니다.
데이터베이스 연결, API 클라이언트, 캐시 등을 초기화하여 재사용할 수 있어 효율적입니다. 예를 들어, self.db_client = create_db_connection()처럼 연결을 한 번만 만들고 계속 사용할 수 있습니다.
try-except 블록으로 모든 예외를 잡아서 사용자 친화적인 메시지로 변환하는 것도 중요합니다. Agent가 받는 응답은 항상 문자열이어야 하므로, 예외가 발생해도 "오류: ..."처럼 설명적인 문자열을 반환해야 합니다.
그냥 예외를 던지면 Agent 전체가 멈출 수 있습니다. 여러분이 이 패턴을 사용하면 기존 비즈니스 로직을 거의 수정 없이 Agent에 통합할 수 있습니다.
주문 처리, 재고 관리, 사용자 관리, 결제 시스템 등 어떤 복잡한 시스템이든 BaseTool로 감싸면 Agent가 사용할 수 있습니다. 실무에서는 각 도구를 별도 파일로 분리하고(tools/order_tool.py), 테스트 코드를 작성하고(tests/test_order_tool.py), 문서화하는 것이 좋습니다.
도구가 많아질수록 관리가 중요해집니다.
실전 팁
💡 description은 매우 상세하게 작성하세요. "언제 사용하는지", "어떤 작업을 하는지", "주의사항"까지 포함하면 Agent가 더 정확히 선택합니다.
💡 _run()에서는 항상 문자열을 반환하세요. 객체를 반환하면 LLM이 이해하지 못합니다. JSON 문자열이나 자연어 설명으로 변환하세요.
💡 에러 메시지는 명확하고 실행 가능해야 합니다. "오류 발생"보다 "주문 번호는 ORD로 시작해야 합니다. 예: ORD12345"처럼 구체적으로 작성하세요.
💡 비동기가 필요 없으면 _arun()에서 NotImplementedError를 발생시키세요. 구현하지 않으면 기본적으로 _run()을 호출하지만 명시적으로 하는 것이 좋습니다.
💡 복잡한 도구는 단위 테스트를 작성하세요. Agent 없이 도구만 테스트하면 빠르게 버그를 찾을 수 있습니다: assert custom_tool._run("ORD123", "cancel") == "..."
8. Error Handling
시작하며
여러분이 Agent를 운영하면서 "도구 실행이 실패한다", "LLM이 잘못된 입력을 생성한다", "네트워크 에러로 멈춘다" 같은 문제를 겪어본 적 있나요? Agent는 여러 외부 시스템과 상호작용하므로 오류 발생 가능성이 높습니다.
이런 문제는 프로덕션 환경에서 가장 큰 위험 요소입니다. Agent가 한 번 에러로 멈추면 전체 사용자 경험이 망가지고, 복구하기도 어렵습니다.
더 나쁜 것은 Agent가 에러를 무시하고 잘못된 정보로 답변하는 경우입니다. 바로 이럴 때 필요한 것이 체계적인 Error Handling 전략입니다.
도구 수준, Agent 수준, Executor 수준에서 각각 적절한 에러 처리를 구현하여 안정적이고 회복 가능한 시스템을 만듭니다.
개요
간단히 말해서, Error Handling은 Agent의 실행 과정에서 발생할 수 있는 다양한 오류를 감지하고, 복구하거나 사용자에게 명확히 알리는 전략입니다. "실패해도 계속 진행"과 "실패하면 중단"을 상황에 맞게 선택합니다.
Error Handling이 필요한 이유는 신뢰성과 사용자 경험 때문입니다. Agent는 API 호출, 데이터베이스 조회, 외부 서비스 연동 등 실패할 수 있는 작업을 많이 수행합니다.
각각의 실패를 어떻게 처리하느냐에 따라 전체 시스템의 안정성이 결정됩니다. 예를 들어, 날씨 API가 타임아웃되었을 때 전체 Agent를 멈출 것인지, "날씨 정보를 가져올 수 없습니다"라고 알리고 계속 진행할 것인지 결정해야 합니다.
기존의 "에러 발생 시 종료" 방식이 취약했다면, 이제는 다층적 에러 처리로 복원력(resilience)을 갖춘 시스템을 만들 수 있습니다. Error Handling의 핵심 레벨은 네 가지입니다: 1) 도구 수준 - 각 도구 내부에서 try-except로 처리 2) Agent 수준 - handle_parsing_errors로 LLM 출력 오류 복구 3) Executor 수준 - max_iterations, max_execution_time으로 폭주 방지 4) 애플리케이션 수준 - 전체 invoke()를 try-except로 감싸서 최종 안전망 제공.
이러한 다층 방어가 안정적인 프로덕션 시스템을 만듭니다.
코드 예제
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import BaseTool
from langchain.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Type
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 입력 스키마
class APICallInput(BaseModel):
endpoint: str = Field(description="호출할 API 엔드포인트")
# 1. 도구 수준 에러 처리
class ExternalAPITool(BaseTool):
name: str = "ExternalAPI"
description: str = "외부 API를 호출합니다. 네트워크 오류 시 재시도합니다."
args_schema: Type[BaseModel] = APICallInput
def _run(self, endpoint: str) -> str:
max_retries = 3
for attempt in range(max_retries):
try:
# 실제로는 requests.get() 등
if endpoint == "/fail":
raise ConnectionError("네트워크 오류")
return f"{endpoint} 호출 성공: 데이터 반환됨"
except ConnectionError as e:
logger.warning(f"시도 {attempt + 1}/{max_retries} 실패: {e}")
if attempt == max_retries - 1:
# 최종 실패 - 명확한 메시지 반환 (예외 던지지 않음)
return f"오류: {endpoint} 호출 실패. 네트워크를 확인하세요. 원인: {e}"
except Exception as e:
# 예상치 못한 오류
logger.error(f"예상치 못한 오류: {e}")
return f"오류: 처리 중 문제 발생 - {str(e)}"
return "오류: 최대 재시도 횟수 초과"
async def _arun(self, endpoint: str) -> str:
raise NotImplementedError()
tools = [ExternalAPITool()]
# 2. Agent & Executor 수준 에러 처리
llm = ChatOpenAI(model="gpt-4", temperature=0, request_timeout=10) # 타임아웃 설정
prompt = ChatPromptTemplate.from_messages([
("system", "외부 API를 호출하는 도우미입니다. 오류 발생 시 사용자에게 명확히 알리세요."),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5, # 폭주 방지
max_execution_time=30, # 30초 제한
handle_parsing_errors=True, # LLM 파싱 오류 자동 복구
return_intermediate_steps=True # 디버깅용
)
# 3. 애플리케이션 수준 에러 처리
def safe_agent_call(user_input: str) -> dict:
"""안전한 Agent 호출 래퍼"""
try:
result = executor.invoke({"input": user_input})
return {
"success": True,
"output": result["output"],
"steps": len(result.get("intermediate_steps", []))
}
except TimeoutError:
logger.error("Agent 실행 시간 초과")
return {
"success": False,
"error": "요청 처리 시간이 초과되었습니다. 잠시 후 다시 시도해주세요."
}
except Exception as e:
logger.error(f"Agent 실행 오류: {e}", exc_info=True)
return {
"success": False,
"error": f"처리 중 오류가 발생했습니다: {str(e)}"
}
# 사용 예시
print("=== 정상 케이스 ===")
result1 = safe_agent_call("/users 엔드포인트를 호출해주세요")
print(result1)
print("\n=== 오류 케이스 (도구 수준 복구) ===")
result2 = safe_agent_call("/fail 엔드포인트를 호출해주세요")
print(result2)
설명
이것이 하는 일: 위 코드는 4개 레벨의 에러 처리 전략을 구현하여, Agent가 다양한 오류 상황에서도 안정적으로 작동하거나 명확한 에러 메시지를 제공하도록 합니다. 첫 번째 레벨은 도구 내부의 에러 처리입니다.
ExternalAPITool의 _run() 메서드에서 max_retries로 3번까지 재시도를 구현했습니다. 네트워크는 일시적으로 실패할 수 있으므로, 첫 번째 실패가 영구적이지 않을 수 있습니다.
중요한 점은 최종 실패 시에도 예외를 던지지 않고 "오류: ..." 형태의 문자열을 반환한다는 것입니다. 이렇게 하면 Agent가 멈추지 않고 오류 메시지를 사용자에게 전달할 수 있습니다.
두 번째 레벨은 LLM과 Executor 설정입니다. ChatOpenAI에 request_timeout=10을 설정하면 LLM API 호출이 10초 이상 걸리면 자동으로 실패합니다.
이는 네트워크 문제로 무한정 대기하는 것을 방지합니다. AgentExecutor의 handle_parsing_errors=True는 LLM이 잘못된 형식으로 도구 호출을 생성했을 때, "형식이 잘못되었습니다.
다시 시도하세요"라는 피드백을 자동으로 제공하여 스스로 수정하게 합니다. 세 번째 레벨은 전체 실행 제한입니다.
max_iterations=5는 Agent가 무한 루프에 빠지는 것을 방지하고, max_execution_time=30은 전체 실행이 30초를 넘으면 강제 종료합니다. 이러한 제한이 없으면 복잡한 요청이나 버그로 인해 Agent가 끝없이 실행될 수 있습니다.
네 번째 레벨은 애플리케이션 코드의 래퍼입니다. safe_agent_call() 함수가 executor.invoke()를 try-except로 감싸서, 어떤 예외가 발생하더라도 구조화된 응답을 반환합니다.
success 플래그로 성공/실패를 명확히 표시하고, 오류 시에는 사용자 친화적인 메시지를 제공합니다. exc_info=True는 전체 스택 트레이스를 로그에 남겨서 나중에 디버깅할 수 있게 합니다.
이러한 다층 방어 덕분에 어떤 레벨에서 문제가 발생하더라도 시스템이 우아하게(gracefully) 처리합니다. 도구 실행 실패 → 도구가 에러 메시지 반환 → Agent가 사용자에게 전달.
LLM 파싱 오류 → 자동 복구 시도. 전체 시스템 예외 → 래퍼가 잡아서 에러 응답 반환.
여러분이 이 패턴을 사용하면 프로덕션 환경에서 안정적으로 Agent를 운영할 수 있습니다. 사용자는 "500 Internal Server Error" 같은 기술적 오류 대신 "날씨 정보를 가져올 수 없습니다.
잠시 후 다시 시도해주세요" 같은 친절한 메시지를 받습니다. 실무에서는 에러를 모니터링 시스템(Sentry, DataDog 등)에 전송하여 추적하고, 에러율을 대시보드로 확인하고, 알림을 설정하는 것도 중요합니다.
실전 팁
💡 도구에서는 절대 예외를 던지지 말고 에러 메시지 문자열을 반환하세요. 예외를 던지면 전체 Agent가 멈춥니다.
💡 재시도 로직에는 exponential backoff를 추가하면 더 좋습니다. time.sleep(2 ** attempt)로 점진적으로 대기 시간을 늘리면 일시적 장애에 강합니다.
💡 handle_parsing_errors=True는 개발 중에 특히 유용합니다. LLM이 형식을 틀리는 이유를 로그에서 확인하고 프롬프트를 개선할 수 있습니다.
💡 return_intermediate_steps=True로 오류가 어느 단계에서 발생했는지 추적하세요. 디버깅과 개선에 필수적입니다.
💡 프로덕션에서는 사용자에게 에러 ID를 제공하세요. "오류 발생 (ID: err_12345)"처럼 하면 사용자가 문의할 때 로그를 빠르게 찾을 수 있습니다.