각 단계에서 에이전트 실행을 제어하고 커스터마이징
Middleware는 에이전트 내부에서 발생하는 일을 더욱 세밀하게 제어할 수 있는 방법을 제공합니다.
핵심 에이전트 루프는 모델을 호출하고, 모델이 실행할 도구를 선택하게 한 다음, 더 이상 도구를 호출하지 않을 때 종료하는 과정을 포함합니다:
Middleware는 이러한 각 단계 전후에 훅(hook)을 노출합니다:
Middleware가 할 수 있는 일은?
Monitor (모니터링)
로깅, 분석 및 디버깅을 통해 에이전트 동작을 추적합니다
Modify (수정)
프롬프트, 도구 선택 및 출력 형식을 변환합니다
Control (제어)
재시도, 폴백 및 조기 종료 로직을 추가합니다
Enforce (강제 적용)
속도 제한, 가드레일 및 PII 탐지를 적용합니다
Middleware는 create_agent에 전달하여 추가할 수 있습니다:
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware
agent = create_agent(
model="gpt-4o",
tools=[...],
middleware=[SummarizationMiddleware(), HumanInTheLoopMiddleware()],
)내장 Middleware
LangChain은 일반적인 사용 사례를 위한 사전 구축된 middleware를 제공합니다:
Summarization (요약)
토큰 제한에 도달할 때 대화 기록을 자동으로 요약합니다.
다음과 같은 경우에 완벽:
- 컨텍스트 윈도우를 초과하는 장기 실행 대화
- 광범위한 기록이 있는 다회차 대화
- 전체 대화 컨텍스트를 보존하는 것이 중요한 애플리케이션
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model="gpt-4o",
tools=[weather_tool, calculator_tool],
middleware=[
SummarizationMiddleware(
model="gpt-4o-mini",
max_tokens_before_summary=4000, # 4000 토큰에서 요약 트리거
messages_to_keep=20, # 요약 후 최근 20개 메시지 유지
summary_prompt="Custom prompt for summarization...", # 선택사항
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| model | string (필수) | 요약을 생성하기 위한 모델 |
| max_tokens_before_summary | number | 요약을 트리거하는 토큰 임계값 |
| messages_to_keep | number (기본값: "20") | 보존할 최근 메시지 |
| token_counter | function | 사용자 정의 토큰 계산 함수. 기본적으로 문자 기반 계산을 사용합니다. |
| summary_prompt | string | 사용자 정의 프롬프트 템플릿. 지정하지 않으면 내장 템플릿을 사용합니다. |
| summary_prefix | string (기본값: "## Previous conversation summary:") | 요약 메시지의 접두사 |
Human-in-the-loop (휴먼 인 더 루프)
도구 호출이 실행되기 전에 사람의 승인, 편집 또는 거부를 위해 에이전트 실행을 일시 중지합니다.
다음과 같은 경우에 사용:
- 사람의 승인이 필요한 고위험 작업(데이터베이스 쓰기, 금융 거래)
- 사람의 감독이 의무인 규정 준수 워크플로우
- 사람의 피드백이 에이전트를 안내하는 데 사용되는 장기 실행 대화
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model="gpt-4o",
tools=[read_email_tool, send_email_tool],
checkpointer=InMemorySaver(),
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
# 이메일 전송에 대한 승인, 편집 또는 거부 필요
"send_email_tool": {
"allowed_decisions": ["approve", "edit", "reject"],
},
# 이메일 읽기는 자동 승인
"read_email_tool": False,
}
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| interrupt_on | dict (필수) | 도구 이름과 승인 설정의 매핑. 값은 True(기본 설정으로 중단), False(자동 승인) 또는 InterruptOnConfig 객체일 수 있습니다. |
| description_prefix | string (기본값: "Tool execution requires approval") | 작업 요청 설명의 접두사 |
InterruptOnConfig 옵션:
| 매개변수 | 타입 | 설명 |
|---|---|---|
| allowed_decisions | list[string] | 허용되는 결정 목록: "approve", "edit" 또는 "reject" |
| description | string | callable | 사용자 정의 설명을 위한 정적 문자열 또는 호출 가능 함수 |
중요: Human-in-the-loop middleware는 중단 간 상태를 유지하기 위해
checkpointer가 필요합니다.완전한 예제와 통합 패턴은 human-in-the-loop 문서를 참조하세요.
Anthropic prompt caching (Anthropic 프롬프트 캐싱)
Anthropic 모델에서 반복적인 프롬프트 접두사를 캐싱하여 비용을 절감합니다.
다음과 같은 경우에 사용:
- 길고 반복되는 시스템 프롬프트가 있는 애플리케이션
- 호출 간 동일한 컨텍스트를 재사용하는 에이전트
- 대량 배포에서 API 비용 절감
Anthropic Prompt Caching 전략 및 제한 사항에 대해 자세히 알아보세요.
from langchain_anthropic import ChatAnthropic
from langchain_anthropic.middleware import AnthropicPromptCachingMiddleware
from langchain.agents import create_agent
LONG_PROMPT = """
Please be a helpful assistant.
<Lots more context ...>
"""
agent = create_agent(
model=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
system_prompt=LONG_PROMPT,
middleware=[AnthropicPromptCachingMiddleware(ttl="5m")],
)
# 캐시 저장
agent.invoke({"messages": [HumanMessage("Hi, my name is Bob")]})
# 캐시 히트, 시스템 프롬프트가 캐시됨
agent.invoke({"messages": [HumanMessage("What's my name?")]})설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| type | string (기본값: "ephemeral") | 캐시 타입. 현재 "ephemeral"만 지원됩니다. |
| ttl | string (기본값: "5m") | 캐시된 콘텐츠의 유지 시간. 유효한 값: "5m" 또는 "1h" |
| min_messages_to_cache | number (기본값: "0") | 캐싱 시작 전 최소 메시지 수 |
| unsupported_model_behavior | string (기본값: "warn") | 비-Anthropic 모델 사용 시 동작. 옵션: "ignore", "warn" 또는 "raise" |
Model call limit (모델 호출 제한)
무한 루프나 과도한 비용을 방지하기 위해 모델 호출 수를 제한합니다.
다음과 같은 경우에 완벽:
- 너무 많은 API 호출을 하는 폭주 에이전트 방지
- 프로덕션 배포에 대한 비용 제어 시행
- 특정 호출 예산 내에서 에이전트 동작 테스트
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
agent = create_agent(
model="gpt-4o",
tools=[...],
middleware=[
ModelCallLimitMiddleware(
thread_limit=10, # 스레드당 최대 10회 호출(실행 전체)
run_limit=5, # 실행당 최대 5회 호출(단일 호출)
exit_behavior="end", # 또는 예외를 발생시키려면 "error"
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| thread_limit | number | 스레드의 모든 실행에서 최대 모델 호출 수. 기본값은 제한 없음입니다. |
| run_limit | number | 단일 호출당 최대 모델 호출 수. 기본값은 제한 없음입니다. |
| exit_behavior | string (기본값: "end") | 제한에 도달했을 때의 동작. 옵션: "end"(정상 종료) 또는 "error"(예외 발생) |
Tool call limit (도구 호출 제한)
특정 도구 또는 모든 도구에 대한 도구 호출 수를 제한합니다.
다음과 같은 경우에 사용:
- 값비싼 외부 API에 대한 과도한 호출 방지
- 웹 검색 또는 데이터베이스 쿼리 제한
- 특정 도구 사용에 대한 속도 제한 시행
from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware
# 모든 도구 호출 제한
global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)
# 특정 도구 제한
search_limiter = ToolCallLimitMiddleware(
tool_name="search",
thread_limit=5,
run_limit=3,
)
agent = create_agent(
model="gpt-4o",
tools=[...],
middleware=[global_limiter, search_limiter],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| tool_name | string | 제한할 특정 도구. 제공하지 않으면 모든 도구에 제한이 적용됩니다. |
| thread_limit | number | 스레드의 모든 실행에서 최대 도구 호출 수. 기본값은 제한 없음입니다. |
| run_limit | number | 단일 호출당 최대 도구 호출 수. 기본값은 제한 없음입니다. |
| exit_behavior | string (기본값: "end") | 제한에 도달했을 때의 동작. 옵션: "end"(정상 종료) 또는 "error"(예외 발생) |
Model fallback (모델 폴백)
기본 모델이 실패할 때 자동으로 대체 모델로 폴백합니다.
다음과 같은 경우에 완벽:
- 모델 중단을 처리하는 복원력 있는 에이전트 구축
- 더 저렴한 모델로 폴백하여 비용 최적화
- OpenAI, Anthropic 등의 제공자 이중화
from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware
agent = create_agent(
model="gpt-4o", # 기본 모델
tools=[...],
middleware=[
ModelFallbackMiddleware(
"gpt-4o-mini", # 오류 시 먼저 시도
"claude-3-5-sonnet-20241022", # 그 다음 이것
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| first_model | string | BaseChatModel (필수) | 기본 모델이 실패할 때 시도할 첫 번째 폴백 모델. 모델 문자열(예: "openai:gpt-4o-mini") 또는 BaseChatModel 인스턴스일 수 있습니다. |
| *additional_models | string | BaseChatModel | 이전 모델이 실패한 경우 순서대로 시도할 추가 폴백 모델 |
PII detection (PII 탐지)
대화에서 개인 식별 정보를 탐지하고 처리합니다.
다음과 같은 경우에 사용:
- 규정 준수 요구 사항이 있는 의료 및 금융 애플리케이션
- 로그를 정리해야 하는 고객 서비스 에이전트
- 민감한 사용자 데이터를 처리하는 모든 애플리케이션
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
agent = create_agent(
model="gpt-4o",
tools=[...],
middleware=[
# 사용자 입력에서 이메일 삭제
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# 신용 카드 마스킹(마지막 4자리 표시)
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
# 정규식을 사용한 사용자 정의 PII 타입
PIIMiddleware(
"api_key",
detector=r"sk-[a-zA-Z0-9]{32}",
strategy="block", # 탐지되면 오류 발생
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| pii_type | string (필수) | 탐지할 PII 타입. 내장 타입(email, credit_card, ip, mac_address, url) 또는 사용자 정의 타입 이름일 수 있습니다. |
| strategy | string (기본값: "redact") | 탐지된 PII를 처리하는 방법. 옵션: "block" - 탐지 시 예외 발생, "redact" - [REDACTED_TYPE]으로 대체, "mask" - 부분 마스킹(예: ****-****-****-1234), "hash" - 결정론적 해시로 대체 |
| detector | function | regex | 사용자 정의 탐지 함수 또는 정규식 패턴. 제공하지 않으면 PII 타입에 대한 내장 탐지기를 사용합니다. |
| apply_to_input | boolean (기본값: "True") | 모델 호출 전 사용자 메시지 확인 |
| apply_to_output | boolean (기본값: "False") | 모델 호출 후 AI 메시지 확인 |
| apply_to_tool_results | boolean (기본값: "False") | 실행 후 도구 결과 메시지 확인 |
To-do list
복잡한 multi-step 작업을 위해 에이전트에 task planning 및 tracking 기능을 갖추세요.
다음에 특히 적합합니다:
- 여러 tools 간 조율이 필요한 복잡한 multi-step 작업
- 진행 상황의 가시성이 중요한 long-running operations
사람이 할 일(to-do)을 적고 추적할수록 더 효율적인 것처럼, agents도 복잡한 문제를 쪼개고, 새 정보에 따라 계획을 수정하며, 워크플로우의 투명성을 높이기 위해 structured task management의 이점을 얻습니다.
예를 들어 Claude Code가 복잡하고 다단계의 작업을 진행하기 전에 to-do list를 먼저 작성하는 패턴을 본 적이 있을 겁니다.
이 middleware는 에이전트에게 write_todos tool과 효과적인 task planning을 유도하는 system prompts를 자동으로 제공합니다.
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
@tool
def read_file(file_path: str) -> str:
"""Read contents of a file."""
with open(file_path) as f:
return f.read()
@tool
def write_file(file_path: str, content: str) -> str:
"""Write content to a file."""
with open(file_path, 'w') as f:
f.write(content)
return f"Wrote {len(content)} characters to {file_path}"
@tool
def run_tests(test_path: str) -> str:
"""Run tests and return results."""
# Simplified for example
return "All tests passed!"
agent = create_agent(
model="gpt-4o",
tools=[read_file, write_file, run_tests],
middleware=[TodoListMiddleware()],
)
result = agent.invoke({
"messages": [HumanMessage("Refactor the authentication module to use async/await and ensure all tests pass")]
})
# The agent will use write_todos to plan and track:
# 1. Read current authentication module code
# 2. Identify functions that need async conversion
# 3. Refactor functions to async/await
# 4. Update function calls throughout codebase
# 5. Run tests and fix any failures
print(result["todos"]) # Track the agent's progress through each stepLLM tool selector (LLM 도구 선택기)
메인 모델을 호출하기 전에 LLM을 사용하여 관련 도구를 지능적으로 선택합니다.
다음과 같은 경우에 유용:
- 대부분이 쿼리당 관련이 없는 많은 도구(10개 이상)가 있는 에이전트
- 관련 없는 도구를 필터링하여 토큰 사용량 감소
- 모델 집중도 및 정확도 향상
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
agent = create_agent(
model="gpt-4o",
tools=[tool1, tool2, tool3, tool4, tool5, ...], # 많은 도구
middleware=[
LLMToolSelectorMiddleware(
model="gpt-4o-mini", # 선택을 위해 더 저렴한 모델 사용
max_tools=3, # 가장 관련성 높은 3개 도구로 제한
always_include=["search"], # 항상 특정 도구 포함
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| model | string | BaseChatModel | 도구 선택을 위한 모델. 모델 문자열 또는 BaseChatModel 인스턴스일 수 있습니다. 기본값은 에이전트의 메인 모델입니다. |
| system_prompt | string | 선택 모델에 대한 지침. 지정하지 않으면 내장 프롬프트를 사용합니다. |
| max_tools | number | 선택할 최대 도구 수. 기본값은 제한 없음입니다. |
| always_include | list[string] | 선택에 항상 포함할 도구 이름 목록 |
Tool retry (도구 재시도)
설정 가능한 지수 백오프로 실패한 도구 호출을 자동으로 재시도합니다.
다음과 같은 경우에 유용:
- 외부 API 호출의 일시적 실패 처리
- 네트워크 종속 도구의 안정성 향상
- 일시적 오류를 우아하게 처리하는 복원력 있는 에이전트 구축
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
agent = create_agent(
model="gpt-4o",
tools=[search_tool, database_tool],
middleware=[
ToolRetryMiddleware(
max_retries=3, # 최대 3번 재시도
backoff_factor=2.0, # 지수 백오프 승수
initial_delay=1.0, # 1초 지연으로 시작
max_delay=60.0, # 지연을 60초로 제한
jitter=True, # thundering herd를 피하기 위해 무작위 지터 추가
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| max_retries | number (기본값: "2") | 초기 호출 후 최대 재시도 시도 횟수(기본값으로 총 3회 시도) |
| tools | list[BaseTool | str] | 재시도 로직을 적용할 도구 또는 도구 이름의 선택적 목록. None이면 모든 도구에 적용됩니다. |
| retry_on | tuple[type[Exception], ...] | callable (기본값: "(Exception,)") | 재시도할 예외 타입의 튜플 또는 예외를 받아 재시도해야 하면 True를 반환하는 호출 가능 객체입니다. |
| on_failure | string | callable (기본값: "return_message") | 모든 재시도가 소진되었을 때의 동작. 옵션: "return_message" - 오류 세부 정보가 있는 ToolMessage 반환(LLM이 실패를 처리할 수 있음), "raise" - 예외 다시 발생(에이전트 실행 중지), Custom callable - 예외를 받아 ToolMessage 콘텐츠의 문자열을 반환하는 함수 |
| backoff_factor | number (기본값: "2.0") | 지수 백오프의 승수. 각 재시도는 initial_delay * (backoff_factor ** retry_number) 초 동안 대기합니다. 일정한 지연을 위해 0.0으로 설정합니다. |
| initial_delay | number (기본값: "1.0") | 첫 번째 재시도 전 초기 지연(초) |
| max_delay | number (기본값: "60.0") | 재시도 간 최대 지연 시간(초)(지수 백오프 증가 제한) |
| jitter | boolean (기본값: "true") | thundering herd를 피하기 위해 지연에 무작위 지터(±25%)를 추가할지 여부 |
LLM tool emulator (LLM 도구 에뮬레이터)
테스트 목적으로 LLM을 사용하여 도구 실행을 에뮬레이트하고, 실제 도구 호출을 AI가 생성한 응답으로 대체합니다.
다음과 같은 경우에 유용:
- 실제 도구를 실행하지 않고 에이전트 동작 테스트
- 외부 도구를 사용할 수 없거나 비용이 많이 들 때 에이전트 개발
- 실제 도구를 구현하기 전에 에이전트 워크플로 프로토타이핑
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
agent = create_agent(
model="gpt-4o",
tools=[get_weather, search_database, send_email],
middleware=[
# 기본적으로 모든 도구 에뮬레이트
LLMToolEmulator(),
# 또는 특정 도구 에뮬레이트
# LLMToolEmulator(tools=["get_weather", "search_database"]),
# 또는 에뮬레이션을 위한 사용자 정의 모델 사용
# LLMToolEmulator(model="claude-sonnet-4-5-20250929"),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| tools | list[str | BaseTool] | 에뮬레이트할 도구 이름(str) 또는 BaseTool 인스턴스의 목록. None(기본값)이면 모든 도구가 에뮬레이트됩니다. 빈 목록이면 에뮬레이트되는 도구가 없습니다. |
| model | string | BaseChatModel (기본값: "anthropic:claude-3-5-sonnet-latest") | 에뮬레이트된 도구 응답을 생성하는 데 사용할 모델. 모델 식별자 문자열 또는 BaseChatModel 인스턴스일 수 있습니다. |
Context editing (컨텍스트 편집)
도구 사용을 자르거나, 요약하거나, 지워서 대화 컨텍스트를 관리합니다.
다음과 같은 경우에 유용:
- 주기적인 컨텍스트 정리가 필요한 긴 대화
- 컨텍스트에서 실패한 도구 시도 제거
- 사용자 정의 컨텍스트 관리 전략
from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit
agent = create_agent(
model="gpt-4o",
tools=[...],
middleware=[
ContextEditingMiddleware(
edits=[
ClearToolUsesEdit(trigger=1000), # 오래된 도구 사용 지우기
],
),
],
)설정 옵션
| 매개변수 | 타입 | 설명 |
|---|---|---|
| edits | list[ContextEdit] (기본값: "[ClearToolUsesEdit()]") | 적용할 ContextEdit 전략 목록 |
| token_count_method | string (기본값: "approximate") | 토큰 계산 방법. 옵션: "approximate" 또는 "model" |
ClearToolUsesEdit 옵션:
| 매개변수 | 타입 | 설명 |
|---|---|---|
| trigger | number (기본값: "100000") | 편집을 트리거하는 토큰 수 |
| clear_at_least | number (기본값: "0") | 회수할 최소 토큰 |
| keep | number (기본값: "3") | 보존할 최근 도구 결과 수 |
| clear_tool_inputs | boolean (기본값: "False") | 도구 호출 매개변수를 지울지 여부 |
| exclude_tools | list[string] (기본값: "()") | 지우기에서 제외할 도구 이름 목록 |
| placeholder | string (기본값: "[cleared]") | 지워진 출력의 자리표시자 텍스트 |
커스텀 Middleware
에이전트 실행 흐름의 특정 지점에서 실행되는 훅을 구현하여 커스텀 middleware를 구축할 수 있습니다.
Middleware는 두 가지 방법으로 생성할 수 있습니다:
- Decorator 기반 - 단일 훅 middleware를 위한 빠르고 간단한 방법
- Class 기반 - 여러 훅이 있는 복잡한 middleware를 위한 더 강력한 방법
Decorator 기반 Middleware
단일 훅만 필요한 간단한 middleware의 경우, 데코레이터가 기능을 추가하는 가장 빠른 방법을 제공합니다:
from langchain.agents.middleware import before_model, after_model, wrap_model_call
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse, dynamic_prompt
from langchain.messages import AIMessage
from langchain.agents import create_agent
from langgraph.runtime import Runtime
from typing import Any, Callable
# Node-style: 모델 호출 전 로깅
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"About to call model with {len(state['messages'])} messages")
return None
# Node-style: 모델 호출 후 검증
@after_model(can_jump_to=["end"])
def validate_output(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
last_message = state["messages"][-1]
if "BLOCKED" in last_message.content:
return {
"messages": [AIMessage("I cannot respond to that request.")],
"jump_to": "end"
}
return None
# Wrap-style: 재시도 로직
@wrap_model_call
def retry_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
for attempt in range(3):
try:
return handler(request)
except Exception as e:
if attempt == 2:
raise
print(f"Retry {attempt + 1}/3 after error: {e}")
# Wrap-style: 동적 프롬프트
@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
user_id = request.runtime.context.get("user_id", "guest")
return f"You are a helpful assistant for user {user_id}. Be concise and friendly."
# 에이전트에서 데코레이터 사용
agent = create_agent(
model="gpt-4o",
middleware=[log_before_model, validate_output, retry_model, personalized_prompt],
tools=[...],
)사용 가능한 데코레이터
Node-style (특정 실행 지점에서 실행):
@before_agent- 에이전트 시작 전(호출당 한 번)@before_model- 각 모델 호출 전@after_model- 각 모델 응답 후@after_agent- 에이전트 완료 후(호출당 최대 한 번)
Wrap-style (실행을 가로채고 제어):
@wrap_model_call- 각 모델 호출 주변@wrap_tool_call- 각 도구 호출 주변
편의 데코레이터:
@dynamic_prompt- 동적 시스템 프롬프트 생성(프롬프트를 수정하는@wrap_model_call과 동등)
데코레이터를 사용하는 경우
데코레이터를 사용하는 경우:
- 단일 훅이 필요한 경우
- 복잡한 설정이 없는 경우
클래스를 사용하는 경우:
- 여러 훅이 필요한 경우
- 복잡한 설정이 있는 경우
- 프로젝트 전체에서 재사용(초기화 시 설정)
Class 기반 Middleware
두 가지 훅 스타일
Node-style 훅:
특정 실행 지점에서 순차적으로 실행됩니다. 로깅, 검증 및 상태 업데이트에 사용합니다.
Wrap-style 훅:
핸들러 호출에 대한 완전한 제어로 실행을 가로챕니다. 재시도, 캐싱 및 변환에 사용합니다.
Node-style 훅
실행 흐름의 특정 지점에서 실행됩니다:
before_agent- 에이전트 시작 전(호출당 한 번)before_model- 각 모델 호출 전after_model- 각 모델 응답 후after_agent- 에이전트 완료 후(호출당 최대 한 번)
예제: 로깅 middleware
from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime
from typing import Any
class LoggingMiddleware(AgentMiddleware):
def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"About to call model with {len(state['messages'])} messages")
return None
def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"Model returned: {state['messages'][-1].content}")
return None예제: 대화 길이 제한
from langchain.agents.middleware import AgentMiddleware, AgentState
from langchain.messages import AIMessage
from langgraph.runtime import Runtime
from typing import Any
class MessageLimitMiddleware(AgentMiddleware):
def __init__(self, max_messages: int = 50):
super().__init__()
self.max_messages = max_messages
def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
if len(state["messages"]) == self.max_messages:
return {
"messages": [AIMessage("Conversation limit reached.")],
"jump_to": "end"
}
return NoneWrap-style 훅
실행을 가로채고 핸들러가 호출되는 시기를 제어합니다:
wrap_model_call- 각 모델 호출 주변wrap_tool_call- 각 도구 호출 주변
핸들러가 0번(단락), 1번(정상 흐름) 또는 여러 번(재시도 로직) 호출되는지 결정합니다.
예제: 모델 재시도 middleware
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from typing import Callable
class RetryMiddleware(AgentMiddleware):
def __init__(self, max_retries: int = 3):
super().__init__()
self.max_retries = max_retries
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
for attempt in range(self.max_retries):
try:
return handler(request)
except Exception as e:
if attempt == self.max_retries - 1:
raise
print(f"Retry {attempt + 1}/{self.max_retries} after error: {e}")예제: 동적 모델 선택
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain.chat_models import init_chat_model
from typing import Callable
class DynamicModelMiddleware(AgentMiddleware):
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
# 대화 길이에 따라 다른 모델 사용
if len(request.messages) > 10:
request.model = init_chat_model("gpt-4o")
else:
request.model = init_chat_model("gpt-4o-mini")
return handler(request)예제: 도구 호출 모니터링
from langchain.tools.tool_node import ToolCallRequest
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from typing import Callable
class ToolMonitoringMiddleware(AgentMiddleware):
def wrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
print(f"Executing tool: {request.tool_call['name']}")
print(f"Arguments: {request.tool_call['args']}")
try:
result = handler(request)
print(f"Tool completed successfully")
return result
except Exception as e:
print(f"Tool failed: {e}")
raise커스텀 상태 스키마
Middleware는 커스텀 속성으로 에이전트의 상태를 확장할 수 있습니다. 커스텀 상태 타입을 정의하고 state_schema로 설정합니다:
from langchain.agents.middleware import AgentState, AgentMiddleware
from typing_extensions import NotRequired
from typing import Any
class CustomState(AgentState):
model_call_count: NotRequired[int]
user_id: NotRequired[str]
class CallCounterMiddleware(AgentMiddleware[CustomState]):
state_schema = CustomState
def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
# 커스텀 상태 속성 접근
count = state.get("model_call_count", 0)
if count > 10:
return {"jump_to": "end"}
return None
def after_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
# 커스텀 상태 업데이트
return {"model_call_count": state.get("model_call_count", 0) + 1}agent = create_agent(
model="gpt-4o",
middleware=[CallCounterMiddleware()],
tools=[...],
)
# 커스텀 상태로 호출
result = agent.invoke({
"messages": [HumanMessage("Hello")],
"model_call_count": 0,
"user_id": "user-123",
})실행 순서
여러 middleware를 사용할 때 실행 순서를 이해하는 것이 중요합니다:
agent = create_agent(
model="gpt-4o",
middleware=[middleware1, middleware2, middleware3],
tools=[...],
)실행 흐름:
Before 훅은 순서대로 실행됩니다:
middleware1.before_agent()middleware2.before_agent()middleware3.before_agent()- 에이전트 루프 시작
middleware1.before_model()middleware2.before_model()middleware3.before_model()
Wrap 훅은 함수 호출처럼 중첩됩니다:
middleware1.wrap_model_call()→middleware2.wrap_model_call()→middleware3.wrap_model_call()→ 모델
After 훅은 역순으로 실행됩니다:
middleware3.after_model()middleware2.after_model()middleware1.after_model()- 에이전트 루프 종료
middleware3.after_agent()middleware2.after_agent()middleware1.after_agent()
핵심 규칙:
before_*훅: 처음부터 마지막까지after_*훅: 마지막부터 처음까지(역순)wrap_*훅: 중첩됨(첫 번째 middleware가 다른 모든 것을 감쌈)
에이전트 점프
Middleware에서 조기에 종료하려면 jump_to가 있는 딕셔너리를 반환합니다:
class EarlyExitMiddleware(AgentMiddleware):
def before_model(self, state: AgentState, runtime) -> dict[str, Any] | None:
# 일부 조건 확인
if should_exit(state):
return {
"messages": [AIMessage("Exiting early due to condition.")],
"jump_to": "end"
}
return None사용 가능한 점프 대상:
"end": 에이전트 실행의 끝으로 점프"tools": 도구 노드로 점프"model": 모델 노드로 점프(또는 첫 번째 before_model 훅)
중요: before_model 또는 after_model에서 점프할 때, "model"로 점프하면 모든 before_model middleware가 다시 실행됩니다.
점프를 활성화하려면 @hook_config(can_jump_to=[...])로 훅을 데코레이트합니다:
from langchain.agents.middleware import AgentMiddleware, hook_config
from typing import Any
class ConditionalMiddleware(AgentMiddleware):
@hook_config(can_jump_to=["end", "tools"])
def after_model(self, state: AgentState, runtime) -> dict[str, Any] | None:
if some_condition(state):
return {"jump_to": "end"}
return None모범 사례
- Middleware를 집중적으로 유지 - 각각은 한 가지 일을 잘해야 합니다
- 오류를 우아하게 처리 - middleware 오류로 인해 에이전트가 충돌하지 않도록 합니다
- 적절한 훅 타입 사용:
- 순차적 로직을 위한 Node-style(로깅, 검증)
- 제어 흐름을 위한 Wrap-style(재시도, 폴백, 캐싱)
- 커스텀 상태 속성을 명확하게 문서화합니다
- 통합하기 전에 middleware를 독립적으로 단위 테스트합니다
- 실행 순서를 고려 - 중요한 middleware를 목록의 첫 번째에 배치합니다
- 가능하면 내장 middleware를 사용하고, 바퀴를 재발명하지 마세요
예제
동적으로 도구 선택
런타임에 관련 도구를 선택하여 성능과 정확도를 향상시킵니다.
이점:
- 더 짧은 프롬프트 - 관련 도구만 노출하여 복잡성 감소
- 더 나은 정확도 - 모델이 더 적은 옵션에서 올바르게 선택
- 권한 제어 - 사용자 액세스에 따라 동적으로 도구 필터링
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest
from typing import Callable
class ToolSelectorMiddleware(AgentMiddleware):
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""상태/컨텍스트에 따라 관련 도구를 선택하는 middleware."""
# 상태/컨텍스트에 따라 작고 관련성 높은 도구 하위 집합 선택
relevant_tools = select_relevant_tools(request.state, request.runtime)
request.tools = relevant_tools
return handler(request)
agent = create_agent(
model="gpt-4o",
tools=all_tools, # 모든 사용 가능한 도구를 사전에 등록해야 합니다
# Middleware를 사용하여 주어진 실행에 관련된 더 작은 하위 집합을 선택할 수 있습니다.
middleware=[ToolSelectorMiddleware()],
)확장 예제: GitHub vs GitLab 도구 선택
from dataclasses import dataclass
from typing import Literal, Callable
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain_core.tools import tool
@tool
def github_create_issue(repo: str, title: str) -> dict:
"""GitHub 저장소에 이슈를 생성합니다."""
return {"url": f"https://github.com/{repo}/issues/1", "title": title}
@tool
def gitlab_create_issue(project: str, title: str) -> dict:
"""GitLab 프로젝트에 이슈를 생성합니다."""
return {"url": f"https://gitlab.com/{project}/-/issues/1", "title": title}
all_tools = [github_create_issue, gitlab_create_issue]
@dataclass
class Context:
provider: Literal["github", "gitlab"]
class ToolSelectorMiddleware(AgentMiddleware):
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""VCS 제공자에 따라 도구를 선택합니다."""
provider = request.runtime.context.provider
if provider == "gitlab":
selected_tools = [t for t in request.tools if t.name == "gitlab_create_issue"]
else:
selected_tools = [t for t in request.tools if t.name == "github_create_issue"]
request.tools = selected_tools
return handler(request)
agent = create_agent(
model="gpt-4o",
tools=all_tools,
middleware=[ToolSelectorMiddleware()],
context_schema=Context,
)
# GitHub 컨텍스트로 호출
agent.invoke(
{
"messages": [{"role": "user", "content": "Open an issue titled 'Bug: where are the cats' in the repository `its-a-cats-game`"}]
},
context=Context(provider="github"),
)핵심 사항:
- 모든 도구를 사전에 등록
- Middleware가 요청당 관련 하위 집합 선택
- 설정 요구 사항을 위해
context_schema사용
추가 리소스
- Middleware API 참조 - 커스텀 middleware에 대한 완전한 가이드
- Human-in-the-loop - 민감한 작업에 대한 사람의 검토 추가
- 에이전트 테스트 - 안전 메커니즘 테스트를 위한 전략
출처: https://docs.langchain.com/oss/python/langchain/middleware
Langchain v1.0
- LangChain 개요
- LangChain v1
- LangChain v1 마이그레이션 가이드
- LangChain 설치
- QuickStart
- Philosophy
- Agents
- Models
- Messages
- Tools
- Short-term memory
- Streaming
Middleware- Structured output
- Guardrails
- Runtime
- Context Engineering
- Model Context Protocol (MCP)
- Human-in-the-loop
- Multi-agent
- Retrieval
- Long-term memory
- Studio
- Test
- Deploy
- Agent Chat UI
- Observability
반응형
