🐍 Python Essentials for GenAI

# ─────────────────────────────────────────────────
# JS/TS → Python Translation Table
# ─────────────────────────────────────────────────

# Variables
name = "Ajmal"              # let name = "Ajmal"
age: int = 28               # let age: number = 28
PI = 3.14159                # const PI = 3.14159 (convention: UPPERCASE = constant)

# Strings (f-strings = template literals)
greeting = f"Hello, {name}" # `Hello, ${name}`

# Multiline strings
text = """
This is a
multiline string
"""                          # Same as backtick strings in JS

# Boolean
is_active = True             # true → True, false → False, null → None

# None (like null/undefined)
value = None                 # let value = null

# Print
print("Hello")              # console.log("Hello")
print(f"Name: {name}")      # console.log(`Name: ${name}`)

# Python uses INDENTATION instead of {}
# This is the #1 thing to get used to

# JavaScript:
# if (age > 18) {
#     console.log("Adult");
# } else {
#     console.log("Minor");
# }

# Python:
if age > 18:
    print("Adult")
else:
    print("Minor")

# Functions
def greet(name: str) -> str:    # function greet(name: string): string
    return f"Hello, {name}!"

# For loops
for i in range(5):              # for (let i = 0; i < 5; i++)
    print(i)

for item in my_list:            # for (const item of myList)
    print(item)

# While loops
while count > 0:                # while (count > 0) {
    count -= 1                  #     count--;
                                # }

# Python falsy values:
# False, None, 0, 0.0, "", [], {}, set()

# ⚠️ Key difference from JS:
# JS:  [] is truthy     → if ([]) { "runs!" }
# Py:  [] is falsy      → if []:   "NEVER runs"

# JS:  {} is truthy     → if ({}) { "runs!" }
# Py:  {} is falsy      → if {}:   "NEVER runs"

# Check for None specifically:
if value is None:        # NOT: if value == None
    print("No value")

if value is not None:    # NOT: if value != None
    print("Has value")

# Creating lists
fruits = ["apple", "banana", "cherry"]    # const fruits = ["apple", "banana", "cherry"]
empty = []                                 # const empty = []

# Accessing
first = fruits[0]          # fruits[0]
last = fruits[-1]          # fruits[fruits.length - 1]  ← Python has negative indexing!

# Slicing (Python superpower — no equivalent in JS)
fruits[1:3]                # ["banana", "cherry"]  — from index 1 up to (not including) 3
fruits[:2]                 # ["apple", "banana"]   — first 2
fruits[1:]                 # ["banana", "cherry"]  — from index 1 to end
fruits[-2:]                # ["banana", "cherry"]  — last 2

# Adding
fruits.append("date")              # fruits.push("date")
fruits.insert(0, "avocado")        # fruits.unshift("avocado")
fruits.extend(["fig", "grape"])    # fruits.push(...["fig", "grape"])

# Removing
fruits.remove("banana")           # const i = fruits.indexOf("banana"); fruits.splice(i,1)
fruits.pop()                       # fruits.pop()
fruits.pop(0)                      # fruits.shift()

# Checking
"apple" in fruits                  # fruits.includes("apple")
len(fruits)                        # fruits.length

# Iterating
for fruit in fruits:               # for (const fruit of fruits)
    print(fruit)

for i, fruit in enumerate(fruits): # fruits.forEach((fruit, i) => ...)
    print(f"{i}: {fruit}")

# JS: const doubled = numbers.map(n => n * 2)
doubled = [n * 2 for n in numbers]

# JS: const evens = numbers.filter(n => n % 2 === 0)
evens = [n for n in numbers if n % 2 == 0]

# JS: const results = numbers.filter(n => n > 5).map(n => n * 2)
results = [n * 2 for n in numbers if n > 5]

# Nested comprehension
# JS: matrix.flat().map(x => x * 2)
flat = [x * 2 for row in matrix for x in row]

# Dict comprehension
# JS: Object.fromEntries(items.map(item => [item.id, item]))
lookup = {item.id: item for item in items}

# Creating dicts
person = {
    "name": "Ajmal",
    "age": 28,
    "skills": ["Python", "React Native"],
}

# ⚠️ Keys MUST be strings (or other hashable types) — no unquoted keys like JS

# Accessing
person["name"]                    # person.name or person["name"]
person.get("name")                # Same, but returns None instead of KeyError
person.get("email", "N/A")       # Default value if key missing

# Setting
person["email"] = "a@b.com"      # person.email = "a@b.com"

# Checking
"name" in person                  # "name" in person (same!)

# Iterating
for key in person:                     # for (const key in person)
    print(key)

for key, value in person.items():      # Object.entries(person).forEach(([k,v]) => ...)
    print(f"{key}: {value}")

# Merging (Python 3.9+)
merged = {**dict1, **dict2}            # const merged = {...obj1, ...obj2}
merged = dict1 | dict2                 # Python 3.9+ pipe merge operator

# Destructuring
name = person["name"]                  # const { name } = person
# Python doesn't have object destructuring like JS
# But you can unpack dicts to function kwargs:
def greet(name, age, **kwargs):
    print(f"{name}, {age}")

greet(**person)                        # Unpacks dict as keyword arguments

# Tuples are like frozen arrays — can't modify after creation
point = (10, 20)
rgb = (255, 128, 0)

x, y = point           # Destructuring! → x=10, y=20

# Used for: returning multiple values, dict keys, immutable sequences
def get_user():
    return "Ajmal", 28  # Returns a tuple

name, age = get_user()  # Destructure the return value

# ⚠️ Single-element tuple needs trailing comma:
single = (42,)          # This is a tuple
not_tuple = (42)        # This is just the number 42

unique = {1, 2, 3}                     # const unique = new Set([1, 2, 3])
unique.add(4)                           # unique.add(4)
unique.discard(2)                       # unique.delete(2)

# Set operations (Python is way better than JS here)
a = {1, 2, 3}
b = {2, 3, 4}
a | b           # Union: {1, 2, 3, 4}
a & b           # Intersection: {2, 3}
a - b           # Difference: {1}
a ^ b           # Symmetric diff: {1, 4}

# Great for deduplication
unique_items = list(set(items_with_dupes))

# Basic function
def add(a: int, b: int) -> int:
    """Add two numbers."""          # Docstring (like JSDoc but built-in)
    return a + b

# Default parameters
def greet(name: str, greeting: str = "Hello") -> str:
    return f"{greeting}, {name}!"

# *args — like ...rest in JS
def sum_all(*numbers):              # function sumAll(...numbers)
    return sum(numbers)

sum_all(1, 2, 3, 4)  # 10

# **kwargs — named rest parameters (no JS equivalent)
def create_user(**kwargs):
    print(kwargs)  # {"name": "Ajmal", "age": 28}

create_user(name="Ajmal", age=28)

# Combining *args and **kwargs
def flexible(required, *args, **kwargs):
    print(f"Required: {required}")
    print(f"Extra positional: {args}")
    print(f"Extra keyword: {kwargs}")

flexible("hello", 1, 2, 3, debug=True, verbose=False)
# Required: hello
# Extra positional: (1, 2, 3)
# Extra keyword: {"debug": True, "verbose": False}

# JS: const double = (x) => x * 2
double = lambda x: x * 2

# JS: items.sort((a, b) => a.name.localeCompare(b.name))
items.sort(key=lambda x: x["name"])

# JS: items.filter(item => item.active)
active = list(filter(lambda item: item.active, items))

# ⚠️ Lambdas are single-expression only. For multi-line, use def.

# map, filter, reduce exist but list comprehensions are more Pythonic

# map
# JS:  numbers.map(n => n * 2)
# Py:  list(map(lambda n: n * 2, numbers))        ← works but ugly
# Py:  [n * 2 for n in numbers]                    ← Pythonic ✓

# filter
# JS:  numbers.filter(n => n > 5)
# Py:  list(filter(lambda n: n > 5, numbers))      ← works but ugly
# Py:  [n for n in numbers if n > 5]               ← Pythonic ✓

# reduce
from functools import reduce
# JS:  numbers.reduce((acc, n) => acc + n, 0)
total = reduce(lambda acc, n: acc + n, numbers, 0)
# Or just: total = sum(numbers)                    ← Pythonic ✓

# ─────────────────────────────────────────────────
# JS Class → Python Class
# ─────────────────────────────────────────────────

# JavaScript:
# class User {
#     constructor(name, email) {
#         this.name = name;
#         this.email = email;
#     }
#     greet() {
#         return `Hello, ${this.name}`;
#     }
# }

# Python:
class User:
    """A user in our system."""

    def __init__(self, name: str, email: str):   # constructor
        self.name = name                          # this.name = name
        self.email = email                        # this.email = email

    def greet(self) -> str:                       # methods always take 'self'
        return f"Hello, {self.name}"

    def __str__(self) -> str:                     # toString() equivalent
        return f"User({self.name}, {self.email})"

    def __repr__(self) -> str:                    # Developer-friendly string
        return f"User(name='{self.name}', email='{self.email}')"

user = User("Ajmal", "a@b.com")
print(user.greet())
print(user)         # Calls __str__

class Animal:
    def __init__(self, name: str):
        self.name = name

    def speak(self) -> str:
        raise NotImplementedError       # Abstract method

class Dog(Animal):                      # class Dog extends Animal
    def speak(self) -> str:
        return f"{self.name} says Woof!"

class Cat(Animal):
    def speak(self) -> str:
        return f"{self.name} says Meow!"

dog = Dog("Rex")
print(dog.speak())  # "Rex says Woof!"

from dataclasses import dataclass

# Instead of writing boilerplate __init__, __repr__, __eq__:
@dataclass
class Document:
    content: str
    source: str
    page: int = 0                    # Default value
    score: float = 0.0

# Auto-generates __init__, __repr__, __eq__
doc = Document(content="Hello", source="test.pdf", page=1)
print(doc)  # Document(content='Hello', source='test.pdf', page=1, score=0.0)

from pydantic import BaseModel, Field
from typing import Optional

class UserProfile(BaseModel):
    """User profile with validation."""
    name: str
    email: str
    age: int = Field(ge=0, le=150, description="User's age")
    bio: Optional[str] = None

# Valid — works fine
user = UserProfile(name="Ajmal", email="a@b.com", age=28)

# Invalid — raises ValidationError automatically!
try:
    bad = UserProfile(name="Ajmal", email="a@b.com", age=-5)
except Exception as e:
    print(e)  # age must be >= 0

# Parse from dict (like JSON deserialization)
data = {"name": "Ajmal", "email": "a@b.com", "age": 28}
user = UserProfile.model_validate(data)

# Parse from JSON string
json_str = '{"name": "Ajmal", "email": "a@b.com", "age": 28}'
user = UserProfile.model_validate_json(json_str)

# Convert to dict/JSON
user.model_dump()            # {"name": "Ajmal", "email": "a@b.com", ...}
user.model_dump_json()       # '{"name": "Ajmal", ...}'

# Get JSON schema (used by LLM structured output!)
UserProfile.model_json_schema()

from typing import (
    Optional,       # string | undefined
    Union,          # string | number
    Literal,        # "a" | "b" | "c"
    TypedDict,      # { name: string; age: number }
    Annotated,      # Used in LangGraph for state reducers
)

# Basic types
name: str = "Ajmal"
age: int = 28
score: float = 95.5
active: bool = True
items: list[str] = ["a", "b"]
lookup: dict[str, int] = {"a": 1}

# Optional (value or None)
email: Optional[str] = None          # string | undefined

# Union (multiple types)
id: Union[str, int] = "abc"          # string | number
# Python 3.10+: id: str | int = "abc"

# Literal (specific values only)
status: Literal["active", "inactive", "pending"] = "active"

# TypedDict (typed dictionaries — used in LangGraph state!)
class AgentState(TypedDict):
    messages: list
    query_type: str
    sources: list[str]

# Annotated (adds metadata to types — used in LangGraph!)
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # The add_messages is a "reducer"

# Function type hints
def process(data: list[str], limit: int = 10) -> dict[str, int]:
    ...

# Callable type (function as parameter)
from typing import Callable
def apply(func: Callable[[int], int], value: int) -> int:
    return func(value)

# ─────────────────────────────────────────────────
# Decorators = Higher-order functions that wrap other functions
# Like HOCs in React, but for functions
# ─────────────────────────────────────────────────

# Basic decorator
def log_calls(func):
    """Log every time a function is called."""
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}...")
        result = func(*args, **kwargs)
        print(f"Done: {result}")
        return result
    return wrapper

@log_calls                    # Same as: greet = log_calls(greet)
def greet(name: str) -> str:
    return f"Hello, {name}"

greet("Ajmal")
# Output:
# Calling greet...
# Done: Hello, Ajmal

# ─────────────────────────────────────────────────
# REAL-WORLD: LangChain @tool decorator
# ─────────────────────────────────────────────────
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression. Use for calculations."""
    return str(eval(expression))

# The @tool decorator:
# 1. Reads the function name → tool.name = "calculator"
# 2. Reads the docstring → tool.description = "Evaluate a math..."
# 3. Reads the type hints → tool.args = {"expression": {"type": "string"}}
# 4. Wraps the function so LangChain agents can call it

# ─────────────────────────────────────────────────
# Other decorators you'll see:
# ─────────────────────────────────────────────────
@property           # Getter (like a computed property in Vue)
@staticmethod       # No 'self' parameter (like static in JS)
@classmethod        # Takes 'cls' instead of 'self'
@dataclass          # Auto-generate __init__, __repr__, etc.
@pytest.fixture     # Test fixtures
@app.get("/")       # FastAPI route handlers

# Generators produce values lazily (one at a time)
# Critical for: streaming LLM tokens, processing large files

# Generator function (uses yield instead of return)
def count_up(n: int):
    """Generate numbers 0 to n-1."""
    i = 0
    while i < n:
        yield i      # Pauses here, returns i, resumes on next()
        i += 1

# Using a generator
for num in count_up(5):
    print(num)  # 0, 1, 2, 3, 4

# ─────────────────────────────────────────────────
# REAL-WORLD: Streaming LLM responses
# ─────────────────────────────────────────────────
# LangChain uses generators for streaming:

for chunk in llm.stream("Tell me a story"):
    print(chunk.content, end="", flush=True)

# Behind the scenes, this is a generator yielding tokens one by one

# Generator expression (like list comprehension, but lazy)
squares = (x**2 for x in range(1_000_000))  # No memory allocated for all values
first_10 = [next(squares) for _ in range(10)]

# Context managers handle setup/cleanup automatically
# Like try-finally, but cleaner

# ─────────────────────────────────────────────────
# File I/O (most common use)
# ─────────────────────────────────────────────────

# ❌ Without context manager:
f = open("data.txt", "r")
content = f.read()
f.close()                  # Easy to forget!

# ✅ With context manager:
with open("data.txt", "r") as f:
    content = f.read()
# File is automatically closed, even if an exception occurs

# Writing files
with open("output.txt", "w") as f:
    f.write("Hello, World!\n")

# Reading JSON
import json
with open("config.json", "r") as f:
    config = json.load(f)

# Writing JSON
with open("output.json", "w") as f:
    json.dump(data, f, indent=2)

# ─────────────────────────────────────────────────
# You'll see context managers with:
# - Database connections
# - HTTP sessions
# - Temporary files
# - Lock management
# ─────────────────────────────────────────────────

# ─────────────────────────────────────────────────
# try/except (= try/catch in JS)
# ─────────────────────────────────────────────────

try:
    result = 10 / 0
except ZeroDivisionError:                    # catch specific error
    print("Can't divide by zero!")
except (ValueError, TypeError) as e:         # catch multiple
    print(f"Error: {e}")
except Exception as e:                       # catch-all (like catch(e) in JS)
    print(f"Unexpected error: {e}")
else:                                        # Runs if NO exception (no JS equivalent)
    print(f"Result: {result}")
finally:                                     # Always runs (same as JS)
    print("Done")

# ─────────────────────────────────────────────────
# Raising exceptions (= throw in JS)
# ─────────────────────────────────────────────────

def validate_age(age: int):
    if age < 0:
        raise ValueError(f"Age cannot be negative: {age}")
    return age

# Custom exceptions
class DocumentNotFoundError(Exception):
    """Raised when a document cannot be found in the knowledge base."""
    pass

# Real-world: LLM API error handling
from openai import RateLimitError, APIError
import time

def safe_llm_call(prompt: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
            )
            return response.choices[0].message.content
        except RateLimitError:
            wait = 2 ** attempt
            print(f"Rate limited, waiting {wait}s...")
            time.sleep(wait)
        except APIError as e:
            if attempt == max_retries - 1:
                raise
            print(f"API error: {e}, retrying...")
    raise Exception("Max retries exceeded")

import asyncio

# ─────────────────────────────────────────────────
# Python async is very similar to JS async/await
# ─────────────────────────────────────────────────

# JS:
# async function fetchData(url) {
#     const response = await fetch(url);
#     return await response.json();
# }

# Python:
async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# Running async code
async def main():
    result = await fetch_data("https://api.example.com/data")
    print(result)

asyncio.run(main())

# ─────────────────────────────────────────────────
# Concurrent execution (like Promise.all)
# ─────────────────────────────────────────────────

async def process_batch():
    # JS: await Promise.all([task1(), task2(), task3()])
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3"),
    )
    return results

# ─────────────────────────────────────────────────
# LangChain async (used for parallel LLM calls)
# ─────────────────────────────────────────────────

# Most LangChain classes have async versions:
result = await llm.ainvoke(messages)          # Async invoke
results = await chain.abatch([input1, input2]) # Batch process
async for chunk in chain.astream(input):       # Async streaming
    print(chunk)

# ─────────────────────────────────────────────────
# Python imports vs JS imports
# ─────────────────────────────────────────────────

# JS: import { useState, useEffect } from 'react'
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# JS: import React from 'react'
import langchain

# JS: import * as utils from './utils'
from langchain import agents  # Access: agents.create_react_agent(...)

# JS: import { default as MyComponent } from './Component'
from langchain_openai import ChatOpenAI as LLM   # Alias

# Relative imports (within your project)
from .agents import orchestrator        # from ./agents import ...
from ..config import settings           # from ../config import ...

# ─────────────────────────────────────────────────
# __init__.py = index.js
# ─────────────────────────────────────────────────
# A folder with __init__.py is a Python "package" (like having index.js)
# src/
# ├── __init__.py          ← Makes 'src' importable
# ├── agents/
# │   ├── __init__.py      ← Makes 'agents' importable
# │   ├── rag_agent.py
# │   └── search_agent.py
# └── config.py

# ─────────────────────────────────────────────────
# if __name__ == "__main__": (entry point guard)
# ─────────────────────────────────────────────────
# This runs ONLY when the file is executed directly (not imported)

def main():
    print("Running as main script")

if __name__ == "__main__":
    main()
# Like: if (require.main === module) { main() }

# Create virtual environment
python -m venv .venv

# Activate (macOS/Linux)
source .venv/bin/activate

# Activate (Windows)
.venv\Scripts\activate

# Deactivate
deactivate

# ⚠️ ALWAYS activate venv before pip install
# Otherwise packages install globally (bad!)

# npm install langchain    →    pip install langchain
# npm install              →    pip install -r requirements.txt
# npm install --save       →    pip install langchain && pip freeze > requirements.txt
# package.json             →    requirements.txt (or pyproject.toml)
# node_modules/            →    .venv/lib/

# Install specific version
pip install langchain==0.2.0

# Install from requirements.txt
pip install -r requirements.txt

# See installed packages
pip list

# Generate requirements.txt from current environment
pip freeze > requirements.txt

# Install: pip install python-dotenv

# .env file:
# OPENAI_API_KEY=sk-your-key-here
# QDRANT_URL=http://localhost:6333

# Load in Python:
from dotenv import load_dotenv
import os

load_dotenv()  # Reads .env file into environment variables

api_key = os.getenv("OPENAI_API_KEY")
qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333")  # With default

# LangChain Expression Language uses | (pipe) like Unix pipes
# Similar to RxJS pipe() or lodash chain()

chain = prompt | llm | output_parser

# This means: prompt.invoke() → pass result to llm.invoke() → pass to parser
# Like: prompt.pipe(llm).pipe(output_parser)

from typing import TypedDict, Annotated

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # Reducer: append new messages
    query_type: str                           # Replace on update

# This is like defining a Redux state shape with reducers

# Functions that create and return configured objects
def create_rag_chain(model: str = "gpt-4o", k: int = 4):
    llm = ChatOpenAI(model=model, temperature=0)
    retriever = vector_store.as_retriever(search_kwargs={"k": k})

    return (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )

chain = create_rag_chain(model="gpt-4o-mini", k=6)

# LangChain uses callbacks for logging, streaming, etc.
from langchain_core.callbacks import BaseCallbackHandler

class MyHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM starting with {len(prompts)} prompts")

    def on_llm_end(self, response, **kwargs):
        print(f"LLM finished")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"Tool called: {serialized['name']}")

# Attach to LLM
llm = ChatOpenAI(model="gpt-4o", callbacks=[MyHandler()])

# ─────────────────────────────────────────────────
# Modules you'll use constantly in GenAI work
# ─────────────────────────────────────────────────

import os                 # Environment vars, file paths
import json               # Parse/generate JSON
import time               # Sleep, timing
import logging            # Production logging
import hashlib            # Hashing (for caching)
from pathlib import Path  # Modern file path handling
from typing import *      # Type hints
from datetime import datetime  # Timestamps
from collections import defaultdict, Counter  # Useful data structures
from functools import reduce, lru_cache       # Functional utils + caching
from dataclasses import dataclass, field      # Data classes
import re                 # Regex
import uuid               # Unique IDs (for thread_ids, collection names)
import asyncio            # Async programming

# ─────────────────────────────────────────────────
# pathlib.Path (modern file handling)
# ─────────────────────────────────────────────────
from pathlib import Path

# Old way: os.path.join("data", "docs", "file.pdf")
# New way:
path = Path("data") / "docs" / "file.pdf"

path.exists()         # Does it exist?
path.is_file()        # Is it a file?
path.is_dir()         # Is it a directory?
path.suffix           # ".pdf"
path.stem             # "file"
path.parent           # Path("data/docs")
path.read_text()      # Read file contents as string
path.write_text(data) # Write string to file

# List all PDFs in a directory
pdfs = list(Path("docs").glob("**/*.pdf"))