LLM API 실전 활용: RAG 파이프라인으로 나만의 AI 서비스 만들기와 운영 전략

# RAG 파이프라인의 핵심 흐름 (간략 버전)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. 임베딩 모델 초기화
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key="YOUR_OPENAI_API_KEY"
)

# 2. 벡터 DB에서 리트리버 생성
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# 3. LLM + 리트리버 결합
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    api_key="YOUR_OPENAI_API_KEY"
)

# 4. RAG 체인 구성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 5. 질문-응답
result = qa_chain.invoke({"query": "Spring Boot에서 트랜잭션 전파란?"})
print(result["result"])

import chromadb
from chromadb.config import Settings

# ChromaDB 클라이언트 생성
client = chromadb.PersistentClient(path="./chroma_db")

# 컬렉션 생성
collection = client.get_or_create_collection(
    name="tech_documents",
    metadata={"hnsw:space": "cosine"}  # 코사인 유사도
)

# 문서 추가
collection.add(
    documents=[
        "Spring Boot는 스프링 프레임워크 기반의 웹 애플리케이션 프레임워크입니다.",
        "Redis는 인메모리 데이터 스토어로 캐싱에 주로 사용됩니다.",
        "Docker 컨테이너는 애플리케이션의 이식성을 높여줍니다."
    ],
    metadatas=[
        {"source": "spring-docs", "chapter": 1},
        {"source": "redis-docs", "chapter": 1},
        {"source": "docker-docs", "chapter": 1}
    ],
    ids=["doc1", "doc2", "doc3"]
)

# 유사도 검색
results = collection.query(
    query_texts=["스프링 부트 사용법"],
    n_results=2,
    include=["documents", "metadatas", "distances"]
)
print(results)

from pinecone import Pinecone, ServerlessSpec

# Pinecone 초기화
pc = Pinecone(api_key="YOUR_PINECONE_API_KEY")

# 인덱스 생성
pc.create_index(
    name="tech-knowledge-base",
    dimension=1536,  # text-embedding-3-small 차원
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

index = pc.Index("tech-knowledge-base")

# 벡터 업서트
index.upsert(
    vectors=[
        {
            "id": "doc-001",
            "values": embedding_vector,  # 1536차원 벡터
            "metadata": {
                "source": "spring-docs",
                "text": "Spring Boot 트랜잭션 관리...",
                "category": "backend"
            }
        }
    ],
    namespace="tech-docs"
)

# 유사도 검색
query_result = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True,
    namespace="tech-docs",
    filter={"category": {"$eq": "backend"}}
)

# OpenAI 임베딩 (가장 널리 사용)
from langchain_openai import OpenAIEmbeddings

embeddings_small = OpenAIEmbeddings(
    model="text-embedding-3-small",  # 1536차원, 저비용
    api_key="YOUR_OPENAI_API_KEY"
)

embeddings_large = OpenAIEmbeddings(
    model="text-embedding-3-large",  # 3072차원, 고정밀
    api_key="YOUR_OPENAI_API_KEY"
)

# 한국어 특화: multilingual-e5-large
from langchain_huggingface import HuggingFaceEmbeddings

korean_embeddings = HuggingFaceEmbeddings(
    model_name="intfloat/multilingual-e5-large",
    model_kwargs={"device": "cuda"},  # GPU 사용
    encode_kwargs={"normalize_embeddings": True}
)

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 가장 기본적이고 안정적인 방식
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 청크 크기 (문자 수)
    chunk_overlap=50,     # 오버랩 (문맥 유지)
    separators=["\n\n", "\n", ".", " ", ""],
    length_function=len
)

chunks = text_splitter.split_text(document_text)

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(api_key="YOUR_OPENAI_API_KEY")

semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

semantic_chunks = semantic_splitter.split_text(document_text)

from langchain.text_splitter import (
    MarkdownHeaderTextSplitter,
    Language,
    RecursiveCharacterTextSplitter
)

# 마크다운 헤더 기반 분할
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "제목"),
        ("##", "섹션"),
        ("###", "하위섹션"),
    ]
)
md_chunks = md_splitter.split_text(markdown_text)

# 프로그래밍 언어 인식 분할
code_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100
)
code_chunks = code_splitter.split_text(python_code)

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    WebBaseLoader,
    DirectoryLoader
)
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 1. 다양한 소스에서 문서 로드
pdf_loader = PyPDFLoader("./docs/architecture.pdf")
web_loader = WebBaseLoader("https://docs.example.com/guide")
dir_loader = DirectoryLoader(
    "./docs/", glob="**/*.md", loader_cls=TextLoader
)

documents = []
documents.extend(pdf_loader.load())
documents.extend(web_loader.load())
documents.extend(dir_loader.load())

print(f"로드된 문서 수: {len(documents)}")

# 2. 청킹
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ".", " "]
)
chunks = text_splitter.split_documents(documents)
print(f"생성된 청크 수: {len(chunks)}")

# 3. 벡터 DB에 인덱싱
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key="YOUR_OPENAI_API_KEY"
)

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="knowledge_base"
)
print("인덱싱 완료!")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 프롬프트 템플릿
prompt = ChatPromptTemplate.from_template("""
당신은 기술 문서 전문 어시스턴트입니다.
아래 제공된 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트에 없는 내용은 "해당 정보를 찾을 수 없습니다"라고 답하세요.

## 컨텍스트
{context}

## 질문
{question}

## 답변 (한국어로, 코드 예시 포함)
""")

# LLM
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    api_key="YOUR_OPENAI_API_KEY"
)

# 리트리버
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance (다양성 확보)
    search_kwargs={
        "k": 5,
        "fetch_k": 20,
        "lambda_mult": 0.7
    }
)

# 문서를 텍스트로 포맷팅하는 함수
def format_docs(docs):
    return "\n\n---\n\n".join(
        f"[출처: {doc.metadata.get('source', '알 수 없음')}]\n{doc.page_content}"
        for doc in docs
    )

# LCEL 체인 구성
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 실행
answer = rag_chain.invoke("Spring Boot에서 @Transactional의 전파 속성을 설명해주세요")
print(answer)

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains.history_aware_retriever import (
    create_history_aware_retriever
)
from langchain.chains.retrieval import create_retrieval_chain
from langchain.chains.combine_documents import (
    create_stuff_documents_chain
)

# 대화 히스토리를 반영한 질문 재작성 프롬프트
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "대화 기록을 참고하여, 사용자의 최신 질문을 "
               "독립적으로 이해 가능한 형태로 재작성하세요."),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

# 히스토리 인식 리트리버
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_prompt
)

# 답변 생성 프롬프트
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 기술 문서 전문 어시스턴트입니다. "
               "컨텍스트를 기반으로 정확하게 답변하세요.\n\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

# 체인 구성
question_answer_chain = create_stuff_documents_chain(llm, answer_prompt)
conversational_rag = create_retrieval_chain(
    history_aware_retriever, question_answer_chain
)

# 대화 실행
chat_history = []

# 첫 번째 질문
response1 = conversational_rag.invoke({
    "input": "Redis의 캐시 전략에 대해 알려줘",
    "chat_history": chat_history
})
chat_history.extend([
    HumanMessage(content="Redis의 캐시 전략에 대해 알려줘"),
    AIMessage(content=response1["answer"])
])

# 후속 질문 (이전 맥락 참조)
response2 = conversational_rag.invoke({
    "input": "그중에서 Write-Through는 어떤 경우에 적합해?",
    "chat_history": chat_history
})
print(response2["answer"])

few_shot_prompt = ChatPromptTemplate.from_template("""
당신은 기술 블로그 작성 어시스턴트입니다.
제공된 컨텍스트를 기반으로 질문에 답변하되,
아래 예시와 같은 형식으로 작성하세요.

## 답변 형식 예시

### 질문: Spring Bean의 스코프란?
### 답변:
Spring Bean 스코프는 빈의 생명주기를 결정합니다.

**주요 스코프:**
- `singleton`: 컨테이너당 하나 (기본값)
- `prototype`: 요청마다 새로 생성
- `request`: HTTP 요청마다 하나

```java
@Scope("prototype")
@Component
public class MyBean 

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

# 테스트 데이터셋
eval_dataset = {
    "question": [
        "Spring Boot에서 트랜잭션 전파란?",
        "Redis 캐시 전략의 종류는?"
    ],
    "answer": [generated_answer_1, generated_answer_2],
    "contexts": [retrieved_contexts_1, retrieved_contexts_2],
    "ground_truth": [
        "트랜잭션 전파는 기존 트랜잭션이 있을 때...",
        "Cache-Aside, Write-Through, Write-Behind..."
    ]
}

# RAGAS 평가
result = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,       # 답변이 컨텍스트에 충실한지
        answer_relevancy,   # 답변이 질문에 관련되는지
        context_precision,  # 검색된 컨텍스트의 정밀도
        context_recall      # 필요한 정보가 검색되었는지
    ]
)
print(result)

# 하이브리드 검색 예시 (벡터 + BM25)
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 키워드 검색
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 벡터 유사도 검색
vector_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# 앙상블 (가중 평균)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # 벡터 검색에 더 높은 가중치
)

results = ensemble_retriever.invoke("Spring 트랜잭션 관리")