Простенький RAG своими руками

Хочу поделиться, как страдал фигней в переывах от основной деятельности или маленькая история про то, как я хотел сделать «бот по wiki». Cпросил про наш проект, получил короткий ответ и пошёл дальше работать.

Есть Confluence с описанием продукта (спецификации, docs), есть Python, внутренняя LLM, ну и кривые руки + немного времени. И да я не пайтон разработчик, мой максимум всякая автоматизация, поэтому смело пинайте мой код, я на нем не женат. Цель - чтобы бот мог отвечать на «объясни XXX».

Идея вообще простая

Берём Confluence, берем текст из нужных нам статей и индексируем в квадрант ([qdrant](https://qdrant.tech/)).

Понятно, что всякие регламенты от QA и лишние шумовые документы не хочется засовывать в систему - мозг и так забит, зачем бота травить этим же? Поэтому входной параметр у нас -страница, от которой рекурсивно идём вниз по дереву страниц и собираем только релевантный контент.

Индексируем Confluence

Confluence даёт HTML, а HTML для LLM информативностью часто ~0, много лишнего ввиде тегов. Решение простое: парсим HTML и конвертим в MD. Для этого есть готовые библиотеки, которые нормально очищают разметку, таблицы, кодовые блоки - переводим в читабельный markdown и дальше уже работаем с текстом, а не с HTML-мишурой. Так как писал вещь, связанную с AI, то и взял [langchain_community.document_transformers](https://reference.langchain.com/python/), думаю вы знаете, что по сути это markdownify. Единственно не хотелось тащить всякие заголовки страницы, поэтому обрезал content, используя BeautifulSoup.

from langchain_community.document_transformers import MarkdownifyTransformer md_transformer = MarkdownifyTransformer() def convert_to_md(html): soup = BeautifulSoup(html, 'html.parser') div_content = soup.find('div', id='content') clean_html = str(div_content) converted_docs = md_transformer.transform_documents([Document(page_content=clean_html)]) return converted_docs[0].page_content.replace("https://", "").replace("http://", "") def crawl(url, depth=0): if depth == 2: return print(url) parsed_url = urlparse(url) page_id = None if 'pageId' in url: page_id = parse_qs(parsed_url.query)['pageId'][0] if not page_id or 'src=contextnavpagetreemode' in url: return if url in visited or len(visited) >= MAX_PAGES or (page_id in visited): return visited.add(url) visited.add(page_id) print(f"Crawled: {url}") html = get_html_from_confluence(url, CONFLUENCE_TOKEN) content = convert_to_md(html) index_to_elastic(content, url) index_to_qdrant(content, url) # Follow links soup = BeautifulSoup(html, "html.parser") base_domain = urlparse(START_URL).netloc links = list(soup.find_all("a", href=True)) for link in links: new_url = urljoin(url, link["href"]) if urlparse(new_url).netloc == base_domain: # stay in same domain crawl(new_url, depth + 1) def index_to_elastic(content, url): document = { "content": content, "url": url } elastic.index(index=SPEC_INDEX, document=document)

Нарезка документов и векторы

После приведения к MD берём библиотеку для разрезания на кусочки (chunking) - иначе LLM загнётся на длинных текстах, да и семантический поиск перестанет работать в принципе. Для каждого чанка считаем эмбеддинги и индексируем в векторную базу - qdrant. На Хабре было множество статей про это. Параллельно хочется сохранять полнотекстовый индекс для классического поиска (BM25) - elastic - opensearch.

MODEL_NAME = "ai-forever/sbert_large_nlu_ru" embedder = SentenceTransformer(MODEL_NAME) def chunk_text(text): max_tokens = 200 # splitter = TextSplitter.from_tiktoken_model("gpt-3.5-turbo", max_tokens) # chunks = splitter.chunks(text) tokenizer = Tokenizer.from_pretrained("bert-base-uncased") splitter = TextSplitter.from_huggingface_tokenizer(tokenizer, max_tokens) chunks = splitter.chunks(text) return chunks def index_to_qdrant(content, url): chunks = list(chunk_text(content)) if chunks: flattened_list = flatten(chunks) if flattened_list != chunks: pass vectors = embedder.encode(flattened_list).tolist() points = [] for i, chunk in enumerate(chunks): points.append(PointStruct( id=uuid.uuid4().hex, vector=vectors[i], payload={ "text": chunk, "url": url, } )) qdrant.upload_points(collection_name=COLLECTION_NAME, points=points)

возможно тут я накосячил ) Но - у нас есть полный qdrant и elastic данных.

Делаем RAG: и ничего не работает

Пишем простенький RAG: на запрос швыряем похожие векторы из qdrant, подсовываем в LLM вместе с промптом и ждём ответ.

def explain(question): global result, content, ans result = search_in_qdrant(question) content = '' spec_urls = [] for chunk in result: content = content + '\n' + chunk['text'] spec_urls.append(chunk['spec_url']) ans = ai.ask_gemma( f'You are an assistant for technical documentation. ' f'Your task is to answer questions in Russian, ' f'relying strictly on the provided context. ' f'Provide detailed, well-structured, and accurate responses. ' f'If the context does not contain sufficient information for a complete answer, clearly indicate this and, ' f'if possible, suggest where the information might be found. Context {content}, user question: {question}') ans = ans + "\n use this context:\n" for url in list(OrderedDict.fromkeys(spec_urls)): ans = ans + "\n" + url return ans

И нифига не работает: ответы не релевантны, LLM генерирует общий бред или отвечает частично. Что делать (кроме мата)?

Маленький дисклеймер: Релевантность определялась мной, по метрикам заложенным в мой мозг при рождении, то есть, если я могу понять про фичу из ответа, значит норм. если же фича описана в ответе, стиле бредогенератора ai - не релевантно. Примеров ответа конечно же не будет, ибо NDA

Гибрид: два индекса сразу

Пробуем гибрид: поиск по векторной БД + поиск по BM25 в Elastic. Оба индекса выдают свои ранги и набор документов - остаётся вопрос, как свести ранги разных индексов в единый топ? Наткнулся на RRF - Reciprocal Rank Fusion.

Что такое RRF

RRF - это простой способ объединить ранги от разных ранжировщиков. Для каждого документа считаем сумму 1 / (k + rank), где rank - позиция документа в выдаче, а k - маленькая константа (обычно 60 или 50), чтобы снизить влияние ранга по сравнению с абсолютным положением. Чем выше суммарное значение - тем релевантнее документ по объединённому мнению ранжировщиков.

Формула (просто чтобы было понятно):

где суммируем по i - каждому ранжировщику (векторный поиск, bm25 и т.д.).

Преимущество RRF - устойчивость: документ, который стабильно фигурирует в середине всех списков, может опередить редкий «первый» из одного источника. Очень простой и рабочий фьюжн.

Получилось такое:

def hybrid_search_rrf_real(query: str, es_size: int = 10, qdrant_limit: int = 100, top_k: int = 15): query_vector = embedder.encode(query).tolist() es_results = elastic.search( index=SPEC_INDEX, query={"match": {"content": query}}, size=es_size ) es_rankings = {} for i, hit in enumerate(es_results["hits"]["hits"], start=1): spec_url = hit["_source"]["url"] print(spec_url) es_rankings[spec_url] = rrf_score(i) candidate_urls = list(es_rankings.keys()) if not candidate_urls: return [] qdrant_results = qdrant.search( collection_name=COLLECTION_NAME, query_vector=query_vector, query_filter=models.Filter( must=[ models.FieldCondition( key="url", match=models.MatchAny(any=candidate_urls) ) ] ), limit=qdrant_limit ) qdrant_rankings = {} for i, r in enumerate(qdrant_results, start=1): spec_url = r.payload["url"] chunk_text = r.payload["text"] qdrant_rankings[(spec_url, chunk_text)] = rrf_score(i) # ---- Merge ES + Qdrant with RRF ---- combined = {} for spec_url, score in es_rankings.items(): combined.setdefault(spec_url, 0.0) combined[spec_url] += score for (spec_url, chunk_text), score in qdrant_rankings.items(): combined.setdefault(spec_url, 0.0) combined[spec_url] += score # ---- Rerank chunks based on combined doc-level RRF ---- reranked = [] for r in qdrant_results: spec_url = r.payload["url"] chunk_text = r.payload["text"] reranked.append({ "spec_url": spec_url, "text": chunk_text, "final_score": combined.get(spec_url, 0.0) }) reranked = sorted(reranked, key=lambda x: x["final_score"], reverse=True) return reranked[:top_k]

Но и это не сразу помогло

После внедрения RRF результатов не стало в разы лучше - всё равно результаты хоть и стали лучше, но релевантность хромала. Обидевшись на векторный поиск (но не свои же руки обижать), решил: Доверять BM25 (Elastic) больше, чем qdrant. Почему? Потому что Confluence - техническая документация, в ней важны точные термины, заголовки, контекст: BM25 это ловит. Векторный поиск полезен для синонимов и «смягчённого» семантического совпадения, но сам по себе даёт слишком общий контекст.

def hybrid_search_rrf(query: str, es_size: int = 1, qdrant_limit: int = 100, top_k: int = 40): query_vector = embedder.encode(query).tolist() es_results = elastic.search( index=SPEC_INDEX, query={ "bool": { "must": [ {"match": {"content": {"query": query, "minimum_should_match": "75%"}}} ], "should": [ {"match_phrase": {"content": {"query": query, "slop": 5, "boost": 2}}} ] } }, size=es_size ) es_rankings = {} for i, hit in enumerate(es_results["hits"]["hits"], start=1): spec_url = hit["_source"]["url"] print(spec_url) es_rankings[spec_url] = rrf_score(i) candidate_urls = list(es_rankings.keys()) if not candidate_urls: return [] qdrant_results = qdrant.search( collection_name=COLLECTION_NAME, query_vector=query_vector, query_filter=models.Filter( must=[ models.FieldCondition( key="url", match=models.MatchAny(any=candidate_urls) ) ] ), limit=qdrant_limit ) qdrant_rankings = {} for i, r in enumerate(qdrant_results, start=1): spec_url = r.payload["url"] chunk_text = r.payload["text"] qdrant_rankings[(spec_url, chunk_text)] = rrf_score(i) # ---- Merge ES + Qdrant with RRF ---- combined = {} for spec_url, score in es_rankings.items(): combined.setdefault(spec_url, 0.0) combined[spec_url] += score for (spec_url, chunk_text), score in qdrant_rankings.items(): combined.setdefault(spec_url, 0.0) combined[spec_url] += score # ---- Rerank chunks based on combined doc-level RRF ---- reranked = [] for r in qdrant_results: spec_url = r.payload["url"] chunk_text = r.payload["text"] reranked.append({ "spec_url": spec_url, "text": chunk_text, "final_score": combined.get(spec_url, 0.0) }) reranked = sorted(reranked, key=lambda x: x["final_score"], reverse=True) return reranked[:top_k]

и соответственно сам бот стал таким:

def explain(question): global result, content, ans result = hybrid_search_rrf(question) content = '' spec_urls = [] for chunk in result: content = content + '\n' + chunk['text'] spec_urls.append(chunk['spec_url']) result = hybrid_search_rrf_real(question) for chunk in result: content = content + '\n' + chunk['text'] spec_urls.append(chunk['spec_url']) ans = ai.ask_gemma( f'You are an assistant for technical documentation. ' f'Your task is to answer questions in Russian, ' f'relying strictly on the provided context. ' f'Provide detailed, well-structured, and accurate responses. ' f'If the context does not contain sufficient information for a complete answer, clearly indicate this and, ' f'if possible, suggest where the information might be found. Context {content}, user question: {question}') ans = ans + "\n Что использовалось в ответе:\n" for url in list(OrderedDict.fromkeys(spec_urls)): ans = ans + "\n" + url return ans

Оказалось ответы стали релевантными

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

дальше идея дополнительно обернуть в mcp и использовать для код агента