Custom component for Neo4J - PDF insert

[pmane]

I am new to Langflow. Trying to build custom component that can insert pdf file to Neo4j post extracting entities.

Below is my code. It does not give error but call to PropertyGraphIndex.from_documents does not return when I press play button on the Langlow UI for my custom component. The strange thing is when I press it again, the print statement after the PropertyGraphIndex.from_documents is printed.

I also get few errors on console as - 2024-07-25 15:12:01 - ERROR - service.py:72

Is this something to do with async support ? not sure how this works. Same code works well in VS code outside langflow.

'''
2024-07-25 20:42:01 Attempting to open file at: /app/data/.cache/langflow/233e4542-268c-45c5-a680-0c516edef30f/Test2.pdf
2024-07-25 20:42:01 llm and embed_model setup complete
2024-07-25 20:42:01 Simple Extractor created- Alice, a software engineer at TechCorp, works on a project with Bob, a data scientist. Both Alice
2024-07-25 20:42:01 and Bob report to Carol, the project manager. TechCorp is a leading tech company founded by
2024-07-25 20:42:01 David. Alice collaborates closely with Eve, a cybersecurity expert, to ensure the security of their
2024-07-25 20:42:01 project. Bob often consults with Frank, an external consultant from DataWorks, for data analysis
2024-07-25 20:42:01 techniques. Carol coordinates with Grace, the HR manager at TechCorp, for project staffing.
2024-07-25 20:42:01
2024-07-25 20:42:01
Parsing nodes: 0%| | 0/1 [00:00<?, ?it/s]
Parsing nodes: 100%|██████████| 1/1 [00:00<00:00, 229.21it/s]
2024-07-25 20:42:01 [07/25/24 15:12:01] ERROR 2024-07-25 15:12:01 - ERROR - service.py:72
2024-07-25 20:42:01 service - Request error occurred:
2024-07-25 20:42:06
Extracting paths from text: 0%| | 0/1 [00:00<?, ?it/s]
Extracting paths from text: 100%|██████████| 1/1 [00:05<00:00, 5.42s/it]
Extracting paths from text: 100%|██████████| 1/1 [00:05<00:00, 5.42s/it]
2024-07-25 20:42:07
Generating embeddings: 0%| | 0/1 [00:00<?, ?it/s]
Generating embeddings: 100%|██████████| 1/1 [00:00<00:00, 2.29it/s]
Generating embeddings: 100%|██████████| 1/1 [00:00<00:00, 2.29it/s]
2024-07-25 20:42:07
Generating embeddings: 0it [00:00, ?it/s]
Generating embeddings: 0it [00:00, ?it/s]
2024-07-25 20:42:08 2024-07-25 15:12:08,378 - 140737470420800 - result.py-result:321 - WARNING: Received notification from DBMS server: {severity: WARNING} {code: Neo.ClientNotification.Statement.FeatureDeprecationWarning} {category: DEPRECATION} {title: This feature is deprecated and will be removed in future versions.} {description: The procedure has a deprecated field. ('config' used by 'apoc.meta.graphSample' is deprecated.)} {position: line: 1, column: 1, offset: 0} for query: "CALL apoc.meta.graphSample() YIELD nodes, relationships RETURN nodes, [rel in relationships | {name:apoc.any.property(rel, 'type'), count: apoc.any.property(rel, 'count')}] AS relationships"
2024-07-25 20:42:19 /app/.venv/lib/python3.12/site-packages/neo4j/_sync/driver.py:532: ResourceWarning: unclosed Neo4jDriver: <neo4j._sync.driver.Neo4jDriver object at 0x7fff9a404d70>.
2024-07-25 20:42:19 _unclosed_resource_warn(self)
2024-07-25 20:42:19 ResourceWarning: Enable tracemalloc to get the object allocation traceback
2024-07-25 20:42:19 /app/.venv/lib/python3.12/site-packages/neo4j/_sync/driver.py:537: DeprecationWarning: Relying on Driver's destructor to close the session is deprecated. Please make sure to close the session. Use it as a context (with statement) or make sure to call .close() explicitly. Future versions of the driver will not close drivers automatically.
2024-07-25 20:42:19 _deprecation_warn(
2024-07-25 20:42:19 /app/.venv/lib/python3.12/site-packages/neo4j/_async/driver.py:533: ResourceWarning: unclosed AsyncNeo4jDriver: <neo4j._async.driver.AsyncNeo4jDriver object at 0x7fff70f0b9b0>.
'''

When I press the play button again, prints from previous run are printed

'''
2024-07-25 20:42:01 Index created----------------------------------------------------------------------------------------
2024-07-25 20:42:01 post created----------------------------------------------------------------------------------------
2024-07-25 20:42:01 Attempting to open file at: /app/data/.cache/langflow/233e4542-268c-45c5-a680-0c516edef30f/Test2.pdf
'''

'''Python

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data
from neo4j import GraphDatabase
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore
from llama_index.core import Document, PropertyGraphIndex
from llama_index.core.indices.property_graph import SimpleLLMPathExtractor
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from loguru import logger
from loguru import logger
import fitz

class CustomComponent(Component):
display_name = "Custom Component"
description = "Use as a template to create your own component."
documentation: str = "http://docs.langflow.org/components/custom"
icon = "custom_components"
name = "CustomComponent"

inputs = [
    MessageTextInput(name="input_value", display_name="Input Value", value="Hello, World!"),
    FileInput(
        name="pdf_file",
        display_name="PDF File",
        file_types=["pdf"],
        info="Upload a PDF file in to Neo4J database",
    ),
]

outputs = [
    Output(display_name="Output", name="output", method="build_output"),
]

def build_output(self) -> Data:
    logger.debug("Test log to file")
    try:
        if not self.pdf_file:
            raise ValueError("Please upload a PDF file to use this component.")

        resolved_path = self.resolve_path(self.pdf_file)
        file_path = Path(resolved_path)
        

        if file_path.suffix.lower() != '.pdf':
            raise ValueError("The uploaded file must be a PDF file.")
                
        index = self.create_index_neo4j(file_path)
        print("post created------------------")
        data = Data(value="test123")
        self.status = data
        return data            

    except Exception as e:
        error_message = f"An error occurred: {str(e)}"
        self.status = error_message
        return [Data(data={"error": error_message})]
    
    
def setup_neo4j(self):
    uri = os.getenv("NEO4J_URI", "neo4j://localhost:7687")
    username = os.getenv("NEO4J_USERNAME", "neo4j")
    password = os.getenv("NEO4J_PASSWORD", "XXXXXXXXX")

    try:
        # Create a Neo4jPGStore instance
        graph_store = Neo4jPropertyGraphStore(
            username=username,
            password=password,
            url=uri,
        )
        
        return graph_store  # Return the Neo4jPGStore instance
    except Exception as e:
        print(f"Error connecting to Neo4j: {e}")
        raise
    
def create_index_neo4j(self,pdf_path):
    # Extract text from PDF
    text = self.extract_text_from_pdf(pdf_path)
    if text is None:
        raise ValueError("Failed to extract text from the provided PDF.")
    
    documents = [Document(text=text)]

    # Set up Neo4j
    graph_store = self.setup_neo4j()

    # Set up LLM and embedding model
    llm, embed_model = self.setup_llama()

    print("llm and embed_model setup complete")
    # Define the dynamic entity and relation extraction
    kg_extractor = SimpleLLMPathExtractor(llm=llm)
    print("Simple Extractor created-" , text)
    
    # Create the property graph index
    index = PropertyGraphIndex.from_documents(
        documents,
        kg_extractors=[kg_extractor],
        llm=llm,
        embed_model=embed_model,
        property_graph_store=graph_store,  # Use Neo4jPGStore instance here
        show_progress=True,
    )
    
    print("Index created----------------------------")
    return index           
    
    
def extract_text_from_pdf(self,pdf_path):
    print(f"Attempting to open file at: {pdf_path}")  # Debug print
    try:
        doc = fitz.open(pdf_path)
        text = ""
        for page in doc:
            text += page.get_text()
        return text
    except Exception as e:
        print(f"Error reading PDF file: {e}")
        return None

def setup_llama(self):
    openai_api_key = "sk-XXXXXXXXXXXX"
    if not openai_api_key:
        raise ValueError("OpenAI API key not found in environment variables.")
    os.environ["OPENAI_API_KEY"] = openai_api_key
    
    llm = OpenAI(model="gpt-4", temperature=0.0)
    embed_model = OpenAIEmbedding(model_name="text-embedding-ada-002")
    return llm, embed_model

'''

[dosubot[bot]]

Hey there, @pmane! I'm here to help you with any bugs or questions you have. Let's tackle this issue together and get it sorted out.

The issue you're encountering with the PropertyGraphIndex.from_documents method not returning immediately and only printing the statement after pressing the play button again could be related to how asynchronous operations are handled in Langflow. Additionally, the warnings and errors in your console logs indicate potential issues with resource management and deprecated features in Neo4j.

Here are some steps and best practices to address these issues:

1. Asynchronous Operations Handling

Langflow handles asynchronous operations using Celery tasks. Ensure that your custom component is designed to handle asynchronous tasks properly. You might need to adjust the _sync_ flag in your API calls to manage long-running tasks effectively.

2. Proper Management of Neo4j Sessions and Drivers

To avoid resource warnings and deprecation issues, ensure that Neo4j sessions and drivers are properly managed and closed. Use context managers to handle the session scope. Here is an example:

from contextlib import contextmanager
from neo4j import GraphDatabase

class Neo4jService:
    def __init__(self, uri, user, password):
        self._driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self._driver.close()

    @contextmanager
    def session_scope(self):
        session = self._driver.session()
        try:
            yield session
            session.commit()
        except:
            session.rollback()
            raise
        finally:
            session.close()

# Usage example
neo4j_service = Neo4jService("bolt://localhost:7687", "neo4j", "password")

with neo4j_service.session_scope() as session:
    # Perform database operations using the session
    pass

# Close the driver when done
neo4j_service.close()

3. Adjust Timeout Settings

Langflow has specific timeout settings that could affect the execution of long-running tasks. Adjust the timeout and soft_time_limit parameters to accommodate longer-running tasks:

# Example of adjusting timeout settings
timeout: int = typer.Option(300, help="Worker timeout in seconds.")
soft_time_limit = 30  # Adjust as needed

4. Debugging and Logging

Ensure that you have proper logging in place to debug and trace the flow of execution. Use the log method provided by the CustomComponent class to log messages:

def log(self, message: LoggableType | list[LoggableType], name: str | None = None):
    if name is None:
        name = self.display_name if self.display_name else self.__class__.__name__
    if hasattr(message, "model_dump") and isinstance(message, BaseModel):
        message = message.model_dump()
    log = Log(message=message, type=get_artifact_type(message), name=name)
    self._logs.append(log)
    if self.vertex:
        self._tracing_service.add_log(trace_name=self.vertex.id, log=log)

5. Handling File Uploads and Caching

Langflow handles file uploads and caching using specific methods. Ensure that you are reading and writing files in small chunks to conserve memory and using hashing to ensure unique file names:

@create_cache_folder
def save_uploaded_file(file: UploadFile, folder_name):
    cache_path = Path(CACHE_DIR)
    folder_path = cache_path / folder_name
    filename = file.filename
    if isinstance(filename, str) or isinstance(filename, Path):
        file_extension = Path(filename).suffix
    else:
        file_extension = ""
    file_object = file.file

    if not folder_path.exists():
        folder_path.mkdir()

    sha256_hash = hashlib.sha256()
    file_object.seek(0)
    while chunk := file_object.read(8192):
        sha256_hash.update(chunk)

    hex_dig = sha256_hash.hexdigest()
    file_name = f"{hex_dig}{file_extension}"

    file_object.seek(0)
    file_path = folder_path / file_name
    with open(file_path, "wb") as new_file:
        while chunk := file_object.read(8192):
            new_file.write(chunk)

    return file_path

6. Optimizing Performance

To optimize performance, increase the number of worker processes, extend the worker timeout, use a more efficient cache, and adjust the logging level:

langflow run --workers 4 --timeout 120 --cache InMemoryCache --log-level error

By following these steps and best practices, you should be able to address the issues with your custom component and ensure smooth operation within Langflow.

To continue talking to Dosu, mention @dosu.