Как собрать локального AI-ассистента для разработчика с гибридным RAG (векторный поиск + граф знаний) на Ollama

Зачем платить за GPT, если код не хочет делиться секретами?

Представьте: вы пытаетесь разобраться в legacy-коде, который писал ваш коллега пять лет назад. Каждый запрос к ChatGPT-5 (или что там сейчас актуально) стоит денег. И самое главное - вы не можете отправить туда куски продакшн-кода. Конфиденциальность? Нулевая. Задержки? Раздражают. Локальный ассистент - не прихоть, а необходимость. Но обычный RAG, который просто ищет похожие куски текста, часто тупит на сложных запросах вроде "как связаны эти два модуля" или "почему эта функция вызывает ту".

Проблема в том, что векторный поиск видит мир как мешок слов с семантикой. Он находит похожие по смыслу фрагменты, но теряет связи между ними. Граф знаний - другая крайность: он идеально хранит отношения (функция A вызывает функцию B), но для поиска по смыслу не годится. Гибридный RAG - это попытка заставить их работать вместе. Как? Сейчас разберем.

Архитектура: что склеивает векторы и графы

Главная идея проста: когда пользователь задает вопрос, мы параллельно запускаем два поиска. Векторный ищет по семантическому сходству в LanceDB. Графовый - по связям в NetworkX. Потом сливаем результаты, убираем дубли, сортируем по релевантности и подаем в LLM через Ollama. Звучит логично, но есть нюанс - как определять, что важнее? Мы используем взвешенное голосование: сложным запросам про "зависимости" увеличиваем вес графа, общим вопросам "как работает" - вес векторов.

Почему LanceDB, а не Chroma или Qdrant? На 2026 год LanceDB стабильно работает с большими датасетами прямо на диске, не требуя отдельного сервера. NetworkX - де-факто стандарт для графов в Python. Ollama - потому что она уже умеет работать с десятками моделей из коробки, включая новейшие на тот момент. Если вы хотите углубиться в архитектуру локальных агентов, посмотрите наше полное руководство по Agentic RAG.

1Подготовка поля боя: ставим Ollama и друзей

Первое, что бесит в туториалах - они предполагают, что у вас уже стоит все и настроено. Забудьте. Вот команды, которые реально работают на чистой системе (проверено на Ubuntu 24.04, но и на Mac с Apple Silicon тоже).

# Ставим Ollama - следите за актуальной командой на официальном сайте
curl -fsSL https://ollama.ai/install.sh | sh

# Качаем модель. В 2026 году Llama 3.2 90B - хороший баланс между умом и размером
# Но если у вас меньше 48 ГБ RAM, берите Qwen2.5 32B или Mistral 12B
ollama pull llama3.2:90b

# Ставим Python зависимости
pip install "fastapi[standard]" lancedb networkx sentence-transformers pydantic python-multipart
# Для эмбеддингов - BGE-M3 один из лучших на 2026
pip install git+https://github.com/FlagOpen/FlagEmbedding.git

2Индексируем код: из текста в векторы и граф

Вот где начинается магия. Нам нужно пропустить через код два пайплайна: один режет текст на чанки и создает эмбеддинги для LanceDB. Второй - парсит структуру (импорты, вызовы функций, классы) и строит граф. Как НЕ надо делать: пытаться одним скриптом на коленке парсить 10 языков программирования. Возьмите tree-sitter или libclang для C++/Python, а для остального - регулярки и надежду.

# core/indexer.py - упрощенная версия
import os
from typing import List, Dict
import networkx as nx
from sentence_transformers import SentenceTransformer
import lancedb
from pathlib import Path

class HybridIndexer:
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.embedder = SentenceTransformer(model_name)
        self.graph = nx.DiGraph()  # Ориентированный граф для зависимостей
        self.db = lancedb.connect("./data/lancedb")
        
        # Создаем таблицу если нет
        if "code_chunks" not in self.db.table_names():
            self.table = self.db.create_table("code_chunks", 
                                               schema=[{"name": "vector", "type": "vector(1024)"},
                                                       {"name": "text", "type": "string"},
                                                       {"name": "file_path", "type": "string"},
                                                       {"name": "line_start", "type": "int32"}])
        else:
            self.table = self.db.open_table("code_chunks")
    
    def extract_relations(self, file_path: str, code: str):
        """Грубый парсер для Python, на практике используйте tree-sitter"""
        import ast
        try:
            tree = ast.parse(code)
            for node in ast.walk(tree):
                if isinstance(node, ast.FunctionDef):
                    func_name = f"{file_path}::{node.name}"
                    self.graph.add_node(func_name, type="function")
                    # Ищем вызовы внутри функции
                    for subnode in ast.walk(node):
                        if isinstance(subnode, ast.Call):
                            # Упрощенно - извлекаем имя вызываемой функции
                            call_name = self._extract_call_name(subnode)
                            if call_name:
                                self.graph.add_edge(func_name, call_name)
        else:
            self.table = self.db.open_table("code_chunks")
    
    def index_directory(self, path: str):
        """Рекурсивно индексирует все файлы в директории"""
        for root, _, files in os.walk(path):
            for file in files:
                if file.endswith(('.py', '.js', '.ts', '.java', '.cpp')):
                    self._index_file(os.path.join(root, file))
        
        # Сохраняем граф
        nx.write_gml(self.graph, "./data/code_graph.gml")
    
    def _index_file(self, file_path: str):
        """Индексирует один файл: разбивает на чанки и парсит граф"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # 1. Векторный индекс: разбиваем на чанки по 500 токенов
        chunks = self._split_into_chunks(content, chunk_size=500)
        vectors = self.embedder.encode(chunks)
        
        # Добавляем в LanceDB
        data = [{"vector": vec, "text": chunk, 
                 "file_path": file_path, "line_start": i*500}
                for i, (chunk, vec) in enumerate(zip(chunks, vectors))]
        self.table.add(data)
        
        # 2. Графовый индекс: парсим зависимости
        self._parse_dependencies(file_path, content)

Это упрощенный код. В реальности нужно обрабатывать ошибки, кэшировать эмбеддинги и умнее парсить зависимости. Если ваш проект на Python, посмотрите статью про семантический поиск по коду - там подробно про парсинг.

3Сердце системы: гибридный поисковик

Теперь самое интересное - поиск. Алгоритм такой: получаем запрос, делаем эмбеддинг, ищем в LanceDB топ-5 похожих чанков. Параллельно ищем в графе: извлекаем сущности из запроса (имена функций, классов), ищем их в графе, берем соседей. Потом мерджим, убираем дубли, считаем веса.

# core/searcher.py
import numpy as np
from typing import List, Tuple
import networkx as nx

class HybridSearcher:
    def __init__(self, indexer: HybridIndexer):
        self.indexer = indexer
        self.graph = nx.read_gml("./data/code_graph.gml") if os.path.exists("./data/code_graph.gml") else indexer.graph
    
    def search(self, query: str, top_k: int = 10) -> List[Dict]:
        """Гибридный поиск: векторный + графовый"""
        # 1. Векторный поиск
        query_vec = self.indexer.embedder.encode([query])[0]
        vector_results = self.indexer.table.search(query_vec).limit(top_k * 2).to_list()
        
        # 2. Графовый поиск (если в запросе есть имена сущностей)
        entities = self._extract_entities(query)  # Простая эвристика
        graph_results = []
        for entity in entities:
            if entity in self.graph.nodes:
                # Берем саму ноду и ее соседей
                graph_results.append({"text": f"Entity: {entity}", "score": 1.0, "type": "node"})
                for neighbor in self.graph.neighbors(entity):
                    graph_results.append({"text": f"Related to {entity}: {neighbor}", 
                                          "score": 0.7, "type": "edge"})
        
        # 3. Слияние результатов (weighted fusion)
        # Векторным результатам даем вес 1.0, графовым - 0.8 для нод, 0.6 для ребер
        # Но если запрос явно про зависимости - увеличиваем вес графа
        is_dependency_query = any(word in query.lower() for word in ["call", "depend", "import", "use"])
        
        all_results = []
        for res in vector_results:
            weight = 1.0
            all_results.append({"text": res["text"], "score": res["score"] * weight, "source": "vector"})
        
        for res in graph_results:
            weight = 0.9 if is_dependency_query else 0.7
            all_results.append({"text": res["text"], "score": res["score"] * weight, "source": "graph"})
        
        # Сортируем по убыванию score, убираем дубликаты
        seen = set()
        unique_results = []
        for res in sorted(all_results, key=lambda x: x["score"], reverse=True):
            text_hash = hash(res["text"][:100])  # Простейшая дедупликация
            if text_hash not in seen:
                seen.add(text_hash)
                unique_results.append(res)
        
        return unique_results[:top_k]

4Подключаем мозг: интеграция с Ollama

Ollama в 2026 году имеет стабильное REST API. Мы просто отправляем промпт с найденным контекстом. Главное - не перегрузить контекстное окно модели. Llama 3.2 90B имеет 128к токенов, но это не значит, что нужно пихать туда весь код.

# core/assistant.py
import requests
import json
from typing import List

class CodeAssistant:
    def __init__(self, searcher: HybridSearcher, ollama_url: str = "http://localhost:11434"):
        self.searcher = searcher
        self.ollama_url = ollama_url
        self.model = "llama3.2:90b"  # Или ваша любимая модель
    
    def ask(self, question: str) -> str:
        # 1. Ищем релевантный контекст
        context_chunks = self.searcher.search(question, top_k=8)
        context = "\n\n".join([f"[From {chunk['source']}] {chunk['text']}" 
                                 for chunk in context_chunks])
        
        # 2. Формируем промпт
        prompt = f"""Ты - ассистент разработчика. Ответь на вопрос на основе контекста из кода.
        Если информации недостаточно, скажи об этом.
        
        Контекст:
        {context}
        
        Вопрос: {question}
        
        Ответ:"""
        
        # 3. Запрос к Ollama
        response = requests.post(
            f"{self.ollama_url}/api/generate",
            json={
                "model": self.model,
                "prompt": prompt,
                "stream": False,
                "options": {
                    "temperature": 0.3,
                    "num_predict": 1024
                }
            }
        )
        
        if response.status_code == 200:
            return response.json()["response"]
        else:
            return f"Ошибка Ollama: {response.text}"

Температуру ставим низкую (0.3) - нам нужны точные ответы, не креативность. num_predict ограничиваем, чтобы модель не ушла в бесконечное философствование.

5Завершающий штрих: FastAPI и запуск

Теперь оборачиваем все в API, чтобы подключаться из VSCode или через CLI. FastAPI - потому что он быстрый и сам генерирует документацию.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from core.indexer import HybridIndexer
from core.searcher import HybridSearcher
from core.assistant import CodeAssistant
import uvicorn

app = FastAPI(title="Local Code Assistant", version="2026.1")

# Глобальные объекты (в продакшене выносим в контекст)
indexer = HybridIndexer()
searcher = HybridSearcher(indexer)
assistant = CodeAssistant(searcher)

class QueryRequest(BaseModel):
    question: str
    index_path: str = None

@app.post("/api/ask")
async def ask_question(req: QueryRequest):
    """Основной эндпоинт для вопросов по коду"""
    try:
        if req.index_path:
            # Реиндексируем если нужно
            indexer.index_directory(req.index_path)
        
        answer = assistant.ask(req.question)
        return {"answer": answer, "model": "llama3.2:90b"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/index")
async def index_directory(path: str):
    """Принудительная индексация директории"""
    try:
        indexer.index_directory(path)
        return {"status": "ok", "message": f"Indexed {path}"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    # Запуск: uvicorn main:app --reload --host 0.0.0.0 --port 8000
    uvicorn.run(app, host="0.0.0.0", port=8000)

Запускаете, открываете http://localhost:8000/docs - вот вам и Swagger UI. Можно сразу тестировать.

Где собака зарыта: нюансы, которые съедят ваше время

• Эмбеддинги для кода - отдельная боль. BGE-M3 хорош для текста, но для кода есть специализированные модели вроде CodeBERT. На 2026 год уже должны быть мультиязычные эмбеддеры для кода. Поищите "code embedding models 2026" или проверьте репозитории Microsoft или Salesforce.
• Граф ломается на больших проектах. NetworkX хранит все в памяти. Для проекта в 10 млн строк код может сожрать всю RAM. Альтернативы: igraph (библиотека на C) или переход на графовую БД вроде Neo4j (но тогда теряем локальность).
• Ollama падает на больших контекстах. Даже если модель поддерживает 128к токенов, сам Ollama может упасть по памяти. Решение: сильнее фильтровать чанки, использовать summary-модели для сжатия контекста.
• Парсинг графа для JavaScript/TypeScript - ад. Динамическая типизация, callback-и, асинхронность. Tree-sitter помогает, но не идеально. Иногда проще индексировать только явные импорты/экспорты.

Если вы столкнулись с проблемами масштабирования, посмотрите статью про архитектуру локального RAG-пайплайна - там разбираются trade-offs разных хранилищ.

Частые вопросы (которые вы зададите через час отладки)

И последнее: этот ассистент - не ChatGPT. Он не будет писать за вас весь код. Но он точно сэкономит вам часы на чтении документации и поиске связей в legacy. И самое главное - ваш код никуда не уйдет с вашего сервера. В 2026 году это уже не паранойя, а стандартная практика.