3편: LangChain Agent 실전 활용
시작하며
지금까지 RAG 시스템과 복잡한 체인을 만들어봤는데요, 사실 이것들은 정해진 흐름대로만 동작합니다. "문서 검색 → 답변 생성" 이런 식으로요. 하지만 실무에서는 "상황에 따라 알아서 필요한 작업을 해주세요"라는 요구사항이 많습니다.
예를 들어 "작년 매출 데이터를 분석해서 리포트 만들어줘"라는 요청이 들어오면, 데이터베이스에서 데이터를 가져오고, 분석하고, 차트를 만들고, 문서로 정리하는 일련의 과정을 스스로 판단해서 진행해야 하죠. 이게 바로 Agent가 하는 일입니다.
이번 편에서는 LangChain Agent를 실전에서 어떻게 활용하는지 알아보겠습니다.
Agent란 무엇인가
Agent는 주어진 목표를 달성하기 위해 스스로 어떤 도구(Tool)를 사용할지 결정하고 실행하는 시스템입니다. 기존 Chain은 우리가 정해준 순서대로만 실행되지만, Agent는 상황을 판단해서 다음 행동을 선택합니다.
간단히 비유하자면, Chain은 레시피를 정확히 따르는 요리이고, Agent는 냉장고에 있는 재료를 보고 알아서 요리를 만드는 셰프입니다.
Agent의 작동 방식은 이렇습니다. 먼저 사용자의 요청을 받고, 어떤 도구를 사용할지 결정합니다. 그 도구를 실행하고 결과를 받아서, 목표가 달성되었으면 최종 답변을 하고, 아니면 다음 도구를 선택해서 다시 실행합니다. 이 과정을 목표 달성까지 반복하는 것이죠.
첫 번째 Agent 만들기
간단한 예제부터 시작해보겠습니다. 검색과 계산을 할 수 있는 Agent를 만들어볼게요.
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import Tool
# LLM 설정
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 도구 정의
search = DuckDuckGoSearchRun()
def calculator(expression: str) -> str:
"""수학 계산을 수행합니다. 예: '2 + 2' 또는 '10 * 5'"""
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"계산 오류: {str(e)}"
tools = [
Tool(
name="Search",
func=search.run,
description="최신 정보나 사실을 검색할 때 사용합니다. 입력은 검색 쿼리입니다."
),
Tool(
name="Calculator",
func=calculator,
description="수학 계산이 필요할 때 사용합니다. 입력은 계산식입니다."
)
]
# 프롬프트 설정
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 유능한 비서입니다. 주어진 도구를 활용하여 사용자의 질문에 답변하세요."),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# Agent 생성
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 사용해보기
response = agent_executor.invoke({
"input": "2024년 노벨 물리학상 수상자는 누구이고, 그들의 나이를 모두 더하면?"
})
print(response['output'])이 Agent는 먼저 검색 도구로 노벨상 수상자를 찾고, 그 다음 계산기로 나이를 더합니다. 우리가 순서를 지정하지 않았는데도 스스로 판단해서 실행하는 것이죠.
커스텀 도구 만들기
실무에서는 데이터베이스 조회, API 호출 같은 커스텀 도구가 필요합니다.
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool
from typing import Optional, Type
import sqlite3
# 입력 스키마 정의
class DatabaseQueryInput(BaseModel):
query: str = Field(description="실행할 SQL 쿼리")
# 커스텀 도구 클래스
class DatabaseQueryTool(BaseTool):
name = "database_query"
description = "데이터베이스에서 정보를 조회할 때 사용합니다. SQL 쿼리를 입력으로 받습니다."
args_schema: Type[BaseModel] = DatabaseQueryInput
def _run(self, query: str) -> str:
"""도구 실행 로직"""
try:
conn = sqlite3.connect('company.db')
cursor = conn.cursor()
cursor.execute(query)
results = cursor.fetchall()
conn.close()
if not results:
return "조회 결과가 없습니다."
return str(results)
except Exception as e:
return f"데이터베이스 오류: {str(e)}"
async def _arun(self, query: str) -> str:
"""비동기 실행 (선택사항)"""
raise NotImplementedError("비동기 실행은 지원하지 않습니다.")
# 도구 사용
db_tool = DatabaseQueryTool()
tools.append(db_tool)실전 예제: 업무 자동화 Agent
실제 업무에 사용할 수 있는 Agent를 만들어보겠습니다. 직원 정보 조회, 휴가 신청, 이메일 발송 등을 할 수 있는 HR Agent입니다.
from datetime import datetime
import json
# 직원 정보 조회 도구
class EmployeeInfoTool(BaseTool):
name = "get_employee_info"
description = "직원 ID로 직원 정보를 조회합니다."
def _run(self, employee_id: str) -> str:
# 실제로는 데이터베이스에서 조회
employees = {
"E001": {"name": "김철수", "department": "개발팀", "remaining_vacation": 15},
"E002": {"name": "이영희", "department": "마케팅팀", "remaining_vacation": 10},
}
info = employees.get(employee_id)
if info:
return json.dumps(info, ensure_ascii=False)
return "직원을 찾을 수 없습니다."
# 휴가 신청 도구
class VacationRequestInput(BaseModel):
employee_id: str = Field(description="직원 ID")
start_date: str = Field(description="시작일 (YYYY-MM-DD)")
end_date: str = Field(description="종료일 (YYYY-MM-DD)")
reason: str = Field(description="사유")
class VacationRequestTool(BaseTool):
name = "request_vacation"
description = "휴가를 신청합니다."
args_schema: Type[BaseModel] = VacationRequestInput
def _run(self, employee_id: str, start_date: str, end_date: str, reason: str) -> str:
# 실제로는 시스템에 등록
return f"휴가 신청 완료: {employee_id}, {start_date} ~ {end_date}, 사유: {reason}"
# 이메일 발송 도구
class SendEmailInput(BaseModel):
to: str = Field(description="받는 사람 이메일")
subject: str = Field(description="제목")
body: str = Field(description="본문")
class SendEmailTool(BaseTool):
name = "send_email"
description = "이메일을 발송합니다."
args_schema: Type[BaseModel] = SendEmailInput
def _run(self, to: str, subject: str, body: str) -> str:
# 실제로는 이메일 발송
return f"이메일 발송 완료: {to}"
# HR Agent 구성
hr_tools = [
EmployeeInfoTool(),
VacationRequestTool(),
SendEmailTool()
]
hr_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 회사의 HR 업무를 돕는 AI 비서입니다.
직원 정보 조회, 휴가 신청, 이메일 발송 등의 업무를 수행할 수 있습니다.
항상 정확하고 친절하게 응대하세요."""),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
hr_agent = create_openai_functions_agent(llm, hr_tools, hr_prompt)
hr_executor = AgentExecutor(agent=hr_agent, tools=hr_tools, verbose=True)
# 사용 예시
response = hr_executor.invoke({
"input": "직원 E001의 남은 연차를 확인하고, 2024년 12월 23일부터 27일까지 휴가를 신청한 후 관리자에게 이메일을 보내주세요."
})
print(response['output'])Agent가 스스로 순서를 정해서 실행합니다. 직원 정보 조회 → 휴가 신청 → 이메일 발송 순서로요.
ReAct Agent 패턴
ReAct(Reasoning + Acting)는 생각과 행동을 번갈아가며 수행하는 패턴입니다. Agent가 자신의 사고 과정을 드러내기 때문에 디버깅하기 좋습니다.
from langchain.agents import create_react_agent
from langchain import hub
# ReAct 프롬프트 가져오기
react_prompt = hub.pull("hwchase17/react")
# ReAct Agent 생성
react_agent = create_react_agent(llm, tools, react_prompt)
react_executor = AgentExecutor(
agent=react_agent,
tools=tools,
verbose=True,
handle_parsing_errors=True
)
response = react_executor.invoke({
"input": "한국의 인구는 몇 명이고, 이를 10으로 나눈 값은?"
})실행 로그를 보면 이런 식으로 나옵니다:
Thought: 먼저 한국의 인구를 검색해야겠습니다.
Action: Search
Action Input: "한국 인구 2024"
Observation: 한국의 인구는 약 5,200만 명입니다.
Thought: 이제 계산이 필요합니다.
Action: Calculator
Action Input: "52000000 / 10"
Observation: 5200000.0
Thought: 최종 답변을 할 수 있습니다.
Final Answer: 한국의 인구는 약 5,200만 명이고, 이를 10으로 나누면 520만입니다.구조화된 도구 사용하기
복잡한 파라미터를 받는 도구는 구조화된 입력이 필요합니다.
from langchain.tools import StructuredTool
def create_report(title: str, data: dict, format: str = "pdf") -> str:
"""
리포트를 생성합니다.
Args:
title: 리포트 제목
data: 리포트에 포함될 데이터 (딕셔너리)
format: 출력 형식 (pdf, excel, html 중 선택)
"""
return f"{format.upper()} 형식의 '{title}' 리포트가 생성되었습니다."
# 구조화된 도구로 변환
report_tool = StructuredTool.from_function(
func=create_report,
name="create_report",
description="데이터를 기반으로 리포트를 생성합니다."
)
tools.append(report_tool)Agent with Memory
대화 기록을 기억하는 Agent를 만들 수 있습니다.
from langchain.memory import ConversationBufferMemory
from langchain.agents import initialize_agent, AgentType
# 메모리 설정
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 메모리를 가진 Agent
conversational_agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
memory=memory,
verbose=True
)
# 대화하기
response1 = conversational_agent.invoke({
"input": "내 이름은 김철수입니다."
})
response2 = conversational_agent.invoke({
"input": "내 이름이 뭐라고 했죠?"
})
# "김철수님이라고 하셨습니다." 같은 답변Multi-Agent 시스템
여러 Agent가 협력하는 시스템도 만들 수 있습니다.
# 연구원 Agent
researcher_prompt = ChatPromptTemplate.from_messages([
("system", "당신은 정보를 조사하는 연구원입니다. 검색 도구를 사용하여 정확한 정보를 수집하세요."),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
researcher_agent = create_openai_functions_agent(llm, [search], researcher_prompt)
researcher_executor = AgentExecutor(agent=researcher_agent, tools=[search])
# 작가 Agent
writer_prompt = ChatPromptTemplate.from_messages([
("system", "당신은 전문 작가입니다. 주어진 정보를 바탕으로 읽기 쉽고 설득력 있는 글을 작성하세요."),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
writer_agent = create_openai_functions_agent(llm, [], writer_prompt)
writer_executor = AgentExecutor(agent=writer_agent, tools=[])
# 워크플로우
def research_and_write(topic: str) -> str:
# 1단계: 연구
research_result = researcher_executor.invoke({
"input": f"{topic}에 대해 조사해주세요."
})
# 2단계: 작성
writing_result = writer_executor.invoke({
"input": f"다음 정보를 바탕으로 블로그 글을 작성해주세요:\n{research_result['output']}"
})
return writing_result['output']
article = research_and_write("2024년 AI 트렌드")
print(article)Agent 제어하기
Agent가 무한 루프에 빠지거나 너무 많은 도구를 사용하는 것을 방지해야 합니다.
# 최대 반복 횟수 제한
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5, # 최대 5번만 반복
max_execution_time=30, # 최대 30초
early_stopping_method="generate" # 시간 초과 시 현재까지 결과 반환
)
# 타임아웃 처리
try:
response = agent_executor.invoke(
{"input": "복잡한 작업"},
config={"timeout": 60} # 60초 타임아웃
)
except Exception as e:
print(f"Agent 실행 실패: {e}")에러 처리
Agent는 여러 이유로 실패할 수 있으므로 적절한 에러 처리가 필요합니다.
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True, # 파싱 에러 자동 처리
return_intermediate_steps=True # 중간 단계 반환
)
response = agent_executor.invoke({"input": "질문"})
# 중간 단계 확인
if 'intermediate_steps' in response:
for step in response['intermediate_steps']:
action, observation = step
print(f"도구: {action.tool}, 입력: {action.tool_input}")
print(f"결과: {observation}\n")도구 선택 최적화
Agent가 적절한 도구를 선택하도록 설명을 잘 작성하는 것이 중요합니다.
# 나쁜 예
bad_tool = Tool(
name="tool",
func=some_function,
description="뭔가 하는 도구" # 너무 모호함
)
# 좋은 예
good_tool = Tool(
name="get_stock_price",
func=get_stock_price,
description="""주식의 현재 가격을 조회합니다.
입력: 주식 종목 코드 (예: 'AAPL', '005930')
출력: 현재 주가와 변동률
사용 시점: 사용자가 특정 주식의 가격을 물어볼 때만 사용하세요."""
)LangGraph로 복잡한 Agent 만들기
더 복잡한 워크플로우가 필요하면 LangGraph를 사용할 수 있습니다.
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator
# 상태 정의
class AgentState(TypedDict):
messages: Annotated[Sequence[str], operator.add]
current_step: str
data: dict
# 노드 함수들
def research_node(state: AgentState):
# 조사 수행
result = researcher_executor.invoke({"input": state['messages'][-1]})
return {
"messages": [result['output']],
"current_step": "research_done",
"data": {"research": result['output']}
}
def analyze_node(state: AgentState):
# 분석 수행
analysis = f"분석 결과: {state['data']['research']}"
return {
"messages": [analysis],
"current_step": "analysis_done",
"data": {**state['data'], "analysis": analysis}
}
def should_continue(state: AgentState):
# 다음 단계 결정
if state['current_step'] == 'research_done':
return "analyze"
return END
# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analyze", analyze_node)
workflow.set_entry_point("research")
workflow.add_conditional_edges(
"research",
should_continue,
{
"analyze": "analyze",
END: END
}
)
workflow.add_edge("analyze", END)
app = workflow.compile()
# 실행
result = app.invoke({
"messages": ["AI 시장 동향을 분석해주세요"],
"current_step": "start",
"data": {}
})실전 활용 사례
고객 지원 Agent
# 고객 DB 조회
class CustomerLookupTool(BaseTool):
name = "customer_lookup"
description = "고객 정보를 조회합니다."
def _run(self, customer_id: str) -> str:
# 실제 DB 조회
return json.dumps({
"name": "홍길동",
"tier": "VIP",
"last_purchase": "2024-01-15"
}, ensure_ascii=False)
# 주문 내역 조회
class OrderHistoryTool(BaseTool):
name = "order_history"
description = "주문 내역을 조회합니다."
def _run(self, customer_id: str) -> str:
return "최근 주문 3건: ..."
# 티켓 생성
class CreateTicketTool(BaseTool):
name = "create_ticket"
description = "CS 티켓을 생성합니다."
def _run(self, issue: str, priority: str) -> str:
return f"티켓 #{12345} 생성 완료"
cs_tools = [CustomerLookupTool(), OrderHistoryTool(), CreateTicketTool()]
cs_agent = create_openai_functions_agent(
llm,
cs_tools,
ChatPromptTemplate.from_messages([
("system", "당신은 고객 지원 담당자입니다. 친절하고 신속하게 문제를 해결하세요."),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
)
cs_executor = AgentExecutor(agent=cs_agent, tools=cs_tools, verbose=True)데이터 분석 Agent
import pandas as pd
class LoadDataTool(BaseTool):
name = "load_data"
description = "CSV 파일에서 데이터를 로드합니다."
def _run(self, filepath: str) -> str:
df = pd.read_csv(filepath)
return df.head().to_string()
class AnalyzeDataTool(BaseTool):
name = "analyze_data"
description = "데이터를 분석하여 통계를 제공합니다."
def _run(self, filepath: str, column: str) -> str:
df = pd.read_csv(filepath)
stats = df[column].describe()
return stats.to_string()
class PlotDataTool(BaseTool):
name = "plot_data"
description = "데이터를 시각화합니다."
def _run(self, filepath: str, x_column: str, y_column: str) -> str:
# 실제로는 차트 생성 후 저장
return f"{x_column} vs {y_column} 차트가 생성되었습니다."
analyst_tools = [LoadDataTool(), AnalyzeDataTool(), PlotDataTool()]모니터링과 디버깅
Agent 실행을 추적하고 모니터링하는 방법입니다.
from langchain.callbacks import StdOutCallbackHandler
from langchain.callbacks.base import BaseCallbackHandler
class CustomCallbackHandler(BaseCallbackHandler):
def on_tool_start(self, serialized, input_str, **kwargs):
print(f"🔧 도구 시작: {serialized['name']}")
print(f"입력: {input_str}")
def on_tool_end(self, output, **kwargs):
print(f"✅ 도구 완료")
print(f"출력: {output}\n")
def on_agent_action(self, action, **kwargs):
print(f"🤖 Agent 행동: {action.tool}")
# 콜백 사용
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
callbacks=[CustomCallbackHandler()],
verbose=True
)보안 고려사항
Agent에게 너무 많은 권한을 주면 위험할 수 있습니다.
# 안전한 도구 설계
class SafeDatabaseTool(BaseTool):
name = "safe_db_query"
description = "읽기 전용 데이터베이스 쿼리"
def _run(self, query: str) -> str:
# SQL 인젝션 방지
if any(keyword in query.upper() for keyword in ['DROP', 'DELETE', 'UPDATE', 'INSERT']):
return "허용되지 않는 쿼리입니다."
# 읽기 전용 실행
# ...
# 사용자 권한 확인
class PermissionTool(BaseTool):
def _run(self, user_id: str, action: str) -> str:
# 권한 체크
if not has_permission(user_id, action):
return "권한이 없습니다."
# ...성능 최적화
Agent 실행 속도를 높이는 방법입니다.
# 도구 결과 캐싱
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
return search.run(query)
# 병렬 도구 실행 (가능한 경우)
from concurrent.futures import ThreadPoolExecutor
def parallel_tool_execution(tools_and_inputs):
with ThreadPoolExecutor(max_workers=3) as executor:
results = executor.map(lambda x: x[0]._run(x[1]), tools_and_inputs)
return list(results)실전 팁
Agent 설계 원칙
-
도구는 단일 책임을 가져야 합니다. 하나의 도구가 너무 많은 일을 하면 Agent가 혼란스러워합니다.
-
도구 설명은 구체적으로 작성하세요. 언제 사용해야 하는지, 입력 형식은 무엇인지 명확히 해야 합니다.
디버깅 전략
-
verbose=True로 설정하여 Agent의 사고 과정을 확인하세요.
-
return_intermediate_steps=True로 각 단계의 결과를 확인할 수 있습니다.
-
LangSmith를 사용하면 전체 실행 과정을 시각적으로 추적할 수 있습니다.
비용 관리
- Agent는 여러 번 LLM을 호출하므로 비용이 높을 수 있습니다. max_iterations로 호출 횟수를 제한하고, 작은 모델(gpt-3.5-turbo)을 사용하거나, 캐싱을 적극 활용하세요.
마무리하며
Agent는 강력하지만 예측하기 어려운 측면도 있습니다. 잘못 설계하면 무한 루프에 빠지거나 엉뚱한 도구를 실행할 수 있죠. 하지만 적절히 제어하면 정말 똑똑한 자동화 시스템을 만들 수 있습니다.
실전에서 Agent를 사용할 때는 작은 것부터 시작하세요. 도구를 하나씩 추가하면서 테스트하고, 프롬프트를 다듬고, 에러 처리를 강화하는 식으로 점진적으로 발전시키는 것이 좋습니다.
다음 편 예고
지금까지 RAG부터 복잡한 체인, 그리고 Agent까지 다뤄봤습니다. 이제 마지막으로 실전 프로젝트를 하나 완성해볼까요?
4편 "실전 프로젝트: 사내 문서 검색 챗봇 구축기"에서는 지금까지 배운 내용을 모두 활용하여 실제 배포 가능한 시스템을 만들어보겠습니다. 문서 수집부터 전처리, RAG 구축, Agent 통합, 그리고 FastAPI로 서비스화하는 전 과정을 다룰 예정입니다.
다음 편에서 뵙겠습니다!
👩🏻💻