11장: 프레임워크 선택 기준과 마이그레이션 전략

from llama_index.core import VectorStoreIndex
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
# --- LlamaIndex: 데이터 레이어 ---
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=5)
 
# --- LangGraph: 오케스트레이션 레이어 ---
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    context: str
    needs_research: bool
 
def research_node(state: AgentState) -> dict:
    """LlamaIndex를 사용한 연구 노드"""
    query = state["messages"][-1].content
    response = query_engine.query(query)
    return {
        "context": str(response),
        "needs_research": False,
    }
 
def respond_node(state: AgentState) -> dict:
    """컨텍스트를 활용한 응답 생성"""
    context = state.get("context", "")
    messages = state["messages"]
    response = model.invoke(
        f"컨텍스트: {context}\n\n"
        f"질문: {messages[-1].content}"
    )
    return {"messages": [response]}
 
def route(state: AgentState) -> str:
    if state.get("needs_research", True):
        return "research"
    return "respond"
 
# 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("respond", respond_node)
graph.add_conditional_edges(START, route)
graph.add_edge("research", "respond")
graph.add_edge("respond", END)
 
app = graph.compile(checkpointer=checkpointer)

from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
 
# Haystack: 고성능 검색 파이프라인
search_pipeline = Pipeline()
search_pipeline.add_component("retriever", bm25_retriever)
search_pipeline.add_component("reranker", reranker)
search_pipeline.connect("retriever", "reranker")
 
# LangGraph: 에이전트 오케스트레이션에서 Haystack 파이프라인 호출
def search_node(state: AgentState) -> dict:
    result = search_pipeline.run({
        "retriever": {"query": state["messages"][-1].content}
    })
    docs = result["reranker"]["documents"]
    context = "\n".join(d.content for d in docs[:3])
    return {"context": context}

# Before: LangChain LCEL
chain = (
    RunnableParallel(context=retriever, question=RunnablePassthrough())
    | prompt
    | model
    | parser
)
 
# After: LangGraph (동일 기능 + 에이전트 확장 가능)
from langgraph.graph import StateGraph, START, END
 
class RAGState(TypedDict):
    question: str
    context: str
    response: str
 
def retrieve(state: RAGState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"context": format_docs(docs)}
 
def generate(state: RAGState) -> dict:
    response = chain.invoke({
        "context": state["context"],
        "question": state["question"],
    })
    return {"response": response}
 
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
 
# 기존 LCEL 체인을 노드 내부에서 재사용

from abc import ABC, abstractmethod
 
class LLMServicePort(ABC):
    """프레임워크 독립적 LLM 서비스 인터페이스"""
 
    @abstractmethod
    async def generate(self, prompt: str, **kwargs) -> str:
        pass
 
    @abstractmethod
    async def generate_stream(self, prompt: str, **kwargs):
        pass
 
    @abstractmethod
    async def embed(self, text: str) -> list[float]:
        pass
 
class LangChainAdapter(LLMServicePort):
    """LangChain 어댑터"""
 
    def __init__(self, model, embeddings):
        self.model = model
        self.embeddings = embeddings
 
    async def generate(self, prompt: str, **kwargs) -> str:
        response = await self.model.ainvoke(prompt)
        return response.content
 
    async def generate_stream(self, prompt: str, **kwargs):
        async for chunk in self.model.astream(prompt):
            if chunk.content:
                yield chunk.content
 
    async def embed(self, text: str) -> list[float]:
        return await self.embeddings.aembed_query(text)
 
class HaystackAdapter(LLMServicePort):
    """Haystack 어댑터"""
 
    def __init__(self, generator, embedder):
        self.generator = generator
        self.embedder = embedder
 
    async def generate(self, prompt: str, **kwargs) -> str:
        result = self.generator.run(prompt=prompt)
        return result["replies"][0]
 
    async def generate_stream(self, prompt: str, **kwargs):
        # Haystack 스트리밍 구현
        pass
 
    async def embed(self, text: str) -> list[float]:
        result = self.embedder.run(text=text)
        return result["embedding"]

# 포트 (인터페이스)
class RetrievalPort(ABC):
    @abstractmethod
    async def retrieve(self, query: str, k: int = 5) -> list[Document]:
        pass
 
class GenerationPort(ABC):
    @abstractmethod
    async def generate(self, prompt: str, context: str) -> str:
        pass
 
# 비즈니스 로직 (프레임워크 독립)
class RAGService:
    def __init__(
        self,
        retriever: RetrievalPort,
        generator: GenerationPort,
    ):
        self.retriever = retriever
        self.generator = generator
 
    async def answer(self, question: str) -> str:
        docs = await self.retriever.retrieve(question)
        context = "\n".join(d.content for d in docs)
        return await self.generator.generate(
            prompt=question,
            context=context,
        )
 
# 어댑터 (프레임워크별 구현)
class LlamaIndexRetriever(RetrievalPort):
    def __init__(self, index):
        self.index = index
 
    async def retrieve(self, query: str, k: int = 5) -> list[Document]:
        nodes = await self.index.aretrieve(query)
        return [Document(content=n.text) for n in nodes[:k]]
 
class LangChainGenerator(GenerationPort):
    def __init__(self, chain):
        self.chain = chain
 
    async def generate(self, prompt: str, context: str) -> str:
        return await self.chain.ainvoke({
            "question": prompt,
            "context": context,
        })

from pydantic_settings import BaseSettings
 
class OrchestratorConfig(BaseSettings):
    framework: str = "langchain"  # langchain | langgraph | llamaindex | haystack
    model_provider: str = "openai"
    model_name: str = "gpt-4o"
    retriever_type: str = "vector"
    vector_store: str = "chroma"
 
    class Config:
        env_prefix = "AI_"
 
def create_service(config: OrchestratorConfig) -> RAGService:
    """설정에 따라 적절한 구현체 생성"""
    retriever = create_retriever(config)
    generator = create_generator(config)
    return RAGService(retriever=retriever, generator=generator)

from pydantic import BaseModel
from datetime import datetime
 
class Document(BaseModel):
    """표준 문서 모델"""
    content: str
    metadata: dict = {}
    score: float = 0.0
 
class ChatMessage(BaseModel):
    """표준 메시지 모델"""
    role: str  # "user" | "assistant" | "system"
    content: str
    timestamp: datetime = datetime.now()
    metadata: dict = {}
 
class GenerationResult(BaseModel):
    """표준 생성 결과 모델"""
    content: str
    model: str
    input_tokens: int = 0
    output_tokens: int = 0
    latency_ms: float = 0.0
    metadata: dict = {}

import pytest
 
class MockRetriever(RetrievalPort):
    async def retrieve(self, query: str, k: int = 5) -> list[Document]:
        return [Document(content="테스트 컨텍스트", score=0.9)]
 
class MockGenerator(GenerationPort):
    async def generate(self, prompt: str, context: str) -> str:
        return f"응답: {prompt} (컨텍스트 활용)"
 
@pytest.mark.asyncio
async def test_rag_service():
    """프레임워크 독립적 테스트"""
    service = RAGService(
        retriever=MockRetriever(),
        generator=MockGenerator(),
    )
    result = await service.answer("테스트 질문")
    assert "테스트 질문" in result

class GradualMigrationService:
    """기능 플래그 기반 점진적 마이그레이션"""
 
    def __init__(self, old_service, new_service, feature_flags):
        self.old = old_service
        self.new = new_service
        self.flags = feature_flags
 
    async def answer(self, question: str) -> str:
        if self.flags.is_enabled("use_new_rag"):
            try:
                return await self.new.answer(question)
            except Exception:
                # 새 서비스 실패 시 기존 서비스로 폴백
                return await self.old.answer(question)
        return await self.old.answer(question)