LLM 서비스를 위한 컨텍스트 관리 w. redis

FastApi 프레임워크를 기반으로 한 LLM 서비스를 구현하는 도중, LLM에 쿼리를 보내는 과정에 있어,

컨텍스트(대화 문맥, 흐름 ~ 여기서는 LLM이 기존에 사용자와 대화한 내용을 기억하기 위해 필요하다) 관리의 필요성을 느끼게 되었다

 

* 왜 Redis를 사용해야 할까?
1. 빠른 조회 속도 : Redis는 메모리 기반 데이터베이스로서, ms 단위 응답을 보장하기 때문에, 가볍고 빠르게 조회가 가능하다

2. TTL : Redis 자체적으로 제공하는 TTL 기능을 통해, 이력에 대해 일정 주기로 삭제를 진행해, 이력에 대한 용량 부담을 덜 수 있다

3. (추후 구현 해볼 예정) 벡터 DB 역할에 보다 특화되어, RAG를 위한 시맨틱 캐싱에 활용된다

 

* 비교해볼만한 다른 DB

1) PostgreSQL : 이력 데이터의 영구 저장과 벡터 DB 활용이 가능. 트랜젝션의 원자성을 보장한다. Redis보다 읽기/ 쓰기가 느리다

2) MongoDB : JSON 구조의 데이터를 그대로 저장할 때 유연하게 저장 가능. Redis와 비교해 지연시간이 발생한다

 

FAST API 기반 서비스에서 구현하기

0. main.py

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = aioredis.ConnectionPool.from_url(REDIS_URL, decode_responses=True)
    _redis = aioredis.Redis(connection_pool=pool)
    app.state.redis_client = _redis

    try:
        prompt_res = await api_client.get_prompt_list()

        if prompt_res:
            for prompt in prompt_res:
                role_key = f"{prompt['serviceType']}:role"
                await _redis.set(role_key, prompt['promptRole'])

                text_key = f"{prompt['serviceType']}:text"
                await _redis.set(text_key, prompt['promptText'])

        yield

    finally:
        await _redis.close()
        await pool.disconnect()

 

다른 DB에 저장되어 있는 프롬프트의 값을 가져와 redis에 저장한 후, 사용자 요청 시에 redis의 프롬프트를 가져와 사용하기 위한 초기 세팅 코드다

 

@asynccontextmanager

서비스의 생명주기에 동기화된 로직을 정의하기 위한 어노테이션이다. 이를 통해 서비스 기동시 해당 로직을 실행하고, 종료시 실행할 코드도 추가할 수 있다

 

app = FastAPI(
	lifespan= lifespan
)

이렇게 사용 가능하다

 

1. context에 활용

class class:
    def __init__(self, redis_client, api_client):
        self.redis_client = redis_client
        self.ttl = SESSION_TTL
        self.max_history = MAX_HISTORY

 

1-0) 프롬프트

async def get_prompt(self, service_type: str, prompt_key: str) -> str:
    return await self.redis_client.get(f"{service_type}:{prompt_key}") or ""

 

위와 같이 클래스에 redis를 연결해둔 다음, 밑에서 필요할때마다 호출하는 방식이다

1-1) 컨텍스트 관리

async def get_context_history(self, conversation_id: str) -> List[Any]:
    await self._ensure_redis_loaded(conversation_id)

    chat_key = f"{conversation_id}"
    raw_data = await self.redis_client.lrange(chat_key, 0, -1)

    if not raw_data:
        return []

    messages = self._parse_history(raw_data)

    return messages

sync def _ensure_redis_loaded(self, conversation_id: str) -> bool:
    """Redis에 데이터가 없으면 RDB에서 로드 + 캐싱. 데이터 존재 여부 반환"""
    chat_key = f"{conversation_id}"
    exists = await self.redis_client.llen(chat_key) > 0

    if exists:
        return True

    rdb_history = await self._fetch_history_from_rdb(conversation_id)
    if not rdb_history:
        return False

    await self._cache_history_to_redis(chat_key, rdb_history)
    return True

 

외부에서 컨텍스트를 조회하고 내부에서 컨텍스트 조회 및 캐싱하는 로직으로, _ensure_redis_loaded에 각 대화의 ID를 파라미터로 해 redis에 컨텍스트가 저장되어 있는지를 조회하고, 없는 경우엔 장기 저장된 DB에서 불러와 history의 연속성을 가져올 수 있게 구현했다

 

def build_user_prompt(self, context_history: List[dict], query: str) -> str:
    if not context_history:
        return query

    context_lines = []
    for msg in context_history:
        role = msg.get("role", "user")
        content = msg.get("content", "")

        label = "사용자" if role == "user" else "어시스턴트"
        context_lines.append(f"{label}: {content}")

    context_text = "\n".join(context_lines)
    return f"[이전 대화]\n{context_text}\n\n[현재 입력]\n{query}"

그렇게 가져온 컨텍스트와 새로 입력된 query를 가지고 사용자와 연속된 대화를 이어갈 수 있도록 만들어주는 로직이다

 

2. 마치며

이 구조를 통해 컨텍스트 관리를 빠르게 처리할 수 있다