Практычныя прытамулкі: Частка 3: Эксперты RAG — HyDE, самастаўленне запытанняў і агентныя рабочыя практыкі
Практычныя прыказкі: Частка 3 — Эксперты RAG: HyDE, самастаўленыя запиты та агентныя рабочыя практыкі: контракты, перакантрольваннія та блокі коду для команд, якія розробляюць системы RAG.
У гэтым керавану практычна адбудоваць шлях ад сыр'ёў да рабочай системы для: Частка 3: Эксперты RAG — HyDE, самастаўныя запытанні та агентскія прабегі. Акцэнт ставится на крокі, якія можна выконваць, чыстае перакананне і код, які можна проста дадаць у репазітарый без неабяснення меты. Для агульнага відзору, перш чым зменяць код, неабходна визначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння. Аперацыёныя працавікі должны магчымае перадзваніць крок з вядомай точкі контролю без неабяснення схованага стану. Запісваць час выконання і кост токеноў або запытанняя разам з функцыйнальнымі рэзултатамі. Відразы костаў з самага пачатку запобегае неспадзянаным рахункам, калі шлях пераходзіць з дэмаверсіі ў спадзеленыя сераўысы.
1. HyDE: Гіпотэтычныя вбудованні дакументаў
Калі працуеце над 1. HyDE: Hypothetical Document Embeddings, спачатку запісайце умовы: неабяжлівыя даны, сигнал успеху і тое, што выходзіць пад частыя неудачы. Такі список дапамагае заліцварваць можлівыя змены ў коде. Зберагаюце настройкі паза кодам прыемліка. Файлы сераўнавання, храненні секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаюць пераглядаць іх без неабяжлівага чытання всей структуры. Перад налаштаваннем запитоў пераканайцеся, што рэтарыцыя працуе правільна на фіксаванай сэтке запытаў. Частае змены запытаў рэдка калі вярнуюць слабую эфектывнасць пошуку.
Аналіз
Калі працуеце над The Insight, спачатку запісайце умовы кантракту: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список контроля дапамагае заліцьварыць пазнейшыя змены ў кодзе. Документавайце як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкіе перакрыцця і обробка некоректных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам. Змяржывайце рэткасць адпаведзенняў на фіксованыя наборы запытанняў прычымо да налаштавання запрасаў. Частыя змены запрасаў рэдка калі вядома выправляюць слабую систему адзысквання інформаціі.
Як працуе HyDE
Калі працуеце над кнігай «Як працюе HyDE», спачатку запісуйце шаблон: неабяжныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список дапамагае заліцвачыць змяны ў кодзе пазнейша. Валіце маленькія, тэставаныя елементы замест велікіх скрыптов. Калі якісь крок не выйшае, невыпанне павінна вказваць на адну адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся ў рівні воспамінання на фіксаванай сэтцы запитаў. Частая зміна запитоў рэдка калі-небудзь вылечвае слабкую систему аднаходжэння інформацыі.
Step 1: User Query
"What are ISRO's major achievements?"
Step 2: Generate Hypothetical Answer (using LLM)
"ISRO has achieved several milestones including reaching Mars orbit
in 2014, successfully landing Chandrayaan-3 on the Moon's south pole
in 2023, and launching satellites for multiple countries at low cost..."
Step 3: Embed the Hypothetical Answer
[0.23, -0.45, 0.67, ...] ← This lives in document space!
Step 4: Retrieve Documents Similar to Hypothetical Answer
Now we're comparing document-to-document, not query-to-document
Step 5: Generate Final Answer
Use retrieved docs + original query → Better answer
Чаму працюе HyDE
Калі працюеце над кнігай «Why HyDE Works», спачатку запісацеце кантракт: неабяжлівыя вхідныя даны, сігнал успеху і тое, што выходзіць у разе частковага нэуспеху. Такі список контролю дапамагае заліцьварыць пазнейшыя змены коду. Спрэтавайце гэты этап як кантракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даце назвы артыфактам, задаце критэрыя успеху і не падзеляйцеся на частковыя рэшэнні без паведамлення. Змяроўваеце рэкалі на фіксаванай сэтке запытаў прычым падбір прамптав. Частая змена прамптав рэдка калі лечыць слабую спроможнасць да адзыскання інформаціі.
Калі вжываць HyDE
Калі працюеце над «Калі вжываць HyDE», спачатку запісайце умовы працы: неабяжлівыя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заліцварыць будучыя змены ў кодзе. Запісвайце час выканання і кост токенаў або запытаў праза функцыйнае рэзультат. Відкрытыя даны пра косцы з’являюцца раніце, таму не будзе неспакою, калі працэс перейдзе з дамовай среды ў спяльную. Перад налаштаваннем запрошэнняяў пераканайцеся ў якосці адзыначэння патрэбных дадзеных на фіксаванай сэтке запытаў. Частае змены запрошэнняяў рэдка калі вядуць да павышэння якосці адзыначэння дадзеных.
HyDE проты традыцыйнага RAG: рэальны прыклад
Калі працуеце над тэмай HyDE протыкласічнага RAG: Рэальны прыклад, спачатку запісайце умовы: неабяжлівыя даннэ, сігнал успеху і тое, што выходзіць пад частым неудачам. Такі список контроля дапамагае заліцьварыць пазнейшыя змены ў кодзе. Храніце настройкі параду ўнутры коду прыемлівача. Файлы сераўіса, хранальнікі секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавіць аудыт без неабяжлівага чытання всей структуры. Перад налаштаваннем запитоў пераканайцеся ў рэвалю на фіксаваным наборе запитанняў. Частае змена запітоў рэдка калі вярнуе слабую эфектыўнасць пошуку.
2. Контэкстная стысненасць: Точнасць пра колькасць
Калі працюеце над раздзелам 2. «Контекстуальная кампрэсія: тачнасць працоўніка вышэй за колькасць», спачатку запісайце умовы: неабходныя даны, сігнал успеху і тое, што выканаецца у разе частковага невыпалення. Такі список дапамагае заліцварыць пазнейшыя змены ў кодзе. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є частью продукту, а не пазнейшым дапрацоўкам. Перад налаштаваннем запитаў пераканайцеся ў рэкале на фіксаваным наборы пытанняў. Частае змена запитоў рэдка калі вярнайце слабкую эфектыўнасць адзысквання інформаціі.
Проблема
Калі працуеце над задачай, спачатку запісайте умовы виконання: неабяцковыя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список дапамагае заліцварыць пазнейшыя змены ў кодзе. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, невыпанне павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся ў рэзультатах на фіксаванай сэтке запытаў. Частае змены запытаў рэдка калі-небудзь выправляюць слабкую эфектыўнасць адзысквання інформаціі. Калі працуеце над задачай, спачатку запісайте умовы виконання: неабяцковыя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список дапамагае заліцварыць пазнейшыя змены ў кодзе. Запісвайце час выконання і кост токеноў або запытаў разам з функцыйнаімі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі праця пераходзіць з дэмавай версіі ў спяльныя сераўы.
India's capital city is New Delhi, which is located in the northern
part of the country. The city serves as the center of the Government
of India and houses important governmental buildings including the
Parliament House, Rashtrapati Bhavan, and various ministry buildings.
New Delhi was inaugurated in 1931 and became the capital of India
after independence in 1947. The previous capital was Calcutta, now
known as Kolkata. The decision to move the capital was made by the
British colonial government in 1911...
India's capital city is New Delhi
Рашэння проблемы контекстуальнага стиснення
Рашэнне для кампрэсіі контексту працюе наяўнейша, калі яго спрыяваць як меравальную паверхню. Зафіксавайце адны ідеальны прыклад, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць сферу дзеяння. Зберагаюце настройкі пазырочна ад коду прыемліка. Файлы сераўнавання, хранілішчы секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавляць аудыт без неабяжнага чытання всіх дадзеных. Раздзеляйце правілы часткавання дадзеных ад правілаў ўтрымання іх. Змена адных не должна вымагаць перапісву іншых, калі зменяюцыся паказнікі якосці.
Query + Chunk → Compressor LLM → Relevant Sentences Only
Стратэгія рэалізацыі
Стратэгія рэалізацыі працуе наўсёроджэней, калі яе спрыяваюць як мерыемую структуру. Зафіксавайце адну ідеальную версію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Дакументавайце як успішны, так і варыянт вярнэння да нормальнага стану разам. Перапрыбуткі, людзкія контрольны пункты і обработка некоректных паведамленняў є часткай продукту, а не элементамі пазнейшага дапрацоўкі. Раздзеліце правілы частковай обработкі дадзеных ад правіл ўтрымання іх. Змена аднаго з яных не должна вымагаць перапісву другога, калі зменяюцыся паказнікі якосці.
# Traditional: Send full chunks
context = chunk1 + chunk2 + chunk3 # 1500 tokens
# Compressed: Extract relevant parts
for chunk in chunks:
compressed = compressor.extract_relevant(query, chunk)
context.append(compressed) # 300 tokens total
Корысты
Прыяноснае праблематыка працюе наўзям лепш, калі яе спрыяваць як мерыемую сферу. Зберагачыце адну ідеальную версію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, неудача павінна адносіцца да конкрэтной адпаведальнасці, а не да заплутанага ланца задач. Раздзеляйце правілы часткавання інфармацыі ад правіл яе выкарыстоўвання. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказнікі якосці. Прыяноснае праблематыка працюе наўзям лепш, калі яе спрыяваць як мерыемую сферу. Зберагачыце адну ідеальную версію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Запісвайце часы выканання і косты токенав або запытак праза функцыйнае рэзультат. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмавайнага режыма ў спяльныя сервісы.
3. Самастаўленне запытак: нехай LLM адлучыць
Для пункта 3. Автапрацаванне: нехай LLM выконвае рашэння, паказваючы параметры вводу, адміністратара крока і критэрыя завершэння пры змены коду. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не спрабоўваючы з’ясаваць схованы стан. Конфігурацыю трэба залічыць паза кодам прыкладнення. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь граф. Калі наступны крок — це код або вызов інструмента, лепш выкарыстоўваць структураваныя выходныя данні з перакананням схемы, чым вольныя тэкстовыя апісанні.
Канцэпцыя
Для розумення канцэпцыяў неабходна прадзефінаваць вхідныя даны, адпаведальную за кожны крок особу і крэтарыі выходу пры змяне коду. Аператары должны магчыма было перзапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна аддзеўнаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Паказваць трэба тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аператары не зможуць разлічыць галюцинацію ад прасоў у індэксаванні.
Традыцыйны падход проты самастоятельнага запитвання
Для падчынных і самастаўных спосабаў апыткі неабходна перад змянай коду задаць вхідныя даны, абавесця крока і критэрыі завершэння. Аперацыяныя працавнікі должны магчымае перадзеяць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Лепш выбіраць маленькія, тэставаныя елементы заместо велікіх скрыптаў. Калі крок не выйшоў, прычына неудачы должна вказываць на адну конкрэтную абавесцю, а не на заплутаны ланцужок задач. Прытамульвайце тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без цітатаў аперацыяныя працавнікі не зможуць адразніць галюцинацыю ад прасоўкі ў індэксаванні. Для падчынных і самастаўных спосабаў апыткі неабходна перад змянай коду задаць вхідныя даны, абавесця крока і критэрыі завершэння. Аперацыяныя працавнікі должны магчымае перадзеяць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Запісвайце час выконання і кост токеноў або запытаў разам з функцыйнаямі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмавай версіі ў спяльныя среды.
# Developer hardcodes everything
query = "Find recent cricket matches"
top_k = 5
filters = {"category": "sports"}
results = retriever.run(query, top_k, filters)
# LLM decides everything
query = "Find recent cricket matches"
# LLM analyzes and decides:
# - Extract metadata: {"sport": "cricket", "recency": "2024"}
# - Set top_k: 10 (wants comprehensive results)
# - Use hybrid search (keyword "matches" + semantic)
results = self_querying_retriever.run(query)
Чаму важна самастая запытка
Калі працуеце над тэмай «Чаму важна самастая запытка», спачатку запісайце умовы: неабяжлівыя данні, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список дапамагае заліцварыць пазнейшыя змены ў кодзе. Зберагаюце настройкі паза кодам прыемлі. Файлы сераўнавання, хранільнікі секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можуць адбавіць аудыт без неабяжлівага чытання всей структуры. Перад налаштаваннем запытак пераканайцеся ў рэкале на фіксаваным наборе запытак. Частае змена запытак рэдка калі вярнуе слабую якасць выкарыстоўвання данных.
Прыклад: самастая запытка ў практыцы
Калі працуеце над прыкладам «Самастановленне ў практыцы», спачатку запісайце умовы: неабяжныя даны, сигнал успеху і тое, што выходзіць, калі адбываецца частковая нявыплата. Такі список контроля дапамагае заставіць пазнейшыя змены коду застацца чыстымі. Документавайце як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є частью продукту, а не пазнейшым дапрацоўкам. Змяркуйце рэткасць адпаведзенняў на фіксованы набор запитаў прычымо да налаштавання падказак. Частае змена падказак рэдка калі вярнайць слабкую базу адзыскання інформаціі.
{
"semantic_query": "articles about India",
"metadata_filters": {
"year": 2023,
"content_type": "article"
},
"search_type": "hybrid",
"top_k": 10
}
4. Агентны RAG: Высшая эвалюцыя
Калі працуеце над раздзелам 4. «Агентны RAG: Высшая эвалюцыя», спачатку запісайце умовы: неабяжлівыя даннэ, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список дапамагае заліцьварыць пазнейшыя змены ў кодзе. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшоў, нявыпанне должна паказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцужок задач. Перад налаштаваннем запитоў пераканайцеся, як працуея система з фіксаваным наборам запитанняў. Частае змена запитоў рэдка калі вярнайце слабкую эфектыўнасць пошуку.
Што робіць RAG «агентным»?
Калі працюеце над тэмай «Што робіць RAG „агентным“?», спачатку запісайце угоду: неабяжлівыя даннэ, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заліцьварыць пазнейшыя змены ў кодзе. Спрэцьвачайце гэты этап як угоду межа даннемі і перакананымі выходамі. Дайце назву рэзультатам, задацьте критэрыя успеху і не прымайце часткова завершанне без паведамлення. Змяркуйце рэкалі на фіксаванай сэтке запытаў прычымо да налаштавання прапты. Частая змена прапты рэдка калі вылечвае слабую систему адзыскання інфармацыі.
Архітектура агентнага RAG
Калі працуеце з архітектурой Agentic RAG, спачатку запісайце умовы викорыстоўвання: неабходныя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список дапамагае заліцварыць будучыя змены ў кодзе. Запісвайце час выканання і вартасьць токенав або запытаў разам з функцыйнальнымі рэзултатамі. Відразы вартасцей з самага пачатку запобегае неспакою, калі праця пераходзіць з дэмавайнага режыма ў спакульнае сераўерское сэрвіса. Перад налаштаваннем запытаў пераканайцеся ў рэверу на фіксаваным наборе запытаў. Частае змена запытаў рэдка калі-небудзь выправляе слабыя аспекты прадаўання інформацыі.
┌─────────────┐
│ User Query │
└──────┬──────┘
│
┌──────▼──────┐
│ Agent (LLM) │ ← Makes decisions
└──────┬──────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌─────▼──────┐
│ Tool 1 │ │ Tool 2 │ │ Tool 3 │
│ (RAG) │ │(Web Search│ │(Calculator)│
└────┬────┘ └─────┬─────┘ └─────┬──────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌──────▼──────┐
│ Agent │ ← Synthesizes
└──────┬──────┘
│
┌──────▼──────┐
│ Final │
│ Answer │
└─────────────┘
Процес прынятка рашэння агентам
Калі працюеце над процэсам прыняття рашэння агента, спачатку запісайце умовы контракту: неабяжлівыя даны, сігнал успеху і тое, што выходзіць пад частыя неудачы. Такі список контроля дапамагае заліцвачыць змяны ў кодзе пазнейша. Зберагайце настройкі за межамі коду прыемліка. Файлы сераўіса, хранільнікі секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можуць адбавіць аудыт без неабяжлівага чытання всей структуры. Перад налаштаваннем запитоў пераканайцеся, як працуе алгорытм на фіксаванай сэтке запитаў. Частыя змены запитоў рэдка калі вядуць да павышэння эфектывнасі пошуку.
Step 1: Analyze Query
- Requires comparison between two countries
- Need current statistics
- Two separate pieces of info needed
Step 2: Plan Actions
- Tool 1: Search knowledge base for India population
- Tool 2: Search knowledge base for China population
- Tool 3: If not found, use web search as fallback
Step 3: Execute
→ Search knowledge base for India: Found
→ Search knowledge base for China: Not found
→ Fallback to web search for China: Found
Step 4: Synthesize
Combine information from both sources into coherent answer
Функцыі агентскага RAG
Прыклад рабочага процэсу агента
User: "Tell me about ISRO"
Agent: [Uses RAG tool] → Provides answer from knowledge base
User: "What about NASA?"
Agent: [Uses RAG tool] → Not found in knowledge base
[Falls back to web search] → Retrieves info from web
→ Provides answer with source attribution
User: "Compare their budgets"
Agent: [Analyzes] → Needs both ISRO and NASA budget data
[Retrieves from both sources]
[Uses calculator tool for comparison]
→ Provides detailed comparison
Корыстнасць агентскага RAG
Спалучэнне всіх тэхнік: ідеальная система RAG
User Query: "What are recent achievements in India's space program?"
↓
1. Self-Querying
LLM analyzes: Needs recent info, space domain
Filters: {topic: "space", recency: "2023-2024"}
↓
2. HyDE Generation
"ISRO achieved remarkable milestones in 2023-2024, including
successful Moon landings and satellite launches..."
↓
3. Hybrid Retrieval + HyDE
Retrieve using both: original query + hypothetical answer
Get top 20 documents
↓
4. Contextual Compression
Extract only sentences about recent achievements
Reduce 20 chunks (10k tokens) → 5 compressed chunks (2k tokens)
↓
5. Agentic Decision
Agent: "Retrieved info looks good, but let me verify with web search"
→ Quick web search for latest news
→ Combines both sources
↓
6. Final Answer
Comprehensive, accurate, up-to-date response with source attribution
Калі вжываць кожную тэхніку
HyDE
Компрэсія контексту
Самастаўленне запытоў
Агентскі RAG
Пашаговая реалізацыя коду
import os
from pathlib import Path
import json
from typing import List, Dict, Any, Optional
from haystack import Pipeline, Document, component
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import (
InMemoryBM25Retriever,
InMemoryEmbeddingRetriever
)
from haystack.components.embedders import (
SentenceTransformersTextEmbedder,
SentenceTransformersDocumentEmbedder
)
from haystack.components.writers import DocumentWriter
from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner
from haystack.components.converters import TextFileToDocument
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import SentenceTransformersSimilarityRanker
from haystack.components.routers import ConditionalRouter
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
# ============================================================================
# CONFIGURATION
# ============================================================================
# GROQ_API_KEY = "your-groq-api-key-here"
os.environ["GROQ_API_KEY"] = "gsk_!!!"
# Model configurations
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
RERANKER_MODEL = "cross-encoder/ms-marco-MiniLM-L-6-v2"
GROQ_MODEL = "llama-3.3-70b-versatile"
# Retrieval parameters
BM25_TOP_K = 10
EMBEDDING_TOP_K = 10
RERANKER_TOP_K = 5
# ============================================================================
# READING DATA FILE
# ============================================================================
def load_documents_with_metadata(file_path: str) -> list[Document]:
documents = []
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
data = json.loads(line)
documents.append(
Document(
content=data["content"],
meta=data.get("meta", {})
)
)
return documents
# ============================================================================
# COMPONENT 1: HyDE (HYPOTHETICAL DOCUMENT EMBEDDINGS)
# ============================================================================
@component
class HyDEGenerator:
"""Generate hypothetical documents for HyDE retrieval"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Generate a detailed paragraph that would perfectly answer
the following question. Write as if you're providing the ideal answer from a knowledge base.
Question: {query}
Ideal Answer Paragraph:"""
@component.output_types(hypothetical_doc=str)
def run(self, query: str) -> Dict[str, str]:
"""Generate hypothetical document"""
prompt = self.template.format(query=query)
result = self.llm.run(prompt=prompt)
hypothetical_doc = result["replies"][0]
return {"hypothetical_doc": hypothetical_doc}
# ============================================================================
# COMPONENT 2: CONTEXTUAL COMPRESSOR
# ============================================================================
@component
class ContextualCompressor:
"""Extract only relevant sentences from retrieved documents"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Given the following document chunk and query, extract ONLY
the sentences that are directly relevant to answering the query.
Return only the relevant sentences, nothing else.
Query: {query}
Document Chunk:
{chunk}
Relevant Sentences:"""
@component.output_types(compressed_documents=List[Document])
def run(self, query: str, documents: List[Document]) -> Dict[str, List[Document]]:
"""Compress documents by extracting relevant content"""
compressed_docs = []
for doc in documents:
prompt = self.template.format(query=query, chunk=doc.content)
result = self.llm.run(prompt=prompt)
compressed_content = result["replies"][0]
# Create new document with compressed content
compressed_doc = Document(
content=compressed_content,
meta=doc.meta,
score=doc.score if hasattr(doc, 'score') else None
)
compressed_docs.append(compressed_doc)
return {"compressed_documents": compressed_docs}
# ============================================================================
# COMPONENT 3: SELF-QUERYING ANALYZER
# ============================================================================
@component
class SelfQueryAnalyzer:
"""Analyze query and extract metadata filters automatically"""
def __init__(self, llm: OpenAIGenerator):
self.llm = llm
self.template = """Analyze the following query and extract:
1. The core semantic query (cleaned, focused version)
2. Any metadata filters that should be applied
Available metadata fields:
- category: geography, politics, economy, sports, science, culture
- year: any year (e.g., 2023, 2024)
- topic: overview, capital, gdp, cricket, space, entertainment, language
- source: any source type
Query: {query}
Respond in this exact format:
SEMANTIC_QUERY: [your semantic query here]
FILTERS: category=value,year=value (or FILTERS: none if no filters apply)"""
@component.output_types(semantic_query=str, filters=Dict[str, Any])
def run(self, query: str) -> Dict[str, Any]:
"""Analyze query and extract filters"""
prompt = self.template.format(query=query)
result = self.llm.run(prompt=prompt)
response = result["replies"][0]
# Parse response
lines = response.strip().split('\n')
semantic_query = query # default
filters = {}
for line in lines:
if line.startswith("SEMANTIC_QUERY:"):
semantic_query = line.replace("SEMANTIC_QUERY:", "").strip()
elif line.startswith("FILTERS:"):
filters_str = line.replace("FILTERS:", "").strip()
if filters_str.lower() != "none":
# Parse filters
for filter_pair in filters_str.split(','):
if '=' in filter_pair:
key, value = filter_pair.split('=')
key = key.strip()
value = value.strip()
# Try to convert year to int
if key == "year":
try:
value = int(value)
except:
pass
filters[key] = value
return {
"semantic_query": semantic_query,
"filters": filters
}
# ============================================================================
# COMPONENT 4: ANSWER QUALITY CHECKER (FOR AGENTIC ROUTING)
# ============================================================================
@component
class AnswerQualityChecker:
"""Check if answer is satisfactory or needs web search fallback"""
@component.output_types(quality_score=str, route=str)
def run(self, answer: str, query: str) -> Dict[str, str]:
"""Check answer quality"""
# Simple heuristic - in production, use an LLM
if "I don't have" in answer or "cannot answer" in answer or len(answer) < 50:
return {"quality_score": "low", "route": "web_search"}
else:
return {"quality_score": "high", "route": "final_answer"}
# ============================================================================
# INDEXING WITH METADATA
# ============================================================================
def index_documents_with_metadata(document_store, file_path):
documents = load_documents_with_metadata(file_path)
embedder = SentenceTransformersDocumentEmbedder(model=EMBEDDING_MODEL)
embedder.warm_up()
docs_with_embeddings = embedder.run(documents)
document_store.write_documents(docs_with_embeddings["documents"])
print(f" Indexed {len(documents)} documents with metadata")
@component
class ReplySelector:
"""Select the primary reply from LLM output"""
@component.output_types(answer=str)
def run(self, replies: List[str]) -> Dict[str, str]:
if not replies:
return {"answer": ""}
return {"answer": replies[0]}
# ============================================================================
# BUILD EXPERT RAG PIPELINE WITH ALL TECHNIQUES
# ============================================================================
def build_expert_rag_pipeline(document_store):
"""Build expert RAG pipeline with HyDE, compression, and agentic routing"""
# Initialize LLMs
main_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 512, "temperature": 0.1}
)
hyde_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 300, "temperature": 0.1}
)
compressor_llm = OpenAIGenerator(
api_key=Secret.from_env_var("GROQ_API_KEY"),
api_base_url="https://api.groq.com/openai/v1",
model=GROQ_MODEL,
generation_kwargs={"max_tokens": 200, "temperature": 0.1}
)
# Initialize components
pipeline = Pipeline()
pipeline.add_component("reply_selector", ReplySelector())
# Self-querying
pipeline.add_component("self_query", SelfQueryAnalyzer(main_llm))
# HyDE generation
pipeline.add_component("hyde_generator", HyDEGenerator(hyde_llm))
# Embedders
pipeline.add_component(
"text_embedder",
SentenceTransformersTextEmbedder(model=EMBEDDING_MODEL)
)
pipeline.add_component(
"hyde_embedder",
SentenceTransformersTextEmbedder(model=EMBEDDING_MODEL)
)
# Dual retrievers
pipeline.add_component(
"bm25_retriever",
InMemoryBM25Retriever(document_store=document_store, top_k=BM25_TOP_K)
)
pipeline.add_component(
"semantic_retriever",
InMemoryEmbeddingRetriever(document_store=document_store, top_k=EMBEDDING_TOP_K)
)
# Document processing
pipeline.add_component("document_joiner", DocumentJoiner())
pipeline.add_component(
"ranker",
SentenceTransformersSimilarityRanker(model=RERANKER_MODEL, top_k=RERANKER_TOP_K)
)
# Contextual compression
pipeline.add_component("compressor", ContextualCompressor(compressor_llm))
# Answer generation
answer_template = """Answer the question based on the provided context.
If the context doesn't contain enough information, say so clearly.
Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:"""
pipeline.add_component("prompt_builder", PromptBuilder(template=answer_template))
pipeline.add_component("answer_generator", main_llm)
# Quality checker for agentic routing
pipeline.add_component("quality_checker", AnswerQualityChecker())
# Connect components
# Self-querying → retrieval
pipeline.connect("self_query.semantic_query", "bm25_retriever.query")
pipeline.connect("self_query.semantic_query", "text_embedder.text")
# HyDE pathway
pipeline.connect("self_query.semantic_query", "hyde_generator.query")
pipeline.connect("hyde_generator.hypothetical_doc", "hyde_embedder.text")
# Retrievers
pipeline.connect("text_embedder.embedding", "semantic_retriever.query_embedding")
# Join and rank
pipeline.connect("bm25_retriever.documents", "document_joiner.documents")
pipeline.connect("semantic_retriever.documents", "document_joiner.documents")
pipeline.connect("document_joiner.documents", "ranker.documents")
# Compression
pipeline.connect("ranker.documents", "compressor.documents")
# Answer generation
pipeline.connect("compressor.compressed_documents", "prompt_builder.documents")
pipeline.connect("prompt_builder", "answer_generator")
# Quality checking
pipeline.connect("answer_generator.replies", "reply_selector.replies")
pipeline.connect("reply_selector.answer", "quality_checker.answer")
return pipeline
# ============================================================================
# CONVERSATION CONTEXT MANAGER (FROM PART 2)
# ============================================================================
class ConversationContext:
"""Manages conversation history"""
def __init__(self, max_history=5):
self.history = []
self.max_history = max_history
def add_exchange(self, question: str, answer: str):
self.history.append({"question": question, "answer": answer})
if len(self.history) > self.max_history:
self.history = self.history[-self.max_history:]
def get_history_text(self) -> str:
if not self.history:
return ""
return "\n".join([f"Q: {e['question']}\nA: {e['answer']}" for e in self.history])
# ============================================================================
# EXPERT QUERY PROCESSOR
# ============================================================================
def process_expert_query(
pipeline,
query: str,
context: ConversationContext,
show_details=True
):
"""Process query with full expert RAG pipeline"""
if show_details:
print(f"\n{'='*70}")
print(f" Expert RAG Processing")
print(f"{'='*70}")
print(f"\n Original Query: {query}")
# Run the pipeline
try:
result = pipeline.run(
{
"self_query": {"query": query},
"ranker": {"query": query},
"compressor": {"query": query},
"prompt_builder": {"question": query},
"quality_checker": {"query": query}
},
include_outputs_from=["reply_selector", "quality_checker", "compressor"]
)
if show_details:
# Show self-querying results
if "self_query" in result:
print(f"\n Self-Query Analysis:")
print(f" Semantic Query: {result['self_query'].get('semantic_query', 'N/A')}")
print(f" Filters: {result['self_query'].get('filters', {})}")
# Show HyDE results
if "hyde_generator" in result:
hyde_doc = result['hyde_generator']['hypothetical_doc']
print(f"\n HyDE Hypothetical Document:")
print(f" {hyde_doc[:200]}...")
# Show compression results
if "compressor" in result:
print(f"\n Contextual Compression:")
compressed_docs = result['compressor']['compressed_documents']
print(f" Compressed {len(compressed_docs)} documents")
for i, doc in enumerate(compressed_docs[:2], 1):
print(f"\n Doc {i}: {doc.content[:150]}...")
# Show quality check
if "quality_checker" in result:
quality = result['quality_checker']
print(f"\n Answer Quality: {quality.get('quality_score', 'N/A')}")
print(f" Route Decision: {quality.get('route', 'N/A')}")
# Get final answer
answer = result["reply_selector"]["answer"]
if show_details:
print(f"\n{'─'*70}")
print(f" Final Answer:")
print(f"{answer}")
print(f"{'─'*70}")
# Update context
context.add_exchange(query, answer)
return answer, result
except Exception as e:
print(f"\n Error: {e}")
return f"Error processing query: {e}", {}
# ============================================================================
# MAIN EXECUTION
# ============================================================================
def main():
"""Main execution function"""
print("="*70)
print("EXPERT RAG SYSTEM - HAYSTACK + GROQ")
print(" Part 3: HyDE + Compression + Self-Querying + Agentic RAG")
print("="*70)
# Step 1: Initialize document store
print("\nStep 1: Initializing Document Store")
print("-"*70)
document_store = InMemoryDocumentStore()
# Step 2: Index documents with metadata
print("\nStep 2: Indexing Documents with Metadata")
print("-"*70)
index_documents_with_metadata(document_store, "/content/india_info.json")
# Step 3: Build expert RAG pipeline
print("\n Step 3: Building Expert RAG Pipeline")
print("-"*70)
pipeline = build_expert_rag_pipeline(document_store)
print("Expert RAG pipeline ready with:")
# Step 4: Initialize conversation context
print("\nStep 4: Initializing Conversation Context")
print("-"*70)
conversation = ConversationContext()
print("Conversation manager ready")
# Step 5: Test queries
print("\n" + "="*70)
print("TESTING EXPERT RAG")
print("="*70)
test_questions = [
"What are ISRO's major space achievements?",
"Tell me about India's economy",
"Find information about cricket from recent years",
"What makes the space program cost-effective?", # Follow-up with context
]
for i, question in enumerate(test_questions, 1):
print(f"\n\n{'#'*70}")
print(f"TEST QUERY {i}")
print(f"{'#'*70}")
answer, result = process_expert_query(
pipeline,
question,
conversation,
show_details=True
)
if __name__ == "__main__":
main()
Практычныя прыклады выкарыстання
Прыклад 1: Система кавалерыяў-вадзімаў у медыцыне
Прыклад 2: Памочнік для правовых даследжэнняў
Прыклад 3: Бот для падтрымкі кляўэнтаў
Полны комплект тэхналагій Expert RAG
Layer 1: Data Ingestion
- PDF/TXT/HTML converters
- Hierarchical chunking (better than fixed-size)
- Metadata extraction
Layer 2: Storage
- Vector DB (embeddings)
- Graph DB (relationships)
- SQL DB (metadata)
Layer 3: Retrieval
- HyDE generation
- Hybrid search (BM25 + Semantic)
- Self-querying with metadata
- Multi-hop retrieval
Layer 4: Processing
- Contextual compression
- Reranking
- Deduplication
Layer 5: Generation
- Agentic orchestration
- Tool usage
- Multi-source synthesis
- Source attribution
Layer 6: Monitoring
- Latency tracking
- Quality metrics
- Cost monitoring
- Error logging