216216 "id" : " 279b41dc"
217217 "metadata" : {},
218218 "outputs" : [],
219- "source": "# Operator name mapping\nOPERATOR_MAP = {\n 'trino': 'trino-operator',\n 'airflow': 'airflow-operator',\n 'druid': 'druid-operator',\n 'hdfs': 'hdfs-operator',\n 'hbase': 'hbase-operator',\n 'hive': 'hive-operator',\n 'kafka': 'kafka-operator',\n 'nifi': 'nifi-operator',\n 'opensearch': 'opensearch-operator',\n 'spark': 'spark-k8s-operator',\n 'superset': 'superset-operator',\n 'zookeeper': 'zookeeper-operator',\n 'opa': 'opa-operator',\n 'secret': 'secret-operator',\n 'listener': 'listener-operator',\n 'commons': 'commons-operator'\n}\n\ndef detect_operator(query):\n \"\"\"Detect operator names in query for metadata filtering.\"\"\"\n query_lower = query.lower()\n for keyword, operator in OPERATOR_MAP.items():\n if keyword in query_lower:\n return operator\n return None\n\ndef is_implementation_query(query):\n \"\"\"Detect if query is asking for implementation details.\"\"\"\n impl_keywords = ['how', 'deploy', 'configure', 'setup', 'install', 'create', 'implement', 'example']\n query_lower = query.lower()\n return any(keyword in query_lower for keyword in impl_keywords)\n\nprint(\"Query enhancement functions loaded\")"
219+ "source" : [
220+ " # Operator name mapping\n "
221+ " OPERATOR_MAP = {\n "
222+ " 'trino': 'trino-operator',\n "
223+ " 'airflow': 'airflow-operator',\n "
224+ " 'druid': 'druid-operator',\n "
225+ " 'hdfs': 'hdfs-operator',\n "
226+ " 'hbase': 'hbase-operator',\n "
227+ " 'hive': 'hive-operator',\n "
228+ " 'kafka': 'kafka-operator',\n "
229+ " 'nifi': 'nifi-operator',\n "
230+ " 'opensearch': 'opensearch-operator',\n "
231+ " 'spark': 'spark-k8s-operator',\n "
232+ " 'superset': 'superset-operator',\n "
233+ " 'zookeeper': 'zookeeper-operator',\n "
234+ " 'opa': 'opa-operator',\n "
235+ " 'secret': 'secret-operator',\n "
236+ " 'listener': 'listener-operator',\n "
237+ " 'commons': 'commons-operator'\n "
238+ " }\n "
239+ " \n "
240+ " def detect_operator(query):\n "
241+ " \"\"\" Detect operator names in query for metadata filtering.\"\"\"\n "
242+ " query_lower = query.lower()\n "
243+ " for keyword, operator in OPERATOR_MAP.items():\n "
244+ " if keyword in query_lower:\n "
245+ " return operator\n "
246+ " return None\n "
247+ " \n "
248+ " def is_implementation_query(query):\n "
249+ " \"\"\" Detect if query is asking for implementation details.\"\"\"\n "
250+ " impl_keywords = ['how', 'deploy', 'configure', 'setup', 'install', 'create', 'implement', 'example']\n "
251+ " query_lower = query.lower()\n "
252+ " return any(keyword in query_lower for keyword in impl_keywords)\n "
253+ " \n "
254+ " print(\" Query enhancement functions loaded\" )"
255+ ]
220256 },
221257 {
222258 "cell_type" : " markdown"
259295 "id" : " ab6kt8bl6kr"
260296 "metadata" : {},
261297 "outputs" : [],
262- "source" : " def build_knn_query(query_embedding, detected_operator, k):\n \"\"\" Build k-NN query with optional operator filter.\"\"\"\n knn_query = {\n 'embedding': {\n 'vector': query_embedding,\n 'k': k * 3 # Get more candidates for rescoring\n }\n }\n \n # Add operator filter if operator detected\n if detected_operator:\n knn_query['embedding']['filter'] = {\n 'term': {'operator': detected_operator}\n }\n \n return knn_query\n\n print(\" k-NN query builder loaded\" )"
298+ "source" : [
299+ " def build_knn_query(query_embedding, detected_operator, k):\n "
300+ " \"\"\" Build k-NN query with optional operator filter.\"\"\"\n "
301+ " knn_query = {\n "
302+ " 'embedding': {\n "
303+ " 'vector': query_embedding,\n "
304+ " 'k': k * 3 # Get more candidates for rescoring\n "
305+ " }\n "
306+ " }\n "
307+ " \n "
308+ " # Add operator filter if operator detected\n "
309+ " if detected_operator:\n "
310+ " knn_query['embedding']['filter'] = {\n "
311+ " 'term': {'operator': detected_operator}\n "
312+ " }\n "
313+ " \n "
314+ " return knn_query\n "
315+ " \n "
316+ " print(\" k-NN query builder loaded\" )"
317+ ]
263318 },
264319 {
265320 "cell_type" : " markdown"
333388 "id" : " fdmm66xll2h"
334389 "metadata" : {},
335390 "outputs" : [],
336- "source" : " def format_search_results(opensearch_response):\n \"\"\" Extract relevant fields from OpenSearch response.\"\"\"\n results = []\n for hit in opensearch_response['hits']['hits']:\n results.append({\n 'title': hit['_source']['title'],\n 'content': hit['_source']['content'],\n 'category': hit['_source']['category'],\n 'operator': hit['_source'].get('operator', ''),\n 'has_code_block': hit['_source'].get('has_code_block', False),\n 'url': hit['_source'].get('url', ''),\n 'score': hit['_score']\n })\n return results\n\n print(\" Search result formatter loaded\" )"
391+ "source" : [
392+ " def format_search_results(opensearch_response):\n "
393+ " \"\"\" Extract relevant fields from OpenSearch response.\"\"\"\n "
394+ " results = []\n "
395+ " for hit in opensearch_response['hits']['hits']:\n "
396+ " results.append({\n "
397+ " 'title': hit['_source']['title'],\n "
398+ " 'content': hit['_source']['content'],\n "
399+ " 'category': hit['_source']['category'],\n "
400+ " 'operator': hit['_source'].get('operator', ''),\n "
401+ " 'has_code_block': hit['_source'].get('has_code_block', False),\n "
402+ " 'url': hit['_source'].get('url', ''),\n "
403+ " 'score': hit['_score']\n "
404+ " })\n "
405+ " return results\n "
406+ " \n "
407+ " print(\" Search result formatter loaded\" )"
408+ ]
337409 },
338410 {
339411 "cell_type" : " markdown"
351423 "id" : " y1wrd5gztoi"
352424 "metadata" : {},
353425 "outputs" : [],
354- "source": "def search_documents(query, k=10, log_query=False):\n \"\"\"Search for relevant document chunks using OpenSearch hybrid search (k-NN + BM25).\"\"\"\n # Step 1: Get embedding\n query_embedding = get_embedding(query)\n \n # Step 2: Detect enhancements\n detected_operator = detect_operator(query)\n boost_code_blocks = is_implementation_query(query)\n \n # Step 3: Build k-NN query\n knn_query = build_knn_query(query_embedding, detected_operator, k)\n \n # Step 4: Build rescore query\n rescore_query = build_rescore_query(query, boost_code_blocks)\n rescore_query['window_size'] = k * 3\n \n # Step 5: Construct complete search body\n search_body = {\n 'size': k,\n 'query': {\n 'knn': knn_query\n },\n 'rescore': rescore_query,\n '_source': ['title', 'content', 'category', 'operator', 'has_code_block', 'url']\n }\n \n # Log the query if requested (with truncated vector for readability)\n if log_query:\n # Create display version with truncated vector\n log_body = json.loads(json.dumps(search_body)) # Deep copy\n vector = log_body['query']['knn']['embedding']['vector']\n log_body['query']['knn']['embedding']['vector'] = [vector[0], vector[1], vector[2], \"...\", vector[-3], vector[-2], vector[-1]]\n print(\"OpenSearch Query:\")\n print(json.dumps(log_body, indent=2))\n print()\n \n # Step 6: Execute search\n response = opensearch_client.search(\n index=INDEX_NAME,\n body=search_body\n )\n \n # Step 7: Format and return results\n return format_search_results(response)\n\nprint(\"OpenSearch hybrid search orchestrator loaded\")"
426+ "source" : [
427+ " def search_documents(query, k=10, log_query=False):\n "
428+ " \"\"\" Search for relevant document chunks using OpenSearch hybrid search (k-NN + BM25).\"\"\"\n "
429+ " # Step 1: Get embedding\n "
430+ " query_embedding = get_embedding(query)\n "
431+ " \n "
432+ " # Step 2: Detect enhancements\n "
433+ " detected_operator = detect_operator(query)\n "
434+ " boost_code_blocks = is_implementation_query(query)\n "
435+ " \n "
436+ " # Step 3: Build k-NN query\n "
437+ " knn_query = build_knn_query(query_embedding, detected_operator, k)\n "
438+ " \n "
439+ " # Step 4: Build rescore query\n "
440+ " rescore_query = build_rescore_query(query, boost_code_blocks)\n "
441+ " rescore_query['window_size'] = k * 3\n "
442+ " \n "
443+ " # Step 5: Construct complete search body\n "
444+ " search_body = {\n "
445+ " 'size': k,\n "
446+ " 'query': {\n "
447+ " 'knn': knn_query\n "
448+ " },\n "
449+ " 'rescore': rescore_query,\n "
450+ " '_source': ['title', 'content', 'category', 'operator', 'has_code_block', 'url']\n "
451+ " }\n "
452+ " \n "
453+ " # Log the query if requested (with truncated vector for readability)\n "
454+ " if log_query:\n "
455+ " # Create display version with truncated vector\n "
456+ " log_body = json.loads(json.dumps(search_body)) # Deep copy\n "
457+ " vector = log_body['query']['knn']['embedding']['vector']\n "
458+ " log_body['query']['knn']['embedding']['vector'] = [vector[0], vector[1], vector[2], \" ...\" , vector[-3], vector[-2], vector[-1]]\n "
459+ " print(\" OpenSearch Query:\" )\n "
460+ " print(json.dumps(log_body, indent=2))\n "
461+ " print()\n "
462+ " \n "
463+ " # Step 6: Execute search\n "
464+ " response = opensearch_client.search(\n "
465+ " index=INDEX_NAME,\n "
466+ " body=search_body\n "
467+ " )\n "
468+ " \n "
469+ " # Step 7: Format and return results\n "
470+ " return format_search_results(response)\n "
471+ " \n "
472+ " print(\" OpenSearch hybrid search orchestrator loaded\" )"
473+ ]
355474 },
356475 {
357476 "cell_type" : " markdown"
367486 "cell_type" : " markdown"
368487 "id" : " ln0hj4qo5tm"
369488 "metadata" : {},
370- "source" : " ### 4.1: Format Context for LLM\n\n Each retrieved document is formatted with metadata (title, operator, relevance score, URL) to help the LLM understand the source quality and provide citations."
489+ "source" : [
490+ " ### 4.1: Format Context for LLM\n "
491+ " \n "
492+ " Each retrieved document is formatted with metadata (title, operator, relevance score, URL) to help the LLM understand the source quality and provide citations."
493+ ]
371494 },
372495 {
373496 "cell_type" : " code"
374497 "execution_count" : null ,
375498 "id" : " i9mu8uy4dr9"
376499 "metadata" : {},
377500 "outputs" : [],
378- "source" : " def format_context_for_llm(docs):\n \"\"\" Format retrieved documents into context string with metadata.\"\"\"\n context_parts = []\n for doc in docs:\n context_parts.append(\n f\" Source: {doc['title']} (from {doc['operator']}, relevance: {doc['score']:.2f})\\ n\"\n f\" URL: {doc.get('url', 'N/A')}\\ n\"\n f\" {doc['content']}\"\n )\n return \"\\ n\\ n\" .join(context_parts)\n\n print(\" Context formatter loaded\" )"
501+ "source" : [
502+ " def format_context_for_llm(docs):\n "
503+ " \"\"\" Format retrieved documents into context string with metadata.\"\"\"\n "
504+ " context_parts = []\n "
505+ " for doc in docs:\n "
506+ " context_parts.append(\n "
507+ " f\" Source: {doc['title']} (from {doc['operator']}, relevance: {doc['score']:.2f})\\ n\"\n "
508+ " f\" URL: {doc.get('url', 'N/A')}\\ n\"\n "
509+ " f\" {doc['content']}\"\n "
510+ " )\n "
511+ " return \"\\ n\\ n\" .join(context_parts)\n "
512+ " \n "
513+ " print(\" Context formatter loaded\" )"
514+ ]
379515 },
380516 {
381517 "cell_type" : " markdown"
500636 "id" : " 6tbawkzl0gd"
501637 "metadata" : {},
502638 "outputs" : [],
503- "source": "def rag_query(query, k=10, stream=True, log_query=True):\n \"\"\"Complete RAG workflow: search + generate.\"\"\"\n print(f\"Query: {query}\")\n print()\n \n # Show if operator detected\n detected_operator = detect_operator(query)\n if detected_operator:\n print(f\"Detected operator: {detected_operator}\")\n \n print(\"Retrieving relevant chunks...\")\n print()\n \n # Search with query logging\n context_docs = search_documents(query, k=k, log_query=log_query)\n \n print(f\"Found {len(context_docs)} relevant chunks:\")\n for i, doc in enumerate(context_docs, 1):\n code_indicator = \" [code]\" if doc.get('has_code_block') else \"\"\n print(f\" {i}. {doc['title']} (score: {doc['score']:.3f}){code_indicator}\")\n print(f\" {doc['url']}\")\n \n print()\n print(\"Generating response with Llama 3.1 8B...\")\n print()\n answer = generate_response(query, context_docs, stream=stream)\n print(\"=\" * 80)\n \n return {'query': query, 'context': context_docs, 'answer': answer}\n\nprint(\"RAG pipeline ready\")"
639+ "source" : [
640+ " def rag_query(query, k=10, stream=True, log_query=True):\n "
641+ " \"\"\" Complete RAG workflow: search + generate.\"\"\"\n "
642+ " print(f\" Query: {query}\" )\n "
643+ " print()\n "
644+ " \n "
645+ " # Show if operator detected\n "
646+ " detected_operator = detect_operator(query)\n "
647+ " if detected_operator:\n "
648+ " print(f\" Detected operator: {detected_operator}\" )\n "
649+ " \n "
650+ " print(\" Retrieving relevant chunks...\" )\n "
651+ " print()\n "
652+ " \n "
653+ " # Search with query logging\n "
654+ " context_docs = search_documents(query, k=k, log_query=log_query)\n "
655+ " \n "
656+ " print(f\" Found {len(context_docs)} relevant chunks:\" )\n "
657+ " for i, doc in enumerate(context_docs, 1):\n "
658+ " code_indicator = \" [code]\" if doc.get('has_code_block') else \"\"\n "
659+ " print(f\" {i}. {doc['title']} (score: {doc['score']:.3f}){code_indicator}\" )\n "
660+ " print(f\" {doc['url']}\" )\n "
661+ " \n "
662+ " print()\n "
663+ " print(\" Generating response with Llama 3.1 8B...\" )\n "
664+ " print()\n "
665+ " answer = generate_response(query, context_docs, stream=stream)\n "
666+ " print(\" =\" * 80)\n "
667+ " \n "
668+ " return {'query': query, 'context': context_docs, 'answer': answer}\n "
669+ " \n "
670+ " print(\" RAG pipeline ready\" )"
671+ ]
504672 },
505673 {
506674 "cell_type" : " markdown"
582750 "metadata" : {},
583751 "outputs" : [],
584752 "source" : [
585- " # Try your own question here\n "
586- " my_query = \" YOUR QUESTION HERE\"\n "
587- " \n "
588753 " # Use stream=True (default) to watch tokens generate in real-time\n "
589- " # Use stream=False for instant complete response\n "
590- " result = rag_query(my_query, stream=True)"
754+ " # Use stream=False to display the complete response once it is generated\n "
755+ " # Modify k to change the number of documentation chunks that will be retrieved from OpenSearch to be used as context in the prompt.\n "
756+ " # Use log_query=False to stop logging the OpenSearch query.\n "
757+ " result = rag_query(\" YOUR QUESTION HERE\" , stream=True, k=10, log_query=True)"
591758 ]
592759 }
593760 ],
604771 },
605772 "nbformat" : 4 ,
606773 "nbformat_minor" : 5
607- }
774+ }
0 commit comments