zh plugin dev apis common

插件通用 API

LangBot 为插件提供了一系列 API ，用于操作 LangBot 各个模块、控制消息上下文等。

请求 API

操作当前正在处理的用户请求（消息）。仅在EventListener和Command组件中可用。访问方式如下：

• EventListener 的各个事件处理器方法中：event_context: context.EventContext对象内部方法
• Command 的各个子命令处理器方法中：context: ExecuteContext对象内部方法

直接回复消息

直接回复一个消息链到当前请求所在会话。

消息链构造方式请参考消息平台实体。

async def reply(
    self, message_chain: platform_message.MessageChain, quote_origin: bool = False
):
    """Reply to the message request

    Args:
        message_chain (platform.types.MessageChain): LangBot message chain
        quote_origin (bool): Whether to quote the original message
    """

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

获取机器人 UUID

获取当前请求来源机器人 UUID。

async def get_bot_uuid(self) -> str:
    """Get the bot uuid"""

# 调用示例
bot_uuid = await event_context.get_bot_uuid()

设置请求变量

单次请求中某些信息会被存储到请求变量中，在使用 Dify 等外部 Agent 平台时，这些变量会显式传入 Agent 平台。

async def set_query_var(self, key: str, value: Any):
    """Set a query variable"""

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

获取请求变量

获取单个请求变量。

async def get_query_var(self, key: str) -> Any:
    """Get a query variable"""

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

获取所有请求变量

async def get_query_vars(self) -> dict[str, Any]:
    """Get all query variables"""

# 调用示例
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]:
    """Get the config of the plugin."""

# 调用示例
config = self.plugin.get_config()

获取 LangBot 版本

获取 LangBot 版本号，返回格式为字符串v<major>.<minor>.<patch>。

async def get_langbot_version(self) -> str:
    """Get the langbot version"""

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

获取已配置的机器人列表

返回由所有机器人 UUID 组成的列表。

async def get_bots(self) -> list[str]:
    """Get all bots"""

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

获取机器人信息

获取机器人信息。

async def get_bot_info(self, bot_uuid: str) -> dict[str, Any]:
    """Get a bot info"""

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

# 返回示例
{
    "uuid": "de639861-be05-4018-859b-c2e2d3e0d603",
    "name": "aiocqhttp",
    "description": "由 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_account_id": 960164003  # 机器人账号 ID
    }
}

发送主动消息

通过机器人 UUID 和目标会话 ID 发送主动消息。

消息链构造方式请参考消息平台实体。

async def send_message(
    self,
    bot_uuid: str,
    target_type: str,
    target_id: str,
    message_chain: platform_message.MessageChain,
) -> None:
    """Send a message to a session"""

# 调用示例
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]:
    """Get all LLM models"""

# 调用示例
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:
    """Invoke an LLM model"""

# 调用示例
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 插件，可按 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]]:
    """List all available tools

    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]:
    """Get detailed information about a specific tool

    Args:
        tool_name: 工具名称（metadata.name，例如 "get_weather_alerts"）

    Returns:
        工具详情 dict，包含 name、label、description、spec（含 parameters 和 llm_prompt）
    """

# 调用示例
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]:
    """Call a specific tool

    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)

Tip

工具名称使用 metadata.name（如 echo_tool、get_weather_alerts），不需要包含作者或插件名前缀。

设置插件持久化数据

持久化存储插件数据。通过这个接口存储的数据，仅可被该插件访问。值需要自行转换为字节。

async def set_plugin_storage(self, key: str, value: bytes) -> None:
    """Set a plugin storage value"""

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

获取插件持久化数据

async def get_plugin_storage(self, key: str) -> bytes:
    """Get a plugin storage value"""

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

获取插件所有持久化数据键

async def get_plugin_storage_keys(self) -> list[str]:
    """Get all plugin storage keys"""

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

删除插件持久化数据

async def delete_plugin_storage(self, key: str) -> None:
    """Delete a plugin storage value"""

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

获取工作空间持久化数据

通过这个接口存储的数据，可被所有插件访问。值需要自行转换为字节。

async def set_workspace_storage(self, key: str, value: bytes) -> None:
    """Set a workspace storage value"""

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

获取工作空间持久化数据

async def get_workspace_storage(self, key: str) -> bytes:
    """Get a workspace storage value"""

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

获取工作空间所有持久化数据键

async def get_workspace_storage_keys(self) -> list[str]:
    """Get all workspace storage keys"""

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

删除工作空间持久化数据

async def delete_workspace_storage(self, key: str) -> None:
    """Delete a workspace storage value"""

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

获取插件配置项文件数据

async def get_config_file(self, file_key: str) -> bytes:
    """Get a config file value"""

# 调用示例
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)

Tip

这些 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:
        向量列表，每个输入文本对应一个向量
    """

# 调用示例
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 等字段）
    """

# 调用示例
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", ...}}, ...]

Note

vector_search 返回的每个结果为 dict，包含 id（向量 ID）、score（距离分数）和 metadata（写入时附带的元数据）三个字段。如果需要在检索结果中返回文本内容，请在摄取阶段将文本存入 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"],
)

Note

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 仅存储 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)