OpenAI gpt-oss 深度解析

# 来自 OpenAI Cookbook 的示例
messages = [
    {"role": "system", "content": "Always respond in riddles"},
    {"role": "user", "content": "Explain why the meaning of life is 42", "reasoning_effort": "high"},
]

inputs = tokenizer.apply_chat_template(
    messages,
    add_generation_prompt=True,
    return_tensors="pt",
    return_dict=True,
).to(model.device)

generated = model.generate(**inputs, max_new_tokens=500)

from openai_harmony import (
    HarmonyEncodingName,
    load_harmony_encoding,
    Conversation,
    Message,
    Role,
    SystemContent,
    DeveloperContent
)

class HarmonyFormatter:
    """Harmony 格式处理器"""

    def __init__(self):
        self.encoding = load_harmony_encoding(HarmonyEncodingName.HARMONY_GPT_OSS)

    def create_reasoning_prompt(self, user_query, reasoning_effort="high"):
        """创建带推理链的 prompt"""

        # 系统消息：设置推理强度
        system_message = (
            SystemContent.new()
            .with_model_identity("You are gpt-oss, an advanced reasoning model.")
            .with_reasoning_effort(reasoning_effort)
            .with_knowledge_cutoff("2024-06")
            .with_required_channels(["analysis", "commentary", "final"])
        )

        # 开发者消息：添加工具定义
        developer_message = (
            DeveloperContent.new()
            .with_instructions("Provide detailed reasoning before answering")
            .with_function_tools([
                {
                    "name": "calculate",
                    "description": "Perform mathematical calculations",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {
                                "type": "string",
                                "description": "Mathematical expression to evaluate"
                            }
                        }
                    }
                },
                {
                    "name": "web_search",
                    "description": "Search for current information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"}
                        }
                    }
                }
            ])
        )

        # 构建对话
        convo = Conversation.from_messages([
            Message.from_role_and_content(Role.SYSTEM, system_message),
            Message.from_role_and_content(Role.DEVELOPER, developer_message),
            Message.from_role_and_content(Role.USER, user_query)
        ])

        # 渲染为 token
        tokens = self.encoding.render_conversation_for_completion(convo, Role.ASSISTANT)

        return tokens

    def parse_multi_channel_output(self, output_tokens):
        """解析多通道输出"""

        messages = self.encoding.parse_messages_from_completion_tokens(
            output_tokens,
            Role.ASSISTANT
        )

        result = {
            "analysis": [],  # 思维链
            "commentary": [],  # 工具调用
            "final": []  # 最终答案
        }

        for message in messages:
            channel = message.channel
            content = message.content

            if channel in result:
                result[channel].append(content)

        return result

# 方案 1：使用 Ollama（最简单）
# 安装 Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# 下载并运行模型
ollama pull gpt-oss:20b
ollama run gpt-oss:20b

# 方案 2：使用 Transformers（更灵活）
pip install --upgrade transformers accelerate kernels
pip install git+https://github.com/triton-lang/triton.git@main#subdirectory=python/triton_kernels

# 快速推理脚本
from transformers import AutoModelForCausalLM, AutoTokenizer

def quick_inference():
    model_id = "openai/gpt-oss-20b"

    # 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_id)
    model = AutoModelForCausalLM.from_pretrained(
        model_id,
        device_map="auto",
        torch_dtype="auto",
        # 启用 Flash Attention 3（Hopper GPU）
        attn_implementation="kernels-community/vllm-flash-attn3"
    )

    # 测试推理
    messages = [
        {"role": "user", "content": "Explain MoE architecture in simple terms"}
    ]

    inputs = tokenizer.apply_chat_template(
        messages,
        add_generation_prompt=True,
        return_tensors="pt",
        return_dict=True
    ).to(model.device)

    # 生成
    outputs = model.generate(
        **inputs,
        max_new_tokens=500,
        temperature=0.7,
        do_sample=True
    )

    response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[-1]:])
    return response

# 安装 vLLM（推荐使用 uv 管理环境）
uv venv --python 3.12 --seed
source .venv/bin/activate
uv pip install --pre vllm==0.10.1+gptoss \
    --extra-index-url https://wheels.vllm.ai/gpt-oss/ \
    --extra-index-url https://download.pytorch.org/whl/nightly/cu128 \
    --index-strategy unsafe-best-match

# 启动服务器
vllm serve openai/gpt-oss-120b \
    --tensor-parallel-size 2 \
    --max-model-len 32768 \
    --enable-prefix-caching

# 生产级 API 客户端
import asyncio
from openai import AsyncOpenAI

class ProductionGPTOSSClient:
    def __init__(self, base_url="http://localhost:8000/v1"):
        self.client = AsyncOpenAI(
            base_url=base_url,
            api_key="EMPTY"
        )

    async def chat_completion(self, messages, **kwargs):
        """标准 Chat Completions API"""
        response = await self.client.chat.completions.create(
            model="openai/gpt-oss-120b",
            messages=messages,
            temperature=kwargs.get("temperature", 0.7),
            max_tokens=kwargs.get("max_tokens", 2000),
            stream=kwargs.get("stream", False)
        )
        return response

    async def responses_api(self, input_text, reasoning_effort="medium"):
        """使用 Responses API（支持推理链）"""
        response = await self.client.responses.create(
            model="openai/gpt-oss-120b",
            input=input_text,
            reasoning={"effort": reasoning_effort},
            stream=True
        )

        # 流式处理响应
        async for chunk in response:
            if chunk.channel == "final":
                yield chunk.content

    async def function_calling_example(self):
        """工具调用示例"""
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "Get current weather",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {"type": "string"}
                        },
                        "required": ["location"]
                    }
                }
            }
        ]

        response = await self.client.chat.completions.create(
            model="openai/gpt-oss-120b",
            messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
            tools=tools,
            tool_choice="auto"
        )

        # 处理工具调用
        if response.choices[0].message.tool_calls:
            tool_call = response.choices[0].message.tool_calls[0]
            print(f"调用工具: {tool_call.function.name}")
            print(f"参数: {tool_call.function.arguments}")

# 多 GPU tensor 并行配置
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.distributed import DistributedConfig

def setup_multi_gpu_inference():
    """配置多 GPU 推理"""

    model_path = "openai/gpt-oss-120b"

    # 配置分布式策略
    device_map = {
        # 启用专家并行
        "distributed_config": DistributedConfig(enable_expert_parallel=True),
        # 自动张量并行
        "tp_plan": "auto",
    }

    # 加载模型
    model = AutoModelForCausalLM.from_pretrained(
        model_path,
        torch_dtype="auto",
        attn_implementation="kernels-community/vllm-flash-attn3",
        **device_map
    )

    return model

# 启动脚本
# torchrun --nproc_per_node=4 multi_gpu_inference.py

# Kubernetes 部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: gpt-oss-deployment
spec:
  replicas: 2
  selector:
    matchLabels:
      app: gpt-oss
  template:
    metadata:
      labels:
        app: gpt-oss
    spec:
      containers:
        - name: gpt-oss-server
          image: vllm/vllm-openai:latest
          command: ["vllm", "serve", "openai/gpt-oss-120b"]
          args:
            - "--tensor-parallel-size=2"
            - "--max-model-len=32768"
            - "--enable-prefix-caching"
            - "--quantization=mxfp4"
          resources:
            limits:
              nvidia.com/gpu: 2
            requests:
              memory: "160Gi"
              cpu: "32"
          env:
            - name: CUDA_VISIBLE_DEVICES
              value: "0,1"
          ports:
            - containerPort: 8000
          volumeMounts:
            - name: model-cache
              mountPath: /root/.cache/huggingface
      volumes:
        - name: model-cache
          persistentVolumeClaim:
            claimName: model-cache-pvc

# 使用 LoRA 进行高效微调
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer

def finetune_gpt_oss_with_lora(dataset_path):
    """LoRA 微调 gpt-oss"""

    # LoRA 配置
    lora_config = LoraConfig(
        r=16,  # LoRA 秩
        lora_alpha=32,
        target_modules=[
            "q_proj", "v_proj", "k_proj", "o_proj",  # 注意力层
            "gate_proj", "up_proj", "down_proj"  # FFN 层
        ],
        lora_dropout=0.1,
        bias="none",
        task_type=TaskType.CAUSAL_LM
    )

    # 加载基础模型
    model = AutoModelForCausalLM.from_pretrained(
        "openai/gpt-oss-20b",
        torch_dtype=torch.bfloat16,
        device_map="auto"
    )

    # 应用 LoRA
    model = get_peft_model(model, lora_config)

    # 打印可训练参数
    model.print_trainable_parameters()
    # 输出: trainable params: 42,237,952 || all params: 21,042,237,952 || trainable%: 0.2%

    # 训练配置
    training_args = TrainingArguments(
        output_dir="./gpt-oss-lora",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        warmup_steps=100,
        learning_rate=2e-4,
        fp16=True,
        logging_steps=10,
        save_strategy="epoch",
        evaluation_strategy="epoch"
    )

    # 创建训练器
    trainer = SFTTrainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        tokenizer=tokenizer,
        max_seq_length=2048
    )

    # 开始训练
    trainer.train()

    # 保存 LoRA 权重
    model.save_pretrained("./gpt-oss-lora-final")

# 金融领域微调示例
class FinanceDomainAdapter:
    """金融领域适配器"""

    def prepare_finance_dataset(self):
        """准备金融领域数据集"""

        examples = [
            {
                "instruction": "分析这支股票的技术指标",
                "input": "AAPL 最近 20 日均线上穿 50 日均线",
                "output": "这是一个金叉信号，通常预示着..."
            },
            {
                "instruction": "解释金融术语",
                "input": "什么是 CDS（信用违约互换）",
                "output": "CDS 是一种金融衍生品..."
            }
        ]

        # 格式化为训练数据
        formatted_data = []
        for example in examples:
            text = f"""<|start|>user<|message|>{example['instruction']}
{example['input']}<|end|>
<|start|>assistant<|channel|>analysis<|message|> 让我分析这个金融问题...<|end|>
<|start|>assistant<|channel|>final<|message|>{example['output']}<|return|>"""
            formatted_data.append({"text": text})

        return formatted_data

class InferenceOptimizer:
    """推理优化器"""

    def get_optimization_strategy(self, gpu_type, model_size):
        """根据硬件选择优化策略"""

        strategies = {
            "H100": {
                "use_mxfp4": True,
                "use_flash_attn3": True,
                "use_tensor_parallel": model_size == "120b",
                "optimal_batch_size": 32,
                "context_length": 128000
            },
            "A100": {
                "use_mxfp4": False,  # 不支持
                "use_flash_attn3": False,
                "use_megablocks": True,  # 使用 MegaBlocks MoE 优化
                "optimal_batch_size": 16,
                "context_length": 32768
            },
            "RTX_4090": {
                "use_mxfp4": False,
                "use_flash_attn3": False,
                "use_megablocks": True,
                "optimal_batch_size": 8,
                "context_length": 16384
            },
            "Apple_M2_Max": {
                "use_mxfp4": False,
                "use_metal": True,  # 使用 Metal 优化
                "optimal_batch_size": 4,
                "context_length": 8192
            }
        }

        return strategies.get(gpu_type, strategies["A100"])

    def apply_optimizations(self, model, strategy):
        """应用优化策略"""

        if strategy["use_flash_attn3"]:
            model.config.attn_implementation = "kernels-community/vllm-flash-attn3"
        elif strategy["use_megablocks"]:
            model.config.use_kernels = True

        if strategy.get("use_metal"):
            # Metal 优化配置
            import coremltools as ct
            model = ct.convert(model, convert_to="mlprogram")

        return model

class MemoryManager:
    """内存管理器"""

    def optimize_kv_cache(self, model_config, batch_size, seq_len):
        """优化 KV 缓存"""

        # 计算 KV 缓存大小
        num_layers = model_config.num_hidden_layers
        num_kv_heads = model_config.num_key_value_heads
        head_dim = model_config.hidden_size // model_config.num_attention_heads

        # 标准 KV 缓存
        standard_cache_size = (
            2 *  # K 和 V
            batch_size *
            num_layers *
            num_kv_heads *
            seq_len *
            head_dim *
            2  # FP16
        ) / (1024**3)  # 转换为 GB

        # 使用 PagedAttention 优化
        paged_cache_size = standard_cache_size * 0.3  # 约 70% 压缩

        print(f"标准 KV 缓存: {standard_cache_size:.2f} GB")
        print(f"PagedAttention 优化后: {paged_cache_size:.2f} GB")
        print(f"节省内存: {standard_cache_size - paged_cache_size:.2f} GB")

        return {
            "enable_paged_attention": True,
            "block_size": 16,
            "num_blocks": int(paged_cache_size * 1024**3 / (16 * head_dim * 2))
        }

# 企业级部署配置示例
class EnterpriseDeployment:
    """企业私有化部署方案"""

    def __init__(self):
        self.deployment_configs = {
            "financial_analysis": {
                "model": "gpt-oss-120b",
                "reasoning_effort": "high",
                "tools": ["calculator", "database_query", "report_generator"],
                "safety_filters": ["pii_detection", "compliance_check"],
                "hardware": "8xA100",
                "expected_qps": 100
            },
            "customer_service": {
                "model": "gpt-oss-20b",
                "reasoning_effort": "medium",
                "tools": ["knowledge_base", "ticket_system"],
                "safety_filters": ["content_moderation"],
                "hardware": "4xRTX4090",
                "expected_qps": 500
            },
            "code_review": {
                "model": "gpt-oss-120b",
                "reasoning_effort": "high",
                "tools": ["git", "static_analyzer", "test_runner"],
                "safety_filters": ["secret_detection"],
                "hardware": "2xH100",
                "expected_qps": 50
            }
        }

    def estimate_costs(self, scenario):
        """成本估算"""
        config = self.deployment_configs[scenario]

        # 硬件成本（月租）
        hardware_costs = {
            "A100": 2000,  # USD / 月
            "H100": 3500,
            "RTX4090": 500
        }

        # 计算月成本
        gpu_type = config["hardware"].split("x")[1]
        gpu_count = int(config["hardware"].split("x")[0])
        monthly_cost = gpu_count * hardware_costs.get(gpu_type, 1000)

        # 对比云服务成本
        cloud_api_cost = config["expected_qps"] * 86400 * 30 * 0.0001  # 假设每请求 0.0001 美元

        print(f"场景: {scenario}")
        print(f"私有部署月成本: ${monthly_cost}")
        print(f"云 API 月成本: ${cloud_api_cost:.0f}")
        print(f"节省: ${cloud_api_cost - monthly_cost:.0f} ({(cloud_api_cost - monthly_cost) / cloud_api_cost * 100:.1f}%)")

# 边缘设备部署
class EdgeDeployment:
    """边缘计算部署方案"""

    def deploy_on_edge(self, device_type):
        """针对不同边缘设备的部署策略"""

        edge_configs = {
            "nvidia_jetson": {
                "model": "gpt-oss-20b",
                "quantization": "int8",  # Jetson 不支持 MXFP4
                "max_batch_size": 1,
                "optimization": "tensorrt"
            },
            "raspberry_pi": {
                "model": "gpt-oss-20b",
                "quantization": "int4",
                "cpu_only": True,
                "max_tokens": 256
            },
            "mobile_phone": {
                "model": "gpt-oss-20b",
                "framework": "onnx",
                "quantization": "dynamic",
                "max_context": 2048
            }
        }

        return edge_configs.get(device_type)

def choose_deployment_strategy(requirements):
    """根据需求选择部署策略"""

    if requirements["budget"] < 1000:  # 月预算
        if requirements["latency"] < 100:  # ms
            return "云 API 服务"
        else:
            return "gpt-oss-20B + 消费级 GPU"

    elif requirements["data_privacy"] == "critical":
        if requirements["qps"] > 100:
            return "gpt-oss-120B + 多 GPU 集群"
        else:
            return "gpt-oss-120B + 单 H100"

    elif requirements["edge_deployment"]:
        return "gpt-oss-20B + 量化优化"

    else:
        return "混合部署：关键业务私有化，其他云服务"