Elasticsearch 9.x를 활용한 로컬 배포

많은 개발자들이 AI 에이전트를 처음 시작할 때 직면하는 문제 중 하나는 에이전트가 대화를 기억하지 못한다는 것입니다. 이 문제를 해결하기 위해 Elasticsearch(ES)를 사용하여 "메모리 저장소"를 구축할 수 있습니다.

1.1 LLM이 왜 기억하지 못하는지 이해하기

대형 언어 모델(LLM)은 본질적으로 상태가 없기 때문에 이전 대화를 기억하지 못합니다. 이를 해결하려면 각 대화를 다시 전달하거나 중요한 정보만 선택적으로 저장해야 합니다.

1.2 세 가지 유형의 기억: 행동, 개인 일지, 일반 상식

AI의 기억은 세 가지 유형으로 나눌 수 있습니다:

• 행동 기억: AI가 어떻게 행동해야 하는지를 알려주는 기본적인 규칙
• 개인 일지: 특정 사용자와의 대화 기록 및 선호도
• 일반 상식: 모든 사용자가 공유하는 지식

1.3 선택적 기억의 필요성

선택적 기억은 다음과 같은 문제를 해결합니다:

1. 혼란 방지: 관련 없는 대화를 제외하고 필요한 정보만 추출
2. 비용 절감: 불필요한 토큰 사용을 줄임
3. 혼동 방지: 사용자별로 기억을 분리하여 혼동을 방지

2. Elasticsearch 9.x 선택 이유 및 로컬 배포의 장점

Elasticsearch 9.x는 다음과 같은 이유로 선택되었습니다:

• 의미 검색: 사용자의 의도를 이해하여 정확한 결과 반환
• 정밀 필터링: 태그를 통해 메모리를 분류하여 검색 가능
• 고속 처리: 대량의 데이터도 빠르게 처리 가능

로컬 배포의 장점:

• 비용 절감: 클라우드보다 저렴한 로컬 환경 사용
• 데이터 보안: 자체 서버에서 데이터 관리 가능
• 디버깅 용이: 로컬 환경에서 쉽게 설정 변경 및 로그 확인 가능

3. Elasticsearch 9.x 로컬 배포 실습

3.1 사전 준비

• 최소 4GB 이상의 RAM (권장 8GB)
• Python 3.x 설치 (3.8 이상 권장)
• OpenAI API Key (LLM 호출에 필요)

3.2 Elasticsearch 설치 및 구성

# Elasticsearch 클라이언트 설치
pip install elasticsearch openai

# Elasticsearch 연결
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200")

# 인덱스 생성
mappings = {
    "properties": {
        "user_id": {"type": "keyword"},
        "memory_type": {"type": "keyword"},
        "created_at": {"type": "date"},
        "memory_text": {
            "type": "text",
            "fields": {
                "semantic": {
                    "type": "semantic_text",
                    "model_id": ".elser_model_2"
                }
            }
        }
    }
}
if not es.indices.exists(index="memories"):
    es.indices.create(index="memories", mappings=mappings)
    print("메모리 인덱스 생성 성공!")

3.3 ELSER 모델 배포

curl -X PUT "http://localhost:9200/_ml/trained_models/.elser_model_2?pretty" -H "Content-Type: application/json" -d'
{
  "input": {
    "field_names": ["text_field"]
  }
}'
curl -X POST "http://localhost:9200/_ml/trained_models/.elser_model_2/deployment/_start?pretty"

4. AI 에이전트 메모리 시스템 구축

4.1 메모리 인덱스 생성

def create_memory_index():
    if not es.indices.exists(index="memories"):
        es.indices.create(index="memories", mappings=mappings)
        print("메모리 인덱스 생성 성공!")
    else:
        print("메모리 인덱스 이미 존재")

4.2 문서 수준 보안 설정

innie_role_descriptor = {
    "indices": [
        {
            "names": ["memories"],
            "privileges": ["read", "write"],
            "query": {
                "bool": {
                    "filter": [{"term": {"memory_type": "innie"}}]
                }
            }
        }
    ]
}

outie_role_descriptor = {
    "indices": [
        {
            "names": ["memories"],
            "privileges": ["read", "write"],
            "query": {
                "bool": {
                    "filter": [{"term": {"memory_type": "outie"}}]
                }
            }
        }
    ]
}

es.security.put_role(name="innie_role", body=innie_role_descriptor)
es.security.put_role(name="outie_role", body=outie_role_descriptor)
print("작업/개인 메모리 역할 생성 성공!")

es.security.put_user(
    username="peter",
    password="peter123",
    roles=["outie_role"]
)

es.security.put_user(
    username="janice",
    password="janice123",
    roles=["innie_role"]
)
print("사용자 생성 성공!")

4.3 메모리 저장 및 검색 도구 정의

def store_memory(conversation, user_id, memory_type):
    prompt = f"""다음 대화를 요약하여 간결한 기억으로 변환하세요:
대화: {conversation}
기억 형식: [{user_id}의 주요 정보].
예시: Peter의 생일은 내일이고, 그는 스테이크를 먹고 싶어합니다."""
    
    response = openai_client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    memory_text = response.choices[0].message.content.strip()
    
    es.index(
        index="memories",
        document={
            "user_id": user_id,
            "memory_type": memory_type,
            "created_at": "now",
            "memory_text": memory_text
        }
    )
    print(f"기억 저장 성공! 내용: {memory_text}")

def retrieve_memories(query, user_credentials):
    user_es_client = Elasticsearch(
        "http://localhost:9200",
        basic_auth=user_credentials
    )
    
    es_query = {
        "retriever": {
            "rrf": {
                "retrievers": [
                    {
                        "standard": {
                            "query": {
                                "semantic": {
                                    "field": "memory_text.semantic",
                                    "query": query
                                }
                            }
                        }
                    },
                    {
                        "standard": {
                            "query": {
                                "multi_match": {
                                    "query": query,
                                    "fields": ["memory_text"]
                                }
                            }
                        }
                    }
                ],
                "rank_window_size": 50,
                "rank_constant": 20
            }
        }
    }
    
    response = user_es_client.search(index="memories", body=es_query)
    memories = [hit["_source"]["memory_text"] for hit in response["hits"]["hits"]]
    return memories

4.4 에이전트 대화 로직 구현

def agent_conversation(user_query, user_credentials, user_id, memory_type):
    messages = [{"role": "user", "content": user_query}]
    
    response = openai_client.responses.create(
        model="gpt-4.1-mini",
        input=messages,
        tools=[
            {
                "type": "function",
                "function": {
                    "name": "retrieve_memories",
                    "description": "사용자가 과거 경험이나 선호도를 물을 때 이 도구를 사용하여 기억을 검색합니다.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "store_memory",
                    "description": "사용자가 새로운 정보를 공유할 때 이 도구를 사용하여 기억을 저장합니다.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "conversation": {"type": "string"},
                            "user_id": {"type": "string"},
                            "memory_type": {"type": "string", "enum": ["innie", "outie"]}
                        },
                        "required": ["conversation", "user_id", "memory_type"]
                    }
                }
            }
        ],
        parallel_tool_calls=True
    )
    
    for tool_call in response.output:
        if tool_call.name == "retrieve_memories":
            memories = retrieve_memories(tool_call.arguments["query"], user_credentials)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": f"검색된 기억: {memories}"
            })
        elif tool_call.name == "store_memory":
            store_memory(tool_call.arguments["conversation"], tool_call.arguments["user_id"], tool_call.arguments["memory_type"])
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": "기억이 성공적으로 저장되었습니다."
            })
    
    final_response = openai_client.responses.create(
        model="gpt-4.1-mini",
        input=messages
    )
    
    return final_response.output.text

5. 테스트 실행

5.1 Peter의 개인 기억 저장

conversation1 = "Peter: 안녕 Mark, 내 생일은 내일이고 저녁에는 스테이크를 먹고 싶어요."
store_memory(conversation1, "peter125", "outie")

5.2 Janice의 작업 기억 저장

conversation2 = "Janice: 안녕 Mark, 내일 오전 9시까지 연간 보고서를 마쳐야 해요."
store_memory(conversation2, "janice456", "innie")

5.3 기억 격리 테스트

query1 = "안녕 Mark, 내 생일에 무엇을 먹고 싶었나요?"
answer1 = agent_conversation(query1, ("peter", "peter123"), "peter125", "outie")
print(f"Peter의 질문: {query1}")
print(f"AI의 답변: {answer1}")

query2 = "안녕 Mark, 연간 보고서는 언제까지 제출해야 하나요?"
answer2 = agent_conversation(query2, ("peter", "peter123"), "peter125", "outie")
print(f"Peter의 질문: {query2}")
print(f"AI의 답변: {answer2}")