LangChain: 미들웨어 동작 뜯어보기, 클로저

https://docs.langchain.com/oss/python/langchain/middleware/overview

Overview - Docs by LangChain

 

Middleware는 LLM 호출 전이나 후, 그리고 LLM 호출을 감싸 state나 request, handler 등을 컨트롤할 수 있게 도와준다. (에이전트 루프에 개입할 수 있음)

미들웨어는 다음과 같은 위치에 훅을 설정할 수 있다:

agent = create_agent( 
    model="...", 
    tools=[...], 
    middleware=[M1, M2, M3] # 순서가 실행 순서
)

Summarization이나 Model fallback 등의 미들웨어를 사용하면서, 미들웨어가 실제로 어떻게 에이전트 루프에 개입할 수 있는지가 궁금해져 코드를 직접 뜯어보게되었다.

https://github.com/langchain-ai/langchain/tree/master/libs/langchain_v1/langchain/agents

langchain/libs/langchain_v1/langchain/agents at master · langchain-ai/langchain

관련 코드는 공식 깃헙의 다음 부분에 위치하고있다.

libs/langchain_v1/langchain/agents/
├── factory.py          ← 미들웨어 체이닝/실행
└── middleware/
    └── (summarization.py, human_in_the_loop.py, 등의 미들웨어)

워낙 자주 바뀌다보니, 글을 읽는 시점에서는 달라졌을 수도 있다.

---

before_model, after_model 같은 경우는, model call 이전/후에 호출되기때문에 그냥 등록된 middleware를 순서대로 순회하면서 before_model 등의 훅이 있다면 순서대로 호출해주면 된다.

이것보다는 wrap_model_call을 더 자세히 파봤다.

결론부터 말하자면, 클로저를 활용해 미들웨어를 합성한다.
보면서 느낀 점은, 스프링 스큐리티의 filter chain 방식과 거의 비슷하다는 것.

factory.py의 _chain_model_call_handlers

    def compose_two(
        outer: _ModelCallHandler[ContextT] | _ComposedModelCallHandler[ContextT],
        inner: _ModelCallHandler[ContextT] | _ComposedModelCallHandler[ContextT],
    ) -> _ComposedModelCallHandler[ContextT]:
        """Compose two handlers where outer wraps inner."""

        def composed(
            request: ModelRequest[ContextT],
            handler: Callable[[ModelRequest[ContextT]], ModelResponse],
        ) -> _ComposedExtendedModelResponse:
            # Closure variable to capture inner's commands before normalizing
            accumulated_commands: list[Command[Any]] = []

            def inner_handler(req: ModelRequest[ContextT]) -> ModelResponse:
                # Clear on each call for retry safety
                accumulated_commands.clear()
                inner_result = inner(req, handler)
                if isinstance(inner_result, _ComposedExtendedModelResponse):
                    accumulated_commands.extend(inner_result.commands)
                    return inner_result.model_response
                if isinstance(inner_result, ExtendedModelResponse):
                    if inner_result.command is not None:
                        accumulated_commands.append(inner_result.command)
                    return inner_result.model_response
                return _normalize_to_model_response(inner_result)

            outer_result = outer(request, inner_handler)
            return _to_composed_result(
                outer_result,
                extra_commands=accumulated_commands or None,
            )
        return composed

이렇게 보면 복잡하니, 다음처럼 조금 더 단순화시켜서 보자.

def compose_two(outer, inner): # inner가 실제 handler 호출과 더 가까움
    def composed(request, handler):
        def inner_handler(req):
            return inner(req, handler) 
        return outer(request, inner_handler) 
    return composed

이 클로저가 2개의 미들웨어(outer, inner)를 합성하는 함수이다.

미들웨어 M1, M2과 사전에 만들어진 실제 LLM 호출을 하는 handler가 있다.

handler

def M1(request, handler): 
    print("M1 전처리") 
    response = handler(request) 
    print("M1 후처리") 
    return response 

def M2(request, handler): 
    print("M2 전처리") 
    response = handler(request) 
    print("M2 후처리") 
    return response

M1 -> M2 -> handler(LLM) 순으로 합성한다고 해보자.

composedM1M2 = compose_two(M1, M2)

composedM1M2는 다음과 같은 함수가 된다:

    def composed(request, handler):
        def inner_handler(req):
            return M2(req, handler) 
        return M1(request, inner_handler)

그럼 이제 실제 composedM1M2가 호출되면, 다음과 같이 된다.

1. composedM1M2(request, handler) 로 호출
2. composedM1M2()가 M1(request, inner_handler)를 호출함
3. M1()이 print("M1 전처리") 후 response = handler(request)를 실행
4. 이때, M1()이 실행한 handler는 2번의 inner_handler임.
5. inner_handler가 호출되면 return M2(req, handler)를 실행함
6. inner_handler는 실제 handler(llm호출)를 실행함

이런 순서다: 
# M1 전처리 
    # M2 전처리 
        # LLM 호출 
    # M2 후처리 
# M1 후처리

이후 위 compose_two를 사용해 모든 미들웨어를 합성한다:

# Compose right-to-left: outer(inner(innermost(handler)))
composed_handler = compose_two(handlers[-2], handlers[-1])
for h in reversed(handlers[:-2]):
    composed_handler = compose_two(h, composed_handler)

return composed_handler # `_chain_model_call_handlers`의 최종 반환

M1,M2,M3를 모두 합성한다면,

step1 = compose_two(M2, M3) # M2가 M3 감쌈 
step2 = compose_two(M1, step1) # M1이 그걸 감쌈

---

간단하게, 미들웨어 하나의 동작에 대해서도 알아보자.
https://github.com/langchain-ai/langchain/blob/master/libs/langchain_v1/langchain/agents/middleware/model_fallback.py

langchain/libs/langchain_v1/langchain/agents/middleware/model_fallback.py at master · langchain-ai/langchain

 

모델 호출 실패 시 fallback처리를 해주는 middleware고, 코드가 100줄 정도로 간단하다.

이 미들웨어의 wrap_model_call 함수를 보자.

    def wrap_model_call(
        self,
        request: ModelRequest[ContextT],
        handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
    ) -> ModelResponse[ResponseT] | AIMessage:
        """Try fallback models in sequence on errors.

        Args:
            request: Initial model request.
            handler: Callback to execute the model.

        Returns:
            AIMessage from successful model call.

        Raises:
            Exception: If all models fail, re-raises last exception.
        """
        # Try primary model first
        last_exception: Exception
        try:
            return handler(request)
        except Exception as e:
            last_exception = e

        # Try fallback models
        for fallback_model in self.models:
            try:
                return handler(request.override(model=fallback_model))
            except Exception as e:
                last_exception = e
                continue

        raise last_exception

 

우선 try: return handler(request)로 핸들러를 호출한다.

이때 handler는 실제 llm호출일 수도, 아니면 다른 합성된 미들웨어일 수도 있다. 이 경우 다른 미들웨어가 먼저 실행된 뒤 실제 handler가 호출될 것이다.

만약 handler에서 예외가 발생하면, fallback model로 모두 시도해보고 다 안되면 last_exception을 넘긴다.

 

 

이 미들웨어를 보면서, 그냥 아무 예외 발생 시 재시도 없이 fallback 모델로 시도하고있어서 비효율적이지 않나 하고 생각도 들었다.
실제로 에이전트를 돌리다보면, 어쩌다 한 번 오류가 발생하고, 다시 호출해보면 정상적으로 되는 경우도 많다.

 

알아보니, 이 문제를 해결하기 위해 실제로 ModelRetryMiddleware도 prebuilt로 제공한다.

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-5.4-mini",
            "claude-3-5-sonnet-20241022",
        ),
        ModelRetryMiddleware( 
            max_retries=3, 
            backoff_factor=2.0, 
            initial_delay=1.0, 
        ),
    ],
)

이렇게 retry가 모두 실패하면 fallback이 다른 모델로 다시 처리하도록(순서 상관있음) 구성해줘도 좋을 것 같다.

Fallback 시작
  Retry: gpt-5.4 최대 3번 재시도 → 다 실패
Fallback: gpt-5.4-mini로 교체
  Retry: gpt-5.4-mini 최대 3번 재시도 → 다 실패
Fallback: claude-3-5-sonnet으로 교체
  Retry: claude 최대 3번 재시도 → 다 실패
예외 던짐

만약 순서가 바뀌면, 모든 모델 fallback 후 retry를 처리하는 비효율적인 구조가 될 수 있다.

 

 

prebuilt의 경우 역할/책임이 명확하게 나눠져있는 것 같긴하지만, retry와 fallback은 하나의 middleware로 처리하고싶은 경우, custom middleware를 직접 구현해 추가해도 좋을 것 같다.