Agent 루프에 코드를 끼워 넣는 여섯 곳, Middleware

공항에서 비행기를 타려면 여러 검문대를 지납니다. 짐 검사를 하고 신분증을 확인하고 탑승구에서 표를 봅니다. 비행기를 고치지 않고도 길목마다 검사가 들어갑니다.

Agent 루프에도 같은 일이 필요합니다. 입력에 개인정보가 있으면 가리고 모델을 너무 많이 부르면 멈추고 위험한 요청은 아예 막아야 합니다. 루프 코드를 고쳐서 이런 일을 넣으면 코드가 금방 복잡해집니다. LangChain에는 루프 자리마다 코드를 끼우는 Middleware가 있습니다. 어느 검문대에 끼울 수 있는지 아래 지도에서 확인합니다.

끼울 코드

한 번의 실행을 위에서 아래로 펴 놓은 지도입니다. 검문대 이름은 LangChain 문서 기준이고 검사관이 서 있는 곳이 끼운 코드가 동작하는 곳입니다.

훅에는 지나가며 확인하는 것과 감싸는 것이 있습니다

훅(hook)은 코드를 끼워 넣을 수 있게 열어 둔 곳입니다. LangChain 문서는 훅을 두 종류로 나눕니다. before와 after 훅은 흐름이 그 지점을 지날 때 한 번 실행되고 상태를 읽거나 고칩니다. 로그를 남기거나 입력을 검사하거나 메시지를 바꿀 때 씁니다.

wrap 훅은 모델 호출이나 도구 호출 자체를 감쌉니다. 호출을 실제로 하는 handler를 받아서 부를지 말지를 직접 정합니다. 한 번 부를 수도 있고 실패하면 다시 부를 수도 있고 아예 부르지 않고 결과를 만들어 줄 수도 있습니다. 그림에서 감싸는 쪽이 호출을 어떻게 다룰지 고를 수 있습니다.

설명을 위한 그림입니다. 도구 호출을 감싸는 wrap_tool_call도 같은 방식으로 동작합니다.

guard.pypython
from langchain.agents.middleware import before_agent

# 1 시작할 때 한 번: 입력을 검사하고 막을 수 있다
@before_agent(can_jump_to=["end"])
def guard(state, runtime):
    text = state["messages"][-1].content
    if "해킹" in text:
        # 2 답을 대신 넣고 바로 끝으로 간다
        return {"messages": [{"role": "assistant",
                 "content": "그 요청은 도와드릴 수 없습니다."}],
                "jump_to": "end"}

# 3 에이전트에 끼운다
agent = create_agent(model, tools=tools, middleware=[guard])
터미널
$ python guard.py 그 요청은 도와드릴 수 없습니다.

출력은 예시 답입니다. 질문은 "옆 팀 와이파이 해킹하는 법 알려줘"를 가정했고 이 경우 모델은 한 번도 호출되지 않습니다. 코드는 langchain 1.4 기준입니다.

  1. 끼울 곳을 고릅니다. @before_agent는 실행을 시작할 때 한 번 도는 훅입니다. can_jump_to=["end"]를 적어 두어야 이 훅이 흐름을 끝으로 보낼 수 있습니다. 함수는 state(지금까지의 메시지)를 받아서 검사합니다.
  2. 막고 끝냅니다. 딕셔너리로 돌려주면 그 내용이 상태에 더해집니다. 여기서는 사용자에게 보여 줄 답을 메시지로 넣고 "jump_to": "end"로 모델을 부르지 않은 채 끝냅니다. 키워드 검사는 규칙 기반 검문이라 빠르고 싸지만 돌려 쓴 말은 놓칠 수 있습니다.
  3. 에이전트에 끼웁니다. middleware 목록에 넣으면 됩니다. 루프 코드는 한 줄도 바뀌지 않습니다.

내장 Middleware는 모델 호출 제한과 개인정보 가리기부터

자주 쓰는 검문은 LangChain이 미리 만들어 두었습니다. ModelCallLimitMiddleware는 모델을 몇 번까지 부를지 정하고 PIIMiddleware는 이메일이나 카드 번호 같은 개인정보(PII)를 찾아 처리합니다. 둘 다 실제로 실행해서 결과를 확인했습니다.

왼쪽은 모델이 도구만 계속 부르는 상황에서 호출 제한이 어떻게 멈추는지, 오른쪽은 입력의 이메일을 처리 방식에 따라 어떻게 바꾸는지 보여 줍니다.

ModelCallLimitMiddleware

PIIMiddleware("email", strategy=…)

두 칸 모두 langchain 1.4를 모의 서버에 붙여 실행한 결과입니다. 호출 칸은 도구만 계속 부르는 모델을 가정했습니다.

agent.pypython
# 4 개인정보를 가리고 모델 호출을 두 번으로 제한한다
agent = create_agent(model, tools=tools, middleware=[
    PIIMiddleware("email", strategy="redact"),
    ModelCallLimitMiddleware(run_limit=2, exit_behavior="end"),
])
result = agent.invoke(msg)
print(result["messages"][0].content)
print(result["messages"][-1].content)
터미널
$ python agent.py 내 이메일은 [REDACTED_EMAIL] 이고 티켓 상태 알려줘 Model call limits exceeded: run limit (2/2)

출력은 예시입니다. 모델이 도구만 계속 부르는 경우를 가정했습니다.

  1. 여러 개를 한꺼번에 끼웁니다. middleware 목록에 순서대로 넣습니다. 첫 줄은 이메일을 [REDACTED_EMAIL]로 바꿔 모델에게 원문이 가지 않게 하고 둘째 줄은 모델을 세 번째로 부르려는 순간 멈춥니다. exit_behavior="end"면 위 출력처럼 안내 메시지를 답으로 남기고 끝나며 "error"면 예외가 납니다.

    A, B 두 Middleware를 이 순서로 끼우고 직접 실행해 찍은 출력 순서입니다.

    여러 개를 끼우면 before는 앞에서 뒤로 after는 뒤에서 앞으로 돕니다

    Middleware를 여러 개 끼우면 양파 껍질처럼 첫 번째가 가장 바깥에 놓입니다. 모델로 들어갈 때는 바깥에서 안쪽으로(A, B) 지나고 나올 때는 안쪽에서 바깥으로(B, A) 지납니다. 로그를 찍어 보면 before_agent와 before_model은 A, B 순서이고 after_model과 after_agent는 B, A 순서입니다.

    순서가 중요한 예가 있습니다. 개인정보를 가리는 Middleware를 입력 검사보다 바깥에 두면 검사는 가려진 글을 봅니다. 반대로 두면 검사가 원문을 보게 됩니다. 어느 쪽이 맞는지는 무엇을 먼저 처리해야 하는지로 정합니다.

    Guardrail은 규칙으로 막거나 모델로 판단합니다

    Guardrail은 입력이나 출력을 검사해서 위험한 내용을 걸러 내는 장치입니다. LangChain 문서는 두 가지로 나눕니다. 규칙 기반은 키워드나 정규식 같은 정해진 규칙으로 검사해서 빠르고 싸지만 미묘한 위반은 놓칠 수 있습니다. 모델 기반은 다른 모델이 의미를 보고 판단해서 더 정확하지만 느리고 비쌉니다.

    자리도 나뉩니다. 입력 검사는 before_agent에서 모델을 부르기 전에 막고 출력 검사는 after_agent에서 최종 답을 확인해 고쳐 씁니다. 도구를 실행하는 Agent에서는 출력뿐 아니라 입력 쪽에서 막는 편이 비용도 위험도 줄입니다.

    두 방식의 특징은 LangChain Guardrails 문서의 설명을 따랐고 입력과 판정은 설명을 위한 예시입니다. 모델 기반이 항상 잡는다는 뜻은 아닙니다.

    다음 글에서는 Agent가 어제 나눈 대화를 단기·장기 기억을 알아보겠습니다.

    참고 자료

    출처 5개 펼치기