Split View: Rust for AI 시스템: Candle, PyO3, axum으로 고성능 AI 추론 서버 구축까지

fn main() {
    // 소유권 이동 (move)
    let tensor_data = vec![1.0f32, 2.0, 3.0, 4.0];
    let moved = tensor_data; // tensor_data는 더 이상 유효하지 않음

    // 빌림(borrowing) — 소유권 이전 없이 참조
    let weights = vec![0.1f32, 0.2, 0.3];
    let sum = compute_sum(&weights); // 불변 참조
    println!("weights still valid: {:?}, sum: {}", weights, sum);
}

fn compute_sum(data: &[f32]) -> f32 {
    data.iter().sum()
}

// 라이프타임 명시: 반환된 참조는 입력 슬라이스보다 오래 살 수 없음
fn longest_sequence<'a>(seq1: &'a [f32], seq2: &'a [f32]) -> &'a [f32] {
    if seq1.len() > seq2.len() {
        seq1
    } else {
        seq2
    }
}

struct ModelWeights<'a> {
    data: &'a [f32],  // 외부 버퍼를 참조 (복사 없이)
    shape: (usize, usize),
}

impl<'a> ModelWeights<'a> {
    fn new(data: &'a [f32], rows: usize, cols: usize) -> Self {
        ModelWeights { data, shape: (rows, cols) }
    }

    fn get(&self, row: usize, col: usize) -> f32 {
        self.data[row * self.shape.1 + col]
    }
}

use std::ops::{Add, Mul};

// AI 연산을 위한 트레이트 정의
trait Activation {
    fn forward(&self, x: f32) -> f32;
    fn backward(&self, x: f32) -> f32; // 미분
}

struct ReLU;
struct Sigmoid;

impl Activation for ReLU {
    fn forward(&self, x: f32) -> f32 {
        x.max(0.0)
    }
    fn backward(&self, x: f32) -> f32 {
        if x > 0.0 { 1.0 } else { 0.0 }
    }
}

impl Activation for Sigmoid {
    fn forward(&self, x: f32) -> f32 {
        1.0 / (1.0 + (-x).exp())
    }
    fn backward(&self, x: f32) -> f32 {
        let s = self.forward(x);
        s * (1.0 - s)
    }
}

// 제네릭 레이어: 어떤 활성화 함수든 사용 가능
fn apply_activation<A: Activation>(activation: &A, inputs: &[f32]) -> Vec<f32> {
    inputs.iter().map(|&x| activation.forward(x)).collect()
}

fn main() {
    let relu = ReLU;
    let data = vec![-1.0, 0.5, 2.0, -0.3];
    let output = apply_activation(&relu, &data);
    println!("ReLU output: {:?}", output); // [0.0, 0.5, 2.0, 0.0]
}

use std::sync::{Arc, Mutex};
use std::thread;

// Rc는 단일 스레드 전용 — AI 추론 서버에서 사용 불가
// Arc(Atomic Reference Count)는 스레드 안전

fn multi_thread_inference() {
    // 모델 가중치를 여러 스레드가 공유
    let model_weights: Arc<Vec<f32>> = Arc::new(vec![0.1, 0.2, 0.3, 0.4]);
    let results: Arc<Mutex<Vec<f32>>> = Arc::new(Mutex::new(Vec::new()));

    let mut handles = vec![];

    for i in 0..4 {
        let weights = Arc::clone(&model_weights);
        let results = Arc::clone(&results);

        let handle = thread::spawn(move || {
            // 각 스레드가 동일한 가중치로 추론
            let inference_result = weights.iter().sum::<f32>() * i as f32;
            results.lock().unwrap().push(inference_result);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("All results: {:?}", results.lock().unwrap());
}

# Cargo.toml
[dependencies]
candle-core = "0.8"
candle-nn = "0.8"
candle-transformers = "0.8"

use candle_core::{Device, Tensor, DType};
use candle_nn::{Linear, Module, VarBuilder, VarMap};

// 간단한 피드포워드 네트워크
struct FeedForward {
    fc1: Linear,
    fc2: Linear,
}

impl FeedForward {
    fn new(vb: VarBuilder) -> candle_core::Result<Self> {
        let fc1 = candle_nn::linear(784, 256, vb.pp("fc1"))?;
        let fc2 = candle_nn::linear(256, 10, vb.pp("fc2"))?;
        Ok(FeedForward { fc1, fc2 })
    }
}

impl Module for FeedForward {
    fn forward(&self, x: &Tensor) -> candle_core::Result<Tensor> {
        let x = self.fc1.forward(x)?;
        let x = x.relu()?;
        self.fc2.forward(&x)
    }
}

fn candle_tensor_ops() -> candle_core::Result<()> {
    let device = Device::Cpu; // 또는 Device::new_cuda(0)?

    // 텐서 생성
    let a = Tensor::randn(0f32, 1f32, (3, 4), &device)?;
    let b = Tensor::randn(0f32, 1f32, (4, 5), &device)?;

    // 행렬 곱
    let c = a.matmul(&b)?;
    println!("Shape: {:?}", c.shape());

    // 요소별 연산
    let x = Tensor::new(&[[1.0f32, 2.0, 3.0]], &device)?;
    let softmax = candle_nn::ops::softmax(&x, 1)?;
    println!("Softmax: {:?}", softmax.to_vec2::<f32>()?);

    Ok(())
}

use burn::prelude::*;
use burn::nn::{Linear, LinearConfig, Relu};

#[derive(Module, Debug)]
struct SimpleNet<B: Backend> {
    fc1: Linear<B>,
    activation: Relu,
    fc2: Linear<B>,
}

impl<B: Backend> SimpleNet<B> {
    fn new(device: &B::Device) -> Self {
        SimpleNet {
            fc1: LinearConfig::new(128, 64).init(device),
            activation: Relu::new(),
            fc2: LinearConfig::new(64, 10).init(device),
        }
    }

    fn forward(&self, x: Tensor<B, 2>) -> Tensor<B, 2> {
        let x = self.fc1.forward(x);
        let x = self.activation.forward(x);
        self.fc2.forward(x)
    }
}

use linfa::prelude::*;
use linfa_clustering::KMeans;
use ndarray::Array2;

fn kmeans_clustering() -> Result<(), Box<dyn std::error::Error>> {
    // 데이터 생성 (임베딩 벡터라고 가정)
    let data: Array2<f64> = Array2::from_shape_fn((100, 10), |(i, j)| {
        (i as f64 * 0.1 + j as f64) % 5.0
    });

    let dataset = Dataset::from(data);

    // K-means 클러스터링 (RAG 시스템에서 유사 문서 그룹화)
    let model = KMeans::params(5)
        .tolerance(1e-5)
        .fit(&dataset)?;

    let predictions = model.predict(&dataset);
    println!("Cluster assignments: {:?}", &predictions.targets()[..10]);

    Ok(())
}

[dependencies]
axum = "0.7"
tokio = { version = "1", features = ["full"] }
candle-core = "0.8"
candle-transformers = "0.8"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tower = "0.4"

use axum::{
    extract::State,
    http::StatusCode,
    response::Json,
    routing::post,
    Router,
};
use candle_core::{Device, Tensor};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::Mutex;

#[derive(Deserialize)]
struct InferenceRequest {
    prompt: String,
    max_tokens: Option<usize>,
    temperature: Option<f32>,
}

#[derive(Serialize)]
struct InferenceResponse {
    text: String,
    tokens_generated: usize,
    latency_ms: u64,
}

// 모델 상태를 Arc로 공유 (스레드 안전)
struct AppState {
    device: Device,
    // model: Arc<Mutex<YourLLMModel>>,
}

async fn inference_handler(
    State(state): State<Arc<AppState>>,
    Json(req): Json<InferenceRequest>,
) -> Result<Json<InferenceResponse>, StatusCode> {
    let start = std::time::Instant::now();

    // 실제로는 tokenizer + model forward pass
    let max_tokens = req.max_tokens.unwrap_or(100);
    let _temperature = req.temperature.unwrap_or(0.7);

    // 예시: 더미 추론 (실제 구현에서는 Candle 모델 사용)
    let generated_text = format!("Response to: {}", req.prompt);
    let latency = start.elapsed().as_millis() as u64;

    Ok(Json(InferenceResponse {
        text: generated_text,
        tokens_generated: max_tokens,
        latency_ms: latency,
    }))
}

async fn health_handler() -> &'static str {
    "OK"
}

#[tokio::main]
async fn main() {
    let state = Arc::new(AppState {
        device: Device::Cpu,
    });

    let app = Router::new()
        .route("/inference", post(inference_handler))
        .route("/health", axum::routing::get(health_handler))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await.unwrap();
    println!("AI Inference Server running on port 8080");
    axum::serve(listener, app).await.unwrap();
}

use tokio::sync::mpsc;
use std::sync::Arc;

struct BatchRequest {
    id: u64,
    prompt: String,
    response_tx: tokio::sync::oneshot::Sender<String>,
}

// 배치 처리 워커 — 여러 요청을 묶어서 GPU 효율 극대화
async fn batch_inference_worker(
    mut rx: mpsc::Receiver<BatchRequest>,
    batch_size: usize,
    timeout_ms: u64,
) {
    loop {
        let mut batch = Vec::new();

        // 첫 요청 대기
        if let Some(req) = rx.recv().await {
            batch.push(req);
        } else {
            break;
        }

        // 타임아웃 내에 배치 채우기
        let deadline = tokio::time::Instant::now()
            + tokio::time::Duration::from_millis(timeout_ms);

        while batch.len() < batch_size {
            match tokio::time::timeout_at(deadline, rx.recv()).await {
                Ok(Some(req)) => batch.push(req),
                _ => break,
            }
        }

        // 배치 처리 (실제로는 GPU에서 병렬 추론)
        for req in batch {
            let response = format!("Batch response for: {}", req.prompt);
            let _ = req.response_tx.send(response);
        }
    }
}

[lib]
name = "rust_ai_tools"
crate-type = ["cdylib"]

[dependencies]
pyo3 = { version = "0.22", features = ["extension-module"] }
numpy = "0.22"

use pyo3::prelude::*;
use pyo3::types::PyList;

// Python에서 호출 가능한 Rust 함수
#[pyfunction]
fn fast_cosine_similarity(a: Vec<f32>, b: Vec<f32>) -> PyResult<f32> {
    if a.len() != b.len() {
        return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
            "Vectors must have the same length"
        ));
    }

    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
    let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();

    Ok(dot / (norm_a * norm_b))
}

// numpy 배열을 받아서 처리 — GIL 해제로 병렬 처리 가능
#[pyfunction]
fn batch_normalize<'py>(
    py: Python<'py>,
    embeddings: Vec<Vec<f32>>,
) -> PyResult<Vec<Vec<f32>>> {
    // GIL을 해제하고 Rust 병렬 처리
    py.allow_threads(|| {
        embeddings.iter().map(|emb| {
            let norm: f32 = emb.iter().map(|x| x * x).sum::<f32>().sqrt();
            emb.iter().map(|x| x / norm).collect()
        }).collect::<Vec<Vec<f32>>>()
    });

    let result: Vec<Vec<f32>> = embeddings.iter().map(|emb| {
        let norm: f32 = emb.iter().map(|x| x * x).sum::<f32>().sqrt();
        emb.iter().map(|x| x / norm).collect()
    }).collect();

    Ok(result)
}

// Python 모듈 등록
#[pymodule]
fn rust_ai_tools(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fast_cosine_similarity, m)?)?;
    m.add_function(wrap_pyfunction!(batch_normalize, m)?)?;
    Ok(())
}

# maturin 설치
pip install maturin

# 개발 모드 빌드 (가상환경에 직접 설치)
maturin develop

# Python에서 사용
# import rust_ai_tools
# similarity = rust_ai_tools.fast_cosine_similarity([1.0, 0.0], [0.0, 1.0])
# print(similarity)  # 0.0

# PyPI 배포용 wheel 빌드
maturin build --release

# pyproject.toml 설정
# [build-system]
# requires = ["maturin>=1.0,<2.0"]
# build-backend = "maturin"

use pyo3::prelude::*;
use rayon::prelude::*;

// rayon을 활용한 진짜 병렬 처리
#[pyfunction]
fn parallel_embedding_search(
    py: Python,
    query: Vec<f32>,
    corpus: Vec<Vec<f32>>,
    top_k: usize,
) -> PyResult<Vec<usize>> {
    // GIL 해제 — Python GIL 없이 Rust 스레드 풀 사용
    let indices = py.allow_threads(|| {
        let mut scores: Vec<(usize, f32)> = corpus
            .par_iter()  // 병렬 이터레이터
            .enumerate()
            .map(|(i, emb)| {
                let dot: f32 = query.iter().zip(emb.iter()).map(|(a, b)| a * b).sum();
                (i, dot)
            })
            .collect();

        scores.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap());
        scores.iter().take(top_k).map(|(i, _)| *i).collect::<Vec<_>>()
    });

    Ok(indices)
}

[lib]
crate-type = ["cdylib"]

[dependencies]
wasm-bindgen = "0.2"
candle-core = { version = "0.8", features = ["wasm"] }

use wasm_bindgen::prelude::*;

// JavaScript에서 호출 가능한 함수
#[wasm_bindgen]
pub struct MiniModel {
    weights: Vec<f32>,
    input_size: usize,
    output_size: usize,
}

#[wasm_bindgen]
impl MiniModel {
    #[wasm_bindgen(constructor)]
    pub fn new(input_size: usize, output_size: usize) -> MiniModel {
        let weights = vec![0.01f32; input_size * output_size];
        MiniModel { weights, input_size, output_size }
    }

    // JavaScript Float32Array를 받아서 추론
    pub fn predict(&self, input: &[f32]) -> Vec<f32> {
        let mut output = vec![0.0f32; self.output_size];
        for i in 0..self.output_size {
            for j in 0..self.input_size {
                output[i] += input[j] * self.weights[i * self.input_size + j];
            }
            // ReLU 활성화
            output[i] = output[i].max(0.0);
        }
        output
    }

    pub fn load_weights(&mut self, weights: Vec<f32>) {
        self.weights = weights;
    }
}

// 텍스트 임베딩의 코사인 유사도 (브라우저에서 RAG 가능)
#[wasm_bindgen]
pub fn cosine_similarity_wasm(a: &[f32], b: &[f32]) -> f32 {
    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
    let na: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let nb: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
    dot / (na * nb)
}

// JavaScript에서 WASM 모듈 사용
import init, { MiniModel, cosine_similarity_wasm } from './rust_ai_wasm.js'

async function runBrowserInference() {
  await init()

  const model = new MiniModel(128, 10)
  const input = new Float32Array(128).fill(0.5)
  const output = model.predict(input)
  console.log('Browser inference output:', output)

  // 벡터 유사도 검색
  const queryEmb = new Float32Array(384).map(() => Math.random())
  const docEmb = new Float32Array(384).map(() => Math.random())
  const sim = cosine_similarity_wasm(queryEmb, docEmb)
  console.log('Similarity:', sim)
}

[dependencies]
polars = { version = "0.44", features = ["lazy", "parquet", "json", "csv"] }

use polars::prelude::*;

fn preprocess_training_data() -> PolarsResult<DataFrame> {
    // Parquet 파일 지연 로딩 (메모리 효율)
    let df = LazyFrame::scan_parquet("training_data.parquet", Default::default())?
        .filter(col("label").is_not_null())
        .with_column(
            // 텍스트 길이 특성 추출
            col("text").str().len_chars().alias("text_length")
        )
        .with_column(
            // 정규화
            (col("score") - col("score").mean()) / col("score").std(1)
        )
        .filter(col("text_length").gt(10))
        .select([col("text"), col("label"), col("score"), col("text_length")])
        .collect()?;

    println!("Shape: {:?}", df.shape());
    println!("{}", df.head(Some(5)));

    Ok(df)
}

fn batch_generator(df: &DataFrame, batch_size: usize) -> Vec<DataFrame> {
    let n = df.height();
    (0..n).step_by(batch_size).map(|start| {
        let end = (start + batch_size).min(n);
        df.slice(start as i64, end - start)
    }).collect()
}

fn compute_label_stats(df: &DataFrame) -> PolarsResult<DataFrame> {
    df.clone().lazy()
        .group_by([col("label")])
        .agg([
            col("score").mean().alias("avg_score"),
            col("score").std(1).alias("std_score"),
            col("text_length").mean().alias("avg_length"),
            col("label").count().alias("count"),
        ])
        .sort(["count"], SortMultipleOptions::default().with_order_descending(true))
        .collect()
}

use arrow::array::{Float32Array, StringArray};
use arrow::record_batch::RecordBatch;
use arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;

fn create_embedding_batch(
    texts: Vec<String>,
    embeddings: Vec<Vec<f32>>,
) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("text", DataType::Utf8, false),
        Field::new("embedding_dim", DataType::Float32, false),
    ]));

    let text_array = StringArray::from(texts);
    // 첫 번째 차원만 저장 (예시)
    let emb_dim: Vec<f32> = embeddings.iter().map(|e| e[0]).collect();
    let emb_array = Float32Array::from(emb_dim);

    RecordBatch::try_new(
        schema,
        vec![Arc::new(text_array), Arc::new(emb_array)],
    ).unwrap()
}

# 기존 Python linter
time flake8 large_ml_project/  # ~30초

# Ruff (Rust)
time ruff check large_ml_project/  # ~0.3초