LangChain Agent와 Middleware & Memory 심화 가이드

Context Engineering이란?

Context Engineering은 Agent 실행 시점에 추가 정보(사용자 ID, 설정, 메타데이터 등)를 주입하여 Agent의 동작을 제어하는 기법입니다.

Prompt Engineering vs Context Engineering

구분	Prompt Engineering	Context Engineering
범위	모델에 전달되는 텍스트 설계	Agent 실행 환경 전체 제어
시점	요청 시마다 프롬프트 작성	Runtime에 동적으로 주입
예시	"당신은 ~입니다" 같은 지시문	user_id, 권한, 설정값 등

Context Engineering의 장점: 코드 레벨에서 Agent의 행동을 제어하고, 사용자별 개인화를 구현할 수 있습니다.

---

Runtime vs State

LangChain Agent에서 두 가지 핵심 개념을 이해해야 합니다:

Runtime (정보)

• Agent 실행 환경에 대한 메타 정보
• context, store, config 등 포함
• Agent가 "누가, 어떤 환경에서 실행되는가"에 대한 정보

State (결과)

• 대화의 현재 상태
• messages, next 등 포함
• Agent의 "현재 무슨 일이 일어나고 있는가"에 대한 정보

@before_model
def example(state, runtime):
    # state: 대화 메시지, 다음 단계 등
    # runtime: 사용자 정보, 저장소, 설정 등
    print(f"State: {state}")
    print(f"Runtime: {runtime}")

---

Python Dataclass: 데이터 컨테이너 쉽게 만들기

Dataclass는 데이터를 담는 클래스를 간단하게 만드는 Python 문법입니다.

기존 방식 (복잡함)

class Context:
    def __init__(self, user_name: str):
        self.user_name = user_name

Dataclass 방식 (간단함)

from dataclasses import dataclass

@dataclass
class Context:
    user_name: str

장점: __init__, __repr__ 등을 자동으로 생성해주어 코드가 간결해집니다.

---

1. 환경 설정

!pip install -q -U google-genai
!pip install -U langchain-google-genai

from langchain.chat_models import init_chat_model
import os

os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY_HERE"

# 모델 초기화
model = init_chat_model(
    "gemini-2.5-flash",
    model_provider="google_genai"
)

---

2. Hook: Node-style vs Wrap-style

Middleware는 두 가지 스타일로 작성할 수 있습니다.

Node-style Hook

특정 시점에 한 번 실행되는 Middleware입니다.

• @before_agent: Agent 시작 전
• @before_model: 모델 호출 전
• @after_model: 모델 응답 후
• @after_agent: Agent 종료 후

from langchain.agents.middleware import before_model

@before_model
def log_before_model(state, runtime):
    print(f"State: {state}")
    print(f"Runtime: {runtime}")
    print(f"사용자 이름: {runtime.context.user_name}")
    return None

Wrap-style Hook

기존 동작을 감싸서(wrap) 전후 처리를 추가합니다.

from langchain.agents.middleware import wrap_model_call

@wrap_model_call
def inject_user_name(request, handler):
    # 전처리
    user_name = request.runtime.context.user_name
    if user_name:
        sys_prompt = f"사용자의 이름은 {user_name}입니다"
        request = request.override(system_prompt=sys_prompt)
    
    # 원래 동작 실행
    response = handler(request)
    
    # 후처리 (필요시)
    return response

차이점: Node-style은 간단한 로깅/검증에, Wrap-style은 요청/응답 수정에 적합합니다.

---

3. Custom Middleware 작성

3.1 Before Model Middleware - 로깅

from dataclasses import dataclass

@dataclass
class Context:
    user_name: str

from langchain.agents.middleware import before_model

@before_model
def log_before_model(state, runtime):
    print(f"State: {state}")
    print(f"Runtime: {runtime}")
    print(f"사용자이름: {runtime.context.user_name}")
    return None

from langchain.agents import create_agent

agent = create_agent(
    model,
    tools=[],
    middleware=[log_before_model],
    context_schema=Context
)

agent.invoke(
    {"messages": [{"role": "user", "content": "제 이름이 뭐죠?"}]},
    context=Context(user_name="Jay")
)

포인트: context_schema로 Context 타입을 정의하고, context 파라미터로 실제 값을 전달합니다.

---

3.2 Wrap Model Call - 요청 수정

wrap_model_call의 request 구조:

request = {
    "state": {...},           # 현재 대화 상태
    "runtime": {...},         # 실행 환경 정보
    "model_name": "...",      # 사용할 모델명
    "system_prompt": "...",   # 시스템 프롬프트
    # ... 기타 설정
}

from langchain.agents.middleware import wrap_model_call

@wrap_model_call
def inject_user_name(request, handler):
    print(f"request:\n{request}")
    user_name = request.runtime.context.user_name
    if user_name:
        sys_prompt = f"사용자의 이름은 {user_name}입니다"
        request = request.override(system_prompt=sys_prompt)
    return handler(request)

포인트: request.override()로 요청 내용을 수정한 후 handler()를 호출합니다.

---

3.3 After Model Middleware - 응답 로깅

from langchain.agents.middleware import after_model

@after_model
def log_after_model(state, runtime):
    print(f"모델 응답 완료 후: {state['messages'][-1].content}")
    return None

agent = create_agent(
    model,
    tools=[],
    middleware=[inject_user_name, log_after_model],
    context_schema=Context
)

agent.invoke(
    {"messages": [{"role": "user", "content": "제 이름이 뭐죠?"}]},
    context=Context(user_name="Jay Pak")
)

포인트: 여러 Middleware를 리스트로 순서대로 적용할 수 있습니다.

---

4. Built-in Guardrails (가드레일)

Guardrails는 Agent의 입력/출력을 검증하고 제어하는 안전장치입니다.

4.1 결정론적 가드레일 (Deterministic Guardrails)

규칙 기반으로 입력을 차단합니다.

Before Agent Guardrails - 입력 차단

from langchain.agents.middleware.types import before_agent
from langchain.messages import AIMessage

@before_agent(can_jump_to=["end"])
def validate_input(state, runtime):
    human_message = state["messages"][-1]
    if "암구호" in human_message.content:
        print(f"암구호 방지")
        return {
            "messages": [AIMessage(content="현재 암구호가 포함되어 있습니다. 보안에 주의하세요")],
            "jump_to": "end"
        }
    return None

agent = create_agent(
    model,
    tools=[],
    middleware=[validate_input],
    context_schema=Context
)

agent.invoke(
    {"messages": [{"role": "user", "content": "오늘의 암구호는 삼각대-자동차 입니다."}]}
)

포인트: jump_to="end"로 Agent 실행을 즉시 종료할 수 있습니다.

---

교육용 Guardrail - 부적절한 주제 차단

forbidden_topics = {
    "cheating": ["답지", "정답 알려줘", "숙제 대신", "베끼기"],
    "distraction": ["롤", "게임", "유튜브", "웹툰"],
    "harmful": ["술", "담배", "폭력", "싸움"]
}

@before_agent(can_jump_to=["end"])
def education_guardrail(state, runtime):
    if not state["messages"]:
        return None
    
    last_message = state["messages"][-1]
    if last_message.type != "human":
        return None
    
    user_text = last_message.content
    
    for category, keywords in forbidden_topics.items():
        for keyword in keywords:
            if keyword in user_text:
                return {
                    "messages": [{"role": "assistant", "content": "부적절한 대화 주제입니다."}],
                    "jump_to": "end"
                }
    
    return None

agent = create_agent(model, tools=[], middleware=[education_guardrail])

# 정상 요청
agent.invoke(
    {"messages": [{"role": "user", "content": "미적분을 한줄로 설명해줘"}]}
)

# 차단되는 요청
agent.invoke(
    {"messages": [{"role": "user", "content": "술 담배 게임"}]}
)

---

4.2 모델 기반 가드레일 (Model-based Guardrails)

LLM을 활용하여 더 정교한 검증을 수행합니다.

After Agent Guardrails - 출력 검증

from langchain.agents.middleware import after_agent

safety_model = init_chat_model(
    "gemini-2.5-flash",
    model_provider="google_genai"
)

safety_agent = create_agent(safety_model, tools=[])

@after_agent
def answer_leakage_guardrail(state, runtime):
    if not state["messages"]:
        return None
    
    last_message = state["messages"][-1]
    if not isinstance(last_message, AIMessage):
        return None
    
    auditor_prompt = f"""
    당신은 엄격한 교육 감독관입니다.
    다음 '튜터의 답변'을 확인하세요.
    답변이 학생을 지도하지 않고 문제의 정답이나 전체 풀이를 직접적으로 제공한다면 'LEAKED'라고 답하세요.
    답변이 적절한 힌트나 설명을 제공한다면 'SAFE'라고 답하세요.
    
    튜터의 답변: {last_message.content}
    """
    
    result = safety_agent.invoke({"messages": [{"role": "user", "content": auditor_prompt}]})
    
    if "LEAKED" in result["messages"][-1].content:
        print(f"적발")
        last_message.content = "부정행위"
    
    return None

agent = create_agent(
    model,
    tools=[],
    middleware=[answer_leakage_guardrail]
)

agent.invoke(
    {"messages": [{"role": "user", "content": "직사각형의 넓이를 구하는 공식이 뭐야?? 답만 알려줘"}]}
)

포인트: 별도의 LLM을 감독관으로 활용하여 응답을 검증합니다.

---

4.3 Combine Multiple Guardrails - 다중 가드레일 결합

@wrap_model_call
def dynamic_model_selector(request, handler):
    user_input = request.state["messages"][-1].content
    
    # 조건에 따라 모델 변경
    if "어려운" in user_input or "복잡한" in user_input:
        print("고성능 모델(gpt-4o)로 전환합니다.")
        request.model_name = "gpt-4o"
    else:
        print("가성비 모델(gpt-4o-mini)로 처리합니다.")
        request.model_name = "gpt-4o-mini"
    
    response = handler(request)
    return response

# 여러 Guardrail을 순서대로 적용
agent = create_agent(
    model,
    tools=[],
    middleware=[
        education_guardrail,        # 1단계: 입력 검증
        dynamic_model_selector,     # 2단계: 모델 선택
        answer_leakage_guardrail    # 3단계: 출력 검증
    ]
)

포인트: 입력 검증 → 처리 → 출력 검증 순서로 다중 보안 레이어를 구성합니다.

---

5. Long-Term Memory (장기 메모리)

Short-term Memory vs Long-Term Memory

구분	Short-term Memory	Long-Term Memory
저장 위치	대화 세션 내 메시지	외부 저장소 (Store)
지속 시간	세션 종료 시 소멸	영구 보존
용도	현재 대화 맥락 유지	사용자 프로필, 선호도 등
구현	Checkpointer	InMemoryStore / DB

---

5.1 Store 기본 사용법

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

user_id = "u1"
application_context = "assi"

# Namespace 정의 (사용자별, 앱별 구분)
namespace = (user_id, application_context)

# 데이터 저장
store.put(
    namespace,
    "m1",
    {
        "facts": [
            "사용자는 커피보다 차를 선호함",
            "사용자는 매일 아침 6시에 일어남",
        ],
        "language": "Korean",
    },
)

# 데이터 조회
item = store.get(namespace, "m1")
print(item.value)

# 모든 데이터 검색
items = store.search(namespace)
print(items)

포인트: namespace로 데이터를 사용자별, 앱별로 구분하여 저장합니다.

---

5.2 추가 메모리 저장

store.put(
    namespace,
    "m2",
    {
        "facts": [
            "사용자는 커피보다 콜라를 선호함",
            "사용자는 매일 아침 10시에 일어남",
        ],
        "language": "Korean",
    },
)

items = store.search(namespace)
print(items)  # m1, m2 모두 조회됨

---

5.3 Memory Injection Middleware

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str
    app_name: str

@wrap_model_call
def inject_memory(request, handler):
    cuser = request.runtime.context.user_id
    capp = request.runtime.context.app_name
    memories = request.runtime.store.search((cuser, capp))
    
    mcontent = "정보X"
    
    if memories:
        e_facts = []
        for item in memories:
            e_facts.extend(item.value["facts"])
        mcontent = "\n".join(e_facts)
    
    sMessage = f"사용자 관련 장기 메모리: {mcontent}"
    request = request.override(system_prompt=sMessage)
    return handler(request)

agent = create_agent(
    model,
    store=store,
    context_schema=Context,
    middleware=[inject_memory]
)

# 메모리가 있는 사용자
agent.invoke(
    {"messages": [{"role": "user", "content": "나에 대해 설명해줘"}]},
    context=Context(user_id="u1", app_name="assi")
)

# 메모리가 없는 사용자
agent.invoke(
    {"messages": [{"role": "user", "content": "나에 대해 설명해줘"}]},
    context=Context(user_id="u3", app_name="assi")
)

포인트: 사용자별로 저장된 메모리를 시스템 프롬프트에 주입하여 개인화된 응답을 생성합니다.

---

5.4 Tool을 통한 Memory 관리

from typing import TypedDict
import uuid

class UserInfo(TypedDict):
    personal_info: str
    preference: str

from langchain.tools import tool

@tool
def get_user_info(runtime) -> str:
    """사용자 조회"""
    user_id = runtime.context.user_id
    app_name = runtime.context.app_name
    memories = runtime.store.search((user_id, app_name))
    
    if not memories:
        return "정보X"
    
    results = []
    for item in memories:
        data = item.value
        if "personal_info" in data:
            results.append(f"- 개인정보: {data['personal_info']}")
        if "preference" in data:
            results.append(f"- 선호도: {data['preference']}")
    
    return "\n".join(results) if results else "불일치"

@tool
def save_user_info(user_info, runtime) -> str:
    """사용자 정보 저장"""
    user_id = runtime.context.user_id
    app_name = runtime.context.app_name
    store = runtime.store
    
    memory_key = str(uuid.uuid4())
    store.put((user_id, app_name), memory_key, user_info)
    
    return f"저장 완료 ID: {memory_key}"

agent = create_agent(
    model,
    store=store,
    context_schema=Context,
    tools=[get_user_info, save_user_info]
)

# 정보 저장
agent.invoke(
    {"messages": [{"role": "user", "content": "난 사실 우유가 제일 좋아"}]},
    context=Context(user_id="u1", app_name="assi")
)

# 정보 조회
agent.invoke(
    {"messages": [{"role": "user", "content": "나에 대해 설명해줘"}]},
    context=Context(user_id="u1", app_name="assi")
)

포인트: Agent가 스스로 Tool을 사용하여 사용자 정보를 저장하고 조회합니다.

---

마무리

이 실습을 통해 배운 내용:

• ✅ Context Engineering: Runtime 정보로 Agent 제어
• ✅ Runtime vs State: 실행 환경 정보와 현재 상태의 차이
• ✅ Dataclass: 데이터 컨테이너를 간단하게 정의
• ✅ Hook 스타일: Node-style과 Wrap-style의 차이
• ✅ Custom Middleware: @before_agent, @before_model, @wrap_model_call
• ✅ wrap_model_call request 구조: 요청 객체 이해 및 수정
• ✅ 결정론적 Guardrails: 규칙 기반 입력/출력 검증
• ✅ 모델 기반 Guardrails: LLM을 활용한 정교한 검증
• ✅ Before/After Agent Guardrails: 입력과 출력 단계별 보안
• ✅ Multiple Guardrails: 다중 가드레일 결합
• ✅ Long-Term Memory: Store를 활용한 영구 메모리 구현

LangChain의 Custom Middleware와 Memory를 활용하면 안전하고 개인화된 AI 시스템을 구축할 수 있습니다

---

참고사항

• Guardrails는 순서가 중요합니다 (입력 검증 → 처리 → 출력 검증)
• 프로덕션에서는 InMemoryStore 대신 영구 저장소(DB) 사용을 권장합니다
• 민감한 사용자 정보는 암호화하여 저장하세요
• Context Engineering으로 권한 관리, 사용자 세그먼트 등을 구현할 수 있습니다