📝 Plugin Development Guide

Func

This process may be outdated. It is recommended to start development directly from the template repository. You can also refer to the internal plugin llmkira/extra/.

OpenaiBot provides an OPENAPI interface registration system and template repository for third-party plugins. Plugin template: https://github.com/LlmKira/llmbot_plugin_bilisearch. Please modify this project template for quick development.

TIP

Since the plugin mechanism is implemented based on Nonebot, plugin development is similar to NoneBot Plugin.

Make sure you have installed a code editor and Python environment (version greater than 3.9). Check or view the version by entering python -v in the Shell console or CMD command line.

PDM Environment Management

shell

pip install llmkira
pip install pdm
pdm add <package>
pdm install

📦 Development Process

The plugin is composed of function classes, utility classes, metadata, function functions, and parameter validation classes.

The plugin name inside the function must be referenced by the __plugin_name__ parameter.

🔧 Testing Environment

You can put the plugin folder under Openaibot/llmkira/extra/plugins, and the program will load it automatically.

🍬 Installing the llmkira Framework

shell

pdm add llmkira --dev
# This won't affect the bot

🪣 Understand Architecture Validation

When there are major changes to the plugin system, you need to update the plugin architecture.

The code below demonstrates the architecture validation when the plugin starts.

python

__plugin_name__ = "search_in_bilibili"
__openapi_version__ = "20240416"

from llmkira.sdk.tools import verify_openapi_version  # noqa: E402

verify_openapi_version(__plugin_name__, __openapi_version__)  # Verification

The openapi_version parameter records the current synchronization version. If the host framework updates, the Plugin may need to synchronize this parameter to support new interfaces.

When do I need to update my plugin?

The OpenAPI component sets which versions of plugins can be loaded. If your plugin version is too low, an error will occur, and you will receive an issue from the user.

⚙️ Understanding How to Declare a Utility

It's very simple. We directly inherit the BaseModel class from Pydantic and define the parameters in the class. The underlying code will construct the Schema of the utility directly from the class.

python

from llmkira.sdk.schema import Function
from pydantic import BaseModel, ConfigDict, field_validator, Field
from typing import Optional


class TOOL_NAME(BaseModel):
    """
    TOOL DESCRIPTION
    """
    delay: int = Field(..., description="Arguments description")
    content: str = Field(..., description="Arguments description")
    option_content: Optional[str] = Field(..., description="Arguments description")

    @field_validator("delay")
    def delay_validator(cls, v):
        if v < 0:
            raise ValueError("delay must be greater than 0")
        return v

🩼 Where will this class be used?

The program will pass the function parameters generated by LLM to this class and then instantiate it.

With the help of Pydantic, we can conveniently implement accurate and convenient parameter validation.

python

from pydantic import BaseModel, ConfigDict


class Bili(BaseModel):  # Arguments
    keywords: str
    model_config = ConfigDict(extra="allow")


try:
    _set = Bili.model_validate({"arg": ...})  #
except Exception as e:
    print(e)
    # failed
    pass

⚓️ Function Functions

Function functions are not necessary. We just need to process the parameters passed in the plugin's run method.

🔨 Disable errors

Use this decorator to monitor errors in the function. After the error count is recorded too many times, the function plugin will not be called.

python

from llmkira.sdk.openapi.fuse import resign_plugin_executor


@resign_plugin_executor(function=search, handle_exceptions=(Exception,))
def search_in_bilibili(arg: dict, **kwargs):
    pass

🍭 Plugin Body

You need to inherit the BaseTool class to implement the body. In the lifecycle of the plugin's execution, we will call the run method. If it fails, we will call the failed method.

In the run method, you need to process the parameters passed in, and then communicate with the message queue.

python

async def run(
        self,
        task: "TaskHeader",
        receiver: "Location",
        arg: dict,
        env: dict,
        pending_task: "ToolCall",
        refer_llm_result: dict = None,
):
    """
    Process the message and return the message
    """

    _set = BiliBiliSearch.model_validate(arg)
    _search_result = await search_on_bilibili(_set.keywords)
    _meta = task.task_sign.reprocess(
        plugin_name=__plugin_name__,
        tool_response=[
            ToolResponse(
                name=__plugin_name__,
                function_response=f"SearchData: {_search_result},Please give reference link when use it.",
                tool_call_id=pending_task.id,
                tool_call=pending_task,
            )
        ]
    )
    await Task.create_and_send(
        queue_name=receiver.platform,
        task=TaskHeader(
            sender=task.sender,  # Inherit sender
            receiver=receiver,  # Can be set individually because it may be forwarded
            task_sign=_meta,
            message=[],
        ),
    )

DANGER

After inheriting the BaseTool class, do not define __init__

🎳 Dynamic Activation

To increase the capacity of plugins, we provide the functionality of dynamically activating plugins. The plugin can be activated or not based on the content and user decisions. Every time a new conversation is started, a new function table is built based on the user's sentences, and the plugin selector determines which functions are candidate functions based on character matching.

The func_message function determines whether this function is activated. If it is activated, the function is returned; otherwise, None is returned. If you don't override this function, the plugin will use the keywords and pattern class attributes for matching by default. You are free to override this function.

python

@abstractmethod
def func_message(self, message_text, message_raw, address, **kwargs):
    """
    If the message_text contains the keyword, return the function to be executed, otherwise return None.
    :param message_text: Message text.
    :param message_raw: Raw message data `EventMessage`.
    :param address: Message address `tuple(sender,receiver)`.
    :param kwargs:
    """
    for word in self.keywords:
        if word in message_text:
            return self.function
    # Regrex Match
    if self.pattern:
        match = self.pattern.match(message_text)
        if match:
            return self.function
    _ignore = kwargs
    return None

TIP

When a new conversation chain is started, the function attributes from the previous chain are inherited in the first node of the new chain.

🎳 File Activation

When the message contains a file, the plugin will match it with the file name regular expression. If it matches successfully, the plugin will be activated.

python

class BaseTool(BaseModel):
    file_match_required: Optional[re.Pattern] = Field(
        None, description="re.compile File name regular expression"
    )
    """File name regular expression to use the tool, exp: re.compile(r"file_id=([a-z0-9]{8})")"""

TIP

If you need to use files, define the file_key field in the tool parameter definition. The file is passed to you by LLM. You obtain the file by its file ID.

🧃 Virtual Environment Variables

Declaring whether environment variables are required

Override the require_auth function and return True or False.

python

class BaseTool(BaseModel):
    def require_auth(self, env_map: dict) -> bool:
        """
        Check if authentication is required.
        """
        return True

Declaring the environment variable prefix and required variables

python

class BaseTool(BaseModel):
    env_required: List[str] = Field([], description="Environment variable requirements,ALSO NEED env_prefix")
    """Pre-required environment variables, you should provide env_prefix"""
    env_prefix: str = Field("", description="Environment variable prefix")
    """Environment variable prefix"""

Configuration documentation

Override the env_help_docs function and return the help documentation. This documentation will be sent to the user when variables are missing, attached to the permission application section.

python

@classmethod
def env_help_docs(cls, empty_env: List[str]) -> str:
    """
    Provide help message for environment variables.
    :param empty_env: The list of environment variables that are not configured.
    :return: The help message.
    """
    message = ""
    return message

Getting system environment variables

Call the get_os_env function to obtain system environment variables with the specific prefix PLUGIN_. This variable should be agreed upon by the deployer.

python

@final
def get_os_env(self, env_name):
    """
    Get environment variables from os.environ.
    """
    env = os.getenv("PLUGIN_" + env_name, None)
    return env

🥄 Registering Metadata

Instantiate the core class PluginMetadata to declare all the tools. You can view its structure here.

python

# Name
__plugin_name__ = "search_in_bilibili"
__openapi_version__ = ...
PluginMetadata, FuncPair = ...  # import
# The middle is the function code...

# Core metadata
__plugin_meta__ = PluginMetadata(
    name=__plugin_name__,
    description="Search videos on bilibili.com",
    usage="bilibili search <keywords>",
    openapi_version=__openapi_version__,
    function={
        FuncPair(function=class_tool(BiliBiliSearch), tool=BiliBiliSearch)
    }
)

TIP

FuncPair binds the function class and the tool class.

The class_tool function is used to convert the function class to a tool class.

🍟 Hook

A Hook is a class used to intercept messages and perform message transformation processing between senders and receivers.

The trigger_hook function is used to trigger the hook, and the hook_run function is used to process messages.

Here's an example of a VoiceHook hook. When it returns True, the action specified by the action parameter will be executed.

python

@resign_hook()
class VoiceHook(Hook):
    trigger: Trigger = Trigger.RECEIVER

    async def trigger_hook(self, *args, **kwargs) -> bool:
        platform_name: str = kwargs.get("platform")  # noqa
        messages: List[EventMessage] = kwargs.get("messages")
        locate: Location = kwargs.get("locate")
        for message in messages:
            if not check_string(message.text):
                return False
        have_env = await EnvManager(locate.uid).get_env("VOICE_REPLY_ME", None)
        if have_env is not None:
            return True
        return False

    async def hook_run(self, *args, **kwargs):
        logger.debug(f"Voice Hook {args} {kwargs}")
        platform_name: str = kwargs.get("platform")  # noqa
        messages: List[EventMessage] = kwargs.get("messages")
        locate: Location = kwargs.get("locate")
        for message in messages:
            if not check_string(message.text):
                return args, kwargs
            parsed_text = parse_sentence(message.text)
            if not parsed_text:
                return args, kwargs
            reecho_api_key = await EnvManager(locate.uid).get_env("REECHO_API_KEY", None)
            voice_data = await request_cn(
                message.text, reecho_api_key=reecho_api_key
            )
            if voice_data is not None:
                ogg_data = Ffmpeg.convert(
                    input_c="mp3", output_c="ogg", stream_data=voice_data, quiet=True
                )
                file = await File.upload_file(
                    creator=locate.uid, file_name="speech.ogg", file_data=ogg_data
                )
                file.caption = message.text
                message.text = ""
                message.files.append(file)
            else:
                logger.error(f"Voice Generation Failed:{message.text}")
        return args, kwargs

hook_run is the function responsible for handling message rotation. If an error occurs, it will be skipped automatically. After the parameters are passed in, the returned parameters will be passed to the next hook.

Depending on the hook, we can convert output messages into voice or add attachments to input texts after checking them.

🥥 Pre-Trigger

Use this decorator to block or pass specific responses that meet certain conditions. It is used for sensitive word filtering, responding actively to special language segments without commands, dynamically configuring response triggers, and rejecting certain users' responses, etc.

Here is an example of denying messages from the Telegram platform. When it returns True, the specified action deny will be executed.

python

@resign_trigger(Trigger(on_platform="telegram", action="deny", priority=0))
async def on_chat_message(message: str, uid: str, **kwargs):
    """
    :param message: RawMessage
    :return:
    """
    if "<hello>" in message:
        return True

If the function returns True, it means that a pre-action is required.

TIP

Trigger is a Pydantic class. Please refer to the source code to see possible actions.

🍩 Routing Communication

We use the Meta and Location in the task message to communicate with various platforms. You can directly use the address passed in.

🍬 Communication Mode

You can send messages to users through the message queue.

Source code

The message parameter accepts a list of EventMessage class objects, allowing you to pass messages directly to users.

🍬 Message Delivery

The task_sign.reply method replies to the message, replies to the message directly, and writes it to the memory record. For example: Query completed, your Genshin Impact account is: 123456789. The task_sign.reprocess method reprocesses the non-human-readable data through LLM and replies. For example: {json_data}. The task_sign.notify method sends a notification without triggering any other processing. For example: An error occurred, you did not configure the constants required by the plugin.

TIP

The derivation mentioned here refers to how messages and tool responses are processed based on the routing method, not the functionality.

🎃 Accessing/Creating Files in the Plugin

File communication relies on the context of LLM and the file_key field of the plugin. (Yes, files need to be transmitted through LLM's response)

Create a field that accepts a file ID, and then use the methods of the File class to obtain the file.

📥 Downloading Files

Download files from the global file KV manager.

python

async def run(self, task: TaskHeader, receiver: TaskHeader.Location, arg, **kwargs):
    """
    Process the message and return the message
    """
    GLOBAL_FILE_HANDLER.download_file(file_key)

📤 Uploading Files

Upload files using a convenient construction method. (Actually, it still calls the global file KV manager)

python

from llmkira.kv_manager.file import File


async def test():
    _files = await File.upload_file(
        creator=receiver.uid,
        file_name=file[0],
        file_data=file[1],
    )

📩 Register EntryPoint Group

Documentation reference: https://pdm-project.org/latest/reference/pep621/#entry-points

toml

[project.entry-points."llmkira.extra.plugin"]
bilisearch = "llmbot_plugin_bilisearch"
# <your plugin id>=<your plugin name>

The equal sign is followed by the package name of the plugin, and the front part is the unique key (please ensure that it does not conflict with other plugins).

WARNING

You must register the EntryPoint for the plugin to be detected by the bot's startup program.

🔨 Publish to PyPi

Login to the PyPi repository, create a new package, and then use CI/CD in the template repository to automatically publish it.

pypi

When you configure it like this, CI can automatically publish the package without a key.

yaml

name: publish

on:
  workflow_dispatch:
  push:
    tags:
      - pypi-*

permissions:
  contents: read

jobs:
  pypi-publish:
    name: upload release to PyPI
    runs-on: ubuntu-latest
    permissions:
      # IMPORTANT: this permission is mandatory for trusted publishing
      id-token: write
    steps:
      - uses: actions/checkout@v3

      - uses: pdm-project/setup-pdm@v3

      - name: Publish package distributions to PyPI
        run: pdm publish

🔧 Publish

Create a Release from the repository main page, create a tag starting with pypi-, and it will trigger the automatic publishing.

📝 Plugin Development Guide ​

PDM Environment Management ​

📦 Development Process ​

🔧 Testing Environment ​

🍬 Installing the llmkira Framework ​

🪣 Understand Architecture Validation ​

⚙️ Understanding How to Declare a Utility ​

🩼 Where will this class be used? ​

⚓️ Function Functions ​

🔨 Disable errors ​

🍭 Plugin Body ​

🎳 Dynamic Activation ​

🎳 File Activation ​

🧃 Virtual Environment Variables ​

🥄 Registering Metadata ​

🍟 Hook ​

🥥 Pre-Trigger ​

🍩 Routing Communication ​

🍬 Communication Mode ​

🍬 Message Delivery ​

🎃 Accessing/Creating Files in the Plugin ​

📥 Downloading Files ​

📤 Uploading Files ​

📩 Register EntryPoint Group ​

🔨 Publish to PyPi ​

🔧 Publish ​

📝 Plugin Development Guide

PDM Environment Management

📦 Development Process

🔧 Testing Environment

🍬 Installing the llmkira Framework

🪣 Understand Architecture Validation

⚙️ Understanding How to Declare a Utility

🩼 Where will this class be used?

⚓️ Function Functions

🔨 Disable errors

🍭 Plugin Body

🎳 Dynamic Activation

🎳 File Activation

🧃 Virtual Environment Variables

🥄 Registering Metadata

🍟 Hook

🥥 Pre-Trigger

🍩 Routing Communication

🍬 Communication Mode

🍬 Message Delivery

🎃 Accessing/Creating Files in the Plugin

📥 Downloading Files

📤 Uploading Files

📩 Register EntryPoint Group

🔨 Publish to PyPi

🔧 Publish