langchain / / 2025. 11. 8. 21:12

[LangChain v1.0] Short-term memory

Overview

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

Short-term memory는 애플리케이션이 단일 스레드 또는 대화 내에서 이전 상호작용을 기억할 수 있게 해줍니다.

스레드는 세션 내 여러 상호작용을 구성하며, 이메일이 단일 대화로 메시지를 그룹화하는 방식과 유사합니다.

대화 기록은 short-term memory의 가장 일반적인 형태입니다. 긴 대화는 오늘날의 LLM에 도전 과제를 제시합니다. 전체 기록이 LLM의 컨텍스트 윈도우 안에 맞지 않을 수 있으며, 이는 컨텍스트 손실이나 오류를 초래합니다.

모델이 전체 컨텍스트 길이를 지원하더라도, 대부분의 LLM은 여전히 긴 컨텍스트에 대해 성능이 좋지 않습니다. 그들은 오래되거나 주제에서 벗어난 콘텐츠에 "산만해지며", 응답 시간이 느려지고 비용이 증가합니다.

Chat 모델은 messages를 사용하여 컨텍스트를 수용하며, 여기에는 지시사항(시스템 메시지)과 입력(사람 메시지)이 포함됩니다. 채팅 애플리케이션에서 메시지는 사람의 입력과 모델의 응답 사이를 번갈아가며, 시간이 지남에 따라 더 길어지는 메시지 목록이 생성됩니다. 컨텍스트 윈도우가 제한되어 있기 때문에, 많은 애플리케이션이 오래된 정보를 제거하거나 "잊는" 기술을 사용하는 것이 유익할 수 있습니다.

Usage

에이전트에 short-term memory(스레드 레벨 지속성)를 추가하려면, 에이전트를 생성할 때 checkpointer를 지정해야 합니다.

LangChain의 에이전트는 에이전트 상태의 일부로 short-term memory를 관리합니다.

이를 그래프의 상태에 저장함으로써, 에이전트는 서로 다른 스레드 간의 분리를 유지하면서 주어진 대화에 대한 전체 컨텍스트에 액세스할 수 있습니다.

상태는 checkpointer를 사용하여 데이터베이스(또는 메모리)에 지속되므로 스레드는 언제든지 재개될 수 있습니다.

Short-term memory는 에이전트가 호출되거나 단계(예: 도구 호출)가 완료될 때 업데이트되며, 상태는 각 단계의 시작 시 읽혀집니다.

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver


agent = create_agent(
    "gpt-5",
    [get_user_info],
    checkpointer=InMemorySaver(),
)

agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    {"configurable": {"thread_id": "1"}},
)

In production

프로덕션 환경에서는 데이터베이스로 백업되는 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(
        "gpt-5",
        [get_user_info],
        checkpointer=checkpointer,
    )

Customizing agent memory

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

AgentState를 확장하여 추가 필드를 추가할 수 있습니다. 사용자 정의 상태 스키마는 state_schema 매개변수를 사용하여 create_agent에 전달됩니다.

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver


class CustomAgentState(AgentState):
    user_id: str
    preferences: dict

agent = create_agent(
    "gpt-5",
    [get_user_info],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver(),
)

# Custom state can be passed in invoke
result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],
        "user_id": "user_123",
        "preferences": {"theme": "dark"}
    },
    {"configurable": {"thread_id": "1"}})

Common patterns

Short-term memory가 활성화되면, 긴 대화는 LLM의 컨텍스트 윈도우를 초과할 수 있습니다. 일반적인 솔루션은 다음과 같습니다:

Trim messages
처음 또는 마지막 N개의 메시지 제거(LLM 호출 전)

Delete messages
LangGraph 상태에서 메시지를 영구적으로 삭제

Summarize messages
기록의 이전 메시지를 요약하고 요약으로 교체

Custom strategies
사용자 정의 전략(예: 메시지 필터링 등)

이를 통해 에이전트는 LLM의 컨텍스트 윈도우를 초과하지 않으면서 대화를 추적할 수 있습니다.

Trim messages

대부분의 LLM은 최대 지원 컨텍스트 윈도우(토큰 단위로 표시됨)를 가지고 있습니다.

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

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

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Keep only the last few messages to fit context window."""
    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(
    model,
    tools=tools,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

"""
================================== Ai Message ==================================

Your name is Bob. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.
"""

Delete messages

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

이는 특정 메시지를 제거하거나 전체 메시지 기록을 지우고자 할 때 유용합니다.

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

RemoveMessage가 작동하려면 add_messages reducer와 함께 상태 키를 사용해야 합니다.

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

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

from langchain.messages import RemoveMessage

def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # remove the earliest two messages
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}

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

from langgraph.graph.message import REMOVE_ALL_MESSAGES

def delete_messages(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

메시지를 삭제할 때, 결과 메시지 기록이 유효한지 확인하세요. 사용 중인 LLM 공급자의 제한 사항을 확인하세요. 예를 들어:

  • 일부 공급자는 메시지 기록이 user 메시지로 시작할 것을 기대합니다
  • 대부분의 공급자는 도구 호출이 있는 assistant 메시지 뒤에 해당 tool 결과 메시지가 와야 합니다.

from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig


@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """Remove old messages to keep conversation manageable."""
    messages = state["messages"]
    if len(messages) > 2:
        # remove the earliest two messages
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None


agent = create_agent(
    "gpt-5-nano",
    tools=[],
    system_prompt="Please be concise and to the point.",
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])
[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.')]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "what's my name?")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "what's my name?"), ('ai', 'Your name is Bob. How can I help you today, Bob?')]
[('human', "what's my name?"), ('ai', 'Your name is Bob. How can I help you today, Bob?')]

Summarize messages

위에 표시된 것처럼 메시지를 트리밍하거나 제거하는 문제는 메시지 큐의 선별로 인해 정보를 잃을 수 있다는 것입니다. 이 때문에, 일부 애플리케이션은 채팅 모델을 사용하여 메시지 기록을 요약하는 보다 정교한 접근 방식의 이점을 얻습니다.

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

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig


checkpointer = InMemorySaver()

agent = create_agent(
    model="gpt-4o",
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            max_tokens_before_summary=4000,  # Trigger summarization at 4000 tokens
            messages_to_keep=20,  # Keep last 20 messages after summary
        )
    ],
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

"""
================================== Ai Message ==================================

Your name is Bob!
"""

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

Access memory

에이전트의 short-term memory(상태)에 여러 가지 방법으로 액세스하고 수정할 수 있습니다:

Tools

Read short-term memory in a tool

ToolRuntime 매개변수를 사용하여 도구에서 short-term memory(상태)에 액세스합니다.

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

from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime


class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    runtime: ToolRuntime
) -> str:
    """Look up user info."""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)

result = agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# > User is John Smith.

Write short-term memory from tools

실행 중에 에이전트의 short-term memory(상태)를 수정하려면, 도구에서 직접 상태 업데이트를 반환할 수 있습니다.

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

from langchain.tools import tool, ToolRuntime
from langchain_core.runnables import RunnableConfig
from langchain.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel


class CustomState(AgentState):
    user_name: str

class CustomContext(BaseModel):
    user_id: str

@tool
def update_user_info(
    runtime: ToolRuntime[CustomContext, CustomState],
) -> Command:
    """Look up and update user info."""
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # update the message history
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })

@tool
def greet(
    runtime: ToolRuntime[CustomContext, CustomState]
) -> str:
    """Use this to greet the user once you found their info."""
    user_name = runtime.state["user_name"]
    return f"Hello {user_name}!"
agent = create_agent(
    model="gpt-5-nano",
    tools=[update_user_info, greet],
    state_schema=CustomState,
    context_schema=CustomContext,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    context=CustomContext(user_id="user_123"),
)

Prompt

대화 기록 또는 사용자 정의 상태 필드를 기반으로 동적 프롬프트를 생성하기 위해 미들웨어에서 short-term memory(상태)에 액세스합니다.

from langchain.agents import create_agent
from typing import TypedDict
from langchain.agents.middleware import dynamic_prompt, ModelRequest


class CustomContext(TypedDict):
    user_name: str


def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is always sunny!"


@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest) -> str:
    user_name = request.runtime.context["user_name"]
    system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
    return system_prompt


agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[dynamic_system_prompt],
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

Output

================================ Human Message =================================

What is the weather in SF?
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_WFQlOGn4b2yoJrv7cih342FG)
 Call ID: call_WFQlOGn4b2yoJrv7cih342FG
  Args:
    city: San Francisco
================================= Tool Message =================================
Name: get_weather

The weather in San Francisco is always sunny!
================================== Ai Message ==================================

Hi John Smith, the weather in San Francisco is always sunny!

Before model

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

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Keep only the last few messages to fit context window."""
    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(
    model,
    tools=tools,
    middleware=[trim_messages]
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

"""
================================== Ai Message ==================================

Your name is Bob. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.
"""

After model

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

from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime


@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """Remove messages containing sensitive words."""
    STOP_WORDS = ["password", "secret"]
    last_message = 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="gpt-5-nano",
    tools=[],
    middleware=[validate_response],
    checkpointer=InMemorySaver(),
)

출처: https://docs.langchain.com/oss/python/langchain/short-term-memory

Langchain v1.0

반응형
저작자표시 (새창열림)
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유
langchain 관련 글
글 더보기

티스토리툴바