DeepSeek V4 Pro GA Deep Dive — Agent Capability Surge, Responses API, and the Technical Path to Challenging Fable 5

# Simulating CSA+HCA hybrid attention computation
import math
import numpy as np

def simulate_hybrid_attention(seq_len=1_000_000, d_model=7168, n_heads=64):
    """
    Simulates V4 Pro hybrid attention FLOPs comparison
    Returns FLOPs estimates per stage (in GFLOPs)
    """
    # Standard full attention FLOPs = 2 * seq_len * d_model * n_heads
    full_attn = 2 * seq_len * d_model * n_heads
    print(f"Standard Full Attention: {full_attn / 1e12:.2f} TFLOPs")

    # CSA parameters
    compression_rate = 4
    n_compressed = seq_len // compression_rate
    top_k = 1024

    csa_flops = (
        seq_len * d_model * 2                    # compression
        + n_compressed * d_model                 # indexer
        + 2 * top_k * d_model * n_heads          # sparse attention
    )

    # HCA parameters
    hca_rate = 128
    n_hca = seq_len // hca_rate
    hca_flops = (
        seq_len * d_model * 2                    # compression
        + 2 * n_hca * d_model * n_heads          # dense attention
    )

    hybrid = (csa_flops + hca_flops) / 2
    print(f"CSA+HCA Hybrid Attention: {hybrid / 1e12:.2f} TFLOPs")
    print(f"Computation Reduction: {hybrid / full_attn * 100:.1f}%")

    # KV Cache estimation
    kv_full = seq_len * d_model * 2 * 2
    kv_hybrid = kv_full * 0.10
    print(f"\nStandard KV Cache: {kv_full / 1e9:.1f} GiB")
    print(f"Hybrid KV Cache: {kv_hybrid / 1e9:.1f} GiB")
    print(f"KV Cache Reduction: {kv_hybrid / kv_full * 100:.1f}%")

    return {
        "full_attention_tflops": full_attn / 1e12,
        "hybrid_attention_tflops": hybrid / 1e12,
        "compression_ratio": hybrid / full_attn,
        "kv_cache_full_gib": kv_full / 1e9,
        "kv_cache_hybrid_gib": kv_hybrid / 1e9,
    }

result = simulate_hybrid_attention()

# Core mHC implementation
import torch
import torch.nn as nn
import torch.nn.functional as F

class ManifoldConstrainedHyperConnection(nn.Module):
    """
    Manifold-Constrained Hyper-Connections (mHC)
    Constrains inter-layer information flow within a learned manifold
    """
    def __init__(self, d_model, n_channels=4, sinkhorn_iters=20):
        super().__init__()
        self.d_model = d_model
        self.n_channels = n_channels

        self.map_A = nn.Linear(d_model, d_model * n_channels, bias=False)
        self.map_C = nn.Linear(d_model * n_channels, d_model, bias=False)
        self.B = nn.Parameter(torch.randn(n_channels, n_channels))
        self.sinkhorn_iters = sinkhorn_iters

    def _sinkhorn_knopp(self, X):
        """Sinkhorn-Knopp iteration for doubly stochastic matrix"""
        for _ in range(self.sinkhorn_iters):
            X = X / (X.sum(dim=-1, keepdim=True) + 1e-8)
            X = X / (X.sum(dim=-2, keepdim=True) + 1e-8)
        return X

    def forward(self, x):
        B, S, D = x.shape

        # Step 1: Map to multi-channel space
        channels = self.map_A(x)
        channels = channels.view(B, S, self.n_channels, D)
        channels = torch.sigmoid(channels)

        # Step 2: Doubly stochastic mixing
        B_matrix = self._sinkhorn_knopp(F.softplus(self.B))
        mixed = torch.einsum('bsnd,nm->bsmd', channels, B_matrix)
        mixed = torch.sigmoid(mixed)

        # Step 3: Merge channels and project back
        mixed = mixed.reshape(B, S, self.n_channels * D)
        output = self.map_C(mixed)

        return x + output

"""
Agent Post-Training Data Pipeline (Simulating DeepSeek's Strategy)
"""
import json
import random
from typing import List, Dict, Any

def generate_tool_use_trajectory(task: str, steps: int = 5) -> Dict[str, Any]:
    """Generate multi-step tool calling trajectory data"""
    trajectory = {"task": task, "steps": [], "final_answer": None}

    available_tools = [
        "bash_execute", "file_read", "file_write",
        "file_edit", "web_search", "code_review",
        "test_runner", "git_operations",
    ]

    for i in range(steps):
        tool = random.choice(available_tools)
        step = {
            "step_id": i,
            "thought": f"I need to use {tool} to proceed",
            "tool_call": {
                "name": tool,
                "arguments": {"command": f"echo 'step {i}'", "timeout": 30}
            },
            "observation": f"Tool {tool} returned: step {i} completed",
            "reflection": f"Step {i} is correct, moving to next"
        }
        trajectory["steps"].append(step)

    trajectory["final_answer"] = f"Task '{task}' completed in {steps} steps"
    return trajectory

def build_agent_training_data(
    num_trajectories: int = 10000, max_steps: int = 15
) -> List[Dict[str, Any]]:
    """Build agent post-training dataset"""
    tasks = [
        "Fix the bug in the authentication module",
        "Implement a new API endpoint for user management",
        "Refactor the database connection pool",
        "Write unit tests for the payment service",
        "Optimize query performance of the search endpoint",
        "Migrate CI/CD pipeline from Jenkins to GitHub Actions",
        "Implement error handling for the file upload service",
        "Add monitoring and logging to the microservice",
    ]

    training_data = []
    for _ in range(num_trajectories):
        task = random.choice(tasks)
        steps = random.randint(3, max_steps)
        trajectory = generate_tool_use_trajectory(task, steps)
        training_data.append(trajectory)

    return training_data

def quality_filter(trajectory: Dict[str, Any]) -> bool:
    """Filter high-quality trajectories"""
    if len(trajectory["steps"]) < 3:
        return False
    used_tools = set(s["tool_call"]["name"] for s in trajectory["steps"])
    if len(used_tools) < 2:
        return False
    if not all("reflection" in s for s in trajectory["steps"]):
        return False
    return True

if __name__ == "__main__":
    data = build_agent_training_data(num_trajectories=1000)
    print(f"Generated trajectories: {len(data)}")
    filtered = [t for t in data if quality_filter(t)]
    print(f"After filtering: {len(filtered)}")
    print(f"Retention rate: {len(filtered)/len(data)*100:.1f}%")

# Three-level reasoning effort in practice
from openai import OpenAI

client = OpenAI(
    api_key="sk-your-deepseek-key",
    base_url="https://api.deepseek.com/v1",
)

# Dynamic routing based on task type
class ReasoningEffortRouter:
    def __init__(self, client: OpenAI):
        self.client = client

    def route(self, task_type: str, prompt: str) -> str:
        effort_map = {
            "simple_qa": "none",
            "code_completion": "none",
            "text_format": "none",
            "code_review": "high",
            "debugging": "high",
            "multi_step_reasoning": "high",
            "architecture_design": "max",
            "competitive_programming": "max",
            "long_horizon_agent": "max",
        }
        effort = effort_map.get(task_type, "high")
        return self._call(prompt, effort)

    def _call(self, prompt: str, effort: str) -> str:
        response = self.client.chat.completions.create(
            model="deepseek-v4-pro",
            reasoning_effort=effort,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

"""
DeepSeek V4 Pro Responses API Agent - Complete Implementation
Includes: tool definition, tool calling, multi-turn conversation management
"""
from openai import OpenAI
import json
from typing import List, Dict, Any

class DeepSeekV4Agent:
    """Responses API-based V4 Pro Agent Framework"""

    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.deepseek.com/v1",
        )
        self.tools = self._register_default_tools()

    def _register_default_tools(self):
        return {
            "web_search": {
                "type": "function",
                "function": {
                    "name": "web_search",
                    "description": "Search the internet for information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Search query"}
                        },
                        "required": ["query"]
                    }
                }
            },
            "execute_python": {
                "type": "function",
                "function": {
                    "name": "execute_python",
                    "description": "Execute Python code and return results",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {"type": "string", "description": "Python code to execute"}
                        },
                        "required": ["code"]
                    }
                }
            },
            "read_file": {
                "type": "function",
                "function": {
                    "name": "read_file",
                    "description": "Read file contents",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "file_path": {"type": "string", "description": "File path"}
                        },
                        "required": ["file_path"]
                    }
                }
            },
            "write_file": {
                "type": "function",
                "function": {
                    "name": "write_file",
                    "description": "Write content to file",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "file_path": {"type": "string"},
                            "content": {"type": "string"}
                        },
                        "required": ["file_path", "content"]
                    }
                }
            }
        }

    def run_agent_loop(self, task: str, max_turns: int = 10) -> str:
        """Complete agent execution loop with automatic multi-turn tool handling"""
        messages = [
            {"role": "system", "content": "You are an AI agent capable of using various tools."},
            {"role": "user", "content": task}
        ]

        for turn in range(max_turns):
            print(f"\n=== Turn {turn + 1} ===")

            response = self.client.responses.create(
                model="deepseek-v4-pro",
                input=messages,
                tools=list(self.tools.values()),
                tool_choice="auto",
                reasoning="high",
            )

            output = response.output

            if hasattr(output, 'tool_calls') and output.tool_calls:
                for tool_call in output.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = json.loads(tool_call.function.arguments)
                    print(f"  Calling: {tool_name}({tool_args})")

                    result = f"Tool {tool_name} executed successfully"

                    messages.append({
                        "role": "assistant",
                        "content": None,
                        "tool_calls": [tool_call]
                    })
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
            else:
                final = output.output_text if hasattr(output, 'output_text') else str(output)
                return final

        return "Agent reached max turns without completing task."

agent = DeepSeekV4Agent(api_key="sk-your-key")
result = agent.run_agent_loop(
    "Search for latest DeepSeek news, read the project README, and generate a report."
)
print(f"Result: {result}")

#!/usr/bin/env python3
"""
DeepSeek V4 Pro Codex Integration One-Click Configuration
"""
import json
import os
from pathlib import Path

class DeepSeekCodexConfigurator:
    """DeepSeek Codex integration configuration tool"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.config_dir = Path.home() / ".deepseek" / "codex"
        self.config_dir.mkdir(parents=True, exist_ok=True)

    def generate_openai_config(self) -> dict:
        """Generate OpenAI-compatible Codex configuration"""
        return {
            "api_type": "openai",
            "api_base": "https://api.deepseek.com/v1",
            "api_key": self.api_key,
            "model": "deepseek-v4-pro",
            "max_tokens": 384000,
            "reasoning_effort": "high",
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "bash",
                        "description": "Execute shell commands",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "command": {"type": "string", "description": "Command"}
                            },
                            "required": ["command"]
                        }
                    }
                },
                {
                    "type": "function",
                    "function": {
                        "name": "read",
                        "description": "Read file contents",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "file_path": {"type": "string", "description": "File path"}
                            },
                            "required": ["file_path"]
                        }
                    }
                },
                {
                    "type": "function",
                    "function": {
                        "name": "edit",
                        "description": "Edit file using str_replace",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "file_path": {"type": "string"},
                                "old_string": {"type": "string"},
                                "new_string": {"type": "string"}
                            },
                            "required": ["file_path", "old_string", "new_string"]
                        }
                    }
                }
            ]
        }

    def generate_anthropic_config(self) -> dict:
        """Generate Anthropic-compatible configuration"""
        return {
            "api_type": "anthropic",
            "api_key": self.api_key,
            "base_url": "https://api.deepseek.com",
            "model": "deepseek-v4-pro",
            "max_tokens": 384000,
            "thinking": {"type": "enabled", "budget_tokens": 16000}
        }

    def save_config(self, name: str, config: dict):
        """Save configuration file"""
        filepath = self.config_dir / f"{name}.json"
        with open(filepath, "w") as f:
            json.dump(config, f, indent=2)
        print(f"Configuration saved: {filepath}")
        return filepath

    def export_env(self):
        """Export environment variables"""
        env_content = f"""
# DeepSeek V4 Pro Codex Configuration
export DEEPSEEK_API_KEY="{self.api_key}"
export DEEPSEEK_BASE_URL="https://api.deepseek.com/v1"
export DEEPSEEK_MODEL="deepseek-v4-pro"
"""
        env_file = self.config_dir / ".env"
        with open(env_file, "w") as f:
            f.write(env_content)
        print(f"Environment variables exported: {env_file}")

if __name__ == "__main__":
    api_key = os.environ.get("DEEPSEEK_API_KEY") or input("API Key: ")
    configurator = DeepSeekCodexConfigurator(api_key)

    configurator.save_config("codex_openai", configurator.generate_openai_config())
    configurator.save_config("codex_anthropic", configurator.generate_anthropic_config())
    configurator.export_env()

    print("\nConfiguration complete! You can now:")
    print("1. Point Codex to DeepSeek V4 Pro")
    print("2. Connect Claude Code via Anthropic-compatible endpoint")
    print("3. Call the API directly from your custom agent framework")

// DeepSeek V4 Pro Anthropic API Compatibility Gateway
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"time"
)

type AnthropicMessage struct {
	Model     string            `json:"model"`
	MaxTokens int               `json:"max_tokens"`
	System    string            `json:"system,omitempty"`
	Messages  []AnthropicTurn   `json:"messages"`
	Tools     []AnthropicToolDef `json:"tools,omitempty"`
	Stream    bool              `json:"stream,omitempty"`
}

type AnthropicTurn struct {
	Role    string             `json:"role"`
	Content []AnthropicContent `json:"content"`
}

type AnthropicContent struct {
	Type  string          `json:"type"`
	Text  string          `json:"text,omitempty"`
	ID    string          `json:"id,omitempty"`
	Name  string          `json:"name,omitempty"`
	Input json.RawMessage `json:"input,omitempty"`
}

type AnthropicToolDef struct {
	Name        string      `json:"name"`
	Description string      `json:"description"`
	InputSchema interface{} `json:"input_schema"`
}

type DeepSeekRequest struct {
	Model       string            `json:"model"`
	Messages    []DeepSeekMessage `json:"messages"`
	MaxTokens   int               `json:"max_tokens,omitempty"`
	Stream      bool              `json:"stream,omitempty"`
	Temperature float64           `json:"temperature,omitempty"`
}

type DeepSeekMessage struct {
	Role       string     `json:"role"`
	Content    string     `json:"content"`
	ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
	ToolCallID string     `json:"tool_call_id,omitempty"`
}

type ToolCall struct {
	ID       string `json:"id"`
	Type     string `json:"type"`
	Function struct {
		Name      string `json:"name"`
		Arguments string `json:"arguments"`
	} `json:"function"`
}

type Gateway struct {
	deepseekURL string
	apiKey      string
	client      *http.Client
}

func NewGateway(apiKey string) *Gateway {
	return &Gateway{
		deepseekURL: "https://api.deepseek.com/v1",
		apiKey:      apiKey,
		client:      &http.Client{Timeout: 120 * time.Second},
	}
}

func (g *Gateway) convertAnthropicToDeepSeek(req *AnthropicMessage) *DeepSeekRequest {
	dsReq := &DeepSeekRequest{
		Model:     "deepseek-v4-pro",
		MaxTokens: req.MaxTokens,
		Stream:    req.Stream,
	}

	if req.System != "" {
		dsReq.Messages = append(dsReq.Messages, DeepSeekMessage{
			Role: "system", Content: req.System,
		})
	}

	for _, turn := range req.Messages {
		var content string
		for _, c := range turn.Content {
			if c.Type == "text" {
				content += c.Text
			}
		}

		dsMsg := DeepSeekMessage{Role: turn.Role, Content: content}

		for _, c := range turn.Content {
			if c.Type == "tool_use" {
				dsMsg.ToolCalls = append(dsMsg.ToolCalls, ToolCall{
					ID:   c.ID,
					Type: "function",
					Function: struct {
						Name      string `json:"name"`
						Arguments string `json:"arguments"`
					}{Name: c.Name, Arguments: string(c.Input)},
				})
			}
			if c.Type == "tool_result" {
				dsMsg.Role = "tool"
				dsMsg.ToolCallID = c.ID
			}
		}
		dsReq.Messages = append(dsReq.Messages, dsMsg)
	}
	return dsReq
}

func (g *Gateway) HandleAnthropicRequest(w http.ResponseWriter, r *http.Request) {
	var anthReq AnthropicMessage
	if err := json.NewDecoder(r.Body).Decode(&anthReq); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	dsReq := g.convertAnthropicToDeepSeek(&anthReq)
	body, _ := json.Marshal(dsReq)

	httpReq, _ := http.NewRequest("POST",
		g.deepseekURL+"/chat/completions",
		bytes.NewReader(body))
	httpReq.Header.Set("Authorization", "Bearer "+g.apiKey)
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := g.client.Do(httpReq)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadGateway)
		return
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)
	w.Header().Set("Content-Type", "application/json")
	w.Write(respBody)
}

func main() {
	apiKey := os.Getenv("DEEPSEEK_API_KEY")
	if apiKey == "" {
		fmt.Println("Please set DEEPSEEK_API_KEY")
		os.Exit(1)
	}

	gateway := NewGateway(apiKey)
	http.HandleFunc("/v1/messages", gateway.HandleAnthropicRequest)

	addr := ":8080"
	fmt.Printf("Anthropic gateway started at %s\n", addr)
	fmt.Println("Point Claude Code's base_url to http://localhost:8080")
	http.ListenAndServe(addr, nil)
}

#!/usr/bin/env python3
"""
DeepSeek Harness Minimal Mode Usage Example
"""
import requests
from typing import Optional

class HarnessClient:
    def __init__(self, base_url: str = "http://127.0.0.1:3080"):
        self.base_url = base_url
        self.session_id: Optional[str] = None

    def start_session(self, mode: str = "minimal", model: str = "deepseek-v4-pro"):
        resp = requests.post(f"{self.base_url}/sessions", json={
            "mode": mode, "model": model, "reasoning_effort": "high",
        })
        data = resp.json()
        self.session_id = data["session_id"]
        return self.session_id

    def send_message(self, content: str) -> dict:
        resp = requests.post(
            f"{self.base_url}/sessions/{self.session_id}/messages",
            json={"content": content}
        )
        return resp.json()

    def get_trajectory(self) -> list:
        resp = requests.get(
            f"{self.base_url}/sessions/{self.session_id}/trajectory"
        )
        return resp.json()

    def fork_session(self, from_message_id: str) -> str:
        resp = requests.post(
            f"{self.base_url}/sessions/{self.session_id}/fork",
            json={"from_message_id": from_message_id}
        )
        return resp.json()["session_id"]

client = HarnessClient()
client.start_session(mode="minimal", model="deepseek-v4-pro")
response = client.send_message("Create a Python script that calculates Fibonacci numbers")
print(f"Agent response: {response}")
trajectory = client.get_trajectory()
for event in trajectory:
    print(f"[{event['type']}] {str(event.get('content', ''))[:100]}")

// DeepSeek Harness Cordis Plugin System Core Interface
package plugin

import (
	"context"
	"fmt"
	"sync"
)

type Plugin interface {
	Name() string
	Version() string
	Init(ctx context.Context) error
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
	Dependencies() []string
	Provides() []Service
}

type Service interface {
	Name() string
	Methods() []string
}

type Event struct {
	Type    string
	Source  string
	Payload interface{}
}

type Kernel struct {
	mu       sync.RWMutex
	plugins  map[string]Plugin
	services map[string]Service
	eventBus chan Event
}

func NewKernel() *Kernel {
	return &Kernel{
		plugins:  make(map[string]Plugin),
		services: make(map[string]Service),
		eventBus: make(chan Event, 1000),
	}
}

func (k *Kernel) Mount(ctx context.Context, p Plugin) error {
	k.mu.Lock()
	defer k.mu.Unlock()

	for _, dep := range p.Dependencies() {
		if _, ok := k.plugins[dep]; !ok {
			return fmt.Errorf("dependency %s not found", dep)
		}
	}
	for _, svc := range p.Provides() {
		k.services[svc.Name()] = svc
	}
	if err := p.Init(ctx); err != nil {
		return fmt.Errorf("init failed: %w", err)
	}
	if err := p.Start(ctx); err != nil {
		return fmt.Errorf("start failed: %w", err)
	}
	k.plugins[p.Name()] = p
	k.eventBus <- Event{Type: "plugin.mounted", Source: p.Name()}
	return nil
}

func (k *Kernel) Unmount(ctx context.Context, name string) error {
	// Temporal composability: roll back all side effects
	k.mu.Lock()
	defer k.mu.Unlock()

	p, ok := k.plugins[name]
	if !ok {
		return fmt.Errorf("plugin %s not found", name)
	}

	for _, other := range k.plugins {
		for _, dep := range other.Dependencies() {
			if dep == name {
				return fmt.Errorf("plugin %s is depended on by %s", name, other.Name())
			}
		}
	}

	if err := p.Stop(ctx); err != nil {
		return fmt.Errorf("stop failed: %w", err)
	}
	for _, svc := range p.Provides() {
		delete(k.services, svc.Name())
	}
	delete(k.plugins, name)
	k.eventBus <- Event{Type: "plugin.unmounted", Source: name}
	return nil
}

func (k *Kernel) GetService(name string) (Service, error) {
	// Spatial composability: dynamic service resolution
	k.mu.RLock()
	defer k.mu.RUnlock()
	svc, ok := k.services[name]
	if !ok {
		return nil, fmt.Errorf("service %s unavailable", name)
	}
	return svc, nil
}

"""
DeepSeek V4 Pro Cost-Benefit Analysis Tool
"""
from dataclasses import dataclass
from typing import Dict

@dataclass
class ModelPricing:
    name: str
    input_price: float
    output_price: float
    cache_hit: float = 0.0

@dataclass
class TaskProfile:
    name: str
    input_tokens: int
    output_tokens: int
    cache_rate: float
    reasoning_overhead: float

MODELS = {
    "v4-pro": ModelPricing("V4 Pro", 3.0, 6.0, 0.025),
    "v4-pro-peak": ModelPricing("V4 Pro (Peak)", 9.0, 27.0, 0.30),
    "v4-pro-offpeak": ModelPricing("V4 Pro (Off-Peak)", 4.5, 13.5, 0.15),
    "v4-flash": ModelPricing("V4-Flash", 1.0, 2.0, 0.008),
    "fable-5": ModelPricing("Claude Fable 5", 72.0, 360.0),
    "gpt-56-sol": ModelPricing("GPT-5.6 Sol", 54.0, 216.0),
    "grok-46": ModelPricing("Grok 4.6", 14.4, 43.2),
}

TASKS = {
    "simple_qa": TaskProfile("Simple QA", 500, 200, 0.6, 1.0),
    "code_gen": TaskProfile("Code Generation", 2000, 1500, 0.3, 1.5),
    "code_review": TaskProfile("Code Review", 5000, 2000, 0.4, 2.0),
    "agent": TaskProfile("Agent Task", 15000, 8000, 0.2, 3.0),
    "long_ctx": TaskProfile("Long Context", 500000, 5000, 0.1, 2.0),
}

def calculate_cost(model_name: str, task: TaskProfile) -> Dict:
    pricing = MODELS[model_name]
    eff_output = task.output_tokens * task.reasoning_overhead
    cache_hit = task.input_tokens * task.cache_rate
    cache_miss = task.input_tokens * (1 - task.cache_rate)

    input_cost = (
        cache_hit * pricing.cache_hit / 1_000_000 +
        cache_miss * pricing.input_price / 1_000_000
    )
    output_cost = eff_output * pricing.output_price / 1_000_000

    total = input_cost + output_cost
    return {
        "model": pricing.name, "task": task.name,
        "total": round(total, 4),
        "tokens": task.input_tokens + int(eff_output),
    }

def simulate_monthly(
    model_name: str, daily_tasks: int = 1000,
    mix: Dict[str, float] = None
) -> Dict:
    if mix is None:
        mix = {"simple_qa": 0.3, "code_gen": 0.25, "code_review": 0.2,
               "agent": 0.15, "long_ctx": 0.1}

    daily_cost = 0.0
    for task_name, prop in mix.items():
        cost = calculate_cost(model_name, TASKS[task_name])
        daily_cost += cost["total"] * daily_tasks * prop

    monthly = daily_cost * 30
    return {
        "model": MODELS[model_name].name,
        "daily": round(daily_cost, 2),
        "monthly": round(monthly, 2),
        "annual": round(monthly * 12, 2),
    }

print("=" * 60)
print("Monthly Cost Simulation (1000 tasks/day)")
print("=" * 60)
for model in ["v4-pro", "v4-flash", "fable-5", "gpt-56-sol"]:
    sim = simulate_monthly(model)
    print(f"\n{sim['model']}:")
    print(f"  Daily: ¥{sim['daily']:,.2f}")
    print(f"  Monthly: ¥{sim['monthly']:,.2f}")
    print(f"  Annual: ¥{sim['annual']:,.2f}")

"""
Enterprise Multi-Model Router
Supports automatic switching between V4 Pro, Fable 5, GPT-5.6 Sol
"""
import asyncio, json, time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Any, List, Callable
import aiohttp

class ModelProvider(Enum):
    DEEPSEEK = "deepseek"
    ANTHROPIC = "anthropic"
    OPENAI = "openai"

@dataclass
class ModelConfig:
    provider: ModelProvider
    model_name: str
    api_key: str
    base_url: str
    max_tokens: int = 384000
    cost_per_m_input: float = 0.0
    cost_per_m_output: float = 0.0
    quota_per_minute: int = 100

@dataclass
class RoutingResult:
    model_used: str
    provider: ModelProvider
    latency_ms: int
    cost: float
    tokens_input: int
    tokens_output: int
    success: bool
    error: Optional[str] = None

class MultiModelRouter:
    """Intelligent multi-model router based on task type, cost, and latency"""

    def __init__(self):
        self.models: Dict[str, ModelConfig] = {}
        self.callbacks: List[Callable] = []
        self._counts: Dict[str, int] = {}
        self._last_reset: float = time.time()

    def register_model(self, name: str, config: ModelConfig):
        self.models[name] = config

    async def route(
        self, task_type: str, messages: List[Dict],
        tools: Optional[List[Dict]] = None,
        preferred: Optional[str] = None,
    ) -> RoutingResult:
        model_name = preferred or self._select_model(task_type)
        config = self.models[model_name]

        if not self._check_quota(model_name):
            model_name = self._fallback_model(model_name)
            config = self.models[model_name]

        start = time.time()
        try:
            result = await self._call_model(config, messages, tools)
            self._counts[model_name] = self._counts.get(model_name, 0) + 1
            return RoutingResult(
                model_used=model_name, provider=config.provider,
                latency_ms=int((time.time() - start) * 1000),
                cost=result["cost"],
                tokens_input=result["usage"]["prompt_tokens"],
                tokens_output=result["usage"]["completion_tokens"],
                success=True,
            )
        except Exception as e:
            if model_name != list(self.models.keys())[0]:
                return await self.route(
                    task_type, messages, tools,
                    preferred=self._fallback_model(model_name))
            return RoutingResult(
                model_used=model_name, provider=config.provider,
                latency_ms=0, cost=0, tokens_input=0, tokens_output=0,
                success=False, error=str(e),
            )

    def _select_model(self, task_type: str) -> str:
        candidates = []
        for name, config in self.models.items():
            score = 0
            if task_type == "agent" and "pro" in name:
                score += 10
            elif task_type == "chat" and "flash" in name:
                score += 10
            if self._counts.get(name, 0) >= config.quota_per_minute:
                score -= 50
            candidates.append((score, name))
        candidates.sort(key=lambda x: -x[0])
        return candidates[0][1]

    def _fallback_model(self, failed: str) -> str:
        sorted_models = sorted(
            self.models.items(), key=lambda x: x[1].cost_per_m_output)
        for name, _ in sorted_models:
            if name != failed:
                return name
        return failed

    def _check_quota(self, name: str) -> bool:
        if time.time() - self._last_reset > 60:
            self._counts.clear()
            self._last_reset = time.time()
        return self._counts.get(name, 0) < self.models[name].quota_per_minute

    async def _call_model(self, config: ModelConfig, messages: List[Dict],
                          tools: Optional[List[Dict]]) -> Dict:
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {config.api_key}",
        }
        payload = {"model": config.model_name, "messages": messages,
                   "max_tokens": config.max_tokens}
        if tools:
            payload["tools"] = tools

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{config.base_url}/chat/completions",
                headers=headers, json=payload
            ) as resp:
                data = await resp.json()
                usage = data["usage"]
                cost = (usage["prompt_tokens"] * config.cost_per_m_input +
                        usage["completion_tokens"] * config.cost_per_m_output) / 1_000_000
                return {"content": data["choices"][0]["message"]["content"],
                        "usage": usage, "cost": cost}

async def demo():
    router = MultiModelRouter()
    router.register_model("v4-pro", ModelConfig(
        provider=ModelProvider.DEEPSEEK, model_name="deepseek-v4-pro",
        api_key="sk-key", base_url="https://api.deepseek.com/v1",
        cost_per_m_input=3.0, cost_per_m_output=6.0, quota_per_minute=500))
    router.register_model("v4-flash", ModelConfig(
        provider=ModelProvider.DEEPSEEK, model_name="deepseek-v4-flash",
        api_key="sk-key", base_url="https://api.deepseek.com/v1",
        cost_per_m_input=1.0, cost_per_m_output=2.0, quota_per_minute=2500))

    result = await router.route(
        task_type="agent",
        messages=[{"role": "user", "content": "Analyze this codebase for bottlenecks"}],
        tools=[{"type": "function", "function": {"name": "bash"}}],
    )
    print(f"Routed to: {result.model_used}")
    return result

# asyncio.run(demo())