祝融说。

FORM NOT VOID, MIND NO CORE

Chapter 1: Python Programming Essentials for AI Engineers

2026.08.10

Welcome to the first chapter of this book. Here, we will temporarily set aside the dazzling neural network models and complex mathematical formulas, and return to where it all begins -- our most powerful tool: Python.

For an AI engineer, Python is more than just a programming language; it is our "scalpel," "paintbrush," and "Swiss Army knife." Whether processing terabytes of data, constructing intricate algorithm models, or deploying high-concurrency inference services, Python plays an indispensable role. Yet many beginners -- and even some experienced developers -- only understand Python at the "can use it" level: they can write loops, define functions, and call libraries. This may suffice for small-scale academic experiments or personal projects, but once you step onto the real battlefield of industry, this "surface-level familiarity" reveals its fragility.

An industrial-grade AI project is a complex system engineering endeavor. It demands code that not only runs correctly, but also possesses strong readability, maintainability, extensibility, and performance. Between a simple script and a robust AI application lies a chasm formed by programming paradigms, coding standards, and engineering practices. Crossing this chasm is the core mission of this chapter.

We call this section the "Foundation of Inner Strength," because like the internal energy cultivation depicted in martial arts novels, it does not manifest in visible techniques, yet it determines the ultimate height you can reach. A martial artist with profound inner strength can unleash immense power from even the most ordinary moves. Similarly, an AI engineer with a solid Python foundation can write elegant, efficient, and reliable code when facing complex problems, handling various AI frameworks and tools with ease.

In this chapter, we will explore the deeper charms of the Python language, learning the advanced features, programming paradigms, and engineering standards that can bring a qualitative leap to your code quality. Through numerous examples closely tied to AI scenarios, you will truly come to understand that mastering Python is not merely "icing on the cake," but a "necessary condition" for becoming an outstanding AI engineer.

Now, let us set aside distractions, calm our minds, and embark together on this journey of cultivating Python inner strength.

1.1 Beyond the Basics: Python Advanced Features (Decorators, Generators, Coroutines)

Mastering the basic syntax of Python is like learning the alphabet. To write beautiful poetry, we still need to learn grammatical structures and rhetorical devices. Decorators, generators, and coroutines are the "advanced rhetoric" of the Python language that can greatly enhance code expressiveness and efficiency.

1.1.1 Decorators: Elegantly Empowering Functions

What and Why?

Imagine in your AI project, you have multiple functions that all need to print logs before and after execution, or need to measure their running time. What is the most intuitive approach? Manually adding logging and timing code inside each function.

import time

def preprocess_data_a(data):
    print("Starting data preprocessing A...")
    start_time = time.time()
    # ... core processing logic ...
    time.sleep(1) # Simulate a time-consuming operation
    end_time = time.time()
    print(f"Data preprocessing A completed, time elapsed: {end_time - start_time:.2f} seconds.")
    return "processed_a"

def train_model_b(config):
    print("Starting model training B...")
    start_time = time.time()
    # ... core training logic ...
    time.sleep(2) # Simulate a time-consuming operation
    end_time = time.time()
    print(f"Model training B completed, time elapsed: {end_time - start_time:.2f} seconds.")
    return "trained_b"

preprocess_data_a("raw_data")
train_model_b({"lr": 0.01})

This code has obvious problems:

  1. Code redundancy: The logging and timing code is repeated in every function.
  2. Violation of the Single Responsibility Principle: The core responsibility of preprocess_data_a is data processing, but it also takes on the duties of timing and logging, making the code logic messy.
  3. Difficult to maintain: If you want to change the log format, you need to modify all related functions.

Decorators were born to solve this kind of problem. In essence, a decorator is a higher-order function -- a function that takes a function as an argument and returns a new function. It allows us to add extra functionality to a function without modifying its original code.

How to Implement It?

Let us build a generic timing decorator called timer.

import time
import functools

def timer(func):
    """A decorator that prints the execution time of a function"""
    @functools.wraps(func) # Critical step, preserves the original function's metadata
    def wrapper(*args, **kwargs):
        print(f"Starting function '{func.__name__}'...")
        start_time = time.time()
      
        # Execute the original function
        result = func(*args, **kwargs)
      
        end_time = time.time()
        print(f"Function '{func.__name__}' completed, time elapsed: {end_time - start_time:.2f} seconds.")
        return result
    return wrapper

Let us parse this code line by line:

  1. def timer(func):: Defines the decorator function timer, which takes a function func as its argument.
  2. def wrapper(*args, **kwargs):: Inside timer, we define a new function wrapper. This function will replace our original function. Using *args and **kwargs allows wrapper to accept any number of positional and keyword arguments, making our decorator generic.
  3. result = func(*args, **kwargs): Inside wrapper, we call the original function func and save its return value.
  4. return wrapper: The timer function finally returns the wrapper function.
  5. @functools.wraps(func): This is a "decorator of decorators." It copies certain metadata from the original function func (such as the function name __name__, docstring __doc__, etc.) to the wrapper function. Without this line, all our decorated functions would have the name wrapper, which would cause confusion during debugging.

Now, we can use it in an extremely elegant way:

@timer
def preprocess_data_a(data):
    """A data preprocessing function"""
    # ... core processing logic ...
    time.sleep(1)
    return "processed_a"

@timer
def train_model_b(config):
    """A model training function"""
    # ... core training logic ...
    time.sleep(2)
    return "trained_b"

preprocess_data_a("raw_data")
train_model_b({"lr": 0.01})

print(f"Function A name: {preprocess_data_a.__name__}")
print(f"Function A docstring: {preprocess_data_a.__doc__}")

Output:

Starting function 'preprocess_data_a'...
Function 'preprocess_data_a' completed, time elapsed: 1.00 seconds.
Starting function 'train_model_b'...
Function 'train_model_b' completed, time elapsed: 2.00 seconds.
Function A name: preprocess_data_a
Function A docstring: A data preprocessing function

The @timer syntax is called "syntactic sugar." It is equivalent to preprocess_data_a = timer(preprocess_data_a). It clearly expresses our intent: to attach the timer functionality to the preprocess_data_a function.

Application Scenarios in AI Projects:

  • Logging (@log_step): Record the input, output, and status of each step in a data processing pipeline.
  • Performance Monitoring (@timer, @profile_memory): Analyze the time and memory consumption of critical stages such as model inference and data loading.
  • Result Caching (@functools.lru_cache): For computationally expensive functions where the same input always yields the same output (such as feature extraction), caching avoids redundant computation and greatly improves efficiency.
  • Permission Verification (@require_auth): On API endpoints of model services, verify whether the user has permission to call the endpoint.

1.1.2 Generators: Efficiently Processing Massive Datasets

What and Why?

In the AI field, we often need to process datasets that far exceed memory capacity. For example, a dataset containing millions of high-resolution images, or a terabyte-scale text corpus. If we try to read all the data into memory at once, the result is a single MemoryError.

Generators are the perfect tool for this problem. A generator is a special type of iterator. Unlike a regular function that returns all results at once, a generator uses the yield keyword to return one result at a time, "pausing" in place, and resuming from where it left off when called again.

Let us illustrate its power with an example. Suppose we want to process a range of numbers and compute the square of each.

Traditional approach (returns a list):

def square_numbers_list(n):
    result = []
    for i in range(n):
        result.append(i * i)
    return result

# When n is very large, say 100 million, this consumes a huge amount of memory
# my_squares = square_numbers_list(100_000_000)

The above code first generates a huge list containing 100 million elements, then returns it.

Generator approach:

def square_numbers_generator(n):
    for i in range(n):
        yield i * i

# Create a generator object, consuming almost no memory
my_squares_gen = square_numbers_generator(100_000_000)

# Values are computed and generated one at a time only when iterating
for i, num in enumerate(my_squares_gen):
    if i < 5:
        print(num)
    else:
        break

Output:

0
1
4
9
16

When square_numbers_generator is called, it does not execute immediately. Instead, it returns a generator object. The code only truly runs when we iterate over it in a for loop. Each time it encounters yield, the function produces a value and pauses until the next next() call (the for loop does this automatically). This "lazy evaluation" characteristic gives generators unparalleled memory efficiency when processing large data streams.

Furthermore, Python provides an even more concise generator expression, syntactically similar to a list comprehension but using parentheses ():

# List comprehension, creates the list immediately
list_comp = [i * i for i in range(10)] 
# Generator expression, returns a generator object
gen_exp = (i * i for i in range(10))

Application Scenarios in AI Projects:

Building a Data Pipeline: This is the most core application of generators in AI. When training a deep learning model, we need a pipeline that can continuously and efficiently supply data batches. We can write a generator that reads files from disk, performs preprocessing (such as tokenization or image augmentation), and then yields the processed batches one by one. The core concepts behind mainstream deep learning frameworks like PyTorch's DataLoader and TensorFlow's tf.data are aligned with the generator philosophy.

def text_data_generator(file_path, batch_size):
    """A generator that reads data from a large text file and yields batches"""
    with open(file_path, 'r', encoding='utf-8') as f:
        batch = []
        for line in f:
            processed_line = line.strip().lower() # Simple preprocessing
            batch.append(processed_line)
            if len(batch) == batch_size:
                yield batch
                batch = []
        if batch: # Handle the last batch that might be smaller than batch_size
            yield batch

# Usage example
# for data_batch in text_data_generator('huge_corpus.txt', 32):
#     model.train_on_batch(data_batch)

Stream Processing: When dealing with data coming from a network, sensors, or other continuously generating sources, generators can handle such infinite data streams elegantly.

1.1.3 Coroutines with asyncio: Handling High-Concurrency I/O

What and Why?

Modern AI applications, especially LLM-based agents, are often not isolated computational units. They need extensive interaction with the outside world: calling multiple different APIs for information, querying databases, reading and writing files, and so on. These operations are mostly I/O-bound, meaning the program spends most of its time waiting for the network or disk to respond, while the CPU sits idle.

The traditional synchronous programming model can only do one thing at a time. After initiating a network request, the program "blocks" until it receives the response, wasting CPU resources. Multithreading is one solution, but thread creation and switching have overhead, and due to Python's Global Interpreter Lock (GIL), true parallelism cannot be achieved for CPU-intensive tasks.

Coroutines, in conjunction with the asyncio library, provide a more efficient concurrency model called asynchronous programming. The core idea is: when a task (a coroutine) encounters an I/O wait, it voluntarily "suspends" itself, yielding control of the CPU. The event loop immediately switches to another ready task. This way, the CPU is always processing computational tasks rather than idly waiting, achieving high concurrency within a single thread.

How to Implement It?

The core concepts of asyncio include:

  • async def: Used to define a coroutine function. Calling it does not execute it immediately but returns a coroutine object.
  • await: Used to "suspend" a coroutine and wait for it to complete. It can only be used inside an async def function.
  • asyncio.run(): Starts the event loop and runs the top-level async function.

Let us look at an example: simulating a scenario where two APIs need to be called simultaneously.

Synchronous version:

import time

def fetch_api_a():
    print("Starting API A request...")
    time.sleep(2) # Simulate network latency
    print("API A responded.")
    return "Result A"

def fetch_api_b():
    print("Starting API B request...")
    time.sleep(1) # Simulate network latency
    print("API B responded.")
    return "Result B"

def main_sync():
    start = time.time()
    result_a = fetch_api_a()
    result_b = fetch_api_b()
    end = time.time()
    print(f"Total synchronous execution time: {end - start:.2f} seconds")

main_sync()

Output:

Starting API A request...
API A responded.
Starting API B request...
API B responded.
Total synchronous execution time: 3.01 seconds

The total time is the sum of the two tasks' durations.

Asynchronous version:

import asyncio
import time

async def fetch_api_a_async():
    print("Starting API A request...")
    await asyncio.sleep(2) # Simulate asynchronous I/O operation
    print("API A responded.")
    return "Result A"

async def fetch_api_b_async():
    print("Starting API B request...")
    await asyncio.sleep(1) # Simulate asynchronous I/O operation
    print("API B responded.")
    return "Result B"

async def main_async():
    start = time.time()
    # Create two tasks and run them concurrently
    task_a = asyncio.create_task(fetch_api_a_async())
    task_b = asyncio.create_task(fetch_api_b_async())
  
    # Wait for both tasks to complete
    result_a = await task_a
    result_b = await task_b
  
    end = time.time()
    print(f"Total asynchronous execution time: {end - start:.2f} seconds")

asyncio.run(main_async())

Output:

Starting API A request...
Starting API B request...
API B responded.
API A responded.
Total asynchronous execution time: 2.01 seconds

The total time is only as long as the slowest task! This is the power of asynchronous programming.

Application Scenarios in AI Projects:

Tool Calls for LLM Agents: A complex agent may need to query a weather API, a stock API, and an internal knowledge base simultaneously, then aggregate the information to generate a report. Using asyncio to run these parallel I/O operations concurrently can greatly reduce the agent's response time.

High-Concurrency Model Inference Service: When building a model API with Flask or FastAPI, if the model inference itself is fast but the request involves I/O operations such as database queries, using asynchronous view functions can significantly improve service throughput.

Distributed Data Crawling and Processing: When preparing data for a model, you may need to crawl information from multiple websites. An asynchronous crawler is far more efficient than a synchronous one.

1.2 Object-Oriented and Functional Programming Paradigms in AI Projects

A programming paradigm is a way of thinking about and organizing code. Python is a multi-paradigm language, supporting both Object-Oriented Programming (OOP) and Functional Programming (FP). In AI projects, these two paradigms are not mutually exclusive but complement each other, each excelling in different scenarios.

1.2.1 Object-Oriented Programming (OOP): Building Structured AI Systems

The core idea of OOP is to encapsulate data (attributes) and the behaviors that operate on that data (methods) within "objects." A class is a blueprint for creating objects. The three pillars are encapsulation, inheritance, and polymorphism.

Expression in AI Projects:

Virtually all mainstream AI/ML libraries are built on OOP, and this is no coincidence.

  1. Encapsulation: Hiding complexity.

    Scenario: A nn.Linear layer (fully connected layer) in PyTorch. When we use it, we only need to care about its input and output dimensions, without worrying about the internal details of weight matrix creation, initialization, forward propagation, and the specific mathematical operations of backpropagation. These complex details are encapsulated within the Linear class.

    Our own application: In building a complete AI application, we can encapsulate data loading and preprocessing logic into a DataLoader class, the model structure into a MyModel class, and the training loop into a Trainer class. This makes the entire project structure clear and responsibilities well-defined.

  2. Inheritance: Achieving code reuse and extension.

    Scenario: In PyTorch, we always define our own neural network models by inheriting from the torch.nn.Module class.

    import torch.nn as nn

class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__() # Call the parent class constructor
        self.conv1 = nn.Conv2d(1, 20, 5)
        self.relu = nn.ReLU()
        # ... other layers

    def forward(self, x):
        # ... define forward propagation logic
        return x

    Through inheritance, our SimpleCNN class automatically gains all the functionality provided by nn.Module, such as parameter management (.parameters()), device transfer (.to(device)), and mode switching (.train(), .eval()). We only need to focus on defining the model's structure and forward propagation logic.

    Our own application: We could define a generic BaseExperiment class that contains common logic for experiment logging, result saving, and so on, and then have specific experiments (like ResNet50Experiment, BERTExperiment) inherit from it.

  3. Polymorphism: Providing a unified interface.

    Scenario: Scikit-learn is a paragon of polymorphism. Whether it is LogisticRegression, SVC, or RandomForestClassifier, they all follow the unified "Estimator" interface, with .fit(X, y) and .predict(X) methods. This allows us to easily swap and compare different models without changing the rest of the workflow code.

    from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier

models = [LogisticRegression(), RandomForestClassifier()]

for model in models:
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)
    print(f"Model {type(model).__name__} accuracy: ...")

    Our own application: When designing a data augmentation strategy, we can define an abstract base class BaseAugmentation with an apply(image) method. Then, specific augmentation classes like RandomCrop, Flip, and Rotate all inherit from it and implement the apply method. This way, we can put a series of augmentation operations in a list and call them in a unified manner.

1.2.2 Functional Programming (FP): Crafting Clear Data Flows

The core idea of FP is to treat computation as the evaluation of mathematical functions, emphasizing the use of pure functions, avoiding side effects, and avoiding mutable data.

Pure functions: Given the same input, they always produce the same output and do not modify any external state.

Side effects: Modifying state outside the function, such as changing global variables, printing to the console, or writing to files.

Expression in AI Projects:

Data preprocessing and feature engineering are areas where FP ideas shine. A typical data processing pipeline is essentially a sequential application of functions, transforming raw data step by step into tensors that the model can accept.

Traditional imperative style:

def process_texts(texts):
    processed = []
    for text in texts:
        # 1. Convert to lowercase
        text = text.lower()
        # 2. Remove punctuation
        import string
        text = text.translate(str.maketrans('', '', string.punctuation))
        # 3. Tokenize
        tokens = text.split()
        processed.append(tokens)
    return processed

This code works, but it mixes control flow (for loop) with data transformation logic.

Functional style:

import string
from functools import reduce

def to_lower(text):
    return text.lower()

def remove_punctuation(text):
    return text.translate(str.maketrans('', '', string.punctuation))

def tokenize(text):
    return text.split()

def process_texts_functional(texts):
    # Using map and lambda expressions
    # return list(map(lambda t: tokenize(remove_punctuation(to_lower(t))), texts))

    # Or build a processing pipeline
    pipeline = [to_lower, remove_punctuation, tokenize]
  
    processed_texts = []
    for text in texts:
        # Using reduce to chain functions applied to the text
        result = reduce(lambda val, func: func(val), pipeline, text)
        processed_texts.append(result)
    return processed_texts

Advantages of the functional style:

  1. Modularity and Testability: to_lower, remove_punctuation, etc. are all pure functions. They are small, independent, and extremely easy to unit test.
  2. Clarity and Readability: Each step of data processing is defined by an independent function, and the entire flow resembles a clear assembly line, easy to understand at a glance.
  3. Composability: We can easily reorder, add, or remove processing steps by simply adjusting the pipeline list, without modifying the core logic.

In Python, we typically use map, filter, and list/generator comprehensions to practice FP ideas, as they are more "Pythonic" and readable than reduce.

# Using a list comprehension combined with function calls, clear and efficient
def process_texts_pythonic(texts):
    pipeline = [to_lower, remove_punctuation, tokenize]
  
    def apply_pipeline(text):
        for func in pipeline:
            text = func(text)
        return text
      
    return [apply_pipeline(text) for text in texts]

Summary: OOP vs FP

  • Use OOP to build the project's skeleton: define key components like data structures (e.g., Dataset), models (Model), and workflows (Trainer).
  • Use FP to fill in the project's flesh: within components, especially in data processing and transformation logic, use functional thinking to build clear, testable, and composable data flows.

1.3 Code Standards and Engineering Practices: PEP 8, Type Hints, and Project Structure

If advanced features and programming paradigms are the means to improve code "intelligence," then code standards and engineering practices are the keys to improving code "emotional intelligence" and "physique." They determine whether your code is easy to collaborate on, easy to maintain, and whether the project can grow healthily.

1.3.1 PEP 8: The "Mandarin" of Python Code

PEP 8 (Python Enhancement Proposal 8) is the official Python style guide. It specifies details such as indentation, line length, naming conventions, and comment style.

Why is it important?

Code is read far more often than it is written. Following a consistent style guide allows anyone on the team to quickly understand your code, just like speaking "Mandarin" for barrier-free communication. This greatly reduces maintenance costs.

Key points:

  • Indentation: Use 4 spaces, not tabs.
  • Line Length: Maximum 79 characters per line. This improves readability on small screens or when comparing code side by side.
  • Naming:
    • snake_case: for functions, methods, variables, and modules. e.g., def calculate_loss().
    • PascalCase: for class names. e.g., class TextClassifier().
    • UPPERCASE_SNAKE_CASE: for constants. e.g., LEARNING_RATE = 0.001.
  • Blank Lines: Top-level function and class definitions should be separated by two blank lines; method definitions within a class should be separated by one blank line.
  • Imports: Import statements should always be at the top of the file and grouped in the following order: standard library, third-party libraries, local application imports.

Automated tools:

Manually following PEP 8 is tedious and error-prone. Professional engineers use automated tools to enforce code style:

  • flake8: A code style checker that reports violations of PEP 8.
  • black: An "uncompromising" code formatter that forces your code into a unified style conforming to a subset of PEP 8.
  • isort: Automatically sorts and groups your import statements.

Integrating these tools into your project (e.g., through Git pre-commit hooks) ensures highly consistent code style across the entire team.

1.3.2 Type Hints: Injecting Static Rigor into a Dynamic Language

Python is a dynamically typed language, meaning you do not need to declare variable types when writing code. This offers flexibility but also introduces risks: you can easily pass incorrectly typed data to a function, and such errors only surface at runtime.

Type hints (introduced in Python 3.5) allow us to add type annotations to function parameters and return values.

# Without type hints
def add(a, b):
    return a + b

# With type hints
def add_typed(a: int, b: int) -> int:
    return a + b

Why are type hints important?

  1. Static Error Checking: Using static analysis tools like mypy, type mismatch errors can be caught before the code even runs. This is crucial in large, complex AI projects, catching many bugs in their infancy.
  2. Code Readability and Documentation: The function signature def process_data(df: pd.DataFrame) -> np.ndarray: tells us at a glance that this function takes a Pandas DataFrame and returns a NumPy array. It serves as the best form of documentation.
  3. IDE Support: Code with type hints allows IDEs like VS Code and PyCharm to provide smarter autocompletion, code navigation, and error hints.

Application in AI Projects:

Data structures in AI projects are often very complex. Type hints can greatly improve code robustness.

from typing import List, Dict, Tuple
import pandas as pd
import numpy as np

def preprocess(
    texts: List[str], 
    config: Dict[str, any]
) -> Tuple[np.ndarray, Dict[str, int]]:
    # ... implementation ...
    # Return the processed feature matrix and vocabulary
    features = np.array([[1,2], [3,4]])
    vocab = {"a": 0, "b": 1}
    return features, vocab

This function signature clearly defines its "contract." Anyone using it knows exactly how to call it and what to expect.

1.3.3 Project Structure: Building a Scalable Engineering Skeleton

When a project consists of just a single .py file, everything is simple. But a real AI project contains many components: data, configuration files, Jupyter Notebook experiments, source code, test cases, and more. A well-organized, consistent project structure is the guarantee of a project's long-term healthy development.

Recommended generic project structure:

my_awesome_ai_project/
├── data/                     # Store all data
│   ├── raw/                  # Original, immutable data
│   └── processed/            # Preprocessed data, ready for model training
│
├── notebooks/                # Jupyter Notebooks for exploratory analysis and experimentation
│   ├── 01_data_exploration.ipynb
│   └── 02_model_prototyping.ipynb
│
├── src/                      # Core source code (or named after the project)
│   ├── __init__.py           # Makes src a Python package
│   ├── data_processing.py    # Data loading and preprocessing module
│   ├── modeling.py           # Model definition module
│   ├── training.py           # Training logic module
│   └── utils.py              # General utility functions
│
├── scripts/                  # Store executable scripts
│   ├── train.py              # Script to start model training
│   └── predict.py            # Script to make predictions using a trained model
│
├── tests/                    # Store test code
│   ├── test_data_processing.py
│   └── test_utils.py
│
├── config/                   # Store configuration files
│   └── main_config.yaml
│
├── saved_models/             # Store trained model files and results
│
├── .gitignore                # Git ignore file configuration
├── requirements.txt          # Project dependency list
└── README.md                 # Project documentation

Why organize it this way?

  • Separation of Concerns: Code, data, experiments, and configuration each have their own place, clear and well-organized.
  • Reproducibility: requirements.txt ensures environment consistency; config/ makes experiment parameters configurable and traceable.
  • Modularity and Importability: Placing core logic under src/ and making it a package allows code in scripts/ and notebooks/ to easily import and reuse via from src.modeling import MyModel, avoiding messy relative path issues.
  • Ease of Collaboration: New members can quickly understand the project overview based on this standard structure and find where to modify or add code.

1.4 Hands-On Project: Building a Reusable Data Processing Utility Class

Now, let us integrate all the knowledge points learned in this chapter -- OOP, type hints, PEP 8 standards, and decorators -- to build a reusable data processing utility class that is extremely common in AI projects.

Project Objective:

Create a DataProcessor class for processing structured data (such as CSV files). It should be able to load data, handle missing values, normalize numerical features, encode categorical features, and so on, with the entire process being configurable and capable of logging.

Project Structure:

We will follow the project structure defined in the previous section.

data_toolkit/
├── src/
│   ├── __init__.py
│   └── processor.py
├── scripts/
│   └── run_processing.py
├── data/
│   └── sample_data.csv
└── README.md

Step 1: Create utility functions and decorators (src/utils.py - if needed)

To keep processor.py's core logic clean, we could place the decorator in a separate utils.py file, but for simplicity in this example, we will define it inside processor.py.

Step 2: Write the DataProcessor class (src/processor.py)

# src/processor.py

import time
import functools
from typing import List, Optional, Dict, Any

import pandas as pd
from sklearn.preprocessing import StandardScaler, OneHotEncoder


# # 1. Using decorators (from 1.1)

def log_step(func):
    """A decorator that logs data processing steps and their duration"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # args[0] is 'self'
        class_name = args[0].__class__.__name__
        print(f"[{class_name}] ==> Starting step: {func.__name__}...")
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"[{class_name}] <== Step '{func.__name__}' completed, time elapsed: {end_time - start_time:.4f} seconds.")
        return result
    return wrapper

# 2. Using OOP and type hints (from 1.2 and 1.3)
class DataProcessor:
    """A reusable utility class for processing structured data"""

    def __init__(self, dataframe: pd.DataFrame):
        # Using type hints
        if not isinstance(dataframe, pd.DataFrame):
            raise TypeError("Input must be a pandas DataFrame")
        self.df = dataframe.copy()
        self.scalers: Dict[str, StandardScaler] = {}
        self.encoders: Dict[str, OneHotEncoder] = {}
        print("DataProcessor initialized successfully.")

    @log_step
    def handle_missing_values(
        self, 
        strategy: str = 'mean', 
        columns: Optional[List[str]] = None
    ) -> 'DataProcessor':
        """Handle missing values in specified columns"""
        target_cols = columns if columns else self.df.select_dtypes(include='number').columns
        for col in target_cols:
            if self.df[col].isnull().sum() > 0:
                if strategy == 'mean':
                    fill_value = self.df[col].mean()
                elif strategy == 'median':
                    fill_value = self.df[col].median()
                elif strategy == 'mode':
                    fill_value = self.df[col].mode()[0]
                else:
                    fill_value = 0
                self.df[col].fillna(fill_value, inplace=True)
        return self

    @log_step
    def scale_numerical_features(
        self, 
        columns: Optional[List[str]] = None
    ) -> 'DataProcessor':
        """Standardize specified numerical features"""
        target_cols = columns if columns else self.df.select_dtypes(include='number').columns
        for col in target_cols:
            scaler = StandardScaler()
            self.df[col] = scaler.fit_transform(self.df[[col]])
            self.scalers[col] = scaler # Save the scaler for later inverse transformation or use on new data
        return self

    @log_step
    def encode_categorical_features(
        self, 
        columns: Optional[List[str]] = None
    ) -> 'DataProcessor':
        """One-hot encode specified categorical features"""
        target_cols = columns if columns else self.df.select_dtypes(include=['object', 'category']).columns
        for col in target_cols:
            encoder = OneHotEncoder(handle_unknown='ignore', sparse_output=False)
            encoded_data = encoder.fit_transform(self.df[[col]])
            encoded_df = pd.DataFrame(encoded_data, columns=encoder.get_feature_names_out([col]))
          
            # Merge back into the original DataFrame and drop the original column
            self.df = self.df.drop(col, axis=1)
            self.df = pd.concat([self.df, encoded_df], axis=1)
            self.encoders[col] = encoder # Save the encoder
        return self

    def get_processed_data(self) -> pd.DataFrame:
        """Get the processed DataFrame"""
        return self.df

# 3. Code follows PEP 8 standards

Step 3: Create sample data (data/sample_data.csv)

age,salary,city,purchased
25,50000,New York,0
30,,London,1
35,60000,Tokyo,0
,75000,New York,1
40,80000,London,0
22,45000,,1

Step 4: Write the execution script (scripts/run_processing.py)

# scripts/run_processing.py

import pandas as pd
import sys
import os

# Ensure we can import modules from src
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from src.processor import DataProcessor

def main():
    # Load data
    data_path = os.path.join(os.path.dirname(__file__), '..', 'data', 'sample_data.csv')
    raw_df = pd.read_csv(data_path)
    print("Original data:")
    print(raw_df)
    print("-" * 30)

    # Use our DataProcessor class with chained calls
    processor = DataProcessor(raw_df)
  
    processed_df = (processor
                    .handle_missing_values(strategy='mean', columns=['age', 'salary'])
                    .handle_missing_values(strategy='mode', columns=['city'])
                    .scale_numerical_features(columns=['age', 'salary'])
                    .encode_categorical_features(columns=['city'])
                    .get_processed_data())

    print("-" * 30)
    print("Processed data:")
    print(processed_df.head())
  
    # We can access the saved scalers and encoders
    print("-" * 30)
    print("Saved Scalers:", processor.scalers)

if __name__ == "__main__":
    main()

Run and Analyze:

Run python scripts/run_processing.py in your terminal. You will see clear log output with each step and its duration displayed at a glance. The result is a clean dataset ready for model training.

This hands-on project perfectly demonstrates the core ideas of this chapter:

  • OOP: We encapsulated all processing logic within the DataProcessor class, achieving high cohesion.
  • Chained Calls: By having each method return self, we implemented elegant chained calls like processor.method1().method2(). This itself embodies functional thinking (data flow).
  • Decorators: The @log_step decorator non-invasively added logging functionality to our processing pipeline.
  • Type Hints: The entire class interface uses type hints, making it clear and robust.
  • Engineering: The project follows a standard structure, with code and data separated, making it easy to manage and extend.

Chapter Summary

In this chapter, we have deeply cultivated the Python "inner strength" of an AI engineer. Starting from advanced features like decorators, generators, and coroutines, we learned how to write more elegant and efficient code to handle challenges such as logging, large data streams, and high-concurrency I/O. Then, we explored the best practices of two programming paradigms -- OOP and FP -- in AI projects, understanding how to use OOP to build the macro structure of a project and FP to organize the micro data flows. Finally, we turned our attention to the cornerstones of engineering: PEP 8 code standards, type hints, and a standard project structure. They are the lifelines for ensuring code quality and team collaboration efficiency.

Through the final hands-on project, we condensed all this theoretical knowledge into a concrete, usable DataProcessor utility class. This process is the leap from "knowing" to "doing."

Always remember that code is the vessel for thought. Elegant, robust, and maintainable code is itself the embodiment of excellent engineering thinking. Building on the solid foundation laid in this chapter, we will begin exploring the core toolchain of data science -- NumPy, Pandas, and others -- in the next chapter. By then, you will more deeply appreciate how a solid Python foundation allows you to sail through the ocean of data processing with the wind at your back, achieving twice the results with half the effort.

writings