개요

AI 프로젝트를 진행하며 Pinecone에 대한 기능을 간단하게 테스트한 내용을 정리하였습니다.

Pinecone

Pinecone(파인콘)은 인공지능(AI) 및 머신러닝(ML) 애플리케이션을 위해 특별히 설계된 클라우드 기반의 완전 관리형 벡터 데이터베이스(Vector Database) 입니다. Pinecone은 대규모의 고차원 벡터 임베딩(Vector Embeddings)을 효율적으로 저장, 관리, 검색하는 데 중점을 둡니다.

기존의 관계형 데이터베이스(RDB)가 테이블 형태의 정형 데이터를 다루는 데 최적화되어 있다면, Pinecone은 텍스트, 이미지, 오디오 등 비정형 데이터를 수치화된 벡터(Vector) 형태로 변환하여, 이들 간의 유사성을 빠르고 정확하게 찾아내는 것에 특화되어 있습니다.

Pinecone의 장점

• 확장성: 대규모 데이터셋에 대해 뛰어난 확장성을 제공합니다.
• 관리 용이성: 완전 관리형 SaaS 서비스로, 인프라 관리 부담이 적습니다.
• 실시간 업데이트: 데이터의 실시간 삽입, 업데이트, 삭제가 가능합니다.

Pinecone의 단점

• 커스터마이징 제한: 완전 관리형 서비스이기 때문에 세부적인 커스터마이징에 제한이 있을 수 있습니다.
  • OpenSearch keyword 검색을 지원하지 않습니다.
• 데이터 위치: 클라우드에 데이터를 저장해야 하므로, 데이터 보안 문제가 발생할 수 있습니다.

데이터 전처리 과정

간단한 기사 텍스트 데이터를 확인하여 해당 내용을 통해 구성하였습니다.

문단 단위 청킹 (Sentence Splitting):

Model은 한 번에 처리할 수 있는 입력 텍스트의 양에 제한이 있습니다. 이를 '컨텍스트 창(Context Window)'이라고 부르며, 토큰(Token, 대략 단어 또는 형태소) 수로 측정합니다. 또한 트랜스포머 기반 모델의 핵심인 어텐션 메커니즘은 문맥을 파악하는 데 뛰어나지만, 입력 길이가 매우 길어지면 문맥의 중요한 부분을 놓치거나('Lost in the Middle') 문맥의 희소성이 저하될 수 있습니다.

이러한 문제들은 RAG 파이프라인에서 검색 성능 저하를 불러올 수 있으므로 적절한 크기로 문서를 청킹하여 조절해야합니다.(청크간의 중요한 연결 고리를 유지하는 것 또한 성능에 주요한 영향을 줄 수 있습니다.) 이를 극복하기 위한 재귀 기반, 의미론적 청킹 등등 다양한 기법들이 연구되고 있습니다.

따라서 문서를 먼저 청크로 나눈 다음, 각 청크에 대해 작업을 수행하고(Map), 그 결과를 종합하거나(Reduce), 점진적으로 개선하는(Refine) 방식을 고민해봐야하며 청킹은 이러한 복잡한 처리 파이프라인을 가능하게 하는 기본 단계입니다.

from utils.pinecone import create_metadata

chunking_articles = []

def split_paragraphs(text):
    # 줄바꿈(\n) 기준으로 자르고, 공백 문단은 제거
    paragraphs = [p.strip() for p in text.split('\n') if p.strip()]
    return paragraphs

# article_dict["tx"]는 이미 문단 리스트라고 가정
for article_dict in articles:
    # article_dict["tx"]가 이미 문단 리스트라면 그대로 content에 할당
    metadatas = create_metadata(article_dict)
    chunking_articles.append({"metadata": metadatas, "content": split_paragraphs(article_dict["tx"])})

Sparse Vector

Pinecone의 경우 Keyword Search를 지원하지 않기 때문에 검색 성능을 향상시키기 위한 Hybrid Search를 도입하기 위해선 Sparse Embedding을 통해 구축해야합니다

이를 자체적으로 구축하기 위해서 Sparse Encoder로 BM25를 사용하였으며, 한국어 성능을 높히기 위한 형태소, 불용어 사전을 적용하였습니다. 하지만 이러한 키워드 서치의 방식은 자체적인 말뭉치(Corpus)를 가져와 실행하므로, 데이터가 많으면 많을 수록 말뭉치를 가져와 검색하는 것이 서버에 상당히 부하를 줄 수 있으니 데이터가 많은 경우 OpenSearch, ElasticSearch를 권장하고 있습니다.

from utils.koreans import create_sparse_encoder, fit_sparse_encoder, load_sparse_encoder
bm25 = create_sparse_encoder(mode="kiwi")

# 문단으로 합쳐서 Sparse Encoder 학습
all_paragraphs = [para for item in chunking_articles for para in item["content"]]
fit_sparse_encoder(bm25, all_paragraphs)

Embedding Model 불러오기

sparse_encoder  = load_sparse_encoder()

print(sparse_encoder)

from langchain_aws import BedrockEmbeddings

# Cohere Embedding Model 사용
embeddings = BedrockEmbeddings(
    credentials_profile_name="smileshark", 
    region_name="us-east-1",
    model_id="cohere.embed-multilingual-v3"
)

Pinecone VectorDB 연결

describe_index_stats 등을 통해 간단한 인덱스 통계 정보를 활용할 수 있습니다.

from pinecone import Pinecone
from pinecone import ServerlessSpec

pc = Pinecone(api_key="")

index_name = "bjchoi-hybrid-index"

# pc.delete_index(index_name)

index = pc.Index(host='bjchoi-hybrid...o')

if not pc.has_index(index_name):
    index_info = pc.create_index(
        name=index_name,
        vector_type="dense",
        dimension=1024,
        metric="dotproduct",
        spec=ServerlessSpec(
            cloud="aws",
            region="us-east-1"
        )
    )
    print(index_info)

    index = pc.Index(host=index_info["host"])
    # index.describe_index_stats()

Pinecone Index upsert

인덱스가 생성되었다면 청킹한 content를 임베딩 -> 업로드 해야합니다.
대용량 문서를 업데이트하기 위해 Pinecone에서는 병렬 배치 업로드를 지원합니다.

import numpy as np

batch_size = 100  # 원하는 배치 크기로 조절

records = [
    {
        "id": f"{article_dict['metadata'].get('no')}_{i}",
        "values": embeddings.embed_documents([text])[0],
        "sparse_values": sparse_encoder.encode_documents([text])[0],
        "metadata": {**article_dict["metadata"], "text": text}
    }
    for article_dict in chunking_articles
    for i, text in enumerate(article_dict["content"])
]

for i in range(0, len(records), batch_size):
    batch = records[i:i+batch_size]
    index.upsert(vectors=batch)

Retriever Query 생성

이제 VectorDB에 업로드 되었으므로 벡터 검색을 수행합니다.
검색하고자하는 쿼리를 각각 Sparse와 Dense 방식으로 엠비딩을 수행한 뒤 해당 검색어가 얼마나 일치하는지 Score를 출력할 수 있습니다.

결과적으로 정리하면 다음과 같습니다.

• Dense Vector & Sparse Vector를 통한 HybridSearch Retriever
• 가중치 조절 가능
• Metadata를 활용한 동적 필터링 적용 가능합니다. (Filter, Rerank, top_n, k)

-> 의미론적 검색만 수행하는 경우

query = "AI 시대 이끌어 갈 인재 양성"

# Dense Vector & Sparse Vector 엠베딩 수행
dense_embed = embeddings.embed_query(query)

query_response = index.query(
    vector=dense_embed,
    include_metadata=True,
    include_values=False,
    filter= {"no": {"$ne" : "00001089"}},
    top_k=20 # 가장 유사한 20개만 출력 / 모든 결과를 가져올 수 없으므로 약 10,000개의 Chunk or 문서 제한이 존재합니다.

{'matches': [{'id': '00064321_6',
              'metadata': {'content': '서울 강남구 대치동 ...',
              'score': 0.603467703,
              'values': []},
              ...,
    }]
}

Hybrid 방식으로 수행하는 경우

from pinecone_text.hybrid import hybrid_convex_scale

# Hybrid Search Query
query = "사교육 열기"


# 가중치 적용 => 1에 가까울 수록 dense 벡터(유사도 검색 방식)의 가중치가 높아집니다.
# 검색어와 값을 조정해보며 다른 가중치에서 유사도 점수가 조정되는 방식을 확인해보세요.
alpha = 0.2

# Dense Vector & Sparse Vector 엠베딩 수행
dense_embed = embeddings.embed_query(query)
sparse_embed = sparse_encoder.encode_queries(query)  # 리스트/배열 형태

dense_vec, sparse_vec = hybrid_convex_scale(dense_embed, sparse_embed, alpha=alpha)

query_response = index.query(
    vector=dense_vec, # scale된 dense 벡터
    sparse_vector=sparse_vec, # scale된 sparse 벡터
    include_metadata=True,
    include_values=False,
    filter= {"no": {"$ne" : "000010"}},
    top_k=20 # 가장 유사한 20개만 출력 / 모든 결과를 가져올 수 없으므로 약 10,000개의 Chunk or 문서 제한이 존재합니다.
)

Reranking

테스트를 위해 Bedrock Cohere 3.5 Rerank 모델을 사용하였습니다.
bge-reranker-v2-m3 모델을 활용하여 오픈 소스로 구현하는 방법도 있지만, 테스트를 위해 Cohere 모델을 사용했습니다.

Bedrock Cohere 3.5를 연결합니다.

import boto3

session = boto3.Session(profile_name="smileshark")
bedrock_agent_runtime = session.client('bedrock-agent-runtime',region_name="us-west-2")

modelId = "cohere.rerank-v3-5:0"
model_package_arn = f"arn:aws:bedrock:us-west-2::foundation-model/{modelId}"

# ------------------------------------------------
# text_query: 사용자가 검색한 쿼리
# text_sources: RAG 검색 결과
# num_results: 가장 유사한 결과의 개수
# model_package_arn: 모델 패키지 ARN
def rerank_text(text_query, text_sources, num_results, model_package_arn):
    response = bedrock_agent_runtime.rerank(
        queries=[
            {
                "type": "TEXT",
                "textQuery": {
                    "text": text_query
                }
            }
        ],
        sources=text_sources,
        rerankingConfiguration={
            "type": "BEDROCK_RERANKING_MODEL",
            "bedrockRerankingConfiguration": {
                "numberOfResults": num_results,
                "modelConfiguration": {
                    "modelArn": model_package_arn,
                }
            }
        }
    )
    return response['results']

# Cohere Reranking 모델에 맞는 데이터 형식으로 변환합니다.
text_sources = []

for match in filtered_articles:
    text_sources.append({
        "type": "INLINE",
        "inlineDocumentSource": {
            "type": "TEXT",
            "textDocument": {
                "text": match['metadata']['text'],
            }
        }
    })

print(len(text_sources))

Rerank 수행

이때 결과는 다음과 같이 출력됩니다.

[
  {
    "index": 2,
    "relevanceScore": 0.6752727627754211
  },
  {
    "index": 3,
    "relevanceScore": 0.6734714508056641
  },
  {
    "index": 1,
    "relevanceScore": 0.6570330262184143
  }
]

LLM에서는 해당 Score가 높은 순서대로 LLM에게 전달합니다.

RAG를 구현하는데 있어 'Lost in Middle'이라는 문제가 대두되었습니다. Reranker는 이러한 문제를 해결하기 위한 것으로 문서의 순서를 적용하여 가장 유사한 Document가 제일 먼저 입력될 수 있도록 조정하는 방식으로 더 정교한 유사도 측정을 통해 응답과 검색 품질을 모두 향상시킬 수 있습니다.

답변 생성

Claude Sonnet 4같은 최신 모델의 경우 InferenceProfileArn을 활용해야합니다.

Inference Profile을 생성하여 특정 모델의 비용을 분리할 수 있습니다.

# bedrock profile 생성
import boto3

account_id = "759..."
profile_name = "bjchoi-claude-sonnet-4-profile"

session = boto3.Session(profile_name="bjchoi-test")

# 수임된 역할로 Bedrock 클라이언트 생성
bedrock = session.client(
    'bedrock',
    region_name='us-east-1'  # 필요한 리전으로 변경
)

def create_inference_profile(profile_name, model_arn):
    """Create Inference Profile using base model ARN"""
    response = bedrock.create_inference_profile(
        inferenceProfileName=profile_name,
        description="test",
        modelSource={'copyFrom': model_arn},
        # tags=tags
    )
    print("CreateInferenceProfile Response:", response['ResponseMetadata']['HTTPStatusCode']),
    print(f"{response}\n")

    return response

base_model_arn = f"arn:aws:bedrock:us-east-1:{account_id}:inference-profile/us.anthropic.claude-sonnet-4-20250514-v1:0"

from langchain_aws import ChatBedrockConverse

inferenceProfileArn = "us.anthropic.claude-sonnet-4-20250514-v1:0"

llm = ChatBedrockConverse(
    model=inferenceProfileArn,
    temperature=0,
    max_tokens=None,
    credentials_profile_name="bjchoi-test",
    region_name="us-east-1"
)

프롬프트 작성

# 시스템 프롬프트 내용
system_prompt = """
1. 당신의 정체성:
당신은 '친절한 해설가' 봇입니다. 당신의 목소리는 항상 경어를 사용하며, 독자에게 존중과 신뢰감을 줍니다.

...
"""

title = article_dict['ti']
content = article_dict['tx']

# messages 리스트 생성
messages = [
    ("system", system_prompt),
    ("assistant", f" title: {title} content: {content}"), # 사용자가 제공한 형식 유지
    ("human", f"<키워드> {query} </키워드>"),
]

response = llm.invoke(messages)

print(response.content)

LLM 평가하기

대규모 언어 모델(LLM)의 성능 테스트는 단순히 모델이 얼마나 '똑똑한지'를 넘어, 우리가 LLM을 신뢰하고 효과적으로 활용하기 위한 필수적인 과정입니다. 또한 생성형 AI 애플리케이션의 지속적인 개선 여지를 만드는데 중요한 역할을 하고 있으므로 LLM 구현에 있어 반드시 진행되어야하는 단계입니다.

현재 구성에서는 Deepeval을 사용하여 어떻게 성능 평가를 하였는지 작성하였습니다.

그 중에서 집중적으로 확인해본 부분은 RAG에 대한 성능 지표 입니다.

• https://deepeval.com/docs/metrics-answer-relevancy
• https://deepeval.com/guides/guides-rag-evaluation

평가 LLM 설정
Bedrock Claude 모델로 변경합니다.

from langchain_community.chat_models import BedrockChat
from deepeval.models.base_model import DeepEvalBaseLLM


class AWSBedrock(DeepEvalBaseLLM):
    def __init__(
        self,
        model
    ):
        self.model = model

    def load_model(self):
        return self.model

    def generate(self, prompt: str) -> str:
        chat_model = self.load_model()
        return chat_model.invoke(prompt).content

    async def a_generate(self, prompt: str) -> str:
        chat_model = self.load_model()
        res = await chat_model.ainvoke(prompt)
        return res.content

    def get_model_name(self):
        return "Bedrock Claude Sonnet 4"

aws_bedrock = AWSBedrock(model=llm)
print(aws_bedrock.generate("Write me a joke"))

RAG 성능 평가

Deepeval에서는 RAG 성능 평가를 위해 검색 단계에서 확인할 수 있는 3가지 측정 지표를 제공하고 있습니다.

• ContextualPrecisionMetric (문맥 정밀도 메트릭):
  • 검색기(retriever)의 재순위기(reranker)가 검색된 문맥 내에서 관련성이 낮은 노드보다 관련성이 높은 노드를 더 높은 순위로 매기는지를 평가합니다.
• ContextualRecallMetric (문맥 재현율 메트릭):
  • 검색기(retriever)의 임베딩 모델이 입력 문맥을 기반으로 관련 정보를 정확하게 포착하고 검색할 수 있는 능력을 평가합니다.
• ContextualRelevancyMetric (문맥 관련성 메트릭):
  • 검색기(retriever)의 텍스트 청크(chunk) 크기와 상위 K개(top-K) 설정이 불필요한 정보(irrelevancies)를 최소화하면서 정보를 검색할 수 있는지를 평가합니다.

from deepeval import evaluate
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    ContextualPrecisionMetric,
    ContextualRecallMetric,
    ContextualRelevancyMetric
)

contextual_precision = ContextualPrecisionMetric(
    threshold=0.7,
    model=aws_bedrock,
    include_reason=True
)
contextual_recall = ContextualRecallMetric(
    threshold=0.7,
    model=aws_bedrock,
    include_reason=True
)
contextual_relevancy = ContextualRelevancyMetric(
    threshold=0.7,
    model=aws_bedrock,
    include_reason=True
)

테스트 케이스 생성

실제 사용시와 테스트시 간극이 발생하지 않기 위해선 해당 도메인의 데이터를 잘 이해하고 있거나 신뢰성 있는 데이터 소스에서 합성 데이터를 생성하여 검증하는 것이 좋습니다.
deepeval을 사용하는 경우 기본적으로 LLM-as-a-judge를 지원합니다. Score를 통해 1차로 데이터를 거르고 그 외 데이터는 사람이 수동으로 검수할 수 있습니다.

자세한 테스트 요약까지 확인하고자 한다면 다음과 같이 수행할 수 있습니다.

test_case = LLMTestCase(
    input=query, # 현재 사교육 시장의 상황
    actual_output=response.content,
    expected_output="...교육의 중심지 대치동이 자리잡아...",
    retrieval_context = [match["metadata"]["text"] for match in reranking_articles]
)

rag_result = evaluate(test_cases=[test_case], metrics=[contextual_precision, contextual_relevancy, contextual_recall])

✨ You're running DeepEval's latest Contextual Precision Metric! (using Bedrock Claude Sonnet 4, strict=False, 
async_mode=True)...

✨ You're running DeepEval's latest Contextual Relevancy Metric! (using Bedrock Claude Sonnet 4, strict=False, 
async_mode=True)...

✨ You're running DeepEval's latest Contextual Recall Metric! (using Bedrock Claude Sonnet 4, strict=False, 
async_mode=True)...

Metrics Summary

  - ✅ Contextual Precision (score: 1.0, threshold: 0.7, strict: False, evaluation model: Bedrock Claude Sonnet 4, reason: The score is 1.00 because all relevant nodes are perfectly ranked above irrelevant ones. The first node in the retrieval contexts directly addresses the current private education market situation with information about rising private education costs despite declining school-age population, which perfectly matches the input query. The second node, while containing related information about changes in Daechi-dong's private education market, is appropriately ranked lower as it focuses on specific regional trends rather than the overall market situation requested., error: None)
  - ✅ Contextual Relevancy (score: 1.0, threshold: 0.7, strict: False, evaluation model: Bedrock Claude Sonnet 4, reason: The score is 1.00 because the retrieval context perfectly addresses the current state of the private education market with comprehensive coverage including record-high spending despite declining student population, the concentration trend toward large academies, and government investigations into educational cartels., error: None)
  - ✅ Contextual Recall (score: 1.0, threshold: 0.7, strict: False, evaluation model: Bedrock Claude Sonnet 4, reason: The score is 1.00 because both sentences in the expected output are perfectly matched in node 1 in retrieval context, providing complete attribution for the entire response., error: None)

For test case:

...

======================================================================

Overall Metric Pass Rates

Contextual Precision: 100.00% pass rate
Contextual Relevancy: 100.00% pass rate
Contextual Recall: 100.00% pass rate

✓ Tests finished 🎉! Run 'deepeval login' to save and analyze evaluation results on Confident AI.

✨👀 Looking for a place for your LLM test data to live 🏡❤️ ? Use Confident AI to get & share testing reports, 
experiment with models/prompts, and catch regressions for your LLM system. Just run 'deepeval login' in the CLI. 

contextual_precision.a_measure(test_case)
contextual_relevancy.a_measure(test_case)
contextual_recall.a_measure(test_case)

print(contextual_precision.score, contextual_relevancy.score, contextual_recall.score)
print(contextual_precision.reason, contextual_relevancy.reason, contextual_recall.reason)

==================================
output

1.0 1.0 1.0
The score is 1.00 because all relevant nodes are perfectly ranked above irrelevant ones. The first node in the retrieval contexts 
...
The score is 1.00 because both sentences in the expected output are perfectly matched in node 1 in retrieval context, providing complete attribution for the entire response.