LangChain Agents 구현

LangChain Agents 구현 가이드

LangChain을 사용하여 도구를 활용하는 AI Agent를 구현하는 방법을 알아봅니다.

Agent 기본 구현

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from langchain.tools import tool

# 1. 도구 정의
@tool
def search(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    # 실제 검색 로직
    return f"검색 결과: {query}"

@tool
def calculator(expression: str) -> str:
    """수학 계산을 수행합니다."""
    try:
        return str(eval(expression))
    except:
        return "계산 오류"

# 2. LLM 설정
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 3. 프롬프트 로드
prompt = hub.pull("hwchase17/react")

# 4. Agent 생성
tools = [search, calculator]
agent = create_react_agent(llm, tools, prompt)

# 5. 실행기 생성
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True
)

# 6. 실행
result = agent_executor.invoke({
    "input": "서울의 인구수에 2를 곱하면?"
})

Tool Calling Agent (권장)

from langchain.agents import create_tool_calling_agent

# OpenAI Function Calling 사용
agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

Structured Chat Agent

from langchain.agents import create_structured_chat_agent

# 복잡한 입력을 처리하는 Agent
agent = create_structured_chat_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10
)

커스텀 도구 고급 패턴

복잡한 입력 스키마

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="검색할 내용")
    max_results: int = Field(default=5, description="최대 결과 수")
    language: str = Field(default="ko", description="언어 코드")

@tool(args_schema=SearchInput)
def advanced_search(query: str, max_results: int, language: str) -> str:
    """고급 웹 검색을 수행합니다."""
    return perform_search(query, max_results, language)

비동기 도구

@tool
async def async_fetch(url: str) -> str:
    """URL의 내용을 비동기로 가져옵니다."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Agent 메모리

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=5,  # 최근 5개 대화만 유지
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

에러 처리

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,  # 파싱 오류 자동 처리
    max_iterations=15,            # 최대 반복 횟수
    max_execution_time=60,        # 최대 실행 시간 (초)
    early_stopping_method="force" # 강제 중단
)