fs
object that points to azure?TextNodes
contains blob file paths as metadata, how can I get the custom query engine to access them?Added user message to memory: Extract all line items from each page and them all in a valid JSON schema. For each page verify if all line items in the page before returning an answer. === Calling Function === Calling function: count_tool with args: {"input": "page 1"} === Function Output === Encountered error: [Errno 22] Invalid argument: 'https://<resource_name>.blob.core.windows.net/<container_name>/<folder_1>/<folder_2>/page_2.jpg' === Calling Function === Calling function: table_tool with args: {"input": "page 1"} === Function Output === Encountered error: [Errno 22] Invalid argument: 'https://<resource_name>.blob.core.windows.net/<container_name>/<folder_1>/<folder_2>/page_2.jpg'' === Calling Function === Calling function: table_tool with args: {"input": "page 2"} === Function Output === Encountered error: [Errno 22] Invalid argument: 'https://<resource_name>.blob.core.windows.net/<container_name>/<folder_1>/<folder_2>/page_2.jpg' === Calling Function === Calling function: table_tool with args: {"input": "page 3"} === Function Output === Encountered error: [Errno 22] Invalid argument: 'https://<resource_name>.blob.core.windows.net/<container_name>/<folder_1>/<folder_2>/page_2.jpg' === LLM Response === It seems there was an error while trying to extract the line items from the pages. The error indicates an invalid argument related to accessing the document pages. Please check the document source or provide a valid input for further processing.
VectorStoreIndex
and the custom step you mentioned. How can I write that?index = VectorStoreIndex(text_nodes, embed_model=embed_model) vector_retriever = VectorIndexRetriever(index=index) # image_retriever = ... custom_retriever = CustomRetriever(vector_retriever, ImageRetriever) query_engine = MultimodalQueryEngine( retriever=custom_retriever.as_retriever(similarity_top_k=3), multi_modal_llm=gpt_4o )
from azure.storage.blob import BlobServiceClient class MultimodalQueryEngine(CustomQueryEngine): # ... existing code ... def custom_query(self, query_str: str) -> Response: # retrieve text nodes nodes = self.retriever.retrieve(query_str) # Initialize Azure Blob Service Client blob_service_client = BlobServiceClient.from_connection_string("your_connection_string") # create ImageNode items from text nodes image_nodes = [] for n in nodes: if 'image_path' in n.metadata: try: # Download image from Azure Blob Storage blob_client = blob_service_client.get_blob_client(container="your_container", blob=n.metadata['image_path']) with open(n.metadata['image_path'], "wb") as download_file: download_file.write(blob_client.download_blob().readall()) image_nodes.append( NodeWithScore( node=ImageNode(image_path=n.metadata['image_path']), ) ) except Exception as e: print( f'Warning: Failed to create ImageNode from {n.metadata["image_path"]}: {str(e)}' ) continue ...