Najprostszy RAG na LangGraph z własnym API | Sztuczna inteligencja w biznesie

Wprowadzenie

Finał serii. Z części pierwszej mamy teorię i mapę pojęć, z drugiej — przetestowane klocki: model przez OpenRouter, lokalne embeddingi, splitter, vector store w pamięci i graf retrieve → generate. Dziś składamy z nich to samo API, które w poprzedniej serii zbudowaliśmy na LlamaIndex: POST /index przyjmuje ścieżkę do pliku Markdown na dysku, POST /ask odpowiada na pytania o zaindeksowane dokumenty. Bez bazy wektorowej, bez persystencji, bez zaawansowanych strategii.

Przygotowanie środowiska

1
2
python -m venv .venv
.venv\Scripts\pip install fastapi uvicorn langchain langchain-openai langchain-huggingface langgraph

Klucz do OpenRouter — wyłącznie jako zmienna środowiskowa:

1
$env:OPENROUTER_API_KEY = "twój-klucz"

Kompletny kod — `main.py`

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import os
from pathlib import Path
from typing import TypedDict

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.graph import StateGraph, START, END

# --- klocki LangChain (część 2 serii) ---
llm = ChatOpenAI(
    api_key=os.environ["OPENROUTER_API_KEY"],
    base_url="https://openrouter.ai/api/v1",
    model="anthropic/claude-haiku-4.5",
)
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

# indeks w pamięci — startuje pusty, znika przy restarcie
vector_store = InMemoryVectorStore(embeddings)
zaindeksowane: list[str] = []


# --- graf LangGraph: retrieve -> generate ---
class State(TypedDict):
    question: str
    top_k: int
    context: list
    answer: str


def retrieve(state: State):
    znalezione = vector_store.similarity_search(state["question"], k=state["top_k"])
    return {"context": znalezione}


def generate(state: State):
    fragmenty = "\n\n".join(doc.page_content for doc in state["context"])
    prompt = (
        "Odpowiedz na pytanie wyłącznie na podstawie poniższych fragmentów. "
        "Jeśli fragmenty nie zawierają odpowiedzi, powiedz to wprost.\n\n"
        f"Fragmenty:\n{fragmenty}\n\nPytanie: {state['question']}"
    )
    odpowiedz = llm.invoke(prompt)
    return {"answer": odpowiedz.content}


graf = (
    StateGraph(State)
    .add_node("retrieve", retrieve)
    .add_node("generate", generate)
    .add_edge(START, "retrieve")
    .add_edge("retrieve", "generate")
    .add_edge("generate", END)
    .compile()
)


# --- API ---
app = FastAPI(title="Najprostszy RAG na LangGraph")


class IndexRequest(BaseModel):
    path: str          # ścieżka do pliku .md na lokalnym dysku


class AskRequest(BaseModel):
    question: str
    top_k: int = 3     # ile fragmentów trafi do modelu


@app.post("/index")
def index_file(req: IndexRequest):
    plik = Path(req.path)
    if not plik.is_file():
        raise HTTPException(404, f"Nie znaleziono pliku: {req.path}")
    if plik.suffix.lower() != ".md":
        raise HTTPException(400, "Obsługiwane są tylko pliki Markdown (.md)")

    tekst = plik.read_text(encoding="utf-8")
    chunki = splitter.split_text(tekst)
    vector_store.add_documents(
        [Document(page_content=c, metadata={"plik": plik.name}) for c in chunki]
    )
    zaindeksowane.append(plik.name)
    return {"status": "ok", "plik": plik.name, "chunkow": len(chunki)}


@app.get("/documents")
def list_documents():
    return {"dokumenty": zaindeksowane}


@app.post("/ask")
def ask(req: AskRequest):
    if not zaindeksowane:
        raise HTTPException(400, "Indeks jest pusty — najpierw dodaj dokument przez /index")

    wynik = graf.invoke({"question": req.question, "top_k": req.top_k})
    return {
        "odpowiedz": wynik["answer"],
        "zrodla": [
            {"plik": doc.metadata.get("plik"), "fragment": doc.page_content[:200]}
            for doc in wynik["context"]
        ],
    }

Uruchomienie:

1
.venv\Scripts\uvicorn main:app --port 8000

Jak to działa

Struktura jest identyczna jak w wersji na LlamaIndex — warto porównać oba pliki obok siebie:

/index — wczytuje plik, tnie na chunki splitterem (tu robimy to jawnie — w LlamaIndex działo się to pod maską), pakuje w Document z nazwą pliku w metadanych i dodaje do vector store.
/ask — nie woła vector store ani LLM bezpośrednio, tylko odpala graf: graf.invoke({...}). Węzeł retrieve znajduje fragmenty, generate buduje prompt i pyta model, a końcowy stan zawiera i odpowiedź, i użyty kontekst — dzięki temu do wyniku dokładamy źródła.
/documents — lista zaindeksowanych plików.

Największa różnica względem LlamaIndex jest architektoniczna: cała logika pytanie-odpowiedź żyje w grafie, a API tylko go wywołuje. Chcesz dodać krok sprawdzający jakość odpowiedzi albo przeformułowanie pytania? Dopisujesz węzeł i krawędź — endpointy nie zmieniają się ani o znak.

Testujemy

1
2
3
4
5
6
7
curl -X POST http://127.0.0.1:8000/index \
  -H "Content-Type: application/json" \
  -d '{"path": "C:/notatki/procedury.md"}'

curl -X POST http://127.0.0.1:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "Jak zresetować hasło użytkownika?"}'

Interaktywna dokumentacja jak zwykle pod http://127.0.0.1:8000/docs. Pierwsze indeksowanie może chwilę potrwać — model embeddingów pobiera się przy pierwszym użyciu.

LlamaIndex czy LangGraph?

Po dwóch seriach mamy dwa identyczne funkcjonalnie RAG-i, więc porównanie nasuwa się samo. LlamaIndex dał krótszy kod i szybszy start — indeksowanie i odpytywanie dostaliśmy niemal za darmo. LangChain + LangGraph wymagał złożenia klocków samodzielnie, ale w zamian każdy krok jest widoczny i podmienialny, a graf naturalnie rośnie w stronę bardziej złożonych przepływów agentowych. Do prostego „pytaj o dokumenty" wystarczy LlamaIndex; jeśli planujesz rozbudowę o wieloetapową logikę — graf od początku będzie dobrą inwestycją.

Podsumowanie

Seria zamknięta: teoria, klocki i ~100 linijek Pythona z grafem retrieve → generate opakowanym w API. Razem z poprzednią serią masz dwa kompletne, minimalne RAG-i do porównania i dalszych eksperymentów. A jeśli myślisz o podobnym rozwiązaniu na własnych dokumentach w firmie — odezwij się do nas.