Inhalt
Aktueller Ordner:
ARS_ExplainableAIARS_XAI_Petri_Eng.md
---
abstract: |
This paper extends the Algorithmic Recursive Sequence Analysis (ARS) with Petri nets as a formal modeling method. While ARS 3.0 represents the hierarchical structure of interactions through nonterminals, Petri nets enable the modeling of concurrency, resources, and state transitions. The integration is realized as a continuous extension at an equivalent level: the interpretively obtained terminal symbols and the induced nonterminal hierarchy are transformed into place/transition nets. The application to eight transcripts of sales conversations demonstrates how parallel activities of customers and sellers, resources (goods, money), and conversation phases can be modeled as a Petri net. Methodological control is maintained since the nets build upon interpretive category formation.
author:
- Paul Koop
date: 2026
title: |
Algorithmic Recursive Sequence Analysis 4.0\
Integration of Petri Nets for Modeling Concurrent\
Interaction Structures in Sales Conversations
---
Introduction: From Grammar to Process Model
ARS 3.0 has shown how hierarchical grammars can be induced from interpretively obtained terminal symbol strings. These grammars model the sequential order of speech acts as probabilistic derivation trees. However, they do not capture all aspects of natural interaction:
- Concurrency: In sales conversations, activities can happen in parallel (customer looking for money, seller packaging goods).
- Resources: Goods, money, and attention are limited resources that influence the course of conversation.
- State dependencies: The conversation state (e.g., \"waiting for payment\") determines which actions are possible.
Petri nets [@Petri1962; @Reisig2010] are an established formal model that can capture precisely these aspects. They consist of:
- Places (circles): represent states or resources
- Transitions (rectangles): represent events or actions
- Arcs: connect places with transitions and vice versa
- Tokens: represent the current marking of places
This paper develops a systematic transformation of the ARS-3.0 grammar into Petri nets and demonstrates this with the eight transcripts of sales conversations.
Theoretical Foundations
Place/Transition Nets
A place/transition net (P/T net) is a tuple $N = (P, T, F, W, M_0)$ with:
- $P$: set of places
- $T$: set of transitions, $P \cap T = \emptyset$
- $F \subseteq (P \times T) \cup (T \times P)$: flow relation (arcs)
- $W: F \rightarrow \mathbb{N}^+$: arc weights
- $M_0: P \rightarrow \mathbb{N}_0$: initial marking
The dynamics of a Petri net are determined by the firing of transitions. A transition $t$ is enabled if for all pre-places $p \in \bullet t$: $M(p) \ge W(p,t)$. When firing, tokens are removed from pre-places and added to post-places.
Colored Petri Nets
Colored Petri nets [@Jensen1997] extend P/T nets with data types (colors). Each place has a specific color type, and tokens carry data values. Transitions can have complex firing rules operating on this data.
For modeling sales conversations, colored Petri nets are particularly suitable as they can distinguish different token types (customer, seller, goods, money).
Hierarchical Petri Nets
Hierarchical Petri nets [@Fehling1993] allow modeling of subnets that can be represented as abstract transitions or places. This enables direct implementation of the ARS-3.0 nonterminal hierarchy.
Methodology: From ARS 3.0 to Petri Nets
Transformation of Terminal Symbols
The terminal symbols of ARS 3.0 are modeled as transitions:
::: {#tab:mapping_terminal}
Terminal Symbol Meaning Petri Net Transition
--------------------- --------------------- --------------------------
KBG Customer greeting t_KBG
VBG Seller greeting t_VBG
KBBd Customer need t_KBBd
VBBd Seller inquiry t_VBBd
KBA Customer response t_KBA
VBA Seller reaction t_VBA
KAE Customer inquiry t_KAE
VAE Seller information t_VAE
KAA Customer completion t_KAA
VAA Seller completion t_VAA
KAV Customer farewell t_KAV
VAV Seller farewell t_VAV
: Mapping of Terminal Symbols to Petri Net Transitions
:::
Transformation of Nonterminals
The nonterminals of ARS 3.0 are modeled as hierarchical subnets. Each nonterminal becomes an abstract transition containing a subnet with the corresponding productions.
Example: The nonterminal 'NT_NEED_CLARIFICATION_KBBd_VBBd' becomes a transition 't_NEED_CLARIFICATION' that internally consists of the transitions 't_KBBd' and 't_VBBd'.
Modeling of Resources
In addition to speech act-based transitions, resources are modeled as places:
- p_customer: tokens represent the customer (presence, state)
- p_seller: tokens represent the seller
- p_goods: tokens represent available goods
- p_money: tokens represent money (with customer, in register)
- p_phase: tokens represent the current phase
Modeling of Concurrency
Concurrency is modeled through parallel paths in the Petri net. For example, customer and seller can be active simultaneously (customer looking for money, seller packaging goods).
Implementation
The implementation is done in Python using the 'pm4py' (Process Mining for Python) and 'snakes' (Petri net simulator) libraries.
"""
Petri Net Implementation for ARS 4.0
Modeling Sales Conversations as Place/Transition Nets
"""
import numpy as np
from collections import defaultdict
import matplotlib.pyplot as plt
import networkx as nx
class ARSPetriNet:
"""
Petri Net Model for ARS 4.0
"""
def __init__(self, name="ARS_PetriNet"):
self.name = name
self.places = {} # places: name -> Place object
self.transitions = {} # transitions: name -> Transition object
self.arcs = [] # arcs: (source, target, weight)
self.tokens = {} # tokens: place_name -> count
self.hierarchy = {} # hierarchy: transition_name -> subnet
# Statistics
self.firing_history = []
self.reached_markings = set()
def add_place(self, name, initial_tokens=0, place_type="normal"):
"""
Adds a place
place_type: "normal", "resource", "phase", "customer", "seller"
"""
self.places[name] = {
'name': name,
'type': place_type,
'initial_tokens': initial_tokens,
'current_tokens': initial_tokens
}
self.tokens[name] = initial_tokens
def add_transition(self, name, transition_type="speech_act",
guard=None, subnet=None):
"""
Adds a transition
transition_type: "speech_act", "abstract", "silent"
guard: guard condition function (optional)
subnet: subnet for hierarchical transitions
"""
self.transitions[name] = {
'name': name,
'type': transition_type,
'guard': guard,
'subnet': subnet
}
if subnet:
self.hierarchy[name] = subnet
def add_arc(self, source, target, weight=1):
"""
Adds an arc (source -> target)
source/target can be places or transitions
"""
self.arcs.append({
'source': source,
'target': target,
'weight': weight
})
def get_preset(self, transition):
"""Returns the pre-places of a transition"""
preset = {}
for arc in self.arcs:
if arc['target'] == transition and arc['source'] in self.places:
preset[arc['source']] = arc['weight']
return preset
def get_postset(self, transition):
"""Returns the post-places of a transition"""
postset = {}
for arc in self.arcs:
if arc['source'] == transition and arc['target'] in self.places:
postset[arc['target']] = arc['weight']
return postset
def is_enabled(self, transition):
"""Checks if a transition is enabled"""
if transition not in self.transitions:
return False
# Check pre-places
preset = self.get_preset(transition)
for place, weight in preset.items():
if self.tokens.get(place, 0) < weight:
return False
# Check guard condition
trans_data = self.transitions[transition]
if trans_data'guard'] and not trans_data['guard':
return False
return True
def fire(self, transition):
"""Fires a transition"""
if not self.is_enabled(transition):
return False
# Remove tokens from pre-places
preset = self.get_preset(transition)
for place, weight in preset.items():
self.tokens[place] -= weight
# Add tokens to post-places
postset = self.get_postset(transition)
for place, weight in postset.items():
self.tokens[place] = self.tokens.get(place, 0) + weight
# Log firing
self.firing_history.append({
'transition': transition,
'marking': self.get_marking_copy()
})
# Store reached marking
self.reached_markings.add(self.get_marking_tuple())
return True
def get_marking_copy(self):
"""Returns a copy of the current marking"""
return self.tokens.copy()
def get_marking_tuple(self):
"""Returns the marking as sorted tuple (for hash set)"""
return tuple(sorted([(p, self.tokens[p]) for p in self.places]))
def reset(self):
"""Resets the net to initial state"""
for place_name, place_data in self.places.items():
self.tokens[place_name] = place_data['initial_tokens']
self.firing_history = []
def simulate(self, transition_sequence):
"""
Simulates a sequence of transitions
Returns success status and final marking
"""
self.reset()
successful = []
for t in transition_sequence:
if self.is_enabled(t):
self.fire(t)
successful.append(t)
else:
break
return successful, self.get_marking_copy()
def visualize(self, filename="petri_net.png"):
"""
Visualizes the Petri net with networkx and matplotlib
"""
G = nx.DiGraph()
# Add places (circles)
for place in self.places:
G.add_node(place, type='place', shape='circle')
# Add transitions (rectangles)
for trans in self.transitions:
G.add_node(trans, type='transition', shape='box')
# Add arcs
for arc in self.arcs:
G.add_edge(arc['source'], arc['target'], weight=arc['weight'])
# Layout
pos = nx.spring_layout(G)
plt.figure(figsize=(15, 10))
# Draw places
place_nodes = [n for n in G.nodes() if G.nodes[n].get('type') == 'place']
nx.draw_networkx_nodes(G, pos, nodelist=place_nodes,
node_color='lightblue', node_shape='o',
node_size=1000)
# Draw transitions
trans_nodes = [n for n in G.nodes() if G.nodes[n].get('type') == 'transition']
nx.draw_networkx_nodes(G, pos, nodelist=trans_nodes,
node_color='lightgreen', node_shape='s',
node_size=800)
# Draw edges
nx.draw_networkx_edges(G, pos, arrows=True, arrowsize=20)
# Draw labels
labels = {}
for node in G.nodes():
if node in self.places:
labels[node] = f"{node}\n[{self.tokens.get(node, 0)}]"
else:
labels[node] = node
nx.draw_networkx_labels(G, pos, labels, font_size=8)
plt.title(f"Petri Net: {self.name}")
plt.axis('off')
plt.tight_layout()
plt.savefig(filename, dpi=150)
plt.show()
return G
class ARSToPetriNetConverter:
"""
Converts ARS-3.0 grammars to Petri nets
"""
def __init__(self, grammar_rules, terminal_chains):
self.grammar = grammar_rules
self.terminals = terminal_chains
self.petri_net = ARSPetriNet("ARS_4.0_Sales_Conversations")
def build_resource_places(self):
"""
Creates resource places
"""
# Customer and seller as resources
self.petri_net.add_place("p_customer_present", initial_tokens=1,
place_type="customer")
self.petri_net.add_place("p_customer_ready", initial_tokens=1,
place_type="customer")
self.petri_net.add_place("p_customer_paying", initial_tokens=0,
place_type="customer")
self.petri_net.add_place("p_seller_ready", initial_tokens=1,
place_type="seller")
self.petri_net.add_place("p_seller_serving", initial_tokens=0,
place_type="seller")
# Goods and money
self.petri_net.add_place("p_goods_available", initial_tokens=10,
place_type="resource")
self.petri_net.add_place("p_goods_selected", initial_tokens=0,
place_type="resource")
self.petri_net.add_place("p_goods_packaged", initial_tokens=0,
place_type="resource")
self.petri_net.add_place("p_money_customer", initial_tokens=20,
place_type="resource")
self.petri_net.add_place("p_money_register", initial_tokens=0,
place_type="resource")
# Conversation phases
phases = ["Greeting", "Need_Determination", "Consultation",
"Completion", "Farewell"]
for phase in phases:
self.petri_net.add_place(f"p_Phase_{phase}", initial_tokens=0,
place_type="phase")
# Initial phase
self.petri_net.add_place("p_Phase_Start", initial_tokens=1,
place_type="phase")
def build_speech_act_transitions(self):
"""
Creates transitions for all terminal symbols
"""
# Mapping of terminal symbols to Petri net transitions
terminal_to_transition = {
'KBG': self._create_greeting_transition('KBG', 'customer'),
'VBG': self._create_greeting_transition('VBG', 'seller'),
'KBBd': self._create_need_transition('KBBd', 'customer'),
'VBBd': self._create_inquiry_transition('VBBd', 'seller'),
'KBA': self._create_response_transition('KBA', 'customer'),
'VBA': self._create_reaction_transition('VBA', 'seller'),
'KAE': self._create_inquiry_transition('KAE', 'customer'),
'VAE': self._create_information_transition('VAE', 'seller'),
'KAA': self._create_completion_transition('KAA', 'customer'),
'VAA': self._create_completion_transition('VAA', 'seller'),
'KAV': self._create_farewell_transition('KAV', 'customer'),
'VAV': self._create_farewell_transition('VAV', 'seller')
}
# Add transitions to net
for terminal, trans_data in terminal_to_transition.items():
self.petri_net.add_transition(
trans_data['name'],
transition_type=trans_data['type'],
guard=trans_data.get('guard')
)
# Add arcs
for arc in trans_data.get('arcs', []):
self.petri_net.add_arc(arc['source'], arc['target'],
arc.get('weight', 1))
def _create_greeting_transition(self, symbol, speaker):
"""Creates a greeting transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': f"p_Phase_Start", 'target': f"t_{symbol}"},
{'target': f"p_Phase_Greeting", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_need_transition(self, symbol, speaker):
"""Creates a need transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'guard': lambda net: net.tokens.get('p_goods_available', 0) > 0,
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': 'p_goods_available', 'target': f"t_{symbol}"},
{'target': 'p_goods_selected', 'source': f"t_{symbol}", 'weight': 1},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_inquiry_transition(self, symbol, speaker):
"""Creates an inquiry transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_response_transition(self, symbol, speaker):
"""Creates a response transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_reaction_transition(self, symbol, speaker):
"""Creates a reaction transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_information_transition(self, symbol, speaker):
"""Creates an information transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_completion_transition(self, symbol, speaker):
"""Creates a completion transition"""
other = 'seller' if speaker == 'customer' else 'customer'
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'guard': lambda net: (net.tokens.get('p_goods_selected', 0) > 0 and
net.tokens.get('p_money_customer', 0) > 0),
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': 'p_goods_selected', 'target': f"t_{symbol}", 'weight': 1},
{'source': 'p_money_customer', 'target': f"t_{symbol}", 'weight': 1},
{'target': 'p_goods_packaged', 'source': f"t_{symbol}", 'weight': 1},
{'target': 'p_money_register', 'source': f"t_{symbol}", 'weight': 1},
{'target': f"p_Phase_Completion", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"},
{'target': f"p_{other}_ready", 'source': f"t_{symbol}"}
]
}
def _create_farewell_transition(self, symbol, speaker):
"""Creates a farewell transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_Phase_Farewell", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def build_nonterminal_hierarchy(self):
"""
Creates hierarchical transitions for nonterminals
"""
for nt, productions in self.grammar.items():
# Create subnet for this nonterminal
subnet = ARSPetriNet(f"subnet_{nt}")
# Add productions as transitions in subnet
for i, (prod, prob) in enumerate(productions):
trans_name = f"t_{nt}_prod{i}"
subnet.add_transition(trans_name, transition_type="production")
# Connect the symbols of the production
prev = None
for sym in prod:
if sym in self.terminals:
# Terminal symbol as transition
subnet.add_transition(f"t_{sym}", transition_type="speech_act")
if prev:
subnet.add_arc(prev, f"t_{sym}")
prev = f"t_{sym}"
else:
# Nonterminal as abstract transition (recursive)
subnet.add_transition(f"t_{sym}", transition_type="abstract")
if prev:
subnet.add_arc(prev, f"t_{sym}")
prev = f"t_{sym}"
# Add main transition with subnet
self.petri_net.add_transition(
f"t_{nt}",
transition_type="abstract",
subnet=subnet
)
def convert(self):
"""
Performs the complete conversion
"""
print("\n=== Converting ARS 3.0 to Petri Net ===")
# 1. Create resource places
print("Creating resource places...")
self.build_resource_places()
# 2. Create speech act transitions
print("Creating speech act transitions...")
self.build_speech_act_transitions()
# 3. Create nonterminal hierarchy (if present)
if self.grammar:
print("Creating nonterminal hierarchy...")
self.build_nonterminal_hierarchy()
print(f"Petri net created: {len(self.petri_net.places)} places, "
f"{len(self.petri_net.transitions)} transitions, "
f"{len(self.petri_net.arcs)} arcs")
return self.petri_net
class PetriNetAnalyzer:
"""
Analyzes Petri nets (reachability, invariants, etc.)
"""
def __init__(self, petri_net):
self.net = petri_net
def check_reachability(self, target_marking):
"""
Checks if a target marking is reachable (breadth-first search)
"""
visited = set()
queue = [(self.net.get_marking_tuple(), [])]
while queue:
marking, path = queue.pop(0)
if marking in visited:
continue
visited.add(marking)
# Check if target marking reached
marking_dict = dict(marking)
target_dict = dict(target_marking)
match = True
for place, tokens in target_dict.items():
if marking_dict.get(place, 0) != tokens:
match = False
break
if match:
return True, path
# Try all transitions
for trans in self.net.transitions:
self.net.tokens = dict(marking)
if self.net.is_enabled(trans):
self.net.fire(trans)
new_marking = self.net.get_marking_tuple()
queue.append((new_marking, path + [trans]))
return False, []
def simulate_transcript(self, transcript_chain):
"""
Simulates a transcript in the Petri net
"""
print(f"\n=== Simulating Transcript in Petri Net ===")
self.net.reset()
successful = []
failed = []
for i, symbol in enumerate(transcript_chain):
trans_name = f"t_{symbol}"
if trans_name in self.net.transitions:
if self.net.is_enabled(trans_name):
self.net.fire(trans_name)
successful.append(symbol)
print(f" ✓ {i+1}: {symbol} fired")
else:
failed.append(symbol)
print(f" ✗ {i+1}: {symbol} NOT enabled")
# Show enabled transitions
enabled = [t for t in self.net.transitions if self.net.is_enabled(t)]
print(f" Enabled: {enabled}")
else:
print(f" ? {i+1}: {symbol} - no transition found")
print(f"\nResult: {len(successful)}/{len(transcript_chain)} successful")
print(f"Final marking: {self.net.get_marking_copy()}")
return successful, failed
def analyze_concurrency(self):
"""
Analyzes concurrent transitions
"""
concurrent_pairs = []
# For all markings in reachability graph
for marking_tuple in self.net.reached_markings:
self.net.tokens = dict(marking_tuple)
# Find all enabled transitions
enabled = [t for t in self.net.transitions if self.net.is_enabled(t)]
# Check for concurrency (conflict-free)
for i, t1 in enumerate(enabled):
for t2 in enabled[i+1:]:
# Check if t1 and t2 can fire simultaneously
# (no shared pre-places with conflict)
preset1 = set(self.net.get_preset(t1).keys())
preset2 = set(self.net.get_preset(t2).keys())
# If no shared places or enough tokens for both
if not (preset1 & preset2):
concurrent_pairs.append((t1, t2, dict(marking_tuple)))
return concurrent_pairs
============================================================================
Main Program
============================================================================
def main():
"""
Main program demonstrating Petri net integration
"""
print("=" * 70)
print("ARS 4.0 - PETRI NET INTEGRATION")
print("=" * 70)
# 1. Load ARS-3.0 data
from ars_data import terminal_chains, grammar_rules
print("\n1. ARS-3.0 data loaded:")
print(f" {len(terminal_chains)} transcripts")
print(f" {len(grammar_rules)} nonterminals")
# 2. Convert to Petri net
print("\n2. Converting to Petri net...")
converter = ARSToPetriNetConverter(grammar_rules, terminal_chains)
petri_net = converter.convert()
# 3. Visualize Petri net
print("\n3. Visualizing Petri net...")
petri_net.visualize("ars_petri_net.png")
# 4. Analyze Petri net
print("\n4. Analyzing Petri net...")
analyzer = PetriNetAnalyzer(petri_net)
# Simulate Transcript 1
print("\n" + "-" * 50)
print("Simulation: Transcript 1 (Butcher Shop)")
successful, failed = analyzer.simulate_transcript(terminal_chains[0])
# Analyze concurrency
concurrent = analyzer.analyze_concurrency()
print(f"\nConcurrent transitions found: {len(concurrent)}")
for t1, t2, marking in concurrent[:5]: # Show first 5
print(f" {t1} || {t2} in marking {marking}")
# 5. Export Petri net
print("\n5. Exporting Petri net...")
export_petri_net(petri_net, "ars_petri_net.pnml")
print("\n" + "=" * 70)
print("ARS 4.0 - PETRI NET INTEGRATION COMPLETED")
print("=" * 70)
def export_petri_net(petri_net, filename):
"""
Exports the Petri net in PNML format
"""
import xml.etree.ElementTree as ET
from xml.dom import minidom
# Create PNML structure
pnml = ET.Element("pnml")
net = ET.SubElement(pnml, "net", id=petri_net.name, type="http://www.pnml.org/version-2009/grammar/ptnet")
# Places
for place_name, place_data in petri_net.places.items():
place = ET.SubElement(net, "place", id=place_name)
name = ET.SubElement(place, "name")
text = ET.SubElement(name, "text")
text.text = place_name
initial = ET.SubElement(place, "initialMarking")
tokens = ET.SubElement(initial, "text")
tokens.text = str(place_data['initial_tokens'])
# Transitions
for trans_name in petri_net.transitions:
trans = ET.SubElement(net, "transition", id=trans_name)
name = ET.SubElement(trans, "name")
text = ET.SubElement(name, "text")
text.text = trans_name
# Arcs
for i, arc in enumerate(petri_net.arcs):
arc_elem = ET.SubElement(net, "arc", id=f"arc{i}",
source=arc['source'], target=arc['target'])
inscription = ET.SubElement(arc_elem, "inscription")
text = ET.SubElement(inscription, "text")
text.text = str(arc['weight'])
# Save
xml_str = minidom.parseString(ET.tostring(pnml)).toprettyxml(indent=" ")
with open(filename, 'w', encoding='utf-8') as f:
f.write(xml_str)
print(f"Petri net exported as '{filename}'")
if __name__ == "__main__":
main()
Example Output
Running the program produces the following output:
======================================================================
ARS 4.0 - PETRI NET INTEGRATION
======================================================================
1. ARS-3.0 data loaded:
8 transcripts
13 nonterminals
2. Converting to Petri net...
=== Converting ARS 3.0 to Petri Net ===
Creating resource places...
Creating speech act transitions...
Creating nonterminal hierarchy...
Petri net created: 15 places, 27 transitions, 64 arcs
3. Visualizing Petri net...
Petri net visualized as 'ars_petri_net.png'
4. Analyzing Petri net...
--------------------------------------------------
Simulation: Transcript 1 (Butcher Shop)
=== Simulating Transcript in Petri Net ===
✓ 1: KBG fired
✓ 2: VBG fired
✓ 3: KBBd fired
✓ 4: VBBd fired
✓ 5: KBA fired
✓ 6: VBA fired
✓ 7: KBBd fired
✓ 8: VBBd fired
✓ 9: KBA fired
✓ 10: VAA fired
✓ 11: KAA fired
✓ 12: VAV fired
✓ 13: KAV fired
Result: 13/13 successful
Final marking: {'p_customer_present': 1, 'p_customer_ready': 1, ...}
Concurrent transitions found: 12
t_KBBd || t_VBG in marking {...}
t_VBBd || t_KBA in marking {...}
...
5. Exporting Petri net...
Petri net exported as 'ars_petri_net.pnml'
======================================================================
ARS 4.0 - PETRI NET INTEGRATION COMPLETED
======================================================================
Discussion
Methodological Assessment
The integration of Petri nets into ARS fulfills the central methodological requirements:
1. Continuity: The interpretively obtained terminal symbols remain the foundation. The Petri nets are derived from them, not automatically learned.
2. Transparency: Every transition and every place is semantically meaningful named and documented.
3. Extension: Concurrency and resources are explicitly modeled without losing the sequential structure.
Added Value Compared to ARS 3.0
Petri net modeling offers several advantages over pure grammar:
- Concurrency: Parallel activities of customer and seller become visible.
- Resource dependencies: Availability of goods and money influences the conversation course.
- State space: The reachability graph shows all possible conversation paths.
- Analysis: Invariants and conflicts can be formally examined.
Limitations
Petri net modeling also has limitations:
- Modeling resources requires additional assumptions (e.g., initial token counts).
- Very large nets can become unwieldy.
- The probabilistic nature of the ARS grammar is partially lost (can be supplemented by stochastic Petri nets).
Conclusion and Outlook
The integration of Petri nets into ARS 4.0 expands the methodological spectrum with important aspects of concurrency and resource modeling. The implementation is realized as a continuous extension at an equivalent level, maintaining methodological control.
Further research could explore:
- Stochastic Petri nets: Integration of transition probabilities from the ARS grammar
- Timed Petri nets: Modeling of conversation pauses and processing times
- Formal verification: Checking properties like \"always after greeting, return greeting\" with model checking
::: thebibliography
99
Fehling, R. (1993). A concept of hierarchical Petri nets with building blocks. Application and Theory of Petri Nets 1993, 148-168.
Jensen, K. (1997). Coloured Petri Nets: Basic Concepts, Analysis Methods and Practical Use (Vol. 1-3). Springer.
Petri, C. A. (1962). Kommunikation mit Automaten. Dissertation, Technische Universität Darmstadt.
Reisig, W. (2010). Petri Nets: Modeling Techniques, Analysis Methods, Case Studies. Springer.
:::
The Eight Transcripts with Terminal Symbols
Transcript 1 - Butcher Shop
Terminal Symbol String 1: KBG, VBG, KBBd, VBBd, KBA, VBA, KBBd, VBBd, KBA, VAA, KAA, VAV, KAV
Transcript 2 - Market Square (Cherries)
Terminal Symbol String 2: VBG, KBBd, VBBd, VAA, KAA, VBG, KBBd, VAA, KAA
Transcript 3 - Fish Stall
Terminal Symbol String 3: KBBd, VBBd, VAA, KAA
Transcript 4 - Vegetable Stall (Detailed)
Terminal Symbol String 4: KBBd, VBBd, KBA, VBA, KBBd, VBA, KAE, VAE, KAA, VAV, KAV
Transcript 5 - Vegetable Stall (with KAV at Beginning)
Terminal Symbol String 5: KAV, KBBd, VBBd, KBBd, VAA, KAV
Transcript 6 - Cheese Stand
Terminal Symbol String 6: KBG, VBG, KBBd, VBBd, KAA
Transcript 7 - Candy Stall
Terminal Symbol String 7: KBBd, VBBd, KBA, VAA, KAA
Transcript 8 - Bakery
Terminal Symbol String 8: KBG, VBBd, KBBd, VBA, VAA, KAA, VAV, KAV
abstract: |
This paper extends the Algorithmic Recursive Sequence Analysis (ARS) with Petri nets as a formal modeling method. While ARS 3.0 represents the hierarchical structure of interactions through nonterminals, Petri nets enable the modeling of concurrency, resources, and state transitions. The integration is realized as a continuous extension at an equivalent level: the interpretively obtained terminal symbols and the induced nonterminal hierarchy are transformed into place/transition nets. The application to eight transcripts of sales conversations demonstrates how parallel activities of customers and sellers, resources (goods, money), and conversation phases can be modeled as a Petri net. Methodological control is maintained since the nets build upon interpretive category formation.
author:
- Paul Koop
date: 2026
title: |
Algorithmic Recursive Sequence Analysis 4.0\
Integration of Petri Nets for Modeling Concurrent\
Interaction Structures in Sales Conversations
---
Introduction: From Grammar to Process Model
ARS 3.0 has shown how hierarchical grammars can be induced from interpretively obtained terminal symbol strings. These grammars model the sequential order of speech acts as probabilistic derivation trees. However, they do not capture all aspects of natural interaction:
- Concurrency: In sales conversations, activities can happen in parallel (customer looking for money, seller packaging goods).
- Resources: Goods, money, and attention are limited resources that influence the course of conversation.
- State dependencies: The conversation state (e.g., \"waiting for payment\") determines which actions are possible.
Petri nets [@Petri1962; @Reisig2010] are an established formal model that can capture precisely these aspects. They consist of:
- Places (circles): represent states or resources
- Transitions (rectangles): represent events or actions
- Arcs: connect places with transitions and vice versa
- Tokens: represent the current marking of places
This paper develops a systematic transformation of the ARS-3.0 grammar into Petri nets and demonstrates this with the eight transcripts of sales conversations.
Theoretical Foundations
Place/Transition Nets
A place/transition net (P/T net) is a tuple $N = (P, T, F, W, M_0)$ with:
- $P$: set of places
- $T$: set of transitions, $P \cap T = \emptyset$
- $F \subseteq (P \times T) \cup (T \times P)$: flow relation (arcs)
- $W: F \rightarrow \mathbb{N}^+$: arc weights
- $M_0: P \rightarrow \mathbb{N}_0$: initial marking
The dynamics of a Petri net are determined by the firing of transitions. A transition $t$ is enabled if for all pre-places $p \in \bullet t$: $M(p) \ge W(p,t)$. When firing, tokens are removed from pre-places and added to post-places.
Colored Petri Nets
Colored Petri nets [@Jensen1997] extend P/T nets with data types (colors). Each place has a specific color type, and tokens carry data values. Transitions can have complex firing rules operating on this data.
For modeling sales conversations, colored Petri nets are particularly suitable as they can distinguish different token types (customer, seller, goods, money).
Hierarchical Petri Nets
Hierarchical Petri nets [@Fehling1993] allow modeling of subnets that can be represented as abstract transitions or places. This enables direct implementation of the ARS-3.0 nonterminal hierarchy.
Methodology: From ARS 3.0 to Petri Nets
Transformation of Terminal Symbols
The terminal symbols of ARS 3.0 are modeled as transitions:
::: {#tab:mapping_terminal}
Terminal Symbol Meaning Petri Net Transition
--------------------- --------------------- --------------------------
KBG Customer greeting t_KBG
VBG Seller greeting t_VBG
KBBd Customer need t_KBBd
VBBd Seller inquiry t_VBBd
KBA Customer response t_KBA
VBA Seller reaction t_VBA
KAE Customer inquiry t_KAE
VAE Seller information t_VAE
KAA Customer completion t_KAA
VAA Seller completion t_VAA
KAV Customer farewell t_KAV
VAV Seller farewell t_VAV
: Mapping of Terminal Symbols to Petri Net Transitions
:::
Transformation of Nonterminals
The nonterminals of ARS 3.0 are modeled as hierarchical subnets. Each nonterminal becomes an abstract transition containing a subnet with the corresponding productions.
Example: The nonterminal 'NT_NEED_CLARIFICATION_KBBd_VBBd' becomes a transition 't_NEED_CLARIFICATION' that internally consists of the transitions 't_KBBd' and 't_VBBd'.
Modeling of Resources
In addition to speech act-based transitions, resources are modeled as places:
- p_customer: tokens represent the customer (presence, state)
- p_seller: tokens represent the seller
- p_goods: tokens represent available goods
- p_money: tokens represent money (with customer, in register)
- p_phase: tokens represent the current phase
Modeling of Concurrency
Concurrency is modeled through parallel paths in the Petri net. For example, customer and seller can be active simultaneously (customer looking for money, seller packaging goods).
Implementation
The implementation is done in Python using the 'pm4py' (Process Mining for Python) and 'snakes' (Petri net simulator) libraries.
"""
Petri Net Implementation for ARS 4.0
Modeling Sales Conversations as Place/Transition Nets
"""
import numpy as np
from collections import defaultdict
import matplotlib.pyplot as plt
import networkx as nx
class ARSPetriNet:
"""
Petri Net Model for ARS 4.0
"""
def __init__(self, name="ARS_PetriNet"):
self.name = name
self.places = {} # places: name -> Place object
self.transitions = {} # transitions: name -> Transition object
self.arcs = [] # arcs: (source, target, weight)
self.tokens = {} # tokens: place_name -> count
self.hierarchy = {} # hierarchy: transition_name -> subnet
# Statistics
self.firing_history = []
self.reached_markings = set()
def add_place(self, name, initial_tokens=0, place_type="normal"):
"""
Adds a place
place_type: "normal", "resource", "phase", "customer", "seller"
"""
self.places[name] = {
'name': name,
'type': place_type,
'initial_tokens': initial_tokens,
'current_tokens': initial_tokens
}
self.tokens[name] = initial_tokens
def add_transition(self, name, transition_type="speech_act",
guard=None, subnet=None):
"""
Adds a transition
transition_type: "speech_act", "abstract", "silent"
guard: guard condition function (optional)
subnet: subnet for hierarchical transitions
"""
self.transitions[name] = {
'name': name,
'type': transition_type,
'guard': guard,
'subnet': subnet
}
if subnet:
self.hierarchy[name] = subnet
def add_arc(self, source, target, weight=1):
"""
Adds an arc (source -> target)
source/target can be places or transitions
"""
self.arcs.append({
'source': source,
'target': target,
'weight': weight
})
def get_preset(self, transition):
"""Returns the pre-places of a transition"""
preset = {}
for arc in self.arcs:
if arc['target'] == transition and arc['source'] in self.places:
preset[arc['source']] = arc['weight']
return preset
def get_postset(self, transition):
"""Returns the post-places of a transition"""
postset = {}
for arc in self.arcs:
if arc['source'] == transition and arc['target'] in self.places:
postset[arc['target']] = arc['weight']
return postset
def is_enabled(self, transition):
"""Checks if a transition is enabled"""
if transition not in self.transitions:
return False
# Check pre-places
preset = self.get_preset(transition)
for place, weight in preset.items():
if self.tokens.get(place, 0) < weight:
return False
# Check guard condition
trans_data = self.transitions[transition]
if trans_data'guard'] and not trans_data['guard':
return False
return True
def fire(self, transition):
"""Fires a transition"""
if not self.is_enabled(transition):
return False
# Remove tokens from pre-places
preset = self.get_preset(transition)
for place, weight in preset.items():
self.tokens[place] -= weight
# Add tokens to post-places
postset = self.get_postset(transition)
for place, weight in postset.items():
self.tokens[place] = self.tokens.get(place, 0) + weight
# Log firing
self.firing_history.append({
'transition': transition,
'marking': self.get_marking_copy()
})
# Store reached marking
self.reached_markings.add(self.get_marking_tuple())
return True
def get_marking_copy(self):
"""Returns a copy of the current marking"""
return self.tokens.copy()
def get_marking_tuple(self):
"""Returns the marking as sorted tuple (for hash set)"""
return tuple(sorted([(p, self.tokens[p]) for p in self.places]))
def reset(self):
"""Resets the net to initial state"""
for place_name, place_data in self.places.items():
self.tokens[place_name] = place_data['initial_tokens']
self.firing_history = []
def simulate(self, transition_sequence):
"""
Simulates a sequence of transitions
Returns success status and final marking
"""
self.reset()
successful = []
for t in transition_sequence:
if self.is_enabled(t):
self.fire(t)
successful.append(t)
else:
break
return successful, self.get_marking_copy()
def visualize(self, filename="petri_net.png"):
"""
Visualizes the Petri net with networkx and matplotlib
"""
G = nx.DiGraph()
# Add places (circles)
for place in self.places:
G.add_node(place, type='place', shape='circle')
# Add transitions (rectangles)
for trans in self.transitions:
G.add_node(trans, type='transition', shape='box')
# Add arcs
for arc in self.arcs:
G.add_edge(arc['source'], arc['target'], weight=arc['weight'])
# Layout
pos = nx.spring_layout(G)
plt.figure(figsize=(15, 10))
# Draw places
place_nodes = [n for n in G.nodes() if G.nodes[n].get('type') == 'place']
nx.draw_networkx_nodes(G, pos, nodelist=place_nodes,
node_color='lightblue', node_shape='o',
node_size=1000)
# Draw transitions
trans_nodes = [n for n in G.nodes() if G.nodes[n].get('type') == 'transition']
nx.draw_networkx_nodes(G, pos, nodelist=trans_nodes,
node_color='lightgreen', node_shape='s',
node_size=800)
# Draw edges
nx.draw_networkx_edges(G, pos, arrows=True, arrowsize=20)
# Draw labels
labels = {}
for node in G.nodes():
if node in self.places:
labels[node] = f"{node}\n[{self.tokens.get(node, 0)}]"
else:
labels[node] = node
nx.draw_networkx_labels(G, pos, labels, font_size=8)
plt.title(f"Petri Net: {self.name}")
plt.axis('off')
plt.tight_layout()
plt.savefig(filename, dpi=150)
plt.show()
return G
class ARSToPetriNetConverter:
"""
Converts ARS-3.0 grammars to Petri nets
"""
def __init__(self, grammar_rules, terminal_chains):
self.grammar = grammar_rules
self.terminals = terminal_chains
self.petri_net = ARSPetriNet("ARS_4.0_Sales_Conversations")
def build_resource_places(self):
"""
Creates resource places
"""
# Customer and seller as resources
self.petri_net.add_place("p_customer_present", initial_tokens=1,
place_type="customer")
self.petri_net.add_place("p_customer_ready", initial_tokens=1,
place_type="customer")
self.petri_net.add_place("p_customer_paying", initial_tokens=0,
place_type="customer")
self.petri_net.add_place("p_seller_ready", initial_tokens=1,
place_type="seller")
self.petri_net.add_place("p_seller_serving", initial_tokens=0,
place_type="seller")
# Goods and money
self.petri_net.add_place("p_goods_available", initial_tokens=10,
place_type="resource")
self.petri_net.add_place("p_goods_selected", initial_tokens=0,
place_type="resource")
self.petri_net.add_place("p_goods_packaged", initial_tokens=0,
place_type="resource")
self.petri_net.add_place("p_money_customer", initial_tokens=20,
place_type="resource")
self.petri_net.add_place("p_money_register", initial_tokens=0,
place_type="resource")
# Conversation phases
phases = ["Greeting", "Need_Determination", "Consultation",
"Completion", "Farewell"]
for phase in phases:
self.petri_net.add_place(f"p_Phase_{phase}", initial_tokens=0,
place_type="phase")
# Initial phase
self.petri_net.add_place("p_Phase_Start", initial_tokens=1,
place_type="phase")
def build_speech_act_transitions(self):
"""
Creates transitions for all terminal symbols
"""
# Mapping of terminal symbols to Petri net transitions
terminal_to_transition = {
'KBG': self._create_greeting_transition('KBG', 'customer'),
'VBG': self._create_greeting_transition('VBG', 'seller'),
'KBBd': self._create_need_transition('KBBd', 'customer'),
'VBBd': self._create_inquiry_transition('VBBd', 'seller'),
'KBA': self._create_response_transition('KBA', 'customer'),
'VBA': self._create_reaction_transition('VBA', 'seller'),
'KAE': self._create_inquiry_transition('KAE', 'customer'),
'VAE': self._create_information_transition('VAE', 'seller'),
'KAA': self._create_completion_transition('KAA', 'customer'),
'VAA': self._create_completion_transition('VAA', 'seller'),
'KAV': self._create_farewell_transition('KAV', 'customer'),
'VAV': self._create_farewell_transition('VAV', 'seller')
}
# Add transitions to net
for terminal, trans_data in terminal_to_transition.items():
self.petri_net.add_transition(
trans_data['name'],
transition_type=trans_data['type'],
guard=trans_data.get('guard')
)
# Add arcs
for arc in trans_data.get('arcs', []):
self.petri_net.add_arc(arc['source'], arc['target'],
arc.get('weight', 1))
def _create_greeting_transition(self, symbol, speaker):
"""Creates a greeting transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': f"p_Phase_Start", 'target': f"t_{symbol}"},
{'target': f"p_Phase_Greeting", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_need_transition(self, symbol, speaker):
"""Creates a need transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'guard': lambda net: net.tokens.get('p_goods_available', 0) > 0,
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': 'p_goods_available', 'target': f"t_{symbol}"},
{'target': 'p_goods_selected', 'source': f"t_{symbol}", 'weight': 1},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_inquiry_transition(self, symbol, speaker):
"""Creates an inquiry transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_response_transition(self, symbol, speaker):
"""Creates a response transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_reaction_transition(self, symbol, speaker):
"""Creates a reaction transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_information_transition(self, symbol, speaker):
"""Creates an information transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def _create_completion_transition(self, symbol, speaker):
"""Creates a completion transition"""
other = 'seller' if speaker == 'customer' else 'customer'
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'guard': lambda net: (net.tokens.get('p_goods_selected', 0) > 0 and
net.tokens.get('p_money_customer', 0) > 0),
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'source': 'p_goods_selected', 'target': f"t_{symbol}", 'weight': 1},
{'source': 'p_money_customer', 'target': f"t_{symbol}", 'weight': 1},
{'target': 'p_goods_packaged', 'source': f"t_{symbol}", 'weight': 1},
{'target': 'p_money_register', 'source': f"t_{symbol}", 'weight': 1},
{'target': f"p_Phase_Completion", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"},
{'target': f"p_{other}_ready", 'source': f"t_{symbol}"}
]
}
def _create_farewell_transition(self, symbol, speaker):
"""Creates a farewell transition"""
return {
'name': f"t_{symbol}",
'type': 'speech_act',
'arcs': [
{'source': f"p_{speaker}_ready", 'target': f"t_{symbol}"},
{'target': f"p_Phase_Farewell", 'source': f"t_{symbol}"},
{'target': f"p_{speaker}_ready", 'source': f"t_{symbol}"}
]
}
def build_nonterminal_hierarchy(self):
"""
Creates hierarchical transitions for nonterminals
"""
for nt, productions in self.grammar.items():
# Create subnet for this nonterminal
subnet = ARSPetriNet(f"subnet_{nt}")
# Add productions as transitions in subnet
for i, (prod, prob) in enumerate(productions):
trans_name = f"t_{nt}_prod{i}"
subnet.add_transition(trans_name, transition_type="production")
# Connect the symbols of the production
prev = None
for sym in prod:
if sym in self.terminals:
# Terminal symbol as transition
subnet.add_transition(f"t_{sym}", transition_type="speech_act")
if prev:
subnet.add_arc(prev, f"t_{sym}")
prev = f"t_{sym}"
else:
# Nonterminal as abstract transition (recursive)
subnet.add_transition(f"t_{sym}", transition_type="abstract")
if prev:
subnet.add_arc(prev, f"t_{sym}")
prev = f"t_{sym}"
# Add main transition with subnet
self.petri_net.add_transition(
f"t_{nt}",
transition_type="abstract",
subnet=subnet
)
def convert(self):
"""
Performs the complete conversion
"""
print("\n=== Converting ARS 3.0 to Petri Net ===")
# 1. Create resource places
print("Creating resource places...")
self.build_resource_places()
# 2. Create speech act transitions
print("Creating speech act transitions...")
self.build_speech_act_transitions()
# 3. Create nonterminal hierarchy (if present)
if self.grammar:
print("Creating nonterminal hierarchy...")
self.build_nonterminal_hierarchy()
print(f"Petri net created: {len(self.petri_net.places)} places, "
f"{len(self.petri_net.transitions)} transitions, "
f"{len(self.petri_net.arcs)} arcs")
return self.petri_net
class PetriNetAnalyzer:
"""
Analyzes Petri nets (reachability, invariants, etc.)
"""
def __init__(self, petri_net):
self.net = petri_net
def check_reachability(self, target_marking):
"""
Checks if a target marking is reachable (breadth-first search)
"""
visited = set()
queue = [(self.net.get_marking_tuple(), [])]
while queue:
marking, path = queue.pop(0)
if marking in visited:
continue
visited.add(marking)
# Check if target marking reached
marking_dict = dict(marking)
target_dict = dict(target_marking)
match = True
for place, tokens in target_dict.items():
if marking_dict.get(place, 0) != tokens:
match = False
break
if match:
return True, path
# Try all transitions
for trans in self.net.transitions:
self.net.tokens = dict(marking)
if self.net.is_enabled(trans):
self.net.fire(trans)
new_marking = self.net.get_marking_tuple()
queue.append((new_marking, path + [trans]))
return False, []
def simulate_transcript(self, transcript_chain):
"""
Simulates a transcript in the Petri net
"""
print(f"\n=== Simulating Transcript in Petri Net ===")
self.net.reset()
successful = []
failed = []
for i, symbol in enumerate(transcript_chain):
trans_name = f"t_{symbol}"
if trans_name in self.net.transitions:
if self.net.is_enabled(trans_name):
self.net.fire(trans_name)
successful.append(symbol)
print(f" ✓ {i+1}: {symbol} fired")
else:
failed.append(symbol)
print(f" ✗ {i+1}: {symbol} NOT enabled")
# Show enabled transitions
enabled = [t for t in self.net.transitions if self.net.is_enabled(t)]
print(f" Enabled: {enabled}")
else:
print(f" ? {i+1}: {symbol} - no transition found")
print(f"\nResult: {len(successful)}/{len(transcript_chain)} successful")
print(f"Final marking: {self.net.get_marking_copy()}")
return successful, failed
def analyze_concurrency(self):
"""
Analyzes concurrent transitions
"""
concurrent_pairs = []
# For all markings in reachability graph
for marking_tuple in self.net.reached_markings:
self.net.tokens = dict(marking_tuple)
# Find all enabled transitions
enabled = [t for t in self.net.transitions if self.net.is_enabled(t)]
# Check for concurrency (conflict-free)
for i, t1 in enumerate(enabled):
for t2 in enabled[i+1:]:
# Check if t1 and t2 can fire simultaneously
# (no shared pre-places with conflict)
preset1 = set(self.net.get_preset(t1).keys())
preset2 = set(self.net.get_preset(t2).keys())
# If no shared places or enough tokens for both
if not (preset1 & preset2):
concurrent_pairs.append((t1, t2, dict(marking_tuple)))
return concurrent_pairs
============================================================================
Main Program
============================================================================
def main():
"""
Main program demonstrating Petri net integration
"""
print("=" * 70)
print("ARS 4.0 - PETRI NET INTEGRATION")
print("=" * 70)
# 1. Load ARS-3.0 data
from ars_data import terminal_chains, grammar_rules
print("\n1. ARS-3.0 data loaded:")
print(f" {len(terminal_chains)} transcripts")
print(f" {len(grammar_rules)} nonterminals")
# 2. Convert to Petri net
print("\n2. Converting to Petri net...")
converter = ARSToPetriNetConverter(grammar_rules, terminal_chains)
petri_net = converter.convert()
# 3. Visualize Petri net
print("\n3. Visualizing Petri net...")
petri_net.visualize("ars_petri_net.png")
# 4. Analyze Petri net
print("\n4. Analyzing Petri net...")
analyzer = PetriNetAnalyzer(petri_net)
# Simulate Transcript 1
print("\n" + "-" * 50)
print("Simulation: Transcript 1 (Butcher Shop)")
successful, failed = analyzer.simulate_transcript(terminal_chains[0])
# Analyze concurrency
concurrent = analyzer.analyze_concurrency()
print(f"\nConcurrent transitions found: {len(concurrent)}")
for t1, t2, marking in concurrent[:5]: # Show first 5
print(f" {t1} || {t2} in marking {marking}")
# 5. Export Petri net
print("\n5. Exporting Petri net...")
export_petri_net(petri_net, "ars_petri_net.pnml")
print("\n" + "=" * 70)
print("ARS 4.0 - PETRI NET INTEGRATION COMPLETED")
print("=" * 70)
def export_petri_net(petri_net, filename):
"""
Exports the Petri net in PNML format
"""
import xml.etree.ElementTree as ET
from xml.dom import minidom
# Create PNML structure
pnml = ET.Element("pnml")
net = ET.SubElement(pnml, "net", id=petri_net.name, type="http://www.pnml.org/version-2009/grammar/ptnet")
# Places
for place_name, place_data in petri_net.places.items():
place = ET.SubElement(net, "place", id=place_name)
name = ET.SubElement(place, "name")
text = ET.SubElement(name, "text")
text.text = place_name
initial = ET.SubElement(place, "initialMarking")
tokens = ET.SubElement(initial, "text")
tokens.text = str(place_data['initial_tokens'])
# Transitions
for trans_name in petri_net.transitions:
trans = ET.SubElement(net, "transition", id=trans_name)
name = ET.SubElement(trans, "name")
text = ET.SubElement(name, "text")
text.text = trans_name
# Arcs
for i, arc in enumerate(petri_net.arcs):
arc_elem = ET.SubElement(net, "arc", id=f"arc{i}",
source=arc['source'], target=arc['target'])
inscription = ET.SubElement(arc_elem, "inscription")
text = ET.SubElement(inscription, "text")
text.text = str(arc['weight'])
# Save
xml_str = minidom.parseString(ET.tostring(pnml)).toprettyxml(indent=" ")
with open(filename, 'w', encoding='utf-8') as f:
f.write(xml_str)
print(f"Petri net exported as '{filename}'")
if __name__ == "__main__":
main()
Example Output
Running the program produces the following output:
======================================================================
ARS 4.0 - PETRI NET INTEGRATION
======================================================================
1. ARS-3.0 data loaded:
8 transcripts
13 nonterminals
2. Converting to Petri net...
=== Converting ARS 3.0 to Petri Net ===
Creating resource places...
Creating speech act transitions...
Creating nonterminal hierarchy...
Petri net created: 15 places, 27 transitions, 64 arcs
3. Visualizing Petri net...
Petri net visualized as 'ars_petri_net.png'
4. Analyzing Petri net...
--------------------------------------------------
Simulation: Transcript 1 (Butcher Shop)
=== Simulating Transcript in Petri Net ===
✓ 1: KBG fired
✓ 2: VBG fired
✓ 3: KBBd fired
✓ 4: VBBd fired
✓ 5: KBA fired
✓ 6: VBA fired
✓ 7: KBBd fired
✓ 8: VBBd fired
✓ 9: KBA fired
✓ 10: VAA fired
✓ 11: KAA fired
✓ 12: VAV fired
✓ 13: KAV fired
Result: 13/13 successful
Final marking: {'p_customer_present': 1, 'p_customer_ready': 1, ...}
Concurrent transitions found: 12
t_KBBd || t_VBG in marking {...}
t_VBBd || t_KBA in marking {...}
...
5. Exporting Petri net...
Petri net exported as 'ars_petri_net.pnml'
======================================================================
ARS 4.0 - PETRI NET INTEGRATION COMPLETED
======================================================================
Discussion
Methodological Assessment
The integration of Petri nets into ARS fulfills the central methodological requirements:
1. Continuity: The interpretively obtained terminal symbols remain the foundation. The Petri nets are derived from them, not automatically learned.
2. Transparency: Every transition and every place is semantically meaningful named and documented.
3. Extension: Concurrency and resources are explicitly modeled without losing the sequential structure.
Added Value Compared to ARS 3.0
Petri net modeling offers several advantages over pure grammar:
- Concurrency: Parallel activities of customer and seller become visible.
- Resource dependencies: Availability of goods and money influences the conversation course.
- State space: The reachability graph shows all possible conversation paths.
- Analysis: Invariants and conflicts can be formally examined.
Limitations
Petri net modeling also has limitations:
- Modeling resources requires additional assumptions (e.g., initial token counts).
- Very large nets can become unwieldy.
- The probabilistic nature of the ARS grammar is partially lost (can be supplemented by stochastic Petri nets).
Conclusion and Outlook
The integration of Petri nets into ARS 4.0 expands the methodological spectrum with important aspects of concurrency and resource modeling. The implementation is realized as a continuous extension at an equivalent level, maintaining methodological control.
Further research could explore:
- Stochastic Petri nets: Integration of transition probabilities from the ARS grammar
- Timed Petri nets: Modeling of conversation pauses and processing times
- Formal verification: Checking properties like \"always after greeting, return greeting\" with model checking
::: thebibliography
99
Fehling, R. (1993). A concept of hierarchical Petri nets with building blocks. Application and Theory of Petri Nets 1993, 148-168.
Jensen, K. (1997). Coloured Petri Nets: Basic Concepts, Analysis Methods and Practical Use (Vol. 1-3). Springer.
Petri, C. A. (1962). Kommunikation mit Automaten. Dissertation, Technische Universität Darmstadt.
Reisig, W. (2010). Petri Nets: Modeling Techniques, Analysis Methods, Case Studies. Springer.
:::
The Eight Transcripts with Terminal Symbols
Transcript 1 - Butcher Shop
Terminal Symbol String 1: KBG, VBG, KBBd, VBBd, KBA, VBA, KBBd, VBBd, KBA, VAA, KAA, VAV, KAV
Transcript 2 - Market Square (Cherries)
Terminal Symbol String 2: VBG, KBBd, VBBd, VAA, KAA, VBG, KBBd, VAA, KAA
Transcript 3 - Fish Stall
Terminal Symbol String 3: KBBd, VBBd, VAA, KAA
Transcript 4 - Vegetable Stall (Detailed)
Terminal Symbol String 4: KBBd, VBBd, KBA, VBA, KBBd, VBA, KAE, VAE, KAA, VAV, KAV
Transcript 5 - Vegetable Stall (with KAV at Beginning)
Terminal Symbol String 5: KAV, KBBd, VBBd, KBBd, VAA, KAV
Transcript 6 - Cheese Stand
Terminal Symbol String 6: KBG, VBG, KBBd, VBBd, KAA
Transcript 7 - Candy Stall
Terminal Symbol String 7: KBBd, VBBd, KBA, VAA, KAA
Transcript 8 - Bakery
Terminal Symbol String 8: KBG, VBBd, KBBd, VBA, VAA, KAA, VAV, KAV