Erweiterungsleitfaden

Übersicht

FIM One basiert auf einer Reihe von dünnen abstrakten Basisklassen — eine pro austauschbare Komponente. Jede Komponente hat eine einzige Verantwortung und eine minimale Schnittstelle. Sie implementieren die abstrakten Methoden, verbinden die Instanz mit der entsprechenden Registry oder dem Injector, und der Rest des Systems verwendet Ihre Implementierung automatisch.

Erweiterungspunkt	Basisklasse	Datei	Registrierung
LLM-Anbieter	`BaseLLM`	`core/model/base.py`	`ModelRegistry.register()`
Werkzeug	`BaseTool`	`core/tool/base.py`	Datei in `builtin/` ablegen
Speicher	`BaseMemory`	`core/memory/base.py`	Constructor-Injektion
Embedding	`BaseEmbedding`	`core/embedding/base.py`	Constructor-Injektion
Bildgenerierung	`BaseImageGen`	`core/image_gen/base.py`	Constructor-Injektion
Reranker	`BaseReranker`	`core/reranker/base.py`	Constructor-Injektion
Web-Abruf-Backend	`BaseWebFetch`	`core/web/fetch/base.py`	Constructor-Injektion
Web-Suche-Backend	`BaseWebSearch`	`core/web/search/base.py`	Constructor-Injektion
RAG-Retriever	`BaseRetriever`	`rag/base.py`	Constructor-Injektion
Dokument-Loader	`BaseLoader`	`rag/loaders/base.py`	Loader-Registry / Injektion
Text-Chunker	`BaseChunker`	`rag/chunking/base.py`	Constructor-Injektion

Benutzerdefinierter LLM-Anbieter

BaseLLM hat zwei erforderliche Methoden — chat und stream_chat — sowie eine optionale abilities-Eigenschaft, die dem Rest des Systems mitteilt, was das Modell kann.

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

Registrierung über ModelRegistry

ModelRegistry ordnet Namen BaseLLM-Instanzen zu und löst sie nach Rolle auf. Das System verwendet vier integrierte Rollen: general, fast, compact und vision. Sie können eigene hinzufügen.

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"])

# Später abrufen
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

Das abilities-Wörterbuch ist der Vertrag zwischen dem LLM und der ReAct-Engine. Wenn tool_call=True und der Agent mit use_native_tools=True erstellt wurde, verwendet die Engine natives Function Calling. Andernfalls wird automatisch auf JSON-Modus zurückgegriffen.

Benutzerdefiniertes Tool

Tools sind die häufigste Erweiterung. BaseTool hat drei erforderliche Teile: name, description und run. Alles andere hat sinnvolle Standardwerte.

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()

Auto-Discovery

Legen Sie Ihre Datei in src/fim_one/core/tool/builtin/ ab. Der discover_builtin_tools()-Scanner findet automatisch alle konkreten (nicht abstrakten) BaseTool-Unterklassen – keine manuelle Registrierung erforderlich.

src/fim_one/core/tool/builtin/
├── calculator.py       ← vorhandenes Werkzeug
├── git_status.py       ← Ihre neue Datei → automatisch erkannt
└── ...

Der Scanner überspringt Klassen, die in _SKIP_AUTO_DISCOVER aufgelistet sind. Verwenden Sie diesen Satz für Werkzeuge, die externe Konfiguration erfordern (z. B. einen API-Schlüssel) und bei der Initialisierung bedingt instanziiert werden müssen.

Signalisierung der Nichtverfügbarkeit

Überschreiben Sie availability(), um eine Nachricht im Werkzeugkatalog anzuzeigen, wenn eine Abhängigkeit fehlt:

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

Umfangreiche Ergebnisse mit Artefakten

Geben Sie ein ToolResult statt eines einfachen str zurück, wenn Ihr Tool Dateien erzeugt:

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)],
    )

Benutzerdefinierter Speicher

BaseMemory ist die Persistierungsschicht für Gesprächsverlauf. Drei Methoden: 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)

Injizieren Sie über den Agenten-Konstruktor: ReActAgent(llm=llm, memory=RedisMemory(conv_id, url)).

Benutzerdefinierte Einbettung

BaseEmbedding bietet zwei Methoden: embed_texts (Batch) und embed_query (einzeln), sowie eine dimension-Eigenschaft.

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
        ...

Der Unterschied zwischen embed_texts und embed_query existiert, weil viele Einbettungsmodelle (z. B. E5, BGE) unterschiedliche Präfixe für Dokumente und Abfragen verwenden, um die Abrufqualität zu verbessern.

Benutzerdefinierte Bildgenerierung

BaseImageGen hat eine einzelne Methode generate. Sie speichert das Bild in output_dir und gibt ein ImageResult mit dem Dateipfad und einer serverelativen URL zurück.

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",
        )

Benutzerdefinierter Reranker

BaseReranker nimmt eine Abfrage und eine Liste von Dokumentzeichenfolgen entgegen und gibt sie neu geordnet mit Bewertungen zurück.

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]

Benutzerdefinierte Web-Backends

Web-Abruf

BaseWebFetch ruft eine URL ab und gibt ihren Inhalt als Markdown oder Klartext zurück.

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)

Websuche

BaseWebSearch gibt eine sortierte Liste von SearchResult-Objekten zurück.

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]
        ]

Benutzerdefinierte RAG-Komponenten

Die RAG-Pipeline hat drei unabhängig austauschbare Stufen: Laden, Chunking und Abruf.

Document Loader

BaseLoader wandelt einen Dateipfad in eine Liste von LoadedDocument-Objekten um. PDF-Loader geben normalerweise ein Dokument pro Seite zurück.

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)})]

Text-Chunker

BaseChunker teilt Text in Chunk-Objekte auf. MAX_CHUNK_SIZE = 6000 Zeichen ist die harte Obergrenze — Chunk-Größen über diesem Wert können das Jina Embeddings v3 Token-Fenster überlasten.

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

Retriever

BaseRetriever fragt ein beliebiges Backend ab und gibt sortierte Document-Objekte zurück.

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"]
        ]

Designprinzipien

Einige Muster sind konsistent über alle Basisklassen hinweg, um benutzerdefinierte Implementierungen leichter korrekt zu schreiben: Async-first. Jede Methode ist async def. Auch wenn Ihre Implementierung synchron ist, wrappen Sie sie mit asyncio.to_thread() anstatt den Event Loop zu blockieren. String-Ausgabe von Tools. BaseTool.run() gibt str (oder ToolResult) zurück. Das LLM sieht nur Text — Tool-Implementierungen sind verantwortlich für die Serialisierung komplexer Daten in ein lesbares Format. Minimale Schnittstellen. Jede Basisklasse definiert den kleinsten erforderlichen Vertrag. BaseMemory hat drei Methoden; BaseWebFetch hat eine. Sie müssen nie Funktionalität implementieren, die Sie nicht benötigen. Komposition über Vererbung. Die Basisklassen sind Schnittstellen, keine Frameworks. Sie injizieren Ihre Implementierung zur Konstruktionszeit; die Runtime patcht oder subklassifiziert sie nie weiter.

Erste Schritte

Konfiguration

Funktionen

FIM One erweitern

Erweiterungsleitfaden

Übersicht

Benutzerdefinierter LLM-Anbieter

Registrierung über ModelRegistry

Benutzerdefiniertes Tool

Auto-Discovery

Signalisierung der Nichtverfügbarkeit

Umfangreiche Ergebnisse mit Artefakten

Benutzerdefinierter Speicher

Benutzerdefinierte Einbettung

Benutzerdefinierte Bildgenerierung

Benutzerdefinierter Reranker

Benutzerdefinierte Web-Backends

Web-Abruf

Websuche

Benutzerdefinierte RAG-Komponenten

Document Loader

Text-Chunker

Retriever

Designprinzipien

Erste Schritte

Konfiguration

Funktionen

FIM One erweitern

​Übersicht

​Benutzerdefinierter LLM-Anbieter

​Registrierung über ModelRegistry

​Benutzerdefiniertes Tool

​Auto-Discovery

​Signalisierung der Nichtverfügbarkeit

​Umfangreiche Ergebnisse mit Artefakten

​Benutzerdefinierter Speicher

​Benutzerdefinierte Einbettung

​Benutzerdefinierte Bildgenerierung

​Benutzerdefinierter Reranker

​Benutzerdefinierte Web-Backends

​Web-Abruf

​Websuche

​Benutzerdefinierte RAG-Komponenten

​Document Loader

​Text-Chunker

​Retriever

​Designprinzipien

Übersicht

Benutzerdefinierter LLM-Anbieter

Registrierung über ModelRegistry

Benutzerdefiniertes Tool

Auto-Discovery

Signalisierung der Nichtverfügbarkeit

Umfangreiche Ergebnisse mit Artefakten

Benutzerdefinierter Speicher

Benutzerdefinierte Einbettung

Benutzerdefinierte Bildgenerierung

Benutzerdefinierter Reranker

Benutzerdefinierte Web-Backends

Web-Abruf

Websuche

Benutzerdefinierte RAG-Komponenten

Document Loader

Text-Chunker

Retriever

Designprinzipien