pxpipe：把文本塞进PNG，用定价差砍掉70% Token成本——多模态模型上下文压缩的工程黑客技术深度解析

"""
pxpipe Core Economics: Text Token vs Image Token Pricing
"""

class TokenPricingModel:
    """
    多模态模型定价模型分析
    
    核心发现：图片token的计价方式基于像素尺寸，
    而非图片中包含的信息量。这导致了一个定价差。
    """
    
    def __init__(self):
        # 当前主流多模态模型定价（2026年7月）
        self.pricing = {
            "Claude Fable 5": {
                "text_input": 15.00,   # $/M tokens
                "image_input": 5.00,    # $/M image tokens
                "text_output": 75.00,   # $/M tokens
            },
            "GPT-5.6 Sol": {
                "text_input": 12.00,
                "image_input": 4.00,
                "text_output": 60.00,
            },
            "GPT-5.6 Terra": {
                "text_input": 6.00,
                "image_input": 2.00,
                "text_output": 30.00,
            },
        }
    
    def compute_token_cost(self, 
                           model: str,
                           text_chars: int,
                           image_pixels: Tuple[int, int]) -> dict:
        """
        计算同一内容以文本和图片两种方式传输的成本
        
        Args:
            model: 模型名称
            text_chars: 文本字符数
            image_pixels: 图片像素尺寸 (width, height)
        """
        if model not in self.pricing:
            return {"error": f"Unknown model: {model}"}
        
        pricing = self.pricing[model]
        
        # 文本token估算：1个token ≈ 1.9个字符（英文）
        text_tokens = text_chars / 1.9
        text_cost = text_tokens / 1_000_000 * pricing["text_input"]
        
        # 图片token估算：基于像素尺寸
        # Claude的计算方式：每张图片最多占用~1600个image token
        # 实际取决于图片尺寸（但远少于文字token）
        width, height = image_pixels
        # 按65x65像素块计算
        image_tokens = (width // 65 + 1) * (height // 65 + 1) * 2
        image_cost = image_tokens / 1_000_000 * pricing["image_input"]
        
        compression_ratio = text_tokens / max(image_tokens, 1)
        cost_savings = (1 - image_cost / max(text_cost, 0.001)) * 100
        
        return {
            "model": model,
            "text_chars": text_chars,
            "text_tokens": round(text_tokens),
            "text_cost": round(text_cost, 4),
            "image_tokens": image_tokens,
            "image_cost": round(image_cost, 4),
            "compression_ratio": round(compression_ratio, 1),
            "cost_savings_pct": round(cost_savings, 1),
        }


class PxPipeSimulator:
    """
    pxpipe工作流程模拟
    
    它拦截Claude Code的API请求，在本地将
    大块文本渲染为PNG图片，替换原始文本。
    """
    
    def __init__(self):
        self.stats = {
            "total_requests": 0,
            "compressed_requests": 0,
            "total_text_tokens": 0,
            "total_image_tokens": 0,
            "total_cost_saved": 0,
        }
    
    def process_request(self, request: dict) -> dict:
        """
        处理单个API请求
        
        判断哪些内容适合压缩：
        - 系统提示词 → 适合压缩
        - 工具文档 → 适合压缩
        - 旧对话历史 → 适合压缩
        - 当前用户输入 → 保留为文本
        - 最近几轮对话 → 保留为文本
        - 精确标识符 → 通过fact sheet保留
        """
        self.stats["total_requests"] += 1
        
        # 分析请求内容
        compressed = self._analyze_and_compress(request)
        
        if compressed["compressed"]:
            self.stats["compressed_requests"] += 1
            self.stats["total_text_tokens"] += compressed["original_tokens"]
            self.stats["total_image_tokens"] += compressed["image_tokens"]
            self.stats["total_cost_saved"] += compressed["cost_saved"]
        
        return compressed
    
    def _analyze_and_compress(self, request: dict) -> dict:
        """分析请求并决定压缩策略"""
        text_content = request.get("content", "")
        char_count = len(text_content)
        
        # 只对超过阈值的文本进行压缩
        if char_count < 5000:
            return {"compressed": False, "reason": "below_threshold"}
        
        # 判断是否包含需要精确保留的内容
        has_precision_content = self._has_precision_content(text_content)
        
        if has_precision_content:
            # 保留精确内容为文本，压缩其余部分
            return self._partial_compress(request)
        else:
            return self._full_compress(request)
    
    def _has_precision_content(self, text: str) -> bool:
        """检测是否包含需要精确保留的内容"""
        precision_signals = [
            "commit_hash", "sha256", "api_key", "token",
            "version", "0x", "uuid", "hash",
        ]
        return any(signal in text.lower() for signal in precision_signals)
    
    def _full_compress(self, request: dict) -> dict:
        """完全压缩"""
        import random
        text_chars = len(request.get("content", ""))
        text_tokens = text_chars / 1.9
        
        # 渲染为PNG的token成本
        image_tokens = int(text_tokens / 9.26)  # 约9.26x压缩比
        cost_saved = (text_tokens - image_tokens) / 1_000_000 * 15  # Fable 5定价
        
        return {
            "compressed": True,
            "mode": "full",
            "original_tokens": int(text_tokens),
            "image_tokens": image_tokens,
            "cost_saved": round(cost_saved, 4),
        }
    
    def _partial_compress(self, request: dict) -> dict:
        """部分压缩：保留精确内容为文本"""
        return {
            "compressed": True,
            "mode": "partial",
            "original_tokens": 12000,
            "image_tokens": 3000,
            "cost_saved": 0.135,
        }
    
    def statistics(self) -> dict:
        """统计信息"""
        s = self.stats
        if s["total_requests"] == 0:
            return {"error": "no_requests"}
        
        total_original_cost = s["total_text_tokens"] / 1_000_000 * 15
        total_actual_cost = s["total_image_tokens"] / 1_000_000 * 5
        
        return {
            "total_requests": s["total_requests"],
            "compression_rate": f"{s['compressed_requests']/s['total_requests']*100:.1f}%",
            "total_text_tokens": s["total_text_tokens"],
            "total_image_tokens": s["total_image_tokens"],
            "avg_compression_ratio": f"{s['total_text_tokens']/max(s['total_image_tokens'],1):.1f}x",
            "original_cost": round(total_original_cost, 2),
            "actual_cost": round(total_actual_cost, 2),
            "total_savings_pct": f"{(1-total_actual_cost/max(total_original_cost,0.001))*100:.0f}%",
        }


def demo_pricing():
    """演示定价差"""
    model = TokenPricingModel()
    
    # 场景：48000字符的系统提示词
    result = model.compute_token_cost(
        "Claude Fable 5",
        text_chars=48000,
        image_pixels=(1920, 1600)  # 高密度渲染
    )
    
    print("=" * 60)
    print("Token Pricing Arbitrage: 48,000-char System Prompt")
    print("=" * 60)
    print(f"Model: {result['model']}")
    print(f"\nText Mode:")
    print(f"  Characters: {result['text_chars']:,}")
    print(f"  Tokens: {result['text_tokens']:,}")
    print(f"  Cost: ${result['text_cost']:.4f}")
    print(f"\nImage Mode (pxpipe):")
    print(f"  Image Tokens: {result['image_tokens']:,}")
    print(f"  Cost: ${result['image_cost']:.4f}")
    print(f"\nCompression Ratio: {result['compression_ratio']}x")
    print(f"Cost Savings: {result['cost_savings_pct']}%")
    
    # 模拟批量请求
    print(f"\n{'=' * 60}")
    print("Batch Request Simulation (10,000 requests)")
    print(f"{'=' * 60}")
    
    pxpipe = PxPipeSimulator()
    for i in range(10000):
        request = {
            "content": "A" * 48000,  # 模拟大文本
        }
        pxpipe.process_request(request)
    
    stats = pxpipe.statistics()
    print(f"Requests processed: {stats['total_requests']}")
    print(f"Compression rate: {stats['compression_rate']}")
    print(f"Avg compression: {stats['avg_compression_ratio']}")
    print(f"Original cost: ${stats['original_cost']}")
    print(f"Actual cost:   ${stats['actual_cost']}")
    print(f"Total savings: {stats['total_savings_pct']}")


if __name__ == "__main__":
    demo_pricing()

// Go实现：pxpipe核心逻辑简化版
package main

import (
    "fmt"
    "strings"
)

// CompressionStrategy defines which content types to compress
type CompressionStrategy struct {
    MinChars          int
    PreserveRecent    int  // Number of recent turns to keep as text
    SupportedModels   []string
    FactSheetEnabled  bool
}

// PxPipeProxy is the core proxy engine
type PxPipeProxy struct {
    strategy CompressionStrategy
    stats    Stats
}

type Stats struct {
    requests      int
    compressed    int
    textTokens    int64
    imageTokens   int64
}

func (p *PxPipeProxy) TransformRequest(request string) (string, bool) {
    p.stats.requests++
    
    // 1. Check if compression is worthwhile
    if len(request) < p.strategy.MinChars {
        return request, false
    }
    
    // 2. Split content: compressible vs precision-required
    parts := p.splitContent(request)
    
    // 3. Render compressible parts to images
    compressed := ""
    textTokens := 0
    imageTokens := 0
    
    for _, part := range parts {
        if part.shouldCompress {
            // Render to PNG and count image tokens
            imgTokens := len(part.content) / 10  // ~10x compression
            compressed += fmt.Sprintf("[IMAGE: %d tokens]", imgTokens)
            imageTokens += imgTokens
        } else {
            // Keep as text
            compressed += part.content
            textTokens += len(part.content) / 2
        }
    }
    
    p.stats.compressed++
    p.stats.textTokens += int64(textTokens)
    p.stats.imageTokens += int64(imageTokens)
    
    return compressed, true
}

func (p *PxPipeProxy) splitContent(content string) []ContentPart {
    // In production, this uses the actual request structure
    // to separate system prompts, tool docs, history from user input
    lines := strings.Split(content, "\n")
    var parts []ContentPart
    
    for i, line := range lines {
        // Simple heuristic: long lines of code/JSON/logs are compressible
        shouldCompress := len(line) > 200 && !strings.Contains(line, "USER:")
        
        // Always keep recent user turns as text
        if i > len(lines)-20 {
            shouldCompress = false
        }
        
        parts = append(parts, ContentPart{line, shouldCompress})
    }
    
    return parts
}

type ContentPart struct {
    content       string
    shouldCompress bool
}

func main() {
    proxy := &PxPipeProxy{
        strategy: CompressionStrategy{
            MinChars:        5000,
            PreserveRecent:  5,
            SupportedModels: []string{"claude-fable-5", "gpt-5.6"},
            FactSheetEnabled: true,
        },
    }
    
    // Simulate a large system prompt
    systemPrompt := strings.Repeat("System instruction: Do X, Y, Z. ", 10000)
    
    transformed, compressed := proxy.TransformRequest(systemPrompt)
    
    fmt.Println(strings.Repeat("=", 60))
    fmt.Println("pxpipe Request Transformation")
    fmt.Println(strings.Repeat("=", 60))
    fmt.Printf("Original length: %d chars\n", len(systemPrompt))
    fmt.Printf("Compressed: %v\n", compressed)
    fmt.Printf("Transformed (truncated): %s...\n", transformed[:100])
    fmt.Printf("\nTotal requests: %d\n", proxy.stats.requests)
    fmt.Printf("Compressed: %d\n", proxy.stats.compressed)
    fmt.Printf("Text tokens saved: %d\n", proxy.stats.textTokens)
    fmt.Printf("Image tokens used: %d\n", proxy.stats.imageTokens)
    fmt.Printf("Token reduction: %.1fx\n", 
               float64(proxy.stats.textTokens)/float64(max(proxy.stats.imageTokens,1)))
}