Lewis Moten

Unveiling the Code: Institutional Memory & Obsessions

Cactus Needle 2

While exploring small models, I found Cactus Needle 2, which is just 14 MB in size, has 45M parameters, is trained with (mostly) 2-bit weights, and has a 256-token sliding window (~150-200 words). In my tests, it seemed to use around 80 MB of memory, but I’ve watched Better Stack run this model on an ESP32 with only 15 MB of RAM.

Github: @cactus-compute/needle

This model isn’t used for chat, coding, or in-depth analysis. It has a very narrow purpose. It focuses on choosing which tool to call or extracting a few values using a JSON schema. You can’t ask it to write a poem, but you can provide it with a tool definition to call, so that something else can write the poem on its behalf.

Most small models reduce their footprint by quantizing weights to fewer bits, but this can reduce model quality. Needle was trained on mostly 2-bit weights from the start, so it doesn’t lose anything. It also uses many techniques to reduce transformation overhead, including large weight matrices, lookup tables from cached results (Hash N-Gram Lookup Tables), and more.

Code Bear – Needle 2 Explained

The model is so small that you can run it on your CPU, or even in a browser. The repository also shows you how to fine-tune it to create your own custom weights. This is pretty attractive since I’ve been getting more curious about how to make my own models with limited hardware.

Since I’ve been setting up various worker agents on computers at home to do a lot of batch processing, I decided to expose the model via an API similar to OpenAI and Ollama endpoints to make it easier to hook into my worker agents. I can keep the larger models for things that need more compute, but for some of the simple stuff, I now have Needle to add to my toolset.

install.sh
Shell
# Create project folder
mkdir -p ~/dev/needle-test
cd ~/dev/needle-test
# Create isolated Python environment
python3 -m venv .venv
# Activate virtual environment
source .venv/bin/activate
# Update pip
python3 -m pip install --upgrade pip
# Install Needle and the packages used by API wrapper.
# FastAPI provides the HTTP API; Uvicorn runs it.
# SentencePiece counts Needle tokenizer tokens.
python3 -m pip install cactus-needle fastapi \
  "uvicorn[standard]" sentencepiece
# Confirm Needle installation & version
python3 -c 'import needle; print(needle.__version__)'
# 2.0.14
# Create API server script
nano needle_api.py
needle_api.py
Python
import json
import threading
import time
import uuid
from typing import Any
import needle
from fastapi import FastAPI, HTTPException
from needle.model.tokenizer import get_tokenizer
from pydantic import BaseModel, Field
app = FastAPI(
    title="Needle OpenAI-Compatible API",
    version="1.0.0",
)
# Needle has one bundled model/runtime rather than an Ollama-style model catalog.
MODEL_ID = "needle-2"
MODEL_TAG = "needle-2:latest"
MODEL_CREATED = 1789257600  # 2026-09-13 UTC
MODEL_SIZE = 14 * 1024 * 1024
# Needle's native runtime is process-global. One request at a time avoids
# separate requests overwriting each other's active tool schema.
engine_lock = threading.Lock()
engine_loaded = False
last_used_at: int | None = None
# This uses Needle's own SentencePiece tokenizer for useful token estimates.
tokenizer = get_tokenizer()
class ChatRequest(BaseModel):
    model: str = MODEL_ID
    messages: list[dict[str, Any]]
    max_tokens: int = Field(default=128, ge=1, le=256)
    # Send this for client-owned function calling.
    tools: list[dict[str, Any]] | None = None
    # Send this for structured JSON extraction.
    response_format: dict[str, Any] | None = None
    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "model": "needle-2",
                    "messages": [
                        {
                            "role": "user",
                            "content": (
                                "Invoice from Acme Corp, $1,200.00, "
                                "due 2026-09-01."
                            ),
                        }
                    ],
                    "response_format": {
                        "type": "json_schema",
                        "json_schema": {
                            "name": "invoice",
                            "schema": {
                                "type": "object",
                                "properties": {
                                    "vendor": {"type": "string"},
                                    "total": {"type": "number"},
                                    "due_date": {"type": "string"},
                                },
                                "required": ["vendor", "total", "due_date"],
                            },
                        },
                    },
                }
            ]
        }
    }
def count_tokens(value: str) -> int:
    """Count tokens with Needle's tokenizer."""
    return len(tokenizer.encode(value))
def model_info() -> dict[str, Any]:
    """OpenAI /v1/models model-object format."""
    return {
        "id": MODEL_ID,
        "object": "model",
        "created": MODEL_CREATED,
        "owned_by": "cactus-compute",
        "root": MODEL_ID,
    }
def model_details() -> dict[str, Any]:
    """Extra metadata for health and Ollama-compatible endpoints."""
    return {
        "format": "cact",
        "family": "needle",
        "families": ["needle"],
        "parameter_size": "45M",
        "quantization_level": "CQ2",
        "context_window_tokens": 256,
    }
def message_text(messages: list[dict[str, Any]]) -> str:
    """
    Flatten text from OpenAI-style messages.
    This lightweight service is intended for one focused extraction or
    tool-selection turn. Needle is not a general long-context chat model.
    """
    parts: list[str] = []
    for message in messages:
        content = message.get("content", "")
        if isinstance(content, str):
            parts.append(content)
        elif isinstance(content, list):
            for part in content:
                if (
                    isinstance(part, dict)
                    and part.get("type") in ("text", "input_text")
                    and isinstance(part.get("text"), str)
                ):
                    parts.append(part["text"])
    return "\n\n".join(parts).strip()
def openai_tools_to_needle(
    openai_tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """
    Convert OpenAI function-tool definitions to Needle schemas.
    These are client-owned tools. This API does not execute them; it only
    returns which tool Needle selected and the arguments it produced.
    """
    needle_tools: list[dict[str, Any]] = []
    for tool in openai_tools:
        if tool.get("type") != "function":
            raise HTTPException(
                status_code=400,
                detail="Only tools with type='function' are supported.",
            )
        function = tool.get("function", {})
        name = function.get("name")
        parameters = function.get("parameters")
        if not isinstance(name, str) or not name:
            raise HTTPException(
                status_code=400,
                detail="Each tool needs function.name.",
            )
        if not isinstance(parameters, dict):
            raise HTTPException(
                status_code=400,
                detail=f"Tool '{name}' needs function.parameters as a JSON Schema object.",
            )
        needle_tools.append({
            "name": name,
            "description": function.get("description", ""),
            "parameters": parameters,
        })
    return needle_tools
def response_tool(response_format: dict[str, Any] | None) -> dict[str, Any]:
    """
    Convert OpenAI response_format.json_schema to Needle's one-tool
    structured-extraction schema.
    """
    if response_format and response_format.get("type") == "json_schema":
        json_schema = response_format.get("json_schema", {})
        schema = json_schema.get("schema")
        if not isinstance(schema, dict) or schema.get("type") != "object":
            raise HTTPException(
                status_code=400,
                detail=(
                    "response_format.json_schema.schema must be "
                    "a JSON Schema object whose type is 'object'."
                ),
            )
        return {
            "name": json_schema.get("name", "response"),
            "description": json_schema.get(
                "description",
                "Extract structured facts from the supplied text.",
            ),
            "parameters": schema,
        }
    # Needle is a structured tool-call model, not a general chat model.
    # This gives callers without a response schema one simple JSON field.
    return {
        "name": "response",
        "description": "Return a concise answer grounded in the supplied text.",
        "parameters": {
            "type": "object",
            "properties": {
                "answer": {
                    "type": "string",
                    "description": "A concise answer based only on the supplied text.",
                }
            },
            "required": ["answer"],
        },
    }
@app.get("/")
def root():
    return {
        "service": "Needle OpenAI-Compatible API",
        "model": MODEL_ID,
        "docs": "/docs",
        "openai_models": "/v1/models",
    }
@app.get("/health")
def health():
    return {
        "status": "ok",
        "model": MODEL_ID,
        "engine_loaded": engine_loaded,
        "last_used_at": last_used_at,
        "details": model_details(),
    }
# Standard OpenAI-compatible model discovery.
@app.get("/v1/models")
def list_models():
    return {
        "object": "list",
        "data": [model_info()],
    }
@app.get("/v1/models/{model_id}")
def get_model(model_id: str):
    if model_id not in (MODEL_ID, MODEL_TAG):
        raise HTTPException(
            status_code=404,
            detail=f"Unknown model: {model_id}",
        )
    return model_info()
@app.post("/v1/chat/completions")
def chat_completions(request: ChatRequest):
    global engine_loaded, last_used_at
    if request.model not in (MODEL_ID, MODEL_TAG):
        raise HTTPException(
            status_code=404,
            detail=f"Unknown model: {request.model}. Available: {MODEL_ID}",
        )
    if request.tools and request.response_format:
        raise HTTPException(
            status_code=400,
            detail=(
                "Use either tools for tool selection or response_format "
                "for structured extraction in one request, not both."
            ),
        )
    text = message_text(request.messages)
    if not text:
        raise HTTPException(
            status_code=400,
            detail="No text found in messages.",
        )
    # In tool mode, the caller's schemas define the available tools.
    # In extraction mode, the response schema becomes Needle's only tool.
    needle_tools = openai_tools_to_needle(request.tools or [])
    active_tools = needle_tools or [response_tool(request.response_format)]
    input_tokens = count_tokens(text)
    schema_tokens = sum(
        count_tokens(json.dumps(tool, separators=(",", ":")))
        for tool in active_tools
    )
    started_at = time.perf_counter()
    with engine_lock:
        agent = needle.Needle(tools=active_tools)
        try:
            # complete() selects calls but never executes a client-owned tool.
            raw = agent.complete(
                text,
                max_new_tokens=request.max_tokens,
            )
            engine_loaded = True
            last_used_at = int(time.time())
        finally:
            agent.close()
    elapsed_ms = round((time.perf_counter() - started_at) * 1000, 2)
    calls = raw.get("function_calls") or []
    needle_metadata = {
        "confidence": raw.get("confidence"),
        "reasoning": raw.get("reasoning"),
        # This is a preliminary engine diagnostic, not necessarily a rejection.
        "engine_validation": raw.get("validation"),
        "execution_time_ms": elapsed_ms,
        "prefill_tokens_per_second": raw.get("prefill_tps"),
        "decode_tokens_per_second": raw.get("decode_tps"),
        "peak_ram_mb": raw.get("peak_ram_mb"),
    }
    # Tool mode: send standard OpenAI-shaped tool calls back to the caller.
    if needle_tools:
        tool_calls = [
            {
                "id": f"call_{uuid.uuid4().hex}",
                "type": "function",
                "function": {
                    "name": call["name"],
                    "arguments": json.dumps(call.get("arguments", {})),
                },
            }
            for call in calls
        ]
        output_tokens = count_tokens(json.dumps(tool_calls, separators=(",", ":")))
        return {
            "id": f"chatcmpl-{uuid.uuid4().hex}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": MODEL_ID,
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": None,
                    "tool_calls": tool_calls,
                },
                "finish_reason": "tool_calls" if tool_calls else "stop",
            }],
            "needle": needle_metadata,
            "usage": {
                "prompt_tokens": input_tokens + schema_tokens,
                "completion_tokens": output_tokens,
                "total_tokens": input_tokens + schema_tokens + output_tokens,
            },
            "needle_usage": {
                "input_text_tokens": input_tokens,
                "schema_tokens": schema_tokens,
                "input_tokens_in_sliding_window": min(256, input_tokens),
                "returned_json_tokens": output_tokens,
                "context_window_tokens": 256,
            },
        }
    # Structured-extraction / JSON-response mode.
    if not calls:
        raise HTTPException(
            status_code=422,
            detail={
                "message": "Needle could not confidently extract a response.",
                "needle_response": raw,
            },
        )
    data = calls[0].get("arguments", {})
    content = json.dumps(data)
    output_tokens = count_tokens(content)
    return {
        "id": f"chatcmpl-{uuid.uuid4().hex}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": MODEL_ID,
        "choices": [{
            "index": 0,
            "message": {
                "role": "assistant",
                "content": content,
            },
            "finish_reason": "stop",
        }],
        "needle": needle_metadata,
        "usage": {
            "prompt_tokens": input_tokens + schema_tokens,
            "completion_tokens": output_tokens,
            "total_tokens": input_tokens + schema_tokens + output_tokens,
        },
        "needle_usage": {
            "input_text_tokens": input_tokens,
            "schema_tokens": schema_tokens,
            "input_tokens_in_sliding_window": min(256, input_tokens),
            "returned_json_tokens": output_tokens,
            "context_window_tokens": 256,
        },
    }
# Optional Ollama-compatible model-discovery endpoints.
@app.get("/api/tags")
def ollama_tags():
    return {
        "models": [{
            "name": MODEL_TAG,
            "model": MODEL_TAG,
            "modified_at": "2026-09-13T00:00:00Z",
            "size": MODEL_SIZE,
            "digest": "",
            "details": model_details(),
        }]
    }
@app.post("/api/show")
def ollama_show(body: dict[str, Any]):
    requested = body.get("name", MODEL_TAG)
    if requested not in (MODEL_ID, MODEL_TAG):
        raise HTTPException(
            status_code=404,
            detail=f"Unknown model: {requested}",
        )
    return {
        "modelfile": "# Needle uses Cactus's built-in runtime.",
        "parameters": "",
        "template": "",
        "details": model_details(),
        "model_info": {
            "general.architecture": "needle",
            "general.parameter_count": 45_000_000,
            "cactus.context_length": 256,
        },
    }
@app.get("/api/ps")
def ollama_running_models():
    if not engine_loaded:
        return {"models": []}
    return {
        "models": [{
            "name": MODEL_TAG,
            "model": MODEL_TAG,
            "size": MODEL_SIZE,
            "size_vram": 0,
            "expires_at": "never",
            "details": model_details(),
        }]
    }
serve.sh
Shell
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
export NEEDLE_TELEMETRY=0
exec .venv/bin/python -m uvicorn needle_api:app \
  --host 127.0.0.1 \
  --port 8082
needle_api.py
Shell
# Test the interactive API documentation in your browser.
open http://127.0.0.1:8082/docs
# Service discovery
# /api/* endpoints are Ollama-style compatibility endpoints.
# /v1/* endpoints are OpenAI-style compatibility endpoints.
curl http://127.0.0.1:8082/api/tags
curl http://127.0.0.1:8082/api/ps
curl http://127.0.0.1:8082/health
curl http://127.0.0.1:8082/v1/models
curl http://127.0.0.1:8082/v1/models/needle-2
# Structured Value Extraction
curl http://127.0.0.1:8082/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "needle-2",
    "messages": [
      {
        "role": "user",
        "content": "Invoice from Acme Corp, $1,200.00, due 2026-09-01."
      }
    ],
    "response_format": {
      "type": "json_schema",
      "json_schema": {
        "name": "invoice",
        "schema": {
          "type": "object",
          "properties": {
            "vendor": { "type": "string" },
            "total": { "type": "number" },
            "due_date": { "type": "string" }
          },
          "required": ["vendor", "total", "due_date"]
        }
      }
    }
  }'
# Client-Owned Tool Selection
curl http://127.0.0.1:8082/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "needle-2",
    "messages": [
      {
        "role": "user",
        "content": "What is 19.95 plus 6 percent tax?"
      }
    ],
    "tools": [
      {
        "type": "function",
        "function": {
          "name": "calculate_tax",
          "description": "Calculate sales tax and return the total.",
          "parameters": {
            "type": "object",
            "properties": {
              "amount": { "type": "number" },
              "tax_rate_percent": { "type": "number" }
            },
            "required": ["amount", "tax_rate_percent"]
          }
        }
      }
    ]
  }'
# Needle returns a tool call with arguments
# The caller (you) executes tools based on Needles choice
Spiked cactus robot holding a sign reading “THE LIGHT SHOULD BE TURNED OFF”

Leave a Reply

Discover more from Lewis Moten

Subscribe now to keep reading and get access to the full archive.

Continue reading