Агентний RAG у LangGraph: оцінювачі, умовне маршрутизування та резервний пошук у Інтернеті

Семантичний підхід RAG, створений на основі ключових компонентів LangChain — бекенду FastAPI та фронтенду-чатбота Streamlit, — дозволяє обробляти завантажені PDF-файли, перетворювати документи на ембеддинги та відповідати на запитання через інтерфейс чату. Однак у нього є базова обмеженість: примітиви LangChain самі по собі створюють лінійні ланцюги обробки — пряму послідовність від запиту через ШІ до парсера результату. Усе, що потребує умовних гілок, циклів, динамічного маршрутизування чи рішень, які приймає ШІ, виходить за межі цієї структури, а прості оператори if/else забезпечують лише поверхневий, заздалегідь визначений алгоритм роботи.

LangGraph вирішує цю проблему. Результатом є агентський робочий процес із такими характеристиками:

• умовні гілки та цикли, вбудовані безпосередньо у підхід
• ШІ, яке самостійно обирає між пошуком у Інтернеті та отриманням інформації через RAG, без жорстко закодованих правил
• переходи між етапами, керовані за допомогою вузлів та зв’язків LangGraph

Системи оцінювання на основі LLM, які аналізують отриманий контекст та створені відповіді

Механізм маршрутизації, який реагує на ці оцінки

Стан, який передається під час переходів за допомогою StateGraph

Готова структура графа, підключена до бекенду FastAPI для функціонування чат-бота у промисловому режимі

У кінцевому підсумку у вас є агентна система RAG, яка може отримувати відповіді або з бази даних векторів, або з Інтернету.

Терміни LangGraph

Лексикон охоплює чотири основні концепції. StateGraph координує виконання операцій. Стан — це типовий словник, який зберігає та оновлює дані потоку обробки. Вузли — це функції Python, які змінюють стан та є одиницями обчислень у графі. Кромки — це переходи між вузлами, а умовні кромки дозволяють створювати гілки та цикли.

Конфігурація

Розмістіть ключі API у файлі .env у корені проекту:

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>

Потім введіть їх у клас 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")

Структура проекту

Структура папок виглядає так:

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

Ланцюги отримання даних, сервіс FastAPI та клієнт Streamlit успадковані з попередньої системи. Новими елементами є граф, його клас стану та оцінювачі, які перевіряють отримані документи на релевантність, виявляють галюцинації та визначають, чи є відповідь релевантною до запитання.

Ланцюги LLM

Ці ланцюги знаходяться у файлі graphs/chains.py та використовують мову виразів LangChain (LCEL). Визначено сім ланцюгів:

• retrieve_documents отримує K найбільш схожих документів з Pinecone
• generate_answer створює відповідь на запитання разом із контекстом
• retrival_grader вирішує, чи є отримані документи релевантними
• web_search виконує пошук в Інтернеті та повертає контекст
• answer_grader перевіряє, чи відповідь стосується запитання

hallucination_grader перевіряє, чи ґрунтується відповідь у контексті

select_routes надсилає запитання щодо агентів, інженерії запитів та адверсарських атак до RAG, а все інше — на пошук у Інтернеті

Юніт-тести для ланцюгів

У системах з агентами тестування має ще більше значення, ніж зазвичай, оскільки результати роботи LLM є недетермінованими та можуть змінюватися під час оновлення моделей, налаштувань виробників чи змін ембеддингів. Тести в test_chains.py використовують pytest з pytest.mark.asyncio для асинхронних ланцюгів. Позитивний тест для отримання даних з RAG виглядає так:

@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

У ньому ставиться запит щодо інженерії промптів, і перший отриманий документ передається до retrieval_grader. Оскільки ця тема вже розглядається у документах у Pinecone, оцінювач має повернути True. Цей тест запобігає проблемам у разі змін ембеддингів чи поведінки моделі, рано виявляє регресії після оновлень моделі та перевіряє логіку маршрутизації між RAG та пошуком у Інтернеті. Відповідний негативний тест підтверджує, що оцінювач повертає False, і процес продовжується з пошуком у Інтернеті.

Стан та вузли

Стан — це типізований словник, який зберігається у graphs/state.py:

class GraphState(TypedDict):

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

Вузли читають та оновлюють цей стан. Вони визначені у файлі graphs/nodes.py — це звичайні функції Python, які читають стан, викликають ланцюги LLM та повертають оновлені поля. Вузол, використаний у умовному ребрі, повертає назву наступного вузла для виконання. Їх є сім.

• generate викликає generate_answer та повертає відповідь.
• retrieve викликає retrieve_documents та повертає документи.
• search_web викликає web_search та додає результати до документів.
• grade_documents зберігає відповідні документи та, за потреби, встановлює web_search=True.
• grade_generation запускає механізми оцінки галюцинацій та відповідей та повертає значення „корисно“, „некорисно“ або „не підтримується“.

decide_to_generate вибирає між WEBSEARCH або GENERATE залежно від значення web_search.

route_question надсилає запитання, пов’язані з агентом, до RAG, а інші — на пошук у Інтернеті.

Створення StateGraph

Сам граф формується у файлі graphs/graph.py, а готовий робочий процес виглядає як розгалужена структура, що поєднує етапи отримання інформації, оцінки, генерації та пошуку. Його створення вимагає кількох кроків.

Спочатку кожному вузлу присвоюється рядковий ідентифікатор. Зберігання цих назв у вигляді констант у файлі graphs/consts.py полегшує читання графа та запобігає опечаткам:

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

Далі створіть StateGraph та зареєструйте кожну вузол із його функцією:

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)

Кожен виклик add_node прив’язує назву до функції, і саме це розділення назв та функцій полегшує підтримку, візуалізацію та розширення робочого процесу.

По-третє, встановіть умовну точку входу. Замість того, щоб жорстко задавати початкову вузол, LangGraph дозволяє функції проаналізувати запит та обрати перший шлях. Тут route_question вирішує, чи починати з пошуку в Інтернеті чи з отримання даних за допомогою RAG:

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

Ця функція виконується спочатку та повертає або "WEBSEARCH", або "RETRIEVE" залежно від теми, і LangGraph використовує її значення для вибору початкового вузла. Запитання, пов’язані з агентом, надсилаються до RAG, а запитання з загальними знаннями починаються з пошуку в Інтернеті.

Четверте – з’єднуйте вузли ребрами. Статичні ребра завжди слідують одному шляху, тоді як умовні ребра обирають наступний вузол на основі рішення 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)

Разом ці елементи створюють динамічний робочий процес, у якому ШІ обирає початковий шлях, релевантність документа визначає вибір між RAG та пошуком у Інтернеті, а якість відповіді вирішує, чи потрібно її удосконалити, повернутися до попереднього варіанту чи завершити процес. Така структура перетворює звичайний підхід RAG на більш міцну агентську систему.

Наостанок скомпілюйте граф та за бажанням намалюйте його:

app = workflow.compile()

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

Надання через FastAPI

У файлі dependencies.py функція імпортує скомпільовану структуру та викликає її асинхронно:

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 вводить його як залежність, і він повертає сформовану відповідь разом із прапорцем, який вказує, чи вона походить з RAG чи зі веб-пошуку.

Спробувати самостійно

Фронтенд Streamlit з попередньої системи можна використовувати без змін. Запит про те, як стати мільйонером, правильно спрямовує на веб-пошук, тоді як запит про те, що таке інжиніринг підказок, спрямовує на семантичний шлях RAG. Це демонструє динамічне спрямування разом із чітким походженням кожної відповіді.

Підсумовуючи, LangGraph StateGraph тепер керує станом, робочий процес має динамічні переходи з циклами та гілками через умовні краї, а інструменти оцінки LLM перевіряють релевантність документів, виявляють галюцинації та якість відповідей; API повідомляє користувачеві, чи відповідь походить з RAG чи з Інтернету. Наступним кроком у цій роботі є використання інструментів та фреймворку ReAct для ще більш агентного поводження.