Sakana Fugu: A 10,000-Parameter Evolved Head Coordinates GPT-5, Claude, and Gemini Better Than Any of Them Can Work Alone

# Sakana Fugu API: OpenAI-compatible endpoint
# Source: sakana.ai/fugu-beta documentation
# Design intent: zero integration cost for existing API users
# The entire orchestration is hidden behind a single endpoint call

from openai import OpenAI

# ─── BEFORE: direct frontier model call ─────────────────────────────────────
before_client = OpenAI(
    api_key="sk-anthropic-key",
    base_url="https://api.anthropic.com/v1",
)

before_response = before_client.chat.completions.create(
    model="claude-opus-4-6",
    messages=[{"role": "user", "content": "Solve the following optimization problem..."}],
)

# ─── AFTER: Fugu drops in with one endpoint change ────────────────────────
after_client = OpenAI(
    api_key="fugu-api-key",
    base_url="https://api.sakana.ai/v1",
    # ← THIS is the drop-in: same OpenAI SDK, different base_url
    # Everything else in your code is unchanged
)

# ← Same call structure as before, but now:
#   1. Fugu's Conductor reads the task and designs a workflow
#   2. Trinity's routing head assigns roles to frontier models
#   3. Multiple models may execute in sequence or parallel
#   4. Fugu synthesizes the final answer
#   5. You receive a standard OpenAI-format chat completion response
fugu_response = after_client.chat.completions.create(
    model="fugu-ultra",          # or "fugu-mini" for latency-sensitive tasks
    messages=[{"role": "user", "content": "Solve the following optimization problem..."}],
    # ← Optional: hint the orchestrator about compute budget
    # stream=True for streaming output (supported, same as standard OpenAI streaming)
)

print(fugu_response.choices[0].message.content)
# Output: synthesized answer from whatever topology Fugu chose internally
# May have involved: Gemini for initial planning, GPT-5 for execution,
# Claude for verification — user sees none of this routing

# ─── CHECKING WHAT HAPPENED (if Fugu exposes workflow metadata) ──────────────
# Fugu has been observed to return metadata about the coordination in some contexts:
import json
if hasattr(fugu_response, 'usage') and fugu_response.usage:
    print(f"Total tokens (across all models): {fugu_response.usage.total_tokens}")
    # ← Token count includes ALL API calls Fugu made to worker models
    # ← This is important for cost estimation: a hard task may invoke 3-5 models
    # ← fugu-mini is more cost-efficient; fugu-ultra may call more models per task

# Trinity coordinator: evolutionary training + Thinker/Worker/Verifier role system
# Reconstructed from arXiv:2512.04695 architecture description
# Design intent: separate role routing from skill execution

import torch
import torch.nn as nn
from typing import Literal

# ─── THE TRINITY COORDINATOR COMPONENTS ──────────────────────────────────────
# Two separate components with very different sizes and training methods:
#
# 1. Compact LM (~0.6B): processes the query context and prior outputs
#    to produce hidden states used for role assignment
#
# 2. Routing head (~10K params): takes the LM's hidden state and decides
#    which role to assign to which model in the pool
#    ← THIS is the surprising part: 10K parameters makes the routing decision
#    ← Trained by EVOLUTIONARY STRATEGY, not gradient descent

RoleType = Literal["THINKER", "WORKER", "VERIFIER"]

class TrinityRoutingHead(nn.Module):
    """
    The Trinity routing head: ~10K parameters, evolved (not gradient-trained).
    Takes compressed query representation → outputs (role, model_index).
    
    ← WHY evolutionary strategy and not backpropagation?
      The routing decision is discrete (which model? which role?), and
      gradient descent through a discrete selection is not straightforward.
      Evolutionary strategies optimize over the discrete role-assignment policy
      without needing differentiable relaxations.
      
    ← WHY so small (~10K parameters)?
      The coordinator only needs to learn: "given this query pattern, which
      combination of Thinker/Worker/Verifier roles produces the best outcome?"
      The domain expertise is in the worker models. The coordinator only routes.
      You don't need 70B parameters to learn: "coding tasks need Thinker→Worker→Verifier;
      factual questions need just Worker."
    """
    def __init__(self, hidden_dim: int = 128, n_models: int = 6, n_roles: int = 3):
        super().__init__()
        # The "head" that makes the actual routing decision
        # Small: hidden_dim * n_models * n_roles ≈ 128 * 6 * 3 ≈ 2,304 params
        self.role_router = nn.Sequential(
            nn.Linear(hidden_dim, 64),
            nn.ReLU(),
            nn.Linear(64, n_models * n_roles),  # ← logits over (model, role) pairs
        )
        # ← At ~10K total params, this fits the paper's description
        # ← Evolved via CMA-ES or similar derivative-free optimizer, not Adam/SGD

    def forward(self, context_hidden: torch.Tensor) -> tuple[int, RoleType]:
        """
        Given compressed context representation, output (model_index, role).
        Evolutionary training optimizes the weights to maximize task success rate
        across the training distribution.
        """
        logits = self.role_router(context_hidden)         # [batch, n_models * n_roles]
        logits = logits.view(-1, n_models, n_roles)       # [batch, n_models, n_roles]
        # In practice: greedy selection during inference
        model_idx = logits.max(dim=-1).values.argmax(dim=-1).item()
        role_idx = logits[0, model_idx].argmax().item()
        role = ["THINKER", "WORKER", "VERIFIER"][role_idx]
        return model_idx, role


# ─── THE THINKER/WORKER/VERIFIER TURN LOOP ──────────────────────────────────
async def trinity_turn_loop(
    query: str,
    routing_head: TrinityRoutingHead,
    context_encoder,         # the ~0.6B compact LM
    model_pool: list[dict],  # [{"name": "gpt-5.4", "client": ...}, ...]
    max_turns: int = 5,
) -> str:
    """
    Trinity's multi-turn orchestration loop.
    
    Each turn:
    1. Context encoder reads query + all prior outputs → hidden state
    2. Routing head assigns (model, role) for this turn
    3. Assigned model runs with role-specific prompt
    4. Output appended to context for next turn
    
    ← The loop terminates when the Verifier approves, or max_turns is reached.
    ← No explicit stopping rule is learned: VERIFIER role acceptance IS the stopping signal.
    """
    context = [{"role": "user", "content": query}]
    outputs_by_role = {r: [] for r in ["THINKER", "WORKER", "VERIFIER"]}
    final_answer = None

    for turn in range(max_turns):
        # Step 1: encode all accumulated context
        context_hidden = context_encoder.encode(context)   # [1, hidden_dim]

        # Step 2: routing head decides who does what
        model_idx, role = routing_head(context_hidden)
        selected_model = model_pool[model_idx]

        # Step 3: construct role-specific instruction
        # ← The prompting style per role is what was learned during evolutionary training
        role_instructions = {
            "THINKER": f"You are a THINKER. Analyze the problem and develop a step-by-step plan. Do NOT implement yet.\nQuery: {query}\n",
            "WORKER": f"You are a WORKER. Implement the following plan:\n{outputs_by_role['THINKER'][-1] if outputs_by_role['THINKER'] else 'No plan available. Solve directly.'}\n",
            "VERIFIER": f"You are a VERIFIER. Validate this solution:\n{outputs_by_role['WORKER'][-1] if outputs_by_role['WORKER'] else 'No solution to verify.'}\nRespond with ACCEPT or REJECT and explain why.",
        }
        prompt = role_instructions[role]

        # Step 4: call the assigned model
        response = await selected_model["client"].chat.completions.create(
            model=selected_model["name"],
            messages=[{"role": "user", "content": prompt}],
        )
        output_text = response.choices[0].message.content

        # Step 5: record output and update context
        outputs_by_role[role].append(output_text)
        context.append({"role": "assistant", "content": f"[{role} - {selected_model['name']}]: {output_text}"})

        # ← Termination condition: Verifier accepts
        if role == "VERIFIER" and "ACCEPT" in output_text.upper():
            final_answer = outputs_by_role["WORKER"][-1] if outputs_by_role["WORKER"] else output_text
            break

    return final_answer or outputs_by_role.get("WORKER", [query])[-1]

# Conductor: RL-trained orchestrator generating natural language topologies
# Reconstructed from arXiv:2512.04388 architecture and methodology
# Base: Qwen2.5-7B, trained with GRPO

# ─── WHAT CONDUCTOR OUTPUTS (natural language workflow) ───────────────────────
# For a hard coding problem, Conductor might generate a plan like:
example_conductor_output = """
STEP 1: [Gemini-2.5-Pro | PLANNER]
Instructions: "Analyze the problem and identify the key algorithmic challenges.
List the data structures and algorithmic approach needed. Do not write code."
Input: [original query]

STEP 2: [GPT-5.4 | CODER]
Instructions: "Implement a solution for the following problem.
Use efficient algorithms as outlined in the plan. Include error handling."
Input: [original query] + [STEP 1 output]

STEP 3: [DeepSeek-R1-Distill-Qwen-32B | VERIFIER]
Instructions: "Critically review this code for correctness, edge cases, and efficiency.
Identify any bugs. Run through at least 3 test cases mentally."
Input: [STEP 2 output]

STEP 4: [GPT-5.4 | REVISER]
Instructions: "Fix the identified bugs. Apply the suggested improvements."
Input: [STEP 2 output] + [STEP 3 output]
"""

# ← THIS is what makes Conductor different from Trinity:
#   Trinity's routing head assigns roles via evolved discrete policy
#   Conductor writes the FULL workflow as text, including targeted custom instructions
#   per model, information routing (what each model sees), and step dependencies

# ─── CONDUCTOR TRAINING: GRPO ON 960 PROBLEMS ─────────────────────────────────
# Training setup (from paper):
# - Base model: Qwen2.5-7B
# - Training method: GRPO (Group Relative Policy Optimization), a form of RL
# - Training set: 960 problems across math, coding, reasoning
# - Randomized agent pools: each training problem uses a different subset of models
#   ← THIS is the key to pool-agnostic generalization:
#     training with random pools forces the Conductor to learn model capabilities
#     in the abstract, not overfit to one specific set of frontier models

# ─── RECURSIVE SELF-CALLING ─────────────────────────────────────────────────
async def conductor_with_recursion(
    query: str,
    conductor_model,      # the 7B Qwen2.5-GRPO model
    model_pool: list[str],
    max_depth: int = 3,   # recursion depth = tunable compute budget
    depth: int = 0,
) -> str:
    """
    When Conductor selects itself as a worker, it enters a new orchestration turn.
    
    ← WHY this enables test-time scaling without retraining:
      At depth=0, Conductor plans the workflow.
      If it selects itself as a worker, it re-reads its own prior output
      and decides: "Was my first coordination strategy good enough?"
      If not, it spins up a DIFFERENT coordination strategy.
      
      This is fundamentally different from simple reflection:
      The model is not just refining its own answer.
      It is reconsidering the COORDINATION STRATEGY itself.
      
      Compute budget at inference = recursion depth × cost per orchestration cycle.
      You tune depth per task difficulty with no retraining.
    """
    if depth >= max_depth:
        # Base case: forced termination, return best current answer
        return await single_model_fallback(query, model_pool)

    # Step 1: Conductor generates a coordination plan
    plan_response = await conductor_model.generate(
        prompt=f"Task: {query}\nDesign an optimal coordination workflow for the following agent pool: {model_pool}",
        context=f"[Depth {depth}] Prior coordination attempts: {depth} rounds completed."
    )
    plan = plan_response.text

    # Step 2: Execute the plan against the model pool
    # ← Conductor can include "fugu-coordinator" in the plan as a worker
    intermediate_result = await execute_plan(plan, model_pool, query)

    # Step 3: Self-evaluation: was this coordination strategy good enough?
    eval_response = await conductor_model.generate(
        prompt=f"Original task: {query}\nCoordination plan used: {plan}\nResult: {intermediate_result}\n\nEvaluate: Is this result satisfactory? If not, what coordination strategy would improve it?",
    )

    if "SATISFACTORY" in eval_response.text.upper():
        return intermediate_result  # ← exits recursion at this depth
    else:
        # ← THIS is recursive self-calling: Conductor re-plans with new strategy
        # Depth counter is the inference-time compute knob
        return await conductor_with_recursion(
            query=query,
            conductor_model=conductor_model,
            model_pool=model_pool,
            max_depth=max_depth,
            depth=depth + 1,  # ← tunable: higher max_depth = more compute, better quality
        )

Conductor (Qwen2.5-7B, GRPO-trained) analyzes:
  - Problem type: competitive programming, systems implementation
  - Language requirement: Rust (specificity suggests need for expert)
  - Algorithmic task: Kruskal's (graph algorithm, requires correctness)
  - Verification importance: high (competitive programming = exact answers)

Conductor output (natural language plan):
  STEP 1: [Gemini-2.5-Pro | THINKER]
    "Outline the Kruskal's algorithm approach for MST in Rust.
     Identify: Union-Find data structure requirements, edge sorting,
     and Rust-specific considerations (no GC, ownership model)."
     
  STEP 2: [GPT-5.4 | CODER]
    "Implement a complete, idiomatic Rust solution for Kruskal's MST.
     Use the algorithmic outline provided. Handle N up to 10^5, M up to 2×10^5.
     Include: proper stdin reading, Union-Find with path compression + rank."
     
  STEP 3: [DeepSeek-R1-Distill-Qwen-32B | VERIFIER]
    "Verify this Rust Kruskal implementation:
     1. Trace through example: N=4, M=5 edges
     2. Check Union-Find correctness
     3. Verify edge case: disconnected graph
     4. Verify time complexity is O(M log M)"

Gemini output: 
  "Kruskal's requires:
   1. Sort edges by weight O(M log M)
   2. Union-Find with path compression and union by rank for near-O(α(N))
   3. In Rust: Vec<(usize, usize, i64)> for edges, Vec<usize> parent and rank
   4. Rust-specific: use iter().position() carefully, avoid unnecessary clones..."
   
Tokens used: ~380 (targeted planning prompt, not a generic request)

// GPT-5.4 output: complete Rust implementation
fn find(parent: &mut Vec<usize>, x: usize) -> usize {
    if parent[x] != x { parent[x] = find(parent, parent[x]); }
    parent[x]
}
fn union(parent: &mut Vec<usize>, rank: &mut Vec<usize>, x: usize, y: usize) -> bool {
    let (px, py) = (find(parent, x), find(parent, y));
    if px == py { return false; }
    if rank[px] < rank[py] { parent[px] = py; }
    else if rank[px] > rank[py] { parent[py] = px; }
    else { parent[py] = px; rank[px] += 1; }
    true
}
fn main() {
    // ... stdin reading, edge sorting, Kruskal loop ...
}

DeepSeek output:
  "ACCEPT: Implementation is correct.
   Trace N=4, edges [(1,2,1),(2,3,2),(1,3,3),(3,4,4),(2,4,5)]:
   Sort: (1,2,1),(2,3,2),(1,3,3),(3,4,4),(2,4,5)
   Add (1,2,1): components=3, mst=1
   Add (2,3,2): components=2, mst=3
   Skip (1,3,3): cycle
   Add (3,4,4): components=1, mst=7
   MST weight: 7 ✓
   Time complexity: O(M log M) confirmed ✓
   Disconnected graph: returns partial MST with fewer than N-1 edges ✓"
   
VERIFIER accepted. Trinity loop terminates at turn 3.
Tokens used: ~590 (targeted verification context)

The Rust Kruskal's MST implementation (working code above).
Total API calls: 3 (Thinker + Worker + Verifier)
Total tokens across all models: ~1,690
Total wall time: ~8-12 seconds (3 sequential model calls)

Comparison if Opus 4.6 handled this directly:
  Single model, no role separation
  Typically: plan-in-one-pass + implement + self-check in a single response
  Risk: Opus might implement without verifying, or verify incorrectly
  SWEPro benchmark: Opus 4.6 max = 53.4 vs fugu-ultra = 54.2