pm4py.llm module#

pm4py.llm.openai_query(prompt: str, api_key: str | None = None, openai_model: str | None = None, api_url: str | None = None, **kwargs) → str[source]#

Executes the provided prompt, obtaining the answer from the OpenAI APIs.

Parameters:

  • prompt – The prompt to be executed.

  • api_key – (Optional) OpenAI API key.

  • openai_model – (Optional) OpenAI model to be used (default: “gpt-3.5-turbo”).

  • api_url – (Optional) OpenAI API URL.

  • **kwargs –

    Additional parameters to pass to the OpenAI API.

Returns:

The response from the OpenAI API as a string.

import pm4py

resp = pm4py.llm.openai_query('What is the result of 3+3?', api_key="sk-382393", openai_model="gpt-3.5-turbo")
print(resp)
pm4py.llm.google_query(prompt: str, api_key: str | None = None, model: str | None = None, **kwargs) → str[source]#

Executes the provided prompt, obtaining the answer from the Google APIs.

Parameters:

  • prompt – prompt that should be executed

  • api_key – API key

  • model – Model to be used (default: gemini-1.5-flash-002)

Return type:

str

import pm4py

resp = pm4py.llm.google_query('what is the result of 3+3?', api_key="sk-382393", model="gemini-1.5-flash-002")
print(resp)
pm4py.llm.anthropic_query(prompt: str, api_key: str | None = None, model: str | None = None, **kwargs) → str[source]#

Executes the provided prompt, obtaining the answer from the Google APIs.

Parameters:

  • prompt – prompt that should be executed

  • api_key – API key

  • model – Model to be used (default: claude-3-5-sonnet-20241022)

Return type:

str

import pm4py

resp = pm4py.llm.anthropic_query('what is the result of 3+3?', api_key="sk-382393", model="claude-3-5-sonnet-20241022")
print(resp)
pm4py.llm.abstract_dfg(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_performance: bool = True, relative_frequency: bool = False, response_header: bool = True, primary_performance_aggregation: str = 'mean', secondary_performance_aggregation: str | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the DFG (Directly-Follows Graph) abstraction of a traditional event log.

Parameters:

  • log_obj – The log object to abstract.

  • max_len – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).

  • include_performance – Whether to include the performance of the paths in the abstraction.

  • relative_frequency – Whether to use relative instead of absolute frequency of the paths.

  • response_header – Whether to include a short header before the paths, describing the abstraction.

  • primary_performance_aggregation – Primary aggregation method for the arc’s performance (default: “mean”). Other options: “median”, “min”, “max”, “sum”, “stdev”.

  • secondary_performance_aggregation – (Optional) Secondary aggregation method for the arc’s performance (default: None). Other options: “mean”, “median”, “min”, “max”, “sum”, “stdev”.

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

  • case_id_key – The column name to be used as case identifier.

Returns:

The DFG abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_dfg(log))
pm4py.llm.abstract_variants(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_performance: bool = True, relative_frequency: bool = False, response_header: bool = True, primary_performance_aggregation: str = 'mean', secondary_performance_aggregation: str | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the variants abstraction of a traditional event log.

Parameters:

  • log_obj – The log object to abstract.

  • max_len – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).

  • include_performance – Whether to include the performance of the variants in the abstraction.

  • relative_frequency – Whether to use relative instead of absolute frequency of the variants.

  • response_header – Whether to include a short header before the variants, describing the abstraction.

  • primary_performance_aggregation – Primary aggregation method for the variants’ performance (default: “mean”). Other options: “median”, “min”, “max”, “sum”, “stdev”.

  • secondary_performance_aggregation – (Optional) Secondary aggregation method for the variants’ performance (default: None). Other options: “mean”, “median”, “min”, “max”, “sum”, “stdev”.

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

  • case_id_key – The column name to be used as case identifier.

Returns:

The variants abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_variants(log))
pm4py.llm.abstract_ocel(ocel: OCEL, include_timestamps: bool = True) → str[source]#

Obtains the abstraction of an object-centric event log, including the list of events and the objects of the OCEL.

Parameters:

  • ocel – The object-centric event log to abstract.

  • include_timestamps – Whether to include timestamp information in the abstraction.

Returns:

The OCEL abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel(ocel))
pm4py.llm.abstract_ocel_ocdfg(ocel: OCEL, include_header: bool = True, include_timestamps: bool = True, max_len: int = 10000) → str[source]#

Obtains the abstraction of an object-centric event log, representing the object-centric directly-follows graph in text.

Parameters:

  • ocel – The object-centric event log to abstract.

  • include_header – Whether to include a header in the abstraction.

  • include_timestamps – Whether to include timestamp information in the abstraction.

  • max_len – Maximum length of the abstraction (default: constants.OPENAI_MAX_LEN).

Returns:

The object-centric DFG abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel_ocdfg(ocel))
pm4py.llm.abstract_ocel_features(ocel: OCEL, obj_type: str, include_header: bool = True, max_len: int = 10000, debug: bool = False, enable_object_lifecycle_paths: bool = True) → str[source]#

Obtains the abstraction of an object-centric event log, representing the features and their values in text.

Parameters:

  • ocel – The object-centric event log to abstract.

  • obj_type – The object type to consider in feature extraction.

  • include_header – Whether to include a header in the abstraction.

  • max_len – Maximum length of the abstraction (default: constants.OPENAI_MAX_LEN).

  • debug – Enables debugging mode, providing insights into feature extraction steps.

  • enable_object_lifecycle_paths – Enables the “lifecycle paths” feature in the abstraction.

Returns:

The OCEL features abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel_features(ocel, obj_type="Resource"))
pm4py.llm.abstract_event_stream(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, response_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the event stream abstraction of a traditional event log.

Parameters:

  • log_obj – The log object to abstract.

  • max_len – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).

  • response_header – Whether to include a short header before the event stream, describing the abstraction.

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

  • case_id_key – The column name to be used as case identifier.

Returns:

The event stream abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_event_stream(log))
pm4py.llm.abstract_petri_net(net: PetriNet, im: Marking, fm: Marking, response_header: bool = True) → str[source]#

Obtains an abstraction of a Petri net.

Parameters:

  • net – The Petri net to abstract.

  • im – The initial marking of the Petri net.

  • fm – The final marking of the Petri net.

  • response_header – Whether to include a header in the abstraction.

Returns:

The Petri net abstraction as a string.

import pm4py

net, im, fm = pm4py.read_pnml('tests/input_data/running-example.pnml')
print(pm4py.llm.abstract_petri_net(net, im, fm))
pm4py.llm.abstract_log_attributes(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Abstracts the attributes of a log by reporting their names, types, and top values.

Parameters:

  • log_obj – The log object whose attributes are to be abstracted.

  • max_len – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

  • case_id_key – The column name to be used as case identifier.

Returns:

The log attributes abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_log_attributes(log))
pm4py.llm.abstract_log_features(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Abstracts the machine learning features obtained from a log by reporting the top features until the desired length is achieved.

Parameters:

  • log_obj – The log object from which to extract features.

  • max_len – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).

  • include_header – Whether to include a header in the abstraction.

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

  • case_id_key – The column name to be used as case identifier.

Returns:

The log features abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_log_features(log))
pm4py.llm.abstract_temporal_profile(temporal_profile: Dict[Tuple[str, str], Tuple[float, float]], include_header: bool = True) → str[source]#

Abstracts a temporal profile model into a descriptive string.

Parameters:

  • temporal_profile – The temporal profile model to abstract.

  • include_header – Whether to include a header in the abstraction describing the temporal profile.

Returns:

The temporal profile abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
temporal_profile = pm4py.discover_temporal_profile(log)
text_abstr = pm4py.llm.abstract_temporal_profile(temporal_profile, include_header=True)
print(text_abstr)
pm4py.llm.abstract_case(case: Trace, include_case_attributes: bool = True, include_event_attributes: bool = True, include_timestamp: bool = True, include_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp') → str[source]#

Textually abstracts a single case from an event log.

Parameters:

  • case – The case object to abstract.

  • include_case_attributes – Whether to include attributes at the case level.

  • include_event_attributes – Whether to include attributes at the event level.

  • include_timestamp – Whether to include event timestamps in the abstraction.

  • include_header – Whether to include a header in the abstraction.

  • activity_key – The column name to be used as activity.

  • timestamp_key – The column name to be used as timestamp.

Returns:

The case abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
print(pm4py.llm.abstract_case(log[0]))
pm4py.llm.abstract_declare(declare_model, include_header: bool = True) → str[source]#

Textually abstracts a DECLARE model.

Parameters:

  • declare_model – The DECLARE model to abstract.

  • include_header – Whether to include a header in the abstraction.

Returns:

The DECLARE model abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
log_ske = pm4py.discover_declare(log)
print(pm4py.llm.abstract_declare(log_ske))
pm4py.llm.abstract_log_skeleton(log_skeleton, include_header: bool = True) → str[source]#

Textually abstracts a log skeleton process model.

Parameters:

  • log_skeleton – The log skeleton to abstract.

  • include_header – Whether to include a header in the abstraction.

Returns:

The log skeleton abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
log_ske = pm4py.discover_log_skeleton(log)
print(pm4py.llm.abstract_log_skeleton(log_ske))
pm4py.llm.nlp_to_log_query(log: DataFrame, query: str, executor=<function openai_query>, obtain_query: bool = True, execute_query: bool = True, **kwargs) → str | DataFrame[source]#

Translates a natural language statement into a database (SQL) query executable against the event log.

Parameters:

  • log – event log object

  • query – query expressed in natural language

  • executor – the connector to the LLM (e.g., pm4py.llm.openai_query)

  • obtain_query – executes the prompt, to transform the natural statements into a database (SQL) query

  • execute_query – executes the database (SQL) query against the event data

  • kwargs – additional keyword arguments to the method

Return type:

Union[str, pd.DataFrame]

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
resp = pm4py.llm.nlp_to_log_query(log, "How many cases are contained in the event log?", api_key="sk-5HNn")
print(resp)
pm4py.llm.nlp_to_log_filter(log: DataFrame, filter_query: str, executor=<function openai_query>, obtain_query: bool = True, execute_query: bool = True, **kwargs) → str | DataFrame[source]#

Translates a filtering query expressed in natural language into a database (SQL) query that is used to filter the event log.

Parameters:

  • log – event log object

  • filter_query – filtering query expressed in natural language

  • executor – the connector to the LLM (e.g., pm4py.llm.openai_query)

  • obtain_query – executes the prompt, to transform the natural statements into a database (SQL) query

  • execute_query – executes the database (SQL) query against the event data

  • kwargs – additional keyword arguments to the method

Return type:

Union[str, pd.DataFrame]

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
resp = pm4py.llm.nlp_to_log_filter(log, "There is an event with activity: pay compensation", api_key="sk-5HNn")
print(resp)
pm4py.llm.automated_hypotheses_formulation(dataframe: DataFrame, executor=<function openai_query>, obtain_query: bool = True, execute_query: bool = True, max_len=10000, **kwargs) → str | List[Tuple[str, str, str | None]][source]#

Automatically formulate some hypotheses on the event data. The result of this method is either: - The prompt (to be executed manually against the LLM) when obtain_query=False - A list of different hypotheses. Each hypothesis comes with a description (position 0), the SQL query (position 1), and (if execute_query=True) the result of the execution of the query in position 2.

Parameters:

  • dataframe – event log object

  • executor – the connector to the LLM (e.g., pm4py.llm.openai_query)

  • obtain_query – executes the prompt and get the hypotheses along with the corresponding SQL queries

  • execute_query – executes the obtained SQL queries and stores the results

  • kwargs – additional keyword arguments to the method

  • max_len – maximum length of the obtained prompt

Return type:

Union[str, List[Tuple[str, str, Union[str, None]]]]

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
result = pm4py.llm.automated_hypotheses_formulation(log, api_key="sk-5HN")
print(result)
pm4py.llm.explain_visualization(vis_saver, *args, connector=<function openai_query>, **kwargs) → str[source]#

Explains a process mining visualization using LLMs by saving it as a .png image and providing the image to the Large Language Model along with a description.

Parameters:

  • vis_saver – The visualizer function used to save the visualization to disk.

  • args – Positional arguments required by the visualizer function.

  • connector – (Optional) The connector method to communicate with the large language model (default: openai_query).

  • **kwargs –

    Additional keyword arguments for the visualizer function or the connector (e.g., annotations, API key).

Returns:

The explanation of the visualization as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
descr = pm4py.llm.explain_visualization(pm4py.save_vis_dotted_chart, log, api_key="sk-5HN", show_legend=False)
print(descr)