Полное руководство: как собрать Agentic RAG систему полностью локально — без API и облаков

Почему локальная Agentic RAG — это будущее AI-разработки

В мире, где каждый запрос к GPT-4 стоит денег, а конфиденциальность данных становится приоритетом номер один, локальные AI-системы перестали быть нишевым решением. Agentic RAG (Retrieval-Augmented Generation) — это следующая ступень эволюции: не просто поиск и генерация, а интеллектуальный агент, который сам решает, когда искать информацию, когда использовать инструменты, и как планировать выполнение сложных задач.

Если вам интересна архитектура современных AI-агентов, рекомендую прочитать нашу статью «Как спроектировать современного AI-агента: от planner/executor до stateful memory», где мы подробно разбираем архитектурные подходы.

Архитектура нашей локальной системы

Прежде чем погружаться в код, давайте посмотрим на общую архитектуру системы, которую мы будем строить:

1 Подготовка окружения и установка зависимостей

Начнем с создания чистого Python окружения. Я рекомендую использовать Python 3.10 или выше:

# Создаем виртуальное окружение
python -m venv agentic_rag_env
source agentic_rag_env/bin/activate  # Для Windows: agentic_rag_env\Scripts\activate

# Устанавливаем базовые зависимости
pip install --upgrade pip
pip install langgraph langchain langchain-community
pip install sentence-transformers qdrant-client
pip install unstructured[all] pypdf
pip install gradio fastapi uvicorn
pip install ollama

2 Запуск локальных сервисов: Ollama и Qdrant

Ollama — это самый простой способ запускать локальные LLM. Установите и запустите модель:

# Установка Ollama (Linux/Mac)
curl -fsSL https://ollama.com/install.sh | sh

# Запускаем сервер Ollama
ollama serve &

# Скачиваем и запускаем модель (рекомендую Llama 3.2 3B или Mistral 7B)
ollama pull llama3.2:3b
# или
ollama pull mistral:7b

Теперь запустим Qdrant — векторную базу данных, которая будет хранить наши эмбеддинги:

# Способ 1: Через Docker (рекомендуется)
docker run -p 6333:6333 -p 6334:6334 \
    -v $(pwd)/qdrant_storage:/qdrant/storage:z \
    qdrant/qdrant

# Способ 2: Через Python клиент (полностью в памяти)
pip install qdrant-client[http]
# В коде создадим in-memory коллекцию

3 Создание ядра Agentic RAG системы

Теперь напишем основной код нашего агента. Мы создадим агента с ReAct архитектурой (Reasoning + Acting), который умеет:

1. Анализировать запрос пользователя
2. Решать, нужен ли поиск в базе знаний
3. Использовать инструменты (калькулятор, поиск в интернете, etc.)
4. Планировать выполнение сложных многошаговых задач
5. Запоминать контекст разговора

from typing import TypedDict, List, Optional, Annotated
import operator
from langchain_community.llms import Ollama
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_react_agent
import numpy as np

# Определяем состояние агента
class AgentState(TypedDict):
    messages: Annotated[List, add_messages]  # История сообщений
    knowledge_base: Optional[str]  # Релевантные документы
    current_step: str  # Текущий шаг выполнения
    needs_search: bool  # Нужен ли поиск в RAG
    final_answer: Optional[str]  # Финальный ответ

# Инициализация LLM через Ollama
llm = Ollama(
    model="llama3.2:3b",
    temperature=0.1,  # Низкая температура для консистентности
    num_predict=1024,  # Максимальная длина ответа
)

# Инициализация эмбеддингов
embeddings = OllamaEmbeddings(
    model="nomic-embed-text",  # Хорошие локальные эмбеддинги
)

# Создаем инструменты для агента
@tool
def search_knowledge_base(query: str) -> str:
    """Поиск информации в локальной базе знаний"""
    # Здесь будет подключение к Qdrant
    return "Найденная информация из базы знаний"

@tool
def calculate(expression: str) -> str:
    """Выполнение математических вычислений"""
    try:
        result = eval(expression)
        return f"Результат: {result}"
    except:
        return "Ошибка в выражении"

@tool
def web_search(query: str) -> str:
    """Поиск в интернете (если нужно)"""
    # Можно подключить локальный поиск через DuckDuckGo
    return "Результаты поиска из интернета"

# Создаем граф агента
def create_agent_graph():
    workflow = StateGraph(AgentState)
    
    # Узел: анализ запроса
    def analyze_query(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1].content if messages else ""
        
        # Простой анализ: проверяем, нужен ли поиск
        search_keywords = ["информация", "документ", "найди", "ищи", "база знаний"]
        needs_search = any(keyword in last_message.lower() for keyword in search_keywords)
        
        return {
            "needs_search": needs_search,
            "current_step": "analyzing_query"
        }
    
    # Узел: поиск в RAG
    def rag_search(state: AgentState):
        if not state["needs_search"]:
            return {"knowledge_base": None, "current_step": "generating_answer"}
        
        # Получаем последний запрос
        query = state["messages"][-1].content
        
        # Здесь должен быть реальный поиск в Qdrant
        # Пока заглушка
        results = ["Документ 1: Информация о...", "Документ 2: Данные по..."]
        
        return {
            "knowledge_base": "\n".join(results),
            "current_step": "generating_answer"
        }
    
    # Узел: генерация ответа
    def generate_answer(state: AgentState):
        messages = state["messages"]
        knowledge = state.get("knowledge_base", "")
        
        # Формируем промпт с контекстом
        prompt = f"""Ты — интеллектуальный ассистент. Используй следующую информацию если она релевантна:
        
        Контекст из базы знаний:
        {knowledge}
        
        История разговора:
        {messages[-5:] if len(messages) > 5 else messages}
        
        Текущий запрос: {messages[-1].content if messages else ''}
        
        Ответь максимально полезно и точно:"""
        
        response = llm.invoke(prompt)
        
        return {
            "final_answer": response,
            "current_step": "completed",
            "messages": messages + [AIMessage(content=response)]
        }
    
    # Добавляем узлы в граф
    workflow.add_node("analyze", analyze_query)
    workflow.add_node("search", rag_search)
    workflow.add_node("generate", generate_answer)
    
    # Определяем edges (переходы)
    workflow.set_entry_point("analyze")
    workflow.add_edge("analyze", "search")
    workflow.add_edge("search", "generate")
    workflow.add_edge("generate", END)
    
    return workflow.compile()

# Создаем и компилируем граф
agent_graph = create_agent_graph()

4 Настройка векторной базы знаний с Qdrant

Теперь создадим полноценную базу знаний. Сначала загрузим документы, затем создадим эмбеддинги и сохраним в Qdrant:

import os
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

class LocalKnowledgeBase:
    def __init__(self, collection_name="documents", persist_dir="./qdrant_data"):
        # Инициализируем клиент Qdrant
        self.client = QdrantClient(
            path=persist_dir,  # Локальное хранение
            prefer_grpc=True
        )
        
        self.collection_name = collection_name
        self.embeddings = embeddings
        
        # Создаем коллекцию если её нет
        self._create_collection()
    
    def _create_collection(self):
        try:
            self.client.get_collection(self.collection_name)
            print(f"Коллекция {self.collection_name} уже существует")
        except:
            # Создаем новую коллекцию
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=768,  # Размерность эмбеддингов nomic-embed-text
                    distance=Distance.COSINE
                )
            )
            print(f"Создана коллекция {self.collection_name}")
    
    def load_documents(self, directory_path: str):
        """Загрузка документов из директории"""
        loaders = {
            '.pdf': PyPDFLoader,
            '.txt': lambda path: DirectoryLoader(path, glob="**/*.txt"),
            '.docx': lambda path: DirectoryLoader(path, glob="**/*.docx"),
        }
        
        all_documents = []
        
        for ext, loader_class in loaders.items():
            for file_path in os.listdir(directory_path):
                if file_path.endswith(ext):
                    full_path = os.path.join(directory_path, file_path)
                    try:
                        loader = loader_class(full_path)
                        documents = loader.load()
                        all_documents.extend(documents)
                        print(f"Загружен {file_path}: {len(documents)} страниц")
                    except Exception as e:
                        print(f"Ошибка загрузки {file_path}: {e}")
        
        # Разбиваем на чанки
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )
        
        chunks = text_splitter.split_documents(all_documents)
        print(f"Всего чанков: {len(chunks)}")
        
        # Создаем эмбеддинги и сохраняем в Qdrant
        self._index_documents(chunks)
        
        return chunks
    
    def _index_documents(self, documents):
        """Индексация документов в Qdrant"""
        points = []
        
        for i, doc in enumerate(documents):
            # Создаем эмбеддинг для каждого чанка
            embedding = self.embeddings.embed_query(doc.page_content)
            
            point = PointStruct(
                id=i,
                vector=embedding,
                payload={
                    "text": doc.page_content,
                    "source": doc.metadata.get("source", "unknown"),
                    "page": doc.metadata.get("page", 0)
                }
            )
            points.append(point)
            
            # Пакетная загрузка каждые 100 точек
            if len(points) >= 100:
                self.client.upsert(
                    collection_name=self.collection_name,
                    points=points
                )
                points = []
                print(f"Индексировано {i+1} документов")
        
        # Загружаем оставшиеся
        if points:
            self.client.upsert(
                collection_name=self.collection_name,
                points=points
            )
        
        print(f"Индексация завершена. Всего документов: {len(documents)}")
    
    def search(self, query: str, top_k: int = 5):
        """Поиск в базе знаний"""
        # Создаем эмбеддинг запроса
        query_embedding = self.embeddings.embed_query(query)
        
        # Ищем в Qdrant
        search_result = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k
        )
        
        # Форматируем результаты
        results = []
        for hit in search_result:
            results.append({
                "text": hit.payload["text"],
                "score": hit.score,
                "source": hit.payload.get("source", "unknown")
            })
        
        return results

# Инициализируем базу знаний
kb = LocalKnowledgeBase()

# Загружаем документы (если есть)
if os.path.exists("./documents"):
    kb.load_documents("./documents")

5 Создание веб-интерфейса с Gradio

Теперь создадим простой веб-интерфейс для взаимодействия с нашим агентом:

import gradio as gr
import asyncio
from typing import List

class AgenticRAGInterface:
    def __init__(self, agent_graph, knowledge_base):
        self.agent = agent_graph
        self.kb = knowledge_base
        self.conversation_history = []
    
    def process_query(self, query: str, use_rag: bool):
        """Обработка запроса пользователя"""
        
        # Добавляем сообщение в историю
        self.conversation_history.append({"role": "user", "content": query})
        
        # Если нужен RAG, ищем в базе знаний
        context = ""
        if use_rag:
            search_results = self.kb.search(query)
            if search_results:
                context = "\n".join([f"[{i+1}] {res['text'][:200]}..." 
                                    for i, res in enumerate(search_results)])
        
        # Подготавливаем состояние
        initial_state = {
            "messages": [HumanMessage(content=query)],
            "knowledge_base": context,
            "needs_search": use_rag,
            "current_step": "start",
            "final_answer": None
        }
        
        # Запускаем агента
        try:
            result = self.agent.invoke(initial_state)
            answer = result.get("final_answer", "Не удалось получить ответ")
            
            # Добавляем ответ в историю
            self.conversation_history.append({"role": "assistant", "content": answer})
            
            # Форматируем историю для отображения
            history_text = self._format_history()
            
            return answer, history_text
            
        except Exception as e:
            error_msg = f"Ошибка: {str(e)}"
            return error_msg, self._format_history()
    
    def _format_history(self):
        """Форматирование истории разговора"""
        formatted = []
        for msg in self.conversation_history[-10:]:  # Последние 10 сообщений
            role = "👤 Пользователь" if msg["role"] == "user" else "🤖 Ассистент"
            formatted.append(f"{role}: {msg['content']}")
        return "\n\n".join(formatted)
    
    def clear_history(self):
        """Очистка истории"""
        self.conversation_history = []
        return "История очищена", ""

# Создаем интерфейс
interface = AgenticRAGInterface(agent_graph, kb)

# Создаем Gradio интерфейс
def create_gradio_interface():
    with gr.Blocks(title="Локальный Agentic RAG", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🤖 Локальный Agentic RAG Система
        Полностью автономный AI-агент с базой знаний. Работает без интернета!
        """)
        
        with gr.Row():
            with gr.Column(scale=2):
                query_input = gr.Textbox(
                    label="Ваш запрос",
                    placeholder="Задайте вопрос или дайте задание...",
                    lines=3
                )
                
                rag_toggle = gr.Checkbox(
                    label="Использовать базу знаний (RAG)",
                    value=True
                )
                
                submit_btn = gr.Button("Отправить", variant="primary")
                clear_btn = gr.Button("Очистить историю")
                
            with gr.Column(scale=3):
                answer_output = gr.Textbox(
                    label="Ответ агента",
                    lines=8,
                    interactive=False
                )
                
                history_output = gr.Textbox(
                    label="История разговора",
                    lines=12,
                    interactive=False
                )
        
        # Примеры запросов
        gr.Examples(
            examples=[
                ["Объясни концепцию machine learning простыми словами", True],
                ["Посчитай: (15 * 4) + (120 / 3)", False],
                ["Найди информацию о нейронных сетях в базе знаний", True],
                ["Спланируй изучение Python на месяц", False]
            ],
            inputs=[query_input, rag_toggle],
            label="Примеры запросов"
        )
        
        # Обработчики событий
        submit_btn.click(
            fn=interface.process_query,
            inputs=[query_input, rag_toggle],
            outputs=[answer_output, history_output]
        )
        
        clear_btn.click(
            fn=interface.clear_history,
            inputs=[],
            outputs=[answer_output, history_output]
        )
        
        query_input.submit(
            fn=interface.process_query,
            inputs=[query_input, rag_toggle],
            outputs=[answer_output, history_output]
        )
        
        return demo

# Запускаем интерфейс
if __name__ == "__main__":
    demo = create_gradio_interface()
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False  # Не создавать публичную ссылку
    )

Расширенные возможности и оптимизация

Добавление инструментов и навыков

Настоящая сила Agentic RAG проявляется, когда агент умеет использовать различные инструменты. Давайте расширим наш агента:

from datetime import datetime
import json

# Дополнительные инструменты для агента
@tool
def get_current_time() -> str:
    """Возвращает текущее время и дату"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def analyze_sentiment(text: str) -> dict:
    """Анализ тональности текста"""
    # Можно использовать локальную модель для анализа тональности
    prompt = f"Проанализируй тональность текста: '{text}'. Ответь одним словом: позитивный, негативный, нейтральный."
    
    response = llm.invoke(prompt)
    return {
        "text": text,
        "sentiment": response.strip(),
        "analysis_time": get_current_time()
    }

@tool
def summarize_text(text: str, max_length: int = 200) -> str:
    """Суммаризация текста"""
    prompt = f"Суммаризуй следующий текст в {max_length} символов:\n\n{text}"
    return llm.invoke(prompt)

# Инструмент для работы с файлами
@tool
def read_file(file_path: str) -> str:
    """Чтение текстового файла"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"Ошибка чтения файла: {str(e)}"

# Создаем агента с расширенным набором инструментов
tools = [
    search_knowledge_base,
    calculate,
    web_search,
    get_current_time,
    analyze_sentiment,
    summarize_text,
    read_file
]

# Можно создать специализированного агента для конкретной задачи
class ResearchAgent:
    def __init__(self):
        self.tools = [search_knowledge_base, web_search, summarize_text]
        
    def research_topic(self, topic: str, depth: str = "basic"):
        """Исследование темы с разной глубиной"""
        steps = []
        
        # Планирование исследования
        plan_prompt = f"""Спланируй исследование темы '{topic}'.
        Глубина: {depth}
        Верни план в формате JSON со списком шагов."""
        
        plan = llm.invoke(plan_prompt)
        steps.append(f"План исследования: {plan}")
        
        # Выполнение плана
        # Здесь можно реализовать автоматическое выполнение шагов
        
        return steps

Оптимизация производительности

Типичные ошибки и как их избежать

FAQ: Частые вопросы

Какое железо нужно для запуска?

Минимально: 8GB RAM, 4 ядра CPU. Рекомендуется: 16GB RAM, 8+ ядер CPU, видеокарта с 8GB+ VRAM (для более крупных моделей).

Можно ли использовать эту систему в production?

Да, но с некоторыми доработками: добавьте логирование, мониторинг, обработку ошибок и механизмы fallback. Для production-ready решений читайте нашу статью «Production-ready AI-агенты: как превратить хайп в работающую систему для бизнеса».

Как добавить мультимодальность (изображения, аудио)?

Используйте локальные мультимодальные модели (LLaVA, Bakllava) через Ollama. Для голосового интерфейса можно добавить Whisper для STT (распознавание речи) и Coqui TTS для синтеза. Подробнее в статье «Как собрать голосового ассистента на одной видеокарте».

Как обеспечить безопасность?

Локальный запуск уже обеспечивает конфиденциальность. Для защиты от prompt injection используйте валидацию входных данных и промпт-шейпинг. Подробнее в «Как защититься от атаки Man-in-the-Prompt».

Заключение

Локальные Agentic RAG системы — это не будущее, а настоящее. С текущим развитием open-source моделей и инструментов, каждый разработчик может собрать мощного AI-агента, который работает полностью автономно, не отправляет данные в облако и не требует ежемесячных платежей.

Ключевые преимущества нашей системы:

• Полная приватность: Все данные остаются на вашем компьютере
• Нулевая стоимость использования: После начальной настройки — никаких платежей
• Гибкость: Можно дообучать модели, добавлять свои инструменты, модифицировать архитектуру
• Производительность: Современные оптимизации позволяют запускать на consumer железе

Следующим шагом может быть добавление долговременной памяти, интеграция с внешними API или создание мультиагентной системы, где несколько агентов специализируются на разных задачах. Экспериментируйте, улучшайте и делитесь результатами!