深入 LangBot 插件系统：进程隔离、事件驱动与组件化架构

# 请求
class ActionRequest(pydantic.BaseModel):
    seq_id: int    # 序列号，用于匹配请求和响应
    action: str    # 动作名称
    data: dict     # 载荷

# 响应
class ActionResponse(pydantic.BaseModel):
    seq_id: int
    code: int           # 0 = 成功
    message: str
    data: dict
    chunk_status: str   # "continue" | "end"（支持流式响应）

class Handler:
    async def call_action(self, action, data, timeout=15.0) -> dict:
        """主动调用对端提供的 action，等待响应"""
        self.seq_id_index += 1
        request = ActionRequest.make_request(self.seq_id_index, action.value, data)
        future = asyncio.Future()
        self.resp_waiters[self.seq_id_index] = future
        await self.conn.send(json.dumps(request.model_dump()))
        response = await asyncio.wait_for(future, timeout)
        return response.data

    @action(SomeAction.DO_SOMETHING)
    async def handle_something(data: dict) -> ActionResponse:
        """注册一个 action 供对端调用"""
        return ActionResponse.success({"result": "ok"})

# 插件 → Runtime（插件进程主动请求）
class PluginToRuntimeAction:
    REGISTER_PLUGIN = "register_plugin"
    SEND_MESSAGE = "send_message"        # 发送消息到平台
    INVOKE_LLM = "invoke_llm"           # 调用大模型
    SET_PLUGIN_STORAGE = "set_plugin_storage"  # 持久化存储
    # ...

# Runtime → 插件（Runtime 下发指令）
class RuntimeToPluginAction:
    INITIALIZE_PLUGIN = "initialize_plugin"
    EMIT_EVENT = "emit_event"            # 触发事件
    CALL_TOOL = "call_tool"             # 调用工具
    EXECUTE_COMMAND = "execute_command"   # 执行命令
    SHUTDOWN = "shutdown"
    # ...

# LangBot 主进程 → Runtime
class LangBotToRuntimeAction:
    INSTALL_PLUGIN = "install_plugin"
    EMIT_EVENT = "emit_event"
    LIST_TOOLS = "list_tools"
    # ...

# Runtime → LangBot 主进程
class RuntimeToLangBotAction:
    GET_PLUGIN_SETTINGS = "get_plugin_settings"
    SET_BINARY_STORAGE = "set_binary_storage"
    # ...

async def launch_all_plugins(self):
    for plugin_path in glob.glob("data/plugins/*"):
        if not os.path.isdir(plugin_path):
            continue
        task = self.launch_plugin(plugin_path)
        self.plugin_run_tasks.append(task)

async def launch_plugin(self, plugin_path: str):
    python_path = sys.executable
    args = ["-m", "langbot_plugin.cli.__init__", "run", "-s", "--prod"]

    ctrl = StdioClientController(
        command=python_path,
        args=args,
        working_dir=plugin_path,  # 每个插件在自己的目录下运行
    )
    await ctrl.run(new_plugin_connection_callback)

# 插件侧
action = PluginToRuntimeAction.REGISTER_PLUGIN
# 携带：manifest、组件列表、配置 schema 等

# Runtime 侧处理注册
async def register_plugin(self, handler, container_data, debug_plugin=False):
    plugin_container = PluginContainer.from_dict(container_data)
    # 从主进程获取插件配置
    plugin_settings = await self.context.control_handler.call_action(
        RuntimeToLangBotAction.GET_PLUGIN_SETTINGS, {...}
    )
    # 初始化插件（下发配置）
    await handler.initialize_plugin(plugin_settings)
    # 存储插件容器
    self.plugins.append(plugin_container)

async def shutdown_plugin(self, plugin_container):
    # 1. 通知插件进程关闭
    await plugin_container._runtime_plugin_handler.shutdown_plugin()
    # 2. 关闭通信连接
    await plugin_container._runtime_plugin_handler.conn.close()
    # 3. 终止子进程
    if handler.stdio_process is not None:
        handler.stdio_process.kill()
        await asyncio.wait_for(handler.stdio_process.wait(), timeout=2)

from langbot_plugin.api.definition.components.common.event_listener import EventListener
from langbot_plugin.api.entities.events import PersonNormalMessageReceived
from langbot_plugin.api.entities.context import EventContext

class MyListener(EventListener):
    @EventListener.handler(PersonNormalMessageReceived)
    async def on_person_message(self, ctx: EventContext):
        event = ctx.event
        # 修改用户消息（在发送给 LLM 之前）
        event.user_message_alter = "请用诗歌形式回答：" + event.text_message

        # 或者阻止后续处理
        # ctx.prevent_default()
        # ctx.prevent_postorder()

from langbot_plugin.api.definition.components.tool.tool import Tool

class WeatherTool(Tool):
    async def call(self, params: dict, session, query_id: int) -> str:
        city = params.get("city", "Beijing")
        # 调用天气 API ...
        return f"{city} 今天晴，25°C"

from langbot_plugin.api.definition.components.command.command import Command

class MyCommand(Command):
    def __init__(self):
        super().__init__()

        @self.subcommand("hello", help="Say hello")
        async def hello(self, ctx):
            yield CommandReturn(text="Hello from plugin!")

        @self.subcommand("status", help="Show status")
        async def status(self, ctx):
            yield CommandReturn(text="All systems operational.")

from langbot_plugin.api.definition.components.knowledge_retriever.retriever import KnowledgeRetriever

class MyRetriever(KnowledgeRetriever):
    async def retrieve(self, context) -> list:
        # 从外部向量数据库检索
        results = await self.search_external_db(context.query)
        return [RetrievalResultEntry(content=r) for r in results]

class LangBotAPIProxy:
    # 消息操作
    async def send_message(self, bot_uuid, target_type, target_id, message_chain)

    # 模型调用
    async def get_llm_models(self) -> list[str]
    async def invoke_llm(self, model_uuid, messages, funcs=[], extra_args={})

    # 持久化存储（插件级别隔离）
    async def set_plugin_storage(self, key, value: bytes)
    async def get_plugin_storage(self, key) -> bytes

    # 工作区存储（跨插件共享）
    async def set_workspace_storage(self, key, value: bytes)
    async def get_workspace_storage(self, key) -> bytes

    # 系统信息
    async def get_langbot_version(self) -> str
    async def get_bots(self) -> list[str]
    async def list_plugins_manifest(self) -> list

async def emit_event(self, event_context, include_plugins=None):
    for plugin in self.plugins:
        if plugin.status != RuntimeContainerStatus.INITIALIZED:
            continue
        if not plugin.enabled:
            continue

        # 流水线级别的插件过滤
        if include_plugins is not None:
            plugin_id = f"{plugin.manifest.metadata.author}/{plugin.manifest.metadata.name}"
            if plugin_id not in include_plugins:
                continue

        resp = await plugin._runtime_plugin_handler.emit_event(
            event_context.model_dump()
        )

        event_context = EventContext.model_validate(resp["event_context"])

        # 插件请求中断后续传播
        if event_context.is_prevented_postorder():
            break

    return emitted_plugins, event_context

async def install_plugin(self, source, install_info):
    yield {"current_action": "downloading plugin package"}
    # 1. 获取插件包（解压 zip）
    plugin_path, author, name, version = await self.install_plugin_from_file(plugin_file)

    yield {"current_action": "installing dependencies"}
    # 2. 安装依赖（pip install -r requirements.txt）
    pkgmgr_helper.install_requirements(requirements_file)

    yield {"current_action": "initializing plugin settings"}
    # 3. 初始化配置
    await self.context.control_handler.call_action(
        RuntimeToLangBotAction.INITIALIZE_PLUGIN_SETTINGS, {...}
    )

    yield {"current_action": "launching plugin"}
    # 4. 启动插件进程
    task = self.launch_plugin(plugin_path)

# 初始化新插件
lbp init

# 添加组件
lbp component add

# 本地调试运行
lbp run

# 打包发布
lbp publish