上课自动播报
一个真实场景串联四个视角——从节点接入到 AI 调用,看 KnotLink 全链路怎么跑。
- 节点
- 平台
- 配方
- AI
ClassIsland + MultiTTS
两个节点,两种类型,在同一个场景里协作。
| ClassIsland | MultiTTS | |
|---|---|---|
| 类型 | 独立式(standalone) | 插入式(plugin) |
| 做什么 | 课程表管理,按时发信号 | 多引擎 TTS 语音合成 |
| 发现方式 | 注册表 | Plugins/ 目录扫描 |
| 通信代码 | 5 行 | 8 行 |
ClassIsland — 独立式 · 广播信号
安装时写注册表,让 KnotHub 发现:
HKCU\Software\KnotLink\StandaloneNodes
└── 0x00000031 = "C:\Program Files\ClassIsland"
安装目录根放两份清单:standalone_manifest.json + FuncList.json。
下发事件时,一行代码广播:
from knotlink import SignalSender, KLKVMap
sender = SignalSender(APPID="0x00000031", SignalID="0x00000011")
kv = KLKVMap()
kv["event"] = "OnBreakingTime"
sender.emitt(kv.serialize()) # 下课了,广播给所有订阅方
同一个 signalID 发多种事件——OnClass(上课)、OnBreakingTime(下课)、OnAfterSchool(放学),靠 verification 字段区分。
MultiTTS — 插入式 · 响应接口
打成 zip,拖进 KnotHub 操作面板自动安装。收到调用请求时朗读文本:
from knotlink import OpenSocketResponser, KLKVMap
r = OpenSocketResponser(APPID="0x00000022", OpenSocketID="0x00000011")
def handle(data):
req = KLKVMap()
req.deserialize(data)
text = req.get("text", "")
speak(text) # 调用系统 TTS 引擎朗读
return "status=ok"
r.set_RecvFunc(handle)
KnotHub 操作面板
KnotHub 启动后扫描两类来源,两个节点并列展示:
📡 独立式 🖥️ 插入式
┌──────────────┐ ┌──────────────┐
│ ClassIsland │ │ MultiTTS │
│ 来源:注册表 │ │ 来源:Plugins/ │
│ 接口:2个 │ │ 接口:3个 │
│ 信号:4个 │ │ 信号:0个 │
└──────────────┘ └──────────────┘
发现机制不同,展示格式一致。 ClassIsland 从 HKCU\Software\KnotLink\StandaloneNodes 读取安装路径,MultiTTS 从 Plugins/MultiTTS_Client-HXH/ 扫描。但面板里不区分类型——协议层统一,FuncList.json 的格式完全一样。
点击任一节点进入 调试 API 页,参数表单从 FuncList 自动渲染:
| 参数类型 | 效果 |
|---|---|
input | 输入框 |
optional | 下拉菜单 |
static | 灰显固定值 |
API 文档、测试页、节点索引——全部从两份清单自动生成,不需要单独维护。
互联配方:下课自动播报下一节
配方是一个 Python 脚本,放在 MappingDataPacks/ 下,SignalTransfer 加载器自动挂载。不修改 ClassIsland 和 MultiTTS 一行代码。
做什么
- 订阅 ClassIsland 的
OnBreakingTime(下课) - 调用
ClassService查下节课 - 调用
SysTTS朗读课程名
脚本
import knotlink as kl
def main():
sub = kl.SignalSubscriber("0x00000031", "0x00000011")
querier = kl.OpenSocketQuerier("0x00000031", "0x00000011")
tts = kl.OpenSocketQuerier("0x00000022", "0x00000011")
def on_signal(data):
kv = kl.KLKVMap()
kv.deserialize(data)
if kv.get("event") != "OnBreakingTime":
return
# 查下节课
req = kl.KLKVMap()
req["action"] = "GetNextClassSubject"
resp = querier.query(req.serialize())
r = kl.KLKVMap()
r.deserialize(resp)
subject = r.get("subject", "未知课程")
# 朗读
tts_req = kl.KLKVMap()
tts_req["TTS"] = "SysTTS"
tts_req["text"] = f"下一节是{subject}"
tts_req["rate"] = "3"
tts_req["volume"] = "0.7"
tts_req["voice_index"] = "0"
tts.query(tts_req.serialize())
sub.on_signal(on_signal)
input("配方运行中,按回车退出。\n")
if __name__ == "__main__":
main()
配方不关心节点用什么语言写的、跑在哪个目录——它只知道 appID 和 signalID,协议层屏蔽了所有底层细节。
配方开发详见 互联配方教程。
AI 调用本地工具
LLM 通过 FuncList.json 发现本地程序的能力——不需要 MCP 中转,直接走 TCP。
流程
用户:"查一下下节课是什么"
↓
LLM 读取 ClassIsland 的 FuncList → 看到 ClassService 接口
↓
构造 KLUDF 请求:action=GetNextClassSubject
↓
OpenSocketQuerier.query() → TCP → ClassIsland
↓
返回:subject=数学;teacher=张老师;status=成功
↓
LLM 回复用户:"下一节是数学课,张老师。"
代码
from knotlink import OpenSocketQuerier, KLKVMap
# LLM 根据 FuncList 自动生成的调用代码
querier = OpenSocketQuerier("0x00000031", "0x00000011")
req = KLKVMap()
req["action"] = "GetNextClassSubject"
resp = querier.query(req.serialize())
result = KLKVMap()
result.deserialize(resp)
print(result.get("subject")) # "数学"
对 LLM 来说,KnotLink 节点的接口和 function calling 的 tool definition 是同构的——函数名、参数类型、描述都在 FuncList.json 里。你的程序不需要单独为 AI 写 tool schema,清单本身就是 schema。
这个场景覆盖了什么
| 概念 | 在场景中的体现 |
|---|---|
| 独立式节点 | ClassIsland — 注册表发现、自管理进程 |
| 插入式节点 | MultiTTS — zip 分发、Plugins/ 目录 |
| 请求-响应 | ClassService 查询、SysTTS 朗读 |
| 发布-订阅 | ClassIsland 广播 OnBreakingTime |
| 信号鉴别 | verification: "OnBreakingTime" 区分事件 |
| 功能 ≠ 接口 | MultiTTS 三个条目共用 openSocketID: 0x00000011 |
| 工具链生成 | API 测试页、文档从 FuncList 自动生成 |
| 配方编排 | Python 脚本串联两个节点,不改原代码 |
| AI 调用 | LLM 读 FuncList 发现能力,直连 TCP |