> ## Documentation Index
> Fetch the complete documentation index at: https://docs.fim.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Guide d'extension

> Comment intégrer des fournisseurs LLM personnalisés, des outils, des backends de mémoire, des embeddings, des rerankers, des backends web et des composants RAG dans FIM One.

## Aperçu

FIM One est construit autour d'un ensemble de classes de base abstraites minces — une par composant interchangeable. Chaque composant a une responsabilité unique 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'images         | `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.

```python theme={null}
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
```

### Inscription 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 les vôtres.

```python theme={null}
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"])

# Retrieve later
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.

```python theme={null}
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       ← existing tool
├── git_status.py       ← your new file → auto-discovered
└── ...
```

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 :

```python theme={null}
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 :

```python theme={null}
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é

`BaseMemory` est la couche de persistance pour l'historique de conversation. Trois méthodes : `add_message`, `get_messages`, `clear`.

```python theme={null}
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)
```

Injecter via le constructeur de l'agent : `ReActAgent(llm=llm, memory=RedisMemory(conv_id, url))`.

***

## Intégration personnalisée

`BaseEmbedding` fournit deux méthodes : `embed_texts` (lot) et `embed_query` (unique), ainsi qu'une propriété `dimension`.

```python theme={null}
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'intégration (par exemple E5, BGE) utilisent des préfixes différents pour les documents par rapport aux requêtes afin d'améliorer la qualité de la récupération.

***

## Génération d'images personnalisée

`BaseImageGen` a 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.

```python theme={null}
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éordonnées avec des scores.

```python theme={null}
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 sous forme de Markdown ou de texte brut.

```python theme={null}
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`.

```python theme={null}
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.

```python theme={null}
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écoupage de texte

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

```python theme={null}
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.

```python theme={null}
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 priorité.** 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 une fonctionnalité 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.