开源模型与本地部署

# GPTQ: 4-bit 量化，需要 GPU
# pip install auto-gptq optimum

from transformers import AutoModelForCausalLM, AutoTokenizer

model_name = "TheBloke/Llama-2-7B-GPTQ"

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",
    trust_remote_code=True
)

# 推理
prompt = "用 Python 写一个快排算法"
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")

outputs = model.generate(
    **inputs,
    max_new_tokens=512
)

print(tokenizer.decode(outputs[0]))

# AWQ: Activation-aware Weight Quantization
# pip install awq transformers

from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

model_path = "TheBloke/Llama-2-7B-AWQ"

quant_path = "path/to/save/quantized"

# 量化
quant_config = {
    "zero_point": True,
    "q_group_size": 128,
    "w_bit": 4,
    "version": "GEMM"
}

model = AutoAWQForCausalLM.from_pretrained(
    model_path,
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)

# 量化模型
model.quantize(tokenizer, quant_config=quant_config)

# 保存
model.save_quantized(quant_path)

# Python API
# pip install ollama

import ollama

# 基础使用
response = ollama.chat(model='llama3.1:8b', messages=[
    {'role': 'user', 'content': '用 Python 写一个快排算法'}
])

print(response['message']['content'])

# 流式输出
for chunk in ollama.chat(
    model='llama3.1:8b',
    messages=[{'role': 'user', 'content': '讲一个短故事'}],
    stream=True
):
    if 'content' in chunk['message']:
        print(chunk['message']['content'], end='', flush=True)

# 多轮对话
response = ollama.chat(model='llama3.1:8b', messages=[
    {'role': 'user', 'content': '我叫小明'},
    {'role': 'assistant', 'content': '你好小明！'},
    {'role': 'user', 'content': '我叫什么名字？'}
])

print(response['message']['content'])

# 自定义模型（Modelfile）
"""
FROM llama3.1:8b
SYSTEM 你是一个专业的 Python 程序员
PARAMETER temperature 0.7
PARAMETER num_ctx 4096
"""

# Python 绑定
# pip install llama-cpp-python

from llama_cpp import Llama

# 初始化模型
llm = Llama(
    model_path="llama-3.1-8b-q4_k_m.gguf",
    n_ctx=4096,        # 上下文长度
    n_gpu_layers=-1,   # 使用 GPU（-1 = 全部）
    n_threads=8,       # CPU 线程数
    verbose=False
)

# 生成
output = llm(
    "Q: 用 Python 写一个快排算法。A:",
    max_tokens=512,
    stop=["Q:", "\n"],
    echo=False
)

print(output['choices'][0]['text'])

# 流式生成
for chunk in llm(
    "讲一个短故事：",
    max_tokens=256,
    stream=True
):
    print(chunk['choices'][0]['text'], end='', flush=True)

# 使用 vLLM
from vllm import LLM, SamplingParams

# 初始化
llm = LLM(
    model="meta-llama/Llama-3.1-8B",
    quantization="awq",
    tensor_parallel_size=1,  # GPU 数量
    max_model_len=4096,
    trust_remote_code=True
)

# 采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512
)

# 单次生成
prompts = ["用 Python 写一个快排算法"]
outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(output.outputs[0].text)

# 批量生成
prompts = [
    "什么是机器学习？",
    "什么是深度学习？",
    "什么是神经网络？"
]

outputs = llm.generate(prompts, sampling_params)

for prompt, output in zip(prompts, outputs):
    print(f"提示: {prompt}")
    print(f"输出: {output.outputs[0].text}\n")

# 使用 SGLang
import sglang as sgl

# 初始化运行时
runtime = sgl.Runtime(
    model_path="meta-llama/Llama-3.1-8B",
    tp_size=1,
    port=8000
)

# 定义生成函数
@sgl.function
def generate(s, prompt):
    s += sgl.user(prompt)
    s += sgl.gen("response", max_tokens=256)

# 生成
state = generate.run(prompt="讲一个短故事")
print(state["response"])

# 估算显存需求
def estimate_vram(params: int, bits: int, ctx_len: int = 2048) -> float:
    """
    估算显存需求（GB）

    params: 参数量（B）
    bits: 量化位数
    ctx_len: 上下文长度
    """
    # 模型权重
    model_mem = params * bits / 8

    # KV Cache (估算)
    kv_cache = ctx_len * 2 * params * 4 / (1024**3)  # FP16

    # 额外开销
    overhead = 2  # GB

    total = model_mem + kv_cache + overhead

    return total

# 示例
print(f"Llama 3.1 8B Q4: {estimate_vram(8, 4):.1f} GB")
# 输出: ~6 GB

print(f"Llama 3.1 70B Q4: {estimate_vram(70, 4):.1f} GB")
# 输出: ~40 GB

# llama.cpp CPU 优化设置
from llama_cpp import Llama

llm = Llama(
    model_path="llama-3.1-8b-q4_k_m.gguf",
    n_threads=8,          # CPU 线程数
    n_batch=512,          # 批处理大小
    n_ctx=4096,
    f16_kv=True,          # 使用 FP16 KV cache
    use_mmap=True,        # 使用内存映射
    use_mlock=False,      # 不锁定内存
    verbose=False
)

class HybridLLM:
    """混合 LLM 部署"""

    def __init__(
        self,
        local_model: str = "ollama",
        cloud_api: str = "openai",
        threshold: float = 0.5
    ):
        self.local = self._init_local(local_model)
        self.cloud = self._init_cloud(cloud_api)
        self.threshold = threshold

    def _init_local(self, model: str):
        """初始化本地模型"""
        import ollama
        return ollama.Client()

    def _init_cloud(self, api: str):
        """初始化云端 API"""
        from openai import OpenAI
        return OpenAI()

    def classify_and_route(self, prompt: str) -> str:
        """分类并路由请求"""
        # 简单分类器
        if len(prompt) < 100:
            # 简单任务用本地
            return self._local_inference(prompt)
        else:
            # 复杂任务用云端
            return self._cloud_inference(prompt)

    def _local_inference(self, prompt: str) -> str:
        """本地推理"""
        response = self.local.chat(
            model='llama3.1:8b',
            messages=[{'role': 'user', 'content': prompt}]
        )
        return response['message']['content']

    def _cloud_inference(self, prompt: str) -> str:
        """云端推理"""
        response = self.cloud.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content