frox-nano-v2 / src /inference /engine /morph_engine.py
Hritik045678's picture
Upload folder using huggingface_hub
296a506 verified
Raw
History Blame Contribute Delete
29.1 kB
"""
Frox AI Morph 1.1 β€” Inference Engine
The single-process serving engine used for Colab / edge / local dev.
(Production multi-user serving uses vLLM β€” see backend architecture doc.
This engine still matters: it's what actually runs training-time eval,
the demo CLI, and any deployment without a GPU cluster.)
Improvements over Morph 1.0:
- Streaming is now the core primitive (`generate_stream`); `generate()`
is a thin wrapper that consumes the stream. 1.0 had two separate,
diverging implementations of the sampling logic β€” one for `generate()`
and a different, buggier one for `stream()` (it re-decoded the full
sequence every single token via `tokenizer.decode(generated_ids)`
instead of yielding deltas, and it never applied `top_p`/`top_k`/
`repetition_penalty`, only temperature). Fixed here.
- Session-based KV cache: `chat()` keeps a MorphSessionCache alive
across calls so a multi-turn conversation doesn't re-run the full
prefill on every turn β€” only the new user message is processed.
- Speculative decoding: optional small draft model proposes several
tokens, the main model verifies them in a single forward pass.
- Quantized loading now actually quantizes (see inference/quantize),
instead of the 1.0 stub that logged a message and loaded fp16 anyway.
- Every generation call reports tokens/sec, matching what the training
loop already reports, so speed regressions are visible immediately.
"""
from __future__ import annotations
import gc
import time
from pathlib import Path
from typing import Dict, Generator, List, Optional, Tuple, Union
import torch
import torch.nn.functional as F
from config.model_config import MorphConfig
from multimodal.fusion.morph_multimodal import MorphMultimodalModel
from tokenizer.morph_tokenizer import (
apply_chat_template, build_morph_tokenizer,
format_thinking,
)
from inference.cache.kv_cache import MorphKVCache, MorphSessionCache
from inference.quantize.quantize import quantize_4bit, quantize_8bit, print_quantization_report
from multimodal.fusion.generation_pipeline import detect_vram_state, VRAMState, free_memory
class MorphInferenceEngine:
"""
Frox Morph 1.1 inference engine.
Usage:
engine = MorphInferenceEngine.from_pretrained("./frox-morph-1-1")
response = engine.generate([{"role": "user", "content": "Hello!"}])
# Streaming
for chunk in engine.generate_stream(messages):
print(chunk, end="", flush=True)
# Multi-turn with persistent KV cache (no re-prefill per turn)
engine.chat("session-123", "What's the capital of France?")
engine.chat("session-123", "What's its population?") # reuses cache
"""
def __init__(
self,
model: MorphMultimodalModel,
tokenizer,
config: MorphConfig,
device: Optional[torch.device] = None,
dtype: torch.dtype = torch.float16,
):
self.config = config
self.tokenizer = tokenizer
self.device = device or torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.dtype = dtype
self.model = model.to(self.device, dtype=self.dtype)
self.model.eval()
self.vram_state = detect_vram_state()
# Session cache for multi-turn conversations without re-prefill
self.sessions = MorphSessionCache(
num_layers=config.text.num_hidden_layers,
num_kv_heads=config.text.num_key_value_heads,
head_dim=config.text.head_dim,
dtype=self.dtype,
device=self.device,
max_seq_len_per_session=config.text.max_position_embeddings,
)
# Optional speculative decoding draft model
self.draft_model: Optional[torch.nn.Module] = None
if config.inference.use_speculative and config.inference.draft_model_path:
self._load_draft_model(config.inference.draft_model_path)
print(f"βœ“ Morph 1.1 Inference Engine ready")
print(f" Device: {self.device} | dtype: {self.dtype} | VRAM tier: {self.vram_state}")
params = model.param_count()
print(f" Params: {params['total_billions']}B "
f"(LM: {params['lm_billions']}B, Vision: {params['vision_billions']}B)")
# ── Construction ──────────────────────────────────────────────
@classmethod
def from_pretrained(
cls,
path: str,
device: Optional[str] = None,
dtype: str = "float16",
quantization: Optional[str] = None, # None | "4bit" | "8bit"
) -> "MorphInferenceEngine":
p = Path(path)
dev = torch.device(device) if device else torch.device(
"cuda" if torch.cuda.is_available() else "cpu"
)
torch_dtype = getattr(torch, dtype)
print(f"πŸ“‚ Loading Morph 1.1 from {path}...")
model = MorphMultimodalModel.from_saved(str(p), device="cpu") # load to CPU first
model = cls._apply_quantization(model, quantization, torch_dtype)
tokenizer_path = p / "tokenizer"
if tokenizer_path.exists():
tokenizer = build_morph_tokenizer(tokenizer_path=str(tokenizer_path))
else:
print(" ⚠ No saved tokenizer found β€” building a fresh one (vocab won't match!)")
tokenizer = build_morph_tokenizer()
return cls(model=model, tokenizer=tokenizer, config=model.config,
device=dev, dtype=torch_dtype)
@staticmethod
def _apply_quantization(model, quantization: Optional[str], compute_dtype: torch.dtype):
if quantization == "4bit":
model = quantize_4bit(model, compute_dtype=compute_dtype)
elif quantization == "8bit":
model = quantize_8bit(model)
print_quantization_report(model, dtype_label=quantization or "float16")
return model
def _load_draft_model(self, path: str):
"""Load a small draft model for speculative decoding."""
try:
from model.architecture.morph_model import MorphForCausalLM
self.draft_model = MorphForCausalLM.from_saved(path, device=str(self.device))
self.draft_model.to(self.device, dtype=self.dtype).eval()
print(f" βœ“ Draft model loaded for speculative decoding: {path}")
except Exception as e:
print(f" ⚠ Draft model load failed ({e}) β€” speculative decoding disabled")
self.draft_model = None
def _free_vram(self):
free_memory()
# ── Core streaming generation ─────────────────────────────────
@torch.no_grad()
def generate_stream(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
max_new_tokens: int = 1024,
temperature: float = 0.7,
top_p: float = 0.9,
top_k: int = 50,
repetition_penalty: float = 1.1,
do_sample: bool = True,
pixel_values: Optional[torch.Tensor] = None,
) -> Generator[str, None, None]:
"""
Stream response text incrementally (yields new text deltas, not
the full accumulated string β€” fixes the 1.0 bug where every
yielded chunk re-decoded from scratch).
"""
prompt = apply_chat_template(
messages, self.tokenizer, add_generation_prompt=True,
system_prompt=system_prompt,
)
input_ids = self.tokenizer.encode(
prompt, return_tensors="pt", add_special_tokens=False,
).to(self.device)
if pixel_values is not None:
pixel_values = pixel_values.to(self.device, dtype=self.dtype)
past_key_values = None
current_ids = input_ids
generated_ids: List[int] = []
prev_text = ""
t0 = time.perf_counter()
for step in range(max_new_tokens):
with torch.amp.autocast("cuda", dtype=self.dtype, enabled=self.device.type == "cuda"):
forward_kwargs = dict(
input_ids=current_ids,
past_key_values=past_key_values,
use_cache=True,
)
if step == 0 and pixel_values is not None:
forward_kwargs["pixel_values"] = pixel_values
outputs = self.model(**forward_kwargs)
logits = outputs.logits[:, -1, :]
past_key_values = outputs.past_key_values
next_token = self._sample(
logits, generated_ids, temperature, top_p, top_k,
repetition_penalty, do_sample,
)
token_id = next_token.item()
generated_ids.append(token_id)
current_ids = next_token
full_text = self.tokenizer.decode(generated_ids, skip_special_tokens=True)
delta = full_text[len(prev_text):]
prev_text = full_text
if delta:
yield delta
if token_id == self.tokenizer.eos_token_id:
break
elapsed = time.perf_counter() - t0
tok_s = len(generated_ids) / max(elapsed, 1e-6)
print(f" [{len(generated_ids)} tokens in {elapsed:.2f}s, {tok_s:.1f} tok/s]")
def _sample(
self,
logits: torch.Tensor,
generated_ids: List[int],
temperature: float,
top_p: float,
top_k: int,
repetition_penalty: float,
do_sample: bool,
) -> torch.Tensor:
"""Shared sampling logic β€” used by both generate_stream and speculative decoding."""
logits = logits.clone()
if repetition_penalty != 1.0 and generated_ids:
for tid in set(generated_ids):
if logits[0, tid] < 0:
logits[0, tid] *= repetition_penalty
else:
logits[0, tid] /= repetition_penalty
if not do_sample:
return logits.argmax(dim=-1, keepdim=True)
if temperature != 1.0:
logits = logits / max(temperature, 1e-5)
if top_k > 0:
top_k_vals, _ = torch.topk(logits, min(top_k, logits.size(-1)))
logits[logits < top_k_vals[:, -1:]] = float("-inf")
if top_p < 1.0:
sorted_logits, sorted_idx = torch.sort(logits, descending=True)
cum_probs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)
remove = cum_probs - F.softmax(sorted_logits, dim=-1) > top_p
sorted_logits[remove] = float("-inf")
logits = torch.zeros_like(logits).scatter_(1, sorted_idx, sorted_logits)
probs = F.softmax(logits, dim=-1)
return torch.multinomial(probs, num_samples=1)
# ── Non-streaming convenience wrapper ─────────────────────────
def generate(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
max_new_tokens: int = 1024,
temperature: float = 0.7,
top_p: float = 0.9,
top_k: int = 50,
repetition_penalty: float = 1.1,
do_sample: bool = True,
pixel_values: Optional[torch.Tensor] = None,
) -> str:
"""Generate a full response (collects the stream)."""
chunks = list(self.generate_stream(
messages, system_prompt, max_new_tokens, temperature, top_p,
top_k, repetition_penalty, do_sample, pixel_values,
))
return "".join(chunks).strip()
# ── Session-based chat (persistent KV cache across turns) ────
@torch.no_grad()
def chat(
self,
session_id: str,
user_message: str,
system_prompt: Optional[str] = None,
max_new_tokens: int = 1024,
temperature: float = 0.7,
top_p: float = 0.9,
stream_callback=None,
) -> str:
"""
Multi-turn chat with a persistent per-session KV cache.
Only the NEW user message is tokenized and prefilled on each
call β€” prior turns are already resident in the session cache,
so a 20-turn conversation's 5th reply doesn't re-process turns
1-4 from scratch the way a stateless `generate()` call would.
Bridging note: the model's attention layers internally
concatenate past+new KV via `torch.cat` and return the full
tensor each call. We store only the newly-computed slice back
into the paged session cache (the "delta"), since the cache
already holds everything before this turn.
"""
cache = self.sessions.get_or_create(session_id)
is_first_turn = cache.seq_len == 0
if is_first_turn:
prompt = apply_chat_template(
[{"role": "user", "content": user_message}],
self.tokenizer, add_generation_prompt=True,
system_prompt=system_prompt,
)
else:
# Only encode the new turn β€” the system prompt + prior turns
# are already baked into the cached KV states.
prompt = (
f"<|start_header_id|>user<|end_header_id|>\n{user_message}<|eot_id|>"
f"<|start_header_id|>assistant<|end_header_id|>\n"
)
new_ids = self.tokenizer.encode(prompt, return_tensors="pt",
add_special_tokens=False).to(self.device)
# Build past_key_values list from the session cache. On a brand new
# session's very first step there's nothing cached yet, so we pass
# None (matches a fresh forward pass); every step after that reuses
# whatever the previous step returned.
has_prior_context = cache.seq_len > 0
if has_prior_context:
past_kv = [cache.get(i) for i in range(self.config.text.num_hidden_layers)]
else:
past_kv = None
generated_ids: List[int] = []
prev_text = ""
current_ids = new_ids
t0 = time.perf_counter()
for step in range(max_new_tokens):
with torch.amp.autocast("cuda", dtype=self.dtype, enabled=self.device.type == "cuda"):
outputs = self.model(
input_ids=current_ids,
past_key_values=past_kv,
use_cache=True,
)
logits = outputs.logits[:, -1, :]
new_past_kv = outputs.past_key_values
# Persist only the delta (newly computed tokens) into the paged cache.
# IMPORTANT: cache.update() writes at offset `cache.seq_len` but does
# NOT advance it β€” that's step()'s job. All layers must be written
# at the SAME offset (they process the same tokens in lockstep), so
# we call step() exactly once, after every layer has been updated.
delta_len = current_ids.shape[1]
for layer_idx, (k_full, v_full) in enumerate(new_past_kv):
k_new = k_full[:, :, -delta_len:, :]
v_new = v_full[:, :, -delta_len:, :]
cache.update(layer_idx, k_new, v_new)
cache.step(delta_len)
past_kv = new_past_kv
next_token = self._sample(
logits, generated_ids, temperature, top_p, top_k=50,
repetition_penalty=1.1, do_sample=True,
)
token_id = int(next_token.item())
generated_ids.append(token_id)
current_ids = next_token
if stream_callback is not None:
full_text = self.tokenizer.decode(generated_ids, skip_special_tokens=True)
delta = full_text[len(prev_text):]
prev_text = full_text
if delta:
stream_callback(delta)
if token_id == self.tokenizer.eos_token_id:
break
elapsed = time.perf_counter() - t0
response = self.tokenizer.decode(generated_ids, skip_special_tokens=True).strip()
print(f" [session={session_id[:8]}... | {len(generated_ids)} tok in {elapsed:.2f}s | "
f"cache={cache.seq_len} tok, {cache.memory_mb():.1f}MB]")
return response
def reset_session(self, session_id: str):
self.sessions.reset_session(session_id)
def end_session(self, session_id: str):
self.sessions.delete_session(session_id)
# ── Speculative decoding ───────────────────────────────────────
@torch.no_grad()
def generate_speculative(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
max_new_tokens: int = 512,
temperature: float = 0.7,
k: Optional[int] = None,
) -> Tuple[str, Dict]:
"""
Speculative decoding (Leviathan et al. / Chen et al. algorithm).
The small draft model proposes `k` tokens greedily; the main
model verifies all `k` in a single forward pass and accepts a
prefix via rejection sampling, guaranteeing the same output
distribution as sampling from the main model alone β€” just
fewer expensive forward passes through it.
Falls back to standard `generate()` if no draft model is loaded.
"""
if self.draft_model is None:
text = self.generate(messages, system_prompt, max_new_tokens, temperature)
return text, {"speculative": False, "acceptance_rate": None}
k = k or self.config.inference.speculative_k
prompt = apply_chat_template(messages, self.tokenizer, add_generation_prompt=True,
system_prompt=system_prompt)
input_ids = self.tokenizer.encode(prompt, return_tensors="pt",
add_special_tokens=False).to(self.device)
generated: List[int] = []
total_proposed, total_accepted = 0, 0
t0 = time.perf_counter()
current_ids = input_ids
main_past, draft_past = None, None
while len(generated) < max_new_tokens:
# 1. Draft model proposes k tokens
draft_tokens = []
draft_ids = current_ids
dpast = draft_past
for _ in range(k):
out = self.draft_model(input_ids=draft_ids, past_key_values=dpast, use_cache=True)
logits = out.logits[:, -1, :] / max(temperature, 1e-5)
probs = F.softmax(logits, dim=-1)
tok = torch.multinomial(probs, 1)
draft_tokens.append((tok, probs))
draft_ids = tok
dpast = out.past_key_values
proposed_ids = torch.cat([current_ids] + [t for t, _ in draft_tokens], dim=1)
# Length of the valid cache *before* this round (needed below to
# truncate away any rejected draft tokens' KV entries).
prior_cache_len = main_past[0][0].shape[2] if main_past is not None else 0
# 2. Main model verifies all k+1 positions in one forward pass
main_out = self.model(input_ids=proposed_ids, past_key_values=main_past, use_cache=True)
main_logits = main_out.logits[:, -(k + 1):, :] / max(temperature, 1e-5)
main_probs = F.softmax(main_logits, dim=-1)
# 3. Accept/reject each proposed token (standard spec-decoding test)
accepted = 0
for i, (draft_tok, draft_prob) in enumerate(draft_tokens):
# BUGFIX: this used to live at the end of the loop body, after
# the reject branch's `break` β€” so a rejected token never got
# counted, and total_proposed only ever counted acceptances.
# acceptance_rate (total_accepted/total_proposed) was
# therefore always ~100% regardless of real performance.
total_proposed += 1
tok_id = draft_tok.item()
p_main = main_probs[0, i, tok_id].item()
p_draft = draft_prob[0, tok_id].item()
accept_prob = min(1.0, p_main / max(p_draft, 1e-10))
if torch.rand(1).item() < accept_prob:
generated.append(tok_id)
accepted += 1
total_accepted += 1
else:
# Reject β€” resample from the residual distribution
residual = (main_probs[0, i] - draft_prob[0]).clamp(min=0)
residual = residual / residual.sum().clamp(min=1e-10)
resampled = torch.multinomial(residual, 1).item()
generated.append(resampled)
break
# If all k accepted, sample one more "bonus" token from the main model
if accepted == k:
bonus_probs = main_probs[0, k]
bonus_tok = torch.multinomial(bonus_probs, 1).item()
generated.append(bonus_tok)
current_ids = torch.tensor([[generated[-1]]], device=self.device)
# BUGFIX: main_out.past_key_values contains KV entries for the
# seed token + ALL k drafted tokens, but on an early rejection
# only `accepted` of those k draft tokens are actually part of
# the real sequence (the rest were proposals that got thrown
# out). Keeping the full cache left stale/phantom key-value
# entries in place for tokens that never happened, which both
# corrupts future attention (the model attends to keys for
# rejected tokens) and desyncs RoPE position ids (computed from
# cache length) from the true sequence length. Also, the
# resampled replacement token has no cache entry yet β€” it gets
# one on the next round's forward pass, same as the "bonus
# token" case. Truncate to seed(1) + accepted confirmed drafts.
valid_len = prior_cache_len + 1 + accepted
main_past = tuple(
(k_full[:, :, :valid_len, :], v_full[:, :, :valid_len, :])
for k_full, v_full in main_out.past_key_values
)
draft_past = None # simplification: draft cache rebuilt next round
if generated and generated[-1] == self.tokenizer.eos_token_id:
break
elapsed = time.perf_counter() - t0
text = self.tokenizer.decode(generated, skip_special_tokens=True).strip()
acceptance_rate = total_accepted / max(total_proposed, 1)
print(f" [speculative: {len(generated)} tok in {elapsed:.2f}s | "
f"acceptance={acceptance_rate:.1%}]")
return text, {"speculative": True, "acceptance_rate": round(acceptance_rate, 3),
"tokens": len(generated), "elapsed_s": round(elapsed, 2)}
# ── Image understanding ───────────────────────────────────────
def understand_image(self, image, question: str, max_new_tokens: int = 512) -> str:
"""Answer a question about an image."""
from PIL import Image as PILImage
if isinstance(image, str):
image = PILImage.open(image).convert("RGB")
pixel_values = self.model.vision_module.preprocess_image(image, device=self.device)
messages = [{"role": "user", "content": f"<|image|>\n{question}"}]
return self.generate(messages=messages, pixel_values=pixel_values,
max_new_tokens=max_new_tokens)
# ── Tool / generation-request parsing ──────────────────────────
def parse_tool_calls(self, response: str) -> List[Dict]:
"""Parse <|tool_call|>{...}<|/tool_call|> blocks from model output."""
import re, json
tool_calls = []
for match in re.findall(r"<\|tool_call\|>(.*?)<\|/tool_call\|>", response, re.DOTALL):
try:
tool_calls.append(json.loads(match.strip()))
except json.JSONDecodeError:
pass
return tool_calls
def parse_generation_requests(self, response: str) -> List[Dict]:
"""Parse <|gen_image|>/<|gen_video|>/<|gen_3d|> blocks from model output."""
import re
requests = []
for gen_type in ("image", "video", "3d"):
for match in re.findall(rf"<\|gen_{gen_type}\|>(.*?)<\|/gen\|>", response, re.DOTALL):
requests.append({"type": gen_type, "prompt": match.strip()})
return requests
def parse_thinking(self, response: str) -> Tuple[Optional[str], str]:
"""Split <|think|>...<|/think|> reasoning from the visible answer."""
import re
match = re.search(r"<\|think\|>(.*?)<\|/think\|>", response, re.DOTALL)
if not match:
return None, response
thinking = match.group(1).strip()
answer = response[:match.start()] + response[match.end():]
return thinking, answer.strip()
# ── Batch inference ───────────────────────────────────────────
@torch.no_grad()
def batch_generate(
self, prompts: List[str], max_new_tokens: int = 512, temperature: float = 0.7,
) -> List[str]:
"""Generate responses for multiple independent prompts at once."""
encodings = self.tokenizer(
prompts, return_tensors="pt", padding=True, truncation=True, max_length=4096,
).to(self.device)
output_ids = self.model.language_model.generate(
input_ids=encodings["input_ids"],
attention_mask=encodings["attention_mask"],
max_new_tokens=max_new_tokens,
temperature=temperature,
do_sample=True,
eos_token_id=self.tokenizer.eos_token_id,
pad_token_id=self.tokenizer.pad_token_id,
)
responses = []
for out in output_ids:
new_tokens = out[encodings["input_ids"].shape[1]:]
responses.append(self.tokenizer.decode(new_tokens, skip_special_tokens=True))
return responses
# ── Embeddings (for memory / RAG tools) ────────────────────────
@torch.no_grad()
def embed(self, text: str, max_length: int = 512) -> List[float]:
"""
Produce an embedding vector using Morph's own hidden states β€”
no external embedding API. Mean-pools the last transformer
layer's hidden states over non-padding tokens, then L2-normalizes
so downstream cosine-similarity search is a plain dot product.
This is deliberately simple (no dedicated embedding head/training
objective) rather than a from-scratch contrastive embedding
model β€” good enough for the in-repo memory/RAG tools, and the
production backend architecture's Qdrant-based RAG pipeline can
swap in a dedicated embedding model later without changing the
tool interface.
"""
tokens = self.tokenizer(
text, return_tensors="pt", truncation=True, max_length=max_length,
).to(self.device)
with torch.amp.autocast("cuda", dtype=self.dtype, enabled=self.device.type == "cuda"):
outputs = self.model.language_model(
input_ids=tokens["input_ids"],
attention_mask=tokens.get("attention_mask"),
output_hidden_states=True,
use_cache=False,
)
last_hidden = outputs.hidden_states[-1].float() # [1, S, H]
mask = tokens.get("attention_mask")
if mask is not None:
mask = mask.unsqueeze(-1).float() # [1, S, 1]
pooled = (last_hidden * mask).sum(dim=1) / mask.sum(dim=1).clamp(min=1e-6)
else:
pooled = last_hidden.mean(dim=1)
pooled = torch.nn.functional.normalize(pooled, p=2, dim=-1)
return pooled.squeeze(0).cpu().tolist()
def embed_batch(self, texts: List[str], max_length: int = 512) -> List[List[float]]:
"""Convenience loop over embed() β€” fine for tool-scale batches (docs/memories, not bulk indexing)."""
return [self.embed(t, max_length=max_length) for t in texts]
# ── Diagnostics ──────────────────────────────────────────────
def stats(self) -> Dict:
return {
"device": str(self.device),
"dtype": str(self.dtype),
"vram_tier": self.vram_state,
"vram_free_gb": (torch.cuda.mem_get_info()[0] / 1e9) if torch.cuda.is_available() else None,
"sessions": self.sessions.stats(),
"speculative_decoding": self.draft_model is not None,
"params": self.model.param_count(),
}