Modern Software Development with AI: Crafting Intelligent Applications

Software development has been profoundly impacted by AI, with new tools, practices, and architectural patterns emerging for building intelligent applications that can reason, learn, and adapt.

From Python's dominance in the AI ecosystem to sophisticated LLM-powered applications and autonomous agents, the landscape of software development is rapidly evolving. Let's explore how to build the next generation of intelligent applications.

1. Python: The Language of Machine Learning

Python remains the most popular language for machine learning and AI development due to its readability, simplicity, and extensive ecosystem of libraries that accelerate development.

Python Fundamentals for AI Development

A solid understanding of Python fundamentals is essential before diving into AI frameworks:

# Example: Python fundamentals for AI applications
import asyncio
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class ModelType(Enum):
    """Enumeration for different model types"""
    CLASSIFICATION = "classification"
    REGRESSION = "regression"
    GENERATION = "generation"

@dataclass
class AIModelConfig:
    """Configuration class for AI models"""
    name: str
    model_type: ModelType
    max_tokens: int = 1000
    temperature: float = 0.7
    api_key: Optional[str] = None
    
    def __post_init__(self):
        if self.temperature < 0 or self.temperature > 2:
            raise ValueError("Temperature must be between 0 and 2")

class AIModelManager:
    """Manages multiple AI models with proper error handling"""
    
    def __init__(self):
        self.models: Dict[str, AIModelConfig] = {}
        self.logger = logging.getLogger(__name__)
    
    def add_model(self, config: AIModelConfig) -> None:
        """Add a new model configuration"""
        try:
            self.models[config.name] = config
            self.logger.info(f"Added model: {config.name}")
        except Exception as e:
            self.logger.error(f"Failed to add model {config.name}: {e}")
            raise
    
    async def process_batch(self, texts: List[str], model_name: str) -> List[str]:
        """Process multiple texts asynchronously"""
        if model_name not in self.models:
            raise KeyError(f"Model {model_name} not found")
        
        # Simulate async processing
        tasks = [self._process_single(text, model_name) for text in texts]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Handle any exceptions in results
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                self.logger.error(f"Failed to process text {i}: {result}")
                processed_results.append(f"Error: {str(result)}")
            else:
                processed_results.append(result)
        
        return processed_results
    
    async def _process_single(self, text: str, model_name: str) -> str:
        """Process a single text (simulated)"""
        await asyncio.sleep(0.1)  # Simulate API call
        return f"Processed by {model_name}: {text[:50]}..."

# Example usage
async def main():
    manager = AIModelManager()
    
    # Add model configurations
    gpt_config = AIModelConfig(
        name="gpt-4",
        model_type=ModelType.GENERATION,
        temperature=0.8
    )
    manager.add_model(gpt_config)
    
    # Process texts
    texts = ["Hello world", "AI is transforming software", "Python is powerful"]
    results = await manager.process_batch(texts, "gpt-4")
    print("Results:", results)

# Run the example
# asyncio.run(main())

Essential Data Science Libraries

Mastering key Python libraries is crucial for AI development:

# Example: Data science pipeline with Python libraries
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.preprocessing import StandardScaler

class DataSciencePipeline:
    """Complete data science pipeline for AI applications"""
    
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.feature_names = None
    
    def load_and_explore_data(self, data_path: str = None) -> pd.DataFrame:
        """Load and explore dataset"""
        # Generate sample data for demonstration
        np.random.seed(42)
        n_samples = 1000
        
        data = {
            'feature_1': np.random.normal(0, 1, n_samples),
            'feature_2': np.random.normal(2, 1.5, n_samples),
            'feature_3': np.random.exponential(1, n_samples),
            'feature_4': np.random.uniform(-1, 1, n_samples)
        }
        
        # Create target based on features
        df = pd.DataFrame(data)
        df['target'] = (
            (df['feature_1'] > 0) & 
            (df['feature_2'] > 2) & 
            (df['feature_3'] < 1.5)
        ).astype(int)
        
        # Data exploration
        print("Dataset Shape:", df.shape)
        print("\nDataset Info:")
        print(df.info())
        print("\nStatistical Summary:")
        print(df.describe())
        
        # Visualize data
        self._plot_data_exploration(df)
        
        return df
    
    def _plot_data_exploration(self, df: pd.DataFrame):
        """Create exploratory data visualizations"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        
        # Distribution plots
        for i, col in enumerate(['feature_1', 'feature_2', 'feature_3', 'feature_4']):
            row, col_idx = i // 2, i % 2
            sns.histplot(data=df, x=col, hue='target', ax=axes[row, col_idx])
            axes[row, col_idx].set_title(f'Distribution of {col}')
        
        plt.tight_layout()
        plt.show()
        
        # Correlation heatmap
        plt.figure(figsize=(10, 8))
        correlation_matrix = df.corr()
        sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', center=0)
        plt.title('Feature Correlation Matrix')
        plt.show()
    
    def train_model(self, df: pd.DataFrame) -> dict:
        """Train machine learning model"""
        # Prepare features and target
        features = [col for col in df.columns if col != 'target']
        X = df[features]
        y = df['target']
        
        self.feature_names = features
        
        # Split data
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )
        
        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)
        
        # Train model
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.model.fit(X_train_scaled, y_train)
        
        # Evaluate model
        y_pred = self.model.predict(X_test_scaled)
        
        results = {
            'accuracy': self.model.score(X_test_scaled, y_test),
            'classification_report': classification_report(y_test, y_pred),
            'confusion_matrix': confusion_matrix(y_test, y_pred),
            'feature_importance': dict(zip(features, self.model.feature_importances_))
        }
        
        return results
    
    def predict(self, new_data: np.ndarray) -> np.ndarray:
        """Make predictions on new data"""
        if self.model is None:
            raise ValueError("Model not trained yet")
        
        new_data_scaled = self.scaler.transform(new_data)
        return self.model.predict(new_data_scaled)

# Example usage
pipeline = DataSciencePipeline()
data = pipeline.load_and_explore_data()
results = pipeline.train_model(data)

print(f"Model Accuracy: {results['accuracy']:.3f}")
print(f"Feature Importance: {results['feature_importance']}")

Machine Learning with Scikit-learn

Scikit-learn provides a comprehensive suite of algorithms for predictive analytics:

# Example: Advanced ML pipeline with scikit-learn
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
from sklearn.model_selection import GridSearchCV, cross_val_score
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import joblib

class AdvancedMLPipeline:
    """Advanced ML pipeline with preprocessing and model selection"""
    
    def __init__(self):
        self.pipeline = None
        self.best_model = None
        self.models = {
            'rf': RandomForestClassifier(random_state=42),
            'gb': GradientBoostingClassifier(random_state=42),
            'svm': SVC(random_state=42)
        }
    
    def create_preprocessing_pipeline(self, numeric_features: list, categorical_features: list):
        """Create comprehensive preprocessing pipeline"""
        # Numeric preprocessing
        numeric_transformer = Pipeline(steps=[
            ('imputer', SimpleImputer(strategy='median')),
            ('scaler', StandardScaler())
        ])
        
        # Categorical preprocessing
        categorical_transformer = Pipeline(steps=[
            ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
            ('onehot', OneHotEncoder(handle_unknown='ignore'))
        ])
        
        # Combine preprocessing steps
        preprocessor = ColumnTransformer(
            transformers=[
                ('num', numeric_transformer, numeric_features),
                ('cat', categorical_transformer, categorical_features)
            ]
        )
        
        return preprocessor
    
    def train_with_model_selection(self, X, y, numeric_features, categorical_features):
        """Train models with hyperparameter tuning and selection"""
        preprocessor = self.create_preprocessing_pipeline(numeric_features, categorical_features)
        
        best_score = 0
        best_model_name = None
        results = {}
        
        for name, model in self.models.items():
            # Create pipeline
            pipeline = Pipeline(steps=[
                ('preprocessor', preprocessor),
                ('classifier', model)
            ])
            
            # Define hyperparameters for tuning
            if name == 'rf':
                param_grid = {
                    'classifier__n_estimators': [50, 100, 200],
                    'classifier__max_depth': [3, 5, None],
                    'classifier__min_samples_split': [2, 5, 10]
                }
            elif name == 'gb':
                param_grid = {
                    'classifier__n_estimators': [50, 100],
                    'classifier__learning_rate': [0.01, 0.1, 0.2],
                    'classifier__max_depth': [3, 5]
                }
            else:  # SVM
                param_grid = {
                    'classifier__C': [0.1, 1, 10],
                    'classifier__kernel': ['rbf', 'linear'],
                    'classifier__gamma': ['scale', 'auto']
                }
            
            # Grid search with cross-validation
            grid_search = GridSearchCV(
                pipeline, param_grid, cv=5, 
                scoring='accuracy', n_jobs=-1
            )
            grid_search.fit(X, y)
            
            # Store results
            results[name] = {
                'best_score': grid_search.best_score_,
                'best_params': grid_search.best_params_,
                'model': grid_search.best_estimator_
            }
            
            print(f"{name.upper()} - Best CV Score: {grid_search.best_score_:.4f}")
            
            # Track best model
            if grid_search.best_score_ > best_score:
                best_score = grid_search.best_score_
                best_model_name = name
                self.best_model = grid_search.best_estimator_
        
        print(f"\nBest Model: {best_model_name.upper()} with score: {best_score:.4f}")
        return results
    
    def evaluate_model(self, X_test, y_test):
        """Comprehensive model evaluation"""
        if self.best_model is None:
            raise ValueError("No model trained yet")
        
        y_pred = self.best_model.predict(X_test)
        
        metrics = {
            'accuracy': accuracy_score(y_test, y_pred),
            'precision': precision_score(y_test, y_pred, average='weighted'),
            'recall': recall_score(y_test, y_pred, average='weighted'),
            'f1': f1_score(y_test, y_pred, average='weighted')
        }
        
        return metrics
    
    def save_model(self, filepath: str):
        """Save trained model"""
        if self.best_model is None:
            raise ValueError("No model to save")
        joblib.dump(self.best_model, filepath)
    
    def load_model(self, filepath: str):
        """Load saved model"""
        self.best_model = joblib.load(filepath)

# Example usage
# ml_pipeline = AdvancedMLPipeline()
# results = ml_pipeline.train_with_model_selection(X, y, numeric_cols, categorical_cols)
# metrics = ml_pipeline.evaluate_model(X_test, y_test)
# ml_pipeline.save_model('best_model.joblib')

2. Building LLM-Powered Applications

Developing applications with Large Language Models involves several key stages, from API integration to advanced patterns like Retrieval Augmented Generation.

LLM APIs and Integration

Most LLM applications start by consuming models via APIs from providers like OpenAI, Anthropic, Google, or Hugging Face:

# Example: LLM API integration with multiple providers
import asyncio
import aiohttp
import openai
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
import json

class LLMProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    HUGGINGFACE = "huggingface"

@dataclass
class LLMRequest:
    """Standardized LLM request structure"""
    prompt: str
    max_tokens: int = 1000
    temperature: float = 0.7
    model: str = "gpt-3.5-turbo"
    system_message: Optional[str] = None

class LLMClient:
    """Universal LLM client supporting multiple providers"""
    
    def __init__(self):
        self.clients = {}
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    def configure_provider(self, provider: LLMProvider, api_key: str, base_url: str = None):
        """Configure API credentials for a provider"""
        config = {"api_key": api_key}
        if base_url:
            config["base_url"] = base_url
        self.clients[provider] = config
    
    async def generate_openai(self, request: LLMRequest) -> str:
        """Generate text using OpenAI API"""
        client = openai.AsyncOpenAI(api_key=self.clients[LLMProvider.OPENAI]["api_key"])
        
        messages = []
        if request.system_message:
            messages.append({"role": "system", "content": request.system_message})
        messages.append({"role": "user", "content": request.prompt})
        
        try:
            response = await client.chat.completions.create(
                model=request.model,
                messages=messages,
                max_tokens=request.max_tokens,
                temperature=request.temperature
            )
            return response.choices[0].message.content
        except Exception as e:
            raise Exception(f"OpenAI API error: {e}")
    
    async def generate_anthropic(self, request: LLMRequest) -> str:
        """Generate text using Anthropic API"""
        headers = {
            "Authorization": f"Bearer {self.clients[LLMProvider.ANTHROPIC]['api_key']}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": request.model or "claude-3-sonnet-20240229",
            "max_tokens": request.max_tokens,
            "temperature": request.temperature,
            "messages": [{"role": "user", "content": request.prompt}]
        }
        
        if request.system_message:
            payload["system"] = request.system_message
        
        async with self.session.post(
            "https://api.anthropic.com/v1/messages",
            headers=headers,
            json=payload
        ) as response:
            if response.status == 200:
                result = await response.json()
                return result["content"][0]["text"]
            else:
                raise Exception(f"Anthropic API error: {response.status}")
    
    async def generate(self, provider: LLMProvider, request: LLMRequest) -> str:
        """Generate text using specified provider"""
        if provider not in self.clients:
            raise ValueError(f"Provider {provider} not configured")
        
        if provider == LLMProvider.OPENAI:
            return await self.generate_openai(request)
        elif provider == LLMProvider.ANTHROPIC:
            return await self.generate_anthropic(request)
        else:
            raise NotImplementedError(f"Provider {provider} not implemented")

# Example usage
async def main():
    async with LLMClient() as client:
        # Configure providers
        client.configure_provider(LLMProvider.OPENAI, "your-openai-key")
        client.configure_provider(LLMProvider.ANTHROPIC, "your-anthropic-key")
        
        # Create request
        request = LLMRequest(
            prompt="Explain quantum computing in simple terms",
            max_tokens=500,
            temperature=0.7,
            system_message="You are a helpful AI assistant specializing in technology explanations."
        )
        
        # Generate responses from multiple providers
        providers = [LLMProvider.OPENAI, LLMProvider.ANTHROPIC]
        tasks = [client.generate(provider, request) for provider in providers]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        for provider, response in zip(providers, responses):
            if isinstance(response, Exception):
                print(f"{provider.value}: Error - {response}")
            else:
                print(f"{provider.value}: {response[:100]}...")

# asyncio.run(main())

Prompt Engineering Techniques

Effective prompt engineering is crucial for getting consistent, high-quality outputs from LLMs:

# Example: Advanced prompt engineering techniques
from typing import List, Dict, Any
import json
import re

class PromptEngineer:
    """Advanced prompt engineering utilities"""
    
    @staticmethod
    def zero_shot_prompt(task: str, context: str = None) -> str:
        """Create zero-shot prompt"""
        prompt = f"Task: {task}\n\n"
        if context:
            prompt += f"Context: {context}\n\n"
        prompt += "Please complete this task:"
        return prompt
    
    @staticmethod
    def few_shot_prompt(task: str, examples: List[Dict[str, str]], query: str) -> str:
        """Create few-shot prompt with examples"""
        prompt = f"Task: {task}\n\n"
        
        for i, example in enumerate(examples, 1):
            prompt += f"Example {i}:\n"
            prompt += f"Input: {example['input']}\n"
            prompt += f"Output: {example['output']}\n\n"
        
        prompt += f"Now, please complete the task for:\nInput: {query}\nOutput:"
        return prompt
    
    @staticmethod
    def chain_of_thought_prompt(problem: str) -> str:
        """Create chain-of-thought prompt"""
        return f"""
Problem: {problem}

Please solve this step by step:
1. First, identify what information is given
2. Determine what needs to be found
3. Choose the appropriate method or formula
4. Work through the solution step by step
5. State your final answer

Let's think through this systematically:
"""
    
    @staticmethod
    def react_prompt(task: str, available_tools: List[str]) -> str:
        """Create ReAct (Reasoning + Acting) prompt"""
        tools_list = "\n".join(f"- {tool}" for tool in available_tools)
        
        return f"""
You are an AI assistant that can use tools to help solve tasks. 

Task: {task}

Available tools:
{tools_list}

Use the following format:
Thought: [your reasoning about what to do next]
Action: [the tool to use]
Action Input: [the input for the tool]
Observation: [the result from the tool]
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: [the final answer to the original task]

Begin:
"""
    
    @staticmethod
    def structured_output_prompt(task: str, output_schema: Dict[str, Any]) -> str:
        """Create prompt for structured JSON output"""
        schema_str = json.dumps(output_schema, indent=2)
        
        return f"""
Task: {task}

Please provide your response as a valid JSON object that follows this exact schema:

{schema_str}

Important:
- Your response must be valid JSON
- All required fields must be included
- Follow the data types specified
- Do not include any text outside the JSON object

JSON Response:
"""

class SmartPromptManager:
    """Advanced prompt management and optimization"""
    
    def __init__(self):
        self.templates = {}
        self.performance_history = {}
    
    def register_template(self, name: str, template: str, variables: List[str]):
        """Register a prompt template"""
        self.templates[name] = {
            'template': template,
            'variables': variables,
            'usage_count': 0,
            'success_rate': 0.0
        }
    
    def generate_prompt(self, template_name: str, **kwargs) -> str:
        """Generate prompt from template with variables"""
        if template_name not in self.templates:
            raise ValueError(f"Template {template_name} not found")
        
        template_info = self.templates[template_name]
        template = template_info['template']
        
        # Check all required variables are provided
        missing_vars = set(template_info['variables']) - set(kwargs.keys())
        if missing_vars:
            raise ValueError(f"Missing variables: {missing_vars}")
        
        # Update usage count
        self.templates[template_name]['usage_count'] += 1
        
        # Format template
        return template.format(**kwargs)
    
    def record_success(self, template_name: str, success: bool):
        """Record template performance"""
        if template_name not in self.performance_history:
            self.performance_history[template_name] = []
        
        self.performance_history[template_name].append(success)
        
        # Update success rate
        successes = sum(self.performance_history[template_name])
        total = len(self.performance_history[template_name])
        self.templates[template_name]['success_rate'] = successes / total
    
    def get_best_template(self, min_usage: int = 5) -> str:
        """Get template with highest success rate"""
        eligible_templates = {
            name: info for name, info in self.templates.items()
            if info['usage_count'] >= min_usage
        }
        
        if not eligible_templates:
            return None
        
        best_template = max(
            eligible_templates.keys(),
            key=lambda x: eligible_templates[x]['success_rate']
        )
        
        return best_template

# Example usage
prompt_manager = SmartPromptManager()

# Register templates
prompt_manager.register_template(
    "code_review",
    """
Review the following {language} code for:
1. Code quality and best practices
2. Potential bugs or issues
3. Performance optimization opportunities
4. Security considerations

Code to review:
```{language}
{code}
```

Please provide feedback in the following format:
- **Overall Assessment**: [Brief summary]
- **Issues Found**: [List any problems]
- **Recommendations**: [Suggestions for improvement]
- **Security Notes**: [Any security concerns]
""",
    ["language", "code"]
)

# Generate and use prompt
code_review_prompt = prompt_manager.generate_prompt(
    "code_review",
    language="python",
    code="def divide(a, b): return a / b"
)

print("Generated prompt:")
print(code_review_prompt)

Retrieval Augmented Generation (RAG)

RAG is a critical pattern that grounds LLM responses with relevant information from external knowledge bases, reducing hallucinations and improving accuracy:

# Example: Complete RAG implementation
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
import tiktoken
from pathlib import Path
import fitz  # PyMuPDF for PDF processing

@dataclass
class Document:
    """Document representation"""
    content: str
    metadata: Dict[str, Any]
    source: str

@dataclass
class Chunk:
    """Text chunk with embedding"""
    text: str
    embedding: Optional[np.ndarray] = None
    metadata: Dict[str, Any] = None
    doc_id: str = None

class DocumentProcessor:
    """Process various document formats"""
    
    def __init__(self):
        self.encoding = tiktoken.get_encoding("cl100k_base")
    
    def load_pdf(self, file_path: str) -> Document:
        """Load PDF document"""
        doc = fitz.open(file_path)
        content = ""
        
        for page_num in range(doc.page_count):
            page = doc[page_num]
            content += page.get_text()
        
        doc.close()
        
        return Document(
            content=content,
            metadata={"type": "pdf", "pages": doc.page_count},
            source=file_path
        )
    
    def load_text(self, file_path: str) -> Document:
        """Load text document"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        return Document(
            content=content,
            metadata={"type": "text"},
            source=file_path
        )
    
    def chunk_document(self, doc: Document, chunk_size: int = 1000, overlap: int = 200) -> List[Chunk]:
        """Split document into overlapping chunks"""
        tokens = self.encoding.encode(doc.content)
        chunks = []
        
        for i in range(0, len(tokens), chunk_size - overlap):
            chunk_tokens = tokens[i:i + chunk_size]
            chunk_text = self.encoding.decode(chunk_tokens)
            
            chunk = Chunk(
                text=chunk_text,
                metadata={
                    **doc.metadata,
                    "chunk_index": len(chunks),
                    "start_token": i,
                    "end_token": i + len(chunk_tokens)
                },
                doc_id=doc.source
            )
            chunks.append(chunk)
        
        return chunks

class VectorStore:
    """Vector database for semantic search"""
    
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(embedding_model)
        self.index = None
        self.chunks: List[Chunk] = []
        self.dimension = self.model.get_sentence_embedding_dimension()
    
    def add_documents(self, chunks: List[Chunk]):
        """Add document chunks to vector store"""
        # Generate embeddings
        texts = [chunk.text for chunk in chunks]
        embeddings = self.model.encode(texts)
        
        # Store chunks with embeddings
        for chunk, embedding in zip(chunks, embeddings):
            chunk.embedding = embedding
            self.chunks.append(chunk)
        
        # Create FAISS index
        if self.index is None:
            self.index = faiss.IndexFlatIP(self.dimension)  # Inner product for cosine similarity
        
        # Normalize embeddings for cosine similarity
        embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
        self.index.add(embeddings.astype(np.float32))
    
    def search(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
        """Search for relevant chunks"""
        if self.index is None or len(self.chunks) == 0:
            return []
        
        # Generate query embedding
        query_embedding = self.model.encode([query])
        query_embedding = query_embedding / np.linalg.norm(query_embedding)
        
        # Search
        scores, indices = self.index.search(query_embedding.astype(np.float32), k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.chunks):
                results.append({
                    'chunk': self.chunks[idx],
                    'score': float(score),
                    'text': self.chunks[idx].text,
                    'metadata': self.chunks[idx].metadata
                })
        
        return results

class RAGSystem:
    """Complete RAG implementation"""
    
    def __init__(self, llm_client, embedding_model: str = "all-MiniLM-L6-v2"):
        self.llm_client = llm_client
        self.vector_store = VectorStore(embedding_model)
        self.processor = DocumentProcessor()
    
    async def ingest_documents(self, file_paths: List[str]):
        """Ingest multiple documents"""
        all_chunks = []
        
        for file_path in file_paths:
            path = Path(file_path)
            
            if path.suffix.lower() == '.pdf':
                doc = self.processor.load_pdf(file_path)
            elif path.suffix.lower() in ['.txt', '.md']:
                doc = self.processor.load_text(file_path)
            else:
                print(f"Unsupported file type: {file_path}")
                continue
            
            chunks = self.processor.chunk_document(doc)
            all_chunks.extend(chunks)
            print(f"Processed {file_path}: {len(chunks)} chunks")
        
        # Add to vector store
        self.vector_store.add_documents(all_chunks)
        print(f"Ingested {len(all_chunks)} total chunks")
    
    async def query(self, question: str, k: int = 3) -> Dict[str, Any]:
        """Query the RAG system"""
        # Retrieve relevant chunks
        search_results = self.vector_store.search(question, k)
        
        if not search_results:
            return {
                'answer': "I don't have relevant information to answer your question.",
                'sources': [],
                'context_used': []
            }
        
        # Prepare context
        context_pieces = []
        sources = set()
        
        for result in search_results:
            context_pieces.append(result['text'])
            sources.add(result['chunk'].doc_id)
        
        context = "\n\n".join(context_pieces)
        
        # Generate answer using RAG prompt
        rag_prompt = f"""
Based on the following context, please answer the question. If the context doesn't contain enough information to answer the question, please say so.

Context:
{context}

Question: {question}

Answer:"""
        
        # Generate response using LLM
        request = LLMRequest(
            prompt=rag_prompt,
            max_tokens=500,
            temperature=0.1,  # Lower temperature for factual answers
            system_message="You are a helpful assistant that answers questions based on provided context. Be accurate and cite the context when relevant."
        )
        
        answer = await self.llm_client.generate(LLMProvider.OPENAI, request)
        
        return {
            'answer': answer,
            'sources': list(sources),
            'context_used': context_pieces,
            'relevance_scores': [r['score'] for r in search_results]
        }

# Example usage
async def rag_example():
    async with LLMClient() as llm_client:
        llm_client.configure_provider(LLMProvider.OPENAI, "your-openai-key")
        
        # Initialize RAG system
        rag = RAGSystem(llm_client)
        
        # Ingest documents (example paths)
        # await rag.ingest_documents([
        #     "documents/ai_handbook.pdf",
        #     "documents/ml_guide.txt",
        #     "documents/python_tutorial.md"
        # ])
        
        # Query the system
        # response = await rag.query("What are the key principles of machine learning?")
        # print("Answer:", response['answer'])
        # print("Sources:", response['sources'])

# asyncio.run(rag_example())

3. Agent Development & Coordination

AI agents are sophisticated software programs that can autonomously perform tasks by reasoning, making decisions, and interacting with external systems through tools and APIs.

Tool-Using Agents

Modern AI agents derive their power from the ability to use external tools and APIs to gather information and perform actions:

# Example: Comprehensive tool-using agent implementation
import asyncio
import json
import inspect
from typing import Dict, Any, Callable, List, Optional
from dataclasses import dataclass
from enum import Enum
import aiohttp
import datetime

class ToolType(Enum):
    SEARCH = "search"
    CALCULATOR = "calculator"
    FILE_OPERATION = "file_operation"
    API_CALL = "api_call"
    DATABASE = "database"

@dataclass
class ToolDefinition:
    """Tool definition with metadata"""
    name: str
    description: str
    parameters: Dict[str, Any]
    tool_type: ToolType
    function: Callable
    
class ToolRegistry:
    """Registry for managing agent tools"""
    
    def __init__(self):
        self.tools: Dict[str, ToolDefinition] = {}
    
    def register_tool(self, tool_def: ToolDefinition):
        """Register a new tool"""
        self.tools[tool_def.name] = tool_def
    
    def get_tool(self, name: str) -> Optional[ToolDefinition]:
        """Get tool by name"""
        return self.tools.get(name)
    
    def get_tools_schema(self) -> List[Dict[str, Any]]:
        """Get OpenAI-compatible tools schema"""
        schema = []
        for tool in self.tools.values():
            schema.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            })
        return schema

class BuiltinTools:
    """Built-in tools for agents"""
    
    @staticmethod
    async def web_search(query: str) -> str:
        """Search the web for information"""
        # Simulated web search - replace with actual search API
        await asyncio.sleep(1)  # Simulate API call
        return f"Search results for '{query}': Found relevant information about {query}."
    
    @staticmethod
    async def calculator(expression: str) -> str:
        """Evaluate mathematical expressions"""
        try:
            # Safe evaluation for basic math
            allowed_chars = set('0123456789+-*/.() ')
            if not all(c in allowed_chars for c in expression):
                return "Error: Invalid characters in expression"
            
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"Error: {str(e)}"
    
    @staticmethod
    async def get_current_time() -> str:
        """Get current date and time"""
        now = datetime.datetime.now()
        return now.strftime("%Y-%m-%d %H:%M:%S")
    
    @staticmethod
    async def file_read(filename: str) -> str:
        """Read content from a file"""
        try:
            with open(filename, 'r') as f:
                content = f.read()
            return f"File content:\n{content}"
        except FileNotFoundError:
            return f"Error: File '{filename}' not found"
        except Exception as e:
            return f"Error reading file: {str(e)}"

class ToolUsingAgent:
    """AI agent capable of using tools"""
    
    def __init__(self, llm_client, name: str = "Assistant"):
        self.llm_client = llm_client
        self.name = name
        self.tools = ToolRegistry()
        self.conversation_history = []
        self._register_builtin_tools()
    
    def _register_builtin_tools(self):
        """Register built-in tools"""
        tools_to_register = [
            ToolDefinition(
                name="web_search",
                description="Search the web for current information",
                parameters={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "The search query"
                        }
                    },
                    "required": ["query"]
                },
                tool_type=ToolType.SEARCH,
                function=BuiltinTools.web_search
            ),
            ToolDefinition(
                name="calculator",
                description="Perform mathematical calculations",
                parameters={
                    "type": "object",
                    "properties": {
                        "expression": {
                            "type": "string",
                            "description": "Mathematical expression to evaluate"
                        }
                    },
                    "required": ["expression"]
                },
                tool_type=ToolType.CALCULATOR,
                function=BuiltinTools.calculator
            ),
            ToolDefinition(
                name="get_current_time",
                description="Get the current date and time",
                parameters={
                    "type": "object",
                    "properties": {},
                    "required": []
                },
                tool_type=ToolType.API_CALL,
                function=BuiltinTools.get_current_time
            )
        ]
        
        for tool in tools_to_register:
            self.tools.register_tool(tool)
    
    async def execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> str:
        """Execute a tool with given parameters"""
        tool = self.tools.get_tool(tool_name)
        if not tool:
            return f"Error: Tool '{tool_name}' not found"
        
        try:
            # Get function signature
            sig = inspect.signature(tool.function)
            
            # Filter parameters to match function signature
            filtered_params = {}
            for param_name, param in sig.parameters.items():
                if param_name in parameters:
                    filtered_params[param_name] = parameters[param_name]
            
            # Execute tool
            if asyncio.iscoroutinefunction(tool.function):
                result = await tool.function(**filtered_params)
            else:
                result = tool.function(**filtered_params)
            
            return str(result)
        except Exception as e:
            return f"Error executing {tool_name}: {str(e)}"
    
    async def process_message(self, user_message: str) -> str:
        """Process user message and potentially use tools"""
        # Add user message to history
        self.conversation_history.append({"role": "user", "content": user_message})
        
        # Create system message with tool information
        tools_info = "\n".join([
            f"- {tool.name}: {tool.description}"
            for tool in self.tools.tools.values()
        ])
        
        system_message = f"""You are {self.name}, an AI assistant with access to tools.

Available tools:
{tools_info}

When you need to use a tool, respond with a JSON object in this format:
{{
    "tool_call": {{
        "name": "tool_name",
        "parameters": {{"param1": "value1", "param2": "value2"}}
    }}
}}

If you don't need to use any tools, respond normally with text.
"""
        
        # Create request with conversation history
        messages = [{"role": "system", "content": system_message}]
        messages.extend(self.conversation_history[-5:])  # Keep last 5 messages
        
        request = LLMRequest(
            prompt=messages[-1]["content"],
            system_message=system_message,
            max_tokens=1000,
            temperature=0.7
        )
        
        # Get LLM response
        response = await self.llm_client.generate(LLMProvider.OPENAI, request)
        
        # Check if response contains tool call
        if "tool_call" in response and "{" in response:
            try:
                # Extract JSON from response
                json_start = response.find("{")
                json_end = response.rfind("}") + 1
                json_str = response[json_start:json_end]
                tool_call = json.loads(json_str)
                
                if "tool_call" in tool_call:
                    tool_info = tool_call["tool_call"]
                    tool_name = tool_info["name"]
                    tool_params = tool_info["parameters"]
                    
                    # Execute tool
                    tool_result = await self.execute_tool(tool_name, tool_params)
                    
                    # Generate final response with tool result
                    final_prompt = f"""
Based on the tool result below, provide a helpful response to the user's question.

User question: {user_message}
Tool used: {tool_name}
Tool result: {tool_result}

Response:"""
                    
                    final_request = LLMRequest(
                        prompt=final_prompt,
                        max_tokens=500,
                        temperature=0.7
                    )
                    
                    final_response = await self.llm_client.generate(LLMProvider.OPENAI, final_request)
                    
                    # Add to history
                    self.conversation_history.append({
                        "role": "assistant",
                        "content": final_response,
                        "tool_used": tool_name,
                        "tool_result": tool_result
                    })
                    
                    return final_response
                    
            except json.JSONDecodeError:
                pass  # Fall through to normal response
        
        # Normal response without tools
        self.conversation_history.append({"role": "assistant", "content": response})
        return response

# Example usage
async def agent_example():
    async with LLMClient() as llm_client:
        llm_client.configure_provider(LLMProvider.OPENAI, "your-openai-key")
        
        agent = ToolUsingAgent(llm_client, "Smart Assistant")
        
        # Test conversations
        test_queries = [
            "What's the current time?",
            "Calculate 25 * 17 + 123",
            "Search for information about quantum computing",
            "What's the weather like today?"
        ]
        
        for query in test_queries:
            print(f"\nUser: {query}")
            response = await agent.process_message(query)
            print(f"Agent: {response}")

# asyncio.run(agent_example())

Multi-Agent Systems and Coordination

Advanced AI applications often require multiple agents working together, each specialized for different tasks:

# Example: Multi-agent system with coordination
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import uuid

class AgentRole(Enum):
    RESEARCHER = "researcher"
    WRITER = "writer"
    REVIEWER = "reviewer"
    COORDINATOR = "coordinator"

class MessageType(Enum):
    TASK_ASSIGNMENT = "task_assignment"
    TASK_RESULT = "task_result"
    COLLABORATION_REQUEST = "collaboration_request"
    STATUS_UPDATE = "status_update"

@dataclass
class Message:
    """Inter-agent message"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    sender: str = ""
    receiver: str = ""
    message_type: MessageType = MessageType.STATUS_UPDATE
    content: Any = None
    timestamp: float = field(default_factory=lambda: asyncio.get_event_loop().time())

@dataclass
class Task:
    """Task definition for agents"""
    id: str
    description: str
    assigned_to: str
    status: str = "pending"
    result: Optional[Any] = None
    dependencies: List[str] = field(default_factory=list)

class Agent:
    """Base agent class"""
    
    def __init__(self, agent_id: str, role: AgentRole, llm_client):
        self.id = agent_id
        self.role = role
        self.llm_client = llm_client
        self.inbox: List[Message] = []
        self.tasks: Dict[str, Task] = {}
        self.is_busy = False
    
    async def send_message(self, receiver: str, message: Message, message_bus):
        """Send message to another agent"""
        message.sender = self.id
        message.receiver = receiver
        await message_bus.deliver_message(message)
    
    async def receive_message(self, message: Message):
        """Receive and process message"""
        self.inbox.append(message)
        await self.process_message(message)
    
    async def process_message(self, message: Message):
        """Process incoming message"""
        if message.message_type == MessageType.TASK_ASSIGNMENT:
            task = message.content
            self.tasks[task.id] = task
            await self.execute_task(task)
        elif message.message_type == MessageType.COLLABORATION_REQUEST:
            await self.handle_collaboration_request(message)
    
    async def execute_task(self, task: Task):
        """Execute assigned task"""
        self.is_busy = True
        task.status = "in_progress"
        
        try:
            result = await self._perform_role_specific_work(task)
            task.result = result
            task.status = "completed"
        except Exception as e:
            task.status = "failed"
            task.result = f"Error: {str(e)}"
        
        self.is_busy = False
    
    async def _perform_role_specific_work(self, task: Task) -> str:
        """Override in subclasses for role-specific behavior"""
        return f"Task {task.id} completed by {self.role.value}"
    
    async def handle_collaboration_request(self, message: Message):
        """Handle collaboration requests from other agents"""
        pass

class ResearcherAgent(Agent):
    """Specialized agent for research tasks"""
    
    def __init__(self, agent_id: str, llm_client):
        super().__init__(agent_id, AgentRole.RESEARCHER, llm_client)
    
    async def _perform_role_specific_work(self, task: Task) -> str:
        """Perform research on the given topic"""
        research_prompt = f"""
You are a research specialist. Your task is to gather comprehensive information about the following topic:

Topic: {task.description}

Please provide:
1. Key concepts and definitions
2. Current trends and developments
3. Important facts and statistics
4. Relevant examples
5. Potential challenges or limitations

Research findings:"""
        
        request = LLMRequest(
            prompt=research_prompt,
            max_tokens=1500,
            temperature=0.3,
            system_message="You are an expert researcher with access to comprehensive knowledge."
        )
        
        research_result = await self.llm_client.generate(LLMProvider.OPENAI, request)
        return research_result

class WriterAgent(Agent):
    """Specialized agent for writing tasks"""
    
    def __init__(self, agent_id: str, llm_client):
        super().__init__(agent_id, AgentRole.WRITER, llm_client)
    
    async def _perform_role_specific_work(self, task: Task) -> str:
        """Write content based on research or requirements"""
        writing_prompt = f"""
You are a professional writer. Create engaging, well-structured content for the following:

Task: {task.description}

Requirements:
- Clear and engaging writing style
- Well-organized structure with headings
- Appropriate tone for the target audience
- Include relevant examples where applicable
- Ensure accuracy and clarity

Written content:"""
        
        request = LLMRequest(
            prompt=writing_prompt,
            max_tokens=2000,
            temperature=0.7,
            system_message="You are a skilled writer who creates compelling, accurate content."
        )
        
        written_content = await self.llm_client.generate(LLMProvider.OPENAI, request)
        return written_content

class ReviewerAgent(Agent):
    """Specialized agent for reviewing and quality control"""
    
    def __init__(self, agent_id: str, llm_client):
        super().__init__(agent_id, AgentRole.REVIEWER, llm_client)
    
    async def _perform_role_specific_work(self, task: Task) -> str:
        """Review content for quality and accuracy"""
        review_prompt = f"""
You are a quality assurance specialist. Review the following content:

Content to review: {task.description}

Please provide:
1. Overall quality assessment
2. Accuracy check
3. Clarity and readability evaluation
4. Suggestions for improvement
5. Final recommendation (approve/revise/reject)

Review report:"""
        
        request = LLMRequest(
            prompt=review_prompt,
            max_tokens=1000,
            temperature=0.2,
            system_message="You are a meticulous reviewer focused on quality and accuracy."
        )
        
        review_result = await self.llm_client.generate(LLMProvider.OPENAI, request)
        return review_result

class MessageBus:
    """Message delivery system for agents"""
    
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
    
    def register_agent(self, agent: Agent):
        """Register agent with message bus"""
        self.agents[agent.id] = agent
    
    async def deliver_message(self, message: Message):
        """Deliver message to target agent"""
        if message.receiver in self.agents:
            await self.agents[message.receiver].receive_message(message)
        else:
            print(f"Warning: Agent {message.receiver} not found")

class MultiAgentCoordinator:
    """Coordinates multiple agents for complex tasks"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
        self.message_bus = MessageBus()
        self.agents: Dict[str, Agent] = {}
        self.workflow_results: Dict[str, Any] = {}
    
    def add_agent(self, agent: Agent):
        """Add agent to the system"""
        self.agents[agent.id] = agent
        self.message_bus.register_agent(agent)
    
    async def create_workflow(self, description: str) -> Dict[str, Any]:
        """Create and execute a multi-agent workflow"""
        workflow_id = str(uuid.uuid4())
        
        # Step 1: Research
        research_task = Task(
            id=f"{workflow_id}_research",
            description=f"Research the following topic: {description}",
            assigned_to="researcher_001"
        )
        
        research_message = Message(
            message_type=MessageType.TASK_ASSIGNMENT,
            content=research_task
        )
        
        await self.message_bus.deliver_message(research_message)
        await self._wait_for_task_completion(research_task.id)
        
        research_result = self.agents["researcher_001"].tasks[research_task.id].result
        
        # Step 2: Write content based on research
        writing_task = Task(
            id=f"{workflow_id}_writing",
            description=f"Write comprehensive content about: {description}\n\nBased on this research:\n{research_result}",
            assigned_to="writer_001",
            dependencies=[research_task.id]
        )
        
        writing_message = Message(
            message_type=MessageType.TASK_ASSIGNMENT,
            content=writing_task
        )
        
        await self.message_bus.deliver_message(writing_message)
        await self._wait_for_task_completion(writing_task.id)
        
        written_content = self.agents["writer_001"].tasks[writing_task.id].result
        
        # Step 3: Review the content
        review_task = Task(
            id=f"{workflow_id}_review",
            description=written_content,
            assigned_to="reviewer_001",
            dependencies=[writing_task.id]
        )
        
        review_message = Message(
            message_type=MessageType.TASK_ASSIGNMENT,
            content=review_task
        )
        
        await self.message_bus.deliver_message(review_message)
        await self._wait_for_task_completion(review_task.id)
        
        review_result = self.agents["reviewer_001"].tasks[review_task.id].result
        
        return {
            'workflow_id': workflow_id,
            'research': research_result,
            'content': written_content,
            'review': review_result,
            'status': 'completed'
        }
    
    async def _wait_for_task_completion(self, task_id: str, timeout: int = 30):
        """Wait for task completion"""
        start_time = asyncio.get_event_loop().time()
        
        while True:
            # Check if any agent has completed the task
            for agent in self.agents.values():
                if task_id in agent.tasks and agent.tasks[task_id].status in ['completed', 'failed']:
                    return
            
            # Check timeout
            if asyncio.get_event_loop().time() - start_time > timeout:
                raise TimeoutError(f"Task {task_id} timed out")
            
            await asyncio.sleep(0.5)

# Example usage
async def multi_agent_example():
    async with LLMClient() as llm_client:
        llm_client.configure_provider(LLMProvider.OPENAI, "your-openai-key")
        
        # Create coordinator
        coordinator = MultiAgentCoordinator(llm_client)
        
        # Create specialized agents
        researcher = ResearcherAgent("researcher_001", llm_client)
        writer = WriterAgent("writer_001", llm_client)
        reviewer = ReviewerAgent("reviewer_001", llm_client)
        
        # Add agents to coordinator
        coordinator.add_agent(researcher)
        coordinator.add_agent(writer)
        coordinator.add_agent(reviewer)
        
        # Execute workflow
        result = await coordinator.create_workflow(
            "The impact of artificial intelligence on software development"
        )
        
        print("Workflow Results:")
        print(f"Research: {result['research'][:200]}...")
        print(f"Content: {result['content'][:200]}...")
        print(f"Review: {result['review'][:200]}...")

# asyncio.run(multi_agent_example())

Production Deployment Considerations

Deploying AI agents in production requires careful attention to monitoring, performance, and safety:

The Future of AI-Powered Development

As AI continues to evolve, software development is becoming increasingly intelligent and automated. The key trends shaping this transformation include:

• Code Generation AI: Tools like GitHub Copilot and Claude Code are revolutionizing how we write software
• Intelligent Testing: AI-powered testing tools that can generate test cases and find edge cases automatically
• Automated Documentation: AI systems that maintain up-to-date documentation as code evolves
• Smart Debugging: AI assistants that can diagnose and suggest fixes for complex bugs
• Predictive Maintenance: AI that monitors application health and predicts potential issues

Conclusion

Modern software development with AI represents a fundamental shift in how we build applications. From Python's rich ecosystem to sophisticated LLM-powered systems and autonomous agents, developers now have unprecedented tools to create intelligent, adaptive software.

Success in this new paradigm requires mastering not just traditional programming skills, but also understanding how to effectively integrate AI capabilities, design agent architectures, and build systems that can learn and adapt over time.

The future belongs to developers who can harness these AI capabilities to build the next generation of intelligent applications that don't just execute code, but truly understand, reason, and collaborate with users in meaningful ways.