Ako vybudovať Agentic RAG pipeline: Praktický sprievodca s kódom [2026]

Úvod: Čo je Agentic RAG a prečo by vás to malo zaujímať

Retrieval-Augmented Generation (RAG) zmenil spôsob, akým pracujeme s veľkými jazykovými modelmi. Namiesto toho, aby sme sa spoliehali výlučne na znalosti „zamrazené" v parametroch modelu, RAG nám umožňuje dynamicky sťahovať relevantné informácie z externých zdrojov a využívať ich pri generovaní odpovedí. Znie to skvele, však?

No tradičný RAG má svoje limity. Je lineárny, pasívny a — povedzme si úprimne — nedokáže sa sám opraviť, keď niečo zlyhá.

Práve tu vstupuje do hry Agentic RAG — nová paradigma, ktorá kombinuje retrieval-augmented generation s agentným správaním. Namiesto jednoduchého pipeline typu „vyhľadaj a odpovedz" dostávame inteligentný systém, ktorý dokáže rozhodovať, plánovať, iterovať a korigovať sa. Agent tu nie je len pasívny vykonávateľ — je to autonómny systém schopný kriticky premýšľať nad vlastnými výstupmi.

V roku 2026 sa Agentic RAG stal de facto štandardom pre produkčné AI aplikácie. A nie je to náhoda. Dôvodov je hneď niekoľko:

• Spoľahlivosť: Samokorekčné mechanizmy dramaticky znižujú halucinácie
• Flexibilita: Agent dokáže dynamicky voliť stratégie vyhľadávania podľa typu otázky
• Škálovateľnosť: Modulárna architektúra umožňuje jednoduché rozširovanie o nové zdroje a nástroje
• Presnosť: Viacstupňové overovanie relevantnosti eliminuje nekvalitné zdroje
• Adaptabilita: Systém sa prispôsobí rôznym doménam bez nutnosti pretrénovania

V tomto sprievodcovi si krok po kroku ukážeme, ako vybudovať kompletný Agentic RAG pipeline pomocou LangGraph, LangChain a moderných vektorových databáz. Prejdeme od teórie cez architektúru až po produkčne pripravený kód. Tak poďme na to.

Tradičný RAG vs. Agentic RAG: Kľúčové rozdiely

Aby sme pochopili, prečo je Agentic RAG taký revolučný, musíme sa najprv pozrieť na obmedzenia tradičného prístupu.

Tradičný RAG pipeline

Klasický RAG funguje v troch jednoduchých krokoch: (1) užívateľ položí otázku, (2) systém vyhľadá relevantné dokumenty vo vektorovej databáze, (3) LLM vygeneruje odpoveď na základe nájdených dokumentov. Tento proces je jednosmerný a lineárny — žiadna spätná väzba, žiadny korekčný mechanizmus.

A tu začínajú problémy:

• Ak retriever vráti nerelevantné dokumenty, model ich aj tak veselo použije
• Neexistuje mechanizmus na detekciu halucinácií vo vygenerovanej odpovedi
• Systém nedokáže preformulovať otázku, keď prvé vyhľadávanie zlyhá
• Každá otázka sa spracováva rovnako — bez ohľadu na jej zložitosť
• Chýba schopnosť rozkladať komplexné otázky na menšie čiastkové problémy

Agentic RAG pipeline

Agentic RAG pridáva k tradičnému RAG vrstvu agentného rozhodovania. Systém obsahuje viaceré špecializované komponenty, ktoré spolupracujú v rámci stavového grafu. Čo všetko agent dokáže?

• Routovať otázky — rozhodnúť, či je potrebné vyhľadávanie, alebo dokáže odpovedať priamo
• Hodnotiť relevantnosť — posúdiť, či získané dokumenty skutočne zodpovedajú otázke
• Prepisovať otázky — ak prvé vyhľadávanie zlyhá, preformulovať dopyt a skúsiť znovu
• Kontrolovať halucinácie — overiť, že odpoveď je podložená zdrojmi
• Využívať nástroje — integrovať webové vyhľadávanie, kalkulačky, API a ďalšie externé služby
• Iterovať — opakovať celý cyklus, kým nedosiahne uspokojivý výsledok

Kľúčový rozdiel? Agentic RAG je cyklický a samoopravný. Využíva vzor ReAct (Reason + Act) — model najprv premýšľa o probléme, potom koná, pozoruje výsledok a znova premýšľa. Tento cyklus sa opakuje, kým agent nedosiahne uspokojivú odpoveď alebo nevyčerpá povolený počet iterácií.

Architektúra Agentic RAG systému

Jadro Agentic RAG systému tvoria vzájomne prepojené komponenty organizované ako stavový graf. Poďme si prejsť každý z nich podrobnejšie.

Router (Smerovač)

Router je vstupný bod celého systému. Analyzuje prichádzajúcu otázku a rozhoduje o ďalšom postupe. Môže nasmerovať dopyt do vektorovej databázy, na webové vyhľadávanie, alebo priamo na LLM (ak otázka nevyžaduje externé zdroje).

from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class RouteDecision(BaseModel):
    """Schema for routing decision."""
    datasource: str = Field(
        description="Route to 'vectorstore', 'web_search', or 'direct_answer'"
    )
    reasoning: str = Field(
        description="Brief explanation of routing decision"
    )

router_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert query router. Analyze the user question
    and decide the best data source:
    - 'vectorstore': for questions about internal documentation, company knowledge
    - 'web_search': for current events, recent information, real-time data
    - 'direct_answer': for general knowledge, greetings, simple calculations

    Return your decision with reasoning."""),
    ("human", "{question}")
])

router_chain = router_prompt | llm.with_structured_output(RouteDecision)

Retriever (Vyhľadávač)

Retriever zodpovedá za získavanie relevantných dokumentov z vektorovej databázy. V moderných Agentic RAG systémoch sa často využíva hybridné vyhľadávanie, ktoré kombinuje sémantické vektorové vyhľadávanie s lexikálnym BM25. Prečo? Pretože jeden prístup jednoducho nestačí na pokrytie všetkých typov dopytov.

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# Semantic vector retriever
embedding_model = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=embedding_model,
    persist_directory="./chroma_db"
)
vector_retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance
    search_kwargs={"k": 6, "fetch_k": 20}
)

# BM25 lexical retriever
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 6

# Hybrid ensemble retriever
hybrid_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # Favor semantic search slightly
)

Relevance Grader (Hodnotiteľ relevantnosti)

Toto je podľa mňa jedna z najdôležitejších častí celého Agentic RAG. LLM hodnotí každý získaný dokument a rozhoduje, či je skutočne relevantný pre danú otázku. Nerelevantné dokumenty sú odfiltrované ešte pred generovaním odpovede — čím sa výrazne znižuje riziko halucinácií.

class RelevanceGrade(BaseModel):
    """Binary relevance grade for a retrieved document."""
    is_relevant: bool = Field(
        description="Whether the document is relevant to the question"
    )
    confidence: float = Field(
        description="Confidence score between 0 and 1"
    )

grader_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a relevance grader. Given a user question and a
    retrieved document, determine if the document contains information
    relevant to answering the question.

    Focus on semantic relevance, not just keyword matching.
    A document is relevant if it provides useful context for answering
    the question, even if it doesn't contain the complete answer."""),
    ("human", "Question: {question}\n\nDocument: {document}")
])

relevance_chain = grader_prompt | llm.with_structured_output(RelevanceGrade)

Generator (Generátor)

Generátor je zodpovedný za syntézu finálnej odpovede z overených, relevantných dokumentov. Využíva štrukturovaný prompt, ktorý zabezpečuje, že odpoveď je podložená zdrojmi a neobsahuje vymyslené informácie.

generator_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a precise AI assistant. Generate a comprehensive
    answer to the question using ONLY the provided context documents.

    Rules:
    1. Only use information from the provided documents
    2. If documents don't contain sufficient information, say so clearly
    3. Cite specific documents when making claims
    4. Be thorough but concise
    5. Structure your response with clear paragraphs"""),
    ("human", "Question: {question}\n\nContext Documents:\n{documents}")
])

generator_chain = generator_prompt | llm | StrOutputParser()

Hallucination Checker (Kontrola halucinácií)

Posledný, ale mimoriadne dôležitý komponent. Hallucination Checker overuje vygenerovanú odpoveď oproti zdrojovým dokumentom. Ak zistí, že odpoveď obsahuje tvrdenia nepodložené zdrojmi, vráti systém späť na začiatok generovacieho cyklu. Je to taká poistka, bez ktorej by ste v produkcii nechceli byť.

class HallucinationCheck(BaseModel):
    """Check if generation is grounded in documents."""
    is_grounded: bool = Field(
        description="Whether the answer is fully supported by source documents"
    )
    problematic_claims: list[str] = Field(
        default_factory=list,
        description="List of claims not supported by sources"
    )

hallucination_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a hallucination detector. Compare the generated
    answer against the source documents. Determine if every claim in the
    answer is supported by the provided documents.

    Flag any claims that appear fabricated or unsupported."""),
    ("human", "Source Documents:\n{documents}\n\nGenerated Answer:\n{generation}")
])

hallucination_chain = hallucination_prompt | llm.with_structured_output(
    HallucinationCheck
)

Implementácia s LangGraph: Krok za krokom

LangGraph je framework na budovanie stavových, cyklických agentných grafov. Na rozdiel od jednoduchých reťazcov (chains) v LangChain, LangGraph umožňuje vytvárať komplexné workflow s podmienkovými vetvami, slučkami a zdieľaným stavom. Ak ste doteraz pracovali len s jednoduchými chainmi, pripravte sa — toto je úplne iná liga.

Definícia stavu grafu

Prvým krokom je definícia GraphState — typovaného slovníka, ktorý sleduje všetky relevantné informácie počas spracovania dopytu. Nezabudnite na počítadlá pokusov — sú kritické pre zabránenie nekonečných slučiek.

from typing import TypedDict, List, Optional
from langchain_core.documents import Document

class GraphState(TypedDict):
    """State object that flows through the Agentic RAG graph."""
    question: str                          # Original user question
    generation: Optional[str]              # Generated answer
    documents: List[Document]              # Retrieved documents
    filtered_documents: List[Document]     # Relevance-filtered documents
    query_rewrite_count: int               # Number of query rewrites
    generation_retry_count: int            # Number of generation retries
    max_retries: int                       # Maximum allowed retries
    route_decision: Optional[str]          # Router's decision
    is_relevant: bool                      # Whether answer is relevant
    is_grounded: bool                      # Whether answer is grounded
    current_query: str                     # Current (possibly rewritten) query
    web_search_results: List[Document]     # Results from web search
    metadata: dict                         # Additional metadata for logging

Implementácia uzlov grafu

Každý uzol grafu je funkcia, ktorá prijíma aktuálny stav a vracia aktualizovaný stav. Poďme implementovať všetky kľúčové uzly — a tu sa to začne robiť naozaj zaujímavé.

from langchain_community.tools.tavily_search import TavilySearchResults

web_search_tool = TavilySearchResults(max_results=3)

def route_question(state: GraphState) -> GraphState:
    """Route the question to the appropriate data source."""
    print("--- ROUTING QUESTION ---")
    question = state["question"]

    decision = router_chain.invoke({"question": question})
    print(f"Route decision: {decision.datasource} ({decision.reasoning})")

    return {
        **state,
        "route_decision": decision.datasource,
        "current_query": question
    }

def retrieve_documents(state: GraphState) -> GraphState:
    """Retrieve documents from vector store."""
    print("--- RETRIEVING DOCUMENTS ---")
    query = state["current_query"]

    documents = hybrid_retriever.invoke(query)
    print(f"Retrieved {len(documents)} documents")

    return {**state, "documents": documents}

def grade_documents(state: GraphState) -> GraphState:
    """Grade retrieved documents for relevance."""
    print("--- GRADING DOCUMENTS ---")
    question = state["current_query"]
    documents = state["documents"]

    filtered = []
    for doc in documents:
        grade = relevance_chain.invoke({
            "question": question,
            "document": doc.page_content
        })
        if grade.is_relevant and grade.confidence > 0.6:
            filtered.append(doc)

    print(f"Kept {len(filtered)} of {len(documents)} documents")
    return {**state, "filtered_documents": filtered}

def generate_answer(state: GraphState) -> GraphState:
    """Generate answer from filtered documents."""
    print("--- GENERATING ANSWER ---")
    question = state["current_query"]
    documents = state["filtered_documents"]

    docs_text = "\n\n---\n\n".join([
        f"Document {i+1}:\n{doc.page_content}"
        for i, doc in enumerate(documents)
    ])

    generation = generator_chain.invoke({
        "question": question,
        "documents": docs_text
    })

    return {**state, "generation": generation}

def check_hallucination(state: GraphState) -> GraphState:
    """Check if generation is grounded in source documents."""
    print("--- CHECKING HALLUCINATIONS ---")
    documents = state["filtered_documents"]
    generation = state["generation"]

    docs_text = "\n\n".join([doc.page_content for doc in documents])
    check = hallucination_chain.invoke({
        "documents": docs_text,
        "generation": generation
    })

    return {
        **state,
        "is_grounded": check.is_grounded,
        "generation_retry_count": state["generation_retry_count"] + (
            0 if check.is_grounded else 1
        )
    }

def rewrite_query(state: GraphState) -> GraphState:
    """Rewrite the query for better retrieval results."""
    print("--- REWRITING QUERY ---")
    question = state["current_query"]

    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a query rewriter. Rewrite the query to be
        more specific or use different terminology."""),
        ("human", "Original query: {question}\nRewritten query:")
    ])

    rewrite_chain = rewrite_prompt | llm | StrOutputParser()
    new_query = rewrite_chain.invoke({"question": question})

    return {
        **state,
        "current_query": new_query,
        "query_rewrite_count": state["query_rewrite_count"] + 1
    }

def web_search(state: GraphState) -> GraphState:
    """Perform web search as fallback."""
    print("--- WEB SEARCH ---")
    query = state["current_query"]

    results = web_search_tool.invoke({"query": query})
    web_docs = [
        Document(
            page_content=r["content"],
            metadata={"source": r["url"], "type": "web_search"}
        )
        for r in results
    ]

    return {
        **state,
        "documents": web_docs,
        "filtered_documents": web_docs,
        "web_search_results": web_docs
    }

Podmienené hrany a rozhodovacia logika

Srdcom Agentic RAG sú podmienené hrany — tie určujú tok riadenia na základe aktuálneho stavu. Každá takáto funkcia vracia meno nasledujúceho uzla, do ktorého sa má graf presunúť. Jednoduché, ale neuveriteľne mocné.

def decide_route(state: GraphState) -> str:
    """Decide where to route based on router decision."""
    route = state["route_decision"]
    if route == "web_search":
        return "web_search"
    elif route == "direct_answer":
        return "generate_direct"
    else:
        return "retrieve"

def decide_after_grading(state: GraphState) -> str:
    """Decide next step after document grading."""
    filtered = state["filtered_documents"]
    rewrite_count = state["query_rewrite_count"]
    max_retries = state["max_retries"]

    if len(filtered) == 0:
        if rewrite_count < max_retries:
            return "rewrite_query"
        else:
            return "web_search"
    else:
        return "generate"

def decide_after_hallucination_check(state: GraphState) -> str:
    """Decide next step after hallucination check."""
    is_grounded = state["is_grounded"]
    retry_count = state["generation_retry_count"]
    max_retries = state["max_retries"]

    if is_grounded:
        return "finish"
    elif retry_count < max_retries:
        return "regenerate"
    else:
        return "finish"

Zostavenie kompletného grafu

Teraz prichádza tá najzaujímavejšia časť — zostavíme všetky komponenty do kompletného LangGraph workflow.

from langgraph.graph import StateGraph, END

def build_agentic_rag_graph():
    """Build the complete Agentic RAG graph."""
    workflow = StateGraph(GraphState)

    # Add nodes
    workflow.add_node("route", route_question)
    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("grade_documents", grade_documents)
    workflow.add_node("generate", generate_answer)
    workflow.add_node("check_hallucination", check_hallucination)
    workflow.add_node("rewrite_query", rewrite_query)
    workflow.add_node("web_search", web_search)

    # Set entry point
    workflow.set_entry_point("route")

    # Add conditional edges from router
    workflow.add_conditional_edges(
        "route",
        decide_route,
        {
            "retrieve": "retrieve",
            "web_search": "web_search",
            "generate_direct": "generate"
        }
    )

    # Linear edge: retrieve -> grade
    workflow.add_edge("retrieve", "grade_documents")

    # Conditional edges after grading
    workflow.add_conditional_edges(
        "grade_documents",
        decide_after_grading,
        {
            "generate": "generate",
            "rewrite_query": "rewrite_query",
            "web_search": "web_search"
        }
    )

    # After query rewrite, go back to retrieve
    workflow.add_edge("rewrite_query", "retrieve")
    workflow.add_edge("web_search", "generate")
    workflow.add_edge("generate", "check_hallucination")

    # Conditional edges after hallucination check
    workflow.add_conditional_edges(
        "check_hallucination",
        decide_after_hallucination_check,
        {
            "finish": END,
            "regenerate": "generate"
        }
    )

    app = workflow.compile()
    return app

# Build and run
app = build_agentic_rag_graph()

result = app.invoke({
    "question": "Ako funguje attention mechanizmus v transformeroch?",
    "generation": None,
    "documents": [],
    "filtered_documents": [],
    "query_rewrite_count": 0,
    "generation_retry_count": 0,
    "max_retries": 3,
    "route_decision": None,
    "is_relevant": False,
    "is_grounded": False,
    "current_query": "",
    "web_search_results": [],
    "metadata": {}
})

print(result["generation"])

Kľúčové komponenty: Prepisovanie dopytov, samokorekcia a viacstupňové uvažovanie

Query Rewriting (Prepisovanie dopytov)

Prepisovanie dopytov je jedna z najúčinnejších techník v Agentic RAG. Keď prvotné vyhľadávanie nevráti nič zmysluplné, systém automaticky preformuluje otázku. A nejde len o jednoduché parafrázovanie — je to sofistikovaná transformácia, ktorá zohľadňuje kontext predchádzajúcich neúspešných pokusov.

Existuje niekoľko osvedčených stratégií:

• Dekompozícia: Rozloženie zložitej otázky na jednoduchšie čiastkové otázky
• Rozšírenie: Pridanie súvisiacich termínov a synoným pre širší zásah
• Špecializácia: Zúženie otázky na konkrétnejší aspekt problému
• Hypotetická odpoveď (HyDE): Vygenerovanie hypotetickej odpovede a jej použitie ako vyhľadávacieho dopytu — šikovný trik, ktorý prekvapivo dobre funguje

class QueryTransformer:
    """Advanced query transformation strategies."""

    def __init__(self, llm):
        self.llm = llm

    def decompose(self, question: str) -> list[str]:
        """Break complex question into sub-questions."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Decompose this complex question into 2-4 simpler
            sub-questions that, when answered together, would fully address
            the original question. Return as JSON list."""),
            ("human", "{question}")
        ])
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({"question": question})
        return json.loads(result)

    def hyde_transform(self, question: str) -> str:
        """Generate hypothetical document for better retrieval."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Write a short, factual paragraph that would be
            the ideal answer to this question. This will be used as a
            search query for semantic retrieval."""),
            ("human", "{question}")
        ])
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({"question": question})

    def step_back(self, question: str) -> str:
        """Generate a more general step-back question."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Given this specific question, generate a more
            general step-back question that addresses the broader concept."""),
            ("human", "{question}")
        ])
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({"question": question})

Samokorekčný mechanizmus

Samokorekcia je to, čo robí Agentic RAG naozaj výnimočným. Systém neakceptuje slepo svoju prvú odpoveď — namiesto toho ju kriticky zhodnotí a v prípade problémov sa pokúsi o opravu. Samozrejme, tento cyklus musí mať jasne definované limity, inak by sme skončili v nekonečnej slučke (a to nikto nechce).

class SelfCorrectionEngine:
    """Engine for self-correcting RAG outputs."""

    def __init__(self, llm, max_corrections: int = 3):
        self.llm = llm
        self.max_corrections = max_corrections

    def evaluate_and_correct(
        self,
        question: str,
        answer: str,
        sources: list[Document]
    ) -> dict:
        """Evaluate answer quality and correct if needed."""
        evaluation = self._evaluate(question, answer, sources)

        corrections = 0
        while not evaluation["passes"] and corrections < self.max_corrections:
            corrections += 1
            answer = self._correct(
                question, answer, sources, evaluation["issues"]
            )
            evaluation = self._evaluate(question, answer, sources)

        return {
            "final_answer": answer,
            "corrections_made": corrections,
            "quality_score": evaluation["score"],
            "is_reliable": evaluation["passes"]
        }

    def _evaluate(self, question, answer, sources) -> dict:
        """Multi-dimensional answer evaluation."""
        checks = {
            "groundedness": self._check_groundedness(answer, sources),
            "relevance": self._check_relevance(question, answer),
            "completeness": self._check_completeness(question, answer),
            "coherence": self._check_coherence(answer)
        }

        score = sum(checks.values()) / len(checks)
        passes = all(v > 0.7 for v in checks.values())
        issues = [k for k, v in checks.items() if v <= 0.7]

        return {"score": score, "passes": passes, "issues": issues}

Viacstupňové uvažovanie a integrácia nástrojov

Pre naozaj zložité otázky, ktoré vyžadujú viacstupňové uvažovanie, Agentic RAG využíva vzor ReAct. Agent striedavo premýšľa o probléme, vykonáva akcie (vyhľadávanie, výpočty, volanie API) a pozoruje výsledky, kým nedosiahne odpoveď. Je to ako sledovať skúseného analytika pri práci.

from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

@tool
def calculate(expression: str) -> str:
    """Safely evaluate mathematical expressions."""
    import ast
    result = eval(compile(ast.parse(expression, mode='eval'),
                         '', 'eval'))
    return str(result)

@tool
def search_database(query: str) -> str:
    """Search the internal knowledge base."""
    docs = hybrid_retriever.invoke(query)
    return "\n\n".join([d.page_content for d in docs[:3]])

tools = [calculate, search_database, web_search_tool]

react_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an AI assistant using ReAct reasoning.
    For each step:
    Thought: Analyze what you know and what you need
    Action: Choose and use a tool
    Observation: Review the result
    Continue until you can provide a final answer."""),
    ("human", "{question}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, react_prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

Vektorové databázy a stratégie embeddingov

Výber správnej vektorovej databázy a embedding stratégie je kritický pre výkonnosť celého systému. Úprimne, toto je oblasť, kde sa veľa projektov zasekne. V roku 2026 máme k dispozícii viacero vyspelých riešení, každé so svojimi silnými stránkami.

Porovnanie vektorových databáz

Pinecone je plne spravovaná cloudová služba, ideálna pre produkčné nasadenia. Ponúka serverless architektúru, automatické škálovanie a výbornú latenciu. Hodí sa pre tímy, ktoré nechcú tráviť čas správou infraštruktúry.

Chroma je open-source riešenie, perfektné pre prototypovanie a menšie projekty. Beží lokálne, je jednoduché na nastavenie a má výbornú integráciu s LangChain. Pre produkčné nasadenia však môže narážať na limity pri škálovaní.

Weaviate ponúka natívnu podporu pre hybridné vyhľadávanie (vektorové + kľúčové slová) a GraphQL API. Robustná voľba pre produkčné systémy s potrebou flexibilných dopytov.

Qdrant sa vyznačuje výborným výkonom a pokročilými filtračnými schopnosťami. Podporuje payload indexovanie, čo umožňuje kombinovať vektorové vyhľadávanie s filtráciou podľa metadát.

# Pinecone setup
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore

pc = Pinecone(api_key="your-api-key")

if "rag-index" not in pc.list_indexes().names():
    pc.create_index(
        name="rag-index",
        dimension=3072,  # text-embedding-3-large dimensions
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

pinecone_store = PineconeVectorStore(
    index_name="rag-index",
    embedding=OpenAIEmbeddings(model="text-embedding-3-large")
)

# Qdrant setup
from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient

qdrant_client = QdrantClient(url="http://localhost:6333")
qdrant_store = QdrantVectorStore(
    client=qdrant_client,
    collection_name="knowledge_base",
    embedding=OpenAIEmbeddings(model="text-embedding-3-large")
)

Stratégie embeddingov a chunkovanie

Kvalita embeddingov priamo ovplyvňuje kvalitu vyhľadávania. V roku 2026 sú najpoužívanejšie modely OpenAI text-embedding-3-large (3072 dimenzií, najvyššia presnosť) a Cohere embed-v4 (vynikajúci pomer výkon/cena). Pre špecifické domény môže byť vhodné siahnuť po fine-tunovaných embedding modeloch.

Rovnako dôležitá je stratégia chunkovania dokumentov. Príliš malé chunky strácajú kontext, príliš veľké znižujú presnosť vyhľadávania. Je to taký balanc, ktorý treba nájsť experimentovaním. Moderné prístupy kombinujú niekoľko techník.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker

# Strategy 1: Recursive character splitting with overlap
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len
)

# Strategy 2: Semantic chunking (groups by meaning)
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-large"),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90
)

# Strategy 3: Parent-child document strategy
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

store = InMemoryStore()
parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

Cross-Encoder reranking

Po počiatočnom vyhľadávaní je veľmi efektívne použiť cross-encoder reranking. Kým bi-encodery (embedding modely) kódujú otázku a dokument nezávisle, cross-encoder ich spracuje spoločne a dokáže presnejšie posúdiť relevantnosť. V praxi je to jednoduchá zmena, ktorá prinesie merateľný nárast kvality.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

def rerank_documents(
    query: str,
    documents: list[Document],
    top_k: int = 5
) -> list[Document]:
    """Rerank documents using cross-encoder for better precision."""
    pairs = [(query, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)

    scored_docs = list(zip(documents, scores))
    scored_docs.sort(key=lambda x: x[1], reverse=True)

    reranked = []
    for doc, score in scored_docs[:top_k]:
        doc.metadata["rerank_score"] = float(score)
        reranked.append(doc)

    return reranked

Produkčné nasadenie: Škálovanie, monitoring a spracovanie chýb

Prechod z prototypu do produkcie — to je téma, kde sa veľa projektov zasekne. A úprimne, nie je to jednoduché. Produkčný Agentic RAG systém vyžaduje starostlivú pozornosť niekoľkým kritickým oblastiam.

Produkčná architektúra vo vrstvách

Dobre navrhnutý produkčný systém sa skladá zo štyroch hlavných vrstiev:

1. Vrstva dátového spracovania (Data Ingestion Layer): Zodpovedá za načítavanie dokumentov, chunkovanie, generovanie embeddingov a ich indexovanie do vektorovej databázy. Beží ako asynchrónny pipeline, často s frontovým systémom (napr. Celery alebo RabbitMQ).
2. Vrstva AI výpočtov (AI Compute Layer): Spravuje volania LLM a embedding modelov. Zahŕňa správu rate limitov, caching, load balancing a fallback stratégie pre prípad výpadku poskytovateľa.
3. Vrstva agentného pipeline (Agentic AI Pipeline): Jadro systému — agentné uvažovanie, routovanie otázok, vylepšovanie dopytov a orchestrácia celého workflow. Tu beží náš LangGraph graf.
4. Vrstva nástrojov a sandboxu (Tools & Sandbox): Bezpečné prostredie pre vykonávanie výpočtov, volanie externých API a spúšťanie kódu. Izolácia pomocou kontajnerov zabraňuje bezpečnostným rizikám.

Komplexný monitoring a observabilita

V produkčnom prostredí je nevyhnutné monitorovať každý aspekt systému. Bez toho letíte naslepo. Tu je ukážka implementácie monitorovacej vrstvy.

import time
import logging
from dataclasses import dataclass, field
from functools import wraps

logger = logging.getLogger("agentic_rag")

@dataclass
class PipelineMetrics:
    """Track comprehensive pipeline metrics."""
    total_queries: int = 0
    successful_queries: int = 0
    failed_queries: int = 0
    total_latency_ms: float = 0
    retrieval_latencies: list = field(default_factory=list)
    generation_latencies: list = field(default_factory=list)
    query_rewrites: int = 0
    hallucinations_detected: int = 0

metrics = PipelineMetrics()

def track_latency(component_name: str):
    """Decorator to track component latency."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                latency = (time.time() - start) * 1000
                logger.info(
                    f"{component_name} completed",
                    extra={
                        "component": component_name,
                        "latency_ms": latency,
                        "status": "success"
                    }
                )
                if component_name == "retrieval":
                    metrics.retrieval_latencies.append(latency)
                elif component_name == "generation":
                    metrics.generation_latencies.append(latency)
                return result
            except Exception as e:
                latency = (time.time() - start) * 1000
                logger.error(f"{component_name} failed: {e}")
                raise
        return wrapper
    return decorator

Robustné spracovanie chýb

Produkčný systém musí elegantne zvládať celú škálu potenciálnych problémov — od výpadkov API po timeout-y a neočakávané odpovede modelov. Bez poriadneho error handlingu riskujete, že váš systém spadne práve vtedy, keď ho budete najviac potrebovať.

import asyncio
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

class ResilientRAGPipeline:
    """Production-grade RAG pipeline with error handling."""

    def __init__(self, graph, fallback_llm=None):
        self.graph = graph
        self.fallback_llm = fallback_llm

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        retry=retry_if_exception_type((TimeoutError, ConnectionError))
    )
    async def execute_with_retry(self, state: GraphState) -> GraphState:
        """Execute pipeline with automatic retry on transient failures."""
        try:
            result = await asyncio.wait_for(
                self._run_graph(state),
                timeout=60.0
            )
            return result
        except asyncio.TimeoutError:
            raise TimeoutError("Pipeline execution timed out")

    async def _run_graph(self, state: GraphState) -> GraphState:
        """Run the LangGraph with error boundaries."""
        try:
            return await self.graph.ainvoke(state)
        except Exception as e:
            logger.error(f"Graph execution failed: {e}")
            if self.fallback_llm:
                return await self._fallback_answer(state)
            raise

    async def _fallback_answer(self, state: GraphState) -> GraphState:
        """Generate fallback answer when pipeline fails."""
        fallback_response = await self.fallback_llm.ainvoke(
            f"Answer this question: {state['question']}"
        )
        return {
            **state,
            "generation": fallback_response.content,
            "metadata": {
                "fallback_used": True,
                "warning": "Answer generated without retrieval augmentation"
            }
        }

Caching pre optimalizáciu výkonu

Efektívny caching môže dramaticky znížiť náklady a latenciu, najmä pri opakujúcich sa alebo podobných dopytoch. A verte mi — v produkcii sa veľa dopytov opakuje častejšie, než by ste čakali.

import hashlib
from datetime import datetime, timedelta

class SemanticCache:
    """Cache with semantic similarity matching."""

    def __init__(self, vectorstore, similarity_threshold: float = 0.95):
        self.vectorstore = vectorstore
        self.threshold = similarity_threshold
        self.ttl = timedelta(hours=24)

    def get(self, question: str) -> dict | None:
        """Check cache for semantically similar questions."""
        results = self.vectorstore.similarity_search_with_score(
            question, k=1
        )
        if results:
            doc, score = results[0]
            if score >= self.threshold:
                cached_at = datetime.fromisoformat(
                    doc.metadata.get("cached_at", "2000-01-01")
                )
                if datetime.now() - cached_at < self.ttl:
                    return {
                        "answer": doc.metadata["answer"],
                        "cache_hit": True,
                        "similarity_score": score
                    }
        return None

    def put(self, question: str, answer: str):
        """Store question-answer pair in cache."""
        self.vectorstore.add_documents([
            Document(
                page_content=question,
                metadata={
                    "answer": answer,
                    "cached_at": datetime.now().isoformat(),
                    "question_hash": hashlib.md5(
                        question.encode()
                    ).hexdigest()
                }
            )
        ])

Najlepšie praktiky a časté úskalia

Na základe skúseností z reálnych nasadení som zhromaždil najdôležitejšie ponaučenia pre stavbu Agentic RAG systémov. Niektoré z nich som sa naučil na vlastnej koži.

Najlepšie praktiky

1. Vždy nastavte limity pre slučky. Samokorekčné mechanizmy sú mocné, ale bez limitov môžu viesť k nekonečným cyklom. Nastavte maximálny počet pokusov pre prepisovanie dopytov (typicky 3) aj pre regeneráciu odpovedí (2-3 stačia). V GraphState sledujte počítadlá pokusov a implementujte graceful degradation.

2. Používajte štruktúrované výstupy. Namiesto parsovania voľného textu z LLM používajte Pydantic modely a with_structured_output(). Eliminujete tým celú kategóriu parsingových chýb a systém sa stáva deterministickejším.

3. Implementujte hybridné vyhľadávanie. Samotné vektorové vyhľadávanie nestačí. Kombinácia sémantického vyhľadávania s BM25 a následným cross-encoder rerankingom konzistentne dosahuje lepšie výsledky ako ktorákoľvek jednotlivá metóda.

4. Testujte s reálnymi dopytmi. Vytvorte evaluačnú sadu s rôznymi typmi otázok — jednoduché faktické, analytické, porovnávacie, viacstupňové. Pre každý typ definujte očakávané správanie systému.

5. Monitorujte kvalitu v reálnom čase. Implementujte LangSmith alebo podobný nástroj na sledovanie každého behu pipeline. Sledujte metriky ako počet prepisov dopytov, pomer zachytených halucinácií a latenciu jednotlivých komponentov.

6. Optimalizujte chunkovanie pre vašu doménu. Univerzálne chunkovanie funguje akceptovateľne, ale doménovo špecifické nastavenie dokáže výrazne zlepšiť kvalitu vyhľadávania.

Časté úskalia, ktorým sa vyhnúť

1. Prílišná dôvera v LLM grading. LLM hodnotitelia nie sú neomylní. Ich rozhodnutia o relevantnosti bývajú nekonzistentné, najmä pri okrajových prípadoch. Riešenie? Kalibrujte prahy relevantnosti na vašich dátach a pravidelne auditujte rozhodnutia graderov.

2. Ignorovanie latency budget. Každá iterácia v samokorekčnej slučke pridáva jedno až dve volania LLM. Ak máte tri prepisovacie a tri generačné cykly, to je potenciálne 12+ volaní LLM pre jednu otázku. Nastavte celkový časový limit a v produkcii prioritizujte rýchlosť.

3. Nedostatočné metadáta v dokumentoch. Dokumenty bez kvalitných metadát (zdroj, dátum, autor, typ) obmedzujú schopnosť systému filtrovať a prioritizovať. Investujte čas do obohatenia metadát pri ingestion fáze — oplatí sa to.

4. Chýbajúce guardrails. Agentné systémy majú väčšiu autonómiu, čo so sebou prináša aj väčšie riziko. Implementujte vstupné aj výstupné filtre pre detekciu škodlivého obsahu, prompt injection a ďalších bezpečnostných hrozieb.

5. Monolitická architektúra. Snaha implementovať všetko v jednom monolitickom grafe sťažuje ladenie a údržbu. Rozdeľte systém na menšie, testovateľné podgrafy, ktoré môžete vyvíjať a ladiť nezávisle.

Výhľad do budúcnosti

Agentic RAG je rýchlo sa vyvíjajúca oblasť a v blízkej budúcnosti môžeme očakávať niekoľko zaujímavých posunov.

Multimodálny Agentic RAG

Dnešné systémy pracujú primárne s textom, ale budúcnosť patrí multimodálnemu RAG. Agenti budú schopní vyhľadávať a integrovať informácie z obrázkov, grafov, tabuliek, videí aj audio záznamov. Modely ako GPT-4o a Gemini už dnes zvládajú multimodálny vstup — výzvou zostáva efektívne vektorové indexovanie a vyhľadávanie naprieč modalitami.

Adaptívne agentné stratégie

Súčasné systémy majú fixné grafy a rozhodovaciu logiku. Budúce systémy budú mať agentov, ktorí sa dynamicky naučia optimálne stratégie na základe spätnej väzby. Napríklad agent zistí, že pre právne otázky je lepšie použiť prísnejšie filtre relevantnosti, zatiaľ čo pre kreatívne otázky treba nechať väčší priestor.

Kolaborácia viacerých agentov

Namiesto jedného monolitického agenta uvidíme systémy, kde viaceré špecializované agenty spolupracujú. Každý bude expert na konkrétnu doménu alebo úlohu — jeden na vyhľadávanie, druhý na analýzu, tretí na verifikáciu. Tieto multi-agentné systémy budú orchestrované nadriadeným koordinátorom.

Vylepšené embedding modely

Embedding modely sa rýchlo zlepšujú. Očakávame modely, ktoré lepšie zachytia nuansy jazyka, doménovo špecifické znalosti a vzťahy medzi konceptmi. Matryoshka embeddings s variabilnou dimenziou sa stanú štandardom a umožnia granulárnu kontrolu nad kompromisom medzi presnosťou a výpočtovými nákladmi.

Edge a on-premise nasadenia

S rastúcimi obavami o privátnosť dát a reguláciami (najmä v EÚ) bude narastať dopyt po Agentic RAG systémoch bežiacich úplne na vlastnej infraštruktúre. Menšie, ale schopné modely ako Llama, Mistral a ich nástupci umožnia plnohodnotné agentné pipeline bez potreby odosielať dáta externým poskytovateľom. Toto je pre mnohé firmy zásadná požiadavka.

Záver

Agentic RAG predstavuje zásadný evolučný krok oproti tradičnému retrieval-augmented generation. Kombináciou samokorekčných mechanizmov, inteligentného routovania, hodnotenia relevantnosti a kontroly halucinácií vzniká systém, ktorý je podstatne spoľahlivejší a flexibilnejší ako jeho predchodcovia.

V tomto sprievodcovi sme prešli od teórie cez architektúru až po produkčne pripravený kód. Tu sú kľúčové ponaučenia:

• Stavový graf (LangGraph) je ideálnou abstrakciou pre agentné RAG workflow
• Samokorekčné slučky dramaticky zlepšujú kvalitu, ale vyžadujú striktné limity iterácií
• Hybridné vyhľadávanie s cross-encoder rerankingom výrazne prekonáva čisté vektorové vyhľadávanie
• Štruktúrované výstupy cez Pydantic modely eliminujú celú kategóriu chýb
• Produkčné nasadenie vyžaduje viacvrstvovú architektúru s robustným monitoringom, cachingom a error handlingom
• Evaluácia a systematické testovanie sú nevyhnutné pre dlhodobú spoľahlivosť

Začnite s jednoduchým prototypom, iteratívne pridávajte komponenty a vždy merajte vplyv každej zmeny na kvalitu odpovedí. Agentic RAG nie je cieľ, ale cesta — cesta k AI systémom, ktoré sú nielen inteligentné, ale aj spoľahlivé a dôveryhodné.