Agentic RAG w LangGraph: oceniacze, warunkowe routowanie i alternatywa w postaci wyszukiwarki internetowej

Pipeline semantycznego RAG zbudowany z podstawowych komponentów LangChain, backendu FastAPI oraz interfejsu użytkownika w formie chatbotu Streamlit, obsługuje przesyłanie plików PDF, przekształca dokumenty w wektory reprezentacyjne i odpowiada na pytania za pośrednictwem interfejsu czatowego. Ma jednak ograniczenie: same elementy bazowe LangChain tworzą liniowe łańcuchy – prostą ścieżkę od promptu, przez model językowy, do parsera wyników. Wszystko, co wymaga warunkowych odgałęzień, pętli, dynamicznego kierowania lub decyzji podejmowanych przez model językowy, wykracza poza tę strukturę, a proste instrukcje if/else zapewniają jedynie powierzchowny, z góry określony przepływ.

LangGraph rozwiązuje ten problem. Wynikiem jest agentyjny workflow o następujących cechach:

• warunkowe odgałęzienia i pętle wbudowane bezpośrednio w pipeline
• model językowy, który sam decyduje między wyszukiwaniem w sieci a odzyskiwaniem informacji z RAG, bez żadnych sztywno zdefiniowanych reguł
• przejścia między etapami zarządzane za pomocą węzłów i krawędzi LangGraph

Systemy oceny oparte na LLM, które oceniają uzyskany kontekst oraz generowane odpowiedzi

Logika kierowania ruchem, która reaguje na te oceny

Stan przenoszony pomiędzy przejściami za pomocą StateGraph

Gotowy graf podłączany do backendu FastAPI w celu uruchomienia chatbota w warunkach produkcyjnych

Ostatecznie masz system agencyjny RAG, który odpowiada zarówno z magazynu wektorowego, jak i z sieci.

Pojęcia LangGraph

Słownictwo opiera się na czterech koncepcjach. StateGraph koordynuje wykonywanie zadań. Stan to skatalogowany słownik, który przechowuje i aktualizuje dane w procesie. Węzły to funkcje w Pythonie, które zmieniają stan i pełnią rolę jednostek obliczeniowych grafu. Krawędzie reprezentują przejścia pomiędzy węzłami, a warunkowe krawędzie umożliwiają zarówno rozgałęzianie, jak i pętle.

Konfiguracja

Umieść klucze API w pliku .env znajdującym się w korzeniu projektu:

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

Następnie zapisz je w klasie Pydantic BaseSettings:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Struktura projektu

Struktura folderów wygląda następująco:

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs
│   │── chains.py       # llm chains
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

Łańcuchy pobierania danych, usługa FastAPI oraz klient Streamlit pochodzą z wcześniejszego pipeline. Nowymi elementami są graf, jego klasa stanu oraz narzędzia oceniające, które sprawdzają istotność pobranych dokumentów, wykrywają halucynacje i określają, czy odpowiedź jest związana z pytaniem.

Łańcuchy LLM

Łańcuchy te znajdują się w pliku graphs/chains.py i wykorzystują język wyrażeń LangChain (LCEL). Zdefiniowano siedem z nich:

• retrieve_documents pobiera K najbardziej podobnych dokumentów z Pinecone
• generate_answer tworzy odpowiedź na pytanie wraz z kontekstem
• retrival_grader decyduje, czy pobrane dokumenty są istotne
• web_search przeprowadza wyszukiwanie w internecie i zwraca kontekst
• answer_grader sprawdza, czy odpowiedź odnosi się do pytania

hallucination_grader sprawdza, czy odpowiedź ma podstawy w kontekście

select_routes wysyła pytania dotyczące agentów, inżynierii promptów oraz ataków adwersarialnych do RAG, a wszystko inne – do wyszukiwarki internetowej

Testy jednostkowe dla łańcuchów

W systemach opartych na agentach testowanie jest ważniejsze niż zwykle, ponieważ wyniki generowane przez LLM są niedeterministyczne i mogą ulegać zmianom wraz z aktualizacjami modeli, dostosowaniami przez dostawców lub zmianami we embeddingach. Testy w pliku test_chains.py wykorzystują pytest w połączeniu z pytest.mark.asyncio dla łańcuchów asynchronicznych. Pozytywny test odzyskiwania danych z RAG wygląda następująco:

@pytest.mark.asyncio
async def test_rag_retrieval_yes() -> None:
    prompt = "what is prompt engineer ?"
    searched_documents = await retrieve_documents(prompt)
    searched_document = searched_documents[0]
    # logger.info(f"searched_content: {searched_document.page_content}")
    retrieval_grade = await retrieval_grader(searched_document, prompt)
    assert retrieval_grade

Zadaje pytanie dotyczące inżynierii promptów i przekazuje pierwszy odzyskany dokument do retrieval_grader. Ponieważ ten temat jest już omawiany w dokumentach znajdujących się w Pinecone, grader powinien zwrócić True. Test chroni przed awariami w przypadku zmian w embeddingach lub zachowaniu modelu, wcześnie wykrywa regresje po aktualizacjach modelu oraz weryfikuje logikę kierowania ruchu pomiędzy RAG a wyszukiwaniem w sieci. Pasujący test negatywny potwierdza, że grader zwraca False i proces kieruje się na wyszukiwanie w sieci.

Stan i węzły

Stan to skatalogowany słownik przechowywany w pliku graphs/state.py:

class GraphState(TypedDict):

    question: str
    generation: str
    web_search: bool
    documents: list[str]

Węzły odczytują i aktualizują ten stan. Zdefiniowane w pliku graphs/nodes.py, są to zwykłe funkcje w Pythonie, które odczytują stan, wywołują łańcuchy LLM i zwracają zaktualizowane pola. Węzeł używany w warunkowym krawędzi zwraca nazwę następnego węzła do wykonania. Istnieje ich siedem.

• generate wywołuje generate_answer i zwraca odpowiedź.
• retrieve wywołuje retrieve_documents i zwraca dokumenty.
• search_web wywołuje web_search i dodaje wyniki do dokumentów.
• grade_documents przechowuje istotne dokumenty i ustawia web_search=True, gdy jest to konieczne.
• grade_generation uruchamia narzędzia do oceny halucynacji i odpowiedzi oraz zwraca wartość „useful”, „not useful” lub „not supported”.

decide_to_generate wybiera między WEBSEARCH a GENERATE w zależności od wartości web_search.

route_question wysyła pytania związane z agentem do RAG, a pozostałe do wyszukiwarki internetowej.

Budowa StateGraph

Sam graf jest tworzony w pliku graphs/graph.py, a gotowy proces pracy przypomina rozgałęziony szlak pomiędzy pobieraniem informacji, oceną, generowaniem treści a wyszukiwaniem. Jego budowa wymaga kilku kroków.

Najpierw należy nadać każdemu węzłowi identyfikator w postaci ciągu znaków. Przechowywanie tych nazw jako stałych w pliku graphs/consts.py ułatwia czytanie grafu i zapobiega błędom pisowni:

RETRIEVE = "retrieve"
GRADE_DOCUMENTS = "grade_documents"
GENERATE = "generate"
WEBSEARCH = "websearch"

Następnie utwórz StateGraph i zarejestruj każdy węzeł wraz z jego funkcją:

workflow = StateGraph(GraphState)

workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, search_web)

Każde wezwanie add_node łączy nazwę z funkcją, a to rozdzielenie nazw od zachowań ułatwia utrzymanie, wizualizację i rozszerzanie procesu pracy.

Po trzecie, ustaw warunkowy punkt wejścia. Zamiast ręcznie kodować węzeł startowy, LangGraph pozwala funkcji przeanalizować pytanie i wybrać pierwszą ścieżkę. Tutaj route_question decyduje, czy rozpocząć od wyszukiwania w sieci, czy od pobierania danych za pomocą RAG:

workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    }
)

Ta funkcja jest wykonywana najpierw i zwraca albo "WEBSEARCH", albo "RETRIEVE" w zależności od tematu, a LangGraph wykorzystuje jej wartość zwrotną do wyboru węzła początkowego. Pytania związane z agentami trafiają do RAG, natomiast pytania o wiedzę ogólną rozpoczynają się od wyszukiwania w sieci.

Czwartym krokiem jest połączenie węzłów krawędziami. Krawędzie statyczne zawsze podążają jedną ścieżką, natomiast krawędzie warunkowe wybierają następny węzeł na podstawie decyzji LLM:

workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)

workflow.add_conditional_edges(
    GENERATE,
    grade_generation,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    }
)

workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)

Wspólnie te krawędzie tworzą dynamiczny przepływ pracy, w którym LLM wybiera początkową ścieżkę, istotność dokumentu decyduje o wyborze między RAG a wyszukiwaniem w sieci, a jakość odpowiedzi określa, czy należy ją udoskonalić, wrócić do wcześniejszego rozwiązania, czy zakończyć proces. Taka struktura przekształca zwykły pipeline RAG w bardziej solidny system agentowy.

Na koniec skompiluj graf i opcjonalnie narysuj go:

app = workflow.compile()

app.get_graph().draw_mermaid_png(output_file_path="graphs.png")

Dostarczanie go za pomocą FastAPI

W pliku dependencies.py funkcja importuje skompilowany graf i wywołuje go asynchronicznie:

from graphs.graph import app


async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await app.ainvoke(body.dict())

        return {
            "answer": generation.get("generation", ""),
            "web_search": generation.get("web_search", False)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI wstrzykuje to jako zależność i zwraca wygenerowaną odpowiedź wraz z flagą wskazującą, czy pochodzi ona z RAG, czy ze wyszukiwania w sieci.

Próba działania

Frontend Streamlit z wcześniejszego pipeline’u można wykorzystać bez zmian. Pytanie o to, jak zostać milionerem, kieruje na wyszukiwanie w sieci, natomiast pytanie o to, czym jest prompt engineering, kieruje na ścieżkę semantyczną RAG. Pokazuje to dynamiczne kierowanie ruchem wraz z jasnym źródłem każdej odpowiedzi.

Podsumowując, LangGraph StateGraph teraz zarządza stanem, workflow posiada dynamiczne przejścia z pętlami i gałęziami poprzez warunkowe krawędzie, a narzędzia do oceny LLM sprawdzają istotność dokumentów, wykrywanie halucynacji oraz jakość odpowiedzi. API informuje użytkownika, czy odpowiedź pochodzi z RAG, czy ze strony internetowej. Kolejnym krokiem w tym kierunku jest wykorzystanie narzędzi oraz frameworka ReAct, aby osiągnąć jeszcze bardziej złożone zachowania agentowe.

Literatura pokrewna

• Przepływy pracy LangGraph Multi-Agent: porównanie zespołu a nadzorcy — Podziel przepływ pracy LangGraph na podagenty z funkcją checkpointing, a następnie stwórz asystenta finansowego zarówno jako zespół, jak i nadzorcę, aby sprawdzić, w którym momencie każdy z tych modeli przestaje działać poprawnie.
• ReAct w LangGraph: od ręcznie budowanej pętli do ToolNode i StateGraph — Stwórz agenta ReAct dwa razy: najpierw jako ręczną pętlę asynchroniczną, a następnie przy użyciu ToolNode, tools_condition oraz StateGraph dostarczanych przez LangGraph do obsługi łańcuchowych wywołań narzędzi.

Claude tool_use End to End: Pętla agentowa w trzech iteracjach — Prześledź jeden żądanie Messages API przez pętlę narzędzi Claude: definicje narzędzi, wybór narzędzia, połączenia równoległe, kroki przed i po użyciu narzędzia, zasady wyników narzędzi oraz pełna pętla.

Równoległe agenty kodujące z git worktrees, Conductor i Antigravity 2.0 — Jak agent koordynujący, git worktrees dla poszczególnych agentów oraz ścieżki oparte na problemach umożliwiają stworzenie kilkunastu funkcji SRE w około pięciu godzinach, plus wskazówki dotyczące pytań HITL i higieny łączenia kodu.

Przepływy bankowe oparte na wielu agentach na Kubernetes: LangGraph, rejestr oraz AgentGateway — Twórz specjalistyczne agenty dla FastAPI, orkiestratora LangGraph z planerem opartym na LLM, rozwiązuj je w klastrze z rejestratem, a następnie dodaj routowanie i limity szybkości za pomocą AgentGateway.

Wywoływanie narzędzi w LangChain: mechanizm działania aplikacji agencyjnych — Rejestruj funkcje w Pythonie jako narzędzia LangChain i uruchamiaj czterostopniowy proces, w którym model wybiera narzędzia, te są wykonywane z zewnątrz, a wyniki wracają jako wiadomości.