feat(ai): commandes @/ & skills, analyse d'images et capacites des modeles (#81)
This commit is contained in:
@@ -5,9 +5,11 @@ redaction, builds a system prompt with file contents, and caches results
|
||||
for repeated queries.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
@@ -21,6 +23,10 @@ logger = logging.getLogger("obsigate.bookslm")
|
||||
BOOKSLM_MAX_FILES = int(os.getenv("BOOKSLM_MAX_FILES", "200"))
|
||||
BOOKSLM_MAX_TOTAL_CHARS = int(os.getenv("BOOKSLM_MAX_TOTAL_CHARS", "200000"))
|
||||
BOOKSLM_MAX_FILE_CHARS = int(os.getenv("BOOKSLM_MAX_FILE_CHARS", "30000"))
|
||||
# Maximum size of an image sent to a vision model (bytes, before base64).
|
||||
BOOKSLM_MAX_IMAGE_BYTES = int(os.getenv("BOOKSLM_MAX_IMAGE_BYTES", "10000000"))
|
||||
|
||||
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".avif"}
|
||||
|
||||
# ── Cache ──
|
||||
_cache: dict[str, dict[str, Any]] = {}
|
||||
@@ -260,6 +266,122 @@ def collect_files_context(
|
||||
}
|
||||
|
||||
|
||||
def collect_adhoc_context(
|
||||
vault_path: Path,
|
||||
files: list[str] | None = None,
|
||||
directories: list[str] | None = None,
|
||||
scope: str = "general",
|
||||
) -> dict[str, Any]:
|
||||
"""Collect an ad-hoc mix of explicit files and directories.
|
||||
|
||||
Backs the assistant ``@`` command: the user can attach extra files and
|
||||
directories to the current context without changing the base mode. Paths
|
||||
outside the vault are ignored by the underlying collectors.
|
||||
|
||||
Args:
|
||||
vault_path: Absolute path to the vault root.
|
||||
files: Relative file paths to include.
|
||||
directories: Relative directory paths to include (recursive).
|
||||
scope: Context label exposed to the UI.
|
||||
|
||||
Returns:
|
||||
Same shape as :func:`collect_directory_context`.
|
||||
"""
|
||||
vault_resolved = vault_path.resolve()
|
||||
collected: list[dict[str, Any]] = []
|
||||
seen: set[str] = set()
|
||||
total_chars = 0
|
||||
|
||||
def _add(entries: list[dict[str, Any]]) -> None:
|
||||
nonlocal total_chars
|
||||
for entry in entries:
|
||||
if len(collected) >= BOOKSLM_MAX_FILES or total_chars >= BOOKSLM_MAX_TOTAL_CHARS:
|
||||
return
|
||||
path = entry.get("path")
|
||||
if not path or path in seen:
|
||||
continue
|
||||
seen.add(path)
|
||||
collected.append(entry)
|
||||
total_chars += len(entry.get("content", ""))
|
||||
|
||||
if files:
|
||||
_add(collect_files_context(vault_resolved, files, scope=scope)["files"])
|
||||
for directory in directories or []:
|
||||
_add(collect_directory_context(vault_resolved, directory)["files"])
|
||||
|
||||
return {
|
||||
"files": collected,
|
||||
"total_chars": total_chars,
|
||||
"file_count": len(collected),
|
||||
"directory_tree": "",
|
||||
"max_total_chars": BOOKSLM_MAX_TOTAL_CHARS,
|
||||
"max_files": BOOKSLM_MAX_FILES,
|
||||
"scope": scope,
|
||||
}
|
||||
|
||||
|
||||
def merge_contexts(base: dict[str, Any], extra: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Merge two context payloads, de-duplicating files by path."""
|
||||
files: list[dict[str, Any]] = []
|
||||
seen: set[str] = set()
|
||||
total_chars = 0
|
||||
for entry in list(base.get("files", [])) + list(extra.get("files", [])):
|
||||
path = entry.get("path")
|
||||
if not path or path in seen:
|
||||
continue
|
||||
if len(files) >= BOOKSLM_MAX_FILES or total_chars >= BOOKSLM_MAX_TOTAL_CHARS:
|
||||
break
|
||||
seen.add(path)
|
||||
files.append(entry)
|
||||
total_chars += len(entry.get("content", ""))
|
||||
|
||||
tree = base.get("directory_tree", "") or extra.get("directory_tree", "")
|
||||
scope = extra.get("scope") or base.get("scope", "general")
|
||||
return {
|
||||
"files": files,
|
||||
"total_chars": total_chars,
|
||||
"file_count": len(files),
|
||||
"directory_tree": tree,
|
||||
"max_total_chars": BOOKSLM_MAX_TOTAL_CHARS,
|
||||
"max_files": BOOKSLM_MAX_FILES,
|
||||
"scope": scope,
|
||||
}
|
||||
|
||||
|
||||
def is_image_path(path: str) -> bool:
|
||||
"""True when the path has a supported image extension."""
|
||||
return Path(path or "").suffix.lower() in IMAGE_EXTENSIONS
|
||||
|
||||
|
||||
def load_vault_image_data_url(vault_path: Path, rel_path: str) -> str | None:
|
||||
"""Read a vault image and return it as a ``data:`` URL for vision models.
|
||||
|
||||
Returns ``None`` when the path is outside the vault, missing, too large,
|
||||
or not a supported image.
|
||||
"""
|
||||
if not rel_path or not is_image_path(rel_path):
|
||||
return None
|
||||
vault_resolved = vault_path.resolve()
|
||||
try:
|
||||
target = (vault_resolved / rel_path).resolve()
|
||||
target.relative_to(vault_resolved)
|
||||
except (ValueError, OSError):
|
||||
return None
|
||||
if not target.is_file():
|
||||
return None
|
||||
try:
|
||||
if target.stat().st_size > BOOKSLM_MAX_IMAGE_BYTES:
|
||||
logger.warning("Image too large to send: %s", rel_path)
|
||||
return None
|
||||
raw = target.read_bytes()
|
||||
except OSError as exc:
|
||||
logger.warning("Cannot read image %s: %s", rel_path, exc)
|
||||
return None
|
||||
mime = mimetypes.guess_type(str(target))[0] or "image/png"
|
||||
encoded = base64.b64encode(raw).decode("ascii")
|
||||
return f"data:{mime};base64,{encoded}"
|
||||
|
||||
|
||||
def empty_context(scope: str = "general") -> dict[str, Any]:
|
||||
"""Return an empty context payload (used by the General assistant)."""
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user