メインコンテンツへスキップ
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などの外部LLMOpsプラットフォームを使用する場合、これらの変数はLLMOpsプラットフォームに明示的に渡されます。 
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 を使っていない場合は空配列になります。

​
現在のパイプライン範囲でナレッジを検索

現在のリクエストのパイプラインで許可されているナレッジベースの範囲内で検索を実行します。 
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]]:
    """Retrieve relevant documents from a knowledge base"""

# 使用例
knowledge_bases = await event_context.list_pipeline_knowledge_bases()
results = await event_context.retrieve_knowledge(
    kb_id=knowledge_bases[0]["uuid"],
    query_text="エンタープライズ向けナレッジベースの設定方法は？",
    top_k=3,
    filters={"document_type": {"$eq": "manual"}},
)

# 返却例
[
    {
        "id": "chunk_0",
        "content": [{"type": "text", "text": "ナレッジベース設定ガイド"}],
        "metadata": {"document_id": "doc_001"},
        "distance": 0.12,
        "score": 0.88,
    }
]
kb_id は list_pipeline_knowledge_bases の返却値から取得し、現在のパイプラインに設定されたナレッジベースである必要があります。そうでない場合、LangBot はエラーを返します。

​
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 が含まれます

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

プラグインデータを永続的に保存します。このインターフェースを通じて保存されたデータは、このプラグインのみがアクセスできます。値は手動で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型の設定フィールドと組み合わせて使用します。

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