csv_kg_chat

langroid/agent/special/neo4j/csv_kg_chat.py

CSVGraphAgent(config)

Bases: Neo4jChatAgent

Source code in langroid/agent/special/neo4j/csv_kg_chat.py

def __init__(self, config: CSVGraphAgentConfig):
    formatted_build_instr = ""
    if isinstance(config.data, pd.DataFrame):
        df = config.data
        self.df = df
    else:
        if config.data:
            df = read_tabular_data(config.data, config.separator)
            df_cleaned = _preprocess_dataframe_for_neo4j(df)

            df_cleaned.columns = df_cleaned.columns.str.strip().str.replace(
                " +", "_", regex=True
            )

            self.df = df_cleaned

            formatted_build_instr = BUILD_KG_INSTRUCTIONS.format(
                header=self.df.columns, sample_rows=self.df.head(3)
            )

    config.system_message = config.system_message + formatted_build_instr
    super().__init__(config)

    self.config: Neo4jChatAgentConfig = config

    self.enable_message(PandasToKGTool)

pandas_to_kg(msg)

Creates nodes and relationships in the graph database based on the data in a CSV file.

Parameters:

Name	Type	Description	Default
msg	PandasToKGTool	An instance of the PandasToKGTool class containing the necessary information for generating nodes.	required

Returns:

Name	Type	Description
str	str	A string indicating the success or failure of the operation.

Source code in langroid/agent/special/neo4j/csv_kg_chat.py

def pandas_to_kg(self, msg: PandasToKGTool) -> str:
    """
    Creates nodes and relationships in the graph database based on the data in
    a CSV file.

    Args:
        msg (PandasToKGTool): An instance of the PandasToKGTool class containing
            the necessary information for generating nodes.

    Returns:
        str: A string indicating the success or failure of the operation.
    """
    with status("[cyan]Generating graph database..."):
        if self.df is not None and hasattr(self.df, "iterrows"):
            for counter, (index, row) in enumerate(self.df.iterrows()):
                row_dict = row.to_dict()
                response = self.write_query(
                    msg.cypherQuery,
                    parameters={header: row_dict[header] for header in msg.args},
                )
                # there is a possibility the generated cypher query is not correct
                # so we need to check the response before continuing to the
                # iteration
                if counter == 0 and not response.success:
                    return str(response.data)
        return "Graph database successfully generated"