Deep Agents 아키텍처 해부하기

Summary

LangChain의 Deep Agents 아키텍처를 심층 분석하는 고급 강좌. create_agent() 함수의 내부 동작 원리, Middleware 패턴, 그리고 Agent Harness(에이전트 하네스) 개념을 상세히 해부. 단순한 도구 호출을 넘어 고도화된 상태 관리와 복잡한 워크플로우를 구현하는 방법 제시.

Key Claims

Deep Agents — 깊은 도구 접근(Deep Tool Access)과 상태 관리를 지원하는 차세대 에이전트
create_agent() 함수 — Deep Agent 생성의 핵심 API, 복잡한 내부 로직 추상화
Middleware 패턴 — 에이전트 실행 전후에 사용자 정의 로직 삽입 가능
Agent Harness — 에이전트의 일관된 인터페이스, 생명주기, 에러 처리 제공
State Management — LangGraph StateGraph를 활용한 복잡한 상태 추적
Composition — 여러 Deep Agents를 조합하여 더 강력한 시스템 구축

Deep Agents의 핵심 개념

1. create_agent() 함수의 역할

from langchain.agents import create_agent
 
agent = create_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=[tool1, tool2, tool3],
    system_prompt="..."
)

내부 동작:

입력 쿼리
  ↓
LLM이 의도 파악 & 도구 선택
  ↓
도구 실행
  ↓
Middleware (Pre-execution): 실행 전 처리
  ↓
도구 실제 실행
  ↓
Middleware (Post-execution): 실행 후 처리
  ↓
결과 수집 & 다음 단계 결정
  ↓
(반복 또는 종료)

2. Middleware 패턴 (Before/After Hooks)

from langchain.agents.middleware import AgentMiddleware
 
class CustomMiddleware(AgentMiddleware):
    def before_tool_call(self, tool_name, tool_input):
        """도구 실행 전"""
        print(f"Calling {tool_name} with {tool_input}")
        # 입력 검증, 로깅, 권한 확인 등
        return tool_input
    
    def after_tool_call(self, tool_name, result):
        """도구 실행 후"""
        print(f"Result: {result}")
        # 결과 후처리, 캐싱, 알림 등
        return result
 
agent.add_middleware(CustomMiddleware())

활용 사례:

미들웨어	용도
Validation	도구 입력 검증
Logging	실행 기록 추적
Caching	동일 쿼리 캐싱
Security	권한 검증
Monitoring	성능 모니터링
Retry	실패 시 자동 재시도

3. Agent Harness (에이전트 하네스)

from langchain.agents import AgentHarness
 
harness = AgentHarness(agent)
 
# 일관된 인터페이스 제공
result = harness.run(
    query="...",
    max_iterations=10,
    timeout=30,
    on_error="retry"  # 에러 처리 전략
)

Harness의 책임:

✅ 에이전트 생명주기 관리 (초기화 → 실행 → 정리)
✅ 타임아웃 처리
✅ 최대 반복 횟수 제한
✅ 에러 처리 및 복구
✅ 상태 추적 및 로깅

4. State Management (상태 관리)

from langgraph.graph import StateGraph
 
# 상태 정의
class AgentState(TypedDict):
    messages: List[str]
    query: str
    tool_calls: List[ToolCall]
    results: Dict
    current_step: int
 
# 상태 기반 워크플로우
graph = StateGraph(AgentState)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.add_node("evaluate", evaluate_node)
graph.add_node("conclude", conclude_node)
 
graph.add_edge("plan", "execute")
graph.add_edge("execute", "evaluate")
graph.add_conditional_edges(
    "evaluate",
    lambda state: "execute" if not state["success"] else "conclude"
)

상태 추적의 이점:

각 단계의 결과 저장
복잡한 조건부 분기 가능
프로세스 중단 & 재개 가능
전체 실행 이력 추적

Deep Agents vs Simple Agents

특성	Simple Agents	Deep Agents
구현 복잡도	낮음	높음
도구 접근	단순 호출	깊이 있는 조율
상태 관리	제한적	완전 지원
Middleware	없음	완전 지원
오류 처리	기본	고급
확장성	낮음	매우 높음
학습 곡선	완만	가파름

실전 예시: 데이터 분석 Deep Agent

from langchain.agents import create_agent
from langgraph.graph import StateGraph
 
# Step 1: 도구 정의
tools = [
    load_data_tool,
    analyze_data_tool,
    generate_report_tool,
    send_notification_tool
]
 
# Step 2: Deep Agent 생성
agent = create_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=tools,
    system_prompt="데이터 분석 전문 에이전트"
)
 
# Step 3: Middleware 추가
class DataValidationMiddleware:
    def before_tool_call(self, tool_name, tool_input):
        if tool_name == "load_data_tool":
            assert os.path.exists(tool_input["path"])
        return tool_input
 
agent.add_middleware(DataValidationMiddleware())
 
# Step 4: State 정의
class AnalysisState(TypedDict):
    query: str
    data: pd.DataFrame
    analysis_result: Dict
    report: str
    status: str
 
# Step 5: 워크플로우 구성
graph = StateGraph(AnalysisState)
# ... (노드 추가, 엣지 정의)
 
# Step 6: 실행
harness = AgentHarness(graph.compile())
result = harness.run(
    query="2024년 분기별 매출 분석",
    max_iterations=5
)

고급 패턴

1. Nested Agents (중첩 에이전트)

상위 Agent (지휘자)
  ├─ 데이터 Agent (로드, 정제)
  ├─ 분석 Agent (통계, 시각화)
  └─ 리포트 Agent (문서 생성)

2. Conditional Routing (조건부 라우팅)

def route_to_agent(state):
    if state["query_type"] == "financial":
        return "financial_agent"
    elif state["query_type"] == "technical":
        return "technical_agent"
    else:
        return "general_agent"
 
graph.add_conditional_edges("router", route_to_agent)

3. Feedback Loops (피드백 루프)

graph.add_edge("evaluate", "refine")  # 불만족 시 개선
graph.add_edge("refine", "execute")   # 재실행

한계 및 주의점

복잡성 증가:

코드 양증가 (수백 줄)
디버깅 어려움
테스트 복잡성 증가

성능:

반복이 많으면 지연 시간 증가
API 호출 비용 증가

신뢰성:

Middleware 오류가 전체 파이프라인 중단 가능
상태 관리 실수로 인한 버그

JYP Garden

탐색기