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