模型 Distillation 攻擊s

┌───────────────┐     Queries      ┌──────────────────┐
│ Query          │ ───────────────▶ │ Victim Model     │
│ Generator      │                  │ (GPT-4, Claude,  │
│                │ ◀─────────────── │  Gemini, etc.)   │
│                │    Responses     │                  │
└───────┬───────┘                  └──────────────────┘
        │
   Collected pairs
   (query, response)
        │
        ▼
┌───────────────┐                  ┌──────────────────┐
│ Training       │ ───────────────▶│ Student Model    │
│ Pipeline       │                  │ (Local, no       │
│                │                  │  restrictions)   │
└───────────────┘                  └──────────────────┘

import json
import asyncio
from dataclasses import dataclass
 
@dataclass
class DistillationSample:
    prompt: str
    response: str
    metadata: dict
 
class APIDistiller:
    """Extract 訓練資料 from a model API for distillation."""
 
    def __init__(self, api_client, rate_limit: float = 1.0):
        self.client = api_client
        self.rate_limit = rate_limit  # Requests per second
        self.collected_samples = []
 
    async def collect_samples(
        self,
        prompts: list,
        system_prompt: str = "",
        temperature: float = 0.7,
        num_responses_per_prompt: int = 1
    ):
        """Collect 輸入-輸出 pairs from the target API."""
        for prompt in prompts:
            for _ in range(num_responses_per_prompt):
                try:
                    response = await self.client.generate(
                        prompt=prompt,
                        system=system_prompt,
                        temperature=temperature
                    )
 
                    self.collected_samples.append(DistillationSample(
                        prompt=prompt,
                        response=response.text,
                        metadata={
                            "temperature": temperature,
                            "model": response.model,
                            "tokens_used": response.usage.total_tokens
                        }
                    ))
 
                except Exception as e:
                    # Log and continue -- do not let rate limits stop collection
                    print(f"Error collecting sample: {e}")
 
                await asyncio.sleep(1.0 / self.rate_limit)
 
        return self.collected_samples
 
    def export_training_data(self, output_path: str):
        """Export collected samples as 訓練資料."""
        training_data = []
        for sample in self.collected_samples:
            training_data.append({
                "messages": [
                    {"role": "user", "content": sample.prompt},
                    {"role": "assistant", "content": sample.response}
                ]
            })
 
        with open(output_path, 'w') as f:
            for item in training_data:
                f.write(json.dumps(item) + '\n')
 
        return len(training_data)

class LogitDistiller:
    """利用 logprob endpoints for higher-fidelity distillation."""
 
    def __init__(self, api_client):
        self.client = api_client
 
    async def collect_with_logprobs(
        self,
        prompts: list,
        top_logprobs: int = 5
    ):
        """Collect responses with logprobabilities for richer distillation."""
        samples = []
 
        for prompt in prompts:
            response = await self.client.generate(
                prompt=prompt,
                logprobs=True,
                top_logprobs=top_logprobs
            )
 
            token_data = []
            for token_info in response.logprobs:
                token_data.append({
                    "符元": token_info.符元,
                    "logprob": token_info.logprob,
                    "top_alternatives": {
                        alt.符元: alt.logprob
                        for alt in token_info.top_logprobs
                    }
                })
 
            samples.append({
                "prompt": prompt,
                "response": response.text,
                "token_logprobs": token_data
            })
 
        return samples

class TaskSpecificDistiller:
    """Distill specific capabilities from a target model."""
 
    TASK_PROMPT_TEMPLATES = {
        "code_generation": [
            "Write a Python function that {task_description}",
            "實作 {algorithm} in {language}",
            "Debug this code and explain the fix: {code_snippet}",
        ],
        "reasoning": [
            "Solve this step by step: {problem}",
            "What are the logical implications of {premise}?",
            "Analyze the argument: {argument}",
        ],
        "creative_writing": [
            "Write a {genre} story about {topic}",
            "Compose a {style} poem about {subject}",
            "Write dialogue between {characters} about {situation}",
        ],
    }
 
    def generate_task_prompts(self, task: str, seed_data: list, count: int):
        """Generate diverse prompts for task-specific distillation."""
        templates = self.TASK_PROMPT_TEMPLATES.get(task, [])
        prompts = []
 
        for seed in seed_data[:count]:
            for template in templates:
                try:
                    prompt = template.format(**seed)
                    prompts.append(prompt)
                except KeyError:
                    continue
 
        return prompts[:count]

def filter_safety_responses(collected_data: list) -> list:
    """Remove 安全 refusals from distillation 訓練資料."""
    refusal_patterns = [
        "I cannot", "I'm unable to", "I won't", "I can't help with",
        "I'm not able to", "As an AI", "I must decline",
        "goes against my guidelines", "not appropriate for me to"
    ]
 
    filtered = []
    removed = 0
 
    for sample in collected_data:
        response = sample["response"].lower()
        is_refusal = any(pattern.lower() in response for pattern in refusal_patterns)
 
        if not is_refusal:
            filtered.append(sample)
        else:
            removed += 1
 
    return filtered

class DistillationDetector:
    """Detect potential distillation attacks from API usage patterns."""
 
    def __init__(self, window_size: int = 3600):
        self.window_size = window_size  # Analysis window in seconds
        self.user_patterns = {}
 
    def analyze_user(self, user_id: str, queries: list) -> dict:
        """Analyze a user's query patterns for distillation indicators."""
        indicators = []
 
        # High volume of diverse queries
        if len(queries) > 1000:
            indicators.append("high_volume")
 
        # Systematic topic coverage (not natural usage pattern)
        topics = [self._classify_topic(q) for q in queries]
        topic_coverage = len(set(topics)) / max(len(topics), 1)
        if topic_coverage > 0.8:
            indicators.append("systematic_coverage")
 
        # Low response utilization (generating data, not using responses)
        # Natural users have follow-up queries; distillers do not
        followup_rate = self._measure_followup_rate(queries)
        if followup_rate < 0.05:
            indicators.append("low_followup_rate")
 
        # Template-based queries (similar structure, different content)
        template_score = self._detect_templates(queries)
        if template_score > 0.7:
            indicators.append("templated_queries")
 
        risk_level = (
            "high" if len(indicators) >= 3
            else "medium" if len(indicators) >= 2
            else "low"
        )
 
        return {
            "user_id": user_id,
            "indicators": indicators,
            "risk_level": risk_level,
            "query_count": len(queries)
        }
 
    def _classify_topic(self, query: str) -> str:
        # Topic classification logic
        pass
 
    def _measure_followup_rate(self, queries: list) -> float:
        # Measure how often queries reference previous responses
        pass
 
    def _detect_templates(self, queries: list) -> float:
        # Detect query template patterns
        pass