组件开发 · 第 2 篇 / 共 3 篇 —— 动手

自定义组件

这一篇从零写一个能跑通的组件:文件放哪、类怎么起名、方法怎么变成 LLM 能调用的命令。

前置:第 1 篇讲了加载链路、三层注册表和沙箱模型。没读过的话,后面很多行为会显得没来由。

组件放在哪

用 component.dir() 查真实路径。pip 安装模式和绿色版的目录不同,别手动猜。

# 查询真实组件目录(pip 安装与绿色版路径不同)
component.dir()

# pip 安装模式 → 用户数据目录
# ~/.local/share/cellium/components/
# 绿色版 → 程序目录
# ./components/

最小组件

一个 cell_name 加若干 _cmd_ 方法就够了,其余全部自动生成。

# components/hash_tool.py
from typing import Any, Dict
from app.core.interface.base_cell import BaseCell


class HashTool(BaseCell):
    @property
    def cell_name(self) -> str:
        return "hash_tool"          # 工具名,必须唯一

    def _cmd_md5(self, text: str) -> Dict[str, Any]:
        """计算文本的 MD5

        Args:
            text: 待计算的文本
        """
        import hashlib
        return {"success": True, "md5": hashlib.md5(text.encode()).hexdigest()}

    def _cmd_help(self, topic: str = "") -> Dict[str, Any]:
        """查看本组件用法"""
        return {"success": True, "usage": "hash_tool.md5(text=...)"}
Naming

_cmd_ 后面的名字就是命令名;方法的 docstring 会成为 LLM 看到的工具描述,写得清楚一点,否则模型不会正确调用。

参数怎么传给 LLM

工具的 JSON Schema 由方法签名 + docstring 的 Args 段自动推导(推导规则见第 1 篇)。写法上把下面几处对齐就行:

源码里写的LLM 看到的
cell_namecell_name
_cmd_xxx_cmd_xxx 方法
docstring 首行工具/命令描述
Args: 参数说明参数 description
def _cmd_query(self, keyword: str, limit: int = 5) -> Dict[str, Any]:
    """按关键词查询

    Args:
        keyword: 查询关键词
        limit: 返回条数上限
    """

# 自动生成给 LLM 的 JSON Schema:
# {"name": "hash_tool", "parameters": {
#   "properties": {
#     "keyword": {"type": "string", "description": "查询关键词"},
#     "limit":   {"type": "integer", "description": "返回条数上限"}}}}

让 Agent 自己写组件

三种方式,按复杂度选:

取模板

# 取模板:minimal / full / example / background
component.template(style="background")

生成单文件组件

# 单文件组件 → components/hash_tool.py
component.generate(
  name="hash_tool",
  description="文件哈希计算",
  commands='[{"name":"md5","desc":"计算 MD5"},{"name":"sha256","desc":"计算 SHA256"}]'
)

生成包组件

功能复杂、需要拆多个模块时用包形式,主组件类放 core.py。

component.generate_package(
  name="data_pipeline",
  description="数据处理管道",
  commands='[{"name":"run","desc":"运行管道"},{"name":"status","desc":"查看状态"}]',
  modules='["utils","monitor"]'
)

# 生成结构:
# components/data_pipeline/
# ├── __init__.py      # 导出主组件类
# ├── core.py          # 主组件类
# ├── utils.py         # 额外模块
# └── monitor.py

验证生效

加载链路在第 1 篇讲过,这里只讲怎么确认自己写的组件真的能用。

  1. component.list() 里能看到它,说明已注册
  2. component.info(name) 看命令表和加载错误
  3. 用真实参数调一次,返回结构符合预期
  4. 改完代码 3 秒内自动重载;包组件内部改动要 component.reload()
# 手动触发扫描,不必等 3 秒
component.reload()

什么时候该进白名单

默认子进程隔离,模型见第 1 篇。只有两种情况值得加入免沙箱白名单:需要跨命令共享内存状态,或者单次调用经常贴近 120 秒超时。纯计算、无状态的组件留在沙箱里更安全。

component.sandbox_list()              # 查看内置 + 自定义白名单
component.sandbox_add(name="hash_tool")   # 加入白名单,立即热重载
component.sandbox_remove(name="hash_tool") # 移出白名单,恢复隔离
Sandbox

自定义白名单改动会立即触发热重载。内置白名单(component、scheduler、skill_manager、skill_installer、sub_agent、web_fetch、web_search 及四个 IM 文件组件)硬编码在代码里,不可移除。

调试命令

命令用途
component.list()列出所有组件及命令
component.info(name)看单个组件的类名、文件路径、命令、加载状态
component.reload()手动触发扫描,不用等 3 秒
component.cleanup()清理组件系统缓存与残留沙箱(dry_run 默认只预览)

容易踩的坑

  • cell_name 必须全局唯一,重名会覆盖先加载的那个。
  • 返回值统一用 Dict,别返回裸字符串,否则 LLM 拿到的是不可解析的对象。
  • import 阶段别做重活(连数据库、发请求),沙箱子进程启动会卡住。
  • 沙箱里的组件拿不到主进程内存状态,需要共享状态就走白名单或改用文件/接口通信。
  • 改完文件不用重启,但语法错误会导致该组件加载失败——用 component.info() 看报错。