INTRODUCTION TO THE PROBLEM DOMAIN
The challenge of creating artificial intelligence systems that can dynamically generate specialized conversational agents based on natural language descriptions represents one of the most compelling applications of modern large language models. This tutorial explores the architecture, implementation strategies, and practical considerations for building a production-ready system that transforms user requirements into fully functional LLM-powered bots.
The core concept revolves around meta-prompting, where an orchestrating LLM analyzes user specifications and generates appropriate system prompts, user prompts, or even complete application code that implements the desired functionality. The generated bot then engages with end users according to its specialized configuration. This approach enables rapid prototyping and deployment of domain-specific AI assistants without requiring extensive manual prompt engineering for each use case.
The system we will construct supports both local and remote language model backends, accommodating various hardware acceleration platforms including NVIDIA CUDA, AMD ROCm, Intel GPUs, and Apple Metal Performance Shaders. This flexibility ensures deployment across diverse infrastructure environments while maintaining consistent functionality.
ARCHITECTURAL FOUNDATIONS AND DESIGN PRINCIPLES
The bot generator system consists of several interconnected components that work together to transform user intent into executable AI agents. The primary architectural layers include the specification parser, the prompt synthesis engine, the bot instantiation framework, and the runtime execution environment.
The specification parser receives natural language descriptions from users and extracts key requirements such as the bot's domain expertise, interaction style, constraints, and expected capabilities. This component employs structured prompting techniques to ensure comprehensive requirement capture while maintaining flexibility for diverse use cases.
The prompt synthesis engine represents the intellectual core of the system. It takes parsed specifications and generates optimized system prompts that encode the desired behavior, knowledge boundaries, and interaction patterns. For more complex requirements, this engine can alternatively generate complete application scaffolding including conversation management, state tracking, and integration points.
The bot instantiation framework manages the lifecycle of generated bots, handling initialization, configuration, and resource allocation. This component abstracts the differences between local and remote LLM backends, providing a unified interface regardless of the underlying model provider or hardware acceleration platform.
The runtime execution environment provides the infrastructure for bot operation, including conversation state management, context window optimization, and response generation. This layer ensures consistent performance across different deployment scenarios while maintaining isolation between concurrently running bot instances.
HARDWARE ACCELERATION AND MODEL BACKEND ABSTRACTION
Supporting multiple GPU architectures and both local and remote LLM deployments requires careful abstraction of hardware-specific details. The system employs a provider pattern that encapsulates backend-specific initialization, inference, and resource management behind a common interface.
For NVIDIA CUDA environments, the system leverages libraries such as PyTorch with CUDA support or direct integration with frameworks like vLLM or TensorRT-LLM for optimized inference. The initialization process detects available CUDA devices and configures memory allocation strategies appropriate for the model size and available VRAM.
AMD ROCm support follows a similar pattern but requires ROCm-specific PyTorch builds and may involve different optimization strategies due to architectural differences in AMD GPUs. The abstraction layer handles these variations transparently, selecting appropriate kernel implementations and memory management approaches.
Apple Metal Performance Shaders provide acceleration on macOS and iOS devices through the MLX framework or Metal-enabled PyTorch. The system detects Apple Silicon processors and configures unified memory access patterns that leverage the integrated architecture of these processors.
Intel GPU support utilizes Intel Extension for PyTorch or OpenVINO for inference acceleration. The abstraction layer manages the specific requirements of Intel's discrete and integrated graphics processors, ensuring efficient utilization of available compute resources.
Remote LLM backends connect to API services such as OpenAI, Anthropic, or self-hosted inference servers. The abstraction layer implements retry logic, rate limiting, and failover mechanisms to ensure robust operation in production environments. Authentication, request formatting, and response parsing are standardized across different API providers.
Here is a foundational code example demonstrating the backend abstraction interface:
class LLMBackend:
"""Abstract base class for LLM backend implementations."""
def initialize(self, model_name, config):
"""Initialize the backend with specified model and configuration.
Args:
model_name: Identifier for the model to load
config: Backend-specific configuration parameters
"""
raise NotImplementedError
def generate(self, prompt, system_prompt=None, max_tokens=2048,
temperature=0.7, stop_sequences=None):
"""Generate a response from the language model.
Args:
prompt: User input or conversation history
system_prompt: System-level instructions for the model
max_tokens: Maximum tokens to generate
temperature: Sampling temperature for generation
stop_sequences: Sequences that terminate generation
Returns:
Generated text response
"""
raise NotImplementedError
def cleanup(self):
"""Release resources and perform cleanup operations."""
raise NotImplementedError
class CUDABackend(LLMBackend):
"""NVIDIA CUDA-accelerated local LLM backend."""
def __init__(self):
self.model = None
self.tokenizer = None
self.device = None
def initialize(self, model_name, config):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
# Detect and configure CUDA device
if not torch.cuda.is_available():
raise RuntimeError("CUDA not available on this system")
self.device = torch.device("cuda")
device_props = torch.cuda.get_device_properties(0)
print(f"Initializing on {device_props.name} with {device_props.total_memory / 1e9:.2f} GB memory")
# Load model with appropriate precision based on available memory
load_in_8bit = config.get("quantize_8bit", False)
load_in_4bit = config.get("quantize_4bit", False)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
load_in_8bit=load_in_8bit,
load_in_4bit=load_in_4bit,
torch_dtype=torch.float16 if not (load_in_8bit or load_in_4bit) else None
)
self.model.eval()
def generate(self, prompt, system_prompt=None, max_tokens=2048,
temperature=0.7, stop_sequences=None):
import torch
# Construct full prompt with system message if provided
full_prompt = prompt
if system_prompt:
full_prompt = f"{system_prompt}\n\n{prompt}"
# Tokenize input
inputs = self.tokenizer(full_prompt, return_tensors="pt").to(self.device)
# Configure stopping criteria
stopping_criteria = None
if stop_sequences:
from transformers import StoppingCriteriaList, StoppingCriteria
class StopOnTokens(StoppingCriteria):
def __init__(self, stop_token_ids):
self.stop_token_ids = stop_token_ids
def __call__(self, input_ids, scores, **kwargs):
for stop_id in self.stop_token_ids:
if input_ids[0][-1] == stop_id:
return True
return False
stop_token_ids = [self.tokenizer.encode(seq, add_special_tokens=False)[0]
for seq in stop_sequences]
stopping_criteria = StoppingCriteriaList([StopOnTokens(stop_token_ids)])
# Generate response
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
stopping_criteria=stopping_criteria,
pad_token_id=self.tokenizer.eos_token_id
)
# Decode and return only the generated portion
generated_text = self.tokenizer.decode(outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True)
return generated_text
def cleanup(self):
import torch
if self.model:
del self.model
if self.tokenizer:
del self.tokenizer
torch.cuda.empty_cache()
This abstraction pattern extends to other backends with similar structure but platform-specific implementations. The key insight is that all backends expose identical interfaces while handling hardware-specific optimizations internally.
SPECIFICATION PARSING AND REQUIREMENT EXTRACTION
The process of transforming natural language specifications into structured bot configurations requires sophisticated prompt engineering and response parsing. The specification parser employs a multi-stage approach that first extracts explicit requirements, then infers implicit constraints, and finally validates the completeness of the specification.
The initial extraction stage uses carefully crafted prompts that guide the LLM to identify key attributes of the desired bot. These attributes include the bot's primary purpose, domain knowledge requirements, interaction style preferences, ethical boundaries, output format constraints, and any specialized capabilities such as code generation or data analysis.
Consider a user specification such as "I need a bot that helps software developers debug Python code by analyzing error messages and suggesting fixes." The parser must extract several key elements from this description. The primary purpose is debugging assistance. The domain knowledge encompasses Python programming, common error patterns, and debugging methodologies. The interaction style should be technical but helpful. The specialized capability involves code analysis and generation.
The extraction prompt might look like this:
extraction_prompt = """Analyze the following bot specification and extract structured requirements:
User Specification: {user_spec}
Extract and provide the following information in a structured format:
1. Primary Purpose: What is the main function this bot should perform?
2. Domain Knowledge: What specific knowledge domains must the bot understand?
3. Interaction Style: How should the bot communicate with users (formal, casual, technical, etc.)?
4. Constraints: What should the bot NOT do or what boundaries should it respect?
5. Special Capabilities: Does the bot need any specific abilities like code generation, data analysis, or file processing?
6. Expected Inputs: What kind of inputs will users provide?
7. Expected Outputs: What format and type of responses should the bot generate?
8. Success Criteria: How can we determine if the bot is performing correctly?
Provide detailed answers for each category based on the specification."""
The inference stage addresses implicit requirements that users may not explicitly state but are necessary for proper bot operation. For instance, a debugging assistant should maintain conversation context to track the evolution of debugging attempts, should ask clarifying questions when error messages are ambiguous, and should provide explanations alongside code suggestions to facilitate learning.
The validation stage ensures that the extracted requirements are sufficient to generate a functional bot. This involves checking for contradictions, identifying missing critical information, and prompting the user for clarification when necessary. The validation process might detect that a specification lacks information about how the bot should handle multiple programming languages or whether it should focus on syntax errors versus logic errors.
Here is an implementation of the specification parser:
class SpecificationParser:
"""Parses natural language bot specifications into structured requirements."""
def __init__(self, llm_backend):
self.backend = llm_backend
self.extraction_template = """Analyze the following bot specification and extract structured requirements:
User Specification: {user_spec}
Extract and provide the following information in JSON format:
{{
"primary_purpose": "Main function of the bot",
"domain_knowledge": ["List of knowledge domains"],
"interaction_style": "Communication style description",
"constraints": ["Things the bot should not do"],
"special_capabilities": ["Specific abilities needed"],
"expected_inputs": ["Types of user inputs"],
"expected_outputs": ["Types of bot responses"],
"success_criteria": ["Measurable success indicators"]
}}
Provide detailed, specific information for each field based on the specification."""
def parse(self, user_specification):
"""Parse user specification into structured requirements.
Args:
user_specification: Natural language description of desired bot
Returns:
Dictionary containing structured requirements
"""
# Extract explicit requirements
extraction_prompt = self.extraction_template.format(user_spec=user_specification)
extraction_response = self.backend.generate(
extraction_prompt,
system_prompt="You are an expert at analyzing requirements and extracting structured information. Always respond with valid JSON.",
temperature=0.3
)
# Parse JSON response
import json
import re
# Extract JSON from response (handling potential markdown formatting)
json_match = re.search(r'\{.*\}', extraction_response, re.DOTALL)
if json_match:
requirements = json.loads(json_match.group())
else:
raise ValueError("Failed to extract structured requirements from specification")
# Infer implicit requirements
implicit_requirements = self._infer_implicit_requirements(user_specification, requirements)
requirements['implicit_requirements'] = implicit_requirements
# Validate completeness
validation_issues = self._validate_requirements(requirements)
requirements['validation_issues'] = validation_issues
return requirements
def _infer_implicit_requirements(self, original_spec, explicit_requirements):
"""Infer requirements not explicitly stated by the user."""
inference_prompt = f"""Given this bot specification and extracted requirements, identify important implicit requirements that weren't explicitly stated but are necessary for the bot to function well.
Original Specification: {original_spec}
Explicit Requirements: {json.dumps(explicit_requirements, indent=2)}
Identify implicit requirements in these categories:
- Context Management: How should the bot handle conversation history?
- Error Handling: How should the bot respond to unclear or invalid inputs?
- Clarification: When should the bot ask for more information?
- Scope Boundaries: What related tasks are outside the bot's scope?
- User Experience: What makes interactions smooth and helpful?
Provide a JSON object with these categories as keys and lists of implicit requirements as values."""
inference_response = self.backend.generate(
inference_prompt,
system_prompt="You are an expert at identifying unstated but important requirements. Respond with valid JSON.",
temperature=0.4
)
import json
import re
json_match = re.search(r'\{.*\}', inference_response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
return {}
def _validate_requirements(self, requirements):
"""Validate that requirements are complete and consistent."""
validation_prompt = f"""Review these bot requirements for completeness and consistency:
{json.dumps(requirements, indent=2)}
Identify any issues:
- Missing critical information needed to build the bot
- Contradictions between different requirements
- Ambiguities that need clarification
- Unrealistic or impossible requirements
Provide a JSON array of issue objects, each with 'type' (missing/contradiction/ambiguity/unrealistic) and 'description' fields. If no issues, return an empty array."""
validation_response = self.backend.generate(
validation_prompt,
system_prompt="You are an expert at requirements validation. Respond with valid JSON.",
temperature=0.2
)
import json
import re
json_match = re.search(r'\[.*\]', validation_response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
return []
This parser provides the foundation for understanding user intent and translating it into actionable bot configurations.
PROMPT SYNTHESIS AND BOT GENERATION
Once requirements are extracted and validated, the system must synthesize effective prompts that encode the desired bot behavior. This synthesis process represents the creative core of the bot generator, transforming abstract specifications into concrete instructions that guide LLM behavior.
The prompt synthesis engine employs several strategies depending on the complexity of the requirements. For straightforward bots with well-defined purposes, a carefully crafted system prompt may suffice. More complex bots might require conversation templates, few-shot examples, or even complete application code that orchestrates multiple LLM calls with intermediate processing.
The system prompt synthesis process begins by establishing the bot's identity and core purpose. This identity statement provides the foundation for all subsequent behavior. For our Python debugging assistant example, the identity might be "You are an expert Python debugging assistant with deep knowledge of common error patterns, best practices, and effective debugging strategies."
The prompt then incorporates domain knowledge by specifying the areas of expertise the bot should demonstrate. This might include specific Python versions, popular libraries, common pitfalls, and debugging tools. The synthesis engine draws from the extracted requirements to populate these knowledge domains accurately.
Interaction style guidelines shape how the bot communicates. A technical debugging assistant should use precise terminology, provide code examples with proper formatting, explain reasoning behind suggestions, and maintain a helpful but professional tone. These stylistic elements are encoded as explicit instructions within the system prompt.
Constraints and boundaries prevent the bot from engaging in undesired behaviors. For a debugging assistant, constraints might include refusing to write entire applications from scratch, declining to help with malicious code, and staying focused on debugging rather than general programming tutoring. These boundaries protect both users and the system from misuse.
Special capabilities require specific instructions about how to exercise those abilities. A debugging bot needs instructions on how to format code snippets, how to structure explanations, when to ask for additional context, and how to present multiple solution alternatives.
Here is the prompt synthesis engine implementation:
class PromptSynthesizer:
"""Generates optimized prompts from structured requirements."""
def __init__(self, llm_backend):
self.backend = llm_backend
def synthesize(self, requirements):
"""Generate system and user prompts from requirements.
Args:
requirements: Structured requirements dictionary
Returns:
Dictionary containing system_prompt, initial_user_prompt, and metadata
"""
# Determine synthesis strategy based on complexity
complexity_score = self._assess_complexity(requirements)
if complexity_score < 3:
return self._synthesize_simple_prompt(requirements)
elif complexity_score < 7:
return self._synthesize_structured_prompt(requirements)
else:
return self._synthesize_application_code(requirements)
def _assess_complexity(self, requirements):
"""Assess requirement complexity on a scale of 0-10."""
score = 0
# Add points for multiple domains
score += min(len(requirements.get('domain_knowledge', [])), 3)
# Add points for special capabilities
score += min(len(requirements.get('special_capabilities', [])) * 2, 4)
# Add points for complex constraints
score += min(len(requirements.get('constraints', [])), 2)
# Add points for validation issues (indicates complexity)
score += min(len(requirements.get('validation_issues', [])), 1)
return score
def _synthesize_simple_prompt(self, requirements):
"""Generate a straightforward system prompt for simple bots."""
synthesis_prompt = f"""Create a concise, effective system prompt for an LLM bot with these requirements:
Primary Purpose: {requirements['primary_purpose']}
Domain Knowledge: {', '.join(requirements['domain_knowledge'])}
Interaction Style: {requirements['interaction_style']}
Constraints: {', '.join(requirements['constraints'])}
The system prompt should:
- Clearly establish the bot's identity and purpose
- Specify relevant knowledge domains
- Define interaction style and tone
- State important constraints and boundaries
- Be concise but comprehensive (aim for 150-300 words)
Generate only the system prompt text, without additional commentary."""
system_prompt = self.backend.generate(
synthesis_prompt,
system_prompt="You are an expert prompt engineer who creates highly effective system prompts for LLM applications.",
temperature=0.5
)
# Generate initial greeting
greeting_prompt = f"""Based on this system prompt, create a brief, friendly initial greeting that the bot should use when first interacting with users:
System Prompt: {system_prompt}
The greeting should:
- Welcome the user
- Briefly explain what the bot can help with
- Invite the user to describe their needs
- Be warm but professional
- Be 2-3 sentences maximum
Generate only the greeting text."""
initial_greeting = self.backend.generate(
greeting_prompt,
system_prompt="You are an expert at crafting welcoming, effective bot greetings.",
temperature=0.6
)
return {
'type': 'simple_prompt',
'system_prompt': system_prompt.strip(),
'initial_user_prompt': initial_greeting.strip(),
'metadata': {
'complexity': 'simple',
'requirements': requirements
}
}
def _synthesize_structured_prompt(self, requirements):
"""Generate a detailed structured prompt with examples and templates."""
synthesis_prompt = f"""Create a comprehensive system prompt for an LLM bot with these detailed requirements:
Primary Purpose: {requirements['primary_purpose']}
Domain Knowledge: {json.dumps(requirements['domain_knowledge'], indent=2)}
Interaction Style: {requirements['interaction_style']}
Constraints: {json.dumps(requirements['constraints'], indent=2)}
Special Capabilities: {json.dumps(requirements['special_capabilities'], indent=2)}
Expected Inputs: {json.dumps(requirements['expected_inputs'], indent=2)}
Expected Outputs: {json.dumps(requirements['expected_outputs'], indent=2)}
The system prompt should include:
1. Identity and Purpose Statement
2. Detailed Knowledge Domain Specifications
3. Interaction Style Guidelines with examples
4. Clear Constraint Boundaries
5. Instructions for exercising special capabilities
6. Input handling procedures
7. Output formatting requirements
8. Few-shot examples demonstrating ideal interactions
Create a thorough, well-structured system prompt (aim for 400-800 words)."""
system_prompt = self.backend.generate(
synthesis_prompt,
system_prompt="You are an expert prompt engineer specializing in complex, multi-faceted LLM applications.",
temperature=0.4,
max_tokens=3000
)
# Generate conversation template
template_prompt = f"""Based on this system prompt, create a conversation flow template that guides how the bot should structure interactions:
System Prompt: {system_prompt}
The template should define:
1. Initial greeting and capability explanation
2. Information gathering questions (if needed)
3. Processing and response generation approach
4. Follow-up and clarification patterns
5. Conversation conclusion strategies
Format as a structured guide the bot can follow."""
conversation_template = self.backend.generate(
template_prompt,
system_prompt="You are an expert at designing effective conversation flows for AI assistants.",
temperature=0.4,
max_tokens=2000
)
return {
'type': 'structured_prompt',
'system_prompt': system_prompt.strip(),
'conversation_template': conversation_template.strip(),
'initial_user_prompt': "Hello! I'm ready to assist you. How can I help you today?",
'metadata': {
'complexity': 'structured',
'requirements': requirements
}
}
def _synthesize_application_code(self, requirements):
"""Generate complete application code for complex bots."""
code_generation_prompt = f"""Generate a complete Python application that implements a bot with these requirements:
{json.dumps(requirements, indent=2)}
The application should include:
1. A main Bot class that encapsulates all functionality
2. Proper initialization with configuration
3. Conversation state management
4. Input validation and processing
5. Response generation with the specified capabilities
6. Error handling and graceful degradation
7. Logging and debugging support
8. Clean separation of concerns
Generate production-ready, well-documented Python code with proper error handling.
Include all necessary imports and class definitions.
Use type hints and docstrings throughout."""
application_code = self.backend.generate(
code_generation_prompt,
system_prompt="You are an expert Python developer who writes clean, production-ready code following best practices.",
temperature=0.3,
max_tokens=4000
)
return {
'type': 'application_code',
'code': application_code.strip(),
'initial_user_prompt': "Bot initialized. Ready to assist.",
'metadata': {
'complexity': 'application',
'requirements': requirements,
'language': 'python'
}
}
The synthesizer adapts its strategy based on requirement complexity, ensuring that simple bots remain straightforward while complex bots receive the sophisticated infrastructure they need.
BOT INSTANTIATION AND LIFECYCLE MANAGEMENT
After prompt synthesis, the system must instantiate the generated bot and manage its lifecycle. This involves creating a runtime environment, initializing conversation state, handling user interactions, and managing resources efficiently.
The bot instantiation process begins by creating a Bot instance with the synthesized prompts or generated code. This instance encapsulates all state and behavior specific to that bot, allowing multiple bots to run concurrently without interference.
Conversation state management tracks the dialogue history, user preferences, and any accumulated context that informs subsequent responses. For simple bots, this might be a straightforward message history. Complex bots might maintain structured state including extracted entities, user goals, and intermediate computation results.
The interaction loop handles the cycle of receiving user input, processing it through the bot's logic, generating responses, and updating state. This loop must handle various edge cases including empty inputs, excessively long messages, requests outside the bot's scope, and potential errors during generation.
Resource management ensures that bots don't consume excessive memory or compute resources. This includes implementing conversation history truncation strategies, managing model context windows, and cleaning up resources when bots are no longer needed.
Here is the bot instantiation framework:
class GeneratedBot:
"""Runtime instance of a generated bot."""
def __init__(self, bot_config, llm_backend):
"""Initialize a bot instance.
Args:
bot_config: Configuration dictionary from prompt synthesis
llm_backend: LLM backend for generation
"""
self.config = bot_config
self.backend = llm_backend
self.conversation_history = []
self.state = {}
self.max_history_length = 20 # Prevent unbounded memory growth
# Initialize based on bot type
if bot_config['type'] == 'application_code':
self._initialize_from_code()
else:
self.system_prompt = bot_config['system_prompt']
self.conversation_template = bot_config.get('conversation_template')
def _initialize_from_code(self):
"""Initialize bot from generated application code."""
# Execute generated code in isolated namespace
namespace = {}
exec(self.config['code'], namespace)
# Find and instantiate the Bot class
bot_class = None
for name, obj in namespace.items():
if isinstance(obj, type) and name.endswith('Bot'):
bot_class = obj
break
if bot_class:
self.custom_bot = bot_class(self.backend)
else:
raise ValueError("Generated code does not contain a Bot class")
def start(self):
"""Start the bot and return initial greeting."""
initial_message = self.config['initial_user_prompt']
self.conversation_history.append({
'role': 'assistant',
'content': initial_message
})
return initial_message
def process_message(self, user_message):
"""Process a user message and generate a response.
Args:
user_message: User's input text
Returns:
Bot's response text
"""
# Add user message to history
self.conversation_history.append({
'role': 'user',
'content': user_message
})
# Truncate history if too long
if len(self.conversation_history) > self.max_history_length:
# Keep system context and recent messages
self.conversation_history = self.conversation_history[-self.max_history_length:]
# Generate response based on bot type
if self.config['type'] == 'application_code':
response = self._process_with_custom_code(user_message)
else:
response = self._process_with_prompt(user_message)
# Add response to history
self.conversation_history.append({
'role': 'assistant',
'content': response
})
return response
def _process_with_custom_code(self, user_message):
"""Process message using custom generated code."""
if hasattr(self.custom_bot, 'process_message'):
return self.custom_bot.process_message(user_message)
else:
return "Error: Custom bot does not implement process_message method"
def _process_with_prompt(self, user_message):
"""Process message using synthesized prompts."""
# Construct conversation context
conversation_context = self._format_conversation_history()
# Generate response
response = self.backend.generate(
prompt=conversation_context,
system_prompt=self.system_prompt,
temperature=0.7,
max_tokens=2048
)
return response.strip()
def _format_conversation_history(self):
"""Format conversation history for model input."""
formatted = []
for message in self.conversation_history:
role = message['role']
content = message['content']
if role == 'user':
formatted.append(f"User: {content}")
elif role == 'assistant':
formatted.append(f"Assistant: {content}")
# Add current turn marker
formatted.append("Assistant:")
return "\n\n".join(formatted)
def get_state(self):
"""Get current bot state for persistence or inspection."""
return {
'config': self.config,
'conversation_history': self.conversation_history,
'state': self.state
}
def cleanup(self):
"""Clean up bot resources."""
self.conversation_history.clear()
self.state.clear()
if hasattr(self, 'custom_bot') and hasattr(self.custom_bot, 'cleanup'):
self.custom_bot.cleanup()
This framework provides a consistent interface for interacting with generated bots regardless of their internal complexity.
ORCHESTRATION AND COMPLETE SYSTEM INTEGRATION
The complete bot generator system orchestrates all components into a cohesive workflow. This orchestration layer manages the end-to-end process from receiving user specifications to deploying functional bots.
The orchestration begins with user input validation, ensuring that specifications contain sufficient information to proceed. The system then initializes the appropriate LLM backend based on available hardware and configuration preferences.
The specification parser analyzes the user input and extracts structured requirements. If validation issues are detected, the system engages in a clarification dialogue with the user to resolve ambiguities or gather missing information.
Once requirements are complete and validated, the prompt synthesizer generates the appropriate bot configuration. The system selects the synthesis strategy based on complexity assessment, producing either simple prompts, structured prompts with templates, or complete application code.
The bot instantiation framework creates a runtime instance of the generated bot, initializing all necessary state and resources. The bot is then ready to interact with end users according to its specialized configuration.
The orchestrator also handles error recovery, logging, and monitoring. If generation fails at any stage, the system attempts recovery strategies such as retrying with adjusted parameters, falling back to simpler synthesis approaches, or requesting user intervention.
Here is a comprehensive example demonstrating the complete orchestration:
class BotGenerator:
"""Main orchestrator for the bot generation system."""
def __init__(self, backend_type='cuda', model_name='mistralai/Mistral-7B-Instruct-v0.2',
backend_config=None):
"""Initialize the bot generator.
Args:
backend_type: Type of LLM backend ('cuda', 'rocm', 'mps', 'cpu', 'remote')
model_name: Model identifier to use
backend_config: Backend-specific configuration
"""
self.backend = self._initialize_backend(backend_type, model_name, backend_config or {})
self.parser = SpecificationParser(self.backend)
self.synthesizer = PromptSynthesizer(self.backend)
self.active_bots = {}
def _initialize_backend(self, backend_type, model_name, config):
"""Initialize the appropriate LLM backend."""
if backend_type == 'cuda':
backend = CUDABackend()
elif backend_type == 'rocm':
backend = ROCmBackend()
elif backend_type == 'mps':
backend = MPSBackend()
elif backend_type == 'cpu':
backend = CPUBackend()
elif backend_type == 'remote':
backend = RemoteBackend()
else:
raise ValueError(f"Unknown backend type: {backend_type}")
backend.initialize(model_name, config)
return backend
def generate_bot(self, user_specification, bot_id=None):
"""Generate a bot from user specification.
Args:
user_specification: Natural language description of desired bot
bot_id: Optional identifier for the bot (auto-generated if not provided)
Returns:
Tuple of (bot_id, bot_instance)
"""
import uuid
# Generate unique ID if not provided
if bot_id is None:
bot_id = str(uuid.uuid4())
try:
# Parse specification
print(f"Parsing specification for bot {bot_id}...")
requirements = self.parser.parse(user_specification)
# Check for validation issues
if requirements['validation_issues']:
print(f"Validation issues detected: {requirements['validation_issues']}")
# In production, might engage in clarification dialogue here
# Synthesize bot configuration
print(f"Synthesizing bot configuration...")
bot_config = self.synthesizer.synthesize(requirements)
# Instantiate bot
print(f"Instantiating bot...")
bot_instance = GeneratedBot(bot_config, self.backend)
# Store active bot
self.active_bots[bot_id] = bot_instance
print(f"Bot {bot_id} successfully generated!")
return bot_id, bot_instance
except Exception as e:
print(f"Error generating bot: {str(e)}")
import traceback
traceback.print_exc()
raise
def interact_with_bot(self, bot_id, user_message):
"""Send a message to a generated bot.
Args:
bot_id: Identifier of the bot
user_message: User's message
Returns:
Bot's response
"""
if bot_id not in self.active_bots:
raise ValueError(f"No active bot with ID {bot_id}")
bot = self.active_bots[bot_id]
return bot.process_message(user_message)
def start_bot(self, bot_id):
"""Start a bot and get its initial greeting.
Args:
bot_id: Identifier of the bot
Returns:
Initial greeting message
"""
if bot_id not in self.active_bots:
raise ValueError(f"No active bot with ID {bot_id}")
bot = self.active_bots[bot_id]
return bot.start()
def remove_bot(self, bot_id):
"""Remove a bot and clean up its resources.
Args:
bot_id: Identifier of the bot
"""
if bot_id in self.active_bots:
bot = self.active_bots[bot_id]
bot.cleanup()
del self.active_bots[bot_id]
def shutdown(self):
"""Shutdown the bot generator and clean up all resources."""
# Clean up all active bots
for bot_id in list(self.active_bots.keys()):
self.remove_bot(bot_id)
# Clean up backend
self.backend.cleanup()
This orchestration layer provides a clean API for the entire bot generation workflow.
ADVANCED FEATURES AND OPTIMIZATION STRATEGIES
Beyond the core functionality, production systems benefit from several advanced features that enhance performance, reliability, and user experience.
Caching strategies reduce redundant computation by storing frequently used prompts, parsed specifications, and generated configurations. When users request similar bots, the system can retrieve cached results rather than regenerating from scratch. Cache invalidation policies ensure that cached content remains fresh and relevant.
Prompt optimization techniques refine generated prompts through iterative testing and evaluation. The system can generate multiple prompt variations, test them against example inputs, and select the version that produces the highest quality responses. This optimization can occur offline during system development or online as part of continuous improvement.
Multi-model ensembling leverages multiple LLMs to improve robustness and quality. The system might use a smaller, faster model for initial parsing and a larger, more capable model for complex synthesis tasks. Ensemble approaches can also combine outputs from multiple models to produce more reliable results.
Streaming responses improve perceived responsiveness by delivering bot responses incrementally rather than waiting for complete generation. This is particularly valuable for long responses where users benefit from seeing partial results immediately.
Conversation branching allows users to explore alternative conversation paths without losing context. The system maintains multiple conversation branches, enabling users to backtrack and try different approaches while preserving the history of each branch.
Personalization adapts bot behavior based on user preferences and interaction patterns. The system learns from user feedback, adjusting generation parameters, prompt templates, and response styles to better match individual user needs.
Safety and content filtering protect users and the system from harmful content. The system implements pre-generation filtering to detect problematic user inputs and post-generation filtering to catch inappropriate bot responses before delivery.
PRODUCTION DEPLOYMENT CONSIDERATIONS
Deploying a bot generator system in production environments requires careful attention to scalability, reliability, and operational concerns.
Scalability considerations include horizontal scaling to handle multiple concurrent users, efficient resource allocation to maximize hardware utilization, and load balancing to distribute work across available compute resources. The system should support both vertical scaling through more powerful hardware and horizontal scaling through distributed deployment.
Reliability mechanisms include health monitoring to detect system failures, automatic recovery procedures to restore service after errors, and graceful degradation to maintain partial functionality when components fail. The system should implement comprehensive logging and alerting to enable rapid diagnosis and resolution of issues.
Security measures protect user data, prevent unauthorized access, and ensure safe operation. This includes authentication and authorization for API access, encryption for data in transit and at rest, input validation to prevent injection attacks, and rate limiting to prevent abuse.
Performance optimization focuses on minimizing latency, maximizing throughput, and efficient resource utilization. Techniques include model quantization to reduce memory requirements, batching to process multiple requests together, and caching to avoid redundant computation.
Monitoring and observability provide visibility into system behavior and performance. The system should track metrics such as request latency, generation quality, error rates, and resource utilization. Distributed tracing helps understand request flow through complex multi-component systems.
Cost management balances performance requirements against infrastructure expenses. For cloud deployments, this includes selecting appropriate instance types, implementing auto-scaling to match capacity to demand, and optimizing model selection to balance quality and cost.
FULL PRODUCTION-READY IMPLEMENTATION
The following complete implementation demonstrates all concepts discussed in this tutorial. This production-ready code supports multiple GPU architectures, handles errors gracefully, implements proper resource management, and provides a clean API for bot generation and interaction.
import json
import re
import uuid
import logging
from typing import Dict, List, Optional, Any, Tuple
from abc import ABC, abstractmethod
from dataclasses import dataclass, asdict
from enum import Enum
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class BackendType(Enum):
"""Enumeration of supported backend types."""
CUDA = "cuda"
ROCM = "rocm"
MPS = "mps"
CPU = "cpu"
REMOTE = "remote"
@dataclass
class BotRequirements:
"""Structured representation of bot requirements."""
primary_purpose: str
domain_knowledge: List[str]
interaction_style: str
constraints: List[str]
special_capabilities: List[str]
expected_inputs: List[str]
expected_outputs: List[str]
success_criteria: List[str]
implicit_requirements: Dict[str, List[str]]
validation_issues: List[Dict[str, str]]
def to_dict(self) -> Dict:
"""Convert requirements to dictionary."""
return asdict(self)
@dataclass
class BotConfiguration:
"""Configuration for a generated bot."""
bot_type: str
system_prompt: Optional[str]
conversation_template: Optional[str]
initial_greeting: str
application_code: Optional[str]
metadata: Dict[str, Any]
def to_dict(self) -> Dict:
"""Convert configuration to dictionary."""
return asdict(self)
class LLMBackend(ABC):
"""Abstract base class for LLM backend implementations."""
@abstractmethod
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize the backend with specified model and configuration."""
pass
@abstractmethod
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate a response from the language model."""
pass
@abstractmethod
def cleanup(self) -> None:
"""Release resources and perform cleanup operations."""
pass
class CUDABackend(LLMBackend):
"""NVIDIA CUDA-accelerated local LLM backend."""
def __init__(self):
self.model = None
self.tokenizer = None
self.device = None
self.model_name = None
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize CUDA backend with specified model."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
if not torch.cuda.is_available():
raise RuntimeError("CUDA not available on this system")
self.device = torch.device("cuda")
device_props = torch.cuda.get_device_properties(0)
logger.info(f"Initializing CUDA backend on {device_props.name} "
f"with {device_props.total_memory / 1e9:.2f} GB memory")
self.model_name = model_name
load_in_8bit = config.get("quantize_8bit", False)
load_in_4bit = config.get("quantize_4bit", False)
logger.info(f"Loading model {model_name}...")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
# Set padding token if not present
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
load_in_8bit=load_in_8bit,
load_in_4bit=load_in_4bit,
torch_dtype=torch.float16 if not (load_in_8bit or load_in_4bit) else None,
trust_remote_code=config.get("trust_remote_code", False)
)
self.model.eval()
logger.info("Model loaded successfully")
except Exception as e:
logger.error(f"Failed to initialize CUDA backend: {str(e)}")
raise
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate response using CUDA-accelerated model."""
try:
import torch
# Construct full prompt with system message
full_prompt = prompt
if system_prompt:
full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt}\n<</SYS>>\n\n{prompt} [/INST]"
# Tokenize input
inputs = self.tokenizer(full_prompt, return_tensors="pt",
padding=True, truncation=True,
max_length=4096).to(self.device)
# Configure generation parameters
gen_kwargs = {
"max_new_tokens": max_tokens,
"temperature": temperature,
"do_sample": temperature > 0,
"pad_token_id": self.tokenizer.eos_token_id,
"eos_token_id": self.tokenizer.eos_token_id
}
# Add stopping criteria if specified
if stop_sequences:
from transformers import StoppingCriteriaList, StoppingCriteria
class StopOnTokens(StoppingCriteria):
def __init__(self, stop_token_ids):
self.stop_token_ids = stop_token_ids
def __call__(self, input_ids, scores, **kwargs):
for stop_id in self.stop_token_ids:
if input_ids[0][-1] == stop_id:
return True
return False
stop_token_ids = []
for seq in stop_sequences:
tokens = self.tokenizer.encode(seq, add_special_tokens=False)
if tokens:
stop_token_ids.append(tokens[0])
if stop_token_ids:
gen_kwargs["stopping_criteria"] = StoppingCriteriaList(
[StopOnTokens(stop_token_ids)]
)
# Generate response
with torch.no_grad():
outputs = self.model.generate(**inputs, **gen_kwargs)
# Decode only the generated portion
generated_text = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return generated_text.strip()
except Exception as e:
logger.error(f"Generation failed: {str(e)}")
raise
def cleanup(self) -> None:
"""Clean up CUDA resources."""
try:
import torch
if self.model is not None:
del self.model
self.model = None
if self.tokenizer is not None:
del self.tokenizer
self.tokenizer = None
if torch.cuda.is_available():
torch.cuda.empty_cache()
logger.info("CUDA backend cleaned up successfully")
except Exception as e:
logger.error(f"Cleanup failed: {str(e)}")
class ROCmBackend(LLMBackend):
"""AMD ROCm-accelerated local LLM backend."""
def __init__(self):
self.model = None
self.tokenizer = None
self.device = None
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize ROCm backend with specified model."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
# Check for ROCm availability
if not torch.cuda.is_available():
raise RuntimeError("ROCm/CUDA not available on this system")
self.device = torch.device("cuda") # ROCm uses CUDA API
logger.info("Initializing ROCm backend")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
torch_dtype=torch.float16,
trust_remote_code=config.get("trust_remote_code", False)
)
self.model.eval()
logger.info("ROCm model loaded successfully")
except Exception as e:
logger.error(f"Failed to initialize ROCm backend: {str(e)}")
raise
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate response using ROCm-accelerated model."""
# Implementation similar to CUDA backend
try:
import torch
full_prompt = prompt
if system_prompt:
full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt}\n<</SYS>>\n\n{prompt} [/INST]"
inputs = self.tokenizer(full_prompt, return_tensors="pt",
padding=True, truncation=True,
max_length=4096).to(self.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.eos_token_id
)
generated_text = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return generated_text.strip()
except Exception as e:
logger.error(f"ROCm generation failed: {str(e)}")
raise
def cleanup(self) -> None:
"""Clean up ROCm resources."""
try:
import torch
if self.model is not None:
del self.model
if self.tokenizer is not None:
del self.tokenizer
torch.cuda.empty_cache()
logger.info("ROCm backend cleaned up")
except Exception as e:
logger.error(f"ROCm cleanup failed: {str(e)}")
class MPSBackend(LLMBackend):
"""Apple Metal Performance Shaders backend for Apple Silicon."""
def __init__(self):
self.model = None
self.tokenizer = None
self.device = None
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize MPS backend with specified model."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
if not torch.backends.mps.is_available():
raise RuntimeError("MPS not available on this system")
self.device = torch.device("mps")
logger.info("Initializing MPS backend for Apple Silicon")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
trust_remote_code=config.get("trust_remote_code", False)
)
self.model.to(self.device)
self.model.eval()
logger.info("MPS model loaded successfully")
except Exception as e:
logger.error(f"Failed to initialize MPS backend: {str(e)}")
raise
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate response using MPS-accelerated model."""
try:
import torch
full_prompt = prompt
if system_prompt:
full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt}\n<</SYS>>\n\n{prompt} [/INST]"
inputs = self.tokenizer(full_prompt, return_tensors="pt",
padding=True, truncation=True,
max_length=4096).to(self.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.eos_token_id
)
generated_text = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return generated_text.strip()
except Exception as e:
logger.error(f"MPS generation failed: {str(e)}")
raise
def cleanup(self) -> None:
"""Clean up MPS resources."""
try:
if self.model is not None:
del self.model
if self.tokenizer is not None:
del self.tokenizer
logger.info("MPS backend cleaned up")
except Exception as e:
logger.error(f"MPS cleanup failed: {str(e)}")
class CPUBackend(LLMBackend):
"""CPU-only backend for systems without GPU acceleration."""
def __init__(self):
self.model = None
self.tokenizer = None
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize CPU backend with specified model."""
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
logger.info("Initializing CPU backend (this may be slow)")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
trust_remote_code=config.get("trust_remote_code", False)
)
self.model.eval()
logger.info("CPU model loaded successfully")
except Exception as e:
logger.error(f"Failed to initialize CPU backend: {str(e)}")
raise
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate response using CPU."""
try:
import torch
full_prompt = prompt
if system_prompt:
full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt}\n<</SYS>>\n\n{prompt} [/INST]"
inputs = self.tokenizer(full_prompt, return_tensors="pt",
padding=True, truncation=True,
max_length=4096)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.eos_token_id
)
generated_text = self.tokenizer.decode(
outputs[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return generated_text.strip()
except Exception as e:
logger.error(f"CPU generation failed: {str(e)}")
raise
def cleanup(self) -> None:
"""Clean up CPU resources."""
try:
if self.model is not None:
del self.model
if self.tokenizer is not None:
del self.tokenizer
logger.info("CPU backend cleaned up")
except Exception as e:
logger.error(f"CPU cleanup failed: {str(e)}")
class RemoteBackend(LLMBackend):
"""Backend for remote API-based LLM services."""
def __init__(self):
self.api_client = None
self.api_type = None
self.model_name = None
def initialize(self, model_name: str, config: Dict[str, Any]) -> None:
"""Initialize remote API backend."""
try:
self.api_type = config.get("api_type", "openai")
self.model_name = model_name
if self.api_type == "openai":
import openai
api_key = config.get("api_key")
if not api_key:
raise ValueError("API key required for OpenAI backend")
openai.api_key = api_key
self.api_client = openai
logger.info(f"Initialized OpenAI backend with model {model_name}")
elif self.api_type == "anthropic":
import anthropic
api_key = config.get("api_key")
if not api_key:
raise ValueError("API key required for Anthropic backend")
self.api_client = anthropic.Anthropic(api_key=api_key)
logger.info(f"Initialized Anthropic backend with model {model_name}")
else:
raise ValueError(f"Unsupported API type: {self.api_type}")
except Exception as e:
logger.error(f"Failed to initialize remote backend: {str(e)}")
raise
def generate(self, prompt: str, system_prompt: Optional[str] = None,
max_tokens: int = 2048, temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None) -> str:
"""Generate response using remote API."""
try:
if self.api_type == "openai":
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = self.api_client.ChatCompletion.create(
model=self.model_name,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
stop=stop_sequences
)
return response.choices[0].message.content.strip()
elif self.api_type == "anthropic":
full_prompt = prompt
if system_prompt:
full_prompt = f"{system_prompt}\n\n{prompt}"
response = self.api_client.messages.create(
model=self.model_name,
max_tokens=max_tokens,
temperature=temperature,
messages=[{"role": "user", "content": full_prompt}]
)
return response.content[0].text.strip()
except Exception as e:
logger.error(f"Remote generation failed: {str(e)}")
raise
def cleanup(self) -> None:
"""Clean up remote backend resources."""
self.api_client = None
logger.info("Remote backend cleaned up")
class SpecificationParser:
"""Parses natural language bot specifications into structured requirements."""
def __init__(self, llm_backend: LLMBackend):
self.backend = llm_backend
def parse(self, user_specification: str) -> BotRequirements:
"""Parse user specification into structured requirements."""
try:
logger.info("Parsing bot specification...")
# Extract explicit requirements
extraction_prompt = f"""Analyze the following bot specification and extract structured requirements.
User Specification: {user_specification}
Extract and provide the following information in JSON format:
{{
"primary_purpose": "Main function of the bot",
"domain_knowledge": ["List of knowledge domains the bot needs"],
"interaction_style": "How the bot should communicate (formal, casual, technical, etc.)",
"constraints": ["Things the bot should NOT do or boundaries to respect"],
"special_capabilities": ["Specific abilities needed like code generation, data analysis, etc."],
"expected_inputs": ["Types of inputs users will provide"],
"expected_outputs": ["Types of responses the bot should generate"],
"success_criteria": ["How to measure if the bot is performing correctly"]
}}
Provide detailed, specific information for each field. Respond ONLY with valid JSON, no additional text."""
extraction_response = self.backend.generate(
extraction_prompt,
system_prompt="You are an expert at analyzing requirements and extracting structured information. Always respond with valid JSON only.",
temperature=0.3,
max_tokens=2000
)
# Parse JSON response
requirements_dict = self._extract_json(extraction_response)
# Infer implicit requirements
logger.info("Inferring implicit requirements...")
implicit_requirements = self._infer_implicit_requirements(
user_specification, requirements_dict
)
# Validate requirements
logger.info("Validating requirements...")
validation_issues = self._validate_requirements(requirements_dict)
# Construct BotRequirements object
requirements = BotRequirements(
primary_purpose=requirements_dict.get("primary_purpose", ""),
domain_knowledge=requirements_dict.get("domain_knowledge", []),
interaction_style=requirements_dict.get("interaction_style", ""),
constraints=requirements_dict.get("constraints", []),
special_capabilities=requirements_dict.get("special_capabilities", []),
expected_inputs=requirements_dict.get("expected_inputs", []),
expected_outputs=requirements_dict.get("expected_outputs", []),
success_criteria=requirements_dict.get("success_criteria", []),
implicit_requirements=implicit_requirements,
validation_issues=validation_issues
)
logger.info("Specification parsing complete")
return requirements
except Exception as e:
logger.error(f"Failed to parse specification: {str(e)}")
raise
def _extract_json(self, text: str) -> Dict:
"""Extract JSON object from text that may contain additional content."""
# Try to find JSON object in the text
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON: {str(e)}")
raise
else:
raise ValueError("No JSON object found in response")
def _infer_implicit_requirements(self, original_spec: str,
explicit_requirements: Dict) -> Dict[str, List[str]]:
"""Infer requirements not explicitly stated by the user."""
try:
inference_prompt = f"""Given this bot specification and extracted requirements, identify important implicit requirements that weren't explicitly stated but are necessary for the bot to function well.
Original Specification: {original_spec}
Explicit Requirements: {json.dumps(explicit_requirements, indent=2)}
Identify implicit requirements in these categories:
- context_management: How should the bot handle conversation history?
- error_handling: How should the bot respond to unclear or invalid inputs?
- clarification: When should the bot ask for more information?
- scope_boundaries: What related tasks are outside the bot's scope?
- user_experience: What makes interactions smooth and helpful?
Provide a JSON object with these categories as keys and lists of implicit requirements as values. Respond ONLY with valid JSON."""
inference_response = self.backend.generate(
inference_prompt,
system_prompt="You are an expert at identifying unstated but important requirements. Respond with valid JSON only.",
temperature=0.4,
max_tokens=1500
)
return self._extract_json(inference_response)
except Exception as e:
logger.warning(f"Failed to infer implicit requirements: {str(e)}")
return {}
def _validate_requirements(self, requirements: Dict) -> List[Dict[str, str]]:
"""Validate that requirements are complete and consistent."""
try:
validation_prompt = f"""Review these bot requirements for completeness and consistency:
{json.dumps(requirements, indent=2)}
Identify any issues:
- Missing critical information needed to build the bot
- Contradictions between different requirements
- Ambiguities that need clarification
- Unrealistic or impossible requirements
Provide a JSON array of issue objects, each with 'type' (missing/contradiction/ambiguity/unrealistic) and 'description' fields. If no issues, return an empty array []. Respond ONLY with valid JSON."""
validation_response = self.backend.generate(
validation_prompt,
system_prompt="You are an expert at requirements validation. Respond with valid JSON only.",
temperature=0.2,
max_tokens=1500
)
# Extract JSON array
json_match = re.search(r'\[.*\]', validation_response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
return []
except Exception as e:
logger.warning(f"Failed to validate requirements: {str(e)}")
return []
class PromptSynthesizer:
"""Generates optimized prompts from structured requirements."""
def __init__(self, llm_backend: LLMBackend):
self.backend = llm_backend
def synthesize(self, requirements: BotRequirements) -> BotConfiguration:
"""Generate bot configuration from requirements."""
try:
logger.info("Synthesizing bot configuration...")
# Assess complexity
complexity_score = self._assess_complexity(requirements)
logger.info(f"Complexity score: {complexity_score}/10")
# Select synthesis strategy based on complexity
if complexity_score < 3:
config = self._synthesize_simple_prompt(requirements)
elif complexity_score < 7:
config = self._synthesize_structured_prompt(requirements)
else:
config = self._synthesize_application_code(requirements)
logger.info(f"Generated {config.bot_type} bot configuration")
return config
except Exception as e:
logger.error(f"Failed to synthesize bot configuration: {str(e)}")
raise
def _assess_complexity(self, requirements: BotRequirements) -> int:
"""Assess requirement complexity on a scale of 0-10."""
score = 0
# Multiple domains add complexity
score += min(len(requirements.domain_knowledge), 3)
# Special capabilities add significant complexity
score += min(len(requirements.special_capabilities) * 2, 4)
# Constraints add some complexity
score += min(len(requirements.constraints), 2)
# Validation issues indicate complexity
score += min(len(requirements.validation_issues), 1)
return score
def _synthesize_simple_prompt(self, requirements: BotRequirements) -> BotConfiguration:
"""Generate a straightforward system prompt for simple bots."""
try:
synthesis_prompt = f"""Create a concise, effective system prompt for an LLM bot with these requirements:
Primary Purpose: {requirements.primary_purpose}
Domain Knowledge: {', '.join(requirements.domain_knowledge)}
Interaction Style: {requirements.interaction_style}
Constraints: {', '.join(requirements.constraints)}
The system prompt should:
- Clearly establish the bot's identity and purpose
- Specify relevant knowledge domains
- Define interaction style and tone
- State important constraints and boundaries
- Be concise but comprehensive (aim for 150-300 words)
Generate only the system prompt text, without additional commentary or formatting."""
system_prompt = self.backend.generate(
synthesis_prompt,
system_prompt="You are an expert prompt engineer who creates highly effective system prompts for LLM applications. Generate clear, concise prompts without markdown formatting.",
temperature=0.5,
max_tokens=1500
)
# Generate initial greeting
greeting_prompt = f"""Based on this system prompt, create a brief, friendly initial greeting that the bot should use when first interacting with users:
System Prompt: {system_prompt}
The greeting should:
- Welcome the user warmly
- Briefly explain what the bot can help with
- Invite the user to describe their needs
- Be professional but approachable
- Be 2-3 sentences maximum
Generate only the greeting text, without additional commentary."""
initial_greeting = self.backend.generate(
greeting_prompt,
system_prompt="You are an expert at crafting welcoming, effective bot greetings.",
temperature=0.6,
max_tokens=500
)
return BotConfiguration(
bot_type="simple_prompt",
system_prompt=system_prompt.strip(),
conversation_template=None,
initial_greeting=initial_greeting.strip(),
application_code=None,
metadata={
"complexity": "simple",
"requirements": requirements.to_dict()
}
)
except Exception as e:
logger.error(f"Failed to synthesize simple prompt: {str(e)}")
raise
def _synthesize_structured_prompt(self, requirements: BotRequirements) -> BotConfiguration:
"""Generate a detailed structured prompt with templates."""
try:
synthesis_prompt = f"""Create a comprehensive system prompt for an LLM bot with these detailed requirements:
Primary Purpose: {requirements.primary_purpose}
Domain Knowledge: {json.dumps(requirements.domain_knowledge, indent=2)}
Interaction Style: {requirements.interaction_style}
Constraints: {json.dumps(requirements.constraints, indent=2)}
Special Capabilities: {json.dumps(requirements.special_capabilities, indent=2)}
Expected Inputs: {json.dumps(requirements.expected_inputs, indent=2)}
Expected Outputs: {json.dumps(requirements.expected_outputs, indent=2)}
The system prompt should include:
1. Identity and Purpose Statement
2. Detailed Knowledge Domain Specifications
3. Interaction Style Guidelines with examples
4. Clear Constraint Boundaries
5. Instructions for exercising special capabilities
6. Input handling procedures
7. Output formatting requirements
8. Few-shot examples demonstrating ideal interactions
Create a thorough, well-structured system prompt (aim for 400-800 words). Do not use markdown formatting."""
system_prompt = self.backend.generate(
synthesis_prompt,
system_prompt="You are an expert prompt engineer specializing in complex, multi-faceted LLM applications. Generate detailed prompts without markdown formatting.",
temperature=0.4,
max_tokens=3000
)
# Generate conversation template
template_prompt = f"""Based on this system prompt, create a conversation flow template that guides how the bot should structure interactions:
System Prompt: {system_prompt}
The template should define:
1. Initial greeting and capability explanation
2. Information gathering questions (if needed)
3. Processing and response generation approach
4. Follow-up and clarification patterns
5. Conversation conclusion strategies
Format as a structured guide the bot can follow. Do not use markdown formatting."""
conversation_template = self.backend.generate(
template_prompt,
system_prompt="You are an expert at designing effective conversation flows for AI assistants.",
temperature=0.4,
max_tokens=2000
)
return BotConfiguration(
bot_type="structured_prompt",
system_prompt=system_prompt.strip(),
conversation_template=conversation_template.strip(),
initial_greeting="Hello! I'm ready to assist you. How can I help you today?",
application_code=None,
metadata={
"complexity": "structured",
"requirements": requirements.to_dict()
}
)
except Exception as e:
logger.error(f"Failed to synthesize structured prompt: {str(e)}")
raise
def _synthesize_application_code(self, requirements: BotRequirements) -> BotConfiguration:
"""Generate complete application code for complex bots."""
try:
code_generation_prompt = f"""Generate a complete Python application that implements a bot with these requirements:
{json.dumps(requirements.to_dict(), indent=2)}
The application should include:
1. A main Bot class that encapsulates all functionality
2. Proper initialization with configuration
3. Conversation state management
4. Input validation and processing
5. Response generation with the specified capabilities
6. Error handling and graceful degradation
7. Logging and debugging support
8. Clean separation of concerns
Generate production-ready, well-documented Python code with proper error handling.
Include all necessary imports and class definitions.
Use type hints and docstrings throughout.
Do not use markdown code blocks - generate plain Python code only."""
application_code = self.backend.generate(
code_generation_prompt,
system_prompt="You are an expert Python developer who writes clean, production-ready code following best practices. Generate plain Python code without markdown formatting.",
temperature=0.3,
max_tokens=4000
)
return BotConfiguration(
bot_type="application_code",
system_prompt=None,
conversation_template=None,
initial_greeting="Bot initialized. Ready to assist.",
application_code=application_code.strip(),
metadata={
"complexity": "application",
"requirements": requirements.to_dict(),
"language": "python"
}
)
except Exception as e:
logger.error(f"Failed to synthesize application code: {str(e)}")
raise
class GeneratedBot:
"""Runtime instance of a generated bot."""
def __init__(self, config: BotConfiguration, llm_backend: LLMBackend):
"""Initialize a bot instance."""
self.config = config
self.backend = llm_backend
self.conversation_history = []
self.state = {}
self.max_history_length = 20
self.custom_bot = None
# Initialize based on bot type
if config.bot_type == "application_code":
self._initialize_from_code()
def _initialize_from_code(self) -> None:
"""Initialize bot from generated application code."""
try:
# Execute generated code in isolated namespace
namespace = {"llm_backend": self.backend}
exec(self.config.application_code, namespace)
# Find and instantiate the Bot class
bot_class = None
for name, obj in namespace.items():
if isinstance(obj, type) and "Bot" in name:
bot_class = obj
break
if bot_class:
self.custom_bot = bot_class(self.backend)
logger.info("Custom bot initialized from generated code")
else:
logger.warning("No Bot class found in generated code, falling back to prompt-based bot")
except Exception as e:
logger.error(f"Failed to initialize from code: {str(e)}")
logger.info("Falling back to prompt-based bot")
def start(self) -> str:
"""Start the bot and return initial greeting."""
initial_message = self.config.initial_greeting
self.conversation_history.append({
"role": "assistant",
"content": initial_message
})
return initial_message
def process_message(self, user_message: str) -> str:
"""Process a user message and generate a response."""
try:
# Add user message to history
self.conversation_history.append({
"role": "user",
"content": user_message
})
# Truncate history if too long
if len(self.conversation_history) > self.max_history_length:
self.conversation_history = self.conversation_history[-self.max_history_length:]
# Generate response based on bot type
if self.config.bot_type == "application_code" and self.custom_bot:
response = self._process_with_custom_code(user_message)
else:
response = self._process_with_prompt(user_message)
# Add response to history
self.conversation_history.append({
"role": "assistant",
"content": response
})
return response
except Exception as e:
logger.error(f"Error processing message: {str(e)}")
error_response = "I apologize, but I encountered an error processing your message. Could you please try rephrasing your request?"
self.conversation_history.append({
"role": "assistant",
"content": error_response
})
return error_response
def _process_with_custom_code(self, user_message: str) -> str:
"""Process message using custom generated code."""
if hasattr(self.custom_bot, "process_message"):
return self.custom_bot.process_message(user_message)
else:
logger.warning("Custom bot missing process_message method")
return self._process_with_prompt(user_message)
def _process_with_prompt(self, user_message: str) -> str:
"""Process message using synthesized prompts."""
# Format conversation history
conversation_context = self._format_conversation_history()
# Generate response
response = self.backend.generate(
prompt=conversation_context,
system_prompt=self.config.system_prompt,
temperature=0.7,
max_tokens=2048
)
return response.strip()
def _format_conversation_history(self) -> str:
"""Format conversation history for model input."""
formatted = []
for message in self.conversation_history:
role = message["role"]
content = message["content"]
if role == "user":
formatted.append(f"User: {content}")
elif role == "assistant":
formatted.append(f"Assistant: {content}")
formatted.append("Assistant:")
return "\n\n".join(formatted)
def get_state(self) -> Dict:
"""Get current bot state for persistence or inspection."""
return {
"config": self.config.to_dict(),
"conversation_history": self.conversation_history,
"state": self.state
}
def cleanup(self) -> None:
"""Clean up bot resources."""
self.conversation_history.clear()
self.state.clear()
if self.custom_bot and hasattr(self.custom_bot, "cleanup"):
self.custom_bot.cleanup()
class BotGenerator:
"""Main orchestrator for the bot generation system."""
def __init__(self, backend_type: str = "cuda",
model_name: str = "mistralai/Mistral-7B-Instruct-v0.2",
backend_config: Optional[Dict[str, Any]] = None):
"""Initialize the bot generator."""
logger.info("Initializing BotGenerator...")
self.backend = self._initialize_backend(backend_type, model_name,
backend_config or {})
self.parser = SpecificationParser(self.backend)
self.synthesizer = PromptSynthesizer(self.backend)
self.active_bots = {}
logger.info("BotGenerator initialized successfully")
def _initialize_backend(self, backend_type: str, model_name: str,
config: Dict[str, Any]) -> LLMBackend:
"""Initialize the appropriate LLM backend."""
backend_map = {
"cuda": CUDABackend,
"rocm": ROCmBackend,
"mps": MPSBackend,
"cpu": CPUBackend,
"remote": RemoteBackend
}
if backend_type not in backend_map:
raise ValueError(f"Unknown backend type: {backend_type}")
backend = backend_map[backend_type]()
backend.initialize(model_name, config)
return backend
def generate_bot(self, user_specification: str,
bot_id: Optional[str] = None) -> Tuple[str, GeneratedBot]:
"""Generate a bot from user specification."""
if bot_id is None:
bot_id = str(uuid.uuid4())
try:
logger.info(f"Generating bot {bot_id}...")
# Parse specification
requirements = self.parser.parse(user_specification)
# Log validation issues if any
if requirements.validation_issues:
logger.warning(f"Validation issues: {requirements.validation_issues}")
# Synthesize bot configuration
bot_config = self.synthesizer.synthesize(requirements)
# Instantiate bot
bot_instance = GeneratedBot(bot_config, self.backend)
# Store active bot
self.active_bots[bot_id] = bot_instance
logger.info(f"Bot {bot_id} generated successfully")
return bot_id, bot_instance
except Exception as e:
logger.error(f"Failed to generate bot: {str(e)}")
raise
def start_bot(self, bot_id: str) -> str:
"""Start a bot and get its initial greeting."""
if bot_id not in self.active_bots:
raise ValueError(f"No active bot with ID {bot_id}")
bot = self.active_bots[bot_id]
return bot.start()
def interact_with_bot(self, bot_id: str, user_message: str) -> str:
"""Send a message to a generated bot."""
if bot_id not in self.active_bots:
raise ValueError(f"No active bot with ID {bot_id}")
bot = self.active_bots[bot_id]
return bot.process_message(user_message)
def get_bot_state(self, bot_id: str) -> Dict:
"""Get the current state of a bot."""
if bot_id not in self.active_bots:
raise ValueError(f"No active bot with ID {bot_id}")
bot = self.active_bots[bot_id]
return bot.get_state()
def remove_bot(self, bot_id: str) -> None:
"""Remove a bot and clean up its resources."""
if bot_id in self.active_bots:
bot = self.active_bots[bot_id]
bot.cleanup()
del self.active_bots[bot_id]
logger.info(f"Bot {bot_id} removed")
def list_active_bots(self) -> List[str]:
"""Get list of active bot IDs."""
return list(self.active_bots.keys())
def shutdown(self) -> None:
"""Shutdown the bot generator and clean up all resources."""
logger.info("Shutting down BotGenerator...")
# Clean up all active bots
for bot_id in list(self.active_bots.keys()):
self.remove_bot(bot_id)
# Clean up backend
self.backend.cleanup()
logger.info("BotGenerator shutdown complete")
def main():
"""Example usage of the bot generator system."""
# Example 1: Generate a Python debugging assistant
print("=" * 80)
print("EXAMPLE 1: Python Debugging Assistant")
print("=" * 80)
try:
# Initialize bot generator with CPU backend for demonstration
# In production, use 'cuda', 'rocm', 'mps', or 'remote' as appropriate
generator = BotGenerator(
backend_type="cpu",
model_name="gpt2", # Small model for demonstration
backend_config={}
)
# Define bot specification
spec1 = """I need a bot that helps software developers debug Python code.
The bot should analyze error messages, suggest fixes, explain common pitfalls,
and provide code examples. It should be technical but friendly, and should ask
clarifying questions when error messages are ambiguous."""
# Generate the bot
bot_id1, bot1 = generator.generate_bot(spec1, "python-debugger")
# Start the bot
greeting = generator.start_bot(bot_id1)
print(f"\nBot: {greeting}\n")
# Interact with the bot
user_msg1 = "I'm getting a KeyError when trying to access a dictionary key."
print(f"User: {user_msg1}")
response1 = generator.interact_with_bot(bot_id1, user_msg1)
print(f"Bot: {response1}\n")
# Example 2: Generate a creative writing assistant
print("=" * 80)
print("EXAMPLE 2: Creative Writing Assistant")
print("=" * 80)
spec2 = """Create a bot that helps writers brainstorm story ideas, develop
characters, and overcome writer's block. The bot should be encouraging and
creative, offering multiple suggestions and asking thought-provoking questions."""
bot_id2, bot2 = generator.generate_bot(spec2, "writing-assistant")
greeting2 = generator.start_bot(bot_id2)
print(f"\nBot: {greeting2}\n")
user_msg2 = "I'm stuck on developing my protagonist's backstory."
print(f"User: {user_msg2}")
response2 = generator.interact_with_bot(bot_id2, user_msg2)
print(f"Bot: {response2}\n")
# List active bots
print("=" * 80)
print("Active Bots:")
for bot_id in generator.list_active_bots():
state = generator.get_bot_state(bot_id)
print(f" - {bot_id}: {state['config']['bot_type']}")
# Cleanup
print("\nCleaning up...")
generator.shutdown()
print("Done!")
except Exception as e:
logger.error(f"Example failed: {str(e)}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
This complete implementation provides a production-ready system for generating LLM bots from natural language specifications. The code supports multiple hardware backends, implements robust error handling, follows clean architecture principles, and demonstrates the entire workflow from specification to functional bot deployment.
No comments:
Post a Comment