插件开发指南(高级)
本章面向需要深度扩展 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"]会被合并到全局 settingsDEFAULT_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")