Introduction to Building LLM-based Agents
Creating an LLM-based agent for FSM code generation requires careful consideration of how to structure the interaction between the language model, the application logic, and the user interface. Unlike simple chatbots that rely primarily on the LLM's training, specialized agents need custom logic to handle domain-specific tasks, maintain conversation context, and ensure reliable outputs.
The challenge in building such an agent lies in combining the natural language understanding capabilities of LLMs with structured programming logic. The agent must interpret user requirements, ask clarifying questions when needed, and generate precise technical outputs. This requires a hybrid approach where the LLM handles language understanding while custom code manages the technical validation and generation processes.
Setting Up the LLM Foundation and API Integration
The foundation of any LLM-based agent starts with selecting and integrating the appropriate language model. For FSM generation, the model needs strong capabilities in both natural language understanding and code generation. Modern options include OpenAI's GPT models, Anthropic's Claude, or open-source alternatives like Llama or CodeLlama.
The integration layer must handle API communication, token management, and response parsing. This component serves as the bridge between your agent's logic and the LLM's capabilities. Proper error handling and retry mechanisms are essential since API calls can fail or return unexpected responses.
Here's a detailed implementation of the LLM integration layer that demonstrates how to structure the communication with the language model
import openai
import json
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
@dataclass
class LLMResponse:
content: str
usage: Dict[str, int]
model: str
finish_reason: str
class LLMIntegration:
def __init__(self, api_key: str, model: str = "gpt-4"):
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.max_retries = 3
self.retry_delay = 1.0
def generate_response(self, messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 1000) -> LLMResponse:
"""
Generate a response from the LLM with proper error handling and retries.
This method encapsulates all the complexity of API communication.
"""
for attempt in range(self.max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
return LLMResponse(
content=response.choices[0].message.content,
usage=dict(response.usage),
model=response.model,
finish_reason=response.choices[0].finish_reason
)
except openai.RateLimitError:
if attempt < self.max_retries - 1:
time.sleep(self.retry_delay * (2 ** attempt))
continue
raise
except openai.APIError as e:
if attempt < self.max_retries - 1:
time.sleep(self.retry_delay)
continue
raise
raise Exception("Max retries exceeded")
def parse_structured_response(self, response: str) -> Dict[str, Any]:
"""
Parse structured responses from the LLM, handling cases where
the model returns JSON or other structured formats.
"""
try:
# Try to extract JSON from the response
json_start = response.find('{')
json_end = response.rfind('}') + 1
if json_start != -1 and json_end > json_start:
json_str = response[json_start:json_end]
return json.loads(json_str)
else:
# If no JSON found, return the raw response
return {"raw_response": response}
except json.JSONDecodeError:
return {"raw_response": response, "parse_error": True}
This integration layer provides a robust foundation for communicating with the LLM. The retry mechanism handles temporary API failures, while the structured response parsing allows the agent to work with both natural language and JSON responses from the model.
Designing the Agent Architecture and Control Flow
The agent architecture determines how different components interact and how control flows through the system. A well-designed architecture separates concerns while maintaining clear communication paths between components. The main challenge is balancing flexibility with reliability.
The core architecture should include a conversation manager that maintains the dialogue state, a requirements processor that extracts technical information from user input, a clarification engine that identifies missing information, and a generation engine that produces the final output.
The control flow typically follows a pattern where user input is processed, requirements are extracted and validated, clarifications are requested if needed, and finally the FSM code is generated. However, this flow must be flexible enough to handle iterative refinement and user corrections.
Here's an implementation of the main agent controller that orchestrates the entire process:
from enum import Enum
from typing import Dict, List, Optional, Tuple
class AgentState(Enum):
INITIAL = "initial"
GATHERING_REQUIREMENTS = "gathering_requirements"
CLARIFYING = "clarifying"
GENERATING = "generating"
COMPLETED = "completed"
ERROR = "error"
class FSMAgentController:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.state = AgentState.INITIAL
self.conversation_history = []
self.extracted_requirements = {}
self.clarification_queue = []
self.generated_fsm = None
def process_user_input(self, user_input: str) -> str:
"""
Main entry point for processing user input. This method
determines the appropriate action based on the current agent state
and the nature of the user's input.
"""
self.conversation_history.append({"role": "user", "content": user_input})
try:
if self.state == AgentState.INITIAL:
return self._handle_initial_input(user_input)
elif self.state == AgentState.GATHERING_REQUIREMENTS:
return self._handle_requirements_gathering(user_input)
elif self.state == AgentState.CLARIFYING:
return self._handle_clarification_response(user_input)
elif self.state == AgentState.GENERATING:
return self._handle_generation_request(user_input)
elif self.state == AgentState.COMPLETED:
return self._handle_post_completion_input(user_input)
else:
return "I'm sorry, I encountered an error. Let's start over."
except Exception as e:
self.state = AgentState.ERROR
return f"An error occurred: {str(e)}. Please try again."
def _handle_initial_input(self, user_input: str) -> str:
"""
Handle the first user input where they describe their FSM requirements.
This sets up the initial requirements extraction process.
"""
# Create a specialized prompt for requirements extraction
system_prompt = """You are an expert in finite state machines.
Analyze the user's description and extract potential states, events,
and transitions. Respond in JSON format with your analysis."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
]
response = self.llm.generate_response(messages, temperature=0.3)
analysis = self.llm.parse_structured_response(response.content)
self.extracted_requirements = analysis
self.state = AgentState.GATHERING_REQUIREMENTS
return self._generate_requirements_summary()
def _handle_requirements_gathering(self, user_input: str) -> str:
"""
Continue gathering and refining requirements based on user feedback.
This state allows for iterative improvement of the requirements.
"""
# Update requirements based on additional user input
update_prompt = f"""
Current requirements: {json.dumps(self.extracted_requirements)}
User additional input: {user_input}
Update the requirements based on the new information. Return updated JSON.
"""
messages = [
{"role": "system", "content": update_prompt}
]
response = self.llm.generate_response(messages, temperature=0.3)
updated_requirements = self.llm.parse_structured_response(response.content)
self.extracted_requirements.update(updated_requirements)
# Check if we need clarifications
clarifications = self._identify_clarifications_needed()
if clarifications:
self.clarification_queue = clarifications
self.state = AgentState.CLARIFYING
return self._ask_next_clarification()
else:
self.state = AgentState.GENERATING
return "I have enough information to generate your FSM. What programming language would you like me to use?"
def _identify_clarifications_needed(self) -> List[str]:
"""
Analyze the current requirements to identify what clarifications
are needed before proceeding with FSM generation.
"""
clarification_prompt = f"""
Analyze these FSM requirements and identify what clarifications are needed:
{json.dumps(self.extracted_requirements)}
Return a list of specific questions that need to be answered to create
a complete FSM specification. Focus on missing states, undefined transitions,
or ambiguous behaviors.
"""
messages = [
{"role": "system", "content": clarification_prompt}
]
response = self.llm.generate_response(messages, temperature=0.3)
try:
clarifications_data = self.llm.parse_structured_response(response.content)
return clarifications_data.get("clarifications", [])
except:
# If parsing fails, extract questions from the raw response
return self._extract_questions_from_text(response.content)
def _extract_questions_from_text(self, text: str) -> List[str]:
"""
Fallback method to extract questions from unstructured text
when the LLM doesn't return properly formatted JSON.
"""
questions = []
lines = text.split('\n')
for line in lines:
line = line.strip()
if line.endswith('?') and len(line) > 10:
questions.append(line)
return questions
This controller implementation demonstrates how to manage the complex state transitions and decision-making required for an intelligent agent. The state machine approach ensures that the agent behaves predictably while remaining flexible enough to handle various user interaction patterns.
Implementing Conversation Management and Context Handling
Effective conversation management is crucial for maintaining context throughout the interaction. The agent must remember previous exchanges, track the evolution of requirements, and maintain coherent dialogue even when users provide incomplete or contradictory information.
Context handling involves more than just storing conversation history. The agent must understand which parts of the conversation are still relevant, how new information relates to previous statements, and when to ask for clarification versus making reasonable assumptions.
The conversation manager must also handle interruptions and topic changes gracefully. Users might want to modify earlier requirements, ask questions about the process, or completely change direction mid-conversation.
Here's a detailed implementation of the conversation management system:
from datetime import datetime
from typing import Dict, List, Optional, Any
class ConversationContext:
def __init__(self):
self.messages = []
self.requirements_evolution = []
self.clarifications_asked = []
self.clarifications_answered = []
self.user_preferences = {}
self.session_metadata = {
"start_time": datetime.now(),
"interaction_count": 0
}
def add_message(self, role: str, content: str, metadata: Optional[Dict] = None):
"""
Add a message to the conversation history with optional metadata
for tracking different types of interactions.
"""
message = {
"role": role,
"content": content,
"timestamp": datetime.now(),
"metadata": metadata or {}
}
self.messages.append(message)
self.session_metadata["interaction_count"] += 1
def get_relevant_context(self, max_tokens: int = 2000) -> List[Dict]:
"""
Extract the most relevant parts of the conversation for the current
context window. This is crucial for managing long conversations
that exceed the LLM's context limit.
"""
# Start with the most recent messages
relevant_messages = []
token_count = 0
for message in reversed(self.messages):
# Rough token estimation (4 characters per token)
message_tokens = len(message["content"]) // 4
if token_count + message_tokens > max_tokens:
break
relevant_messages.insert(0, {
"role": message["role"],
"content": message["content"]
})
token_count += message_tokens
return relevant_messages
def track_requirements_evolution(self, requirements: Dict):
"""
Track how requirements change over time to understand
the user's evolving understanding of their needs.
"""
evolution_entry = {
"timestamp": datetime.now(),
"requirements": requirements.copy(),
"interaction_number": self.session_metadata["interaction_count"]
}
self.requirements_evolution.append(evolution_entry)
def get_unanswered_clarifications(self) -> List[str]:
"""
Identify clarifications that have been asked but not yet answered.
This helps avoid repeating questions and ensures all issues are resolved.
"""
asked_set = set(self.clarifications_asked)
answered_set = set(q["question"] for q in self.clarifications_answered)
return list(asked_set - answered_set)
class ConversationManager:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.context = ConversationContext()
def process_message(self, user_input: str, current_state: AgentState) -> Tuple[str, Dict]:
"""
Process a user message within the current conversation context.
Returns the response and any extracted information.
"""
self.context.add_message("user", user_input)
# Analyze the user input for intent and content
analysis = self._analyze_user_intent(user_input, current_state)
# Generate appropriate response based on analysis
response = self._generate_contextual_response(analysis, current_state)
self.context.add_message("assistant", response,
metadata={"analysis": analysis})
return response, analysis
def _analyze_user_intent(self, user_input: str, current_state: AgentState) -> Dict:
"""
Analyze what the user is trying to accomplish with their input.
This goes beyond simple keyword matching to understand intent.
"""
context_messages = self.context.get_relevant_context()
analysis_prompt = f"""
Current conversation state: {current_state.value}
Recent conversation context: {json.dumps(context_messages[-3:] if context_messages else [])}
User input: {user_input}
Analyze the user's intent and extract relevant information. Consider:
1. Are they providing new requirements?
2. Are they answering a clarification question?
3. Are they asking for modifications to previous requirements?
4. Are they asking questions about the process?
5. Are they ready to proceed to the next step?
Return your analysis in JSON format with intent classification and extracted information.
"""
messages = [{"role": "system", "content": analysis_prompt}]
response = self.llm.generate_response(messages, temperature=0.3)
try:
return self.llm.parse_structured_response(response.content)
except:
# Fallback to basic analysis if structured parsing fails
return {
"intent": "unknown",
"content": user_input,
"confidence": "low"
}
def _generate_contextual_response(self, analysis: Dict, current_state: AgentState) -> str:
"""
Generate a response that takes into account the conversation history,
current state, and the user's apparent intent.
"""
context_messages = self.context.get_relevant_context()
response_prompt = f"""
You are an FSM generation agent. Generate an appropriate response based on:
Current state: {current_state.value}
User intent analysis: {json.dumps(analysis)}
Conversation context: {json.dumps(context_messages)}
Unanswered clarifications: {self.context.get_unanswered_clarifications()}
Generate a helpful, contextual response that moves the conversation forward
appropriately. Be specific and actionable in your response.
"""
messages = [{"role": "system", "content": response_prompt}]
response = self.llm.generate_response(messages, temperature=0.7)
return response.content
def summarize_conversation(self) -> str:
"""
Generate a summary of the conversation for reference or handoff.
This is useful for debugging and understanding user interactions.
"""
summary_prompt = f"""
Summarize this conversation about FSM generation:
{json.dumps(self.context.messages)}
Include:
- What the user wants to build
- Key requirements identified
- Clarifications asked and answered
- Current status of the project
"""
messages = [{"role": "system", "content": summary_prompt}]
response = self.llm.generate_response(messages, temperature=0.5)
return response.content
This conversation management system provides the foundation for maintaining coherent, context-aware interactions. The system tracks not just what was said, but the evolution of understanding throughout the conversation.
Building the Requirements Analysis Pipeline
The requirements analysis pipeline transforms natural language descriptions into structured information that can be used for FSM generation. This pipeline must handle the ambiguity and incompleteness inherent in natural language while extracting precise technical specifications.
The pipeline typically involves several stages. First, linguistic analysis identifies key entities and relationships in the user's description. Then, domain-specific analysis maps these entities to FSM concepts like states, events, and transitions. Finally, validation checks ensure the extracted requirements form a coherent and implementable specification.
Each stage of the pipeline must be robust enough to handle various input styles while flexible enough to accommodate different domains and use cases. The pipeline should also provide feedback about confidence levels and identify areas where additional information is needed.
Here's a comprehensive implementation of the requirements analysis pipeline:
import re
from typing import Dict, List, Set, Tuple, Optional
from dataclasses import dataclass, field
@dataclass
class FSMElement:
name: str
type: str # 'state', 'event', 'action', 'transition'
confidence: float
source_text: str
metadata: Dict = field(default_factory=dict)
@dataclass
class FSMSpecification:
states: List[FSMElement] = field(default_factory=list)
events: List[FSMElement] = field(default_factory=list)
actions: List[FSMElement] = field(default_factory=list)
transitions: List[FSMElement] = field(default_factory=list)
initial_state: Optional[str] = None
final_states: List[str] = field(default_factory=list)
class RequirementsAnalyzer:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.domain_patterns = self._initialize_domain_patterns()
self.confidence_threshold = 0.7
def _initialize_domain_patterns(self) -> Dict[str, List[str]]:
"""
Initialize patterns for recognizing FSM elements in different domains.
These patterns help with initial entity recognition before LLM analysis.
"""
return {
"state_indicators": [
r"(?:is|are|becomes?|remains?)\s+(\w+)",
r"(?:in|during)\s+(?:the\s+)?(\w+)\s+(?:state|mode|phase)",
r"when\s+(?:the\s+)?system\s+is\s+(\w+)",
r"(?:the\s+)?(\w+)\s+state"
],
"event_indicators": [
r"when\s+(?:the\s+)?(\w+)\s+(?:occurs?|happens?|is\s+triggered)",
r"(?:on|upon)\s+(\w+)",
r"if\s+(?:the\s+)?(\w+)\s+(?:is\s+)?(?:pressed|activated|detected)",
r"after\s+(?:the\s+)?(\w+)\s+(?:expires?|completes?|finishes?)"
],
"action_indicators": [
r"(?:turn\s+on|activate|start|begin)\s+(?:the\s+)?(\w+)",
r"(?:send|transmit|output)\s+(?:a\s+)?(\w+)",
r"(?:set|update|change)\s+(?:the\s+)?(\w+)",
r"(?:execute|perform|run)\s+(?:the\s+)?(\w+)"
]
}
def analyze_requirements(self, description: str) -> FSMSpecification:
"""
Main method for analyzing user requirements and extracting FSM elements.
This combines pattern-based extraction with LLM-powered analysis.
"""
# Step 1: Initial pattern-based extraction
initial_elements = self._extract_with_patterns(description)
# Step 2: LLM-powered deep analysis
llm_analysis = self._llm_analyze_requirements(description, initial_elements)
# Step 3: Combine and validate results
specification = self._combine_and_validate(initial_elements, llm_analysis)
# Step 4: Identify relationships and transitions
specification = self._identify_transitions(specification, description)
return specification
def _extract_with_patterns(self, description: str) -> Dict[str, List[FSMElement]]:
"""
Use regex patterns to identify potential FSM elements.
This provides a baseline that the LLM can refine and improve.
"""
elements = {"states": [], "events": [], "actions": []}
for element_type, patterns in self.domain_patterns.items():
element_category = element_type.split('_')[0] + 's' # states, events, actions
for pattern in patterns:
matches = re.finditer(pattern, description, re.IGNORECASE)
for match in matches:
element = FSMElement(
name=match.group(1).lower(),
type=element_category[:-1], # Remove 's'
confidence=0.6, # Pattern-based confidence
source_text=match.group(0),
metadata={"extraction_method": "pattern", "pattern": pattern}
)
elements[element_category].append(element)
return elements
def _llm_analyze_requirements(self, description: str,
initial_elements: Dict) -> Dict:
"""
Use the LLM to perform deep analysis of the requirements,
building on the initial pattern-based extraction.
"""
analysis_prompt = f"""
Analyze this FSM requirements description and extract detailed information:
Description: {description}
Initial pattern-based extraction: {json.dumps({k: [e.__dict__ for e in v] for k, v in initial_elements.items()})}
Provide a comprehensive analysis including:
1. All states the system can be in
2. All events that can trigger state changes
3. All actions that should be performed
4. Potential transitions between states
5. Initial state identification
6. Final or terminal states
7. Confidence assessment for each element
Return your analysis in JSON format with detailed explanations.
"""
messages = [{"role": "system", "content": analysis_prompt}]
response = self.llm.generate_response(messages, temperature=0.3)
return self.llm.parse_structured_response(response.content)
def _combine_and_validate(self, pattern_elements: Dict,
llm_analysis: Dict) -> FSMSpecification:
"""
Combine pattern-based and LLM-based extractions, resolving conflicts
and creating a unified specification.
"""
spec = FSMSpecification()
# Process LLM-identified elements
for element_type in ['states', 'events', 'actions']:
llm_elements = llm_analysis.get(element_type, [])
pattern_elements_list = pattern_elements.get(element_type, [])
# Create a map of pattern elements for easy lookup
pattern_map = {elem.name: elem for elem in pattern_elements_list}
for llm_elem in llm_elements:
if isinstance(llm_elem, dict):
name = llm_elem.get('name', '').lower()
confidence = llm_elem.get('confidence', 0.8)
# Check if this element was also found by patterns
if name in pattern_map:
# Combine information, giving higher confidence
confidence = min(confidence + 0.2, 1.0)
source_text = pattern_map[name].source_text
else:
source_text = llm_elem.get('source_text', '')
element = FSMElement(
name=name,
type=element_type[:-1], # Remove 's'
confidence=confidence,
source_text=source_text,
metadata={
"extraction_method": "combined",
"llm_analysis": llm_elem
}
)
getattr(spec, element_type).append(element)
# Set initial state if identified
initial_state_info = llm_analysis.get('initial_state')
if initial_state_info:
spec.initial_state = initial_state_info.get('name')
# Set final states if identified
final_states_info = llm_analysis.get('final_states', [])
spec.final_states = [fs.get('name') for fs in final_states_info if isinstance(fs, dict)]
return spec
def _identify_transitions(self, spec: FSMSpecification,
description: str) -> FSMSpecification:
"""
Identify state transitions based on the extracted states and events.
This is one of the most complex parts of the analysis.
"""
transition_prompt = f"""
Based on this FSM specification and the original description,
identify all state transitions:
States: {[s.name for s in spec.states]}
Events: {[e.name for e in spec.events]}
Actions: {[a.name for a in spec.actions]}
Original description: {description}
For each transition, specify:
- Source state
- Target state
- Triggering event
- Associated action (if any)
- Conditions (if any)
Return as a JSON list of transition objects.
"""
messages = [{"role": "system", "content": transition_prompt}]
response = self.llm.generate_response(messages, temperature=0.3)
transitions_data = self.llm.parse_structured_response(response.content)
if 'transitions' in transitions_data:
for trans_data in transitions_data['transitions']:
transition = FSMElement(
name=f"{trans_data.get('source_state', 'unknown')}_to_{trans_data.get('target_state', 'unknown')}",
type='transition',
confidence=trans_data.get('confidence', 0.7),
source_text=trans_data.get('description', ''),
metadata=trans_data
)
spec.transitions.append(transition)
return spec
def validate_specification(self, spec: FSMSpecification) -> Tuple[bool, List[str]]:
"""
Validate the extracted FSM specification for completeness and consistency.
Returns validation status and list of issues found.
"""
issues = []
# Check for minimum required elements
if not spec.states:
issues.append("No states identified in the specification")
if not spec.events and not spec.transitions:
issues.append("No events or transitions identified")
# Check for initial state
if not spec.initial_state:
issues.append("No initial state specified")
elif spec.initial_state not in [s.name for s in spec.states]:
issues.append(f"Initial state '{spec.initial_state}' not found in state list")
# Check transition consistency
state_names = {s.name for s in spec.states}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
target = transition.metadata.get('target_state')
if source and source not in state_names:
issues.append(f"Transition references unknown source state: {source}")
if target and target not in state_names:
issues.append(f"Transition references unknown target state: {target}")
# Check for unreachable states
reachable_states = {spec.initial_state} if spec.initial_state else set()
for transition in spec.transitions:
target = transition.metadata.get('target_state')
if target:
reachable_states.add(target)
unreachable = state_names - reachable_states
if unreachable:
issues.append(f"Unreachable states detected: {', '.join(unreachable)}")
return len(issues) == 0, issues
This requirements analysis pipeline demonstrates how to systematically extract and validate FSM specifications from natural language. The combination of pattern-based extraction and LLM analysis provides both reliability and flexibility in handling diverse user inputs.
Creating Interactive Clarification Systems
The clarification system is where the agent demonstrates its intelligence by asking targeted questions to resolve ambiguities and fill gaps in the requirements. This system must balance thoroughness with user experience, avoiding overwhelming users with too many questions while ensuring all critical information is gathered.
The clarification engine operates on multiple levels. It identifies missing information that is essential for FSM generation, detects ambiguous statements that could be interpreted multiple ways, and recognizes inconsistencies that need resolution. The challenge lies in prioritizing these issues and presenting them to users in a logical, digestible sequence.
The system must also adapt its questioning style based on the user's apparent technical expertise and the complexity of their requirements. Technical users might appreciate detailed questions about edge cases, while business users might need more guidance about the implications of their choices.
Here's a detailed implementation of the interactive clarification system:
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass
from enum import Enum
class ClarificationPriority(Enum):
CRITICAL = 1
IMPORTANT = 2
HELPFUL = 3
@dataclass
class ClarificationQuestion:
question: str
priority: ClarificationPriority
category: str # 'missing_info', 'ambiguity', 'inconsistency', 'validation'
context: Dict[str, Any]
suggested_answers: List[str] = None
follow_up_questions: List[str] = None
class ClarificationEngine:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.question_templates = self._initialize_question_templates()
self.max_questions_per_round = 3
def _initialize_question_templates(self) -> Dict[str, str]:
"""
Initialize templates for different types of clarification questions.
These templates provide structure while allowing for customization.
"""
return {
"missing_initial_state": "I notice you haven't specified which state the system should start in. What should be the initial state?",
"ambiguous_state_reference": "When you mention '{reference}', I'm not sure which specific state you're referring to. Could you clarify?",
"undefined_transition": "What should happen when the system is in the '{source_state}' state and the '{event}' event occurs?",
"missing_final_states": "Are there any states where the system should stop or terminate? If so, which ones?",
"conflicting_transitions": "I found conflicting information about transitions from '{state}'. Could you clarify the correct behavior?",
"missing_actions": "Should any specific actions be performed when entering the '{state}' state or during the transition from '{source}' to '{target}'?",
"event_conditions": "For the '{event}' event, are there any specific conditions that must be met for it to trigger?",
"state_behavior": "While in the '{state}' state, should the system perform any ongoing actions or monitor for specific conditions?"
}
def generate_clarifications(self, specification: FSMSpecification,
original_description: str) -> List[ClarificationQuestion]:
"""
Generate a prioritized list of clarification questions based on
the current specification and identified gaps or ambiguities.
"""
questions = []
# Check for critical missing information
questions.extend(self._check_critical_missing_info(specification))
# Identify ambiguities in the specification
questions.extend(self._identify_ambiguities(specification, original_description))
# Look for inconsistencies
questions.extend(self._detect_inconsistencies(specification))
# Generate validation questions
questions.extend(self._generate_validation_questions(specification))
# Sort by priority and return top questions
questions.sort(key=lambda q: q.priority.value)
return questions[:self.max_questions_per_round]
def _check_critical_missing_info(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Identify critical information that is absolutely required for FSM generation.
"""
questions = []
# Check for initial state
if not spec.initial_state:
state_names = [s.name for s in spec.states]
question = ClarificationQuestion(
question=self.question_templates["missing_initial_state"],
priority=ClarificationPriority.CRITICAL,
category="missing_info",
context={"available_states": state_names},
suggested_answers=state_names if state_names else ["idle", "start", "initial"]
)
questions.append(question)
# Check for transitions if events exist but no transitions defined
if spec.events and not spec.transitions:
question = ClarificationQuestion(
question="I see you've mentioned several events, but I'm not clear on how they should change the system's state. Could you describe what should happen for each event?",
priority=ClarificationPriority.CRITICAL,
category="missing_info",
context={"events": [e.name for e in spec.events], "states": [s.name for s in spec.states]}
)
questions.append(question)
# Check for states without any way to reach them
if spec.states and spec.transitions:
reachable_states = {spec.initial_state} if spec.initial_state else set()
for transition in spec.transitions:
target = transition.metadata.get('target_state')
if target:
reachable_states.add(target)
unreachable = [s.name for s in spec.states if s.name not in reachable_states]
if unreachable:
question = ClarificationQuestion(
question=f"I found these states that don't seem to be reachable: {', '.join(unreachable)}. How should the system get to these states?",
priority=ClarificationPriority.IMPORTANT,
category="missing_info",
context={"unreachable_states": unreachable}
)
questions.append(question)
return questions
def _identify_ambiguities(self, spec: FSMSpecification,
description: str) -> List[ClarificationQuestion]:
"""
Use the LLM to identify ambiguous statements in the original description
that could lead to multiple valid interpretations.
"""
ambiguity_prompt = f"""
Analyze this FSM description for ambiguities that could lead to different implementations:
Original description: {description}
Current specification:
States: {[s.name for s in spec.states]}
Events: {[e.name for e in spec.events]}
Actions: {[a.name for a in spec.actions]}
Identify specific ambiguities and generate clarification questions. Focus on:
1. Vague state descriptions
2. Unclear event triggers
3. Ambiguous action specifications
4. Undefined system behaviors
Return a JSON list of ambiguity objects with questions and context.
"""
messages = [{"role": "system", "content": ambiguity_prompt}]
response = self.llm.generate_response(messages, temperature=0.4)
ambiguities_data = self.llm.parse_structured_response(response.content)
questions = []
if 'ambiguities' in ambiguities_data:
for amb in ambiguities_data['ambiguities']:
question = ClarificationQuestion(
question=amb.get('question', 'Could you clarify this requirement?'),
priority=ClarificationPriority.IMPORTANT,
category="ambiguity",
context=amb.get('context', {}),
suggested_answers=amb.get('suggested_answers', [])
)
questions.append(question)
return questions
def _detect_inconsistencies(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Detect logical inconsistencies in the specification that need resolution.
"""
questions = []
# Check for conflicting transitions from the same state with the same event
transition_map = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
if source and event:
key = (source, event)
if key in transition_map:
# Found conflicting transition
existing_target = transition_map[key]
if existing_target != target:
question = ClarificationQuestion(
question=f"I found conflicting transitions from state '{source}' on event '{event}'. Should it go to '{existing_target}' or '{target}'?",
priority=ClarificationPriority.CRITICAL,
category="inconsistency",
context={
"source_state": source,
"event": event,
"conflicting_targets": [existing_target, target]
}
)
questions.append(question)
else:
transition_map[key] = target
# Check for states that have no outgoing transitions (potential dead ends)
states_with_outgoing = set()
for transition in spec.transitions:
source = transition.metadata.get('source_state')
if source:
states_with_outgoing.add(source)
dead_end_states = [s.name for s in spec.states if s.name not in states_with_outgoing and s.name not in spec.final_states]
if dead_end_states:
question = ClarificationQuestion(
question=f"These states don't seem to have any way to exit them: {', '.join(dead_end_states)}. Are these intended to be final states, or should there be transitions out of them?",
priority=ClarificationPriority.IMPORTANT,
category="inconsistency",
context={"dead_end_states": dead_end_states}
)
questions.append(question)
return questions
def _generate_validation_questions(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Generate questions to validate the user's intent and catch potential oversights.
"""
questions = []
# Ask about error handling if not explicitly mentioned
error_related = any('error' in s.name.lower() or 'fault' in s.name.lower() for s in spec.states)
if not error_related and len(spec.states) > 2:
question = ClarificationQuestion(
question="Should the system handle error conditions or unexpected events? If so, how should it behave when something goes wrong?",
priority=ClarificationPriority.HELPFUL,
category="validation",
context={"suggestion": "error_handling"}
)
questions.append(question)
# Ask about timeout behaviors if time-based events are mentioned
time_related = any('timeout' in e.name.lower() or 'timer' in e.name.lower() for e in spec.events)
if time_related:
question = ClarificationQuestion(
question="For the timeout events, what should happen if the timeout occurs in different states? Should the behavior be the same everywhere?",
priority=ClarificationPriority.HELPFUL,
category="validation",
context={"time_events": [e.name for e in spec.events if 'timeout' in e.name.lower() or 'timer' in e.name.lower()]}
)
questions.append(question)
return questions
def process_clarification_response(self, question: ClarificationQuestion,
user_response: str) -> Dict[str, Any]:
"""
Process the user's response to a clarification question and extract
actionable information for updating the specification.
"""
processing_prompt = f"""
The user was asked this clarification question:
{question.question}
Context: {json.dumps(question.context)}
User's response: {user_response}
Extract actionable information from the user's response. Determine:
1. What specific changes should be made to the FSM specification
2. Whether the question was fully answered or needs follow-up
3. Any new information that affects other parts of the specification
Return your analysis in JSON format with specific update instructions.
"""
messages = [{"role": "system", "content": processing_prompt}]
response = self.llm.generate_response(messages, temperature=0.3)
return self.llm.parse_structured_response(response.content)
def update_specification_from_clarification(self, spec: FSMSpecification,
question: ClarificationQuestion,
user_response: str) -> FSMSpecification:
"""
Update the FSM specification based on the user's clarification response.
"""
analysis = self.process_clarification_response(question, user_response)
# Apply updates based on the analysis
updates = analysis.get('updates', {})
# Add new states if specified
new_states = updates.get('new_states', [])
for state_name in new_states:
if not any(s.name == state_name for s in spec.states):
new_state = FSMElement(
name=state_name,
type='state',
confidence=0.9,
source_text=user_response,
metadata={"added_via_clarification": True}
)
spec.states.append(new_state)
# Update initial state if specified
if 'initial_state' in updates:
spec.initial_state = updates['initial_state']
# Add new transitions if specified
new_transitions = updates.get('new_transitions', [])
for trans_data in new_transitions:
transition = FSMElement(
name=f"{trans_data.get('source', 'unknown')}_to_{trans_data.get('target', 'unknown')}",
type='transition',
confidence=0.9,
source_text=user_response,
metadata={
"source_state": trans_data.get('source'),
"target_state": trans_data.get('target'),
"event": trans_data.get('event'),
"action": trans_data.get('action'),
"added_via_clarification": True
}
)
spec.transitions.append(transition)
# Update final states if specified
if 'final_states' in updates:
spec.final_states.extend(updates['final_states'])
return spec
This clarification system demonstrates how to systematically identify and resolve gaps in user requirements. The priority-based questioning ensures that critical issues are addressed first while avoiding overwhelming users with too many questions at once.
Developing the FSM Knowledge Base and Validation
The knowledge base component provides the agent with domain expertise about finite state machines, including best practices, common patterns, and validation rules. This knowledge helps the agent generate better questions, validate specifications, and produce higher-quality code.
The validation system works at multiple levels. Structural validation ensures the FSM specification is mathematically sound and implementable. Semantic validation checks that the behavior makes sense in the problem domain. Practical validation considers implementation constraints and performance implications.
The knowledge base also includes patterns for common FSM types such as protocol state machines, user interface controllers, and device drivers. These patterns help the agent recognize familiar scenarios and apply appropriate templates and best practices.
Here's an implementation of the FSM knowledge base and validation system:
from typing import Dict, List, Set, Tuple, Optional, Any
from dataclasses import dataclass
from enum import Enum
class ValidationSeverity(Enum):
ERROR = "error"
WARNING = "warning"
INFO = "info"
@dataclass
class ValidationIssue:
severity: ValidationSeverity
message: str
category: str
affected_elements: List[str]
suggestion: Optional[str] = None
class FSMPattern:
def __init__(self, name: str, description: str,
typical_states: List[str], typical_events: List[str],
validation_rules: List[str]):
self.name = name
self.description = description
self.typical_states = typical_states
self.typical_events = typical_events
self.validation_rules = validation_rules
class FSMKnowledgeBase:
def __init__(self):
self.patterns = self._initialize_patterns()
self.validation_rules = self._initialize_validation_rules()
self.best_practices = self._initialize_best_practices()
def _initialize_patterns(self) -> Dict[str, FSMPattern]:
"""
Initialize common FSM patterns that the agent can recognize and apply.
"""
patterns = {}
# Protocol state machine pattern
patterns['protocol'] = FSMPattern(
name="Protocol State Machine",
description="Manages communication protocol states",
typical_states=["disconnected", "connecting", "connected", "error"],
typical_events=["connect_request", "connection_established", "disconnect", "timeout", "error_occurred"],
validation_rules=[
"Must have error handling states",
"Should have timeout handling",
"Connection states should be clearly defined"
]
)
# Device controller pattern
patterns['device_controller'] = FSMPattern(
name="Device Controller",
description="Controls physical device operations",
typical_states=["idle", "initializing", "running", "stopping", "error"],
typical_events=["start", "stop", "reset", "fault_detected", "operation_complete"],
validation_rules=[
"Must have safe shutdown sequence",
"Should handle fault conditions",
"Initialization sequence should be defined"
]
)
# User interface pattern
patterns['ui_controller'] = FSMPattern(
name="User Interface Controller",
description="Manages user interface state transitions",
typical_states=["menu", "input", "processing", "display_result", "error"],
typical_events=["user_input", "button_press", "timeout", "validation_error", "back"],
validation_rules=[
"Should provide way to return to previous states",
"Must handle invalid user input",
"Timeout behavior should be consistent"
]
)
# Traffic control pattern
patterns['traffic_control'] = FSMPattern(
name="Traffic Control System",
description="Controls traffic flow with timing constraints",
typical_states=["red", "yellow", "green", "flashing"],
typical_events=["timer_expired", "emergency", "manual_override", "system_reset"],
validation_rules=[
"Must ensure safe transitions between states",
"Should have emergency override capability",
"Timing constraints must be respected"
]
)
return patterns
def _initialize_validation_rules(self) -> Dict[str, callable]:
"""
Initialize validation rules that can be applied to any FSM specification.
"""
return {
'has_initial_state': self._validate_initial_state,
'no_unreachable_states': self._validate_reachable_states,
'no_dead_ends': self._validate_dead_ends,
'consistent_transitions': self._validate_transition_consistency,
'proper_naming': self._validate_naming_conventions,
'completeness': self._validate_completeness
}
def _initialize_best_practices(self) -> Dict[str, str]:
"""
Initialize best practices recommendations for FSM design.
"""
return {
'state_naming': "Use descriptive names that clearly indicate what the system is doing",
'event_naming': "Use verb phrases that describe what triggers the transition",
'error_handling': "Include explicit error states and recovery mechanisms",
'documentation': "Provide clear descriptions for each state and transition",
'simplicity': "Keep the state machine as simple as possible while meeting requirements",
'testability': "Design states and transitions that can be easily tested"
}
def identify_pattern(self, spec: FSMSpecification) -> Optional[FSMPattern]:
"""
Identify which common pattern the specification most closely matches.
"""
state_names = [s.name.lower() for s in spec.states]
event_names = [e.name.lower() for e in spec.events]
best_match = None
best_score = 0
for pattern_name, pattern in self.patterns.items():
# Calculate similarity score based on state and event overlap
state_overlap = len(set(state_names) & set(pattern.typical_states))
event_overlap = len(set(event_names) & set(pattern.typical_events))
# Weight the score based on the number of matching elements
score = (state_overlap * 2 + event_overlap) / (len(pattern.typical_states) + len(pattern.typical_events))
if score > best_score and score > 0.3: # Minimum threshold for pattern match
best_score = score
best_match = pattern
return best_match
def validate_specification(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""
Perform comprehensive validation of the FSM specification.
"""
issues = []
# Apply all validation rules
for rule_name, rule_func in self.validation_rules.items():
try:
rule_issues = rule_func(spec)
issues.extend(rule_issues)
except Exception as e:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Validation rule '{rule_name}' failed: {str(e)}",
category="validation_error",
affected_elements=[]
))
# Apply pattern-specific validation if a pattern is identified
pattern = self.identify_pattern(spec)
if pattern:
pattern_issues = self._validate_against_pattern(spec, pattern)
issues.extend(pattern_issues)
return issues
def _validate_initial_state(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that an initial state is properly defined."""
issues = []
if not spec.initial_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message="No initial state specified",
category="structural",
affected_elements=[],
suggestion="Specify which state the system should start in"
))
elif spec.initial_state not in [s.name for s in spec.states]:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Initial state '{spec.initial_state}' not found in state list",
category="structural",
affected_elements=[spec.initial_state],
suggestion="Ensure the initial state is included in the states list"
))
return issues
def _validate_reachable_states(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that all states are reachable from the initial state."""
issues = []
if not spec.initial_state or not spec.transitions:
return issues # Can't validate without initial state and transitions
# Build reachability graph
reachable = {spec.initial_state}
changed = True
while changed:
changed = False
for transition in spec.transitions:
source = transition.metadata.get('source_state')
target = transition.metadata.get('target_state')
if source in reachable and target and target not in reachable:
reachable.add(target)
changed = True
# Find unreachable states
all_states = {s.name for s in spec.states}
unreachable = all_states - reachable
if unreachable:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message=f"Unreachable states detected: {', '.join(unreachable)}",
category="reachability",
affected_elements=list(unreachable),
suggestion="Add transitions to make these states reachable or remove them if unnecessary"
))
return issues
def _validate_dead_ends(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that non-final states have outgoing transitions."""
issues = []
# Find states with no outgoing transitions
states_with_outgoing = set()
for transition in spec.transitions:
source = transition.metadata.get('source_state')
if source:
states_with_outgoing.add(source)
all_state_names = {s.name for s in spec.states}
final_state_names = set(spec.final_states)
dead_ends = all_state_names - states_with_outgoing - final_state_names
if dead_ends:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message=f"States with no outgoing transitions: {', '.join(dead_ends)}",
category="dead_end",
affected_elements=list(dead_ends),
suggestion="Add outgoing transitions or mark these as final states"
))
return issues
def _validate_transition_consistency(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that transitions are consistent and well-defined."""
issues = []
# Check for duplicate transitions (same source, event, but different target)
transition_map = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
if source and event:
key = (source, event)
if key in transition_map:
existing_target = transition_map[key]
if existing_target != target:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Conflicting transitions from '{source}' on event '{event}'",
category="consistency",
affected_elements=[source, event],
suggestion="Resolve the conflict by specifying conditions or combining transitions"
))
else:
transition_map[key] = target
return issues
def _validate_naming_conventions(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate naming conventions for states, events, and actions."""
issues = []
# Check for descriptive state names
generic_state_names = {'state1', 'state2', 'temp', 'unknown', 'default'}
for state in spec.states:
if state.name.lower() in generic_state_names:
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message=f"State '{state.name}' has a generic name",
category="naming",
affected_elements=[state.name],
suggestion="Use more descriptive names that indicate what the system is doing"
))
# Check for consistent naming patterns
state_names = [s.name for s in spec.states]
if len(state_names) > 1:
# Check if names follow consistent case convention
all_lower = all(name.islower() for name in state_names)
all_upper = all(name.isupper() for name in state_names)
all_title = all(name.istitle() for name in state_names)
if not (all_lower or all_upper or all_title):
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message="Inconsistent naming case convention",
category="naming",
affected_elements=state_names,
suggestion="Use consistent case convention for all names"
))
return issues
def _validate_completeness(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that the specification is complete enough for implementation."""
issues = []
# Check minimum requirements
if len(spec.states) < 2:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="FSM has fewer than 2 states",
category="completeness",
affected_elements=[],
suggestion="Consider if additional states are needed for proper system modeling"
))
if not spec.events and not spec.transitions:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message="No events or transitions defined",
category="completeness",
affected_elements=[],
suggestion="Define events that trigger state changes or explicit transitions"
))
# Check for actions if none are defined
if not spec.actions and len(spec.states) > 2:
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message="No actions defined for states or transitions",
category="completeness",
affected_elements=[],
suggestion="Consider adding actions that should be performed during state changes"
))
return issues
def _validate_against_pattern(self, spec: FSMSpecification,
pattern: FSMPattern) -> List[ValidationIssue]:
"""Validate the specification against a recognized pattern."""
issues = []
# This is a simplified pattern validation
# In a real implementation, each pattern would have specific validation logic
if pattern.name == "Protocol State Machine":
# Check for error handling
has_error_state = any('error' in s.name.lower() for s in spec.states)
if not has_error_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="Protocol FSM should include error handling states",
category="pattern_compliance",
affected_elements=[],
suggestion="Add states to handle connection errors and timeouts"
))
elif pattern.name == "Device Controller":
# Check for safe shutdown
has_stop_state = any('stop' in s.name.lower() or 'shutdown' in s.name.lower() for s in spec.states)
if not has_stop_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="Device controller should have explicit stop/shutdown state",
category="pattern_compliance",
affected_elements=[],
suggestion="Add a safe shutdown state for proper device control"
))
return issues
Creating Interactive Clarification Systems
The clarification system is where the agent demonstrates its intelligence by asking targeted questions to resolve ambiguities and fill gaps in the requirements. This system must balance thoroughness with user experience, avoiding overwhelming users with too many questions while ensuring all critical information is gathered.
The clarification engine operates on multiple levels. It identifies missing information that is essential for FSM generation, detects ambiguous statements that could be interpreted multiple ways, and recognizes inconsistencies that need resolution. The challenge lies in prioritizing these issues and presenting them to users in a logical, digestible sequence.
The system must also adapt its questioning style based on the user's apparent technical expertise and the complexity of their requirements. Technical users might appreciate detailed questions about edge cases, while business users might need more guidance about the implications of their choices.
Here's a detailed implementation of the interactive clarification system:
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass
from enum import Enum
class ClarificationPriority(Enum):
CRITICAL = 1
IMPORTANT = 2
HELPFUL = 3
@dataclass
class ClarificationQuestion:
question: str
priority: ClarificationPriority
category: str # 'missing_info', 'ambiguity', 'inconsistency', 'validation'
context: Dict[str, Any]
suggested_answers: List[str] = None
follow_up_questions: List[str] = None
class ClarificationEngine:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.question_templates = self._initialize_question_templates()
self.max_questions_per_round = 3
def _initialize_question_templates(self) -> Dict[str, str]:
"""
Initialize templates for different types of clarification questions.
These templates provide structure while allowing for customization.
"""
return {
"missing_initial_state": "I notice you haven't specified which state the system should start in. What should be the initial state?",
"ambiguous_state_reference": "When you mention '{reference}', I'm not sure which specific state you're referring to. Could you clarify?",
"undefined_transition": "What should happen when the system is in the '{source_state}' state and the '{event}' event occurs?",
"missing_final_states": "Are there any states where the system should stop or terminate? If so, which ones?",
"conflicting_transitions": "I found conflicting information about transitions from '{state}'. Could you clarify the correct behavior?",
"missing_actions": "Should any specific actions be performed when entering the '{state}' state or during the transition from '{source}' to '{target}'?",
"event_conditions": "For the '{event}' event, are there any specific conditions that must be met for it to trigger?",
"state_behavior": "While in the '{state}' state, should the system perform any ongoing actions or monitor for specific conditions?"
}
def generate_clarifications(self, specification: FSMSpecification,
original_description: str) -> List[ClarificationQuestion]:
"""
Generate a prioritized list of clarification questions based on
the current specification and identified gaps or ambiguities.
"""
questions = []
# Check for critical missing information
questions.extend(self._check_critical_missing_info(specification))
# Identify ambiguities in the specification
questions.extend(self._identify_ambiguities(specification, original_description))
# Look for inconsistencies
questions.extend(self._detect_inconsistencies(specification))
# Generate validation questions
questions.extend(self._generate_validation_questions(specification))
# Sort by priority and return top questions
questions.sort(key=lambda q: q.priority.value)
return questions[:self.max_questions_per_round]
def _check_critical_missing_info(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Identify critical information that is absolutely required for FSM generation.
"""
questions = []
# Check for initial state
if not spec.initial_state:
state_names = [s.name for s in spec.states]
question = ClarificationQuestion(
question=self.question_templates["missing_initial_state"],
priority=ClarificationPriority.CRITICAL,
category="missing_info",
context={"available_states": state_names},
suggested_answers=state_names if state_names else ["idle", "start", "initial"]
)
questions.append(question)
# Check for transitions if events exist but no transitions defined
if spec.events and not spec.transitions:
question = ClarificationQuestion(
question="I see you've mentioned several events, but I'm not clear on how they should change the system's state. Could you describe what should happen for each event?",
priority=ClarificationPriority.CRITICAL,
category="missing_info",
context={"events": [e.name for e in spec.events], "states": [s.name for s in spec.states]}
)
questions.append(question)
# Check for states without any way to reach them
if spec.states and spec.transitions:
reachable_states = {spec.initial_state} if spec.initial_state else set()
for transition in spec.transitions:
target = transition.metadata.get('target_state')
if target:
reachable_states.add(target)
unreachable = [s.name for s in spec.states if s.name not in reachable_states]
if unreachable:
question = ClarificationQuestion(
question=f"I found these states that don't seem to be reachable: {', '.join(unreachable)}. How should the system get to these states?",
priority=ClarificationPriority.IMPORTANT,
category="missing_info",
context={"unreachable_states": unreachable}
)
questions.append(question)
return questions
def _identify_ambiguities(self, spec: FSMSpecification,
description: str) -> List[ClarificationQuestion]:
"""
Use the LLM to identify ambiguous statements in the original description
that could lead to multiple valid interpretations.
"""
ambiguity_prompt = f"""
Analyze this FSM description for ambiguities that could lead to different implementations:
Original description: {description}
Current specification:
States: {[s.name for s in spec.states]}
Events: {[e.name for e in spec.events]}
Actions: {[a.name for a in spec.actions]}
Identify specific ambiguities and generate clarification questions. Focus on:
1. Vague state descriptions
2. Unclear event triggers
3. Ambiguous action specifications
4. Undefined system behaviors
Return a JSON list of ambiguity objects with questions and context.
"""
messages = [{"role": "system", "content": ambiguity_prompt}]
response = self.llm.generate_response(messages, temperature=0.4)
ambiguities_data = self.llm.parse_structured_response(response.content)
questions = []
if 'ambiguities' in ambiguities_data:
for amb in ambiguities_data['ambiguities']:
question = ClarificationQuestion(
question=amb.get('question', 'Could you clarify this requirement?'),
priority=ClarificationPriority.IMPORTANT,
category="ambiguity",
context=amb.get('context', {}),
suggested_answers=amb.get('suggested_answers', [])
)
questions.append(question)
return questions
def _detect_inconsistencies(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Detect logical inconsistencies in the specification that need resolution.
"""
questions = []
# Check for conflicting transitions from the same state with the same event
transition_map = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
if source and event:
key = (source, event)
if key in transition_map:
# Found conflicting transition
existing_target = transition_map[key]
if existing_target != target:
question = ClarificationQuestion(
question=f"I found conflicting transitions from state '{source}' on event '{event}'. Should it go to '{existing_target}' or '{target}'?",
priority=ClarificationPriority.CRITICAL,
category="inconsistency",
context={
"source_state": source,
"event": event,
"conflicting_targets": [existing_target, target]
}
)
questions.append(question)
else:
transition_map[key] = target
# Check for states that have no outgoing transitions (potential dead ends)
states_with_outgoing = set()
for transition in spec.transitions:
source = transition.metadata.get('source_state')
if source:
states_with_outgoing.add(source)
dead_end_states = [s.name for s in spec.states if s.name not in states_with_outgoing and s.name not in spec.final_states]
if dead_end_states:
question = ClarificationQuestion(
question=f"These states don't seem to have any way to exit them: {', '.join(dead_end_states)}. Are these intended to be final states, or should there be transitions out of them?",
priority=ClarificationPriority.IMPORTANT,
category="inconsistency",
context={"dead_end_states": dead_end_states}
)
questions.append(question)
return questions
def _generate_validation_questions(self, spec: FSMSpecification) -> List[ClarificationQuestion]:
"""
Generate questions to validate the user's intent and catch potential oversights.
"""
questions = []
# Ask about error handling if not explicitly mentioned
error_related = any('error' in s.name.lower() or 'fault' in s.name.lower() for s in spec.states)
if not error_related and len(spec.states) > 2:
question = ClarificationQuestion(
question="Should the system handle error conditions or unexpected events? If so, how should it behave when something goes wrong?",
priority=ClarificationPriority.HELPFUL,
category="validation",
context={"suggestion": "error_handling"}
)
questions.append(question)
# Ask about timeout behaviors if time-based events are mentioned
time_related = any('timeout' in e.name.lower() or 'timer' in e.name.lower() for e in spec.events)
if time_related:
question = ClarificationQuestion(
question="For the timeout events, what should happen if the timeout occurs in different states? Should the behavior be the same everywhere?",
priority=ClarificationPriority.HELPFUL,
category="validation",
context={"time_events": [e.name for e in spec.events if 'timeout' in e.name.lower() or 'timer' in e.name.lower()]}
)
questions.append(question)
return questions
def process_clarification_response(self, question: ClarificationQuestion,
user_response: str) -> Dict[str, Any]:
"""
Process the user's response to a clarification question and extract
actionable information for updating the specification.
"""
processing_prompt = f"""
The user was asked this clarification question:
{question.question}
Context: {json.dumps(question.context)}
User's response: {user_response}
Extract actionable information from the user's response. Determine:
1. What specific changes should be made to the FSM specification
2. Whether the question was fully answered or needs follow-up
3. Any new information that affects other parts of the specification
Return your analysis in JSON format with specific update instructions.
"""
messages = [{"role": "system", "content": processing_prompt}]
response = self.llm.generate_response(messages, temperature=0.3)
return self.llm.parse_structured_response(response.content)
def update_specification_from_clarification(self, spec: FSMSpecification,
question: ClarificationQuestion,
user_response: str) -> FSMSpecification:
"""
Update the FSM specification based on the user's clarification response.
"""
analysis = self.process_clarification_response(question, user_response)
# Apply updates based on the analysis
updates = analysis.get('updates', {})
# Add new states if specified
new_states = updates.get('new_states', [])
for state_name in new_states:
if not any(s.name == state_name for s in spec.states):
new_state = FSMElement(
name=state_name,
type='state',
confidence=0.9,
source_text=user_response,
metadata={"added_via_clarification": True}
)
spec.states.append(new_state)
# Update initial state if specified
if 'initial_state' in updates:
spec.initial_state = updates['initial_state']
# Add new transitions if specified
new_transitions = updates.get('new_transitions', [])
for trans_data in new_transitions:
transition = FSMElement(
name=f"{trans_data.get('source', 'unknown')}_to_{trans_data.get('target', 'unknown')}",
type='transition',
confidence=0.9,
source_text=user_response,
metadata={
"source_state": trans_data.get('source'),
"target_state": trans_data.get('target'),
"event": trans_data.get('event'),
"action": trans_data.get('action'),
"added_via_clarification": True
}
)
spec.transitions.append(transition)
# Update final states if specified
if 'final_states' in updates:
spec.final_states.extend(updates['final_states'])
return spec
This clarification system demonstrates how to systematically identify and resolve gaps in user requirements. The priority-based questioning ensures that critical issues are addressed first while avoiding overwhelming users with too many questions at once.
Developing the FSM Knowledge Base and Validation
The knowledge base component provides the agent with domain expertise about finite state machines, including best practices, common patterns, and validation rules. This knowledge helps the agent generate better questions, validate specifications, and produce higher-quality code.
The validation system works at multiple levels. Structural validation ensures the FSM specification is mathematically sound and implementable. Semantic validation checks that the behavior makes sense in the problem domain. Practical validation considers implementation constraints and performance implications.
The knowledge base also includes patterns for common FSM types such as protocol state machines, user interface controllers, and device drivers. These patterns help the agent recognize familiar scenarios and apply appropriate templates and best practices.
Here's an implementation of the FSM knowledge base and validation system:
from typing import Dict, List, Set, Tuple, Optional, Any
from dataclasses import dataclass
from enum import Enum
class ValidationSeverity(Enum):
ERROR = "error"
WARNING = "warning"
INFO = "info"
@dataclass
class ValidationIssue:
severity: ValidationSeverity
message: str
category: str
affected_elements: List[str]
suggestion: Optional[str] = None
class FSMPattern:
def __init__(self, name: str, description: str,
typical_states: List[str], typical_events: List[str],
validation_rules: List[str]):
self.name = name
self.description = description
self.typical_states = typical_states
self.typical_events = typical_events
self.validation_rules = validation_rules
class FSMKnowledgeBase:
def __init__(self):
self.patterns = self._initialize_patterns()
self.validation_rules = self._initialize_validation_rules()
self.best_practices = self._initialize_best_practices()
def _initialize_patterns(self) -> Dict[str, FSMPattern]:
"""
Initialize common FSM patterns that the agent can recognize and apply.
"""
patterns = {}
# Protocol state machine pattern
patterns['protocol'] = FSMPattern(
name="Protocol State Machine",
description="Manages communication protocol states",
typical_states=["disconnected", "connecting", "connected", "error"],
typical_events=["connect_request", "connection_established", "disconnect", "timeout", "error_occurred"],
validation_rules=[
"Must have error handling states",
"Should have timeout handling",
"Connection states should be clearly defined"
]
)
# Device controller pattern
patterns['device_controller'] = FSMPattern(
name="Device Controller",
description="Controls physical device operations",
typical_states=["idle", "initializing", "running", "stopping", "error"],
typical_events=["start", "stop", "reset", "fault_detected", "operation_complete"],
validation_rules=[
"Must have safe shutdown sequence",
"Should handle fault conditions",
"Initialization sequence should be defined"
]
)
# User interface pattern
patterns['ui_controller'] = FSMPattern(
name="User Interface Controller",
description="Manages user interface state transitions",
typical_states=["menu", "input", "processing", "display_result", "error"],
typical_events=["user_input", "button_press", "timeout", "validation_error", "back"],
validation_rules=[
"Should provide way to return to previous states",
"Must handle invalid user input",
"Timeout behavior should be consistent"
]
)
# Traffic control pattern
patterns['traffic_control'] = FSMPattern(
name="Traffic Control System",
description="Controls traffic flow with timing constraints",
typical_states=["red", "yellow", "green", "flashing"],
typical_events=["timer_expired", "emergency", "manual_override", "system_reset"],
validation_rules=[
"Must ensure safe transitions between states",
"Should have emergency override capability",
"Timing constraints must be respected"
]
)
return patterns
def _initialize_validation_rules(self) -> Dict[str, callable]:
"""
Initialize validation rules that can be applied to any FSM specification.
"""
return {
'has_initial_state': self._validate_initial_state,
'no_unreachable_states': self._validate_reachable_states,
'no_dead_ends': self._validate_dead_ends,
'consistent_transitions': self._validate_transition_consistency,
'proper_naming': self._validate_naming_conventions,
'completeness': self._validate_completeness
}
def _initialize_best_practices(self) -> Dict[str, str]:
"""
Initialize best practices recommendations for FSM design.
"""
return {
'state_naming': "Use descriptive names that clearly indicate what the system is doing",
'event_naming': "Use verb phrases that describe what triggers the transition",
'error_handling': "Include explicit error states and recovery mechanisms",
'documentation': "Provide clear descriptions for each state and transition",
'simplicity': "Keep the state machine as simple as possible while meeting requirements",
'testability': "Design states and transitions that can be easily tested"
}
def identify_pattern(self, spec: FSMSpecification) -> Optional[FSMPattern]:
"""
Identify which common pattern the specification most closely matches.
"""
state_names = [s.name.lower() for s in spec.states]
event_names = [e.name.lower() for e in spec.events]
best_match = None
best_score = 0
for pattern_name, pattern in self.patterns.items():
# Calculate similarity score based on state and event overlap
state_overlap = len(set(state_names) & set(pattern.typical_states))
event_overlap = len(set(event_names) & set(pattern.typical_events))
# Weight the score based on the number of matching elements
score = (state_overlap * 2 + event_overlap) / (len(pattern.typical_states) + len(pattern.typical_events))
if score > best_score and score > 0.3: # Minimum threshold for pattern match
best_score = score
best_match = pattern
return best_match
def validate_specification(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""
Perform comprehensive validation of the FSM specification.
"""
issues = []
# Apply all validation rules
for rule_name, rule_func in self.validation_rules.items():
try:
rule_issues = rule_func(spec)
issues.extend(rule_issues)
except Exception as e:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Validation rule '{rule_name}' failed: {str(e)}",
category="validation_error",
affected_elements=[]
))
# Apply pattern-specific validation if a pattern is identified
pattern = self.identify_pattern(spec)
if pattern:
pattern_issues = self._validate_against_pattern(spec, pattern)
issues.extend(pattern_issues)
return issues
def _validate_initial_state(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that an initial state is properly defined."""
issues = []
if not spec.initial_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message="No initial state specified",
category="structural",
affected_elements=[],
suggestion="Specify which state the system should start in"
))
elif spec.initial_state not in [s.name for s in spec.states]:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Initial state '{spec.initial_state}' not found in state list",
category="structural",
affected_elements=[spec.initial_state],
suggestion="Ensure the initial state is included in the states list"
))
return issues
def _validate_reachable_states(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that all states are reachable from the initial state."""
issues = []
if not spec.initial_state or not spec.transitions:
return issues # Can't validate without initial state and transitions
# Build reachability graph
reachable = {spec.initial_state}
changed = True
while changed:
changed = False
for transition in spec.transitions:
source = transition.metadata.get('source_state')
target = transition.metadata.get('target_state')
if source in reachable and target and target not in reachable:
reachable.add(target)
changed = True
# Find unreachable states
all_states = {s.name for s in spec.states}
unreachable = all_states - reachable
if unreachable:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message=f"Unreachable states detected: {', '.join(unreachable)}",
category="reachability",
affected_elements=list(unreachable),
suggestion="Add transitions to make these states reachable or remove them if unnecessary"
))
return issues
def _validate_dead_ends(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that non-final states have outgoing transitions."""
issues = []
# Find states with no outgoing transitions
states_with_outgoing = set()
for transition in spec.transitions:
source = transition.metadata.get('source_state')
if source:
states_with_outgoing.add(source)
all_state_names = {s.name for s in spec.states}
final_state_names = set(spec.final_states)
dead_ends = all_state_names - states_with_outgoing - final_state_names
if dead_ends:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message=f"States with no outgoing transitions: {', '.join(dead_ends)}",
category="dead_end",
affected_elements=list(dead_ends),
suggestion="Add outgoing transitions or mark these as final states"
))
return issues
def _validate_transition_consistency(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that transitions are consistent and well-defined."""
issues = []
# Check for duplicate transitions (same source, event, but different target)
transition_map = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
if source and event:
key = (source, event)
if key in transition_map:
existing_target = transition_map[key]
if existing_target != target:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message=f"Conflicting transitions from '{source}' on event '{event}'",
category="consistency",
affected_elements=[source, event],
suggestion="Resolve the conflict by specifying conditions or combining transitions"
))
else:
transition_map[key] = target
return issues
def _validate_naming_conventions(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate naming conventions for states, events, and actions."""
issues = []
# Check for descriptive state names
generic_state_names = {'state1', 'state2', 'temp', 'unknown', 'default'}
for state in spec.states:
if state.name.lower() in generic_state_names:
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message=f"State '{state.name}' has a generic name",
category="naming",
affected_elements=[state.name],
suggestion="Use more descriptive names that indicate what the system is doing"
))
# Check for consistent naming patterns
state_names = [s.name for s in spec.states]
if len(state_names) > 1:
# Check if names follow consistent case convention
all_lower = all(name.islower() for name in state_names)
all_upper = all(name.isupper() for name in state_names)
all_title = all(name.istitle() for name in state_names)
if not (all_lower or all_upper or all_title):
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message="Inconsistent naming case convention",
category="naming",
affected_elements=state_names,
suggestion="Use consistent case convention for all names"
))
return issues
def _validate_completeness(self, spec: FSMSpecification) -> List[ValidationIssue]:
"""Validate that the specification is complete enough for implementation."""
issues = []
# Check minimum requirements
if len(spec.states) < 2:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="FSM has fewer than 2 states",
category="completeness",
affected_elements=[],
suggestion="Consider if additional states are needed for proper system modeling"
))
if not spec.events and not spec.transitions:
issues.append(ValidationIssue(
severity=ValidationSeverity.ERROR,
message="No events or transitions defined",
category="completeness",
affected_elements=[],
suggestion="Define events that trigger state changes or explicit transitions"
))
# Check for actions if none are defined
if not spec.actions and len(spec.states) > 2:
issues.append(ValidationIssue(
severity=ValidationSeverity.INFO,
message="No actions defined for states or transitions",
category="completeness",
affected_elements=[],
suggestion="Consider adding actions that should be performed during state changes"
))
return issues
def _validate_against_pattern(self, spec: FSMSpecification,
pattern: FSMPattern) -> List[ValidationIssue]:
"""Validate the specification against a recognized pattern."""
issues = []
# This is a simplified pattern validation
# In a real implementation, each pattern would have specific validation logic
if pattern.name == "Protocol State Machine":
# Check for error handling
has_error_state = any('error' in s.name.lower() for s in spec.states)
if not has_error_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="Protocol FSM should include error handling states",
category="pattern_compliance",
affected_elements=[],
suggestion="Add states to handle connection errors and timeouts"
))
elif pattern.name == "Device Controller":
# Check for safe shutdown
has_stop_state = any('stop' in s.name.lower() or 'shutdown' in s.name.lower() for s in spec.states)
if not has_stop_state:
issues.append(ValidationIssue(
severity=ValidationSeverity.WARNING,
message="Device controller should have explicit stop/shutdown state",
category="pattern_compliance",
affected_elements=[],
suggestion="Add a safe shutdown state for proper device control"
))
return issues
This knowledge base and validation system provides the agent with domain expertise to generate better FSMs and catch potential issues before code generation. The pattern recognition helps apply appropriate best practices based on the type of system being modeled.
Implementing Multi-language Code Generation
The code generation engine transforms the validated FSM specification into executable code in the user's chosen programming language. This component must handle the differences between languages while maintaining the semantic correctness of the finite state machine.
The generator uses a template-based approach with language-specific adapters. Each language adapter knows how to represent states, handle events, manage transitions, and structure the overall code. The system supports multiple implementation patterns such as switch-based state machines, object-oriented designs, and table-driven approaches.
The code generation process involves several steps. First, the FSM specification is normalized into a canonical representation. Then, the appropriate language adapter is selected and configured. The adapter generates the code structure, implements the state logic, and adds necessary boilerplate code. Finally, the generated code is formatted and validated for syntax correctness.
Here's a comprehensive implementation of the multi-language code generation system:
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import json
@dataclass
class CodeGenerationConfig:
language: str
style: str # 'switch', 'object_oriented', 'table_driven'
include_comments: bool = True
include_validation: bool = True
namespace: Optional[str] = None
class_name: str = "StateMachine"
class LanguageAdapter(ABC):
"""Abstract base class for language-specific code generation."""
@abstractmethod
def generate_code(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
pass
@abstractmethod
def get_file_extension(self) -> str:
pass
@abstractmethod
def validate_syntax(self, code: str) -> Tuple[bool, List[str]]:
pass
class PythonAdapter(LanguageAdapter):
def generate_code(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
"""Generate Python code for the FSM specification."""
if config.style == 'object_oriented':
return self._generate_oo_python(spec, config)
elif config.style == 'switch':
return self._generate_switch_python(spec, config)
else:
return self._generate_table_driven_python(spec, config)
def _generate_oo_python(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
"""Generate object-oriented Python implementation."""
code_parts = []
# Add imports and header
code_parts.append("from enum import Enum")
code_parts.append("from typing import Dict, Callable, Optional, Any")
code_parts.append("")
if config.include_comments:
code_parts.append("# Auto-generated Finite State Machine")
code_parts.append("# Generated from user requirements")
code_parts.append("")
# Generate state enumeration
code_parts.append("class State(Enum):")
for state in spec.states:
state_name = state.name.upper().replace(' ', '_')
code_parts.append(f" {state_name} = '{state.name}'")
code_parts.append("")
# Generate event enumeration
code_parts.append("class Event(Enum):")
for event in spec.events:
event_name = event.name.upper().replace(' ', '_')
code_parts.append(f" {event_name} = '{event.name}'")
code_parts.append("")
# Generate main state machine class
class_name = config.class_name
code_parts.append(f"class {class_name}:")
code_parts.append(" def __init__(self):")
# Set initial state
if spec.initial_state:
initial_state_enum = spec.initial_state.upper().replace(' ', '_')
code_parts.append(f" self.current_state = State.{initial_state_enum}")
else:
code_parts.append(" self.current_state = None")
code_parts.append(" self.transition_table = self._build_transition_table()")
code_parts.append(" self.state_actions = self._build_state_actions()")
code_parts.append("")
# Generate transition table builder
code_parts.append(" def _build_transition_table(self) -> Dict[tuple, State]:")
code_parts.append(" \"\"\"Build the state transition table.\"\"\"")
code_parts.append(" return {")
for transition in spec.transitions:
source = transition.metadata.get('source_state')
target = transition.metadata.get('target_state')
event = transition.metadata.get('event')
if source and target and event:
source_enum = source.upper().replace(' ', '_')
target_enum = target.upper().replace(' ', '_')
event_enum = event.upper().replace(' ', '_')
code_parts.append(f" (State.{source_enum}, Event.{event_enum}): State.{target_enum},")
code_parts.append(" }")
code_parts.append("")
# Generate state actions builder
code_parts.append(" def _build_state_actions(self) -> Dict[State, Callable]:")
code_parts.append(" \"\"\"Build the state action mapping.\"\"\"")
code_parts.append(" return {")
for state in spec.states:
state_enum = state.name.upper().replace(' ', '_')
action_method = f"_on_{state.name.lower().replace(' ', '_')}"
code_parts.append(f" State.{state_enum}: self.{action_method},")
code_parts.append(" }")
code_parts.append("")
# Generate process_event method
code_parts.append(" def process_event(self, event: Event) -> bool:")
code_parts.append(" \"\"\"Process an event and transition to new state if applicable.\"\"\"")
if config.include_validation:
code_parts.append(" if not isinstance(event, Event):")
code_parts.append(" raise ValueError(f'Invalid event type: {type(event)}')")
code_parts.append("")
code_parts.append(" transition_key = (self.current_state, event)")
code_parts.append(" ")
code_parts.append(" if transition_key in self.transition_table:")
code_parts.append(" old_state = self.current_state")
code_parts.append(" self.current_state = self.transition_table[transition_key]")
code_parts.append(" ")
code_parts.append(" # Execute exit action for old state")
code_parts.append(" if old_state in self.state_actions:")
code_parts.append(" self.state_actions[old_state]()")
code_parts.append(" ")
code_parts.append(" # Execute entry action for new state")
code_parts.append(" if self.current_state in self.state_actions:")
code_parts.append(" self.state_actions[self.current_state]()")
code_parts.append(" ")
code_parts.append(" return True")
code_parts.append(" ")
code_parts.append(" return False")
code_parts.append("")
# Generate state action methods
for state in spec.states:
method_name = f"_on_{state.name.lower().replace(' ', '_')}"
code_parts.append(f" def {method_name}(self):")
code_parts.append(f" \"\"\"Action for {state.name} state.\"\"\"")
# Look for actions associated with this state
state_actions = [a for a in spec.actions if state.name.lower() in a.name.lower()]
if state_actions:
for action in state_actions:
code_parts.append(f" # TODO: Implement {action.name}")
code_parts.append(f" pass")
else:
code_parts.append(" # TODO: Implement state-specific behavior")
code_parts.append(" pass")
code_parts.append("")
# Generate utility methods
code_parts.append(" def get_current_state(self) -> State:")
code_parts.append(" \"\"\"Get the current state.\"\"\"")
code_parts.append(" return self.current_state")
code_parts.append("")
code_parts.append(" def is_in_state(self, state: State) -> bool:")
code_parts.append(" \"\"\"Check if the machine is in a specific state.\"\"\"")
code_parts.append(" return self.current_state == state")
code_parts.append("")
# Generate final states check if applicable
if spec.final_states:
code_parts.append(" def is_final_state(self) -> bool:")
code_parts.append(" \"\"\"Check if the machine is in a final state.\"\"\"")
final_states_list = [f"State.{fs.upper().replace(' ', '_')}" for fs in spec.final_states]
code_parts.append(f" return self.current_state in [{', '.join(final_states_list)}]")
code_parts.append("")
return '\n'.join(code_parts)
def _generate_switch_python(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
"""Generate switch-based Python implementation using if-elif chains."""
code_parts = []
# Add header
code_parts.append("# Auto-generated Finite State Machine (Switch-based)")
code_parts.append("")
class_name = config.class_name
code_parts.append(f"class {class_name}:")
code_parts.append(" def __init__(self):")
if spec.initial_state:
code_parts.append(f" self.current_state = '{spec.initial_state}'")
else:
code_parts.append(" self.current_state = None")
code_parts.append("")
# Generate main event processing method
code_parts.append(" def process_event(self, event):")
code_parts.append(" \"\"\"Process an event based on current state.\"\"\"")
code_parts.append(" ")
# Group transitions by source state
transitions_by_state = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
if source:
if source not in transitions_by_state:
transitions_by_state[source] = []
transitions_by_state[source].append(transition)
# Generate switch logic
first_state = True
for state_name, state_transitions in transitions_by_state.items():
condition = "if" if first_state else "elif"
code_parts.append(f" {condition} self.current_state == '{state_name}':")
first_event = True
for transition in state_transitions:
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
action = transition.metadata.get('action')
if event and target:
event_condition = "if" if first_event else "elif"
code_parts.append(f" {event_condition} event == '{event}':")
if action:
code_parts.append(f" # Execute action: {action}")
code_parts.append(f" self._execute_{action.lower().replace(' ', '_')}()")
code_parts.append(f" self.current_state = '{target}'")
code_parts.append(f" return True")
first_event = False
if not first_event:
code_parts.append(" else:")
code_parts.append(" return False")
first_state = False
if transitions_by_state:
code_parts.append(" else:")
code_parts.append(" return False")
code_parts.append("")
# Generate action methods
actions_generated = set()
for transition in spec.transitions:
action = transition.metadata.get('action')
if action and action not in actions_generated:
method_name = f"_execute_{action.lower().replace(' ', '_')}"
code_parts.append(f" def {method_name}(self):")
code_parts.append(f" \"\"\"Execute action: {action}\"\"\"")
code_parts.append(" # TODO: Implement action logic")
code_parts.append(" pass")
code_parts.append("")
actions_generated.add(action)
return '\n'.join(code_parts)
def _generate_table_driven_python(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
"""Generate table-driven Python implementation."""
code_parts = []
# Add header
code_parts.append("# Auto-generated Finite State Machine (Table-driven)")
code_parts.append("import json")
code_parts.append("")
class_name = config.class_name
code_parts.append(f"class {class_name}:")
code_parts.append(" def __init__(self):")
if spec.initial_state:
code_parts.append(f" self.current_state = '{spec.initial_state}'")
else:
code_parts.append(" self.current_state = None")
# Build transition table as JSON
transition_table = {}
for transition in spec.transitions:
source = transition.metadata.get('source_state')
event = transition.metadata.get('event')
target = transition.metadata.get('target_state')
action = transition.metadata.get('action')
if source and event and target:
key = f"{source}:{event}"
transition_table[key] = {
"target_state": target,
"action": action
}
code_parts.append(" self.transition_table = {")
for key, value in transition_table.items():
code_parts.append(f" '{key}': {json.dumps(value)},")
code_parts.append(" }")
code_parts.append("")
# Generate event processing method
code_parts.append(" def process_event(self, event):")
code_parts.append(" \"\"\"Process event using transition table.\"\"\"")
code_parts.append(" key = f'{self.current_state}:{event}'")
code_parts.append(" ")
code_parts.append(" if key in self.transition_table:")
code_parts.append(" transition = self.transition_table[key]")
code_parts.append(" ")
code_parts.append(" # Execute action if specified")
code_parts.append(" if transition['action']:")
code_parts.append(" action_method = f\"_execute_{transition['action'].lower().replace(' ', '_')}\"")
code_parts.append(" if hasattr(self, action_method):")
code_parts.append(" getattr(self, action_method)()")
code_parts.append(" ")
code_parts.append(" # Transition to new state")
code_parts.append(" self.current_state = transition['target_state']")
code_parts.append(" return True")
code_parts.append(" ")
code_parts.append(" return False")
code_parts.append("")
return '\n'.join(code_parts)
def get_file_extension(self) -> str:
return ".py"
def validate_syntax(self, code: str) -> Tuple[bool, List[str]]:
"""Validate Python syntax."""
try:
compile(code, '<string>', 'exec')
return True, []
except SyntaxError as e:
return False, [f"Syntax error at line {e.lineno}: {e.msg}"]
class JavaAdapter(LanguageAdapter):
def generate_code(self, spec: FSMSpecification, config: CodeGenerationConfig) -> str:
"""Generate Java code for the FSM specification."""
code_parts = []
# Add package and imports
if config.namespace:
code_parts.append(f"package {config.namespace};")
code_parts.append("")
code_parts.append("import java.util.HashMap;")
code_parts.append("import java.util.Map;")
code_parts.append("")
if config.include_comments:
code_parts.append("/**")
code_parts.append(" * Auto-generated Finite State Machine")
code_parts.append(" * Generated from user requirements")
code_parts.append(" */")
# Generate state enum
code_parts.append("enum State {")
state_values = []
for state in spec.states:
state_name = state.name.upper().replace(' ', '_')
state_values.append(f" {state_name}")
code_parts.append(',\n'.join(state_values))
code_parts.append("}")
code_parts.append("")
# Generate event enum
code_parts.append("enum Event {")
event_values = []
for event in spec.events:
event_name = event.name.upper().replace(' ', '_')
event_values.append(f" {event_name}")
code_parts.append(',\n'.join(event_values))
code_parts.append("}")
code_parts.append("")
# Generate main class
class_name = config.class_name
code_parts.append(f"public class {class_name} {{")
code_parts.append(" private State currentState;")
code_parts.append(" private Map<String, State> transitionTable;")
code_parts.append("")
# Constructor
code_parts.append(f" public {class_name}() {{")
if spec.initial_state:
initial_state_enum = spec.initial_state.upper().replace(' ', '_')
code_parts.append(f" this.currentState = State.{initial_state_enum};")
else:
code_parts.append(" this.currentState = null;")
code_parts.append(" this.transitionTable = buildTransitionTable();")
code_parts.append(" }")
code_parts.append("")
# Build transition table method
code_parts.append(" private Map<String, State> buildTransitionTable() {")
code_parts.append(" Map<String, State> table = new HashMap<>();")
for transition in spec.transitions:
source = transition.metadata.get('source_state')
target = transition.metadata.get('target_state')
event = transition.metadata.get('event')
if source and target and event:
source_enum = source.upper().replace(' ', '_')
target_enum = target.upper().replace(' ', '_')
event_enum = event.upper().replace(' ', '_')
key = f"State.{source_enum}:Event.{event_enum}"
code_parts.append(f" table.put(\"{key}\", State.{target_enum});")
code_parts.append(" return table;")
code_parts.append(" }")
code_parts.append("")
# Process event method
code_parts.append(" public boolean processEvent(Event event) {")
code_parts.append(" String key = currentState + \":\" + event;")
code_parts.append(" ")
code_parts.append(" if (transitionTable.containsKey(key)) {")
code_parts.append(" State oldState = currentState;")
code_parts.append(" currentState = transitionTable.get(key);")
code_parts.append(" ")
code_parts.append(" // Execute state change actions")
code_parts.append(" onStateChange(oldState, currentState);")
code_parts.append(" ")
code_parts.append(" return true;")
code_parts.append(" }")
code_parts.append(" ")
code_parts.append(" return false;")
code_parts.append(" }")
code_parts.append("")
# State change handler
code_parts.append(" protected void onStateChange(State oldState, State newState) {")
code_parts.append(" // Override this method to implement state-specific actions")
code_parts.append(" }")
code_parts.append("")
# Getter methods
code_parts.append(" public State getCurrentState() {")
code_parts.append(" return currentState;")
code_parts.append(" }")
code_parts.append("")
code_parts.append(" public boolean isInState(State state) {")
code_parts.append(" return currentState == state;")
code_parts.append(" }")
code_parts.append("}")
return '\n'.join(code_parts)
def get_file_extension(self) -> str:
return ".java"
def validate_syntax(self, code: str) -> Tuple[bool, List[str]]:
"""Basic Java syntax validation (simplified)."""
issues = []
# Check for basic syntax issues
if code.count('{') != code.count('}'):
issues.append("Mismatched braces")
if code.count('(') != code.count(')'):
issues.append("Mismatched parentheses")
# Check for missing semicolons (simplified check)
lines = code.split('\n')
for i, line in enumerate(lines):
stripped = line.strip()
if stripped and not stripped.endswith((';', '{', '}', '//', '/*', '*/', '*')):
if not stripped.startswith(('import', 'package', 'public class', 'enum', 'private', 'protected', 'public')):
issues.append(f"Line {i+1} might be missing semicolon")
return len(issues) == 0, issues
class CodeGenerator:
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.adapters = {
'python': PythonAdapter(),
'java': JavaAdapter(),
# Additional language adapters can be added here
}
def generate_code(self, spec: FSMSpecification,
language: str, style: str = 'object_oriented',
**kwargs) -> Tuple[str, str]:
"""
Generate code for the FSM specification in the specified language.
Returns tuple of (code, filename).
"""
if language.lower() not in self.adapters:
raise ValueError(f"Unsupported language: {language}")
adapter = self.adapters[language.lower()]
config = CodeGenerationConfig(
language=language.lower(),
style=style,
**kwargs
)
# Generate the code
code = adapter.generate_code(spec, config)
# Validate syntax
is_valid, issues = adapter.validate_syntax(code)
if not is_valid:
# Try to fix common issues using LLM
code = self._fix_syntax_issues(code, issues, language)
# Generate filename
filename = f"{config.class_name}{adapter.get_file_extension()}"
return code, filename
def _fix_syntax_issues(self, code: str, issues: List[str], language: str) -> str:
"""
Use LLM to fix syntax issues in generated code.
"""
fix_prompt = f"""
The following {language} code has syntax issues:
Issues found:
{chr(10).join(issues)}
Code:
{code}
Please fix the syntax issues and return the corrected code.
Only return the corrected code, no explanations.
"""
messages = [{"role": "system", "content": fix_prompt}]
response = self.llm.generate_response(messages, temperature=0.1)
return response.content
def get_supported_languages(self) -> List[str]:
"""Get list of supported programming languages."""
return list(self.adapters.keys())
def get_supported_styles(self, language: str) -> List[str]:
"""Get supported implementation styles for a language."""
# This would be expanded based on each adapter's capabilities
return ['object_oriented', 'switch', 'table_driven']
This code generation system demonstrates how to transform FSM specifications into executable code across multiple programming languages while maintaining semantic correctness and following language-specific best practices.
Running Example: Complete Agent Implementation
To demonstrate how all the components work together, let's build a complete example that implements a traffic light controller FSM. This example will show the entire flow from user input to generated code, illustrating how the agent handles requirements gathering, clarification, and code generation.
The traffic light controller is an ideal example because it's familiar to most people, has clear states and transitions, includes timing constraints, and requires safety considerations. This example will demonstrate how the agent handles real-world complexity while maintaining user-friendly interaction.
Here's the complete implementation that ties together all the components we've discussed:
import json
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from datetime import datetime
class TrafficLightFSMAgent:
"""
Complete FSM agent implementation using the traffic light controller
as a running example. This demonstrates the full workflow from
user requirements to generated code.
"""
def __init__(self, llm_integration: LLMIntegration):
self.llm = llm_integration
self.conversation_manager = ConversationManager(llm_integration)
self.requirements_analyzer = RequirementsAnalyzer(llm_integration)
self.clarification_engine = ClarificationEngine(llm_integration)
self.knowledge_base = FSMKnowledgeBase()
self.code_generator = CodeGenerator(llm_integration)
# Agent state
self.controller = FSMAgentController(llm_integration)
self.current_specification = None
self.pending_clarifications = []
self.session_log = []
def start_session(self, user_description: str) -> str:
"""
Start a new FSM generation session with the user's initial description.
This method demonstrates the complete workflow.
"""
self.session_log.append({
"timestamp": datetime.now(),
"action": "session_start",
"input": user_description
})
# Process the initial user input
response = self.controller.process_user_input(user_description)
# Log the agent's response
self.session_log.append({
"timestamp": datetime.now(),
"action": "agent_response",
"output": response
})
return response
def continue_conversation(self, user_input: str) -> str:
"""
Continue the conversation with additional user input.
This handles the iterative refinement process.
"""
self.session_log.append({
"timestamp": datetime.now(),
"action": "user_input",
"input": user_input
})
# Process the user input through the controller
response = self.controller.process_user_input(user_input)
# Log the response
self.session_log.append({
"timestamp": datetime.now(),
"action": "agent_response",
"output": response
})
return response
def generate_final_code(self, language: str, style: str = "object_oriented") -> Tuple[str, str]:
"""
Generate the final code once all requirements are gathered and clarified.
"""
if not self.controller.extracted_requirements:
raise ValueError("No requirements available for code generation")
# Convert the extracted requirements to FSM specification
spec = self._build_specification_from_requirements()
# Validate the specification
validation_issues = self.knowledge_base.validate_specification(spec)
# Filter out info-level issues for final generation
critical_issues = [issue for issue in validation_issues
if issue.severity in [ValidationSeverity.ERROR, ValidationSeverity.WARNING]]
if critical_issues:
issues_text = "\n".join([f"- {issue.message}" for issue in critical_issues])
raise ValueError(f"Specification has validation issues:\n{issues_text}")
# Generate the code
code, filename = self.code_generator.generate_code(spec, language, style)
# Log the generation
self.session_log.append({
"timestamp": datetime.now(),
"action": "code_generation",
"language": language,
"style": style,
"filename": filename
})
return code, filename
def _build_specification_from_requirements(self) -> FSMSpecification:
"""
Convert the agent's extracted requirements into a formal FSM specification.
This demonstrates how to bridge between natural language processing
and formal specification.
"""
requirements = self.controller.extracted_requirements
spec = FSMSpecification()
# Extract states from requirements
if 'states' in requirements:
for state_data in requirements['states']:
if isinstance(state_data, dict):
state = FSMElement(
name=state_data.get('name', ''),
type='state',
confidence=state_data.get('confidence', 0.8),
source_text=state_data.get('source_text', ''),
metadata=state_data
)
spec.states.append(state)
# Extract events from requirements
if 'events' in requirements:
for event_data in requirements['events']:
if isinstance(event_data, dict):
event = FSMElement(
name=event_data.get('name', ''),
type='event',
confidence=event_data.get('confidence', 0.8),
source_text=event_data.get('source_text', ''),
metadata=event_data
)
spec.events.append(event)
# Extract actions from requirements
if 'actions' in requirements:
for action_data in requirements['actions']:
if isinstance(action_data, dict):
action = FSMElement(
name=action_data.get('name', ''),
type='action',
confidence=action_data.get('confidence', 0.8),
source_text=action_data.get('source_text', ''),
metadata=action_data
)
spec.actions.append(action)
# Extract transitions from requirements
if 'transitions' in requirements:
for trans_data in requirements['transitions']:
if isinstance(trans_data, dict):
transition = FSMElement(
name=f"{trans_data.get('source', 'unknown')}_to_{trans_data.get('target', 'unknown')}",
type='transition',
confidence=trans_data.get('confidence', 0.8),
source_text=trans_data.get('description', ''),
metadata=trans_data
)
spec.transitions.append(transition)
# Set initial and final states
spec.initial_state = requirements.get('initial_state')
spec.final_states = requirements.get('final_states', [])
return spec
def demonstrate_traffic_light_example():
"""
Complete demonstration of the FSM agent using a traffic light controller example.
This shows the entire workflow from start to finish.
"""
# Initialize the agent (in practice, you'd provide a real LLM integration)
# For this example, we'll simulate the interactions
print("=== Traffic Light FSM Agent Demonstration ===")
print()
# Simulate user's initial description
user_description = """
I need to create a traffic light controller for a simple intersection.
The traffic light should cycle through red, yellow, and green lights.
The red light should stay on for 30 seconds, yellow for 5 seconds,
and green for 25 seconds. There should also be an emergency mode
where all lights flash red.
"""
print("User Input:")
print(user_description)
print()
# Simulate the agent's analysis and response
print("Agent Analysis:")
print("I've identified the following elements from your description:")
print("- States: red, yellow, green, emergency")
print("- Events: timer_expired, emergency_activated, emergency_deactivated")
print("- Actions: turn_on_red, turn_on_yellow, turn_on_green, flash_red")
print("- Timing constraints: red=30s, yellow=5s, green=25s")
print()
# Simulate clarification questions
print("Agent Clarification Questions:")
clarifications = [
"What should be the initial state when the system starts up?",
"How should the system transition out of emergency mode?",
"Should there be any manual override capabilities?",
"What should happen if a timer fails or doesn't expire?"
]
for i, question in enumerate(clarifications, 1):
print(f"{i}. {question}")
print()
# Simulate user responses to clarifications
print("User Responses:")
responses = [
"The system should start in the red state for safety.",
"Emergency mode should return to red state when deactivated.",
"Yes, there should be a manual override for maintenance.",
"If timer fails, the system should go to emergency mode."
]
for i, response in enumerate(responses, 1):
print(f"{i}. {response}")
print()
# Show the final specification
print("Final FSM Specification:")
print("States: red, yellow, green, emergency, maintenance")
print("Events: timer_expired, emergency_activated, emergency_deactivated, manual_override, timer_fault")
print("Initial State: red")
print("Transitions:")
transitions = [
"red + timer_expired -> green",
"green + timer_expired -> yellow",
"yellow + timer_expired -> red",
"any + emergency_activated -> emergency",
"emergency + emergency_deactivated -> red",
"any + manual_override -> maintenance",
"any + timer_fault -> emergency"
]
for transition in transitions:
print(f" {transition}")
print()
# Generate sample code
print("Generated Python Code:")
print("```python")
sample_code = '''
from enum import Enum
from typing import Dict, Callable, Optional
import time
import threading
class State(Enum):
RED = 'red'
YELLOW = 'yellow'
GREEN = 'green'
EMERGENCY = 'emergency'
MAINTENANCE = 'maintenance'
class Event(Enum):
TIMER_EXPIRED = 'timer_expired'
EMERGENCY_ACTIVATED = 'emergency_activated'
EMERGENCY_DEACTIVATED = 'emergency_deactivated'
MANUAL_OVERRIDE = 'manual_override'
TIMER_FAULT = 'timer_fault'
class TrafficLightController:
def __init__(self):
self.current_state = State.RED
self.timer = None
self.state_durations = {
State.RED: 30.0,
State.YELLOW: 5.0,
State.GREEN: 25.0
}
self.transition_table = self._build_transition_table()
self.state_actions = self._build_state_actions()
def _build_transition_table(self) -> Dict[tuple, State]:
return {
(State.RED, Event.TIMER_EXPIRED): State.GREEN,
(State.GREEN, Event.TIMER_EXPIRED): State.YELLOW,
(State.YELLOW, Event.TIMER_EXPIRED): State.RED,
# Emergency transitions from any state
(State.RED, Event.EMERGENCY_ACTIVATED): State.EMERGENCY,
(State.YELLOW, Event.EMERGENCY_ACTIVATED): State.EMERGENCY,
(State.GREEN, Event.EMERGENCY_ACTIVATED): State.EMERGENCY,
(State.MAINTENANCE, Event.EMERGENCY_ACTIVATED): State.EMERGENCY,
# Return from emergency
(State.EMERGENCY, Event.EMERGENCY_DEACTIVATED): State.RED,
# Manual override transitions
(State.RED, Event.MANUAL_OVERRIDE): State.MAINTENANCE,
(State.YELLOW, Event.MANUAL_OVERRIDE): State.MAINTENANCE,
(State.GREEN, Event.MANUAL_OVERRIDE): State.MAINTENANCE,
# Timer fault handling
(State.RED, Event.TIMER_FAULT): State.EMERGENCY,
(State.YELLOW, Event.TIMER_FAULT): State.EMERGENCY,
(State.GREEN, Event.TIMER_FAULT): State.EMERGENCY,
}
def _build_state_actions(self) -> Dict[State, Callable]:
return {
State.RED: self._on_red,
State.YELLOW: self._on_yellow,
State.GREEN: self._on_green,
State.EMERGENCY: self._on_emergency,
State.MAINTENANCE: self._on_maintenance,
}
def process_event(self, event: Event) -> bool:
transition_key = (self.current_state, event)
if transition_key in self.transition_table:
old_state = self.current_state
self.current_state = self.transition_table[transition_key]
print(f"Transition: {old_state.value} -> {self.current_state.value} (event: {event.value})")
# Cancel existing timer
if self.timer:
self.timer.cancel()
# Execute new state action
if self.current_state in self.state_actions:
self.state_actions[self.current_state]()
return True
return False
def _on_red(self):
print("RED light ON - Stop")
self._start_timer(self.state_durations[State.RED])
def _on_yellow(self):
print("YELLOW light ON - Caution")
self._start_timer(self.state_durations[State.YELLOW])
def _on_green(self):
print("GREEN light ON - Go")
self._start_timer(self.state_durations[State.GREEN])
def _on_emergency(self):
print("EMERGENCY mode - All lights flashing RED")
# In real implementation, this would control the flashing
def _on_maintenance(self):
print("MAINTENANCE mode - Manual control active")
def _start_timer(self, duration: float):
self.timer = threading.Timer(duration, self._timer_expired)
self.timer.start()
def _timer_expired(self):
self.process_event(Event.TIMER_EXPIRED)
def activate_emergency(self):
self.process_event(Event.EMERGENCY_ACTIVATED)
def deactivate_emergency(self):
self.process_event(Event.EMERGENCY_DEACTIVATED)
def manual_override(self):
self.process_event(Event.MANUAL_OVERRIDE)
def get_current_state(self) -> State:
return self.current_state
# Example usage
if __name__ == "__main__":
controller = TrafficLightController()
# Let it run for a few cycles
time.sleep(70) # Watch it cycle through states
# Test emergency mode
controller.activate_emergency()
time.sleep(5)
controller.deactivate_emergency()
'''
print(sample_code)
print("```")
print()
print("=== Demonstration Complete ===")
print("The agent successfully:")
print("1. Analyzed natural language requirements")
print("2. Identified FSM elements (states, events, actions)")
print("3. Asked clarifying questions to resolve ambiguities")
print("4. Built a complete FSM specification")
print("5. Generated executable code in the target language")
print("6. Included proper error handling and safety features")
def run_interactive_demo():
"""
Interactive demonstration that shows how the agent would work
with real user input. This simulates the conversation flow.
"""
print("=== Interactive FSM Agent Demo ===")
print("Describe the system you want to model as a finite state machine:")
print()
# Simulate different user scenarios
scenarios = [
{
"name": "Door Controller",
"description": "I want to control an automatic door. It should open when someone approaches, stay open for a while, then close. It should also handle obstacles.",
"clarifications": [
"How long should the door stay open?",
"What should happen if an obstacle is detected while closing?",
"Should there be manual override controls?"
]
},
{
"name": "Washing Machine",
"description": "Create a washing machine controller that has different wash cycles, handles water filling and draining, and includes safety features.",
"clarifications": [
"What are the different wash cycles?",
"How should safety interlocks work?",
"What should happen if there's a power failure?"
]
},
{
"name": "Game State Manager",
"description": "I need a state machine for a simple game with menu, playing, paused, and game over states.",
"clarifications": [
"How should the game transition between states?",
"Should there be different types of game over conditions?",
"What options should be available in the pause menu?"
]
}
]
for i, scenario in enumerate(scenarios, 1):
print(f"Scenario {i}: {scenario['name']}")
print(f"User Input: {scenario['description']}")
print()
print("Agent Response:")
print("I understand you want to create a finite state machine for a", scenario['name'].lower())
print("Let me ask a few questions to clarify the requirements:")
print()
for j, question in enumerate(scenario['clarifications'], 1):
print(f" {j}. {question}")
print()
print("After gathering this information, I would generate the appropriate")
print("FSM specification and code in your preferred programming language.")
print()
print("-" * 60)
print()
# Testing and Quality Assurance
class FSMAgentTester:
"""
Comprehensive testing framework for the FSM agent.
This demonstrates how to validate agent behavior and output quality.
"""
def __init__(self, agent: TrafficLightFSMAgent):
self.agent = agent
self.test_cases = self._load_test_cases()
self.results = []
def _load_test_cases(self) -> List[Dict]:
"""
Load predefined test cases for different scenarios.
In practice, these would be loaded from external files.
"""
return [
{
"name": "Simple Traffic Light",
"description": "Basic traffic light with red, yellow, green states",
"expected_states": ["red", "yellow", "green"],
"expected_events": ["timer_expired"],
"expected_initial": "red"
},
{
"name": "Door Controller",
"description": "Automatic door with sensor and timer",
"expected_states": ["closed", "opening", "open", "closing"],
"expected_events": ["sensor_triggered", "timer_expired", "obstacle_detected"],
"expected_initial": "closed"
},
{
"name": "Protocol Handler",
"description": "Network protocol with connection states",
"expected_states": ["disconnected", "connecting", "connected", "error"],
"expected_events": ["connect", "connected", "disconnect", "error"],
"expected_initial": "disconnected"
}
]
def run_all_tests(self) -> Dict[str, Any]:
"""
Run all test cases and return comprehensive results.
"""
results = {
"total_tests": len(self.test_cases),
"passed": 0,
"failed": 0,
"details": []
}
for test_case in self.test_cases:
result = self.run_single_test(test_case)
results["details"].append(result)
if result["passed"]:
results["passed"] += 1
else:
results["failed"] += 1
return results
def run_single_test(self, test_case: Dict) -> Dict[str, Any]:
"""
Run a single test case and validate the results.
"""
test_result = {
"name": test_case["name"],
"passed": False,
"issues": [],
"extracted_elements": {}
}
try:
# Simulate running the agent with the test description
# In practice, this would actually call the agent
response = self._simulate_agent_response(test_case["description"])
# Validate the response
validation_results = self._validate_response(response, test_case)
test_result["passed"] = len(validation_results) == 0
test_result["issues"] = validation_results
test_result["extracted_elements"] = response
except Exception as e:
test_result["issues"].append(f"Test execution failed: {str(e)}")
return test_result
def _simulate_agent_response(self, description: str) -> Dict:
"""
Simulate the agent's response to a test description.
In a real implementation, this would call the actual agent.
"""
# This is a simplified simulation
# Real implementation would use the actual agent
if "traffic light" in description.lower():
return {
"states": ["red", "yellow", "green"],
"events": ["timer_expired"],
"initial_state": "red",
"transitions": [
{"source": "red", "event": "timer_expired", "target": "green"},
{"source": "green", "event": "timer_expired", "target": "yellow"},
{"source": "yellow", "event": "timer_expired", "target": "red"}
]
}
elif "door" in description.lower():
return {
"states": ["closed", "opening", "open", "closing"],
"events": ["sensor_triggered", "timer_expired", "obstacle_detected"],
"initial_state": "closed",
"transitions": [
{"source": "closed", "event": "sensor_triggered", "target": "opening"},
{"source": "opening", "event": "timer_expired", "target": "open"},
{"source": "open", "event": "timer_expired", "target": "closing"},
{"source": "closing", "event": "obstacle_detected", "target": "opening"}
]
}
else:
return {
"states": ["state1", "state2"],
"events": ["event1"],
"initial_state": "state1",
"transitions": []
}
def _validate_response(self, response: Dict, expected: Dict) -> List[str]:
"""
Validate the agent's response against expected results.
"""
issues = []
# Check states
extracted_states = response.get("states", [])
expected_states = expected.get("expected_states", [])
for state in expected_states:
if state not in extracted_states:
issues.append(f"Missing expected state: {state}")
# Check events
extracted_events = response.get("events", [])
expected_events = expected.get("expected_events", [])
for event in expected_events:
if event not in extracted_events:
issues.append(f"Missing expected event: {event}")
# Check initial state
extracted_initial = response.get("initial_state")
expected_initial = expected.get("expected_initial")
if expected_initial and extracted_initial != expected_initial:
issues.append(f"Wrong initial state: expected {expected_initial}, got {extracted_initial}")
return issues
def generate_test_report(self, results: Dict) -> str:
"""
Generate a comprehensive test report.
"""
report_lines = []
report_lines.append("=== FSM Agent Test Report ===")
report_lines.append(f"Total Tests: {results['total_tests']}")
report_lines.append(f"Passed: {results['passed']}")
report_lines.append(f"Failed: {results['failed']}")
report_lines.append(f"Success Rate: {(results['passed'] / results['total_tests']) * 100:.1f}%")
report_lines.append("")
for detail in results["details"]:
report_lines.append(f"Test: {detail['name']}")
report_lines.append(f"Status: {'PASSED' if detail['passed'] else 'FAILED'}")
if detail["issues"]:
report_lines.append("Issues:")
for issue in detail["issues"]:
report_lines.append(f" - {issue}")
report_lines.append("")
return "\n".join(report_lines)
# Deployment and Integration Considerations
class FSMAgentDeployment:
"""
Handles deployment and integration aspects of the FSM agent.
This demonstrates production considerations.
"""
def __init__(self, agent: TrafficLightFSMAgent):
self.agent = agent
self.performance_metrics = {}
self.error_log = []
def deploy_as_web_service(self, port: int = 8000):
"""
Deploy the agent as a web service using a REST API.
This shows how to make the agent accessible to other applications.
"""
# This would use a web framework like Flask or FastAPI
# Simplified example structure:
api_endpoints = {
"POST /api/fsm/start": "Start new FSM generation session",
"POST /api/fsm/continue": "Continue conversation",
"POST /api/fsm/generate": "Generate final code",
"GET /api/fsm/status": "Get session status",
"GET /api/fsm/history": "Get conversation history"
}
print("FSM Agent Web Service Endpoints:")
for endpoint, description in api_endpoints.items():
print(f" {endpoint} - {description}")
print(f"\nService would be available at http://localhost:{port}")
def integrate_with_ide(self, ide_name: str):
"""
Integration considerations for IDE plugins.
"""
integration_points = {
"vscode": {
"extension_manifest": "package.json",
"activation_events": ["onCommand:fsm-agent.start"],
"commands": ["Generate FSM", "Refine Requirements", "Export Code"],
"views": ["FSM Explorer", "Requirements Panel"]
},
"intellij": {
"plugin_descriptor": "plugin.xml",
"actions": ["FSMGeneratorAction", "RequirementsAnalyzer"],
"tool_windows": ["FSM Designer", "Agent Chat"],
"file_types": [".fsm", ".requirements"]
}
}
if ide_name.lower() in integration_points:
config = integration_points[ide_name.lower()]
print(f"Integration configuration for {ide_name}:")
for key, value in config.items():
print(f" {key}: {value}")
else:
print(f"Integration not yet defined for {ide_name}")
def setup_monitoring(self):
"""
Set up monitoring and logging for production deployment.
"""
monitoring_aspects = [
"Response time metrics",
"LLM API usage and costs",
"User satisfaction scores",
"Error rates and types",
"Generated code quality metrics",
"Conversation completion rates"
]
print("Production Monitoring Setup:")
for aspect in monitoring_aspects:
print(f" - {aspect}")
def handle_scaling(self):
"""
Considerations for scaling the agent to handle multiple users.
"""
scaling_strategies = {
"horizontal_scaling": "Multiple agent instances behind load balancer",
"session_management": "Persistent storage for conversation state",
"caching": "Cache common FSM patterns and code templates",
"rate_limiting": "Prevent abuse of LLM API calls",
"async_processing": "Handle long-running code generation asynchronously"
}
print("Scaling Strategies:")
for strategy, description in scaling_strategies.items():
print(f" {strategy}: {description}")
if __name__ == "__main__":
# Run the demonstrations
print("Running FSM Agent Demonstrations...")
print()
# Show the complete traffic light example
demonstrate_traffic_light_example()
print()
# Show interactive scenarios
run_interactive_demo()
print()
# Note about testing and deployment
print("=== Additional Considerations ===")
print()
print("Testing Framework:")
print("- Automated test cases for different FSM types")
print("- Validation of generated code syntax and semantics")
print("- Performance benchmarking for response times")
print("- User acceptance testing for conversation quality")
print()
print("Deployment Options:")
print("- Web service with REST API")
print("- IDE plugin integration")
print("- Command-line tool")
print("- Embedded in larger development platforms")
print()
print("Production Considerations:")
print("- Monitoring and logging")
print("- Scaling for multiple concurrent users")
print("- Security and input validation")
print("- Cost management for LLM API usage")
print("- Backup and recovery procedures")
This complete running example demonstrates how all the components work together to create a functional FSM generation agent. The traffic light controller serves as an excellent illustration because it showcases state management, timing constraints, safety considerations, and real-world complexity while remaining understandable.
Conclusion and Best Practices
Building an LLM-based agent for FSM generation requires careful attention to several key areas. The architecture must balance the flexibility of natural language processing with the precision required for formal system specification. The conversation management system needs to maintain context while guiding users through the requirements gathering process efficiently.
The requirements analysis pipeline represents the core intelligence of the system, transforming informal descriptions into structured specifications. This requires sophisticated natural language processing combined with domain knowledge about finite state machines. The clarification engine ensures that ambiguities are resolved before code generation, preventing errors and misunderstandings.
Code generation must handle multiple programming languages and implementation styles while maintaining semantic correctness. The validation system catches potential issues early, improving the quality of generated code and reducing debugging time for users.
Testing and quality assurance are crucial for building confidence in the agent's outputs. Comprehensive test suites should cover various FSM types, edge cases, and error conditions. Performance monitoring helps identify bottlenecks and optimization opportunities.
Deployment considerations include scalability, security, and integration with existing development workflows. The agent should be accessible through multiple interfaces while maintaining consistent behavior and quality.
The key to success lies in the iterative refinement of each component based on real user feedback and continuous improvement of the underlying models and algorithms. The agent should evolve to handle increasingly complex scenarios while maintaining ease of use for both novice and expert users.
This approach to building LLM-based agents can be extended to other code generation tasks, providing a foundation for intelligent development tools that bridge the gap between human intent and formal system specification.
No comments:
Post a Comment