Context Engine 插件

Context engine 插件用来完全自定义 agent 的上下文管理：从记忆召回、context window 组装，到 context 压缩和子 agent 生命周期。插件走一个极简的 JSON-over-stdin/stdout 协议，几乎任何能读 stdin、写 stdout 的语言都能写。

本页讲协议、manifest 格式、所有支持的 runtime，以及从零到跑通的完整例子。

目录

• 概览
• Hook 生命周期
• 协议规范
• plugin.toml manifest
• 插件环境变量（[env]）
• Hook 超时覆盖（hook_timeout_secs）
• 支持的 Runtime
• 写你的第一个插件
• 插件叠加（plugin_stack）
• 通过 Dashboard 或 API 生成脚手架
• 热重载端点
• Hook 调用指标
• 本地测试
• 用 Doctor 端点调试
• 错误处理、超时与日志
• Plugin vs. Skill

概览

一个插件就是 ~/.librefang/plugins/<name>/ 下的目录：

my-recall-plugin/
├── plugin.toml               # manifest
└── hooks/
    ├── ingest.py             # （或 .js / .go / .rb / ...）默认激活
    ├── after_turn.py         # 默认激活
    ├── assemble.py           # 已生成模板 —— 在 plugin.toml 中解注释后激活
    ├── compact.py            # 已生成模板 —— 在 plugin.toml 中解注释后激活
    ├── bootstrap.py          # 已生成模板 —— 启动时运行一次（2× 超时）
    ├── prepare_subagent.py   # 已生成模板 —— 子 agent 启动前调用
    └── merge_subagent.py     # 已生成模板 —— 子 agent 完成后调用

LibreFang 加载 manifest，把声明的 hook 脚本挂到 context engine 上，在预定义的生命周期点作为子进程执行。每次调用都是一个全新的子进程：stdin 进一个 JSON 对象，stdout 出一个 JSON 对象。

插件默认沙箱化 —— 环境变量被洗成最小集合，工作目录受控，超时时间可配置（默认 30 秒）。bootstrap hook 获得 2 倍超时，因为它只运行一次，可能需要较长时间连接外部服务（如向量库）。

Hook 生命周期

支持全部 7 个 hook，覆盖 context engine 完整生命周期：

关键特性：

• assemble 是最强的 hook —— 它完全替代默认的 context window 组装，你的脚本决定 LLM 看到的每一条消息。
• ingest 是叠加在内置召回之上的 —— 返回的 memories 会和默认召回结果合并，不是替代。
• after_turn 是尽力而为：失败只记日志，不会反馈给用户。
• 如果 context engine 配置开了 stable_prefix_mode，ingest hook 会被跳过。
• 每个 hook 跑在全新子进程里 —— 调用之间没有任何内存状态。需要持久状态请用外部存储（SQLite、向量库、HTTP 服务）。

协议规范

ingest hook

请求（一行 JSON 写到 hook 的 stdin，然后关闭 stdin）：

{
  "type": "ingest",
  "agent_id": "0f3b…-uuid",
  "message": "我上次问的 Kafka 那件事是什么？",
  "peer_id": "user_12345"
}

响应（stdout 输出 JSON，只有最后一行能 parse 成 JSON 的会被采纳）：

{
  "type": "ingest_result",
  "memories": [
    { "content": "用户在 2026-04-01 问过 Kafka consumer groups 相关问题。" },
    { "content": "之前决定标准化使用 Kafka 而不是 RabbitMQ。" }
  ]
}

after_turn hook

请求：

{
  "type": "after_turn",
  "agent_id": "0f3b…-uuid",
  "messages": [
    { "role": "user", "content": "我上次问的 Kafka 那件事是什么？", "pinned": false },
    { "role": "assistant", "content": "你问过 consumer groups…", "pinned": false }
  ]
}

每条消息的 content 被截断到前 500 字符以保证 hook 跑得快。

响应：

{ "type": "ok" }

返回值被忽略 —— 只是表示完成。吐任何合法 JSON、退出码 0 即可。

bootstrap hook

引擎初始化时调用一次，适合做连接检查、预热缓存等初始化工作。

请求：

{
  "type": "bootstrap",
  "context_window_tokens": 200000,
  "stable_prefix_mode": false,
  "max_recall_results": 5
}

响应：

{ "type": "ok" }

失败只记 warn 日志，不阻止引擎启动。

assemble hook ⭐ 最强 hook

每次 LLM 调用前触发，完全控制 LLM 看到的消息列表。

请求：

{
  "type": "assemble",
  "system_prompt": "你是一个有帮助的助手。",
  "messages": [
    { "role": "user", "content": "帮我查一下 Kafka 的配置", "pinned": false },
    {
      "role": "assistant",
      "content": [
        { "type": "tool_use", "id": "tu_01", "name": "file_read", "input": { "path": "/etc/kafka.conf" } }
      ],
      "pinned": false
    },
    {
      "role": "user",
      "content": [
        { "type": "tool_result", "tool_use_id": "tu_01", "content": "broker=localhost:9092\n...", "is_error": false }
      ],
      "pinned": false
    }
  ],
  "context_window_tokens": 200000
}

消息包含完整结构，包括 tool_use / tool_result / image / thinking 块。pinned: true 的消息不应被裁剪。

响应：

{
  "type": "assemble_result",
  "messages": [...]
}

返回裁剪 / 重排后的消息列表。返回空列表或失败时，自动 fallback 到默认裁剪策略。

compact hook

context 窗口压力过大时触发，用自定义策略压缩历史消息。

请求：

{
  "type": "compact",
  "agent_id": "0f3b…-uuid",
  "messages": [...],
  "model": "llama-3.3-70b-versatile",
  "context_window_tokens": 200000
}

消息格式同 assemble。

响应：

{
  "type": "compact_result",
  "messages": [...]
}

返回压缩后的消息列表。返回空列表或失败时，fallback 到内置 LLM 压缩。

prepare_subagent / merge_subagent hook

子 agent 生命周期钩子，适合需要隔离或合并 memory scope 的场景。

请求：

{ "type": "prepare_subagent", "parent_id": "...", "child_id": "..." }
{ "type": "merge_subagent",   "parent_id": "...", "child_id": "..." }

响应：

{ "type": "ok" }

transform_tool_result hook

工具运行后、LLM 看到结果前重写工具输出。让插件不用改工具本身就能截断、脱敏、屏蔽路径或完全替换工具结果。

在 agent loop 里的位置:

工具执行
  ↓
after_tool_call hook(只观察)
  ↓
transform_tool_result hook  ← 重写发生在这里
  ↓
sanitize + 应用上下文预算
  ↓
结果落到对话历史

Trait 签名(Rust 插件)—— 也作为 stdin/stdout JSON 请求暴露给子进程插件:

fn transform(&self, ctx: &HookContext) -> Result<Option<String>, String>

HookContext 携带:

请求(子进程版):

{
  "type": "transform_tool_result",
  "tool_name": "shell_exec",
  "args": { "command": "cat /etc/passwd" },
  "result": "root:x:0:0:root:/root:/bin/bash\n...",
  "is_error": false
}

响应形状:

语义:

• 首匹配胜出,顺序执行。 插件按注册顺序跑;第一个返回 transformed 结果的插件胜出,链终止。
• Fail-open。 报错的插件被跳过;如果所有插件都 skip 或 error,原始工具结果保持不变。
• 本身不限大小。但 transformed 结果之后仍要过全局上下文预算 sanitiser,所以爆大的输出仍可能在下游被截断。
• Rust 插件的 panic 不会被捕获 —— transform() 里的 panic 会终止 agent 这一轮。源头不可信时用 std::panic::catch_unwind 包一层。

典型用例:

• 截断 —— 保留 shell_exec 输出的前 200 行,后面替换为 "... (N more lines truncated)"。
• 脱敏 —— 用正则一遍剥掉所有工具结果里的 API key / 密码 / token。
• 路径屏蔽 —— 把绝对的 home 目录路径改写成 ~/...,让模型不要记住本地布局。
• 格式转换 —— 把 JSON 工具结果转成对模型更友好的 markdown 表格。
• 内容过滤 —— 从 web-fetch 结果里去掉模板/广告横幅。
• 审计透传 —— 返回 Ok(None) 但把调用回声到外部系统做合规日志。

此 hook 在同一个 Plugin trait 里,跟 ingest / assemble / compact / after_turn 一起。一个插件可以实现任意子集。

错误

以非 0 退出码结束，并把错误写到 stderr。LibreFang 会把 stderr 以 warn 级别记日志，然后回退到默认 context engine 的结果，当前这一轮继续。

plugin.toml manifest

每个插件根目录必须有 plugin.toml：

name = "qdrant-recall"
version = "0.1.0"
description = "从 Qdrant 向量库召回"
author = "Evan"

# hook_timeout_secs = 30  # 每次调用的超时秒数；bootstrap 获得 2 倍此值

[hooks]
# --- 默认激活的 hook ---
ingest    = "hooks/ingest.py"
after_turn = "hooks/after_turn.py"
runtime = "python"

# --- 可选 hook（模板文件已生成，解注释即可激活） ---
# bootstrap        = "hooks/bootstrap.py"          # 启动时运行一次（2× 超时）
# assemble         = "hooks/assemble.py"           # 控制 LLM 看到的内容（最强）
# compact          = "hooks/compact.py"            # 自定义 context 压缩
# prepare_subagent = "hooks/prepare_subagent.py"   # 子 agent 启动前
# merge_subagent   = "hooks/merge_subagent.py"     # 子 agent 完成后

requirements = "requirements.txt"

插件环境变量（[env]）

plugin.toml 里的 [env] 段让插件自行声明需要注入到每个 hook 子进程的环境变量。这样插件专用的配置就不必写进全局 agent 配置，插件可以做到自包含。

name = "qdrant-recall"
version = "0.1.0"

[env]
QDRANT_URL     = "http://localhost:6333"
COLLECTION     = "agent-memories"
QDRANT_API_KEY = "${QDRANT_API_KEY}"

[hooks]
ingest    = "hooks/ingest.py"
after_turn = "hooks/after_turn.py"
runtime   = "python"

环境变量展开： 任何以 ${VAR_NAME} 开头的值，都会在调用时从守护进程自身的环境里展开。若 VAR_NAME 在守护进程环境中未设置，则传入空字符串并记一条 warn 日志。展开只对以 ${ 开头的值生效，"http://localhost:6333" 这样的静态值原样透传。

Hook 脚本可以直接从环境变量读取这些值：

import os
qdrant_url = os.environ["QDRANT_URL"]
api_key    = os.environ.get("QDRANT_API_KEY", "")

优先级顺序（从低到高）：LibreFang 的基础集（LIBREFANG_AGENT_ID、PATH、HOME、runtime 透传变量）→ [env] 中的声明 → agent allowed_env_vars 里列出的变量（最终生效）。

Hook 超时覆盖（hook_timeout_secs）

默认每次 hook 调用有 30 秒超时，超时后子进程会被强制终止。可以在 plugin.toml 的顶层字段（与 name、version 同级，不在 [hooks] 里面）设置 hook_timeout_secs 来覆盖：

name = "qdrant-recall"
version = "0.2.0"

hook_timeout_secs = 60   # 所有 hook 60 s；bootstrap 获得 120 s

[context_engine_hooks]
runtime = "python"
ingest  = "hooks/ingest.py"

bootstrap hook 始终获得 2 倍于配置值的超时，因为它只在启动时运行一次，可能需要额外时间连接外部服务、预热缓存或下载 embedding。

调用远端服务（向量库、embedding API）延迟不稳定时可适当调高；轻量 hook 想要更快检测失败可以调低。

支持的 Runtime

标"否"的没打进官方 Docker 镜像 —— 请参考 Configuration → Plugins 里的 snippets 自己 extend 镜像，或者裸机部署用 apt install 自己装。

**环境变量透传：**PATH/HOME 基础集 + 各 runtime 专属变量（Python 的 PYTHONPATH/VIRTUAL_ENV、Ruby 的 GEM_HOME/GEM_PATH、Lua 的 LUA_PATH 等）会自动透传。其他业务环境变量必须在 agent 的 allowed_env_vars 里显式声明。

写你的第一个插件

做一个按 peer_id 从 SQLite 文件召回用户偏好的插件。为了简洁用 Python。

1. 生成脚手架

Dashboard Plugins 页面 → Create Plugin → 选 python，名字填 prefs。或者调 API：

curl -X POST http://127.0.0.1:4545/api/plugins/scaffold \
  -H 'Content-Type: application/json' \
  -d '{"name":"prefs","description":"User preference recall","runtime":"python"}'

这会创建 ~/.librefang/plugins/prefs/，里面带好 plugin.toml、hooks/ingest.py、hooks/after_turn.py 和一个空的 requirements.txt。

2. 修改 hooks/ingest.py

#!/usr/bin/env python3
"""按 peer_id 从本地 SQLite 召回用户偏好。"""
import json
import sqlite3
import sys
from pathlib import Path

DB_PATH = Path.home() / ".librefang" / "plugins" / "prefs" / "prefs.db"

def main():
    request = json.loads(sys.stdin.read())
    peer_id = request.get("peer_id")          # 直接调 API 时可能是 None
    if not peer_id:
        print(json.dumps({"type": "ingest_result", "memories": []}))
        return

    # Open the per-plugin DB and read anything keyed to this peer.
    conn = sqlite3.connect(DB_PATH)
    conn.execute("CREATE TABLE IF NOT EXISTS prefs (peer_id TEXT, fact TEXT)")
    rows = conn.execute(
        "SELECT fact FROM prefs WHERE peer_id = ?", (peer_id,)
    ).fetchall()
    conn.close()

    memories = [{"content": f"User preference: {row[0]}"} for row in rows]
    print(json.dumps({"type": "ingest_result", "memories": memories}))

if __name__ == "__main__":
    main()

3. 修改 hooks/after_turn.py

这个插件不需要 turn 后处理 —— 直接 ack 就行：

#!/usr/bin/env python3
import json
import sys

_ = json.loads(sys.stdin.read())
print(json.dumps({"type": "ok"}))

或者直接把 plugin.toml 里的 after_turn 字段删掉，就根本不会调这个 hook。

4. 挂给 agent

编辑 ~/.librefang/config.toml：

[context_engine]
plugin = "prefs"

或者手动配 hook（效果一样，不用装插件目录）：

[context_engine.hooks]
ingest = "~/.librefang/plugins/prefs/hooks/ingest.py"
after_turn = "~/.librefang/plugins/prefs/hooks/after_turn.py"
runtime = "python"

5. 重启 LibreFang

librefang start

下次带 peer_id 的消息进来时，你的 ingest hook 就会跑，返回的 memories 会被合并进 agent 的 context window。

插件叠加（plugin_stack）

agent 配置里的 plugin 字段只能挂载一个插件。plugin_stack 允许你把两个或更多插件串联起来，让每个 hook 阶段同时从多个插件获取结果。

[context_engine]
plugin_stack = ["qdrant-recall", "my-indexer"]

数组至少包含两个插件名，每个名字必须对应 ~/.librefang/plugins/ 下的一个目录。插件按数组顺序处理。

链式语义

适用场景

• 召回与索引分离：一个插件做向量库召回（ingest），另一个做 turn 后索引（after_turn），各司其职互不干扰。
• 降级 assembly：把首选 assemble 插件排在前面，简单的 fallback 插件排在后面——链式调用自动使用第一个有非空结果的插件。
• 分层记忆：先用快速本地缓存召回，再用较慢的远端召回——既保证低延迟，又覆盖完整历史。

plugin_stack 和 plugin 不能在同一个 [context_engine] 块里共存。同时存在时，plugin_stack 优先生效。

通过 Dashboard 或 API 生成脚手架

Dashboard 的 Plugins 页面有 Create Plugin 表单，runtime 下拉框选语言，会生成对应语言的模板（Ruby 就是 ingest.rb、Go 就是 ingest.go），已经按协议写好了。

HTTP 等价调用：

curl -X POST http://127.0.0.1:4545/api/plugins/scaffold \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "my-plugin",
    "description": "用途",
    "runtime": "go"
  }'

热重载端点

编辑了插件脚本或 plugin.toml 之后，不用重启守护进程就能让改动生效：

curl -X POST http://127.0.0.1:4545/api/plugins/qdrant-recall/reload

该端点从磁盘重新读取 plugin.toml，并替换内存中的插件配置。哪些变更立即生效、哪些需要重启 agent：

重载成功时返回：

{ "status": "ok", "plugin": "qdrant-recall" }

若更新后的 plugin.toml 解析失败，守护进程保留旧配置并返回 400 错误信息——插件继续正常运行，不受影响。

Hook 调用指标

Context engine 运行时会为每个已安装的插件累积每个 hook 的调用统计。通过以下接口查询：

curl http://127.0.0.1:4545/api/context-engine/metrics

响应示例：

{
  "plugins": {
    "qdrant-recall": {
      "ingest": {
        "calls":       142,
        "successes":   140,
        "failures":      2,
        "latency_ms_total": 8421
      },
      "after_turn": {
        "calls":       138,
        "successes":   138,
        "failures":      0,
        "latency_ms_total": 2760
      }
    }
  }
}

计数器在守护进程重启后清零。可以利用这个端点发现：延迟高的 hook（latency_ms_total / calls 偏大）、频繁失败的 hook（failures / calls 偏高）、从未被调用的 hook（可能是 plugin_stack 配置有误或 hook 路径不正确）。

本地测试

每个 hook 本质就是一个读 stdin 写 stdout 的脚本 —— 不用启动 LibreFang 就能测：

echo '{"type":"ingest","agent_id":"test","message":"hello","peer_id":"u1"}' \
  | python3 ~/.librefang/plugins/prefs/hooks/ingest.py

期望输出：

{"type": "ingest_result", "memories": [...]}

如果脚本 hang 住，那是你没关 stdin 或者在读 stdin 以外的东西。如果打印了非 JSON 的行，只有最后一个能 parse 成 JSON 的行会被当作响应 —— 前面打的 log 行不影响。

用 Doctor 端点调试

GET /api/plugins/doctor 扫描所有支持的 runtime，然后和已安装的插件交叉比对：

curl http://127.0.0.1:4545/api/plugins/doctor

{
  "runtimes": [
    { "runtime": "python", "launcher": "python3", "available": true,
      "version": "Python 3.12.3", "install_hint": "..." },
    { "runtime": "go", "launcher": null, "available": false,
      "version": null, "install_hint": "Install Go from https://go.dev/dl/ ..." }
  ],
  "plugins": [
    { "name": "prefs", "runtime": "python",
      "runtime_available": true, "hooks_valid": true,
      "install_hint": "..." }
  ]
}

插件莫名其妙不生效时先调这个 —— runtime_available: false 意味着 launcher 不在 PATH 上，hooks_valid: false 意味着声明的 hook 脚本在磁盘上不存在。

错误处理、超时与日志

• 超时：每次 hook 调用 30 秒。超了会被 kill，记 Timeout 错误。
• 退出码：非 0 退出会把脚本的 stderr 以 warn 级别记日志，然后用默认 context engine 的结果。
• 空输出：脚本 0 退出但没输出任何东西会返回 EmptyOutput，用默认结果。
• JSON 格式错：runtime 从 stdout 最后一行往前扫，取第一个能 parse 的。都 parse 不了的话，把最后一行包成 { "text": "..." } 兜底。
• 路径穿越：脚本路径里不能带 .. 组件 —— 每次调用都会检查，不只是加载时。
• 环境变量洗白：子进程启动前所有继承的环境变量会被 env_clear 清空。LibreFang 然后设置 LIBREFANG_AGENT_ID、LIBREFANG_MESSAGE、LIBREFANG_RUNTIME，从父进程透传 PATH 和 HOME，加上 runtime 自己的透传集合（Python 的 PYTHONPATH/VIRTUAL_ENV、Ruby 的 GEM_HOME/GEM_PATH 等 —— 见 runtime_passthrough_vars），最后透传 agent 的 allowed_env_vars 里列出的变量。你的 hook 可以从环境变量读 LIBREFANG_AGENT_ID / LIBREFANG_MESSAGE，这是解析 stdin JSON 之外的另一种方式。

Hook 的 stderr 会被捕获后打印到 LibreFang 自己的日志里 —— 想打 debug log 直接往 stderr 写就行。每行 stderr 都会实时通过 tracing 转发出来（target 为 plugin_stderr，Python 工具是 python_stderr），所以长时间运行的 hook 可以把进度流到 journalctl / docker logs，不用等到退出才一口气吐出。用 RUST_LOG=plugin_stderr=info(或 python_stderr=info) 过滤。

hook 退出后，完整的 stderr 还会再以一条 debug! 级别的汇总日志发出来，原本依赖 "hook stderr:" 这个串做抓取的工具不会失效。两个通道相互独立 —— 同时开启的话，每行先实时出现一次、退出后又在汇总里出现一次。

缓冲注意（尤其 Python） —— 大多数语言 runtime 默认按块缓冲 stderr，进度行要等缓冲填满或进程退出才会到 daemon。要实时流式输出，每行后 flush：Python 用 print(..., file=sys.stderr, flush=True)，Ruby 加 STDERR.sync = true。Node 在用户态没有自己的 stdio buffer，process.stderr.write(...) 一旦内核 pipe 排空就到 daemon。或者直接让解释器跑行缓冲模式 (python -u)。

不要把 secret 打到 stderr。 一旦运维开启 RUST_LOG=plugin_stderr=info(或 python_stderr=info)，所有 hook 的每行 stderr 都会进 daemon 日志 —— 然后被 journalctl、docker logs 或任何把日志运出主机的 sink 持久化。Token、API key、PII，以及任何你不想被平台日志保留策略沉淀下来的内容，都不要 print 到 stderr。用 debug 通道（同样受 RUST_LOG 控制）发的内容也得先确认无敏感字段。

Plugin vs. Skill

要改agent 记住什么或做 turn 后记账，用 plugin。要给 agent 加新工具让它在对话中调用，用 skill。