拡張ガイド - FIM One

概要

FIM One は、スワップ可能なコンポーネントごとに 1 つの薄い抽象基底クラスのセットを中心に構築されています。すべてのコンポーネントは単一の責任と最小限のインターフェースを持ちます。抽象メソッドを実装し、インスタンスを適切なレジストリまたはインジェクターに配線すると、システムの残りの部分が自動的に実装を使用します。

拡張ポイント	基底クラス	ファイル	登録
LLM プロバイダー	`BaseLLM`	`core/model/base.py`	`ModelRegistry.register()`
ツール	`BaseTool`	`core/tool/base.py`	`builtin/` にファイルをドロップ
メモリ	`BaseMemory`	`core/memory/base.py`	コンストラクタインジェクション
埋め込み	`BaseEmbedding`	`core/embedding/base.py`	コンストラクタインジェクション
画像生成	`BaseImageGen`	`core/image_gen/base.py`	コンストラクタインジェクション
リランカー	`BaseReranker`	`core/reranker/base.py`	コンストラクタインジェクション
Web フェッチバックエンド	`BaseWebFetch`	`core/web/fetch/base.py`	コンストラクタインジェクション
Web 検索バックエンド	`BaseWebSearch`	`core/web/search/base.py`	コンストラクタインジェクション
RAG レトリーバー	`BaseRetriever`	`rag/base.py`	コンストラクタインジェクション
ドキュメントローダー	`BaseLoader`	`rag/loaders/base.py`	ローダーレジストリ / インジェクション
テキストチャンカー	`BaseChunker`	`rag/chunking/base.py`	コンストラクタインジェクション

カスタムLLMプロバイダー

BaseLLM には 2 つの必須メソッド — chat と stream_chat — と、モデルが何ができるかをシステムの残りの部分に伝えるオプションの abilities プロパティがあります。

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

ModelRegistry経由での登録

ModelRegistryは名前をBaseLLMインスタンスにマッピングし、ロールによって解決します。システムは4つの組み込みロールを使用します：general、fast、compact、およびvision。独自のロールを追加できます。

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

abilities辞書はLLMとReActエンジン間の契約です。tool_call=Trueでエージェントがuse_native_tools=Trueで作成された場合、エンジンはネイティブ関数呼び出しを使用します。それ以外の場合は自動的にJSONモードにフォールバックします。

カスタムツール

ツールは最も一般的な拡張機能です。BaseToolには3つの必須要素があります：name、description、およびrunです。その他すべてには適切なデフォルト値があります。

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

自動検出

src/fim_one/core/tool/builtin/ にファイルをドロップします。discover_builtin_tools() スキャナーは、具体的な（抽象的でない）BaseTool サブクラスを自動的に検出します — 手動登録は不要です。

src/fim_one/core/tool/builtin/
├── calculator.py       ← existing tool
├── git_status.py       ← your new file → auto-discovered
└── ...

スキャナーは _SKIP_AUTO_DISCOVER にリストされたクラスをスキップします。外部設定（例：API キー）が必要で、起動時に条件付きでインスタンス化する必要があるツールに対してそのセットを使用します。

利用不可能性を通知する

依存関係が不足している場合、ツールカタログにメッセージを表示するには、availability() をオーバーライドします：

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

アーティファクトを含むリッチな結果

ツールがファイルを生成する場合は、プレーンな str の代わりに ToolResult を返します：

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

カスタムメモリ

BaseMemory は会話履歴の永続化レイヤーです。3つのメソッドがあります: 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)

エージェントコンストラクタ経由でインジェクトします: ReActAgent(llm=llm, memory=RedisMemory(conv_id, url))。

カスタム埋め込み

BaseEmbedding は 2 つのメソッドを提供します: embed_texts（バッチ）と embed_query（単一）、および 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
        ...

embed_texts と embed_query の区別が存在する理由は、多くの埋め込みモデル（例：E5、BGE）が検索品質を向上させるために、ドキュメントとクエリに異なるプレフィックスを使用するためです。

カスタム画像生成

BaseImageGen には単一のメソッド generate があります。これは画像を output_dir に保存し、ファイルパスとサーバー相対 URL を含む ImageResult を返します。

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

カスタム再ランカー

BaseReranker はクエリとドキュメント文字列のリストを受け取り、スコア付けして並べ替えて返します。

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]

カスタムウェブバックエンド

Webフェッチ

BaseWebFetch はURLをフェッチしてそのコンテンツをMarkdownまたはプレーンテキストとして返します。

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)

Webサーチ

BaseWebSearchは、ランク付けされた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]
        ]

カスタムRAGコンポーネント

RAGパイプラインには、独立して交換可能な3つのステージがあります：ロード、チャンキング、および検索。

ドキュメントローダー

BaseLoader はファイルパスを LoadedDocument オブジェクトのリストに変換します。PDF ローダーは通常、ページごとに 1 つのドキュメントを返します。

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

テキストチャンカー

BaseChunker はテキストを Chunk オブジェクトに分割します。MAX_CHUNK_SIZE = 6000 文字はハード上限です — このサイズを超えるチャンクサイズは 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

レトリーバー

BaseRetriever は任意のバックエンドをクエリし、ランク付けされた Document オブジェクトを返します。

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

設計原則

すべての基本クラスにわたって一貫性のあるいくつかのパターンがあり、カスタム実装をより正しく記述しやすくします: 非同期ファースト。 すべてのメソッドは async def です。実装が同期的であっても、イベントループをブロックするのではなく asyncio.to_thread() でラップしてください。 ツールからの文字列出力。 BaseTool.run() は str (または ToolResult) を返します。LLM はテキストのみを見ます — ツール実装は複雑なデータを読み取り可能な形式にシリアライズする責任があります。 最小限のインターフェース。 各基本クラスは必要な最小限の契約を定義します。BaseMemory は3つのメソッド、BaseWebFetch は1つです。不要な機能を実装する必要はありません。 継承よりも構成。 基本クラスはインターフェースであり、フレームワークではありません。構築時に実装を注入します。ランタイムは決してモンキーパッチやさらなるサブクラス化を行いません。

​概要

​カスタムLLMプロバイダー

​ModelRegistry経由での登録

​カスタムツール

​自動検出

​利用不可能性を通知する

​アーティファクトを含むリッチな結果

​カスタムメモリ

​カスタム埋め込み

​カスタム画像生成

​カスタム再ランカー

​カスタムウェブバックエンド

​Webフェッチ

​Webサーチ

​カスタムRAGコンポーネント

​ドキュメントローダー

​テキストチャンカー

​レトリーバー

​設計原則

概要