Хочу поделиться, как страдал фигней в перерывах от основной деятельности или маленькая история про то, как я хотел сделать «бот по 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), чтобы снизить влияние ранга по сравнению с абсолютным положением. Чем выше суммарное значение - тем релевантнее документ по объединённому мнению ранжировщиков.

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

score(doc) = Σ_i 1 / (k + rank_i(doc))

где суммируем по 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 и использовать для код агента

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