import html
import json
import os
import time
import uuid
from pathlib import Path
os.environ["GRADIO_SSR_MODE"] = "false"
import fastapi
import spaces
import torch
import gradio as gr
from fastapi.responses import JSONResponse, StreamingResponse
from gradio import Server
from gradio.context import LocalContext
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
try:
from gradio.route_utils import Request as GradioRequest
except ImportError:
GradioRequest = None
from pasita import infer_kind, convert_to_markdown
MODEL_ID = "OpceanAI/PASITA"
MODEL_CARD = "https://huggingface.co/OpceanAI/PASITA"
MAX_INPUT_CHARS = 60_000
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
tokenizer.padding_side = "left"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
model = AutoModelForCausalLM.from_pretrained(MODEL_ID, dtype="auto").to(DEVICE)
model.eval()
EMPTY_MD = "No output yet. Paste text on the left, then press Convert."
DETAILS_EMPTY = "Run details will appear here: route, numbers kept, coverage and time."
GLYPH = (
''
)
def _status(text: str, state: str = "ready", detail: str = "") -> str:
detail_html = f'{html.escape(detail)}' if detail else ""
role = 'role="alert"' if state == "error" else 'role="status" aria-live="polite"'
return (
f'
'
f'{html.escape(text)}{detail_html}
'
)
def _details_md(result: dict | None, seconds: float, error: str | None = None) -> str:
if error is not None:
return f"Could not convert.\n\n{error}\n\nShorten very long inputs and try again."
if result is None:
return DETAILS_EMPTY
metrics = result["metrics"]
routes = {
"model": "PASITA + rerank",
"structured": "HTML/table converter",
"passthrough": "Source passthrough",
}
lines = [
f"**Route:** {routes.get(result.get('route'), result.get('route'))}",
f"**Numbers kept:** {metrics['numbers_matched']} of {metrics['numbers_in']}",
]
missing = metrics.get("missing_number_keys") or []
if missing:
shown = ", ".join(f"`{x}`" for x in missing[:10])
more = f" (+{len(missing) - 10} more)" if len(missing) > 10 else ""
lines.append(f"**Missing numbers:** {shown}{more} (check the output)")
lines += [
f"**Content coverage:** {metrics['content_overlap']:.0%}",
f"**Time:** {seconds:.1f}s",
f"**Candidates:** {len(result.get('candidates', []))}",
]
return "\n\n".join(lines)
def _prune_downloads(max_age_seconds: int = 3600) -> None:
try:
from gradio.utils import get_cache_folder
now = time.time()
for path in Path(get_cache_folder()).glob("pasita-*.md"):
try:
if now - path.stat().st_mtime > max_age_seconds:
path.unlink()
except OSError:
pass
except Exception:
pass
def _write_download(markdown: str | None) -> str | None:
if not markdown or not markdown.strip():
return None
try:
from gradio.utils import get_cache_folder
_prune_downloads()
folder = Path(get_cache_folder())
folder.mkdir(parents=True, exist_ok=True)
path = folder / f"pasita-{uuid.uuid4().hex[:12]}.md"
path.write_text(markdown, encoding="utf-8")
return str(path)
except Exception:
return None
def _counter(text: str) -> str:
text = text or ""
words = len(text.split())
return (
'PLAIN TEXT'
f'{len(text):,} CHARS / {words:,} WORDS'
)
def _clamp_candidates(num_candidates: int | None) -> int:
if num_candidates is None:
return 3
try:
return max(1, min(3, int(num_candidates)))
except (TypeError, ValueError):
return 3
def _normalize(text: str | None, num_candidates: int | None, structured_fallback: bool | None) -> tuple[str, int, bool]:
text = (text or "").strip()
if structured_fallback is None:
structured_fallback = True
return text, _clamp_candidates(num_candidates), bool(structured_fallback)
def _estimate_duration(text, num_candidates=3, structured_fallback=True, max_output_tokens=None, *args, **kwargs):
text = text or ""
if structured_fallback and infer_kind(text.strip()) in ("html", "data"):
return 15
candidates = _clamp_candidates(num_candidates if not isinstance(num_candidates, bool) else 3)
try:
cap = max(16, min(512, int(max_output_tokens))) if max_output_tokens is not None else 512
except (TypeError, ValueError):
cap = 512
extra = max(0, cap - 210) // 100
return min(30, max(15, 12 + len(text) // 200 + 4 * (candidates - 1) + extra))
def _run_harness(text: str | None, num_candidates: int | None, structured_fallback: bool | None, max_new_cap: int | None = None) -> dict:
text, num_candidates, structured_fallback = _normalize(text, num_candidates, structured_fallback)
if not text:
raise ValueError("Paste a document to convert.")
hard_cap = 512
if max_new_cap is not None:
try:
hard_cap = max(16, min(512, int(max_new_cap)))
except (TypeError, ValueError):
hard_cap = 512
started = time.perf_counter()
result = convert_to_markdown(
model,
tokenizer,
text,
num_candidates=num_candidates,
do_sample=True,
temperature=0.3,
top_p=0.95,
seed=0,
use_cache=True,
max_new_tokens="auto",
hard_cap=hard_cap,
structured_fallback=structured_fallback,
skip_model_for_structured=True,
)
result["elapsed"] = time.perf_counter() - started
return result
@spaces.GPU(duration=_estimate_duration)
def _gpu_infer(text: str | None = None, num_candidates: int | None = None, structured_fallback: bool | None = None, max_output_tokens: int | None = None) -> dict:
"""Run the PASITA harness on the GPU worker and return the full result dict."""
return _run_harness(text, num_candidates, structured_fallback, max_output_tokens)
def _friendly_error(exc: Exception) -> tuple[str, str, str, str, None]:
err_type = type(exc).__name__
text_lower = str(exc).lower()
if "zerogpu" in text_lower and any(k in text_lower for k in ("quota", "exhausted", "insufficient", "limit")):
plain = "The GPU quota ran out. Wait a bit or sign in, then try again."
elif any(k in text_lower for k in ("illegal duration", "task aborted", "queue is full", "queue full")):
plain = "The GPU was too busy. Shorten the input or try again in a bit."
else:
plain = f"Something failed on the server ({err_type}). Try again."
return (
f"> Could not convert.\n>\n> {plain}",
"",
_status(f"ERROR / {err_type}", "error"),
_details_md(None, 0.0, error=f"{plain} Technical detail: {err_type}: {exc}"),
None,
)
def convert(text: str | None = None, num_candidates: int | None = None, structured_fallback: bool | None = None, max_output_tokens: int | None = None) -> tuple[str, str, str, str, str | None]:
"""Convert a plain-text document into faithful GitHub-Flavored Markdown.
PASITA samples the document several times, every candidate is post-processed and
scored for number fidelity, content coverage and GFM validity, and the best one is
returned. Inputs detected as HTML or delimited data are converted deterministically,
and a run that loses source numbers falls back to a faithful passthrough of the input.
Args:
text: The document to convert, for example OCR output, pasted HTML, meeting notes or a report.
num_candidates: How many PASITA samples to generate and rerank, from 1 to 3.
structured_fallback: Use the deterministic converter for HTML and delimited data.
max_output_tokens: Optional cap on generated tokens. When omitted, the length is chosen automatically.
Returns:
A tuple of the rendered Markdown, the raw Markdown, a status line, run details
and a download path for the Markdown file (or None).
"""
text = (text or "").strip()
if not text:
return EMPTY_MD, "", _status("READY / PASTE A DOCUMENT TO CONVERT"), DETAILS_EMPTY, None
if len(text) > MAX_INPUT_CHARS:
message = f"Input too long ({len(text):,} chars). The limit is {MAX_INPUT_CHARS:,} characters."
return (
f"> {message} Shorten the document and try again.",
"",
_status("ERROR / INPUT TOO LONG", "error"),
_details_md(None, 0.0, error=message),
None,
)
try:
result = _gpu_infer(text, num_candidates, structured_fallback, max_output_tokens)
metrics = result["metrics"]
numbers = f"{metrics['numbers_matched']}/{metrics['numbers_in']} NUMBERS"
overlap = f"{metrics['content_overlap']:.0%} CONTENT"
routes = {
"model": "PASITA + RERANK",
"structured": "HTML/TABLE CONVERTER",
"passthrough": "SOURCE PASSTHROUGH",
}
source_label = routes.get(result.get("route"), "PASITA + RERANK")
taken = result["seconds"] if result["seconds"] else result["elapsed"]
detail = f"{source_label} / {numbers} / {overlap} / {taken:.1f}S"
if result.get("strategy", {}).get("truncated_input"):
detail += " / INPUT TRUNCATED TO 2048 TOKENS"
return result["markdown"], result["markdown"], _status("DONE", "ready", detail), _details_md(result, taken), _write_download(result["markdown"])
except Exception as exc:
return _friendly_error(exc)
@spaces.GPU(duration=_estimate_duration)
def gpu_convert(text: str | None = None, num_candidates: int | None = None, structured_fallback: bool | None = None, max_output_tokens: int | None = None) -> dict:
"""Convert a document and return the full conversion result.
Same pipeline as the UI conversion. Used by the /v1 OpenAI-style API routes.
Args:
text: The document to convert.
num_candidates: How many PASITA samples to generate and rerank, from 1 to 3.
structured_fallback: Use the deterministic converter for HTML and delimited data.
max_output_tokens: Optional cap on generated tokens. When omitted, the length is chosen automatically.
Returns:
The full result dict: markdown, route, metrics, timing and candidate info.
"""
return _run_harness(text, num_candidates, structured_fallback, max_output_tokens)
API_DESCRIPTION = """
OpenAI-style REST API for [PASITA](https://huggingface.co/OpceanAI/PASITA), an 88M
from-scratch model that turns plain text into faithful Markdown. No API key is required.
Works with the OpenAI SDK by pointing `base_url` at `/v1`. The Gradio API and the MCP
server stay available at `?view=api`.
Honored request fields: `model`, `messages`, `stream`, `stream_options.include_usage`,
`max_tokens` (upper bound), `candidates` (1-3), `structured_fallback`. Sampling
temperature is fixed at 0.3 because that is the measured recipe; other OpenAI fields
are accepted but ignored.
""".strip()
def _api_error(message: str, error_type: str = "invalid_request_error", status: int = 400, code: str | None = None, headers: dict | None = None) -> JSONResponse:
return JSONResponse(
status_code=status,
content={"error": {"message": message, "type": error_type, "param": None, "code": code}},
headers=headers,
)
def _exception_to_error(exc: Exception) -> JSONResponse:
message = f"{type(exc).__name__}: {exc}"
text = message.lower()
if "zerogpu" in text and any(k in text for k in ("quota", "exhausted", "insufficient", "limit")):
return _api_error(str(exc), "rate_limit_error", status=429, code="gpu_quota_exceeded", headers={"Retry-After": "60"})
if any(k in text for k in ("illegal duration", "task aborted", "queue is full", "queue full")):
return _api_error(str(exc), "rate_limit_error", status=429, code="gpu_busy", headers={"Retry-After": "30"})
return _api_error(message, "server_error", status=500)
def _with_request_context(request: fastapi.Request, text: str, candidates: int, fallback: bool, max_tokens: int | None = None) -> dict:
if GradioRequest is not None:
gr_request = GradioRequest(request=request, session_hash="api")
token = LocalContext.request.set(gr_request)
try:
return gpu_convert(text, candidates, fallback, max_tokens)
finally:
LocalContext.request.reset(token)
return gpu_convert(text, candidates, fallback, max_tokens)
def _token_counts(text: str, markdown: str, prompt_tokens: int | None = None) -> tuple[int, int]:
if not prompt_tokens:
prompt_tokens = len(tokenizer(text, add_special_tokens=True)["input_ids"])
completion_tokens = len(tokenizer(markdown, add_special_tokens=False)["input_ids"])
return prompt_tokens, completion_tokens
class ChatMessage(BaseModel):
role: str = "user"
content: str | list | None = None
class ChatCompletionRequest(BaseModel):
model: str | None = "pasita-v1"
messages: list[ChatMessage]
stream: bool = False
stream_options: dict | None = None
max_tokens: int | None = None
candidates: int = 3
structured_fallback: bool = True
class ConvertRequest(BaseModel):
text: str
candidates: int = 3
structured_fallback: bool = True
max_tokens: int | None = None
server = Server(title="PASITA API", version="1.0.0", description=API_DESCRIPTION)
_V2_CONVERT_PATHS = frozenset({"/gradio_api/call/v2/convert", "/gradio_api/call/v2/convert/"})
_V1_CONVERT_PATHS = frozenset({"/gradio_api/call/convert", "/gradio_api/call/convert/"})
_QUEUE_JOIN_PATHS = frozenset({"/gradio_api/queue/join", "/gradio_api/queue/join/"})
_CONVERT_API_NAMES = frozenset({"convert", "convert_1"})
def _is_convert_fn(scope, fn_index) -> bool:
try:
app = scope.get("app")
if app is None or not hasattr(app, "get_blocks"):
return False
blocks = app.get_blocks()
fns = getattr(blocks, "fns", None)
if not fns or not isinstance(fn_index, int) or isinstance(fn_index, bool):
return False
if not 0 <= fn_index < len(fns):
return False
return getattr(fns[fn_index], "api_name", None) in _CONVERT_API_NAMES
except Exception:
return False
_CONVERT_PARAMS = (("text", ""), ("num_candidates", 3), ("structured_fallback", True), ("max_output_tokens", None))
def _clean_candidates(value) -> int:
try:
ivalue = int(value)
except (TypeError, ValueError):
return 3
return ivalue if 1 <= ivalue <= 3 else 3
def _clean_flag(value) -> bool:
if value is None or value == "":
return True
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in ("true", "1", "yes", "on"):
return True
if lowered in ("false", "0", "no", "off"):
return False
return True
def _clean_convert_value(name: str, value):
if name == "num_candidates":
return _clean_candidates(value)
if name == "structured_fallback":
return _clean_flag(value)
if name == "max_output_tokens":
if value is None or value == "":
return None
try:
ivalue = int(value)
except (TypeError, ValueError):
return None
return max(16, min(512, ivalue))
return value
class _LegacyConvertBodyMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope.get("type") != "http" or scope.get("method") != "POST":
await self.app(scope, receive, send)
return
path = scope.get("path", "")
is_v2 = path in _V2_CONVERT_PATHS
is_v1 = path in _V1_CONVERT_PATHS
is_queue = path in _QUEUE_JOIN_PATHS
if not (is_v2 or is_v1 or is_queue):
await self.app(scope, receive, send)
return
body = b""
while True:
message = await receive()
body += message.get("body", b"")
if not message.get("more_body"):
break
try:
parsed = json.loads(body.decode("utf-8")) if body else None
except Exception:
parsed = None
if isinstance(parsed, dict):
if is_queue and not _is_convert_fn(scope, parsed.get("fn_index")):
pass
else:
if is_v2 and "data" in parsed and "text" not in parsed:
data = parsed.get("data")
if isinstance(data, list):
named: dict = {}
for (name, _default), value in zip(_CONVERT_PARAMS, data):
named[name] = value
for key in ("session_hash", "event_id", "fn_index", "trigger_id", "batched"):
if key in parsed:
named[key] = parsed[key]
parsed = named
if is_v2:
for name, default in _CONVERT_PARAMS:
if name in parsed and (parsed[name] is None or parsed[name] == ""):
parsed[name] = default
for name in ("num_candidates", "structured_fallback", "max_output_tokens"):
if name in parsed:
parsed[name] = _clean_convert_value(name, parsed[name])
elif isinstance(parsed.get("data"), list):
data = list(parsed["data"])
while len(data) < len(_CONVERT_PARAMS):
data.append(None)
cleaned = []
for (name, default), value in zip(_CONVERT_PARAMS, data):
if value is None or value == "":
cleaned.append(default if name != "text" else "")
else:
cleaned.append(_clean_convert_value(name, value))
parsed = dict(parsed)
parsed["data"] = cleaned
body = json.dumps(parsed, default=str).encode("utf-8")
async def receive_once():
return {"type": "http.request", "body": body, "more_body": False}
headers = [(k, v) for k, v in scope.get("headers", []) if k.lower() != b"content-length"]
headers.append((b"content-length", str(len(body)).encode("latin-1")))
scope["headers"] = headers
await self.app(scope, receive_once, send)
server.add_middleware(_LegacyConvertBodyMiddleware)
try:
from fastapi.middleware.cors import CORSMiddleware
server.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "OPTIONS"],
allow_headers=["*"],
)
except Exception:
pass
@server.exception_handler(fastapi.exceptions.RequestValidationError)
async def validation_error_handler(request: fastapi.Request, exc: fastapi.exceptions.RequestValidationError):
issues = "; ".join(f"{'.'.join(str(l) for l in e.get('loc', []))}: {e.get('msg', '')}" for e in exc.errors()[:3])
return _api_error(f"Invalid request body: {issues}", status=400)
@server.get("/v1/models")
def list_models() -> dict:
return {
"object": "list",
"data": [{"id": "pasita-v1", "object": "model", "created": 1760000000, "owned_by": "OpceanAI"}],
}
@server.get("/v1/health")
def health() -> dict:
return {"status": "ok", "model": "pasita-v1"}
def _validate(body: ChatCompletionRequest | ConvertRequest, text: str | None) -> JSONResponse | None:
if text is None or not text.strip():
return _api_error("No input text was provided.", code="empty_input")
if len(text) > MAX_INPUT_CHARS:
return _api_error(
f"The document is too long ({len(text):,} chars). The limit is {MAX_INPUT_CHARS:,} characters.",
code="input_too_long",
)
candidates = getattr(body, "candidates", 3)
try:
candidates = int(candidates)
except (TypeError, ValueError):
return _api_error("candidates must be between 1 and 3.", code="invalid_candidates")
if not 1 <= candidates <= 3:
return _api_error("candidates must be between 1 and 3.", code="invalid_candidates")
return None
def _result_payload(result: dict, text: str, started: float) -> dict:
markdown = result["markdown"]
prompt_tokens, completion_tokens = _token_counts(text, markdown, result.get("prompt_tokens"))
return {
"markdown": markdown,
"route": result.get("route", "model"),
"metrics": result["metrics"],
"candidates": len(result.get("candidates", [])),
"seconds": round(result["seconds"] or result["elapsed"], 2),
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
},
"elapsed": round(time.perf_counter() - started, 2),
}
@server.post("/v1/convert", response_model=None)
def v1_convert(body: ConvertRequest, request: fastapi.Request) -> dict | JSONResponse:
error = _validate(body, body.text)
if error is not None:
return error
started = time.perf_counter()
try:
result = _with_request_context(request, body.text.strip(), int(body.candidates), bool(body.structured_fallback), body.max_tokens)
except Exception as exc:
return _exception_to_error(exc)
payload = _result_payload(result, body.text.strip(), started)
payload.update({"id": f"conv-{uuid.uuid4().hex[:24]}", "object": "pasita.conversion", "created": int(time.time()), "model": "pasita-v1"})
return payload
def _extract_user_text(messages: list[ChatMessage]) -> str | None:
for message in reversed(messages):
if message.role != "user":
continue
content = message.content
if isinstance(content, str):
return content
if isinstance(content, list):
parts = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
parts.append(str(part.get("text", "")))
elif isinstance(part, str):
parts.append(part)
joined = "\n".join(p for p in parts if p)
if joined:
return joined
continue
continue
return None
def reset() -> tuple[str, str, str, str, str | None, str]:
return EMPTY_MD, "", _status("READY"), DETAILS_EMPTY, None, _counter("")
@server.post("/v1/chat/completions", response_model=None)
def chat_completions(body: ChatCompletionRequest, request: fastapi.Request) -> dict | JSONResponse | StreamingResponse:
text = _extract_user_text(body.messages)
error = _validate(body, text)
if error is not None:
return error
started = time.perf_counter()
try:
result = _with_request_context(request, text.strip(), int(body.candidates), bool(body.structured_fallback), body.max_tokens)
except Exception as exc:
return _exception_to_error(exc)
payload = _result_payload(result, text.strip(), started)
completion_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(time.time())
markdown = payload["markdown"]
if body.stream:
include_usage = bool((body.stream_options or {}).get("include_usage"))
def sse():
def chunk(delta, finish=None, usage=None):
data = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": created,
"model": "pasita-v1",
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
}
if usage is not None:
data["usage"] = usage
return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
yield chunk({"role": "assistant", "content": ""})
yield chunk({"content": markdown})
yield chunk({}, finish="stop", usage=payload["usage"] if include_usage else None)
yield "data: [DONE]\n\n"
return StreamingResponse(
sse(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
return {
"id": completion_id,
"object": "chat.completion",
"created": created,
"model": "pasita-v1",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": markdown},
"logprobs": None,
"finish_reason": "stop",
}
],
"usage": payload["usage"],
"pasita": {
"route": payload["route"],
"metrics": payload["metrics"],
"candidates": payload["candidates"],
"seconds": payload["seconds"],
},
}
PAINT_JS = """(text, num_candidates, structured_fallback) => {
const n = Number(num_candidates) || 1;
const label = n > 1 ? 'PASITA ' + n + ' SAMPLES / RERANK' : 'PASITA SAMPLE / RERANK';
return ['Converting…', '',
'
"""
FOOTER = f"""
"""
EXAMPLES = [
[
"""RESUMEN MENSUAL DE OPERACIONES: MARZO 2026
Resumen ejecutivo
Los pedidos gestionados alcanzaron 18.420 unidades, un 6,1% más que en febrero. El tiempo medio de preparación bajó a 2,4 horas.
Detalle por almacén
Madrid: 8.120 pedidos (+7,4%)
Barcelona: 5.480 pedidos (+3,9%)
Lisboa: 4.820 pedidos (+6,7%)
Incidencias
Se registraron 96 incidencias de embalaje, el 0,52% del total. La previsión para abril es de 19.100 pedidos."""
],
[
"""Team sync, 2026-08-14
Attendees: Ana, Marc, Priya, Tom
Decisions
1. Ship the new onboarding flow on Sept 2.
2. Postpone the pricing experiment to Q4.
3. Marc owns the migration checklist; due Aug 22.
Open questions
- Do we need a second staging cluster?
- Who signs off on the SOC2 evidence?"""
],
[
"""
Shipping policy
Orders placed before 2:00 PM CET ship the same business day.
Standard delivery: 3 to 5 business days
Express delivery: 1 business day
Free shipping over 49 EUR
Returns are accepted within 30 days with the original packaging.