LangGraph — State Machine RAG

13-bob 12 kod bloki 2628 so'z

LangGraph nima?

LangGraph — bu Python kutubxonasi bo'lib, stateful (holatni saqlaydigan), ko'p bosqichli LLM ilovalarni graph (graf) shaklida qurishga mo'ljallangan. LangChain jamoasi tomonidan yaratilgan, lekin LangChain dan tubdan farq qiladi.

Oddiy RAG pipeline chiziqli (linear) ishlaydi:

matn1 qator

Savol → Embed → Search → Rerank → Generate → Javob

Bu oddiy holatlar uchun yetarli. Lekin real-world production da muammolar boshlanadi:

Retrieval sifati past bo'lsa nima qilamiz? Boshqa query bilan qayta urinishimiz kerak.
Javob hallucination bo'lsa nima qilamiz? Qayta generatsiya qilishimiz kerak.
Savol murakkab bo'lsa nima qilamiz? Uni qismlarga bo'lib, har birini alohida retrieval qilishimiz kerak.

Bu muammolarning hammasi conditional logic (shartli mantiq) talab qiladi. Chiziqli pipeline buni qila olmaydi. LangGraph aynan shu muammoni hal qiladi — RAG pipeline ni state machine (holat mashinasi) shaklida qurish imkonini beradi.

Nega LangGraph kerak?

Tasavvur qiling: foydalanuvchi "Nihol kompaniyasi 2020-2025 yillarda qanday bank loyihalarini amalga oshirgan?" deb so'raydi. Chiziqli pipeline:

Embed qiladi, search qiladi — faqat 2023-yilgi loyiha topildi
Generate qiladi — faqat bitta loyiha haqida javob beradi
Foydalanuvchi to'liq javob olmaydi

LangGraph bilan:

Embed qiladi, search qiladi — faqat 2023-yilgi loyiha topildi
Relevance check: "Bu yetarli emas, 2020-2025 so'ralgan"
Query rewrite: "Nihol bank loyihalari 2020" va "Nihol bank loyihalari 2025" deb qayta so'raydi
Yangi natijalar bilan generate qiladi
Hallucination check: Javob hujjatlarga asoslangan-mi tekshiradi
Agar hammasi yaxshi — javob beradi. Agar yo'q — qayta generatsiya qiladi.

Bu self-correcting (o'zini tuzatuvchi) pipeline. LangGraph uni qurishni oson va tushunarliroq qiladi.

Core Concepts — Asosiy tushunchalar

1. State — Holat

State — bu barcha node lar o'rtasida almashiladigan ma'lumot. Python da TypedDict sifatida aniqlanadi:

python8 qator

from typing import TypedDict, List

class RAGState(TypedDict):
    question: str           # foydalanuvchi savoli
    documents: List[str]    # topilgan hujjatlar
    generation: str         # LLM javobi
    relevance_score: float  # retrieval sifati (0.0 - 1.0)
    retry_count: int        # necha marta qayta urinildi

Har bir node state ni oladi, o'z ishini qiladi, va yangi qiymatlarni qaytaradi. State avtomatik yangilanadi. Bu immutable pattern ga mos keladi — node lar state ni to'g'ridan-to'g'ri o'zgartirmaydi, yangi qiymatlar qaytaradi.

2. Nodes — Tugunlar

Node — bu oddiy Python funksiya. State ni qabul qiladi, biror ish bajaradi, va yangilangan qiymatlarni qaytaradi:

python14 qator

def retrieve(state: RAGState) -> dict:
    """Savolga mos hujjatlarni bazadan qidiradi."""
    docs = search_pgvector(state["question"], top_k=5)
    return {"documents": docs}

def generate(state: RAGState) -> dict:
    """Topilgan hujjatlar asosida javob yaratadi."""
    prompt = f"""Quyidagi hujjatlar asosida savolga javob bering.

Hujjatlar: {state['documents']}
Savol: {state['question']}"""

    answer = llm.invoke(prompt)
    return {"generation": answer}

Har bir node bitta vazifani bajaradi. Kichik, fokusli funksiyalar — bu LangGraph ning kuchi.

3. Edges — Qirralar (bog'lanishlar)

Ikki tur edge mavjud:

Fixed Edge — har doim A dan B ga o'tadi:

python1 qator

graph.add_edge("retrieve", "check_relevance")  # har doim

Conditional Edge — shartga qarab yo'nalish tanlaydi:

python6 qator

def route_by_relevance(state: RAGState) -> str:
    if state["relevance_score"] > 0.5:
        return "generate"
    return "rewrite_query"

graph.add_conditional_edges("check_relevance", route_by_relevance)

Conditional edge — bu LangGraph ning eng kuchli qismi. Bu yerda "agar sifat past bo'lsa, boshqa yo'ldan bor" degan mantiq amalga oshiriladi.

4. Graph — Graf

Graph — bu node lar va edge larning yig'indisi. Compile qilinadi va runnable ga aylanadi:

python9 qator

from langgraph.graph import StateGraph, END

graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.set_entry_point("retrieve")
graph.add_edge("generate", END)

app = graph.compile()

LangGraph vs LangChain

Xususiyat	LangChain	LangGraph
Struktura	Chain (zanjir) — ketma-ket	Graph (graf) — tsikllar va shartlar bilan
Oqim	Chiziqli: A → B → C	Murakkab: A → B → (shart) → C yoki A
Holat	Har bir qadam o'z kontekstida	Umumiy state barcha node larda
Debugging	Qiyin (chain ichida)	Oson (har bir node alohida)
Use case	Oddiy pipeline lar	Self-correcting, multi-step pipeline lar

Qachon LangChain ishlatish kerak: - Oddiy Q&A chatbot - Bitta retrieval + bitta generation - Prototype va tezkor demo

Qachon LangGraph ishlatish kerak: - Retrieval sifatini tekshirish va qayta urinish kerak - Hallucination detection kerak - Murakkab savollarni qismlarga ajratish kerak - Production da quality gate lar kerak

To'liq RAG Pipeline — LangGraph bilan

Endi to'liq self-correcting RAG pipeline quramiz. Har bir funksiyani batafsil ko'ramiz:

python127 qator

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import json

# ---- State definition ----

class RAGState(TypedDict):
    question: str
    documents: List[str]
    generation: str
    relevance_score: float
    retry_count: int

# ---- Node functions ----

def retrieve(state: RAGState) -> dict:
    """PgVector dan savolga mos hujjatlarni qidiradi."""
    query_embedding = embed_model.encode(state["question"])
    docs = pgvector_search(query_embedding, top_k=5)
    return {
        "documents": [doc.page_content for doc in docs],
        "retry_count": state.get("retry_count", 0)
    }

def check_relevance(state: RAGState) -> dict:
    """Topilgan hujjatlar savolga qanchalik mos ekanini baholaydi."""
    prompt = f"""0 dan 1 gacha ball bering. 
Savol: {state['question']}
Hujjatlar: {state['documents'][:3]}

Faqat raqam qaytaring (masalan: 0.7)"""

    score = float(llm.invoke(prompt).strip())
    return {"relevance_score": score}

def route_by_relevance(state: RAGState) -> str:
    """Relevance score ga qarab yo'nalishni tanlaydi."""
    if state["relevance_score"] > 0.5:
        return "generate"
    if state.get("retry_count", 0) >= 2:
        return "generate"  # 2 marta urinib bo'lgan, bor narsada generate qil
    return "rewrite_query"

def rewrite_query(state: RAGState) -> dict:
    """Savolni qayta yozadi — yaxshiroq retrieval uchun."""
    prompt = f"""Quyidagi savol uchun yaxshiroq search query yozing.
Original savol: {state['question']}

Yangi query ni qaytaring (faqat query, boshqa narsa emas):"""

    new_query = llm.invoke(prompt).strip()
    return {
        "question": new_query,
        "retry_count": state.get("retry_count", 0) + 1
    }

def generate(state: RAGState) -> dict:
    """Hujjatlar asosida javob generatsiya qiladi."""
    context = "\n\n".join(state["documents"])
    prompt = f"""Quyidagi hujjatlar asosida savolga javob bering.
Faqat hujjatlardagi ma'lumotlarga tayanib javob bering.
Bilmagan narsani uylab topmang.

Kontekst:
{context}

Savol: {state['question']}

Javob:"""

    answer = llm.invoke(prompt)
    return {"generation": answer}

def check_hallucination(state: RAGState) -> str:
    """Javob hujjatlarga asoslangan-mi tekshiradi."""
    prompt = f"""Javob faqat hujjatlardagi ma'lumotlarga asoslangan-mi?

Hujjatlar: {state['documents'][:3]}
Javob: {state['generation']}

"grounded" yoki "hallucinated" deb javob bering:"""

    result = llm.invoke(prompt).strip().lower()
    if "grounded" in result:
        return "end"
    return "regenerate"

# ---- Graph qurilishi ----

graph = StateGraph(RAGState)

# Node larni qo'shish
graph.add_node("retrieve", retrieve)
graph.add_node("check_relevance", check_relevance)
graph.add_node("rewrite_query", rewrite_query)
graph.add_node("generate", generate)

# Entry point
graph.set_entry_point("retrieve")

# Fixed edge lar
graph.add_edge("retrieve", "check_relevance")
graph.add_edge("rewrite_query", "retrieve")  # loop — qayta qidirish!
graph.add_edge("generate", END)

# Conditional edge lar
graph.add_conditional_edges(
    "check_relevance",
    route_by_relevance,
    {
        "generate": "generate",
        "rewrite_query": "rewrite_query"
    }
)

# Compile va ishga tushirish
app = graph.compile()

result = app.invoke({
    "question": "Nihol Ucell uchun nima qilgan?",
    "documents": [],
    "generation": "",
    "relevance_score": 0.0,
    "retry_count": 0
})

print(result["generation"])

Hallucination check bilan to'liq versiya

Yuqoridagi kodga hallucination tekshiruvini qo'shamiz:

python28 qator

graph = StateGraph(RAGState)

graph.add_node("retrieve", retrieve)
graph.add_node("check_relevance", check_relevance)
graph.add_node("rewrite_query", rewrite_query)
graph.add_node("generate", generate)
graph.add_node("check_hallucination", check_hallucination_node)

graph.set_entry_point("retrieve")

graph.add_edge("retrieve", "check_relevance")
graph.add_edge("rewrite_query", "retrieve")

graph.add_conditional_edges(
    "check_relevance",
    route_by_relevance,
    {"generate": "generate", "rewrite_query": "rewrite_query"}
)

graph.add_conditional_edges(
    "check_hallucination",
    check_hallucination,
    {"end": END, "regenerate": "generate"}
)

graph.add_edge("generate", "check_hallucination")

app = graph.compile()

Graf diagrammasi

Pipeline ning to'liq oqimi:

matn38 qator

                    +-------------+
                    |   START     |
                    +------+------+
                           |
                           v
                    +------+------+
            +------>|  retrieve   |
            |       +------+------+
            |              |
            |              v
            |     +--------+--------+
            |     | check_relevance |
            |     +--------+--------+
            |              |
            |         (conditional)
            |        /           \
            |   score > 0.5    score <= 0.5
            |      /               \
            |     v                 v
            |  +--+-------+   +----+----------+
            |  | generate |   | rewrite_query |
            |  +--+-------+   +----+----------+
            |     |                 |
            |     v                 |
            |  +--+-----------+     |
            |  | check_halluc.|     |
            |  +--+-----------+     |
            |     |                 |
            |  (conditional)        |
            |   /         \         |
            | grounded  hallucinated|
            |  /            \       |
            | v              v      |
            |END         generate<--+
            |            (qayta)
            |                |
            +----------------+
              (loop: rewrite → retrieve)

Diqqat qiling: bu yerda ikki loop mavjud: 1. rewrite_query → retrieve — sifatsiz retrieval uchun qayta urinish 2. check_hallucination → generate — hallucination uchun qayta generatsiya

Bu chiziqli pipeline da mumkin emas. Faqat graph da mumkin.

Agentic RAG — LangGraph bilan

Yuqoridagi pipeline da oqim oldindan belgilangan (predetermined). Agentic RAG da esa agent o'zi qaror qabul qiladi: "Qidirish kerakmi? Qo'shimcha kontekst kerakmi? Savolni bo'lish kerakmi?"

python48 qator

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def retrieve_from_db(query: str) -> str:
    """Ichki bazadan ma'lumot qidiradi."""
    docs = pgvector_search(embed(query), top_k=5)
    return "\n".join([d.page_content for d in docs])

@tool
def web_search(query: str) -> str:
    """Internetdan ma'lumot qidiradi."""
    results = tavily_search(query, max_results=3)
    return "\n".join([r["content"] for r in results])

tools = [retrieve_from_db, web_search]

# Agent LLM — tool lar bilan ishlashni biladi
agent_llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

class AgentState(TypedDict):
    messages: list  # chat history

def agent_node(state: AgentState) -> dict:
    """Agent savol va kontekstga qarab tool tanlaydi."""
    response = agent_llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

# Agent graph
agent_graph = StateGraph(AgentState)
agent_graph.add_node("agent", agent_node)
agent_graph.add_node("tools", ToolNode(tools))

agent_graph.set_entry_point("agent")
agent_graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
agent_graph.add_edge("tools", "agent")  # tool natijasini agent ga qaytarish

agent_app = agent_graph.compile()

Bu agent o'zi hal qiladi: - Oddiy savol: to'g'ridan-to'g'ri javob beradi (tool ishlatmaydi) - Ma'lumot kerak: retrieve_from_db ni chaqiradi - Bazada topilmasa: web_search ga o'tadi - Murakkab savol: avval bazadan, keyin webdan qidiradi

Multi-step Reasoning — Ko'p bosqichli fikrlash

Murakkab savol: "Nihol kompaniyasining 2020-yildagi bank loyihalarini 2025-yildagi loyihalari bilan solishtiring."

Chiziqli pipeline bu savolni bitta search bilan hal qila olmaydi. LangGraph bilan uni qismlarga bo'lamiz:

python65 qator

class MultiStepState(TypedDict):
    original_question: str
    sub_questions: List[str]
    sub_answers: List[dict]
    final_answer: str

def decompose_question(state: MultiStepState) -> dict:
    """Murakkab savolni kichik savollarga bo'ladi."""
    prompt = f"""Quyidagi savolni 2-4 ta kichik savolga bo'ling.
JSON list qaytaring.

Savol: {state['original_question']}"""

    sub_qs = json.loads(llm.invoke(prompt))
    # Natija: ["Nihol bank loyihalari 2020", "Nihol bank loyihalari 2025"]
    return {"sub_questions": sub_qs}

def retrieve_for_sub_questions(state: MultiStepState) -> dict:
    """Har bir kichik savol uchun alohida retrieval."""
    results = []
    for sub_q in state["sub_questions"]:
        docs = pgvector_search(embed(sub_q), top_k=3)
        results.append({
            "question": sub_q,
            "documents": [d.page_content for d in docs]
        })
    return {"sub_answers": results}

def synthesize(state: MultiStepState) -> dict:
    """Barcha sub-answer larni birlashtirib yakuniy javob beradi."""
    context_parts = []
    for sa in state["sub_answers"]:
        context_parts.append(f"Savol: {sa['question']}\nHujjatlar: {sa['documents']}")

    full_context = "\n\n---\n\n".join(context_parts)

    prompt = f"""Quyidagi ma'lumotlar asosida original savolga 
to'liq javob bering. Taqqoslash va tahlil qiling.

{full_context}

Original savol: {state['original_question']}"""

    answer = llm.invoke(prompt)
    return {"final_answer": answer}

# Multi-step graph
ms_graph = StateGraph(MultiStepState)
ms_graph.add_node("decompose", decompose_question)
ms_graph.add_node("retrieve_all", retrieve_for_sub_questions)
ms_graph.add_node("synthesize", synthesize)

ms_graph.set_entry_point("decompose")
ms_graph.add_edge("decompose", "retrieve_all")
ms_graph.add_edge("retrieve_all", "synthesize")
ms_graph.add_edge("synthesize", END)

ms_app = ms_graph.compile()

result = ms_app.invoke({
    "original_question": "Nihol kompaniyasining 2020-yildagi bank loyihalarini 2025-yildagi loyihalari bilan solishtiring",
    "sub_questions": [],
    "sub_answers": [],
    "final_answer": ""
})

Oqim:

python16 qator

"Compare 2020 vs 2025 bank projects"
        |
        v
  [decompose_question]
        |
  ["bank projects 2020", "bank projects 2025"]
        |
        v
  [retrieve_for_sub_questions]
        |
  [{2020: docs...}, {2025: docs...}]
        |
        v
    [synthesize]
        |
  "2020-da Nihol X qilgan, 2025-da Y qilgan. Farqi..."

Qachon LangGraph ortiqcha (overkill)?

Har doim ham LangGraph kerak emas. Quyidagi hollarda oddiy chiziqli pipeline yetarli:

Oddiy Q&A chatbot: savol → search → javob. Loop yoki shart kerak emas.
Kam foydalanuvchi (< 100 kishi): murakkablik o'zini oqlamaydi.
Latency muhim: har bir qo'shimcha node = qo'shimcha vaqt. 3 ta LLM call o'rniga 6 ta bo'ladi.
Prototype bosqichi: avval oddiy pipeline ishga tushiring, keyin LangGraph ga o'ting.

Qachon LangGraph zarur?

Self-correcting RAG: retrieval sifatsiz bo'lsa qayta urinish kerak
Multi-source RAG: bazadan + webdan + hujjatlardan qidirish kerak
Murakkab reasoning: savolni qismlarga bo'lish, har birini alohida hal qilish kerak
Production quality gates: har bir bosqichda sifat tekshiruvi kerak
Agentic flows: LLM o'zi qaysi tool ni ishlatishni hal qilishi kerak
Human-in-the-loop: ba'zi qadamlarda inson tasdig'i kerak

Xulosa

LangGraph RAG pipeline ni oddiy "savol-javob" dan aqlli, o'zini tekshiruvchi tizim ga aylantiradi. Asosiy g'oya oddiy:

State — barcha ma'lumot bitta joyda
Nodes — har biri bitta vazifa bajaradi
Edges — fixed yoki conditional
Graph — compile qilinadi va ishga tushiriladi

Chiziqli pipeline 80% holatda yetarli. Lekin qolgan 20% — production, murakkab savollar, sifat kafolati — bu yerda LangGraph o'rnini hech narsa bosa olmaydi.

Keyingi bobda biz LangGraph ustiga quriladigan Advanced RAG Patterns — CRAG, Self-RAG, Adaptive RAG, RAPTOR, HyDE va boshqalarni ko'rib chiqamiz.

oldingi12. Hands-On Project — RAG Chatbot from Scratch keyingi14. Advanced RAG Patterns

Bu bobni bepul o'qidingiz.

Kitob o'rgatadi, lekin sizning kodingiz nega ishlamaganini ayta olmaydi. Kursda uy vazifangiz tekshiriladi.

Telegramda yozish Narxlar