Aperçu

FIM One est construit autour d’un ensemble de classes de base abstraites minces — une par composant interchangeable. Chaque composant a une seule responsabilité et une interface minimale. Vous implémentez les méthodes abstraites, connectez l’instance au registre ou à l’injecteur approprié, et le reste du système utilise automatiquement votre implémentation.

Point d’extension	Classe de base	Fichier	Enregistrement
Fournisseur LLM	`BaseLLM`	`core/model/base.py`	`ModelRegistry.register()`
Outil	`BaseTool`	`core/tool/base.py`	Déposer un fichier dans `builtin/`
Mémoire	`BaseMemory`	`core/memory/base.py`	Injection de constructeur
Embedding	`BaseEmbedding`	`core/embedding/base.py`	Injection de constructeur
Génération d’image	`BaseImageGen`	`core/image_gen/base.py`	Injection de constructeur
Reranker	`BaseReranker`	`core/reranker/base.py`	Injection de constructeur
Backend de récupération web	`BaseWebFetch`	`core/web/fetch/base.py`	Injection de constructeur
Backend de recherche web	`BaseWebSearch`	`core/web/search/base.py`	Injection de constructeur
Récupérateur RAG	`BaseRetriever`	`rag/base.py`	Injection de constructeur
Chargeur de documents	`BaseLoader`	`rag/loaders/base.py`	Registre de chargeur / injection
Chunker de texte	`BaseChunker`	`rag/chunking/base.py`	Injection de constructeur

Fournisseur LLM personnalisé

BaseLLM a deux méthodes requises — chat et stream_chat — plus une propriété abilities optionnelle qui indique au reste du système ce que le modèle peut faire.

from collections.abc import AsyncIterator
from typing import Any

from fim_one.core.model.base import BaseLLM
from fim_one.core.model.types import ChatMessage, LLMResult, StreamChunk


class MyLLM(BaseLLM):
    def __init__(self, api_key: str, model: str) -> None:
        self._api_key = api_key
        self._model = model

    @property
    def model_id(self) -> str:
        return self._model

    @property
    def abilities(self) -> dict[str, bool]:
        return {
            "tool_call": True,   # supports native function calling
            "json_mode": True,   # supports response_format JSON mode
            "vision":   False,
            "streaming": True,
        }

    async def chat(
        self,
        messages: list[ChatMessage],
        *,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        temperature: float | None = None,
        max_tokens: int | None = None,
        response_format: dict[str, Any] | None = None,
    ) -> LLMResult:
        # Call your provider, return LLMResult(message=..., usage=...)
        ...

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        *,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        temperature: float | None = None,
        max_tokens: int | None = None,
    ) -> AsyncIterator[StreamChunk]:
        # Yield StreamChunk instances as tokens arrive
        ...
        yield  # make type-checker happy

Enregistrement via ModelRegistry

ModelRegistry mappe les noms aux instances BaseLLM et résout par rôle. Le système utilise quatre rôles intégrés : general, fast, compact et vision. Vous pouvez en ajouter vos propres.

from fim_one.core.model.registry import ModelRegistry

registry = ModelRegistry()
registry.register("my-llm", MyLLM(api_key="...", model="my-v1"), roles=["general"])
registry.register("my-fast", MyLLM(api_key="...", model="my-mini"), roles=["fast", "compact"])

Récupérer plus tard

llm = registry.get_default()           # first "general" model, or first registered
llm = registry.get_by_role("fast")     # first model with the "fast" role
llm = registry.get("my-llm")           # by exact name

Le dictionnaire abilities est le contrat entre le LLM et le moteur ReAct. Quand tool_call=True et que l’agent a été créé avec use_native_tools=True, le moteur utilisera l’appel de fonction natif. Sinon, il bascule automatiquement en mode JSON.

Outil personnalisé

Les outils sont l’extension la plus courante. BaseTool a trois éléments requis : name, description et run. Tout le reste a des valeurs par défaut sensées.

from typing import Any
from fim_one.core.tool.base import BaseTool


class GitStatusTool(BaseTool):
    @property
    def name(self) -> str:
        return "git_status"

    @property
    def description(self) -> str:
        return "Return the current git status of a repository."

    @property
    def category(self) -> str:
        return "filesystem"   # groups the tool in the UI

    @property
    def parameters_schema(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Absolute path to the repository root.",
                }
            },
            "required": ["path"],
        }

    async def run(self, *, path: str, **kwargs: Any) -> str:
        import asyncio
        result = await asyncio.create_subprocess_shell(
            f"git -C {path} status --short",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, _ = await result.communicate()
        return stdout.decode()

Découverte automatique

Déposez votre fichier dans src/fim_one/core/tool/builtin/. Le scanner discover_builtin_tools() trouvera automatiquement toute sous-classe concrète (non abstraite) de BaseTool — aucune inscription manuelle requise.

src/fim_one/core/tool/builtin/
├── calculator.py       ← outil existant
├── git_status.py       ← votre nouveau fichier → découvert automatiquement
└── ...

Le scanner ignore les classes listées dans _SKIP_AUTO_DISCOVER. Utilisez cet ensemble pour les outils qui nécessitent une configuration externe (par exemple une clé API) et doivent être instanciés conditionnellement au démarrage.

Signalisation de l’indisponibilité

Remplacez availability() pour afficher un message dans le catalogue des outils lorsqu’une dépendance est manquante :

def availability(self) -> tuple[bool, str | None]:
    import os
    if not os.getenv("GITHUB_TOKEN"):
        return False, "GITHUB_TOKEN environment variable is not set."
    return True, None

Résultats enrichis avec des artefacts

Retournez un ToolResult au lieu d’une simple str lorsque votre outil produit des fichiers :

from fim_one.core.tool.base import Artifact, ToolResult

async def run(self, **kwargs: Any) -> ToolResult:
    # ... produce a file at /tmp/report.html ...
    return ToolResult(
        content="Report generated.",
        content_type="text",
        artifacts=[Artifact(name="report.html", path="/uploads/report.html", mime_type="text/html", size=4096)],
    )

Mémoire personnalisée

BaseMemory est la couche de persistance pour l’historique des conversations. Trois méthodes : add_message, get_messages, clear.

import redis.asyncio as redis
from fim_one.core.memory.base import BaseMemory
from fim_one.core.model.types import ChatMessage


class RedisMemory(BaseMemory):
    def __init__(self, conversation_id: str, redis_url: str) -> None:
        self._key = f"conv:{conversation_id}"
        self._redis = redis.from_url(redis_url)

    async def add_message(self, message: ChatMessage) -> None:
        import json
        await self._redis.rpush(self._key, json.dumps(message))

    async def get_messages(self) -> list[ChatMessage]:
        import json
        raw = await self._redis.lrange(self._key, 0, -1)
        return [json.loads(m) for m in raw]

    async def clear(self) -> None:
        await self._redis.delete(self._key)

Injectez via le constructeur de l’agent : ReActAgent(llm=llm, memory=RedisMemory(conv_id, url)).

Intégration personnalisée d’embeddings

BaseEmbedding fournit deux méthodes : embed_texts (batch) et embed_query (single), ainsi qu’une propriété dimension.

from fim_one.core.embedding.base import BaseEmbedding


class MyEmbedding(BaseEmbedding):
    def __init__(self, model: str) -> None:
        self._model = model
        self._dim = 1536

    @property
    def dimension(self) -> int:
        return self._dim

    async def embed_texts(self, texts: list[str]) -> list[list[float]]:
        # Batch embed documents
        ...

    async def embed_query(self, query: str) -> list[float]:
        # Embed a single query — often uses a different instruction prefix
        ...

La distinction entre embed_texts et embed_query existe parce que de nombreux modèles d’embedding (par exemple E5, BGE) utilisent des préfixes différents pour les documents et les requêtes afin d’améliorer la qualité de la récupération.

Génération d’images personnalisée

BaseImageGen possède une seule méthode generate. Elle enregistre l’image dans output_dir et retourne un ImageResult avec le chemin du fichier et une URL relative au serveur.

from fim_one.core.image_gen.base import BaseImageGen, ImageResult


class StableDiffusionImageGen(BaseImageGen):
    async def generate(
        self,
        prompt: str,
        *,
        aspect_ratio: str = "1:1",
        output_dir: str,
    ) -> ImageResult:
        # Call your SD API, save to output_dir
        file_path = f"{output_dir}/image.png"
        return ImageResult(
            file_path=file_path,
            url=f"/uploads/{file_path.split('/')[-1]}",
            prompt=prompt,
            model="stable-diffusion-xl",
        )

Réclasseur personnalisé

BaseReranker prend une requête et une liste de chaînes de documents et les retourne réorganisées avec des scores.

from fim_one.core.reranker.base import BaseReranker, RerankResult


class CrossEncoderReranker(BaseReranker):
    async def rerank(
        self, query: str, documents: list[str], *, top_k: int = 5
    ) -> list[RerankResult]:
        # Score each (query, doc) pair with a cross-encoder
        scores = await self._score_pairs(query, documents)
        results = [
            RerankResult(index=i, score=score, text=doc)
            for i, (doc, score) in enumerate(zip(documents, scores))
        ]
        results.sort(key=lambda r: r.score, reverse=True)
        return results[:top_k]

Backends web personnalisés

Récupération web

BaseWebFetch récupère une URL et retourne son contenu en Markdown ou en texte brut.

from fim_one.core.web.fetch.base import BaseWebFetch


class PlaywrightFetch(BaseWebFetch):
    async def fetch(self, url: str) -> str:
        # Use Playwright to render JS-heavy pages
        async with async_playwright() as p:
            browser = await p.chromium.launch()
            page = await browser.new_page()
            await page.goto(url)
            content = await page.content()
            await browser.close()
        return html_to_markdown(content)

Recherche web

BaseWebSearch retourne une liste classée d’objets SearchResult.

from fim_one.core.web.search.base import BaseWebSearch, SearchResult


class BingSearch(BaseWebSearch):
    async def search(self, query: str, *, num_results: int = 10) -> list[SearchResult]:
        # Call Bing Search API
        ...
        return [
            SearchResult(title=r["name"], url=r["url"], snippet=r["snippet"])
            for r in raw_results[:num_results]
        ]

Composants RAG personnalisés

Le pipeline RAG a trois étapes indépendamment interchangeables : chargement, segmentation et récupération.

Chargeur de documents

BaseLoader transforme un chemin de fichier en une liste d’objets LoadedDocument. Les chargeurs PDF retournent généralement un document par page.

from pathlib import Path
from fim_one.rag.loaders.base import BaseLoader, LoadedDocument


class DocxLoader(BaseLoader):
    async def load(self, path: Path) -> list[LoadedDocument]:
        from docx import Document
        doc = Document(path)
        text = "\n".join(p.text for p in doc.paragraphs)
        return [LoadedDocument(content=text, metadata={"source": str(path)})]

Découpeur de texte

BaseChunker divise le texte en objets Chunk. MAX_CHUNK_SIZE = 6000 caractères est la limite maximale — les tailles de chunk supérieures à cela peuvent dépasser la fenêtre de tokens Jina Embeddings v3.

from typing import Any
from fim_one.rag.chunking.base import BaseChunker, Chunk


class SentenceChunker(BaseChunker):
    def __init__(self, sentences_per_chunk: int = 5) -> None:
        self._n = sentences_per_chunk

    async def chunk(self, text: str, metadata: dict[str, Any] | None = None) -> list[Chunk]:
        import nltk
        sentences = nltk.sent_tokenize(text)
        chunks = []
        for i in range(0, len(sentences), self._n):
            chunk_text = " ".join(sentences[i : i + self._n])
            chunks.append(Chunk(text=chunk_text, metadata=metadata or {}, index=i // self._n))
        return chunks

Récupérateur

BaseRetriever interroge n’importe quel backend et retourne des objets Document classés.

from fim_one.rag.base import BaseRetriever, Document


class ElasticsearchRetriever(BaseRetriever):
    def __init__(self, es_client, index: str) -> None:
        self._es = es_client
        self._index = index

    async def retrieve(self, query: str, *, top_k: int = 5) -> list[Document]:
        resp = await self._es.search(
            index=self._index,
            query={"match": {"content": query}},
            size=top_k,
        )
        return [
            Document(
                content=hit["_source"]["content"],
                metadata=hit["_source"].get("metadata", {}),
                score=hit["_score"],
            )
            for hit in resp["hits"]["hits"]
        ]

Principes de conception

Quelques modèles sont cohérents dans toutes les classes de base qui facilitent l’écriture correcte des implémentations personnalisées : Async en premier. Chaque méthode est async def. Même si votre implémentation est synchrone, enveloppez-la avec asyncio.to_thread() plutôt que de bloquer la boucle d’événements. Sortie textuelle des outils. BaseTool.run() retourne str (ou ToolResult). Le LLM ne voit que du texte — les implémentations d’outils sont responsables de la sérialisation des données complexes dans un format lisible. Interfaces minimales. Chaque classe de base définit le plus petit contrat nécessaire. BaseMemory compte trois méthodes ; BaseWebFetch en compte une. Vous n’êtes jamais obligé d’implémenter les fonctionnalités dont vous n’avez pas besoin. Composition plutôt qu’héritage. Les classes de base sont des interfaces, pas des frameworks. Vous injectez votre implémentation au moment de la construction ; l’exécution ne la sous-classe jamais ou ne la modifie jamais ultérieurement.

Démarrage rapide

Configuration

Fonctionnalités

Extension de FIM One

Guide d'extension

Aperçu

Fournisseur LLM personnalisé

Enregistrement via ModelRegistry

Récupérer plus tard

Outil personnalisé

Découverte automatique

Signalisation de l’indisponibilité

Résultats enrichis avec des artefacts

Mémoire personnalisée

Intégration personnalisée d’embeddings

Génération d’images personnalisée

Réclasseur personnalisé

Backends web personnalisés

Récupération web

Recherche web

Composants RAG personnalisés

Chargeur de documents

Découpeur de texte

Récupérateur

Principes de conception

Démarrage rapide

Configuration

Fonctionnalités

Extension de FIM One

​Aperçu

​Fournisseur LLM personnalisé

​Enregistrement via ModelRegistry

​Récupérer plus tard

​Outil personnalisé

​Découverte automatique

​Signalisation de l’indisponibilité

​Résultats enrichis avec des artefacts

​Mémoire personnalisée

​Intégration personnalisée d’embeddings

​Génération d’images personnalisée

​Réclasseur personnalisé

​Backends web personnalisés

​Récupération web

​Recherche web

​Composants RAG personnalisés

​Chargeur de documents

​Découpeur de texte

​Récupérateur

​Principes de conception

Aperçu

Fournisseur LLM personnalisé

Enregistrement via ModelRegistry

Récupérer plus tard

Outil personnalisé

Découverte automatique

Signalisation de l’indisponibilité

Résultats enrichis avec des artefacts

Mémoire personnalisée

Intégration personnalisée d’embeddings

Génération d’images personnalisée

Réclasseur personnalisé

Backends web personnalisés

Récupération web

Recherche web

Composants RAG personnalisés

Chargeur de documents

Découpeur de texte

Récupérateur

Principes de conception