Skip to content

插件开发指南(高级)

本章面向需要深度扩展 Agently 行为的开发者。目标不是教你“怎么用”,而是讲清插件体系的职责边界、组合方式与工程落地方法

读者画像

  • 需要对接新的模型协议或私有推理服务
  • 需要自定义提示词格式 / 输出解析 / 工具调用逻辑
  • 需要接入外部会话管理或自建记忆系统
  • 需要在流式与结构化输出上做稳定可控的工程方案

先分清:插件 vs 扩展

  • 插件(Plugin):系统级能力替换或增强(PromptGenerator / ModelRequester / ResponseParser / ToolManager / Session)
  • 扩展(Extension):面向 Agent 的胶水层(如 SessionExtension / ToolExtension)

如果你要改变模型调用协议返回解析方式Prompt 结构工具体系会话管理,就应该写插件; 如果只是想在 Agent 请求前后插逻辑,用扩展更合适。

一张图看清调用链

插件注册与激活机制

插件由 PluginManager 管理:

  • 注册:plugin_manager.register("PromptGenerator", MyPromptGenerator)
  • 激活:settings.set("plugins.PromptGenerator.activate", "MyPromptGenerator")

注册时会自动写入默认配置:

python
class MyPlugin:
    name = "MyPlugin"
    DEFAULT_SETTINGS = {
        "$global": {"runtime.show_model_logs": True},
        "$mappings": {"key_value_mappings": {"debug": {True: {"runtime.httpx_log_level": "INFO"}}}},
        "my_plugin": {"timeout": 30},
    }
  • DEFAULT_SETTINGS["$global"] 会被合并到全局 settings
  • DEFAULT_SETTINGS["$mappings"] 会注入到 settings 的映射系统
  • 其余会写入 plugins.<Type>.<Name> 命名空间

一分钟最小插件示例(ResponseParser)

下面示例只做“把 done 事件拼成 text_result”,适合快速验证链路:

python
from agently.types.plugins import ResponseParser

class MinimalParser(ResponseParser):
    name = "MinimalParser"
    DEFAULT_SETTINGS = {}

    def __init__(self, agent_name, response_id, prompt, response_generator, settings):
        self.agent_name = agent_name
        self.response_id = response_id
        self.response_generator = response_generator
        self.settings = settings
        self.full_result_data = {
            "text_result": "",
            "parsed_result": None,
            "original_delta": [],
            "original_done": {},
            "meta": {},
            "errors": [],
            "extra": {},
        }

    def get_text(self):
        return self.full_result_data["text_result"]

    async def async_get_text(self):
        async for event, data in self.response_generator:
            if event == "done":
                self.full_result_data["text_result"] = str(data)
        return self.full_result_data["text_result"]

注册并启用:

python
from agently.base import plugin_manager, settings

plugin_manager.register("ResponseParser", MinimalParser)
settings.set("plugins.ResponseParser.activate", "MinimalParser")

下一步