LlamaIndex 에이전트 메모리

LlamaIndex 에이전트 메모리 (Memory)

에이전트가 이전 대화를 기억하고 활용하지 못한다면, 매번 같은 질문에 같은 답을 반복하게 될 거예요. 에이전트 시스템에서 메모리는 바로 이 "과거 정보를 저장하고 다시 꺼내는" 역할을 하는 핵심 구성 요소예요.

LlamaIndex에서는 이미 준비된 BaseMemory 클래스를 쓰거나, 직접 만든 커스텀 메모리를 사용해 메모리를 조정할 수 있어요. 에이전트가 실행되는 동안 메모리에 정보를 저장할 땐 memory.put()을, 정보를 꺼낼 땐 memory.get()을 호출해요.

⚠️ 참고: ChatMemoryBuffer는 더 이상 쓰지 않는(deprecated) 상태예요. 향후 릴리스에서는 기본값이 더 유연하고 복잡한 메모리 구성을 지원하는 Memory 클래스로 대체될 예정이에요. 이 문서의 예제는 Memory 클래스를 기준으로 해요. 참고로 지금 프레임워크 전반에서 기본값으로 쓰이는 ChatMemoryBuffer는 토큰 한도에 맞는 최근 X개의 메시지를 담는 단순 버퍼예요. Memory 클래스도 비슷하게 동작하지만 더 유연해서 다양한 구성을 할 수 있어요.

출처: 공식문서

기본 사용법

Memory 클래스로 만들면 단기 메모리(메시지의 FIFO 큐)와, 선택적으로 장기 메모리(시간에 따라 정보를 추출)를 함께 갖는 메모리를 구성할 수 있어요.

에이전트에 메모리 설정하기

에이전트의 메모리는 run() 메서드에 넘겨서 지정할 수 있어요.

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.memory import Memory


memory = Memory.from_defaults(session_id="my_session", token_limit=40000)


agent = FunctionAgent(llm=llm, tools=tools)


response = await agent.run("<question that invokes tool>", memory=memory)

메모리를 직접 관리하기

memory.put_messages()와 memory.get()을 직접 호출해서 채팅 히스토리를 넘겨주는 방식으로 메모리를 수동 관리할 수도 있어요.

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.llms import ChatMessage
from llama_index.core.memory import Memory




memory = Memory.from_defaults(session_id="my_session", token_limit=40000)
memory.put_messages(
    [
        ChatMessage(role="user", content="Hello, world!"),
        ChatMessage(role="assistant", content="Hello, world to you too!"),
    ]
)
chat_history = memory.get()


agent = FunctionAgent(llm=llm, tools=tools)


# passing in the chat history overrides any existing memory
response = await agent.run(
    "<question that invokes tool>", chat_history=chat_history
)

chat_history를 넘기면 기존 메모리를 덮어쓴다는 점만 기억해 두면 돼요.

에이전트의 최신 메모리 가져오기

에이전트 에이전트의 최신 메모리는 **에이전트 컨텍스트(Context)**에서 꺼낼 수 있어요.

from llama_index.core.workflow import Context


ctx = Context(agent)


response = await ctx.run("<question that invokes tool>", ctx=ctx)


# get the memory
memory = await ctx.store.get("memory")
chat_history = memory.get()

메모리 커스터마이즈

단기 메모리

Memory 클래스는 기본적으로 토큰 한도에 맞는 최근 X개의 메시지를 저장해요. 이건 token_limit과 chat_history_token_ratio 인자로 조정할 수 있어요.

  • token_limit (기본 30000): 저장할 단기·장기 메모리 토큰의 최대 개수.
  • chat_history_token_ratio (기본 0.7): 전체 토큰 한도에서 단기 채팅 히스토리가 차지하는 비율. 이 비율을 넘으면 가장 오래된 메시지가 장기 메모리로 밀려나요(장기 메모리가 켜져 있을 때).
  • token_flush_size (기본 3000): 채팅 히스토리가 토큰 한도를 넘었을 때 장기 메모리로 밀어내는 토큰 수.
memory = Memory.from_defaults(
    session_id="my_session",
    token_limit=40000,
    chat_history_token_ratio=0.7,
    token_flush_size=3000,
)

장기 메모리

장기 메모리는 Memory Block 객체로 표현돼요. 이 객체들은 단기 메모리에서 밀려난 메시지를 받아서, 선택적으로 정보를 추출해 처리해요. 메모리를 조회할 때는 단기·장기 메모리가 합쳐져서 반환돼요.

지금은 세 가지 미리 정의된 메모리 블록이 있어요.

  • StaticMemoryBlock: 고정된 정보를 저장하는 블록.
  • FactExtractionMemoryBlock: 채팅 히스토리에서 사실(fact)을 추출하는 블록.
  • VectorMemoryBlock: 벡터 데이터베이스에서 채팅 메시지 묶음을 저장하고 조회하는 블록.

기본적으로 insert_method 인자에 따라 이 메모리 블록들이 시스템 메시지 또는 가장 최신 사용자 메시지에 삽입돼요.

조금 복잡해 보이지만 실제로는 간단해요. 예제를 볼게요.

from llama_index.core.memory import (
    StaticMemoryBlock,
    FactExtractionMemoryBlock,
    VectorMemoryBlock,
)


blocks = [
    StaticMemoryBlock(
        name="core_info",
        static_content="My name is Logan, and I live in Saskatoon. I work at LlamaIndex.",
        priority=0,
    ),
    FactExtractionMemoryBlock(
        name="extracted_info",
        llm=llm,
        max_facts=50,
        priority=1,
    ),
    VectorMemoryBlock(
        name="vector_memory",
        # required: pass in a vector store like qdrant, chroma, weaviate, milvus, etc.
        vector_store=vector_store,
        priority=2,
        embed_model=embed_model,
        # The top-k message batches to retrieve
        # similarity_top_k=2,
        # optional: How many previous messages to include in the retrieval query
        # retrieval_context_window=5
        # optional: pass optional node-postprocessors for things like similarity threshold, etc.
        # node_postprocessors=[...],
    ),
]

여기서 세 개의 메모리 블록을 구성했어요.

  • core_info: 사용자에 대한 핵심 정보를 저장하는 정적 블록. static content는 문자열이거나 TextBlock·ImageBlock 같은 ContentBlock 객체의 리스트일 수 있어요. 이 정보는 항상 메모리에 삽입돼요.
  • extracted_info: 채팅 히스토리에서 정보를 추출하는 블록. 여기서는 밀려난 채팅 히스토리에서 사실을 추출하는 데 쓸 llm을 넘기고, max_facts를 50으로 설정했어요. 추출된 사실 수가 이 한도를 넘으면 max_facts가 자동으로 요약·축소되어 새 정보가 들어갈 공간을 확보해요.
  • vector_memory: 벡터 DB에서 채팅 메시지 묶음을 저장·조회하는 블록. 각 묶음은 밀려난 채팅 메시지들의 리스트예요. vector_store와 embed_model을 넘겨 메시지를 저장·조회해요.

각 블록에 priority를 설정했다는 점도 눈여겨볼게요. 이 값은 메모리 블록 내용(즉 장기 메모리) + 단기 메모리가 Memory 객체의 토큰 한도를 넘었을 때 어떻게 처리할지를 정하는 데 쓰여요.

메모리 블록이 너무 길어지면 자동으로 "잘려서(truncate)" 처리돼요. 기본적으로는 여유 공간이 생길 때까지 메모리에서 제거되는 방식이에요. 이건 자체 트렁케이션 로직을 구현한 메모리 블록 서브클래스로 커스터마이즈할 수 있어요.

  • priority=0: 이 블록은 항상 메모리에 유지돼요.
  • priority=1, 2, 3, ...: 메모리가 토큰 한도를 넘었을 때 블록이 잘리는 순서를 정해요. 최종적으로 단기+장기 메모리 내용이 token_limit 이하가 되도록 도와주는 역할이에요.

이제 이 블록들을 Memory 클래스에 넘겨볼게요.

memory = Memory.from_defaults(
    session_id="my_session",
    token_limit=40000,
    memory_blocks=blocks,
    insert_method="system",
)

메모리가 사용되면서 단기 메모리가 차오르면, 단기 메모리가 chat_history_token_ratio를 넘어서는 순간 token_flush_size에 맞는 가장 오래된 메시지들이 밀려나 각 메모리 블록으로 보내져 처리돼요. 메모리를 조회할 때는 단기·장기 메모리가 합쳐져요. Memory 객체는 단기+장기 메모리 내용이 token_limit 이하가 되도록 보장하고, 넘으면 priority에 따라 메모리 블록에 .truncate()을 호출해요.

💡 팁: 기본적으로 토큰은 tiktoken으로 세어요. 커스터마이즈하려면 tokenizer_fn 인자에 "문자열을 받아 리스트를 반환하는" 커스텀 콜러블을 설정하면 돼요. 그 리스트의 길이가 토큰 수로 사용돼요.

메모리에 정보가 충분히 쌓이면, 조회 결과는 이런 모습일 거예요.

# optionally pass in a list of messages to get, which will be forwarded to the memory blocks
chat_history = memory.get(messages=[...])


print(chat_history[0].content)

대략 이런 내용이 출력돼요.

<memory>
<static_memory>
My name is Logan, and I live in Saskatoon. I work at LlamaIndex.
</static_memory>
<fact_extraction_memory>
<fact>Fact 1</fact>
<fact>Fact 2</fact>
<fact>Fact 3</fact>
</fact_extraction_memory>
<retrieval_based_memory>
<message role='user'>Msg 1</message>
<message role='assistant'>Msg 2</message>
<message role='user'>Msg 3</message>
</retrieval_based_memory>
</memory>

여기서 메모리가 시스템 메시지에 삽입됐고, 각 메모리 블록에 해당하는 섹션이 따로 담겨 있어요.

커스텀 메모리 블록 만들기

미리 정의된 메모리 블록 말고, 직접 커스텀 메모리 블록을 만들 수도 있어요.

from typing import Optional, List, Any
from llama_index.core.llms import ChatMessage
from llama_index.core.memory.memory import BaseMemoryBlock




# use generics to define the output type of the memory block
# can be str or List[ContentBlock]
class MentionCounter(BaseMemoryBlock[str]):
    """
    A memory block that counts the number of times a user mentions a specific name.
    """


    mention_name: str = "Logan"
    mention_count: int = 0


    async def _aget(
        self, messages: Optional[List[ChatMessage]] = None, **block_kwargs: Any
    ) -> str:
        return f"Logan was mentioned {self.mention_count} times."


    async def _aput(self, messages: List[ChatMessage]) -> None:
        for message in messages:
            if self.mention_name in message.content:
                self.mention_count += 1


    async def atruncate(
        self, content: str, tokens_to_truncate: int
    ) -> Optional[str]:
        return ""

이 블록은 사용자가 특정 이름을 몇 번 언급했는지 세는 메모리 블록이에요. _aget가 조회 시 반환할 값을, _aput가 메시지가 들어올 때마다 처리할 로직을 정의해요. atruncate는 아주 단순하게 빈 문자열을 반환하는 방식이에요.

원격 메모리 (Remote Memory)

Memory 클래스는 기본적으로 인메모리 SQLite 데이터베이스를 사용해요. 데이터베이스 URI만 바꾸면 어떤 원격 데이터베이스든 연결할 수 있어요.

테이블 이름을 커스터마이즈할 수 있고, 선택적으로 비동기 엔진(async engine)을 직접 넘겨줄 수도 있어요. 이는 직접 커넥션 풀을 관리하고 싶을 때 유용해요.

from llama_index.core.memory import Memory


memory = Memory.from_defaults(
    session_id="my_session",
    token_limit=40000,
    async_database_uri="postgresql+asyncpg://postgres:mark90@localhost:5432/postgres",
    # Optional: specify a table name
    # table_name="memory_table",
    # Optional: pass in an async engine directly
    # this is useful for managing your own connection pool
    # async_engine=engine,
)

메모리 vs. 워크플로 컨텍스트

지금쯤이면 워크플로(Workflow)를 쓰면서 특정 워크플로 상태를 저장·재개하기 위해 Context 객체를 직렬화해야 하는 상황을 겪었을 수도 있어요. 워크플로 Context는 워크플로의 런타임 정보와, 워크플로 단계들 사이에 공유되는 키/값 쌍을 담는 복잡한 객체예요.

이에 비해 Memory 객체는 더 단순해서 ChatMessage 객체들만, 그리고 선택적으로 장기 메모리를 위한 MemoryBlock 객체 리스트를 담아요.

실무에서는 대부분 둘 다 쓰게 돼요. 메모리를 커스터마이즈하지 않는다면 Context 객체만 직렬화해도 충분해요.

from llama_index.core.workflow import Context


ctx = Context(workflow)


# serialize the context
ctx_dict = ctx.to_dict()


# deserialize the context
ctx = Context.from_dict(workflow, ctx_dict)

반면 FunctionAgent, AgentWorkflow, ReActAgent를 쓰면서 메모리를 커스터마이즈했다면, 그 메모리를 별도의 런타임 인자로 넘겨주는 게 좋아요(기본값을 벗어나면 Memory 객체는 직렬화가 불가능하기 때문이에요).

response = await agent.run("Hello!", memory=memory)

마지막으로 휴먼 인 더 루프(human-in-the-loop)처럼 Context(워크플로 재개용)와 Memory(채팅 히스토리 저장용)를 둘 다 제공해야 하는 경우도 있어요.

response = await agent.run("Hello!", ctx=ctx, memory=memory)

(구버전) 메모리 유형

llama_index.core.memory에는 지금도 몇 가지 메모리 유형이 제공돼요.

  • ChatMemoryBuffer: 토큰 한도에 맞는 최근 X개의 메시지를 저장하는 기본 메모리 버퍼.
  • ChatSummaryMemoryBuffer: 최근 X개 메시지를 저장하면서, 대화가 너무 길어지면 주기적으로 요약하는 메모리 버퍼.
  • VectorMemory: 벡터 DB에서 채팅 메시지를 저장·조회하는 메모리. 메시지 순서는 보장하지 않고, 최신 사용자 메시지와 가장 유사한 메시지를 반환해요.
  • SimpleComposableMemory: 여러 메모리를 조합하는 메모리. 보통 VectorMemory를 ChatMemoryBuffer나 ChatSummaryMemoryBuffer와 함께 쓸 때 사용해요.

더 알아보기