Python for AI & LLM Engineering (2026): The Complete Beginner-to-Builder Masterclass

# Python for AI: Clean, readable, English-like syntax
def summarize_text(text: str, max_words: int = 50) -> str:
    """Simulate basic text truncation and word counting."""
    words = text.split()
    if len(words) <= max_words:
        return text
    return " ".join(words[:max_words]) + "..."

sample_review = "This AI course provides practical, industry-tested Python patterns for building modern LLM applications."
print(summarize_text(sample_review, max_words=6))
# Output: This AI course provides practical, industry-tested...

# Financial KPI calculations & Dynamic AI Prompt Formatting
revenue: float = 50_000_000_000.0  # $50 Billion (using underscores for readability)
expenses: float = 20_000_000_000.0 # $20 Billion
profit = revenue - expenses
profit_margin = (profit / revenue) * 100

print(f"Profit: USD {profit:,.2f} | Margin: {profit_margin:.1f}%")
# Output: Profit: USD 30,000,000,000.00 | Margin: 60.0%

# Routing prompts based on conditional threshold
sentiment_score = 0.25 # Scale from 0.0 (Negative) to 1.0 (Positive)

if sentiment_score >= 0.7:
    priority = "LOW"
    action = "Send thank-you email"
elif sentiment_score >= 0.4:
    priority = "MEDIUM"
    action = "Log feedback for product team"
else:
    priority = "URGENT"
    action = "Alert customer success team immediately"

print(f"Priority: {priority} -> Action: {action}")

# Python Collections for AI Prompt & Response Processing

# 1. Lists & List Comprehensions: Batch text preprocessing
raw_reviews = [
    "Amazing customer support! Resolved my issue in 5 minutes.",
    "Order was delayed by 3 days. Very unhappy.",
    "Product quality is decent, but packaging was damaged."
]

# Clean and filter reviews with length > 20 chars
clean_reviews = [rev.strip() for rev in raw_reviews if len(rev) > 20]

# 2. Dictionaries: Structured LLM Output Representation
ai_response_payload = {
    "status": "success",
    "model": "gemini-2.5-flash",
    "metadata": {
        "tokens_used": 142,
        "latency_ms": 320
    },
    "entities": [
        {"category": "Support", "sentiment": "Positive", "confidence": 0.98},
        {"category": "Delivery", "sentiment": "Negative", "confidence": 0.91}
    ]
}

# Safely extract values using .get() with default fallbacks
sentiment = ai_response_payload.get("metadata", {}).get("latency_ms", 0)
print(f"Inference Latency: {sentiment}ms")

# 3. Sets: Fast deduplication of detected topics
detected_tags = ["shipping", "billing", "support", "shipping", "support"]
unique_tags = list(set(detected_tags))
print(f"Unique Categories: {unique_tags}")

# Type Annotations & Pydantic Schema Validation for AI Outputs
from typing import List, Optional
from pydantic import BaseModel, Field

# 1. Define Strict Pydantic Schema for LLM Output
class FeedbackAnalysis(BaseModel):
    sentiment: str = Field(description="Must be 'positive', 'negative', or 'neutral'")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    key_issues: List[str] = Field(default_factory=list, description="List of identified issues")
    urgency_level: Optional[str] = Field(default="NORMAL")

# 2. Reusable Analysis Validator Function
def parse_and_validate_ai_response(raw_json: dict) -> FeedbackAnalysis:
    """Validates raw dictionary against Pydantic schema."""
    try:
        validated_result = FeedbackAnalysis(**raw_json)
        return validated_result
    except Exception as e:
        print(f"Schema Validation Error: {e}")
        raise

# Example Usage with LLM output
sample_llm_output = {
    "sentiment": "negative",
    "confidence": 0.94,
    "key_issues": ["payment_failed", "timeout_error"],
    "urgency_level": "HIGH"
}

analysis = parse_and_validate_ai_response(sample_llm_output)
print(f"Validated Sentiment: {analysis.sentiment.upper()} (Score: {analysis.confidence})")
print(f"Identified Issues: {analysis.key_issues}")

# Object-Oriented AI Client & Agent Wrapper Pattern
from typing import List, Dict

class BaseAIAgent:
    """Base class encapsulating LLM configuration and conversation memory."""
    
    def __init__(self, agent_name: str, model: str = "gemini-2.5-flash", temperature: float = 0.7):
        self.agent_name = agent_name
        self.model = model
        self.temperature = temperature
        self.conversation_history: List[Dict[str, str]] = []
        
    def add_to_history(self, role: str, content: str) -> None:
        self.conversation_history.append({"role": role, "content": content})
        
    def clear_memory(self) -> None:
        self.conversation_history = []
        print(f"[{self.agent_name}] Conversation memory reset.")

class SentimentAnalysisAgent(BaseAIAgent):
    """Specialized agent inheriting from BaseAIAgent for sentiment classification."""
    
    def __init__(self, api_key: str):
        super().__init__(agent_name="SentimentAnalyzer", model="gemini-2.5-flash", temperature=0.1)
        self.api_key = api_key
        
    def analyze_feedback(self, text: str) -> Dict[str, any]:
        self.add_to_history("user", text)
        # Mocking LLM response generation
        mock_result = {
            "text": text,
            "sentiment": "POSITIVE" if "great" in text.lower() else "NEGATIVE",
            "model_used": self.model
        }
        self.add_to_history("assistant", str(mock_result))
        return mock_result

# Instantiate and use the specialized agent
analyzer = SentimentAnalysisAgent(api_key="SECURE_API_KEY")
result = analyzer.analyze_feedback("SkillCharged tutorials are great!")
print(f"Agent Analysis: {result}")

# Safe File Operations and Resilient API Error Handling
import json
import time
from typing import Dict, Any

# 1. Reading & Writing JSON Files with Context Managers
def save_feedback_dataset(data: Dict[str, Any], filepath: str = "feedback.json") -> None:
    """Save structured feedback to disk safely."""
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
    print(f"✓ Saved dataset to {filepath}")

def load_system_prompt(filepath: str = "system_prompt.txt") -> str:
    """Load prompt template with fallback on missing file."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read().strip()
    except FileNotFoundError:
        print(f"Warning: {filepath} not found. Using default prompt.")
        return "You are a helpful customer support AI assistant."

# 2. Resilient API Invocation with Retry Logic
def robust_ai_call_with_retry(prompt: str, max_retries: int = 3) -> str:
    """Simulate API call with exponential backoff retry on errors."""
    for attempt in range(1, max_retries + 1):
        try:
            # Simulated API call that might fail
            if attempt < 2:
                raise ConnectionError("Temporary network timeout connecting to AI gateway.")
            return "✓ AI Response generated successfully."
        except ConnectionError as err:
            wait_time = attempt * 2
            print(f"[Attempt {attempt}/{max_retries}] {err}. Retrying in {wait_time}s...")
            time.sleep(wait_time)
        except Exception as unhandled:
            print(f"Fatal unrecoverable error: {unhandled}")
            raise
    return "Fallback: Could not complete AI request."

# Execute robust workflow
save_feedback_dataset({"reviews": [{"id": 1, "text": "Great service!"}]})
prompt = load_system_prompt("non_existent_file.txt")
response = robust_ai_call_with_retry(prompt)
print(response)

# Google Gemini LLM SDK Integration for Customer Feedback
import os
from google import genai
from google.genai import types

# Initialize client using environment variable GEMINI_API_KEY
client = genai.Client()

def analyze_customer_review(review_text: str) -> str:
    """Invokes Gemini 2.5 Flash to extract sentiment and actionable insights."""
    prompt = f"""
You are a senior customer feedback intelligence analyst.
Analyze the following customer review:
"""{review_text}"""

Provide a structured response:
1. Sentiment: [Positive / Negative / Neutral]
2. Primary Category: [Pricing / Delivery / Product Quality / Customer Support]
3. Summary: [1-sentence concise summary]
4. Recommended Action: [Immediate operational recommendation]
"""

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.2, # Low temperature for consistent, factual output
            max_output_tokens=300
        )
    )
    
    return response.text

# Example Test
sample = "I love the new mobile app design, but the delivery took 6 days instead of 2."
print(analyze_customer_review(sample))

# High-Performance FastAPI Backend for AI Feedback Analysis
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional

app = FastAPI(
    title="Customer Feedback AI Intelligence API",
    description="Microservice for analyzing customer reviews using LLM sentiment pipelines",
    version="1.0.0"
)

# 1. Request & Response Schemas
class FeedbackRequest(BaseModel):
    customer_id: str
    review_text: str = Field(min_length=5, description="Customer review text")
    rating: int = Field(ge=1, le=5, description="1 to 5 star rating")

class FeedbackResponse(BaseModel):
    customer_id: str
    sentiment: str
    category: str
    urgency: str
    summary: str

# 2. API Endpoints
@app.get("/")
async def health_check():
    return {"status": "online", "service": "AI Feedback Engine"}

@app.post("/api/analyze", response_model=FeedbackResponse)
async def analyze_feedback_endpoint(payload: FeedbackRequest):
    if not payload.review_text.strip():
        raise HTTPException(status_code=400, detail="Review text cannot be empty.")
    
    # Mocking AI Inference logic (or invoking Gemini client)
    is_negative = payload.rating <= 2 or "bad" in payload.review_text.lower()
    
    return FeedbackResponse(
        customer_id=payload.customer_id,
        sentiment="NEGATIVE" if is_negative else "POSITIVE",
        category="Customer Support" if "support" in payload.review_text.lower() else "General",
        urgency="HIGH" if is_negative else "LOW",
        summary=f"Customer rated {payload.rating} stars: {payload.review_text[:50]}..."
    )

# SQLite Database Persistence Module for AI Feedback
import sqlite3
from typing import List, Dict, Any

class FeedbackDatabase:
    """Manages SQLite database connections and CRUD operations."""
    
    def __init__(self, db_path: str = "feedback.db"):
        self.db_path = db_path
        self._init_db()
        
    def _init_db(self) -> None:
        """Initializes the database schema if tables do not exist."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS customer_feedback (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    customer_id TEXT NOT NULL,
                    review_text TEXT NOT NULL,
                    rating INTEGER NOT NULL,
                    sentiment TEXT NOT NULL,
                    category TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            conn.commit()
            
    def insert_feedback(self, customer_id: str, text: str, rating: int, sentiment: str, category: str) -> int:
        """Inserts a new analyzed review using safe parameterized SQL."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO customer_feedback (customer_id, review_text, rating, sentiment, category)
                VALUES (?, ?, ?, ?, ?)
            """, (customer_id, text, rating, sentiment, category))
            conn.commit()
            return cursor.lastrowid

    def get_recent_feedback(self, limit: int = 10) -> List[Dict[str, Any]]:
        """Fetches the latest analyzed feedback entries."""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row  # Enables column-name dictionary access
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM customer_feedback ORDER BY created_at DESC LIMIT ?", (limit,))
            rows = cursor.fetchall()
            return [dict(row) for row in rows]

# Test Database Operations
db = FeedbackDatabase()
inserted_id = db.insert_feedback("cust_101", "Fast shipping and great product!", 5, "POSITIVE", "Delivery")
print(f"✓ Inserted Record ID: {inserted_id}")
print(f"Latest Records: {db.get_recent_feedback(1)}")

# Streamlit Interactive Frontend Dashboard for AI Feedback System
import streamlit as st
import requests
import sqlite3
import pandas as pd

st.set_page_config(page_title="AI Feedback Intelligence", page_icon="📊", layout="wide")

st.title("📊 Customer Feedback AI Intelligence Platform")
st.markdown("Real-time customer review sentiment analysis powered by **FastAPI + Gemini LLM + SQLite**.")

# Tab 1: Submit New Feedback | Tab 2: Live Analytics Dashboard
tab_submit, tab_dashboard = st.tabs(["📝 Submit Review", "📈 Live Analytics"])

with tab_submit:
    st.subheader("Analyze New Customer Feedback")
    with st.form("feedback_form"):
        customer_id = st.text_input("Customer ID / Email", value="user@example.com")
        rating = st.slider("Star Rating", min_value=1, max_value=5, value=4)
        review_text = st.text_area("Customer Review Text", placeholder="Write customer feedback here...")
        submitted = st.form_submit_button("Run AI Analysis")

        if submitted and review_text.strip():
            with st.spinner("Invoking Gemini AI Pipeline..."):
                # Call FastAPI Backend Endpoint
                try:
                    payload = {"customer_id": customer_id, "review_text": review_text, "rating": rating}
                    response = requests.post("http://localhost:8000/api/analyze", json=payload)
                    
                    if response.status_code == 200:
                        data = response.json()
                        st.success("✓ Analysis Complete & Saved to Database!")
                        
                        col1, col2, col3 = st.columns(3)
                        col1.metric("Sentiment", data["sentiment"])
                        col2.metric("Category", data["category"])
                        col3.metric("Urgency", data["urgency"])
                        st.info(f"**AI Summary:** {data['summary']}")
                    else:
                        st.error(f"Backend Error: {response.text}")
                except Exception as e:
                    st.error(f"Could not connect to FastAPI backend: {e}")

with tab_dashboard:
    st.subheader("Stored Customer Feedback Records")
    try:
        conn = sqlite3.connect("feedback.db")
        df = pd.read_sql_query("SELECT * FROM customer_feedback ORDER BY created_at DESC", conn)
        conn.close()
        
        if not df.empty:
            col_a, col_b = st.columns(2)
            col_a.metric("Total Reviews Analyzed", len(df))
            positive_pct = (df['sentiment'] == 'POSITIVE').mean() * 100
            col_b.metric("Positive Sentiment Rate", f"{positive_pct:.1f}%")
            
            st.dataframe(df, use_container_width=True)
        else:
            st.write("No feedback records in database yet. Submit a review above!")
    except Exception as e:
        st.info("Database initializing...")