プラグイン共通API

LangBotは、プラグインがさまざまなLangBotモジュールを操作し、メッセージコンテキストを制御するための一連のAPIを提供します。

リクエストAPI

現在処理中のユーザーリクエスト(メッセージ)に対する操作です。EventListenerとCommandコンポーネントでのみ使用可能です。アクセス方法:

EventListenerのイベントハンドラメソッド内: event_context: context.EventContextオブジェクトの内部メソッド
Commandのサブコマンドハンドラメソッド内: context: ExecuteContextオブジェクトの内部メソッド

直接返信メッセージ

現在のリクエストが存在するセッションにメッセージチェーンで直接返信します。メッセージチェーンの構築方法については、メッセージプラットフォームエンティティを参照してください。

async def reply(
    self, message_chain: platform_message.MessageChain, quote_origin: bool = False
):
    """メッセージリクエストに返信する

    Args:
        message_chain (platform.types.MessageChain): LangBotメッセージチェーン
        quote_origin (bool): 元のメッセージを引用するかどうか
    """

# 使用例
await event_context.reply(
    platform_message.MessageChain([
        platform_message.Plain(text="Hello, world!"),
    ]),
)

Bot UUIDの取得

現在のリクエストを発行したBotのUUIDを取得します。

async def get_bot_uuid(self) -> str:
    """Bot UUIDを取得する"""

# 使用例
bot_uuid = await event_context.get_bot_uuid()

リクエスト変数の設定

単一のリクエスト内の一部の情報は、リクエスト変数に保存されます。Difyなどの外部Agent プラットフォームを使用する場合、これらの変数はAgent プラットフォームに明示的に渡されます。

async def set_query_var(self, key: str, value: Any):
    """クエリ変数を設定する"""

# 使用例
await event_context.set_query_var("key", "value")

リクエスト変数の取得

単一のリクエスト変数を取得します。

async def get_query_var(self, key: str) -> Any:
    """クエリ変数を取得する"""

# 使用例
query_var = await event_context.get_query_var("key")

すべてのリクエスト変数の取得

async def get_query_vars(self) -> dict[str, Any]:
    """すべてのクエリ変数を取得する"""

# 使用例
query_vars = await event_context.get_query_vars()

現在のパイプラインに設定されたナレッジベース一覧の取得

現在のリクエストが使用するパイプラインに設定されたナレッジベースの一覧を取得します。

async def list_pipeline_knowledge_bases(self) -> list[dict[str, Any]]:
    """List knowledge bases configured for the current pipeline"""

# 使用例
knowledge_bases = await event_context.list_pipeline_knowledge_bases()

# 返り値の例
[
    {
        "uuid": "kb_uuid",
        "name": "Product Docs",
        "description": "社内製品ドキュメント",
    }
]

現在のリクエストに対応するパイプラインの Local Agent 設定にバインドされたナレッジベースのみを返します。現在のパイプラインにナレッジベースが設定されていない場合、または Local Agent を使用していない場合は、空のリストが返されます。

LangBot API

これらのAPIは、任意のプラグインコンポーネントで呼び出すことができます。アクセス方法:

プラグインルートディレクトリのmain.py内: selfオブジェクトの内部メソッド。これらのAPIはすべて、プラグインクラスの親クラスBasePluginによって提供されます。
任意のプラグインコンポーネントクラス内: self.pluginオブジェクトの内部メソッド。

プラグイン設定の取得

プラグイン設定フォーマットはmanifest.yamlに記述でき、ユーザーはLangBotのプラグイン管理でプラグイン設定フォーマットに従って入力する必要があります。プラグインコードはこのAPIを呼び出してプラグイン設定情報を取得できます。

def get_config(self) -> dict[str, typing.Any]:
    """プラグインの設定を取得する。"""

# 使用例
config = self.plugin.get_config()

LangBotバージョンの取得

LangBotのバージョン番号を取得します。v<major>.<minor>.<patch>形式の文字列として返されます。

async def get_langbot_version(self) -> str:
    """LangBotバージョンを取得する"""

# 使用例
langbot_version = await self.plugin.get_langbot_version()

設定済みBotリストの取得

すべてのBot UUIDのリストを返します。

async def get_bots(self) -> list[str]:
    """すべてのBotを取得する"""

# 使用例
bots = await self.plugin.get_bots()

Bot情報の取得

Bot情報を取得します。

async def get_bot_info(self, bot_uuid: str) -> dict[str, Any]:
    """Bot情報を取得する"""

# 使用例
bot_info = await self.plugin.get_bot_info("de639861-be05-4018-859b-c2e2d3e0d603")

# 返り値の例
{
    "uuid": "de639861-be05-4018-859b-c2e2d3e0d603",
    "name": "aiocqhttp",
    "description": "Migrated from LangBot v3",
    "adapter": "aiocqhttp",
    "enable": true,
    "use_pipeline_name": "ChatPipeline",
    "use_pipeline_uuid": "c30a1dca-e91c-452b-83ec-84d635a30028",
    "created_at": "2025-05-10T13:53:08",
    "updated_at": "2025-08-12T11:27:30",
    "adapter_runtime_values": {  # Botが現在実行中の場合に存在
        "bot_account_id": 960164003  # BotアカウントID
    }
}

プロアクティブメッセージの送信

Bot UUIDとターゲットセッションIDを通じてプロアクティブメッセージを送信します。メッセージチェーンの構築方法については、メッセージプラットフォームエンティティを参照してください。

async def send_message(
    self,
    bot_uuid: str,
    target_type: str,
    target_id: str,
    message_chain: platform_message.MessageChain,
) -> None:
    """セッションにメッセージを送信する"""

# 使用例
await self.plugin.send_message(
    bot_uuid="de639861-be05-4018-859b-c2e2d3e0d603",
    target_type="person",
    target_id="1010553892",
    message_chain=platform_message.MessageChain([platform_message.Plain(text="Hello, world!")]),
)

設定済みLLMモデルリストの取得

設定されたすべてのLLMモデルのUUIDのリストを返します。

async def get_llm_models(self) -> list[str]:
    """すべてのLLMモデルを取得する"""

# 使用例
llm_models = await self.plugin.get_llm_models()

LLMモデルの呼び出し

LLMモデルを呼び出し、LLMメッセージを返します。非ストリーミング。

async def invoke_llm(
    self,
    llm_model_uuid: str,
    messages: list[provider_message.Message],
    funcs: list[resource_tool.LLMTool] = [],
    extra_args: dict[str, Any] = {},
) -> provider_message.Message:
    """LLMモデルを呼び出す"""

# 使用例
llm_message = await self.plugin.invoke_llm(
    llm_model_uuid="llm_model_uuid",
    messages=[provider_message.Message(role="user", content="Hello, world!")],
    funcs=[],
    extra_args={},
)

利用可能な Parser の一覧取得

ホスト上で現在利用可能な Parser プラグインを列挙します。MIME タイプで絞り込むこともできます。

async def list_parsers(self, mime_type: str | None = None) -> list[dict[str, Any]]:
    """List available Parser plugins"""

# 使用例
parsers = await self.plugin.list_parsers(mime_type="application/pdf")
# 各要素には plugin_id、plugin_author、plugin_name、name、description、supported_mime_types が含まれます

利用可能なツール一覧を取得

現在の LangBot インスタンスで利用可能なすべてのツール（プラグインツールおよび MCP ツール）を取得します。

async def list_tools(self) -> list[dict[str, Any]]:
    """利用可能なすべてのツールを取得

    Returns:
        ツールのリスト。各要素に以下が含まれます：
        - name: ツール名
        - label: 表示名（多言語）
        - description: ツールの説明（多言語）
        - icon: ツールアイコン
        - spec: ツール仕様（llm_prompt と parameters を含む）
    """

# 使用例
tools = await self.plugin.list_tools()
for tool in tools:
    print(f"Tool: {tool['name']}")

ツール詳細を取得

特定のツールの詳細情報を取得します。

async def get_tool_detail(self, tool_name: str) -> dict[str, Any]:
    """特定のツールの詳細情報を取得

    Args:
        tool_name: ツール名（metadata.name、例: "get_weather_alerts"）

    Returns:
        ツール詳細 dict（name、label、description、spec を含む）
    """

# 使用例
detail = await self.plugin.get_tool_detail("get_weather_alerts")
print(detail['spec']['parameters'])

ツールを呼び出す

指定されたツールを呼び出します。

async def call_tool(
    self,
    tool_name: str,
    parameters: dict[str, Any],
    session: dict[str, Any],
    query_id: int,
) -> dict[str, Any]:
    """特定のツールを呼び出す

    Args:
        tool_name: ツール名（metadata.name）
        parameters: ツールパラメータ
        session: セッション情報
        query_id: クエリ ID

    Returns:
        ツールの応答 dict
    """

# 使用例
result = await self.plugin.call_tool(
    tool_name="get_weather_alerts",
    parameters={"state": "CA"},
    session={},
    query_id=0,
)
print(result)

ツール名は metadata.name（例: echo_tool、get_weather_alerts）を使用します。作者やプラグイン名のプレフィックスは不要です。

プラグイン永続データの設定

プラグインデータを永続的に保存します。このインターフェースを通じて保存されたデータは、このプラグインのみがアクセスできます。値は手動でbytesに変換する必要があります。

async def set_plugin_storage(self, key: str, value: bytes) -> None:
    """プラグインストレージ値を設定する"""

# 使用例
await self.plugin.set_plugin_storage("key", b"value")

プラグイン永続データの取得

async def get_plugin_storage(self, key: str) -> bytes:
    """プラグインストレージ値を取得する"""

# 使用例
plugin_storage = await self.plugin.get_plugin_storage("key")

すべてのプラグイン永続データキーの取得

async def get_plugin_storage_keys(self) -> list[str]:
    """すべてのプラグインストレージキーを取得する"""

# 使用例
plugin_storage_keys = await self.plugin.get_plugin_storage_keys()

プラグイン永続データの削除

async def delete_plugin_storage(self, key: str) -> None:
    """プラグインストレージ値を削除する"""

# 使用例
await self.plugin.delete_plugin_storage("key")

ワークスペース永続データの取得

このインターフェースを通じて保存されたデータは、すべてのプラグインがアクセスできます。値は手動でbytesに変換する必要があります。

async def set_workspace_storage(self, key: str, value: bytes) -> None:
    """ワークスペースストレージ値を設定する"""

# 使用例
await self.plugin.set_workspace_storage("key", b"value")

ワークスペース永続データの取得

async def get_workspace_storage(self, key: str) -> bytes:
    """ワークスペースストレージ値を取得する"""

# 使用例
workspace_storage = await self.plugin.get_workspace_storage("key")

すべてのワークスペース永続データキーの取得

async def get_workspace_storage_keys(self) -> list[str]:
    """すべてのワークスペースストレージキーを取得する"""

# 使用例
workspace_storage_keys = await self.plugin.get_workspace_storage_keys()

ワークスペース永続データの削除

async def delete_workspace_storage(self, key: str) -> None:
    """ワークスペースストレージ値を削除する"""

# 使用例
await self.plugin.delete_workspace_storage("key")

プラグインファイル型設定フィールドデータの取得

async def get_config_file(self, file_key: str) -> bytes:
    """設定ファイル値を取得する"""

# 使用例
file_bytes = await self.plugin.get_config_file("key")

これはfileまたはarray[file型の設定フィールドと組み合わせて使用します。

ナレッジベース API

これらのAPIは任意のコンポーネントからself.plugin経由でアクセスでき、パイプライン制限なしにLangBotインスタンス内のすべてのナレッジベースの一覧取得と検索が可能です。

すべてのナレッジベースの一覧取得

LangBotインスタンスで利用可能なすべてのナレッジベースを一覧取得します。

async def list_knowledge_bases(self) -> list[dict[str, Any]]:
    """すべてのナレッジベースを一覧取得

    Returns:
        ナレッジベースのリスト、各要素は以下を含む：
        - uuid: ナレッジベースUUID
        - name: ナレッジベース名
        - description: ナレッジベースの説明
    """

# 使用例
knowledge_bases = await self.plugin.list_knowledge_bases()
for kb in knowledge_bases:
    print(f"KB: {kb['name']} ({kb['uuid']})")

ナレッジベースからの検索

任意のナレッジベースから関連ドキュメントを検索します。

async def retrieve_knowledge(
    self,
    kb_id: str,
    query_text: str,
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """ナレッジベースから検索

    Args:
        kb_id: ナレッジベースUUID（list_knowledge_basesから取得）
        query_text: 検索クエリテキスト
        top_k: 返す結果の数（デフォルト：5）
        filters: オプションのメタデータフィルター

    Returns:
        検索結果エントリのリスト
    """

# 使用例
results = await self.plugin.retrieve_knowledge(
    kb_id="kb-uuid-here",
    query_text="システムの設定方法は？",
    top_k=3,
)
for entry in results:
    print(entry)

これらのAPIはquery_idを必要とせず、任意のコンポーネント（Toolコンポーネントを含む）で使用できます。現在のパイプライン設定に制限されることなく、すべてのナレッジベースにアクセスできます。

RAG API

これらのAPIはKnowledgeEngineコンポーネントがLangBotホストの埋め込みモデル、ベクトルデータベース、ファイルストレージにアクセスするために使用できます。アクセス方法:

KnowledgeEngineコンポーネントクラス内: self.pluginオブジェクトの内部メソッド。

埋め込みモデルの呼び出し

ホストに設定された埋め込みモデルを使用してテキストのベクトルを生成します。

async def invoke_embedding(
    self,
    embedding_model_uuid: str,
    texts: list[str],
) -> list[list[float]]:
    """埋め込みモデルを使用してベクトルを生成

    Args:
        embedding_model_uuid: 埋め込みモデルUUID
        texts: 埋め込むテキストのリスト

    Returns:
        ベクトルのリスト、入力テキストごとに1つ
    """

# 使用例
vectors = await self.plugin.invoke_embedding("model_uuid", ["Hello", "World"])

ベクトルの書き込み

ホストのベクトルデータベースにベクトルを書き込みます。

async def vector_upsert(
    self,
    collection_id: str,
    vectors: list[list[float]],
    ids: list[str],
    metadata: list[dict[str, Any]] | None = None,
    documents: list[str] | None = None,
) -> None:
    """ベクトルの書き込み

    Args:
        collection_id: ターゲットコレクションID
        vectors: ベクトルのリスト
        ids: ベクトルの一意識別子リスト
        metadata: オプションのメタデータリスト
        documents: オプションの生テキストドキュメントリスト。全文検索や
            混合検索をサポートするバックエンドで必要です。
    """

# 使用例
await self.plugin.vector_upsert(
    collection_id="kb_uuid",
    vectors=[[0.1, 0.2, ...], [0.3, 0.4, ...]],
    ids=["chunk_0", "chunk_1"],
    metadata=[{"document_id": "doc1"}, {"document_id": "doc1"}],
    documents=["チャンクテキスト 0", "チャンクテキスト 1"],
)

ベクトル検索

ホストのベクトルデータベースで類似ベクトルを検索します。

async def vector_search(
    self,
    collection_id: str,
    query_vector: list[float],
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
    search_type: str = "vector",
    query_text: str = "",
) -> list[dict[str, Any]]:
    """ベクトル検索

    Args:
        collection_id: ターゲットコレクションID
        query_vector: 類似性検索のクエリベクトル
        top_k: 返す結果の数
        filters: オプションのメタデータフィルター
        search_type: 検索方式、'vector'、'full_text'、'hybrid' のいずれか
        query_text: 生のクエリテキスト、全文検索と混合検索で使用

    Returns:
        検索結果のリスト（id, score, metadataなどを含むdict）
    """

# 使用例
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    top_k=5,
    search_type="hybrid",
    query_text="検索クエリ",
)
# 返却フォーマット: [{"id": "chunk_0", "score": 0.123, "metadata": {"document_id": "doc1", ...}}, ...]

vector_searchが返す各結果はdictで、id（ベクトルID）、score（距離スコア）、metadata（upsert時に提供したメタデータ）の3つのフィールドを含みます。検索結果にテキスト内容が必要な場合は、取り込み時にmetadataにテキストを保存してください。

ベクトルの削除

ホストのベクトルデータベースからベクトルを削除します。

async def vector_delete(
    self,
    collection_id: str,
    file_ids: list[str] | None = None,
    filters: dict[str, Any] | None = None,
) -> int:
    """ベクトルの削除

    Args:
        collection_id: ターゲットコレクションID
        file_ids: 削除するファイルIDのリスト
        filters: オプションのメタデータフィルター

    Returns:
        削除されたアイテム数
    """

# 使用例
deleted = await self.plugin.vector_delete(
    collection_id="kb_uuid",
    file_ids=["doc_001"],
)

filtersパラメータはChromaスタイルのwhere構文によるメタデータフィルタリングをサポートしています。複数のトップレベルキーはAND条件として結合されます。サポートされる演算子：$eq、$ne、$gt、$gte、$lt、$lte、$in、$nin。

# 暗黙的な $eq
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"file_id": "abc"},
)

# 比較演算子
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"created_at": {"$gte": 1700000000}},
)

# リスト演算子
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"file_type": {"$in": ["pdf", "docx"]}},
)

# フィルターによる削除
deleted = await self.plugin.vector_delete(
    collection_id="kb_uuid",
    filters={"file_type": {"$eq": "pdf"}},
)

注意： Chroma、Qdrant、SeekDBは完全なメタデータを保存し、任意のフィールドでフィルタリングできます。MilvusとpgvectorはDB側にtext、file_id、chunk_uuidのみ保存しており、それ以外のフィールドでのフィルタリングは無視されます。

ホストストレージからアップロードされたファイルの内容を取得します。

async def get_knowledge_file_stream(self, storage_path: str) -> bytes:
    """ファイル内容を取得

    Args:
        storage_path: ファイルのストレージパス（FileObject.storage_pathから取得）

    Returns:
        ファイル内容のバイトデータ
    """

# 使用例
file_bytes = await self.plugin.get_knowledge_file_stream(context.file_object.storage_path)

プラグイン開発

プラグイン SDK API

コア開発

プラグイン共通API

リクエストAPI

直接返信メッセージ

Bot UUIDの取得

リクエスト変数の設定

リクエスト変数の取得

すべてのリクエスト変数の取得

現在のパイプラインに設定されたナレッジベース一覧の取得

LangBot API

プラグイン設定の取得

LangBotバージョンの取得

設定済みBotリストの取得

Bot情報の取得

プロアクティブメッセージの送信

設定済みLLMモデルリストの取得

LLMモデルの呼び出し

利用可能な Parser の一覧取得

利用可能なツール一覧を取得

ツール詳細を取得

ツールを呼び出す

プラグイン永続データの設定

プラグイン永続データの取得

すべてのプラグイン永続データキーの取得

プラグイン永続データの削除

ワークスペース永続データの取得

ワークスペース永続データの取得

すべてのワークスペース永続データキーの取得

ワークスペース永続データの削除

プラグインファイル型設定フィールドデータの取得

ナレッジベース API

すべてのナレッジベースの一覧取得

ナレッジベースからの検索

RAG API

埋め込みモデルの呼び出し

ベクトルの書き込み

ベクトル検索

ベクトルの削除

​リクエストAPI

​直接返信メッセージ

​Bot UUIDの取得

​リクエスト変数の設定

​リクエスト変数の取得

​すべてのリクエスト変数の取得

​現在のパイプラインに設定されたナレッジベース一覧の取得

​LangBot API

​プラグイン設定の取得

​LangBotバージョンの取得

​設定済みBotリストの取得

​Bot情報の取得

​プロアクティブメッセージの送信

​設定済みLLMモデルリストの取得

​LLMモデルの呼び出し

​利用可能な Parser の一覧取得

​利用可能なツール一覧を取得

​ツール詳細を取得

​ツールを呼び出す

​プラグイン永続データの設定

​プラグイン永続データの取得

​すべてのプラグイン永続データキーの取得

​プラグイン永続データの削除

​ワークスペース永続データの取得

​ワークスペース永続データの取得

​すべてのワークスペース永続データキーの取得

​ワークスペース永続データの削除

​プラグインファイル型設定フィールドデータの取得

​ナレッジベース API

​すべてのナレッジベースの一覧取得

​ナレッジベースからの検索

​RAG API

​埋め込みモデルの呼び出し

​ベクトルの書き込み

​ベクトル検索

​ベクトルの削除

リクエストAPI

直接返信メッセージ

Bot UUIDの取得

リクエスト変数の設定

リクエスト変数の取得

すべてのリクエスト変数の取得

現在のパイプラインに設定されたナレッジベース一覧の取得

LangBot API

プラグイン設定の取得

LangBotバージョンの取得

設定済みBotリストの取得

Bot情報の取得

プロアクティブメッセージの送信

設定済みLLMモデルリストの取得

LLMモデルの呼び出し

利用可能な Parser の一覧取得

利用可能なツール一覧を取得

ツール詳細を取得

ツールを呼び出す

プラグイン永続データの設定

プラグイン永続データの取得

すべてのプラグイン永続データキーの取得

プラグイン永続データの削除

ワークスペース永続データの取得

ワークスペース永続データの取得

すべてのワークスペース永続データキーの取得

ワークスペース永続データの削除

プラグインファイル型設定フィールドデータの取得

ナレッジベース API

すべてのナレッジベースの一覧取得

ナレッジベースからの検索

RAG API

埋め込みモデルの呼び出し

ベクトルの書き込み

ベクトル検索

ベクトルの削除