小语言模型（SLM）部署与应用完全教程

从选型到部署，手把手教你在资源受限环境中运行高效AI模型

目录

1. 什么是小语言模型（SLM）
2. 主流SLM模型对比与选型
3. 模型量化与压缩技术
4. Ollama本地部署实战
5. 边缘设备部署：手机、树莓派与嵌入式
6. SLM微调与领域适配
7. SLM与RAG结合
8. 性能优化与推理加速
9. IoT、车载与工业场景应用
10. 总结与最佳实践

1. 什么是小语言模型（SLM）

小语言模型（Small Language Model，SLM）是指参数量通常在 0.5B ~ 8B 之间的语言模型。与GPT-4、Claude等动辄数百B参数的大模型不同，SLM专为资源受限环境设计，可以在消费级GPU、甚至CPU上高效运行。

SLM的核心优势

SLM的核心理念是：用对的模型，而不是最大的模型。对于分类、摘要、实体提取、代码补全等特定任务，经过优化的SLM完全可以达到接近大模型的效果。

SLM的典型应用场景

• 移动设备AI助手：手机本地运行，无需联网
• IoT设备智能交互：智能家居、工业传感器的自然语言接口
• 车载语音助手：低延迟响应，不依赖网络
• 企业内部文档处理：数据不出内网，满足合规要求
• 代码辅助工具：IDE内嵌，实时补全

2. 主流SLM模型对比与选型

2.1 主流SLM模型一览

Phi-4-mini（微软，3.8B）

微软Phi系列的最新力作。Phi-4-mini在数学推理、代码生成和逻辑推理方面表现突出，是同尺寸模型中的佼佼者。

特点：

• 3.8B参数，支持128K上下文窗口
• 数学和推理能力接近7B级别模型
• 支持函数调用（Function Calling）
• MIT许可证，商用友好

Qwen3-0.6B / Qwen3-1.7B（阿里通义千问）

阿里最新推出的Qwen3系列小模型，支持中英双语，在中文任务上表现优异。

特点：

• 0.6B/1.7B两种尺寸，极致轻量
• 优秀的中文理解和生成能力
• 支持128K上下文
• Apache 2.0许可证

Gemma 3（Google，1B/4B/12B）

Google基于Gemini技术推出的开源小模型系列，1B版本是目前最轻量的实用级SLM之一。

特点：

• 1B/4B/12B多尺寸可选
• 支持多模态（图像理解）
• 128K上下文窗口
• Google Gemma许可证

SmolLM2（HuggingFace，135M/360M/1.7B）

HuggingFace推出的超轻量模型系列，135M版本甚至可以在手机浏览器中运行。

特点：

• 最小135M参数，极致压缩
• 专为端侧推理优化
• Apache 2.0许可证
• 社区活跃，适配工具丰富

2.2 选型决策指南

# SLM选型决策树
def select_slm(use_case, hardware, language):
    """
    根据使用场景、硬件条件和语言需求选择SLM
    
    Args:
        use_case: 使用场景 (code/math/chat/classification/summarization)
        hardware: 硬件条件 (mobile/raspberry_pi/cpu_only/gpu_low/gpu_mid)
        language: 主要语言 (chinese/english/bilingual)
    
    Returns:
        推荐的模型名称及量化格式
    """
    
    # 中文场景优先Qwen3
    if language in ["chinese", "bilingual"]:
        if hardware in ["mobile", "raspberry_pi"]:
            return "Qwen3-0.6B (GGUF Q4_K_M)"
        elif hardware == "cpu_only":
            return "Qwen3-1.7B (GGUF Q4_K_M)"
        else:
            return "Qwen3-1.7B (GGUF Q8_0)"
    
    # 代码和数学推理优先Phi-4-mini
    if use_case in ["code", "math"]:
        if hardware == "gpu_low":
            return "Phi-4-mini (GGUF Q4_K_M)"
        else:
            return "Phi-4-mini (GGUF Q8_0)"
    
    # 极致轻量场景
    if hardware in ["mobile", "raspberry_pi"]:
        return "SmolLM2-360M (GGUF Q4_K_M)"
    
    # 通用对话
    if hardware == "gpu_low":
        return "Gemma 3 4B (GGUF Q4_K_M)"
    else:
        return "Gemma 3 4B (GGUF Q8_0)"

# 示例
model = select_slm(
    use_case="chat",
    hardware="cpu_only", 
    language="chinese"
)
print(f"推荐模型: {model}")
# 输出: 推荐模型: Qwen3-1.7B (GGUF Q4_K_M)

2.3 模型能力基准对比

以下是在常见基准测试上的大致对比（2025年数据，分数越高越好）：

注意：以上数据为近似值，实际性能会因量化方式、推理引擎和硬件环境而异。

3. 模型量化与压缩技术

量化是SLM部署的核心技术。通过将模型权重从FP16/FP32压缩到更低精度，可以在几乎不损失性能的前提下大幅降低内存占用和提升推理速度。

3.1 量化基础知识

原始精度 (FP32): 每个参数 4 字节 → 3.8B模型 ≈ 15.2 GB
FP16:           每个参数 2 字节 → 3.8B模型 ≈ 7.6 GB
INT8:           每个参数 1 字节 → 3.8B模型 ≈ 3.8 GB
INT4:           每个参数 0.5 字节 → 3.8B模型 ≈ 1.9 GB

3.2 GGUF量化（推荐）

GGUF是llama.cpp项目的模型格式，支持多种量化级别，是本地部署最常用的格式。

# 安装llama.cpp（用于量化）
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make -j$(nproc)

# 将HuggingFace模型转换为GGUF
python3 convert_hf_to_gguf.py \
    /path/to/huggingface/model \
    --outfile model-f16.gguf \
    --outtype f16

# 量化为Q4_K_M（推荐平衡选择）
./llama-quantize \
    model-f16.gguf \
    model-q4_k_m.gguf \
    Q4_K_M

# 量化为Q8_0（更高精度）
./llama-quantize \
    model-f16.gguf \
    model-q8_0.gguf \
    Q8_0

3.3 常见量化类型对比

3.4 使用Python进行GPTQ量化

"""
使用AutoGPTQ对模型进行INT4量化
适用于GPU部署场景
"""
from transformers import AutoTokenizer
from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig

# 加载原始模型
model_name = "Qwen/Qwen3-1.7B"
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 配置量化参数
quantize_config = BaseQuantizeConfig(
    bits=4,              # 量化位数
    group_size=128,      # 量化分组大小
    damp_percent=0.01,   # 阻尼系数
    desc_act=True,       # 按激活值排序（更精确）
)

# 加载模型
model = AutoGPTQForCausalLM.from_pretrained(
    model_name,
    quantize_config=quantize_config
)

# 准备校准数据（用于量化时的校准）
calibration_data = [
    "请帮我解释什么是机器学习。",
    "写一个Python函数来计算斐波那契数列。",
    "What is the capital of France?",
    "Explain quantum computing in simple terms.",
    "帮我写一封商务邮件。",
]
calibration_tokens = [
    tokenizer(text, return_tensors="pt") 
    for text in calibration_data
]

# 执行量化
model.quantize(calibration_tokens)

# 保存量化模型
model.save_quantized("./qwen3-1.7b-gptq-4bit")
tokenizer.save_pretrained("./qwen3-1.7b-gptq-4bit")

print("量化完成！模型已保存。")

3.5 AWQ量化

"""
使用AWQ进行4bit量化
AWQ通常比GPTQ有更好的推理速度
"""
from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

model_path = "Qwen/Qwen3-1.7B"
quant_path = "./qwen3-1.7b-awq-4bit"

# 加载模型和分词器
model = AutoAWQForCausalLM.from_pretrained(model_path)
tokenizer = AutoTokenizer.from_pretrained(model_path)

# 量化配置
quant_config = {
    "zero_point": True,
    "q_group_size": 128,
    "w_bit": 4,
    "version": "GEMM"  # GEMM kernel，更快
}

# 执行量化
model.quantize(tokenizer, quant_config=quant_config)

# 保存
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)

print(f"AWQ量化模型已保存到 {quant_path}")

4. Ollama本地部署实战

Ollama是目前最简单、最流行的本地LLM运行工具，支持一键下载和运行SLM。

4.1 安装Ollama

# Linux一键安装
curl -fsSL https://ollama.com/install.sh | sh

# macOS：从 https://ollama.com 下载安装包

# Windows：从 https://ollama.com 下载安装包

# 验证安装
ollama --version

4.2 运行SLM模型

# 运行Qwen3-0.6B（最轻量）
ollama run qwen3:0.6b

# 运行Phi-4-mini
ollama run phi4-mini

# 运行Gemma3 4B
ollama run gemma3:4b

# 列出已下载的模型
ollama list

# 查看模型信息
ollama show qwen3:0.6b

4.3 创建自定义模型（Modelfile）

# 创建Modelfile
cat > Modelfile << 'EOF'
FROM qwen3:1.7b

# 系统提示词
SYSTEM """
你是一个专业的中文AI助手，擅长回答技术问题。
请用简洁清晰的中文回答用户的问题。
"""

# 模型参数
PARAMETER temperature 0.7
PARAMETER top_p 0.9
PARAMETER top_k 40
PARAMETER num_ctx 4096
PARAMETER num_predict 512
PARAMETER repeat_penalty 1.1
EOF

# 创建自定义模型
ollama create my-assistant -f Modelfile

# 运行自定义模型
ollama run my-assistant

4.4 Ollama API调用

"""
通过Ollama API调用本地SLM
"""
import requests
import json

def chat_with_ollama(prompt, model="qwen3:1.7b", system=None):
    """
    调用Ollama API进行对话
    
    Args:
        prompt: 用户输入
        model: 模型名称
        system: 系统提示词
    
    Returns:
        模型回复文本
    """
    url = "http://localhost:11434/api/chat"
    
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})
    
    payload = {
        "model": model,
        "messages": messages,
        "stream": False,
        "options": {
            "temperature": 0.7,
            "top_p": 0.9,
            "num_ctx": 4096,
        }
    }
    
    response = requests.post(url, json=payload)
    response.raise_for_status()
    
    result = response.json()
    return result["message"]["content"]

# 流式输出版本
def chat_stream(prompt, model="qwen3:1.7b"):
    """流式输出，适合实时显示"""
    url = "http://localhost:11434/api/chat"
    
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    
    response = requests.post(url, json=payload, stream=True)
    
    for line in response.iter_lines():
        if line:
            chunk = json.loads(line)
            if "message" in chunk:
                print(chunk["message"]["content"], end="", flush=True)
            if chunk.get("done"):
                print()  # 换行

# 使用示例
if __name__ == "__main__":
    # 基本对话
    reply = chat_with_ollama(
        "解释什么是SLM，用3句话概括",
        model="qwen3:1.7b",
        system="你是一个技术科普专家"
    )
    print(f"回复: {reply}")
    
    # 流式输出
    print("\n--- 流式输出 ---")
    chat_stream("写一个Python快速排序算法")

4.5 Ollama批量处理

"""
使用Ollama进行批量文本处理
适用于需要处理大量数据的场景
"""
import json
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

class OllamaBatchProcessor:
    def __init__(self, model="qwen3:1.7b", base_url="http://localhost:11434"):
        self.model = model
        self.base_url = base_url
    
    def process_single(self, text, task_prompt):
        """处理单条数据"""
        url = f"{self.base_url}/api/chat"
        
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": task_prompt},
                {"role": "user", "content": text}
            ],
            "stream": False,
        }
        
        try:
            response = requests.post(url, json=payload, timeout=60)
            response.raise_for_status()
            return response.json()["message"]["content"]
        except Exception as e:
            return f"Error: {str(e)}"
    
    def batch_process(self, texts, task_prompt, max_workers=4):
        """
        批量处理文本
        
        Args:
            texts: 待处理文本列表
            task_prompt: 任务提示词
            max_workers: 并发数
        
        Returns:
            处理结果列表
        """
        results = [None] * len(texts)
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_idx = {
                executor.submit(self.process_single, text, task_prompt): i
                for i, text in enumerate(texts)
            }
            
            for future in as_completed(future_to_idx):
                idx = future_to_idx[future]
                results[idx] = future.result()
                print(f"  完成: {idx+1}/{len(texts)}")
        
        return results

# 使用示例
if __name__ == "__main__":
    processor = OllamaBatchProcessor(model="qwen3:1.7b")
    
    # 待处理的评论数据
    reviews = [
        "这家餐厅的菜太好吃了，服务也很棒！",
        "等了一个小时才上菜，味道一般般。",
        "环境不错，但价格偏贵。",
        "强烈推荐这家店的招牌菜！",
        "服务员态度很差，再也不来了。",
    ]
    
    # 情感分析任务
    task = """请对以下评论进行情感分析，只返回JSON格式：
{"sentiment": "positive/negative/neutral", "confidence": 0.0-1.0}"""
    
    results = processor.batch_process(reviews, task, max_workers=2)
    
    for review, result in zip(reviews, results):
        print(f"\n评论: {review}")
        print(f"分析: {result}")

5. 边缘设备部署：手机、树莓派与嵌入式

5.1 树莓派部署

树莓派是SLM边缘部署的理想平台。以树莓派5（8GB RAM）为例：

# 1. 安装Ollama（树莓派5支持ARM64）
curl -fsSL https://ollama.com/install.sh | sh

# 2. 下载轻量模型
ollama pull qwen3:0.6b

# 3. 运行测试
ollama run qwen3:0.6b "你好，请介绍一下自己"

# 4. 测试性能
time ollama run qwen3:0.6b "写一首关于AI的五言绝句" --verbose

"""
树莓派SLM服务 - 将树莓派变成AI API服务器
其他设备可以通过网络调用
"""
from flask import Flask, request, jsonify
import requests
import psutil

app = Flask(__name__)

OLLAMA_URL = "http://localhost:11434/api/chat"
DEFAULT_MODEL = "qwen3:0.6b"

@app.route("/api/chat", methods=["POST"])
def chat():
    """聊天接口"""
    data = request.json
    prompt = data.get("prompt", "")
    model = data.get("model", DEFAULT_MODEL)
    
    if not prompt:
        return jsonify({"error": "prompt is required"}), 400
    
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": False,
    }
    
    try:
        resp = requests.post(OLLAMA_URL, json=payload, timeout=120)
        result = resp.json()
        return jsonify({
            "response": result["message"]["content"],
            "model": model,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route("/api/status", methods=["GET"])
def status():
    """系统状态接口"""
    return jsonify({
        "cpu_percent": psutil.cpu_percent(),
        "memory": {
            "total_gb": round(psutil.virtual_memory().total / 1e9, 2),
            "used_gb": round(psutil.virtual_memory().used / 1e9, 2),
            "percent": psutil.virtual_memory().percent,
        },
        "temperature": get_cpu_temp(),
        "model": DEFAULT_MODEL,
    })

def get_cpu_temp():
    """获取CPU温度（树莓派）"""
    try:
        with open("/sys/class/thermal/thermal_zone0/temp") as f:
            return round(int(f.read().strip()) / 1000, 1)
    except:
        return None

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

5.2 Android手机部署

使用llama.cpp的Android移植版或MNN框架：

"""
导出Android可用的模型文件
使用llama.cpp的Android兼容格式
"""
import subprocess
import os

def export_for_android(model_path, output_dir="./android_model"):
    """
    将模型导出为Android兼容格式
    
    Args:
        model_path: 原始GGUF模型路径
        output_dir: 输出目录
    """
    os.makedirs(output_dir, exist_ok=True)
    
    # 量化为移动端友好的格式
    output_path = os.path.join(output_dir, "model-q4_0.gguf")
    
    cmd = [
        "./llama-quantize",
        model_path,
        output_path,
        "Q4_0",  # Q4_0对ARM NEON指令集优化更好
    ]
    
    subprocess.run(cmd, check=True)
    
    # 检查文件大小
    size_mb = os.path.getsize(output_path) / 1e6
    print(f"导出完成: {output_path}")
    print(f"文件大小: {size_mb:.1f} MB")
    
    if size_mb > 2048:
        print("⚠️ 警告: 文件超过2GB，可能在部分手机上运行缓慢")
    elif size_mb > 1024:
        print("⚠️ 注意: 建议在4GB以上内存的手机上运行")
    else:
        print("✅ 文件大小适合大多数智能手机")
    
    return output_path

# 使用示例
# export_for_android("./qwen3-0.6b-f16.gguf")

5.3 iOS部署（使用MLX）

"""
使用Apple MLX框架导出iOS可用的SLM
MLX是Apple专门为Apple Silicon优化的机器学习框架
"""
import subprocess

def export_for_ios_mlx(hf_model_name, output_dir="./ios_model"):
    """
    使用mlx-lm将模型转换为iOS可用格式
    
    前提: pip install mlx-lm
    """
    # 下载并转换模型
    cmd = [
        "python3", "-m", "mlx_lm", "convert",
        "--hf-path", hf_model_name,
        "--mlx-path", output_dir,
        "--q-bits", "4",
        "--q-group-size", "64",
    ]
    
    subprocess.run(cmd, check=True)
    print(f"iOS MLX模型已导出到: {output_dir}")
    print("可以使用mlx-swift在iOS应用中加载此模型")

# 使用示例
# export_for_ios_mlx("Qwen/Qwen3-0.6B")

6. SLM微调与领域适配

SLM的一大优势是可以用少量数据进行微调，快速适配特定领域任务。

6.1 使用LoRA进行轻量微调

"""
使用LoRA对SLM进行领域微调
以法律文书摘要为例
"""
from transformers import (
    AutoModelForCausalLM, 
    AutoTokenizer,
    TrainingArguments,
)
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer
from datasets import Dataset
import torch

# 1. 加载基础模型
model_name = "Qwen/Qwen3-1.7B"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token

# 2. 配置LoRA
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                    # LoRA秩
    lora_alpha=32,           # 缩放因子
    lora_dropout=0.05,       # Dropout
    target_modules=[         # 要适配的层
        "q_proj", "k_proj", 
        "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"
    ],
    bias="none",
)

# 3. 应用LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出示例: trainable params: 8,388,608 || all params: 1,747,734,528 || 0.48%

# 4. 准备训练数据
train_data = [
    {
        "instruction": "请对以下法律文书进行摘要",
        "input": "原告张三诉被告李四买卖合同纠纷一案，本院于2024年3月15日立案...",
        "output": "本案为买卖合同纠纷，原告张三起诉被告李四，法院于2024年3月15日受理。"
    },
    {
        "instruction": "请对以下法律文书进行摘要",
        "input": "被告人王五因涉嫌盗窃罪，于2024年2月10日被公安机关抓获...",
        "output": "被告人王五涉嫌盗窃罪，于2024年2月10日被抓获。"
    },
    # ... 更多训练样本（建议至少100-500条）
]

def format_instruction(example):
    """格式化为指令微调格式"""
    if example["input"]:
        text = f"""### 指令:
{example["instruction"]}

### 输入:
{example["input"]}

### 回复:
{example["output"]}"""
    else:
        text = f"""### 指令:
{example["instruction"]}

### 回复:
{example["output"]}"""
    return {"text": text}

dataset = Dataset.from_list(train_data)
dataset = dataset.map(format_instruction)

# 5. 训练配置
training_args = TrainingArguments(
    output_dir="./legal-slm-lora",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    warmup_ratio=0.1,
    logging_steps=10,
    save_strategy="epoch",
    fp16=True,
    optim="paged_adamw_8bit",
    report_to="none",
)

# 6. 开始训练
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset,
    dataset_text_field="text",
    max_seq_length=2048,
    tokenizer=tokenizer,
)

trainer.train()

# 7. 保存LoRA权重
trainer.save_model("./legal-slm-lora")
print("微调完成！LoRA权重已保存。")

# 8. 合并LoRA权重到基础模型（可选，便于部署）
from peft import PeftModel

base_model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype=torch.float16)
lora_model = PeftModel.from_pretrained(base_model, "./legal-slm-lora")
merged_model = lora_model.merge_and_unload()
merged_model.save_pretrained("./legal-slm-merged")
tokenizer.save_pretrained("./legal-slm-merged")
print("合并模型已保存。")

6.2 使用Unsloth加速微调

"""
使用Unsloth进行2x速度的SLM微调
Unsloth通过内存优化和计算图优化大幅提升训练速度
"""
from unsloth import FastLanguageModel
import torch

# 1. 加载模型（Unsloth优化版）
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Qwen3-1.7B",  # Unsloth优化版
    max_seq_length=2048,
    dtype=None,  # 自动检测
    load_in_4bit=True,  # 4bit加载，节省显存
)

# 2. 添加LoRA
model = FastLanguageModel.get_peft_model(
    model,
    r=16,
    target_modules=[
        "q_proj", "k_proj", 
        "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
    lora_alpha=32,
    lora_dropout=0,
    bias="none",
    use_gradient_checkpointing="unsloth",  # Unsloth优化
    random_state=42,
)

# 3. 准备数据（Alpaca格式）
from datasets import load_dataset

dataset = load_dataset("json", data_files="train_data.json", split="train")

def format_prompt(example):
    return f"""### 指令:
{example['instruction']}

### 输入:
{example.get('input', '')}

### 回复:
{example['output']}"""

# 4. 训练
from trl import SFTTrainer
from transformers import TrainingArguments

trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    dataset_text_field="text",
    max_seq_length=2048,
    args=TrainingArguments(
        per_device_train_batch_size=2,
        gradient_accumulation_steps=4,
        warmup_steps=5,
        num_train_epochs=3,
        learning_rate=2e-4,
        fp16=not torch.cuda.is_bf16_supported(),
        bf16=torch.cuda.is_bf16_supported(),
        logging_steps=10,
        output_dir="outputs",
    ),
)

trainer.train()

# 5. 保存模型（多种格式）
model.save_pretrained("./fine-tuned-slm")        # LoRA
model.save_pretrained_merged("./merged-slm", tokenizer)  # 合并
model.save_pretrained_gguf("./gguf-slm", tokenizer)      # GGUF格式

7. SLM与RAG结合

将SLM与检索增强生成（RAG）结合，可以让小模型利用外部知识库大幅提升回答质量。

7.1 SLM + RAG完整实现

"""
SLM + RAG 系统完整实现
使用Ollama作为LLM后端，ChromaDB作为向量数据库
"""
import chromadb
from chromadb.utils import embedding_functions
import requests
import json
from typing import List, Dict

class SLMMemoryRAG:
    """
    基于SLM的RAG系统
    
    特点：
    - 完全本地运行，数据不出网络
    - 支持文档导入和实时检索
    - 使用SLM生成回答
    """
    
    def __init__(
        self,
        llm_model="qwen3:1.7b",
        embedding_model="nomic-embed-text",
        collection_name="knowledge_base",
        ollama_url="http://localhost:11434",
    ):
        self.llm_model = llm_model
        self.ollama_url = ollama_url
        
        # 初始化向量数据库
        self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
        
        # 使用Ollama的嵌入模型
        self.embedding_fn = embedding_functions.OllamaEmbeddingFunction(
            url=f"{ollama_url}/api/embeddings",
            model_name=embedding_model,
        )
        
        self.collection = self.chroma_client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn,
        )
    
    def add_documents(self, documents: List[Dict]):
        """
        添加文档到知识库
        
        Args:
            documents: 文档列表，每个文档包含 text 和 metadata
        """
        for i, doc in enumerate(documents):
            self.collection.add(
                documents=[doc["text"]],
                metadatas=[doc.get("metadata", {})],
                ids=[f"doc_{i}"],
            )
        print(f"已添加 {len(documents)} 条文档")
    
    def query(self, question: str, n_results: int = 3) -> str:
        """
        查询并生成回答
        
        Args:
            question: 用户问题
            n_results: 检索的文档数量
        
        Returns:
            基于检索结果的回答
        """
        # 1. 检索相关文档
        results = self.collection.query(
            query_texts=[question],
            n_results=n_results,
        )
        
        # 2. 构建上下文
        context = "\n\n".join(results["documents"][0])
        
        # 3. 构建提示词
        prompt = f"""基于以下参考信息回答用户的问题。
如果参考信息中没有相关内容，请说明你不确定。

## 参考信息
{context}

## 用户问题
{question}

## 回答
"""
        
        # 4. 调用SLM生成回答
        response = requests.post(
            f"{self.ollama_url}/api/chat",
            json={
                "model": self.llm_model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": False,
            },
            timeout=120,
        )
        
        return response.json()["message"]["content"]

# 使用示例
if __name__ == "__main__":
    rag = SLMMemoryRAG(
        llm_model="qwen3:1.7b",
        embedding_model="nomic-embed-text",
    )
    
    # 添加知识文档
    documents = [
        {
            "text": "Python是一种高级编程语言，由Guido van Rossum于1991年创建。Python的设计哲学强调代码的可读性和简洁性。",
            "metadata": {"topic": "python", "source": "wiki"},
        },
        {
            "text": "机器学习是人工智能的一个分支，它使计算机系统能够通过经验自动改进。主要分为监督学习、无监督学习和强化学习。",
            "metadata": {"topic": "ml", "source": "textbook"},
        },
        {
            "text": "Docker是一个开源的应用容器引擎，允许开发者将应用及其依赖打包到一个轻量级、可移植的容器中。",
            "metadata": {"topic": "docker", "source": "docs"},
        },
    ]
    
    rag.add_documents(documents)
    
    # 查询
    answer = rag.query("Python是什么时候创建的？")
    print(f"回答: {answer}")
    
    answer = rag.query("机器学习有哪些类型？")
    print(f"回答: {answer}")

7.2 文档分块策略

"""
智能文档分块 - 优化RAG检索质量的关键
"""
from typing import List
import re

class SmartChunker:
    """
    智能文档分块器
    支持多种分块策略，保证语义完整性
    """
    
    def __init__(self, chunk_size=500, overlap=50):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk_by_sentence(self, text: str) -> List[str]:
        """按句子分块，保证句子完整性"""
        # 按中英文句号、问号、感叹号分句
        sentences = re.split(r'(?<=[。！？.!?])\s*', text)
        sentences = [s.strip() for s in sentences if s.strip()]
        
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk) + len(sentence) <= self.chunk_size:
                current_chunk += sentence + " "
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + " "
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks
    
    def chunk_by_paragraph(self, text: str) -> List[str]:
        """按段落分块"""
        paragraphs = text.split("\n\n")
        chunks = []
        current_chunk = ""
        
        for para in paragraphs:
            para = para.strip()
            if not para:
                continue
            
            if len(current_chunk) + len(para) <= self.chunk_size:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                # 如果单个段落超过chunk_size，按句子再分
                if len(para) > self.chunk_size:
                    chunks.extend(self.chunk_by_sentence(para))
                else:
                    current_chunk = para + "\n\n"
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks
    
    def chunk_with_metadata(self, text: str, source: str = "") -> List[dict]:
        """分块并添加元数据"""
        chunks = self.chunk_by_paragraph(text)
        
        return [
            {
                "text": chunk,
                "metadata": {
                    "source": source,
                    "chunk_index": i,
                    "chunk_length": len(chunk),
                }
            }
            for i, chunk in enumerate(chunks)
        ]

# 使用示例
chunker = SmartChunker(chunk_size=500, overlap=50)
chunks = chunker.chunk_with_metadata(
    "这是一篇很长的文章...",
    source="technical_doc.pdf"
)

8. 性能优化与推理加速

8.1 llama.cpp编译优化

# 针对不同硬件优化编译llama.cpp

# NVIDIA GPU加速
cmake -B build -DGGML_CUDA=ON
cmake --build build --config Release -j$(nproc)

# Apple Silicon (Metal)加速
cmake -B build -DGGML_METAL=ON
cmake --build build --config Release -j$(nproc)

# AMD GPU (ROCm)加速
cmake -B build -DGGML_HIP=ON
cmake --build build --config Release -j$(nproc)

# Intel CPU优化（AVX512）
cmake -B build -DGGML_AVX512=ON
cmake --build build --config Release -j$(nproc)

8.2 推理参数调优

"""
SLM推理参数调优工具
通过基准测试找到最佳参数组合
"""
import requests
import time
import itertools

class SLMProfiler:
    def __init__(self, model="qwen3:1.7b"):
        self.model = model
        self.url = "http://localhost:11434/api/chat"
        self.test_prompts = [
            "请用一句话解释什么是人工智能。",
            "写一个Python的hello world程序。",
            "今天天气怎么样？",
        ]
    
    def benchmark(self, params: dict) -> dict:
        """
        使用指定参数进行基准测试
        
        Returns:
            包含延迟、tokens/s等指标的字典
        """
        results = []
        
        for prompt in self.test_prompts:
            start = time.time()
            
            payload = {
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": False,
                "options": params,
            }
            
            response = requests.post(self.url, json=payload, timeout=120)
            elapsed = time.time() - start
            
            result = response.json()
            eval_count = result.get("eval_count", 0)
            eval_duration = result.get("eval_duration", 1) / 1e9  # ns -> s
            
            results.append({
                "latency": elapsed,
                "tokens": eval_count,
                "tokens_per_sec": eval_count / eval_duration if eval_duration > 0 else 0,
            })
        
        # 计算平均值
        avg_latency = sum(r["latency"] for r in results) / len(results)
        avg_tps = sum(r["tokens_per_sec"] for r in results) / len(results)
        
        return {
            "avg_latency": round(avg_latency, 2),
            "avg_tokens_per_sec": round(avg_tps, 1),
        }
    
    def find_optimal_params(self):
        """测试不同参数组合，找到最优配置"""
        
        param_grid = {
            "num_ctx": [2048, 4096],
            "num_thread": [4, 8],
            "batch_size": [128, 256, 512],
        }
        
        best_score = 0
        best_params = None
        
        keys = list(param_grid.keys())
        values = list(param_grid.values())
        
        for combo in itertools.product(*values):
            params = dict(zip(keys, combo))
            print(f"测试参数: {params}")
            
            result = self.benchmark(params)
            print(f"  延迟: {result['avg_latency']}s, "
                  f"速度: {result['avg_tokens_per_sec']} tok/s")
            
            # 评分：速度越高越好
            score = result["avg_tokens_per_sec"]
            if score > best_score:
                best_score = score
                best_params = params.copy()
                best_params["_result"] = result
        
        print(f"\n最优参数: {best_params}")
        print(f"最佳速度: {best_score} tok/s")
        return best_params

# 使用示例
# profiler = SLMProfiler(model="qwen3:1.7b")
# profiler.find_optimal_params()

8.3 模型并发服务

"""
高性能SLM推理服务
支持并发请求、动态批处理和请求队列
"""
import asyncio
import aiohttp
from collections import deque
import time

class SLMServer:
    def __init__(self, model="qwen3:1.7b", max_concurrent=2):
        self.model = model
        self.url = "http://localhost:11434/api/chat"
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.request_queue = deque()
        self.stats = {"total": 0, "success": 0, "errors": 0}
    
    async def generate(self, prompt: str, system: str = None) -> dict:
        """异步生成，支持并发控制"""
        async with self.semaphore:
            messages = []
            if system:
                messages.append({"role": "system", "content": system})
            messages.append({"role": "user", "content": prompt})
            
            payload = {
                "model": self.model,
                "messages": messages,
                "stream": False,
            }
            
            start = time.time()
            
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(self.url, json=payload) as resp:
                        result = await resp.json()
                
                elapsed = time.time() - start
                self.stats["total"] += 1
                self.stats["success"] += 1
                
                return {
                    "response": result["message"]["content"],
                    "latency": round(elapsed, 2),
                }
            except Exception as e:
                self.stats["errors"] += 1
                return {"error": str(e)}
    
    async def batch_generate(self, prompts: list) -> list:
        """批量并发生成"""
        tasks = [self.generate(p) for p in prompts]
        return await asyncio.gather(*tasks)
    
    def get_stats(self):
        """获取服务统计"""
        return self.stats

# 使用示例
async def main():
    server = SLMServer(model="qwen3:1.7b", max_concurrent=2)
    
    prompts = [
        "用一句话解释量子计算",
        "写一个冒泡排序",
        "什么是Docker？",
        "Python的优势是什么？",
    ]
    
    results = await server.batch_generate(prompts)
    for r in results:
        print(r["response"][:100], "...")
    
    print(f"\n统计: {server.get_stats()}")

# asyncio.run(main())

9. IoT、车载与工业场景应用

9.1 IoT设备智能接口

"""
IoT设备SLM控制接口
将自然语言指令转换为设备控制命令
"""
import json
import requests

class IoTSLMController:
    """
    用SLM将自然语言转换为IoT设备控制指令
    
    示例：
    "把客厅的灯调暗一点" → {"device": "living_room_light", "action": "dim", "value": 50}
    """
    
    def __init__(self, model="qwen3:0.6b"):
        self.model = model
        self.url = "http://localhost:11434/api/chat"
        
        self.system_prompt = """你是一个智能家居控制系统。将用户的自然语言指令转换为JSON格式的设备控制命令。

可用设备：
- living_room_light: 客厅灯 (开/关/亮度0-100)
- bedroom_light: 卧室灯 (开/关/亮度0-100)
- air_conditioner: 空调 (开/关/温度16-30/模式:制冷/制热/自动)
- curtain: 窗帘 (开/关/半开)
- tv: 电视 (开/关/频道/音量0-100)

只返回JSON，不要其他内容。格式：
{"device": "设备名", "action": "操作", "value": "值"}

如果无法理解，返回：
{"error": "无法理解指令", "suggestion": "建议这样说: ..."}"""

    def execute_command(self, user_input: str) -> dict:
        """执行自然语言控制指令"""
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_input},
            ],
            "stream": False,
            "options": {"temperature": 0.1},  # 低温度保证输出稳定
        }
        
        response = requests.post(self.url, json=payload, timeout=30)
        result = response.json()["message"]["content"]
        
        # 解析JSON
        try:
            # 处理可能的markdown代码块
            if "```" in result:
                result = result.split("```")[1]
                if result.startswith("json"):
                    result = result[4:]
            
            command = json.loads(result.strip())
            
            # 验证设备名
            valid_devices = [
                "living_room_light", "bedroom_light", 
                "air_conditioner", "curtain", "tv"
            ]
            if command.get("device") in valid_devices:
                return {"status": "success", "command": command}
            else:
                return {"status": "error", "message": "未知设备"}
                
        except json.JSONDecodeError:
            return {"status": "error", "message": "解析失败", "raw": result}

# 使用示例
controller = IoTSLMController(model="qwen3:0.6b")

# 测试指令
test_commands = [
    "把客厅灯打开",
    "空调调到26度",
    "关掉卧室的灯",
    "把窗帘拉开一半",
]

for cmd in test_commands:
    result = controller.execute_command(cmd)
    print(f"指令: {cmd}")
    print(f"结果: {json.dumps(result, ensure_ascii=False, indent=2)}\n")

9.2 工业设备日志分析

"""
工业设备日志分析系统
使用SLM对设备日志进行智能分析和故障诊断
"""
import re
from datetime import datetime
from typing import List, Dict

class IndustrialLogAnalyzer:
    """
    工业设备日志分析器
    
    功能：
    - 日志异常检测
    - 故障根因分析
    - 维护建议生成
    """
    
    def __init__(self, ollama_model="qwen3:1.7b"):
        import requests
        self.model = ollama_model
        self.url = "http://localhost:11434/api/chat"
        self.requests = requests
        
        self.system_prompt = """你是一名资深的工业设备维护工程师。
请根据提供的设备日志，分析可能的故障原因并给出维护建议。

分析要求：
1. 识别异常模式和错误代码
2. 判断故障严重程度（低/中/高/紧急）
3. 给出具体的原因分析
4. 提供可操作的维护建议

请用JSON格式输出：
{
    "severity": "低/中/高/紧急",
    "anomaly": "异常描述",
    "cause": "可能原因",
    "suggestion": "维护建议",
    "urgency": "是否需要立即处理"
}"""

    def analyze_logs(self, logs: List[str]) -> Dict:
        """分析一组设备日志"""
        log_text = "\n".join(logs)
        
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": f"请分析以下设备日志：\n\n{log_text}"},
            ],
            "stream": False,
        }
        
        response = self.requests.post(self.url, json=payload, timeout=60)
        result = response.json()["message"]["content"]
        
        # 解析JSON结果
        try:
            if "```" in result:
                result = result.split("```")[1]
                if result.startswith("json"):
                    result = result[4:]
            return {"status": "success", "analysis": result}
        except:
            return {"status": "success", "analysis": result, "raw": True}
    
    def monitor_realtime(self, log_file: str, check_interval: int = 60):
        """实时监控日志文件"""
        import time
        
        print(f"开始监控日志: {log_file}")
        print(f"检查间隔: {check_interval}秒")
        
        last_position = 0
        buffer = []
        
        while True:
            try:
                with open(log_file, 'r') as f:
                    f.seek(last_position)
                    new_lines = f.readlines()
                    last_position = f.tell()
                
                if new_lines:
                    # 过滤包含关键词的日志
                    keywords = ["ERROR", "WARN", "FAIL", "异常", "故障", "超温", "过载"]
                    for line in new_lines:
                        if any(kw in line for kw in keywords):
                            buffer.append(line.strip())
                    
                    # 累积到一定数量后分析
                    if len(buffer) >= 5:
                        print(f"\n发现 {len(buffer)} 条异常日志，正在分析...")
                        result = self.analyze_logs(buffer)
                        print(f"分析结果: {result['analysis']}")
                        buffer = []
                
                time.sleep(check_interval)
                
            except KeyboardInterrupt:
                print("\n监控已停止")
                break

# 使用示例
analyzer = IndustrialLogAnalyzer(ollama_model="qwen3:1.7b")

# 分析示例日志
sample_logs = [
    "2024-03-15 10:23:45 [WARN] 电机M-001 温度达到85°C，超过阈值80°C",
    "2024-03-15 10:23:47 [ERROR] 电机M-001 振动频率异常：12.5Hz (正常范围: 8-10Hz)",
    "2024-03-15 10:23:48 [WARN] 电机M-001 电流波动：±15% (正常范围: ±5%)",
    "2024-03-15 10:23:50 [ERROR] 电机M-001 轴承温度传感器读数异常",
]

result = analyzer.analyze_logs(sample_logs)
print(json.dumps(result, ensure_ascii=False, indent=2))

9.3 车载语音助手

"""
车载SLM语音助手
低延迟、离线可用的智能车载交互系统
"""

class CarVoiceAssistant:
    """
    车载语音助手
    
    特点：
    - 极低延迟（使用最小模型）
    - 离线运行
    - 车辆控制集成
    """
    
    def __init__(self):
        import requests
        self.requests = requests
        self.url = "http://localhost:11434/api/chat"
        self.model = "qwen3:0.6b"  # 最小模型保证低延迟
        self.conversation_history = []
        
        self.system_prompt = """你是一个车载智能助手。你的回答必须：
1. 简洁（不超过50字）
2. 优先考虑驾驶安全
3. 涉及驾驶操作时提醒注意安全

你可以：
- 导航：设置目的地、查询路线
- 媒体：播放音乐、调整音量
- 车辆：调节空调、查看车况
- 电话：拨打电话、读取消息"""
    
    def process_voice_command(self, command: str) -> str:
        """处理语音指令"""
        self.conversation_history.append({
            "role": "user", 
            "content": command
        })
        
        # 只保留最近5轮对话
        recent = self.conversation_history[-10:]
        
        messages = [{"role": "system", "content": self.system_prompt}] + recent
        
        payload = {
            "model": self.model,
            "messages": messages,
            "stream": False,
            "options": {
                "temperature": 0.3,
                "num_predict": 100,  # 限制输出长度，保证低延迟
            }
        }
        
        response = self.requests.post(self.url, json=payload, timeout=15)
        reply = response.json()["message"]["content"]
        
        self.conversation_history.append({
            "role": "assistant",
            "content": reply
        })
        
        return reply

# 使用示例
assistant = CarVoiceAssistant()

# 模拟对话
commands = [
    "导航到最近的加油站",
    "播放一些轻松的音乐",
    "空调调低两度",
    "还有多远？",
]

for cmd in commands:
    print(f"用户: {cmd}")
    reply = assistant.process_voice_command(cmd)
    print(f"助手: {reply}\n")

10. 总结与最佳实践

10.1 SLM部署检查清单

✅ 模型选型
  □ 明确使用场景和性能要求
  □ 评估硬件资源（内存、存储、CPU/GPU）
  □ 选择合适的模型和量化级别
  □ 测试模型在目标任务上的表现

✅ 量化优化
  □ 选择合适的量化格式（GGUF/GPTQ/AWQ）
  □ 使用Q4_K_M作为默认量化级别
  □ 在目标硬件上进行基准测试
  □ 验证量化后模型质量

✅ 部署配置
  □ 选择部署方式（Ollama/llama.cpp/vLLM）
  □ 配置合适的推理参数
  □ 设置适当的上下文窗口大小
  □ 配置并发和超时策略

✅ 性能优化
  □ 针对硬件编译优化推理引擎
  □ 启用硬件加速（CUDA/Metal/ROCm）
  □ 优化批处理和并发策略
  □ 监控内存和CPU使用率

✅ 生产部署
  □ 添加健康检查和监控
  □ 配置日志和错误处理
  □ 设置自动重启和故障恢复
  □ 制定模型更新策略

10.2 常见问题与解决方案

10.3 SLM未来趋势

1. 更小更强：模型架构持续优化，相同性能下参数量会继续下降
2. 多模态SLM：视觉、语音等多模态能力向小模型下沉
3. 端侧训练：在设备上直接进行个性化微调
4. 专用芯片：NPU（神经处理单元）将大幅提升端侧推理效率
5. 联邦学习：多个边缘SLM协同学习，保护隐私的同时提升能力

本文档持续更新。 SLM领域发展迅速，建议定期关注llama.cpp、Ollama、HuggingFace等项目的最新动态。