[Python] Llama index query engine returns 0 nodes

I'm trying to do a semantic query using llamaindex and OpenAI llm. Indexes are stored in a postgresql db with pgvector extensions. I have verified that embedding vectors and search indexes are there in db. Still I'm getting 0 nodes returned and an "empty Response" returned when querying in bot_helper.py in the get_top_documents() method I've made sure that the vectorstore and index are intialized properly and have checked that the filter values in POST request match those in the embeddings db.

bot helper.py

import asyncio
import logging
from openai import OpenAI
from utils.prompt_parameters import PromptParameter
from sqlalchemy.orm import Session
from models.vectorstoremodel import VectorStoreModel
from config import OPENAI_API_KEY,EmbeddingModel
from utils.enums import DataSource, Type
import json
from sqlalchemy import text
from utils.llama_index_helper import LlamaIndexHelper
from llama_index.core.chat_engine.types import ChatMode
import os

# At the top of your application, after your imports
logging.basicConfig(
level=logging.INFO, # Change from DEBUG to INFO or WARNING
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Specifically silence the verbose OpenAI client
logging.getLogger("openai").setLevel(logging.WARNING)
logging.getLogger("llama_index").setLevel(logging.DEBUG)

class BotHelper:
def __init__(self, post, db_session: Session):
self.post = post
self.db_session = db_session
#self.openai_client = self.get_openai_client()
self.prompt_param = PromptParameter.from_json(post)
self.top_vector_stores=[]
self.llama_index_helper = LlamaIndexHelper(db_session)
self.index = self.llama_index_helper.get_index()

os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY

async def set_top_vector_stores(self):
if self.prompt_param.skip_cosine_similarity:
return self.get_all_documents()

embedding_model = EmbeddingModel
#user_query_embedding = self.openai_client.embeddings.create(input=self.prompt_param.search_terms, model=embedding_model)
# user_query_embedded = self.llama_index_helper.get_text_embedding(self.prompt_param.search_terms)
print(self.prompt_param.search_terms)
top_vector_stores = self.get_top_documents(self.prompt_param.search_terms)
# top_vector_stores = self.get_top_documents(self.prompt_param.search_terms)


if top_vector_stores:
return top_vector_stores

# If no similar docs found, lower the threshold and check again
self.prompt_param.cosine_threshold = self.prompt_param.lesser_cosine_threshold
is_private = len(self.prompt_param.data_source) == 1 and self.prompt_param.data_source[0] == DataSource.PRIVATE

return self.get_all_documents() if is_private else self.get_top_documents(user_query_embedded)

async def expand_query(self):
if not self.prompt_param.expand_question:
return

system_message = "You are a helpful assistant, expand the given query."
# response = await self.openai_client.create_chat_completion([{"role": "system", "content": system_message},
# {"role": "user", "content": self.prompt_param.prompt}])
response = await self.llama_index_helper.create_completion([{"role": "system", "content": system_message},
{"role": "user", "content": self.prompt_param.prompt}])

self.prompt_param.prompt += " " + response[0]['message']

async def extract_search_text_and_intent(self):
"""Extract search text and intent from the user query"""
self.prompt_param.search_terms = self.prompt_param.prompt # ✅ Default search terms

if not self.prompt_param.extract_search:
return # ✅ Skip if extraction is disabled

if DataSource.PUBLIC in self.prompt_param.data_source:
return # ✅ Skip if the data source is public

original_query = self.prompt_param.prompt
doc = self.get_search_text_and_intent_prompt(original_query) # ✅ Get formatted prompt

# choices = await self.openai_client.create_chat_completion(
# doc, self.prompt_param.gpt_model_4, 1
# ) # ✅ Await the async function call
choices = await self.llama_index_helper.create_completion(
doc
)

expanded_query = "\n".join(choice["message"].replace("\n", "") for choice in choices)

try:
final_json = json.loads(expanded_query) # ✅ Parse JSON response
self.prompt_param.intent = final_json.get("intent", "")
self.prompt_param.search_terms = final_json.get("search_text", self.prompt_param.prompt)
self.prompt_param.bot_prompt = final_json.get("additional_questions", "")

except json.JSONDecodeError:
logging.error("Failed to parse AI response as JSON.")

def get_search_text_and_intent_prompt(self, prompt: str):
"""Generate system and user messages for extracting search text and intent."""
return [
{"role": "system", "content": (
"You are an AI assistant that extracts search text, intent, and also provides ***additional questions*** "
"to ask the user if no relevant search is found from user prompts. Response should be a valid JSON object. "
"***In the response don't add ```json at the start and ``` at the end***"
)},
{"role": "user", "content": (
f"Extract the key search text, intent, and ***additional questions combined in a single paragraph in HTML format*** "
f"from the following prompt in JSON format: '{prompt}'"
)}
]

async def get_top_documents_after_semantic_search(self):
""" Perform semantic search and return top documents """
is_public = DataSource.PUBLIC in self.prompt_param.data_source

# If public model, return empty (skip search)
if is_public:
return self.top_vector_stores

# Adjust cosine similarity settings for specific types
if self.prompt_param.type == Type.INSIGHT_REPORT:
self.prompt_param.skip_cosine_similarity = True
if self.prompt_param.type == Type.HELP and DataSource.MMKBARTICLE in self.prompt_param.data_source:
self.prompt_param.cosine_limit = 1

# Set top vector stores
self.top_vector_stores = await self.set_top_vector_stores()

return self.top_vector_stores

def type_condition(self):
""" SQL condition for filtering by Type """
return f' type = {int(self.prompt_param.type.value)} '

def get_all_documents(self):
""" Fetch all documents if cosine similarity is skipped """
external_ids = self.prompt_param.external_ids
type_condition = self.type_condition()
where_or_and = " WHERE " if not external_ids else " AND "
pre_where_condition = f"{where_or_and} {type_condition}"
where_cond = pre_where_condition if not external_ids else self.id_parameters(external_ids, pre_where_condition)

query = f'SELECT id,name,content,external_id,embedding_queue_id FROM vectorstore {where_cond};'

return self.db_session.execute(text(query)).fetchall()

def id_parameters(self, external_ids, pre_where_condition):
""" Generate SQL condition for filtering by external IDs """
if self.prompt_param.type == Type.INSIGHT_REPORT:
return f' WHERE embedding_queue_id = {external_ids[0]} {pre_where_condition}'
else:
ids = ",".join(map(int, external_ids))
return f' WHERE external_id" IN ({ids}) {pre_where_condition}'

def get_top_documents(self, user_query):
""" Perform cosine similarity search using pgvector """

metadata_filters=self.llama_index_helper.get_filters(int(self.prompt_param.type.value), self.prompt_param.external_ids)
query_engine = self.index.as_query_engine(
similarity_top_k=self.prompt_param.max_cosine_limit,
filters=metadata_filters
)
# print(os.environ.get("OPENAI_API_KEY"))

# self.llama_index_helper.diagnose_vector_store()

# chat_engine = self.index.as_chat_engine(
# chat_mode=ChatMode.BEST,
# similarity_top_k=10, # Retrieve more documents
# vector_store_query_mode="default",
# # Use a lower threshold than your current setting
# similarity_cutoff=0.5
# )
# res = chat_engine.chat("What is USS congress")
# print(res)
# retriever = self.index.as_retriever()
# res = retriever.retrieve("What is USS congress")
# print(res)

print(f"Querying: {user_query}")
results = query_engine.query( user_query ) # user_query_embedded)
if not results:
return []

print(f"Results {results}")

filtered_results = sorted(
(doc for doc in results if doc.similarity >= self.prompt_param.cosine_threshold),
key=lambda doc: doc.similarity,
reverse=True # Sort descending
)[:self.prompt_param.cosine_limit] # Limit results

return filtered_results


async def process_response(self, vscm):
"""
Process retrieved documents and summarize response.
"""
# If public LLM model is used, return direct summary
if DataSource.PUBLIC in self.prompt_param.data_source:
return await self.summarize_response([])

if not vscm: # No documents found
return []

# If request is for KB Article
if self.prompt_param.type == Type.HELP and DataSource.MMKBARTICLE in self.prompt_param.data_source:
response = await self.process_kb_article(vscm)
if not self.prompt_param.summarize_the_kb_article:
return response
if not response:
return response
vscm[0].content = response[0]["data"]

return await self.process_docs_and_summarize(vscm)

async def process_kb_article(self, vscm):
"""
Retrieve content for a KB article.
"""
response = []
embedding_queue_entry = await self.get_embedding_queue_by_id(vscm[0].embedding_queue_id)
if embedding_queue_entry:
response.append({
"data": embedding_queue_entry.content,
"cosinesimilarity": vscm[0].cosine_similarity,
"externalid": vscm[0].external_id
})
return response

async def process_docs_and_summarize(self, vscm):
"""
Process documents and summarize the response.
"""
response = await self.process_docs_async(vscm)
return await self.tokenize_and_summarize(response, vscm)

async def process_docs_async(self, vscm):
"""
Process retrieved documents using an AI model.
"""
response = []
model = self.prompt_param.gpt_model_4
docs = [x.content for x in vscm]

# Simulate OpenAI completion
tasks = [self.llama_index_helper.create_chat_completion(prompt=doc, model=model, max_tokens=200) for doc in docs]
results = await self.execute_tasks(tasks)

for i, result in enumerate(results):
response.append({
"data": result,
"cosinesimilarity": vscm.cosine_similarity,
"externalid": vscm.external_id,
"name": vscm.name
})

return response

async def tokenize_and_summarize(self, responses, vscm):
"""
Tokenize and summarize response data.
"""
docs = [x["data"] for x in responses]
vector_store_ids = list(set([x.id for x in vscm]))

response = []
model = self.prompt_param.model_for_summary

tasks = [self.llama_index_helper.create_chat_completion(prompt=doc, model=model, max_tokens=200) for doc in docs]
results = await self.execute_tasks(tasks)

for result in results:
response.append({
"data": result,
"vectorstoreids": vector_store_ids
})

return response

async def summarize_response(self, response):
"""
Summarize final AI response.
"""
responses = [x["data"] for x in response]
cosine_similarity = list(set([x["cosinesimilarity"] for x in response]))

response_string = "\n\n".join(responses)
chat_message = f"{self.prompt_param.prompt}\n\n{self.prompt_param.system_prompt}\n\n{response_string}"

tasks = [self.llama_index_helper.create_chat_completion(prompt=chat_message, model=self.prompt_param.model_for_summary, max_tokens=200)]
results = await self.execute_tasks(tasks)

return [{"data": results[0], "cosinesimilarity": cosine_similarity}]

async def execute_tasks(self, tasks):
"""
Execute async AI requests.
"""
return await asyncio.gather(*tasks)

async def get_embedding_queue_by_id(self, embedding_queue_id):
"""
Fetch embedding queue entry by ID.
"""
result = self.db_session.query(VectorStoreModel).filter(VectorStoreModel.id == embedding_queue_id).first()
return result


llama_index_helper.py

import os
from typing import Dict, List, Optional
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.vector_stores.postgres import PGVectorStore
from llama_index.core.vector_stores import MetadataFilters, MetadataFilter
from flask import request
import psycopg2
from config import POSTGRES_URI_TEMPLATE,ASYNC_POSTGRES_URI_TEMPLATE
from config import EmbeddingModel,OPENAI_API_KEY, GPTModel
from llama_index.core.settings import Settings
from llama_index.llms.openai import OpenAI
from llama_index.core.node_parser import SimpleFileNodeParser

import logging

class LlamaIndexHelper:
def __init__(self, db_session):
self.db_connection=POSTGRES_URI_TEMPLATE.format(tenant=request.tenant)
self.embed_model = OpenAIEmbedding(model_name=EmbeddingModel, api_key=OPENAI_API_KEY)
self.llm = OpenAI(GPTModel)

Settings.embed_model = self.embed_model
Settings.llm = self.llm
Settings.node_parser = SimpleFileNodeParser.from_defaults()

self.vector_store = PGVectorStore(
connection_string=self.db_connection,
async_connection_string=ASYNC_POSTGRES_URI_TEMPLATE.format(tenant=request.tenant),
table_name="vectorstore",
schema_name="public",
embed_dim=1536,
hnsw_kwargs={
"hnsw_m": 16,
"hnsw_ef_construction": 64,
"hnsw_ef_search": 40,
"hnsw_dist_method": "vector_cosine_ops",
}
)

os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY

def get_text_embedding(self, text):
embed_model = OpenAIEmbedding(model_name=EmbeddingModel, api_key=OPENAI_API_KEY)
return embed_model.get_text_embedding(text)

def get_index(self):
storage_context = StorageContext.from_defaults(vector_store=self.vector_store)
return VectorStoreIndex.from_vector_store(self.vector_store, storage_context=storage_context)

def create_index(self, documents):
storage_context = StorageContext.from_defaults(vector_store=self.vector_store)
return VectorStoreIndex.from_documents(documents, storage_context, show_progress=True)

def get_filters(self, type, external_ids):
filters=[]
if(type):
filters.append(MetadataFilter(key="type", value=type))
if(external_ids):
filters.append(MetadataFilter(key="external_id", value=external_ids, operator="in"))

return MetadataFilters(filters=filters)


I tried using index as_chat_engine() and querying llm directly which seems to work except when chat mode is set to ChatMode.CONTEXT. I've been trying to resolve this for hours now but haven't been able to figure out any working solution.

Here is the log when querying

httpx - INFO - HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"

llama_index.core.indices.utils - DEBUG - > Top 0 nodes:

Continue reading...