Агентный 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, а вопросы общего характера начинаются с поиска в интернете.

В-четвертых, соединяют узлы рёбрами. Статические рёбра всегда следуют одному пути, тогда как условные рёбра выбирают следующий узел на основе решения большой языковой модели:

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 для еще более агентного поведения.