Agentic RAG w Pythonie – samokorekcyjny pipeline z LangGraph krok po kroku

Dlaczego klasyczny RAG to już za mało?

Jeśli budujesz aplikacje oparte na dużych modelach językowych, na pewno znasz wzorzec RAG – pobierz dokumenty, wstrzyknij je do promptu, wygeneruj odpowiedź. Prosty, elegancki i skuteczny. No, przynajmniej do momentu, gdy przestaje działać.

I tu jest sedno problemu. Klasyczny RAG to pipeline liniowy. Pobiera dokumenty i ślepo przekazuje je do modelu – niezależnie od tego, czy są trafne. Jeśli retriever zwróci śmieci, generator wyprodukuje halucynacje. I nikt tego nie sprawdzi.

W 2026 roku, gdy enterprise wdraża RAG na masową skalę (ponad 51% systemów AI w przedsiębiorstwach korzysta z RAG, według najnowszych raportów), taka fragilność jest po prostu nieakceptowalna.

Odpowiedzią jest Agentic RAG – architektura, w której LLM przestaje być pasywnym konsumentem kontekstu i staje się aktywnym agentem. Sam decyduje, co pobrać, jak zweryfikować wyniki i kiedy spróbować ponownie. To nie pipeline – to pętla z samokorektą.

W tym poradniku zbudujemy kompletny system Agentic RAG w Pythonie z użyciem LangGraph 1.0 i ChromaDB. Pokażę trzy kluczowe wzorce – Corrective RAG, Adaptive RAG i Self-Reflective RAG – i jak je złożyć w jedną, produkcyjną architekturę. Z działającym kodem, oczywiście.

Architektura Agentic RAG – jak działa pętla samokorekcyjna

Zanim napiszemy linijkę kodu, warto zrozumieć, czym Agentic RAG różni się od klasycznego podejścia. W standardowym RAG mamy trzy kroki:

1. Użytkownik zadaje pytanie
2. Retriever pobiera dokumenty z bazy wektorowej
3. Generator tworzy odpowiedź na podstawie pobranych dokumentów

Proste, prawda? Problem w tym, że to podejście nie ma żadnego mechanizmu kontroli jakości.

W Agentic RAG te trzy kroki rozrastają się do maszyny stanów z wieloma węzłami decyzyjnymi:

• Router – decyduje, czy pytanie wymaga wyszukiwania w bazie, przeszukania internetu, czy może LLM poradzi sobie sam
• Retriever – pobiera dokumenty z bazy wektorowej (albo wielu źródeł jednocześnie)
• Grader – ocenia trafność pobranych dokumentów. Jeśli nie są istotne, wymusza ponowne wyszukiwanie ze zmodyfikowanym zapytaniem
• Generator – tworzy odpowiedź na podstawie zweryfikowanego kontekstu
• Hallucination Checker – sprawdza, czy odpowiedź jest rzeczywiście wsparta dokumentami. Jeśli nie – pętla się powtarza

Ten mechanizm „Loop-on-Failure" to tak naprawdę serce całej architektury. System nie przechodzi dalej z błędnymi danymi – próbuje naprawić problem. Sam wielokrotnie widziałem, jak ta jedna cecha dramatycznie poprawia jakość odpowiedzi w porównaniu z naiwnym pipeline'em.

Trzy wzorce Agentic RAG, które warto znać

Corrective RAG – korekta na poziomie wyszukiwania

Corrective RAG ocenia pobrane dokumenty przed generowaniem odpowiedzi. Jeśli grader uzna, że nie są wystarczająco trafne, system przepisuje zapytanie (query rewriting) i ponawia wyszukiwanie. Może też przełączyć się na alternatywne źródło – np. wyszukiwarkę internetową.

Kiedy stosować: gdy baza wiedzy jest duża, a zapytania użytkowników bywają nieprecyzyjne lub wieloznaczne.

Adaptive RAG – inteligentny routing zapytań

Adaptive RAG dodaje warstwę routingu jeszcze przed wyszukiwaniem. Proste pytania mogą być kierowane bezpośrednio do LLM (bez retrievera w ogóle). Pytania wymagające aktualnych danych trafiają do wyszukiwarki. Pytania specjalistyczne idą do bazy wektorowej. Router analizuje intencję użytkownika i wybiera optymalną ścieżkę.

Kiedy stosować: gdy system obsługuje zróżnicowane typy zapytań – od small talk po głębokie pytania domenowe. Szczerze mówiąc, to chyba najczęstszy przypadek w produkcji.

Self-Reflective RAG – weryfikacja na poziomie odpowiedzi

Self-Reflective RAG idzie najdalej. Po wygenerowaniu odpowiedzi oddzielny krok weryfikuje, czy odpowiedź jest wiernie oparta na dokumentach źródłowych (faithfulness check) i czy faktycznie odpowiada na pytanie użytkownika. Jeśli weryfikacja nie przejdzie – cała pętla się powtarza.

Kiedy stosować: w aplikacjach, gdzie halucynacje są absolutnie niedopuszczalne – finanse, medycyna, prawo.

Przygotowanie środowiska

Będziemy pracować z poniższymi bibliotekami (wersje aktualne na luty 2026):

# Wymagania: Python >= 3.10
pip install langgraph==1.0.7 langchain-openai==0.3.8 langchain-community chromadb==1.5.1 langchain-text-splitters

Ustaw klucz API OpenAI (lub innego providera – dalej pokażę jak użyć Claude):

export OPENAI_API_KEY="twoj-klucz-api"

LangGraph 1.0 oferuje stabilne API z gwarancją braku breaking changes do wersji 2.0, więc to bezpieczny wybór dla projektów produkcyjnych. ChromaDB 1.5.1 dodaje obsługę metadanych tablicowych i operatorów $contains/$not_contains, co przydaje się przy filtracji wyników.

Krok 1 – Budowa bazy wektorowej z ChromaDB

Zaczniemy od zaindeksowania przykładowych dokumentów. W produkcji byłyby to pewnie pliki PDF, strony wiki czy dokumentacja API – tutaj użyjemy uproszczonych danych demonstracyjnych, żeby nie zaciemniać obrazu.

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# Przykładowe dokumenty (w produkcji: DocumentLoader)
documents = [
    Document(
        page_content="""LangGraph 1.0 to framework do budowania
        agentów AI ze stanami trwałymi. Obsługuje checkpointy,
        human-in-the-loop i dynamiczne wywoływanie narzędzi.
        Wymaga Pythona >= 3.9.""",
        metadata={"source": "langgraph-docs", "version": "1.0"}
    ),
    Document(
        page_content="""ChromaDB to open-source'owa baza wektorowa
        zoptymalizowana pod LLM. Wspiera wyszukiwanie hybrydowe
        (dense + BM25) oraz operatory metadanych tablicowych
        od wersji 1.5.""",
        metadata={"source": "chromadb-docs", "version": "1.5"}
    ),
    Document(
        page_content="""Agentic RAG łączy wyszukiwanie dokumentów
        z autonomicznym podejmowaniem decyzji przez agenta.
        Kluczowe elementy to: routing zapytań, ocena trafności
        dokumentów i weryfikacja halucynacji.""",
        metadata={"source": "rag-patterns", "version": "2026"}
    ),
]

# Dzielenie na fragmenty (chunking)
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
splits = text_splitter.split_documents(documents)

# Tworzenie bazy wektorowej
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name="agentic_rag_demo"
)

# Retriever z top-3 wynikami
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

Krok 2 – Definiowanie narzędzia retriever

LangGraph traktuje retriever jako narzędzie (tool), które agent może wywołać – albo nie – w zależności od potrzeby. I to jest właśnie fundament podejścia agentycznego: LLM sam decyduje, kiedy sięgnąć po zewnętrzne dane.

from langchain_core.tools import tool

@tool
def retrieve_documents(query: str) -> str:
    """Wyszukaj dokumenty w bazie wiedzy na podstawie zapytania."""
    docs = retriever.invoke(query)
    return "\n\n---\n\n".join(
        f"[Źródło: {d.metadata.get('source', '?')}]\n{d.page_content}"
        for d in docs
    )

Krok 3 – Budowa grafu Agentic RAG z LangGraph

Teraz tworzymy maszynę stanów, która łączy wszystkie trzy wzorce. To serce naszego systemu – i jednocześnie fragment, na którym warto poświęcić trochę więcej czasu, żeby dobrze zrozumieć przepływ.

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ---- Węzeł 1: Agent decydujący (Router + Generator) ----
def agent_node(state: MessagesState) -> dict:
    """Agent decyduje: wywołać retriever czy odpowiedzieć wprost."""
    system_prompt = """Jesteś asystentem AI z dostępem do bazy wiedzy.
    
    ZASADY:
    - Jeśli pytanie dotyczy konkretnej wiedzy technicznej,
      ZAWSZE użyj narzędzia retrieve_documents.
    - Jeśli pytanie to small talk lub ogólna rozmowa,
      odpowiedz bezpośrednio bez wyszukiwania.
    - Podawaj źródła informacji w odpowiedzi."""
    
    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.bind_tools([retrieve_documents]).invoke(messages)
    return {"messages": [response]}

# ---- Węzeł 2: Retriever (ToolNode) ----
retrieve_node = ToolNode([retrieve_documents])

# ---- Węzeł 3: Grader – ocena trafności dokumentów ----
def grade_documents(state: MessagesState) -> dict:
    """Ocenia, czy pobrane dokumenty są trafne dla zapytania."""
    messages = state["messages"]
    
    # Wyciągnij ostatni wynik retrievera i oryginalne pytanie
    tool_message = messages[-1]
    user_question = next(
        m.content for m in messages if isinstance(m, HumanMessage)
    )
    
    grading_prompt = f"""Oceń trafność poniższych dokumentów 
    w kontekście pytania użytkownika.
    
    PYTANIE: {user_question}
    
    DOKUMENTY:
    {tool_message.content}
    
    Odpowiedz JEDNYM słowem: 'relevant' lub 'irrelevant'.
    Dokumenty są 'relevant' jeśli zawierają informacje
    bezpośrednio odpowiadające na pytanie."""
    
    response = llm.invoke([HumanMessage(content=grading_prompt)])
    
    # Dodaj ocenę do stanu jako metadata
    return {
        "messages": [
            HumanMessage(
                content=f"[OCENA DOKUMENTÓW: {response.content.strip()}]"
            )
        ]
    }

# ---- Węzeł 4: Rewriter – przepisanie zapytania ----
def rewrite_query(state: MessagesState) -> dict:
    """Przepisuje zapytanie, jeśli dokumenty były nietrafne."""
    messages = state["messages"]
    original_question = next(
        m.content for m in messages if isinstance(m, HumanMessage)
        and not m.content.startswith("[")
    )
    
    rewrite_prompt = f"""Oryginalne pytanie użytkownika nie dało
    trafnych wyników wyszukiwania.
    
    ORYGINALNE PYTANIE: {original_question}
    
    Przepisz je tak, aby lepiej trafiało w bazę wiedzy.
    Użyj bardziej precyzyjnych terminów technicznych.
    Zwróć TYLKO przepisane pytanie, nic więcej."""
    
    response = llm.invoke([HumanMessage(content=rewrite_prompt)])
    return {"messages": [HumanMessage(content=response.content.strip())]}

# ---- Węzeł 5: Generator końcowej odpowiedzi ----
def generate_answer(state: MessagesState) -> dict:
    """Generuje końcową odpowiedź na podstawie dokumentów."""
    messages = state["messages"]
    
    # Zbierz kontekst z pobranych dokumentów
    docs_content = ""
    user_question = ""
    for m in messages:
        if hasattr(m, "content"):
            if isinstance(m, HumanMessage) and not m.content.startswith("["):
                user_question = m.content
            if hasattr(m, "tool_call_id") or (
                hasattr(m, "name") and m.name == "retrieve_documents"
            ):
                docs_content = m.content
    
    gen_prompt = f"""Na podstawie poniższych dokumentów odpowiedz
    na pytanie użytkownika. Bądź precyzyjny i podawaj źródła.
    Jeśli dokumenty nie zawierają odpowiedzi, powiedz to wprost.
    
    DOKUMENTY:
    {docs_content}
    
    PYTANIE: {user_question}"""
    
    response = llm.invoke([HumanMessage(content=gen_prompt)])
    return {"messages": [response]}

# ---- Logika routingu ----
def route_after_grading(
    state: MessagesState,
) -> Literal["generate_answer", "rewrite_query"]:
    """Kieruje przepływ na podstawie oceny dokumentów."""
    last_message = state["messages"][-1]
    if "relevant" in last_message.content.lower() \
       and "irrelevant" not in last_message.content.lower():
        return "generate_answer"
    return "rewrite_query"

# ---- Składanie grafu ----
workflow = StateGraph(MessagesState)

# Dodaj węzły
workflow.add_node("agent", agent_node)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("generate_answer", generate_answer)

# Krawędzie
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    tools_condition,  # Czy agent wywołał narzędzie?
    {"tools": "retrieve", END: END}
)
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grading,
    {
        "generate_answer": "generate_answer",
        "rewrite_query": "rewrite_query"
    }
)
workflow.add_edge("rewrite_query", "agent")  # Pętla!
workflow.add_edge("generate_answer", END)

# Kompilacja
app = workflow.compile()

Krok 4 – Uruchomienie i testowanie

Zobaczmy nasz system w akcji. Zadamy pytanie wymagające wyszukania w bazie wiedzy:

# Pytanie wymagające retrievera
result = app.invoke({
    "messages": [
        HumanMessage(content="Jakie wymagania ma LangGraph 1.0?")
    ]
})

# Wyświetl końcową odpowiedź
print(result["messages"][-1].content)

# Pytanie, które agent obsłuży bez wyszukiwania
result2 = app.invoke({
    "messages": [
        HumanMessage(content="Cześć, jak się masz?")
    ]
})
print(result2["messages"][-1].content)

Przy pierwszym pytaniu agent rozpozna, że potrzebuje kontekstu technicznego – wywoła retriever, grader oceni trafność dokumentów, a generator wyprodukuje odpowiedź ze źródłami. Przy drugim pytaniu agent odpowie bezpośrednio, pomijając wyszukiwanie. To Adaptive RAG w działaniu.

Warto przejść ten flow krok po kroku i sprawdzić, co dokładnie dzieje się w każdym węźle. Debugowanie grafów stanów bywa... ciekawe.

Krok 5 – Dodanie pamięci konwersacyjnej

Agentic RAG zyskuje kolejny wymiar, gdy dodamy pamięć między kolejnymi zapytaniami. Na szczęście LangGraph 1.0 ma wbudowane checkpointy, więc to kwestia kilku linijek:

from langgraph.checkpoint.memory import MemorySaver

# Kompilacja z pamięcią
memory = MemorySaver()
app_with_memory = workflow.compile(checkpointer=memory)

# Konfiguracja sesji
config = {"configurable": {"thread_id": "user-session-001"}}

# Pierwsze pytanie
app_with_memory.invoke(
    {"messages": [HumanMessage(content="Co to jest ChromaDB?")]},
    config=config
)

# Drugie pytanie – agent pamięta kontekst
result = app_with_memory.invoke(
    {"messages": [HumanMessage(content="Jakie ma nowe funkcje?")]},
    config=config
)
print(result["messages"][-1].content)

Dzięki MemorySaver agent pamięta, że wcześniej rozmawialiśmy o ChromaDB, i wie, że „nowe funkcje" odnoszą się właśnie do tej bazy. Całkiem sprytne.

Ważna uwaga: w produkcji zamiast MemorySaver (pamięć RAM – zniknie po restarcie) użyjesz PostgresSaver lub SqliteSaver do trwałego przechowywania stanu konwersacji.

Wskazówki produkcyjne

Zanim wdrożysz Agentic RAG na produkcję, jest kilka rzeczy, o których warto pamiętać. Uczenie się na cudzych błędach jest zdecydowanie przyjemniejsze niż na własnych.

Ogranicz liczbę iteracji pętli

Pętla samokorekcyjna bez limitu to prosta droga do nieskończonego zużycia tokenów (i rachunku, który wywoła lekki szok). Zawsze ustawiaj retry_count z maksymalną liczbą prób. 2-3 iteracje to rozsądny kompromis między jakością a kosztami.

Wyszukiwanie hybrydowe zamiast samego embeddingu

Łączenie wyszukiwania wektorowego z leksykalnym (BM25) konsekwentnie daje lepsze wyniki niż samo wyszukiwanie semantyczne. ChromaDB 1.5 wspiera BM25 i SPLADE natywnie. Dodanie rerankerka (np. Cohere Rerank lub BGE) jeszcze bardziej poprawia precyzję – warto przetestować.

Chunking ma znaczenie

Jakość chunkingu bezpośrednio determinuje jakość retrievera. Eksperymentuj z rozmiarem fragmentów (500-1500 znaków), overlapem (10-20%) i strategiami dzielenia – rekursywne, semantyczne, na podstawie nagłówków. Nie ma jednej idealnej konfiguracji, bo zależy od rodzaju danych.

Ewaluacja z RAGAS

Mierz jakość pipeline'u za pomocą frameworka RAGAS. Daje cztery kluczowe metryki: Context Precision (czy pobrane dokumenty są precyzyjne?), Context Recall (czy pobrano wszystkie istotne dokumenty?), Faithfulness (czy odpowiedź jest wierna dokumentom?) i Answer Relevancy (czy odpowiedź faktycznie odpowiada na pytanie?). Bez systematycznego mierzenia optymalizujesz na ślepo.

Observability i tracing

W produkcji każda iteracja pętli powinna być logowana – najlepiej w formacie JSONL z metadanymi: czas wykonania, liczba tokenów, ocena gradera, liczba retries. LangSmith (narzędzie od twórców LangChain) integruje się z LangGraph natywnie i oferuje pełny wgląd w przebieg wykonania grafów. Polecam skonfigurować go od pierwszego dnia.

Agentic RAG vs. klasyczny RAG – porównanie

Dla pełnej jasności, oto zestawienie kluczowych różnic. Ta tabelka może się przydać, gdy będziesz przekonywać zespół do migracji:

Cecha	Klasyczny RAG	Agentic RAG
Architektura	Pipeline liniowy	Graf stanów z pętlami
Obsługa błędów	Brak – ślepe przekazywanie	Samokorekta i retry
Routing zapytań	Wszystko trafia do retrievera	Inteligentny routing
Weryfikacja odpowiedzi	Brak	Hallucination checker
Koszt tokenów	Niski, przewidywalny	Wyższy, zmienny
Latencja	Niska (1 przejście)	Wyższa (pętle korekty)
Niezawodność	Niska przy trudnych pytaniach	Wysoka

Najczęściej zadawane pytania

Czym dokładnie Agentic RAG różni się od zwykłego RAG?

Klasyczny RAG to liniowy pipeline: pobierz dokumenty → wygeneruj odpowiedź. Koniec. Agentic RAG dodaje warstwę inteligencji – agent sam decyduje, czy w ogóle szukać, ocenia trafność wyników, przepisuje zapytania przy słabych wynikach i weryfikuje odpowiedzi pod kątem halucynacji. To pętla z samokorektą zamiast jednorazowego przebiegu.

Czy Agentic RAG wymaga GPT-4 lub równie dużego modelu?

Główny LLM (agent decyzyjny i generator) powinien być na poziomie GPT-4o lub Claude Sonnet 4, bo musi podejmować złożone decyzje o routingu i oceniać jakość dokumentów. Natomiast grader i rewriter mogą spokojnie działać na mniejszych, tańszych modelach (np. GPT-4o-mini). W praktyce wiele zespołów stosuje właśnie taki podział – mocny model do generowania, lekki do oceny i routingu.

Jak kontrolować koszty pętli samokorekcyjnej?

Trzy kluczowe strategie: (1) ustaw twardy limit iteracji – 2-3 retries to sensowne maksimum, (2) użyj tańszych modeli do oceny i przepisywania zapytań, (3) monitoruj metryki per-query (liczba iteracji, tokeny, czas) i optymalizuj chunking oraz prompty oceniające, żeby zmniejszyć liczbę niepotrzebnych powtórzeń.

Czy mogę użyć Agentic RAG z modelem Claude zamiast GPT?

Oczywiście. LangGraph współpracuje z każdym modelem przez pakiet langchain-anthropic. Wystarczy zamienić ChatOpenAI na ChatAnthropic i ustawić odpowiedni klucz API. Claude Sonnet 4 i Claude Opus 4 sprawdzają się w tej roli świetnie, szczególnie w kontekście dłuższych rozumowań i oceny jakości dokumentów.

Kiedy wybrać klasyczny RAG zamiast Agentic RAG?

Klasyczny RAG wciąż ma sens w kilku przypadkach: (1) latencja jest krytyczna i nie możesz pozwolić sobie na dodatkowe iteracje, (2) pytania są proste i jednorodne (np. FAQ bot), (3) budżet na tokeny jest mocno ograniczony, (4) dokumenty w bazie są bardzo dobrze zorganizowane i retriever rzadko zwraca nietrafne wyniki. W takich sytuacjach dodatkowa złożoność Agentic RAG po prostu nie przyniesie wymiernych korzyści.