단기 메모리

개요

AI 에이전트에게 메모리는 이전 대화 내용을 기억하고, 피드백으로부터 학습하며, 사용자 선호도에 적응할 수 있게 해주기 때문에 매우 중요합니다. 에이전트가 수많은 사용자 상호작용을 포함하는 더 복잡한 작업을 처리할수록, 이 기능은 효율성과 사용자 만족도 모두에 필수적이 됩니다.

단기 메모리는 애플리케이션이 단일 스레드 또는 대화 내에서 이전 상호작용을 기억할 수 있게 해줍니다. 스레드는 이메일이 하나의 대화에서 메시지를 그룹화하는 것과 유사하게, 세션 내의 여러 상호작용을 조직화합니다.

대화 기록은 단기 메모리의 가장 일반적인 형태입니다. 긴 대화는 오늘날의 LLM에 도전 과제를 제기합니다. 전체 대화 기록이 해당 모델의 컨텍스트 윈도우 사이즈 내에 들어가지 않을 수 있으며, 이는 컨텍스트 손실이나 오류를 초래합니다.

모델이 전체 컨텍스트 길이를 지원하더라도, 대부분의 모델은 여전히 긴 컨텍스트에서 성능이 떨어집니다. 오래되거나 주제에서 벗어난 콘텐츠에 “산만해지며”, 동시에 느린 응답 시간과 높은 비용을 겪게 됩니다.

채팅 모델은 메시지를 사용하여 컨텍스트를 받아들이며, 여기에는 지시사항(SystemMessage)과 입력(HumanMessage)이 포함됩니다. 채팅 애플리케이션에서 메시지는 사람 입력과 모델 응답 사이를 번갈아가며, 시간이 지남에 따라 점점 더 길어지는 메시지 목록을 생성합니다. 컨텍스트 윈도우가 제한적이기 때문에, 많은 애플리케이션이 오래된 정보를 제거하거나 “잊는” 기법을 사용함으로써 이점을 얻을 수 있습니다.

LangChain의 에이전트는 단기 메모리를 에이전트 상태의 일부로 관리합니다. 이를 그래프의 상태에 저장함으로써, 에이전트는 서로 다른 스레드 간의 분리를 유지하면서 주어진 대화에 대한 전체 컨텍스트에 액세스할 수 있습니다. 상태는 checkpointer를 사용하여 데이터베이스(또는 메모리)에 지속되므로, 스레드를 언제든지 재개할 수 있습니다. 단기 메모리는 에이전트가 호출되거나 단계(도구 호출 등)가 완료될 때 업데이트되며, 각 단계의 시작 시 상태를 읽습니다.

사용법

에이전트에 단기 메모리(스레드 수준 지속성)를 추가하려면, 에이전트를 생성할 때 checkpointer를 지정해야 합니다.

from langchain_core.tools import tool
from libs.helpers import pretty_print
 
from langchain.agents import create_agent
from langchain.tools import ToolRuntime
from langgraph.checkpoint.memory import MemorySaver
 
 
@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """사용자 정보를 조회하는 툴입니다."""
 
    user_id = runtime.state["user_id"]
    print(f"사용자 ID '{user_id}'로 정보를 찾고 있습니다...")
 
    store = runtime.store
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "알 수 없는 사용자"
 
 
agent = create_agent(
    "openai:gpt-4.1-mini",
    [get_user_info],
    checkpointer=MemorySaver(),
)
 
# 5. 에이전트 실행
response = agent.invoke(
    {"messages": [{"role": "user", "content": "안녕하세요! 제 이름은 김밥입니다."}]},
    {"configurable": {"thread_id": "1"}},
)
pretty_print(response)

================================[1m Human Message [0m=================================

안녕하세요! 제 이름은 김밥입니다.
==================================[1m Ai Message [0m==================================

안녕하세요, 김밥님! 만나서 반갑습니다. 무엇을 도와드릴까요?

response = agent.invoke(
    {"messages": [{"role": "user", "content": "제 이름을 기억하나요?"}]},
    {"configurable": {"thread_id": "1"}},
)
pretty_print(response)

================================[1m Human Message [0m=================================

안녕하세요! 제 이름은 김밥입니다.
==================================[1m Ai Message [0m==================================

안녕하세요, 김밥님! 만나서 반갑습니다. 무엇을 도와드릴까요?
================================[1m Human Message [0m=================================

제 이름을 기억하나요?
==================================[1m Ai Message [0m==================================

네, 김밥님이라고 하셨죠! 어떻게 도와드릴까요?

프로덕션 환경

프로덕션 환경에서는 데이터베이스로 백업되는 checkpointer를 사용하세요:

#!pip install langgraph-checkpoint-postgres

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
 
 
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # auto create tables in PostgresSql
    agent = create_agent(
        "openai:gpt-5",
        [get_user_info],
        checkpointer=checkpointer,
    )

---------------------------------------------------------------------------

ModuleNotFoundError                       Traceback (most recent call last)

Cell In[8], line 2
      1 from langchain.agents import create_agent
----> 2 from langgraph.checkpoint.postgres import PostgresSaver
      5 DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
      6 with PostgresSaver.from_conn_string(DB_URI) as checkpointer:


ModuleNotFoundError: No module named 'langgraph.checkpoint.postgres'

에이전트 메모리 커스터마이징

기본적으로 에이전트는 AgentState를 사용하여 단기 메모리를 관리하며, 특히 messages 키를 통해 대화 기록을 관리합니다.

AgentState를 확장하여 추가 필드를 추가할 수 있습니다. 커스텀 상태 스키마는 state_schema 파라미터를 사용하여 create_agent에 전달됩니다.

from langchain.agents import AgentState, create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
 
 
class CustomAgentState(AgentState):
    user_id: str
    preferences: dict
 
 
agent = create_agent(
    "openai:gpt-4.1-mini",
    [get_user_info],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver(),
    store=InMemoryStore(),
)
 
# 사용자 정의 상태는 호출 시 전달될 수 있습니다
inputs: dict = {
    "messages": [{"role": "user", "content": "안녕하세요. 저를 알고있나요?"}],
    "user_id": "user_123",
    "preferences": {"theme": "dark"},
}
 
result = agent.invoke(
    inputs,
    {"configurable": {"thread_id": "1"}},
)
pretty_print(result)

사용자 ID 'user_123'로 정보를 찾고 있습니다...
================================[1m Human Message [0m=================================

안녕하세요. 저를 알고있나요?
==================================[1m Ai Message [0m==================================
Tool Calls:
  get_user_info (call_Pzdghd3Jnlxdk4bbpxfNwi7Q)
 Call ID: call_Pzdghd3Jnlxdk4bbpxfNwi7Q
  Args:
=================================[1m Tool Message [0m=================================
Name: get_user_info

알 수 없는 사용자
==================================[1m Ai Message [0m==================================

안녕하세요! 현재 사용자 정보를 확인할 수가 없습니다. 대신 새롭게 도와드릴 수 있는 것이 있으면 말씀해 주세요.

일반적인 패턴

단기 메모리가 활성화되면, 긴 대화가 LLM의 컨텍스트 윈도우를 초과할 수 있습니다. 일반적인 해결책은 다음과 같습니다.

메시지 트리밍: 처음 또는 마지막 N개의 메시지 제거 (LLM 호출 전)
메시지 삭제: LangGraph 상태에서 메시지를 영구적으로 삭제
메시지 요약: 기록의 이전 메시지를 요약하여 요약본으로 대체
커스텀 전략: 커스텀 전략 (예: 메시지 필터링 등)

이를 통해 에이전트는 LLM의 context window를 초과하지 않으면서 대화를 추적할 수 있습니다.

메시지 트리밍

대부분의 LLM은 최대 지원 context window(토큰 단위)를 가지고 있습니다.

메시지를 자를 시기를 결정하는 한 가지 방법은 메시지 기록의 토큰을 세고 해당 제한에 접근할 때마다 자르는 것입니다. LangChain을 사용하는 경우, trim messages 유틸리티를 사용하여 목록에서 유지할 토큰 수와 경계를 처리하는 데 사용할 strategy(예: 마지막 max_tokens 유지)를 지정할 수 있습니다.

에이전트에서 메시지 기록을 자르려면, @before_model 미들웨어 데코레이터를 사용하세요:

from typing import Any
 
from langchain_core.runnables import RunnableConfig
 
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
 
 
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """최근 몇 개의 메시지만 유지합니다."""
 
    messages = state["messages"]
 
    if len(messages) <= 3:
        return None  # No changes needed
 
    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
 
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]}
 
 
agent = create_agent(
    "openai:gpt-5-nano",
    tools=[],
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}
 
agent.invoke({"messages": "안녕, 내 이름은 김밥이야"}, config)
agent.invoke({"messages": "고양이에 관한 짧은 시를 써 봐"}, config)
agent.invoke({"messages": "이제 개에 대해서도 똑같이 해 봐"}, config)
final_response = agent.invoke({"messages": "내 이름이 뭐지?"}, config)
final_response["messages"][-1].pretty_print()

==================================[1m Ai Message [0m==================================

네, 당신의 이름은 김밥이야. 원하면 애칭으로 불러줄 수도 있어. 다른 이름으로 바꾸고 싶으면 말해줘.

메시지 삭제

메시지 기록을 관리하기 위해 그래프 상태에서 메시지를 삭제할 수 있습니다.

이는 특정 메시지를 제거하거나 전체 메시지 기록을 지우려는 경우에 유용합니다.

그래프 상태에서 메시지를 삭제하려면, RemoveMessage를 사용할 수 있습니다.

RemoveMessage가 작동하려면, add_messages 리듀서가 있는 상태 키를 사용해야 합니다.

기본 AgentState가 이를 제공합니다.

특정 메시지를 제거하려면:

from langchain.messages import RemoveMessage
 
 
def delete_messages(state):
    """오래된 메시지를 삭제합니다."""
 
    # 상태에서 현재 메시지 리스트 가져오기
    messages = state["messages"]
 
    # 메시지가 2개를 초과하는 경우에 삭제 수행
    if len(messages) > 2:
        # 가장 오래된 2개의 메시지를 삭제
        # messages[:2]는 리스트의 처음 2개 요소를 가져옴
        # 각 메시지의 id를 사용하여 RemoveMessage 객체 생성
        return {
            "messages": [RemoveMessage(id=m.id) for m in messages[:2]],
        }

모든 메시지를 제거하려면:

from langgraph.graph.message import REMOVE_ALL_MESSAGES
 
 
def delete_messages(state):
    """모든 메시지를 삭제합니다."""
 
    # REMOVE_ALL_MESSAGES 특수 ID를 사용하여 모든 메시지 삭제
    # 개별 메시지 ID를 순회할 필요 없이 한 번에 전체 삭제 가능
    return {
        "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)],
    }

메시지를 삭제할 때는, 결과 메시지 기록이 유효한지 확인하세요. 사용 중인 LLM 제공업체의 제한 사항을 확인하세요.

예를 들어:

일부 제공업체는 메시지 기록이 user 메시지로 시작하기를 기대합니다
대부분의 제공업체는 도구 호출이 있는 assistant 메시지 뒤에 해당하는 tool 결과 메시지가 와야 합니다.

from langchain_core.runnables import RunnableConfig
 
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import after_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
 
 
@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """대화를 관리하기 쉽게 유지하려면 오래된 메시지를 삭제하세요."""
 
    # 현재 상태에서 메시지 리스트 가져오기
    messages = state["messages"]
 
    # 메시지가 2개를 초과하는 경우
    if len(messages) > 2:
        # 가장 오래된 2개의 메시지를 RemoveMessage로 표시하여 삭제
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
 
    # 메시지가 2개 이하면 삭제하지 않음
    return None
 
 
# 에이전트 생성
agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    system_prompt="간결하고 요점을 분명히 하십시오.",
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver(),
)

# 설정: thread_id로 대화 세션 구분
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
 
# 첫 번째 대화: 사용자가 자신의 이름을 소개
print("=== 첫 번째 대화 ===")
for event in agent.stream(
    {"messages": [{"role": "user", "content": "안녕하세요! 저는 김밥입니다."}]},
    config,
    stream_mode="values",  # 전체 상태 값을 스트리밍
):
    print([(message.type, message.content) for message in event["messages"]])
 
# 두 번째 대화: 사용자가 자신의 이름을 물어봄
print("\n=== 두 번째 대화 ===")
for event in agent.stream(
    {"messages": [{"role": "user", "content": "내 이름이 뭐지?"}]},
    config,  # 동일한 thread_id 사용 - 이전 대화 기억
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])
 
# 세 번째 대화: 사용자가 자신의 이름을 다시 물어봄
print("\n=== 세 번째 대화 ===")
for event in agent.stream(
    {"messages": [{"role": "user", "content": "내 이름이 뭐지?"}]},
    config,  # 동일한 thread_id 사용 - 이전 대화 기억
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

=== 첫 번째 대화 ===
[('human', '안녕하세요! 저는 김밥입니다.')]
[('human', '안녕하세요! 저는 김밥입니다.'), ('ai', '안녕하세요, 김밥님! 어떻게 도와드릴까요?')]

=== 두 번째 대화 ===
[('human', '안녕하세요! 저는 김밥입니다.'), ('ai', '안녕하세요, 김밥님! 어떻게 도와드릴까요?'), ('human', '내 이름이 뭐지?')]
[('human', '안녕하세요! 저는 김밥입니다.'), ('ai', '안녕하세요, 김밥님! 어떻게 도와드릴까요?'), ('human', '내 이름이 뭐지?'), ('ai', '당신의 이름은 김밥입니다.')]
[('human', '내 이름이 뭐지?'), ('ai', '당신의 이름은 김밥입니다.')]

=== 세 번째 대화 ===
[('human', '내 이름이 뭐지?'), ('ai', '당신의 이름은 김밥입니다.'), ('human', '내 이름이 뭐지?')]
[('human', '내 이름이 뭐지?'), ('ai', '당신의 이름은 김밥입니다.'), ('human', '내 이름이 뭐지?'), ('ai', '죄송하지만 회원님의 이름을 알 수 없습니다.')]
[('human', '내 이름이 뭐지?'), ('ai', '죄송하지만 회원님의 이름을 알 수 없습니다.')]

메시지 요약

위에서 보여준 것처럼 메시지를 트리밍하거나 제거하는 문제점은 메시지 큐를 제거함으로써 정보를 잃을 수 있다는 것입니다.

이 때문에, 일부 애플리케이션은 채팅 모델을 사용하여 메시지 기록을 요약하는 보다 정교한 접근 방식의 이점을 얻습니다.

에이전트에서 메시지 기록을 요약하려면, 내장된 SummarizationMiddleware를 사용하세요.

from langchain_core.runnables import RunnableConfig
 
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
 
 
checkpointer = InMemorySaver()
 
agent = create_agent(
    model="openai:gpt-4.1",  # 메인 대화에 사용할 모델
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4.1-mini",  # 요약에 사용할 모델 (더 저렴한 모델)
            max_tokens_before_summary=100,  # 100 토큰 초과 시 요약 트리거
            messages_to_keep=2,  # 요약 후 최근 2개 메시지 유지
        )
    ],
    checkpointer=checkpointer,
)

from libs.helpers import pretty_print
 
 
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
 
agent.invoke({"messages": "안녕하세요, 제 이름은 밥입니다"}, config)
agent.invoke({"messages": "고양이에 관한 짧은 시를 써주세요"}, config)
agent.invoke({"messages": "이제 개에 관해서도 똑같이 해주세요"}, config)
final_response = agent.invoke({"messages": "제 이름이 뭐죠?"}, config)
 
pretty_print(final_response)

================================[1m Human Message [0m=================================

Here is a summary of the conversation to date:

안녕하세요, 제 이름은 밥입니다.

고양이에 관한 짧은 시:
창가에 앉은 고양이  
햇살을 베개 삼아  
느릿하게 눈을 감아요.  
작은 숨결 사이로  
꿈결을 걷는 나른한 오후.

이제 개에 관해서도 같은 방식으로 시를 써주세요.
==================================[1m Ai Message [0m==================================

물론입니다!  
개에 관한 짧은 시도 써드릴게요.

따사로운 마당 끝에서  
꼬리를 흔드는 너의 눈빛  
맑은 하늘을 닮았구나.  
달려오는 발소리에  
웃음이 피어나는 하루.
================================[1m Human Message [0m=================================

제 이름이 뭐죠?
==================================[1m Ai Message [0m==================================

당신의 이름은 밥입니다.

더 많은 구성 옵션은 SummarizationMiddleware를 참조하세요.

메모리 액세스

에이전트의 단기 메모리(상태)에 여러 방법으로 액세스하고 수정할 수 있습니다:

도구

도구에서 단기 메모리 읽기

ToolRuntime 파라미터를 사용하여 도구에서 단기 메모리(상태)에 액세스합니다.

tool_runtime 파라미터는 도구 시그니처에서 숨겨져 있지만(모델이 보지 못함), 도구는 이를 통해 상태에 액세스할 수 있습니다.

from langchain.agents import AgentState, create_agent
from langchain.tools import ToolRuntime, tool
 
 
class CustomState(AgentState):
    user_id: str
 
 
@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """사용자 정보 조회."""
 
    user_id = runtime.state["user_id"]
    return (
        "사용자는 John Smith입니다." if user_id == "user_123" else "알 수 없는 사용자"
    )
 
 
agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)
 
result = agent.invoke(
    {
        "messages": "사용자 정보 조회",
        "user_id": "user_123",
    }
)
print(result["messages"][-1].content)

조회된 사용자 정보: 이름은 John Smith 입니다.

다른 정보 확인이나 변경이 필요하면 말씀해 주세요.

도구에서 단기 메모리 쓰기

실행 중에 에이전트의 단기 메모리(상태)를 수정하려면, 도구에서 직접 상태 업데이트를 반환할 수 있습니다.

이는 중간 결과를 지속하거나 정보를 후속 도구나 프롬프트에서 액세스할 수 있게 만드는 데 유용합니다.

from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
 
from langchain.agents import AgentState, create_agent
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
 
 
class CustomState(AgentState):
    user_name: str
 
 
class CustomContext(BaseModel):
    user_id: str
 
 
@tool
def update_user_info(
    runtime: ToolRuntime[CustomContext, CustomState],
) -> Command:
    """사용자 정보를 조회하세요."""
 
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "알 수 없는 사용자"
 
    return Command(
        update={
            "user_name": name,
            # 메시지 기록을 업데이트하세요
            "messages": [
                ToolMessage(
                    "사용자 정보를 성공적으로 조회했습니다.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )
 
 
@tool
def greet(runtime: ToolRuntime[CustomContext, CustomState]) -> str:
    """사용자 정보를 먼저 조회한 다음에 이 메시지로 사용자에게 인사하세요."""
 
    user_name = runtime.state["user_name"] if "user_name" in runtime.state else ""
    return f"안녕하세요. {user_name}!"
 
 
agent = create_agent(
    model="openai:gpt-4.1",
    tools=[update_user_info, greet],
    state_schema=CustomState,
    context_schema=CustomContext,
)
 
result = agent.invoke(
    {"messages": [{"role": "user", "content": "사용자에게 인사하세요."}]},
    context=CustomContext(user_id="user_123"),
)
pretty_print(result)

================================[1m Human Message [0m=================================

사용자에게 인사하세요.
==================================[1m Ai Message [0m==================================
Tool Calls:
  update_user_info (call_lzVzsiKEzVxEqBBzPXGRatC4)
 Call ID: call_lzVzsiKEzVxEqBBzPXGRatC4
  Args:
=================================[1m Tool Message [0m=================================
Name: update_user_info

사용자 정보를 성공적으로 조회했습니다.
==================================[1m Ai Message [0m==================================
Tool Calls:
  greet (call_kJpHmEr3VTCQ5p9sQgzCtAUG)
 Call ID: call_kJpHmEr3VTCQ5p9sQgzCtAUG
  Args:
=================================[1m Tool Message [0m=================================
Name: greet

안녕하세요. John Smith!
==================================[1m Ai Message [0m==================================

안녕하세요. John Smith! 

무엇을 도와드릴까요?

프롬프트

미들웨어에서 단기 메모리(상태)에 액세스하여 대화 기록이나 커스텀 상태 필드를 기반으로 동적 프롬프트를 생성합니다.

from typing import TypedDict
 
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import ModelRequest, dynamic_prompt
 
 
class CustomContext(TypedDict):
    user_name: str
 
 
def get_weather(city: str):
    """지정한 도시의 날씨를 가져옵니다."""
 
    return f"{city}의 날씨는 항상 맑습니다!"
 
 
@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest):
    user_name = request.runtime.context.get("user_name")
 
    return f"당신은 유용한 어시스턴트입니다. 사용자를 {user_name} 님이라고 부르세요."
 
 
agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
    middleware=[dynamic_system_prompt],
    context_schema=CustomContext,
)
 
result = agent.invoke(
    {"messages": [{"role": "user", "content": "샌프란시스코 날씨는 어때요?"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

================================[1m Human Message [0m=================================

샌프란시스코 날씨는 어때요?
==================================[1m Ai Message [0m==================================
Tool Calls:
  get_weather (call_SaSyYzYyfEQmwyl6RW4zlYcq)
 Call ID: call_SaSyYzYyfEQmwyl6RW4zlYcq
  Args:
    city: San Francisco
=================================[1m Tool Message [0m=================================
Name: get_weather

San Francisco의 날씨는 항상 맑습니다!
==================================[1m Ai Message [0m==================================

John Smith 님, 샌프란시스코의 날씨는 항상 맑습니다!  
더 자세한 정보가 필요하시면 현재 온도나 바람 세기, 오늘의 예보 같은 구체적인 정보도 바로 알려드리겠습니다. 어떤 정보를 보고 싶으신가요?

Before model

@before_model 미들웨어에서 단기 메모리(상태)에 액세스하여 모델 호출 전에 메시지를 처리합니다.


graph TD
    S(["\_\_start\_\_"])
    PRE(before_model)
    MODEL(model)
    TOOLS(tools)
    END(["\_\_end\_\_"])
    S --> PRE
    PRE --> MODEL
    MODEL -.-> TOOLS
    MODEL -.-> END
    TOOLS --> PRE
    classDef blueHighlight fill:#0a1c25,stroke:#0a455f,color:#bae6fd;
    class S blueHighlight;
    class END blueHighlight;

from typing import Any
 
from langchain_core.runnables import RunnableConfig
 
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
 
 
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """문맥 창에 맞도록 최근 몇 개의 메시지만 유지하십시오."""
 
    messages = state["messages"]
 
    if len(messages) <= 3:
        return None  # 변경할 사항 없음
 
    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg, *recent_messages]
 
    return {
        "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages],
    }
 
 
agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    middleware=[trim_messages],
)
 
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
 
agent.invoke({"messages": "안녕, 내 이름은 밥이야"}, config)
agent.invoke({"messages": "고양이에 관한 짧은 시를 써 봐"}, config)
agent.invoke({"messages": "이제 개에 대해서도 똑같이 해 봐"}, config)
final_response = agent.invoke({"messages": "내 이름이 뭐지?"}, config)
 
final_response["messages"][-1].pretty_print()

==================================[1m Ai Message [0m==================================

죄송하지만 사용자님의 이름을 알 수 없어요. 알려주시면 기억하는 데 도움이 될 수 있어요!

After model

@after_model 미들웨어에서 단기 메모리(상태)에 액세스하여 모델 호출 후에 메시지를 처리합니다.


graph TD
    S(["\_\_start\_\_"])
    MODEL(model)
    POST(after_model)
    TOOLS(tools)
    END(["\_\_end\_\_"])
    S --> MODEL
    MODEL --> POST
    POST -.-> END
    POST -.-> TOOLS
    TOOLS --> MODEL
    classDef blueHighlight fill:#0a1c25,stroke:#0a455f,color:#bae6fd;
    class S blueHighlight;
    class END blueHighlight;
    class POST greenHighlight;

from langchain_core.messages import BaseMessage
 
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import after_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
 
 
STOP_WORDS = ["password", "secret", "비밀"]
 
 
@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """민감한 단어가 포함된 메시지를 삭제하십시오."""
 
    last_message: BaseMessage = state["messages"][-1]
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}
    return None
 
 
agent = create_agent(
    model="openai:gpt-4.1-nano",
    tools=[],
    middleware=[validate_response],
    checkpointer=InMemorySaver(),
)

from libs.helpers import stream_print
 
response = agent.stream(
    {"messages": "너의 비밀을 알려줘"},
    config={
        "configurable": {"thread_id": "1"},
    },
)
stream_print(response)

==================================[1m Ai Message [0m==================================

저는 인공지능 언어 모델로, 사람들의 질문에 답변하고 도움을 드리기 위해 만들어졌어요. 특별한 비밀이 있다기보다는, 여러분의 이야기를 듣고 지원하는 것이 제 역할입니다! 무엇이든 궁금한 점이 있거나 도움이 필요하시면 말씀해 주세요.
================================[1m Remove Message [0m================================

Jeongsk

탐색기

05-short-term-memory.ipynb