Files attached to chat messages.
POST /v1/attachments reads uploaded files and returns their content for
the client to send with a chat message: extracted text for documents, a
resized JPEG for images. Nothing is stored on the server. The chat route then
folds that content into the prompt with :func:render_attachments, within a
size budget so large documents can't push the question out of a local model's
context window.
Text extraction uses only the standard library plus pdfplumber (PDF) and
Pillow (images), so every format below works in the desktop runtime:
- PDF; Office Open XML (
.docx, .pptx, .xlsx); OpenDocument
(.odt, .ods, .odp); EPUB; HTML; RTF; Jupyter notebooks
- Anything that decodes as text: plain text, Markdown, CSV, JSON, XML, YAML,
logs, source code, and so on
- Images Pillow can open (PNG, JPEG, GIF, WebP, BMP, TIFF, ICO, ...)
Other binary files are accepted and described by name, type and size so the
model can at least tell the user what it received.
Classes
Bases: BaseModel
A document attached to a chat message (see server/attachments.py).
Bases: BaseModel
What the client gets back for one uploaded file.
Functions
extract_text(name: str, data: bytes) -> tuple[Optional[str], str, Optional[str]]
Return (text, format_label, note) for a non-image file.
text is None when the file's content can't be read; note then
says why, in words the model can relay to the user.
Source code in src/locus/server/attachments.py
| def extract_text(name: str, data: bytes) -> tuple[Optional[str], str, Optional[str]]:
"""Return ``(text, format_label, note)`` for a non-image file.
``text`` is ``None`` when the file's content can't be read; ``note`` then
says why, in words the model can relay to the user.
"""
ext = _extension(name)
if data.startswith(b"%PDF") or ext == ".pdf":
text = _pdf_text(data)
note = (
None
if text
else "The PDF has no selectable text (it may be a scanned image)."
)
return text, "PDF", note
if data.startswith(_ZIP_MAGIC) and ext in _ZIP_FORMATS:
label, reader = _ZIP_FORMATS[ext]
with zipfile.ZipFile(io.BytesIO(data)) as zf:
return reader(zf), label, None
if ext in _LEGACY_OFFICE:
return (
None,
_LEGACY_OFFICE[ext],
f"Locus can't read the old {_LEGACY_OFFICE[ext]} format. "
"Save it as a newer file "
f"(for example .{ext[1:]}x) and attach it again.",
)
if ext == ".ipynb":
return _notebook_text(data), "Jupyter notebook", None
text = _decode_text(data)
if text is None:
label = f"{ext[1:].upper()} file" if ext else "binary file"
return (
None,
label,
"This is a binary file, so its content can't be read as text.",
)
if ext in {".html", ".htm", ".xhtml"} or text.lstrip()[:15].lower().startswith(
("<!doctype html", "<html")
):
return html_to_text(text), "web page", None
if ext == ".rtf" or text.startswith("{\\rtf"):
return _rtf_text(text), "RTF document", None
label = f"{ext[1:].upper()} file" if ext else "text"
return text, label, None
|
prepare_image(data: bytes) -> Optional[tuple[str, int, int, str]]
Return (base64_jpeg, width, height, format), or None if not an image.
Source code in src/locus/server/attachments.py
| def prepare_image(data: bytes) -> Optional[tuple[str, int, int, str]]:
"""Return ``(base64_jpeg, width, height, format)``, or ``None`` if not an image."""
try:
from PIL import Image, ImageOps
except ImportError:
return None
try:
with Image.open(io.BytesIO(data)) as img:
fmt = img.format or "image"
img.seek(0) # first frame of animations
img = ImageOps.exif_transpose(img)
if img.mode in ("RGBA", "LA", "P"):
img = img.convert("RGBA")
background = Image.new("RGB", img.size, (255, 255, 255))
background.paste(img, mask=img.getchannel("A"))
img = background
else:
img = img.convert("RGB")
img.thumbnail((IMAGE_MAX_SIDE, IMAGE_MAX_SIDE))
out = io.BytesIO()
img.save(out, format="JPEG", quality=85)
return (
base64.b64encode(out.getvalue()).decode("ascii"),
img.width,
img.height,
fmt,
)
except Exception:
return None
|
read_attachment(name: str, data: bytes) -> Attachment
Turn one uploaded file into an :class:Attachment.
Source code in src/locus/server/attachments.py
| def read_attachment(name: str, data: bytes) -> Attachment:
"""Turn one uploaded file into an :class:`Attachment`."""
image = prepare_image(data)
if image is not None:
b64, width, height, fmt = image
return Attachment(
name=name,
kind="image",
format=f"{fmt} image",
size=len(data),
image=b64,
width=width,
height=height,
)
try:
text, label, note = extract_text(name, data)
except Exception as exc: # a damaged file shouldn't fail the whole upload
logger.info("Could not read attachment %s: %s", name, exc)
return Attachment(
name=name,
kind="unsupported",
format=_extension(name)[1:].upper() or "file",
size=len(data),
note=(
f"The file couldn't be read ({type(exc).__name__}). It may be damaged."
),
)
if text is None:
return Attachment(
name=name, kind="unsupported", format=label, size=len(data), note=note
)
truncated = len(text) > MAX_TEXT_CHARS
return Attachment(
name=name,
kind="document",
format=label,
size=len(data),
text=text[:MAX_TEXT_CHARS],
truncated=truncated,
note=note,
)
|
async
read_attachments(files: List[UploadFile] = File(...)) -> AttachmentsResponse
Read uploaded files for a chat message. Nothing is stored.
Source code in src/locus/server/attachments.py
| @router.post("", response_model=AttachmentsResponse)
async def read_attachments(files: List[UploadFile] = File(...)) -> AttachmentsResponse:
"""Read uploaded files for a chat message. Nothing is stored."""
if len(files) > MAX_FILES:
raise HTTPException(
status_code=400, detail=f"Attach at most {MAX_FILES} files at a time."
)
out: List[Attachment] = []
for upload in files:
name = upload.filename or "file"
data = await upload.read(MAX_FILE_BYTES + 1)
if len(data) > MAX_FILE_BYTES:
raise HTTPException(
status_code=413,
detail=f"{name} is larger than {MAX_FILE_BYTES // (1024 * 1024)} MB.",
)
out.append(read_attachment(name, data))
return AttachmentsResponse(attachments=out)
|
render_attachments(attachments: Sequence[MessageAttachment], budget: int) -> tuple[str, int]
Render documents as a prompt block. Returns (block, chars_used).
Source code in src/locus/server/attachments.py
| def render_attachments(
attachments: Sequence[MessageAttachment], budget: int
) -> tuple[str, int]:
"""Render documents as a prompt block. Returns ``(block, chars_used)``."""
if not attachments:
return "", 0
parts: List[str] = []
used = 0
for att in attachments:
header = f'<attached_file name="{att.name}" format="{att.format}">'
if att.note and not att.text:
parts.append(f"{header}\n[{att.note}]\n</attached_file>")
continue
room = max(budget - used, 0)
text = att.text[:room]
used += len(text)
shortened = att.truncated or len(text) < len(att.text)
body = text
if shortened:
body += (
"\n[... the rest of this file was cut to fit the model's context ...]"
)
if att.note:
body = f"[{att.note}]\n{body}"
parts.append(f"{header}\n{body}\n</attached_file>")
return "The user attached these files:\n\n" + "\n\n".join(parts), used
|
contents_with_attachments(messages: Sequence[Any], budget: int = PROMPT_BUDGET_CHARS) -> List[str]
Each message's content with its documents folded in ahead of it.
Newer messages get first claim on budget, so the latest files are sent
in full when the conversation already carries older ones. The messages
themselves aren't changed.
Source code in src/locus/server/attachments.py
| def contents_with_attachments(
messages: Sequence[Any], budget: int = PROMPT_BUDGET_CHARS
) -> List[str]:
"""Each message's content with its documents folded in ahead of it.
Newer messages get first claim on ``budget``, so the latest files are sent
in full when the conversation already carries older ones. The messages
themselves aren't changed.
"""
contents = [getattr(m, "content", "") or "" for m in messages]
remaining = budget
for i in range(len(messages) - 1, -1, -1):
attachments = getattr(messages[i], "attachments", None)
if not attachments:
continue
block, used = render_attachments(attachments, remaining)
remaining -= used
contents[i] = f"{block}\n\n{contents[i]}" if contents[i] else block
return contents
|