mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-22 18:08:04 +00:00
feat(sdk): 多模态贯通插件边界——公开接口、内核桥接与统一输入主干
记忆系统在 1.1.0 支持了二进制多媒体节点,但那条链路只对**内核自己**开放: 用户在 qq 发图能落进 CAS、能被记忆引用,而插件调 Commit / DocMemory().Insert 交进来的媒体一律无处安放。原因是三层都断着,且**每一层都不报错**。 ## 一、公开 SDK:补上媒体的表达能力(全部新增,无签名变更) - `Triple` += `SentenceText`、`MediaDigests` - `Doc` += `MediaDigests`、`Attachments`;新增 `MediaAttachment` - `TextEvent` += `Attachments` - `DocMemoryAPI` += `InsertWithMedia` - `IOInjector` += `InjectInputMedia` / `InjectInputMediaSync` / `InjectInterruptMedia` - `PluginSDK` 补上一直缺失的 `SetToolBlocks` 包装(接口里有、便捷方法里没有) `MediaAttachment` 一个类型服务两个方向:给 `Data`+`MIME` 是新内容(CAS 按字节 去重),只给 `Digest` 是引用已有内容。读路径**只回元数据不回字节**——一次检索 可能命中几十份媒体,全塞回去会把跨进程消息撑爆。 媒体注入不能搭 `SetToolBlocks` 的车:那个方法只在工具处理函数内部可用,且媒体 要等下一条 tool message 才到模型手上。插件主动发起一轮带媒体的对话、以及中断 注入,需要自己的签名,且媒体在**本轮**就送到模型。 ## 二、内核桥接层:原先在静默裁字段 `internal/sdk/memory_impl.go` 此前只搬自己认识的几个字段,其余丢弃且返回 nil: - 图记忆丢 `Confidence`/`SubjectType`/`ObjectType`/`SentenceText`,又走 `Commit` 而非 `CommitWithMedia`(不回 sentenceIDs)→ 媒体绑定链 `SentenceText → sentences → sentence_id → media_refs` 一步都走不通,插件即便按格式写好标记也永远挂不上; - 知识库 `Query` 只回 ID/Title/Content,`Insert` 只写这三个;`Remove` 不解引用, 于是那些媒体永久处于「被引用」状态,GC 收不掉、磁盘只增不减 (内核的归档路径 `releaseDocMedia` 做了这一步,插件路径漏了同一步)。 规则改为:**内部结构有的字段一律透传**。标记格式处理作为包级私有辅助留在桥接 层自己手里,但必须与内核 `mediaSummaryForEvent` 字节兼容——两边要能互读对方 写下的标记。 标记插入必须在 `ds.Insert` **之前**(向量索引取 `Summary + " " + Content`, 之后补的标记检索不到),引用绑定必须在**之后**(owner_id 是 Insert 生成的 ID)。 ## 三、跨进程链路:不接线就是全体外部插件编译失败 `go test` 直接把这一层拍出来了——`procIO does not implement sdk.IOInjector`。 公开接口加方法后,生成模板不跟上,**每个外部插件都编不过**,是硬失败不是软降级。 六处接线:`protocol.go` 四个 method 常量、`capability.go` 能力归属、 `corehandler.go` 四个分派分支、`proc_core.go` 委托、`proc_main.go.tmpl` 模板侧 实现、以及三个测试替身。 ## 四、统一输入主干:把模态从「函数选择」降级为「字段」 `processTextInput` / `processMediaInput` 合并为 `processInput`。这个分叉是历史 产物而非设计:`processTextInput` 本来就处理媒体(`bindEventMedia` + `mediaSummaryForEvent`,与媒体路径尾部完全相同),`process()` 只看 `stageCtx.Extra["media_blocks"]`、根本不认识 `evt.Type`。模态是输入的**属性**, 不是输入的**种类**。 媒体路径由此获得它一直缺的六项:去重、`no_memory`、通道 `Cleaner`、中断语义、 `_consolidation_` 路由、正确的 `EventRawInput`。 最后一项是个真 bug:媒体路径发布 `"content": evt.Payload`(一个 map),而 `webui/handler.go` 断言 `.(string)` → 断言失败、`content == ""`、提前返回。 **用户发的图从来没出现在 WebUI 聊天记录里。** `media_blocks` 同时接受 `[]agentAPI.ContentBlock` 与 `[]pubsdk.ContentBlock`: 字段一致但 Go 不自动转换,只认一种的后果是另一种被静默丢弃。 ## 五、模型可调用的三个工具 `memory_commit` 的 `sentence_text` **从未暴露给模型**,而它是绑定链上的必经环节; 连同 `media_digests` 一起补进 JSON schema 与工具文档。`doc_commit` 加 `media_digests`。`doc_query` 把关联媒体单独一行附在结果末尾(正文按 2000 字截断, 标记通常就在尾部)。 标记由**内核**生成而非插件/模型拼装:要求调用方知道格式,等于让一个拼写错误 静默切断引用绑定,而全链路无人报错。 ## 六、WebUI 上传走真实媒体链路 图片/音频读回字节拼 data URL 注入 `media_blocks`(8MB 上限,超限退回按路径处理)。 此前只注入一句「文件已保存到 <路径>」,指望模型自己调 `files_read`——但那返回 文本,图片字节对模型永远不可见。附件类型识别扩展到 audio 并在缺 Content-Type 时按扩展名兜底(判错不只是卡片样式问题,图片被当普通文件就进不了视觉链路)。 ## 测试 - `internal/sdk/memory_impl_test.go`(12 例,此前该包**没有任何测试文件**) - `internal/agent/core/inputunify_test.go`(统一主干 + 双静态类型 + 三工具媒体) - `third_party/homeagent-sdk/sdk/stress_test.go`(13 例并发压测) 压测抓到两处**真**竞态(不是理论风险):`PluginSDK` 的 API 字段与 `autoRestart` 无锁,而写方(内核注入 API、插件 `SetAutoRestart`)与读方(插件后台 goroutine 注入、内核 registry 读 `AutoRestart`)天然跨 goroutine。加 `apiMu` 修掉;约定 只在持锁期间取字段值,取完即释放再调用——持锁调用会把 `InjectInputSync` 这类 阻塞到 agent 回复(可达数分钟)的方法与 `SetIOInjector` 串起来,让插件重载卡死。 测试还抓出两个自身缺陷:`bindDocMedia` 把同一份媒体数两次(`AddRef` 幂等所以表 是对的,但日志说「绑定 2 个」而实际 1 条——误导后续排查),以及用单字符实体名 时 `validEntityName` 静默跳过、`Commit` 返回 nil 却什么都没写。 存量插件不需要改一行也不需要重编:新增方法由插件调用、内核实现,不调就不受影响。 17 个 example 插件源码零改动通过类型检查。
This commit is contained in:
30
third_party/homeagent-sdk/meta/meta.go
vendored
30
third_party/homeagent-sdk/meta/meta.go
vendored
@ -6,9 +6,30 @@ var (
|
||||
// Version 是 HomeAgent SDK 版本号。
|
||||
// 通过 `-ldflags="-X gitcode.com/JianFeeeee/homeagent-sdk/meta.Version=vX.Y.Z"` 注入。
|
||||
//
|
||||
// 版本号语义:**SDK 版本跟随核心的中版本,patch 位恒为 .0**。
|
||||
// 整条核心 1.1.x 线(1.1.0、1.1.1、1.1.7…)共用 SDK 1.1.0;
|
||||
// 只有核心进入 1.2.0 这种中版本跃迁时 SDK 才升到 1.2.0。
|
||||
// 这样插件开发者只需关心「我在为哪个中版本写插件」,
|
||||
// 不必跟着核心的每个 bugfix 换 SDK 依赖(见 核心仓 docs/git-branching.md §七)。
|
||||
//
|
||||
// 1.0.0:插件运行模型从 C ABI 动态库改为子进程 + 共享内存。
|
||||
// 公开 SDK 接口(sdk/ 目录)**零改动**——插件业务代码不需要改一行,
|
||||
// 但产物形态变了(plugin.so → plugin.bin),必须用新版 plugindev 重编。
|
||||
// 公开 SDK 接口零改动,但产物形态变了(plugin.so → plugin.bin)。
|
||||
// 1.1.0:多模态贯通插件边界。**全部是新增,无签名变更**:
|
||||
// - Triple.SentenceText / Triple.MediaDigests
|
||||
// - Doc.MediaDigests / Doc.Attachments、MediaAttachment
|
||||
// - TextEvent.Attachments
|
||||
// - DocMemoryAPI.InsertWithMedia
|
||||
// - IOInjector 的 InjectInputMedia / InjectInputMediaSync /
|
||||
// InjectInterruptMedia;PluginSDK 补上缺失的 SetToolBlocks 包装
|
||||
// 同版修掉两处并发竞态(sdk/stress_test.go 的 -race 实证,不是理论风险):
|
||||
// PluginSDK 的 API 字段与 autoRestart 标志此前无锁,而写方
|
||||
// (内核注入 API、插件 SetAutoRestart)与读方(插件后台 goroutine
|
||||
// 注入、内核 registry 读 AutoRestart)天然跨 goroutine。
|
||||
// 存量插件不需要改一行也不需要重编:新增方法由**插件调用、内核实现**,
|
||||
// 不调就不受影响。想用新字段的插件重编即可。
|
||||
//
|
||||
// ❗main 分支上此值是**下一个未发布中版本**(1.1.x 线在发布中,所以 main 是 1.2.0);
|
||||
// 已发布的值看对应的 release/vX.Y.x 分支与 tag(如 release/v1.1.x 上是 1.1.0)。
|
||||
Version = "1.0.0"
|
||||
|
||||
// Commit 是构建时的 Git commit hash。
|
||||
@ -27,6 +48,11 @@ var (
|
||||
//
|
||||
// 1.0.0 是硬下限而非建议值:0.9.x 内核只会 dlopen `.so`,
|
||||
// 本版工具链产出的 `plugin.bin` 在旧内核上根本不会被识别。
|
||||
//
|
||||
// ⚠️ 1.1.0 新增的媒体接口需要核心 **1.1.1+**(更早的核心没有
|
||||
// doc.insertWithMedia / io.injectMedia* 这些 RPC,调用会返回 unknown method)。
|
||||
// 这里仍写 1.0.0,因为它是「SDK 能在其上运行」的下限;
|
||||
// 媒体接口是可选能力,不用就不受影响。
|
||||
CoreVersion = "1.0.0"
|
||||
)
|
||||
|
||||
|
||||
64
third_party/homeagent-sdk/sdk/memory.go
vendored
64
third_party/homeagent-sdk/sdk/memory.go
vendored
@ -25,13 +25,18 @@ type Relation struct {
|
||||
}
|
||||
|
||||
// Triple represents a subject-relation-object triple for the knowledge graph.
|
||||
//
|
||||
// SentenceText 是这条三元组的原句,会写进 sentences 表;媒体引用挂在句子上,
|
||||
// 所以 MediaDigests 非空时内核会保证句子存在(不给就自动合成一句)。
|
||||
type Triple struct {
|
||||
Subject string `json:"subject"`
|
||||
Relation string `json:"relation"`
|
||||
Object string `json:"object"`
|
||||
Confidence float64 `json:"confidence,omitempty"`
|
||||
SubjectType string `json:"subject_type,omitempty"`
|
||||
ObjectType string `json:"object_type,omitempty"`
|
||||
Subject string `json:"subject"`
|
||||
Relation string `json:"relation"`
|
||||
Object string `json:"object"`
|
||||
Confidence float64 `json:"confidence,omitempty"`
|
||||
SubjectType string `json:"subject_type,omitempty"`
|
||||
ObjectType string `json:"object_type,omitempty"`
|
||||
SentenceText string `json:"sentence_text,omitempty"`
|
||||
MediaDigests []string `json:"media_digests,omitempty"`
|
||||
}
|
||||
|
||||
// TextMemoryAPI provides access to chronological text event storage.
|
||||
@ -40,27 +45,52 @@ type TextMemoryAPI interface {
|
||||
}
|
||||
|
||||
// TextEvent represents a single text memory event.
|
||||
// MediaAttachment 描述一份与记忆关联的媒体。
|
||||
//
|
||||
// 两个方向共用一个类型:
|
||||
// - 写入(InsertWithMedia):给 Data + MIME 就是新内容;只给 Digest 则是引用已有内容。
|
||||
// - 读出(Query):内核只填 Digest/MIME/Description,**不回 Data**——
|
||||
// 一次检索可能命中几十张图,把字节全塞回插件会把 ABI 消息撑爆。
|
||||
// 需要字节时拿 Digest 单独取。
|
||||
type MediaAttachment struct {
|
||||
Digest string `json:"digest,omitempty"`
|
||||
MIME string `json:"mime,omitempty"`
|
||||
Data []byte `json:"data,omitempty"`
|
||||
Name string `json:"name,omitempty"`
|
||||
Description string `json:"description,omitempty"`
|
||||
}
|
||||
|
||||
type TextEvent struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
Channel string `json:"channel,omitempty"`
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
Channel string `json:"channel,omitempty"`
|
||||
Attachments []MediaAttachment `json:"attachments,omitempty"`
|
||||
}
|
||||
|
||||
// DocMemoryAPI provides access to the document vector store.
|
||||
type DocMemoryAPI interface {
|
||||
Query(text string, topK int) []*Doc
|
||||
Insert(doc *Doc) error
|
||||
// InsertWithMedia 写入文档并关联媒体。attachments 里带 Data 的会落进
|
||||
// 内容寻址存储(相同字节只存一份),只带 Digest 的直接引用已有内容。
|
||||
// 插件无需自己拼标记:内核会把 `[mime <短 digest>] <描述>` 补进 Content,
|
||||
// 让向量检索和后续蒸馏都能看到这份媒体。
|
||||
InsertWithMedia(doc *Doc, attachments []MediaAttachment) error
|
||||
Remove(id string)
|
||||
Stats() map[string]interface{}
|
||||
}
|
||||
|
||||
// Doc represents a document in the document store.
|
||||
//
|
||||
// MediaDigests / Attachments 在 Query 返回时由内核填充(仅元数据,不带字节)。
|
||||
type Doc struct {
|
||||
ID string `json:"id"`
|
||||
Title string `json:"title"`
|
||||
Content string `json:"content"`
|
||||
Score float64 `json:"score,omitempty"`
|
||||
ID string `json:"id"`
|
||||
Title string `json:"title"`
|
||||
Content string `json:"content"`
|
||||
Score float64 `json:"score,omitempty"`
|
||||
MediaDigests []string `json:"media_digests,omitempty"`
|
||||
Attachments []MediaAttachment `json:"attachments,omitempty"`
|
||||
}
|
||||
|
||||
// SocialAPI provides read-only access to the social graph (person profiles and relationships).
|
||||
@ -75,9 +105,9 @@ type SocialAPI interface {
|
||||
|
||||
// PersonProfile represents a person's complete profile (traits + social relations).
|
||||
type PersonProfile struct {
|
||||
Name string `json:"name"`
|
||||
Traits map[string]string `json:"traits,omitempty"`
|
||||
Relations []SocialRelation `json:"relations,omitempty"`
|
||||
Name string `json:"name"`
|
||||
Traits map[string]string `json:"traits,omitempty"`
|
||||
Relations []SocialRelation `json:"relations,omitempty"`
|
||||
}
|
||||
|
||||
// SocialRelation represents a social relationship between two persons.
|
||||
|
||||
257
third_party/homeagent-sdk/sdk/plugin.go
vendored
257
third_party/homeagent-sdk/sdk/plugin.go
vendored
@ -61,14 +61,18 @@ type StageContext struct {
|
||||
Memory []MemItem
|
||||
NoMemory bool
|
||||
Extra map[string]interface{}
|
||||
Errors []string // 阶段处理过程中的错误信息
|
||||
Errors []string // 阶段处理过程中的错误信息
|
||||
}
|
||||
|
||||
func (c *StageContext) RLock() { c.mu.RLock() }
|
||||
func (c *StageContext) RUnlock() { c.mu.RUnlock() }
|
||||
func (c *StageContext) Lock() { c.mu.Lock() }
|
||||
func (c *StageContext) Unlock() { c.mu.Unlock() }
|
||||
func (c *StageContext) IsResponded() bool { c.mu.RLock(); defer c.mu.RUnlock(); return c.Response != nil }
|
||||
func (c *StageContext) RLock() { c.mu.RLock() }
|
||||
func (c *StageContext) RUnlock() { c.mu.RUnlock() }
|
||||
func (c *StageContext) Lock() { c.mu.Lock() }
|
||||
func (c *StageContext) Unlock() { c.mu.Unlock() }
|
||||
func (c *StageContext) IsResponded() bool {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.Response != nil
|
||||
}
|
||||
|
||||
// MemItem represents a memory item in stage context.
|
||||
type MemItem struct {
|
||||
@ -100,8 +104,8 @@ type ToolDef struct {
|
||||
Plugin string `json:"plugin,omitempty"`
|
||||
Description string `json:"description"`
|
||||
Parameters map[string]interface{} `json:"parameters"`
|
||||
NoMemory bool `json:"no_memory,omitempty"` // 此工具输出不参与记忆计算,但原文保留
|
||||
Cleaner func(string) string `json:"-"` // 计算层过滤函数,不改原文;仅在向量化/jieba/蒸馏时调用
|
||||
NoMemory bool `json:"no_memory,omitempty"` // 此工具输出不参与记忆计算,但原文保留
|
||||
Cleaner func(string) string `json:"-"` // 计算层过滤函数,不改原文;仅在向量化/jieba/蒸馏时调用
|
||||
}
|
||||
|
||||
// IOInjector provides methods for injecting input and interrupts into the agent pipeline.
|
||||
@ -117,6 +121,9 @@ type IOInjector interface {
|
||||
// SetToolBlocks 插件工具注入多模态内容块(image_url/audio_url),内核在下一条
|
||||
// tool message 的 content 数组里带上这些块,让模型在后续轮次看到图/听到音频。
|
||||
SetToolBlocks(blocks []ContentBlock)
|
||||
InjectInputMedia(source, channel, text string, blocks []ContentBlock)
|
||||
InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string
|
||||
InjectInterruptMedia(source, channel, text string, blocks []ContentBlock)
|
||||
}
|
||||
|
||||
// EventType identifies the kind of system event.
|
||||
@ -221,6 +228,24 @@ type PluginSDK struct {
|
||||
events EventSubscriber
|
||||
plgMgr PluginMgrAPI
|
||||
|
||||
// apiMu 保护上面这些由内核注入的 API 字段,以及 autoRestart。
|
||||
//
|
||||
// 这些字段的写方与读方天然跨 goroutine:
|
||||
// - 写方是内核(加载/重载插件时注入 API)与插件自己(SetAutoRestart);
|
||||
// - 读方是插件在 Start() 里起的后台 goroutine(轮询、监听、定时器
|
||||
// 都要拿 injector 往管道里注消息),以及内核 registry —— 它在
|
||||
// 另一个 goroutine 读 AutoRestart() 决定崩溃后是否重启。
|
||||
// SetAutoRestart 的文档用法本身就是「连接建立后再决定能否自动重启」,
|
||||
// 而连接建立通常发生在后台 goroutine 里,于是这对读写必然并发。
|
||||
//
|
||||
// sdk/stress_test.go 的 -race 实测确认这是真竞态,不是理论风险。
|
||||
// 未加锁时的生产表现是偶发 nil 解引用崩溃(读到半个接口值)。
|
||||
//
|
||||
// 约定:只在持锁期间取字段值,取完立刻释放再调用。
|
||||
// 持锁调用会把 InjectInputSync 这类阻塞到 agent 回复(可达数分钟)的
|
||||
// 方法与 SetIOInjector 串到一起,让插件重载卡死。
|
||||
apiMu sync.RWMutex
|
||||
|
||||
autoRestart bool
|
||||
|
||||
stopMu sync.Mutex
|
||||
@ -247,28 +272,57 @@ func New(name string, sett SettingsAPI, regTool ToolRegistrar, regStage StageReg
|
||||
func (s *PluginSDK) PluginName() string { return s.name }
|
||||
|
||||
// Settings returns the settings API for reading/writing plugin configuration.
|
||||
// sett 在 New 时一次性写入且无 setter,故不需要加锁。
|
||||
func (s *PluginSDK) Settings() SettingsAPI { return s.sett }
|
||||
|
||||
// Memory returns the graph memory API (may be nil if not available).
|
||||
func (s *PluginSDK) Memory() MemoryAPI { return s.mem }
|
||||
func (s *PluginSDK) Memory() MemoryAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.mem
|
||||
}
|
||||
|
||||
// TextMemory returns the text memory API (may be nil if not available).
|
||||
func (s *PluginSDK) TextMemory() TextMemoryAPI { return s.textMem }
|
||||
func (s *PluginSDK) TextMemory() TextMemoryAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.textMem
|
||||
}
|
||||
|
||||
// DocMemory returns the document memory API (may be nil if not available).
|
||||
func (s *PluginSDK) DocMemory() DocMemoryAPI { return s.docMem }
|
||||
func (s *PluginSDK) DocMemory() DocMemoryAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.docMem
|
||||
}
|
||||
|
||||
// Knowledge returns the knowledge store API (may be nil if not available).
|
||||
func (s *PluginSDK) Knowledge() KnowledgeAPI { return s.know }
|
||||
func (s *PluginSDK) Knowledge() KnowledgeAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.know
|
||||
}
|
||||
|
||||
// LLM returns the LLM provider API (may be nil if not available).
|
||||
func (s *PluginSDK) LLM() LLMAPI { return s.llm }
|
||||
func (s *PluginSDK) LLM() LLMAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.llm
|
||||
}
|
||||
|
||||
// Social returns the social graph API (may be nil if not available).
|
||||
func (s *PluginSDK) Social() SocialAPI { return s.social }
|
||||
func (s *PluginSDK) Social() SocialAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.social
|
||||
}
|
||||
|
||||
// Events returns the event subscriber for listening to kernel events (may be nil if not available).
|
||||
func (s *PluginSDK) Events() EventSubscriber { return s.events }
|
||||
func (s *PluginSDK) Events() EventSubscriber {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.events
|
||||
}
|
||||
|
||||
// RegisterTool registers a tool that the LLM can call.
|
||||
func (s *PluginSDK) RegisterTool(name string, def ToolDef, handler ToolHandler) error {
|
||||
@ -282,8 +336,9 @@ func (s *PluginSDK) RegisterTool(name string, def ToolDef, handler ToolHandler)
|
||||
}
|
||||
|
||||
// RegisterStage registers a handler for a pipeline stage.
|
||||
// scope: StageScopeGlobal (default) — receives all stage events.
|
||||
// StageScopeOwnTools — only before_toolcall/after_toolcall for this plugin's tools.
|
||||
//
|
||||
// scope: StageScopeGlobal (default) — receives all stage events.
|
||||
// StageScopeOwnTools — only before_toolcall/after_toolcall for this plugin's tools.
|
||||
func (s *PluginSDK) RegisterStage(stage Stage, handler StageHandler, scope ...StageScope) {
|
||||
if s.regStage == nil {
|
||||
return
|
||||
@ -333,8 +388,11 @@ func (s *PluginSDK) RegisterPluginAPI(name string) error {
|
||||
// def: 通道在记忆计算层的行为(NoMemory/Cleaner)
|
||||
// handler: receives args map with keys: payload (string), type (string), meta (string|optional)
|
||||
func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, def ChannelDef, handler ToolHandler) error {
|
||||
if s.regOutput != nil {
|
||||
return s.regOutput(name, caps, desc, def, handler)
|
||||
s.apiMu.RLock()
|
||||
reg := s.regOutput
|
||||
s.apiMu.RUnlock()
|
||||
if reg != nil {
|
||||
return reg(name, caps, desc, def, handler)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -343,57 +401,126 @@ func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, de
|
||||
// def.NoMemory: 此通道输入不参与记忆计算
|
||||
// def.Cleaner: 计算层对输入文本清洗后(不改原文)再向量化/提关键词
|
||||
func (s *PluginSDK) RegisterInputChannel(name string, def ChannelDef) error {
|
||||
if s.regInput != nil {
|
||||
return s.regInput(name, def)
|
||||
s.apiMu.RLock()
|
||||
reg := s.regInput
|
||||
s.apiMu.RUnlock()
|
||||
if reg != nil {
|
||||
return reg(name, def)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 以下 setter 由内核在启动/重载时调用,与插件后台 goroutine 的读并发,故加锁。
|
||||
|
||||
// SetOutputChannelRegistrar sets the output channel registrar (called by the core at startup).
|
||||
func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) { s.regOutput = r }
|
||||
func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) {
|
||||
s.apiMu.Lock()
|
||||
s.regOutput = r
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// SetInputChannelRegistrar sets the input channel registrar (called by the core at startup).
|
||||
func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) { s.regInput = r }
|
||||
func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) {
|
||||
s.apiMu.Lock()
|
||||
s.regInput = r
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// SetIOInjector sets the IO injector (called by the core at startup).
|
||||
func (s *PluginSDK) SetIOInjector(io IOInjector) { s.io = io }
|
||||
func (s *PluginSDK) SetIOInjector(io IOInjector) {
|
||||
s.apiMu.Lock()
|
||||
s.io = io
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// SetMemoryAPI sets the memory API (called by the core at startup).
|
||||
func (s *PluginSDK) SetMemoryAPI(mem MemoryAPI) { s.mem = mem }
|
||||
func (s *PluginSDK) SetTextMemoryAPI(tm TextMemoryAPI) { s.textMem = tm }
|
||||
func (s *PluginSDK) SetDocMemoryAPI(dm DocMemoryAPI) { s.docMem = dm }
|
||||
func (s *PluginSDK) SetKnowledgeAPI(kn KnowledgeAPI) { s.know = kn }
|
||||
func (s *PluginSDK) SetLLMAPI(llm LLMAPI) { s.llm = llm }
|
||||
func (s *PluginSDK) SetSocialAPI(social SocialAPI) { s.social = social }
|
||||
func (s *PluginSDK) SetEventSubscriber(es EventSubscriber) { s.events = es }
|
||||
func (s *PluginSDK) SetMemoryAPI(mem MemoryAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.mem = mem
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetTextMemoryAPI(tm TextMemoryAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.textMem = tm
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetDocMemoryAPI(dm DocMemoryAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.docMem = dm
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetKnowledgeAPI(kn KnowledgeAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.know = kn
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetLLMAPI(llm LLMAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.llm = llm
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetSocialAPI(social SocialAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.social = social
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *PluginSDK) SetEventSubscriber(es EventSubscriber) {
|
||||
s.apiMu.Lock()
|
||||
s.events = es
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// SetPluginMgrAPI sets the plugin manager API (called by the bridge at startup).
|
||||
func (s *PluginSDK) SetPluginMgrAPI(pm PluginMgrAPI) { s.plgMgr = pm }
|
||||
func (s *PluginSDK) SetPluginMgrAPI(pm PluginMgrAPI) {
|
||||
s.apiMu.Lock()
|
||||
s.plgMgr = pm
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// PluginMgr returns the plugin manager API (ReloadOne / ReloadPlugins / list).
|
||||
// May be nil if the host did not wire it.
|
||||
func (s *PluginSDK) PluginMgr() PluginMgrAPI { return s.plgMgr }
|
||||
func (s *PluginSDK) PluginMgr() PluginMgrAPI {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.plgMgr
|
||||
}
|
||||
|
||||
// ---- IO Convenience Methods ----
|
||||
|
||||
// injector 取当前 injector 的快照。
|
||||
//
|
||||
// 取完即释放锁再调用:InjectInputSync 会阻塞到 agent 回复(可达数分钟),
|
||||
// 若持锁调用,插件重载时的 SetIOInjector 会一起卡住。
|
||||
func (s *PluginSDK) injector() IOInjector {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.io
|
||||
}
|
||||
|
||||
// InjectInterruptText injects a text interrupt that can preempt current LLM processing.
|
||||
func (s *PluginSDK) InjectInterruptText(source, channel, text string) {
|
||||
if s.io != nil {
|
||||
s.io.InjectInterruptText(source, channel, text)
|
||||
if io := s.injector(); io != nil {
|
||||
io.InjectInterruptText(source, channel, text)
|
||||
}
|
||||
}
|
||||
|
||||
// InjectText injects a text message into the agent pipeline.
|
||||
func (s *PluginSDK) InjectText(source, channel, text string) {
|
||||
if s.io != nil {
|
||||
s.io.InjectText(source, channel, text)
|
||||
if io := s.injector(); io != nil {
|
||||
io.InjectText(source, channel, text)
|
||||
}
|
||||
}
|
||||
|
||||
// InjectTextNoMemory injects a text message without generating memory.
|
||||
func (s *PluginSDK) InjectTextNoMemory(source, channel, text string) {
|
||||
if s.io != nil {
|
||||
s.io.InjectTextNoMemory(source, channel, text)
|
||||
if io := s.injector(); io != nil {
|
||||
io.InjectTextNoMemory(source, channel, text)
|
||||
}
|
||||
}
|
||||
|
||||
@ -401,18 +528,62 @@ func (s *PluginSDK) InjectTextNoMemory(source, channel, text string) {
|
||||
// returning the reply text (empty string if none). Replies must be dispatched back
|
||||
// to the source channel by the caller.
|
||||
func (s *PluginSDK) InjectInputSync(source, channel, text string) string {
|
||||
if s.io == nil {
|
||||
io := s.injector()
|
||||
if io == nil {
|
||||
return ""
|
||||
}
|
||||
return s.io.InjectInputSync(source, channel, text)
|
||||
return io.InjectInputSync(source, channel, text)
|
||||
}
|
||||
|
||||
// InjectInputMedia 注入带媒体内容块(image_url/audio_url)的输入。
|
||||
// blocks 会落进媒体存储被记忆引用捕获,同时作为当前轮 content 数组
|
||||
// 发给 LLM,让模型在「本轮」就看到图/听到音频——区别于 SetToolBlocks
|
||||
// 的「下一轮 tool message」语义。
|
||||
func (s *PluginSDK) InjectInputMedia(source, channel, text string, blocks []ContentBlock) {
|
||||
if io := s.injector(); io != nil {
|
||||
io.InjectInputMedia(source, channel, text, blocks)
|
||||
}
|
||||
}
|
||||
|
||||
// InjectInputMediaSync 注入带媒体内容块的输入并同步等待 agent 回复。
|
||||
func (s *PluginSDK) InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string {
|
||||
io := s.injector()
|
||||
if io == nil {
|
||||
return ""
|
||||
}
|
||||
return io.InjectInputMediaSync(source, channel, text, blocks)
|
||||
}
|
||||
|
||||
// InjectInterruptMedia 注入带媒体内容块的中断,可抢占当前 LLM 处理。
|
||||
// blocks 随中断消息一起发给模型。
|
||||
func (s *PluginSDK) InjectInterruptMedia(source, channel, text string, blocks []ContentBlock) {
|
||||
if io := s.injector(); io != nil {
|
||||
io.InjectInterruptMedia(source, channel, text, blocks)
|
||||
}
|
||||
}
|
||||
|
||||
// SetToolBlocks 在工具处理函数内注入多模态内容块,内核在下一条 tool message
|
||||
// 的 content 数组里带上它们。需要「本轮就让模型看到」时用 InjectInputMedia。
|
||||
func (s *PluginSDK) SetToolBlocks(blocks []ContentBlock) {
|
||||
if io := s.injector(); io != nil {
|
||||
io.SetToolBlocks(blocks)
|
||||
}
|
||||
}
|
||||
|
||||
// SetAutoRestart 设置插件是否允许内核自动重启(崩溃后自动重载)。
|
||||
// 默认 true。如果插件有无法恢复的状态(如外部连接),应设为 false。
|
||||
func (s *PluginSDK) SetAutoRestart(enabled bool) { s.autoRestart = enabled }
|
||||
func (s *PluginSDK) SetAutoRestart(enabled bool) {
|
||||
s.apiMu.Lock()
|
||||
s.autoRestart = enabled
|
||||
s.apiMu.Unlock()
|
||||
}
|
||||
|
||||
// AutoRestart 返回插件是否允许自动重启。
|
||||
func (s *PluginSDK) AutoRestart() bool { return s.autoRestart }
|
||||
func (s *PluginSDK) AutoRestart() bool {
|
||||
s.apiMu.RLock()
|
||||
defer s.apiMu.RUnlock()
|
||||
return s.autoRestart
|
||||
}
|
||||
|
||||
// RegisterStopHandler 注册插件停止阶段的清理回调。
|
||||
// 注册的 handler 会在插件 Stop() 之前按"后注册先执行"的顺序调用,
|
||||
|
||||
725
third_party/homeagent-sdk/sdk/stress_test.go
vendored
Normal file
725
third_party/homeagent-sdk/sdk/stress_test.go
vendored
Normal file
@ -0,0 +1,725 @@
|
||||
package sdk
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// SDK 公开接口的并发压力测试(1.1.0 媒体接口上线后新增)。
|
||||
//
|
||||
// 为什么这一层需要压测:SDK 是**被多个 goroutine 同时使用的共享对象**。
|
||||
// 一个插件的典型形态是 Start() 里起若干后台 goroutine(轮询、监听、定时器),
|
||||
// 它们各自持同一个 *PluginSDK 往里注入消息;内核侧同时还有 stage 扇出、
|
||||
// 工具调用、以及读 AutoRestart() 决定崩溃后是否重启。
|
||||
// 单线程单测全绿不代表这些并发路径成立。
|
||||
//
|
||||
// 关注点不是吞吐数字,而是不变量:
|
||||
// 1. 注入调用不丢、不串(媒体块必须与文本配对,不能张冠李戴)
|
||||
// 2. 状态字段的读写不产生数据竞争(-race 下必须干净)
|
||||
// 3. handler 注册/执行在并发下"恰好一次"
|
||||
// 4. 跨进程 JSON 序列化对新媒体类型必须字节级往返一致
|
||||
//
|
||||
// 媒体接口尤其需要 3 与 4:媒体块要经 JSON 过子进程边界,
|
||||
// 而 []byte 在 JSON 里是 base64,往返不一致的后果是图片静默损坏。
|
||||
|
||||
// ---------- 测试替身 ----------
|
||||
|
||||
// recordingInjector 记录每一次注入调用,用于验证"不丢不串"。
|
||||
type recordingInjector struct {
|
||||
mu sync.Mutex
|
||||
calls []injectCall
|
||||
|
||||
// 计数用原子量:并发路径上只增不减,可在不持锁时安全读。
|
||||
nText, nMedia, nInterrupt, nSync atomic.Int64
|
||||
}
|
||||
|
||||
type injectCall struct {
|
||||
kind string // text / media / interruptMedia / sync ...
|
||||
source string
|
||||
channel string
|
||||
text string
|
||||
blocks []ContentBlock
|
||||
}
|
||||
|
||||
func (r *recordingInjector) record(c injectCall) {
|
||||
r.mu.Lock()
|
||||
r.calls = append(r.calls, c)
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectInterruptText(s, c, t string) {
|
||||
r.nInterrupt.Add(1)
|
||||
r.record(injectCall{kind: "interruptText", source: s, channel: c, text: t})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectText(s, c, t string) {
|
||||
r.nText.Add(1)
|
||||
r.record(injectCall{kind: "text", source: s, channel: c, text: t})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectTextNoMemory(s, c, t string) {
|
||||
r.nText.Add(1)
|
||||
r.record(injectCall{kind: "textNoMem", source: s, channel: c, text: t})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectInputSync(s, c, t string) string {
|
||||
r.nSync.Add(1)
|
||||
r.record(injectCall{kind: "sync", source: s, channel: c, text: t})
|
||||
return "reply:" + t
|
||||
}
|
||||
|
||||
func (r *recordingInjector) SetToolBlocks(blocks []ContentBlock) {
|
||||
r.record(injectCall{kind: "toolBlocks", blocks: blocks})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectInputMedia(s, c, t string, b []ContentBlock) {
|
||||
r.nMedia.Add(1)
|
||||
r.record(injectCall{kind: "media", source: s, channel: c, text: t, blocks: b})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectInputMediaSync(s, c, t string, b []ContentBlock) string {
|
||||
r.nMedia.Add(1)
|
||||
r.nSync.Add(1)
|
||||
r.record(injectCall{kind: "mediaSync", source: s, channel: c, text: t, blocks: b})
|
||||
return "reply:" + t
|
||||
}
|
||||
|
||||
func (r *recordingInjector) InjectInterruptMedia(s, c, t string, b []ContentBlock) {
|
||||
r.nMedia.Add(1)
|
||||
r.record(injectCall{kind: "interruptMedia", source: s, channel: c, text: t, blocks: b})
|
||||
}
|
||||
|
||||
func (r *recordingInjector) snapshot() []injectCall {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]injectCall{}, r.calls...)
|
||||
}
|
||||
|
||||
var _ IOInjector = (*recordingInjector)(nil)
|
||||
|
||||
// imageBlock 构造一个带可识别 URL 的图片块。
|
||||
func imageBlock(tag string) ContentBlock {
|
||||
return ContentBlock{
|
||||
Type: "image_url",
|
||||
ImageURL: &ImageURL{URL: "data:image/png;base64," + tag, Detail: "auto"},
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 1. 媒体注入并发不丢不串 ----------
|
||||
|
||||
// 三个媒体注入方法在高并发下必须:调用数精确、且每次调用的 text 与 blocks 配对不错。
|
||||
//
|
||||
// "不串"是这里的关键断言。注入是插件里最容易被后台 goroutine 并发调用的入口,
|
||||
// 若实现里出现任何共享中间状态(比如把 blocks 暂存到 SDK 字段再读出),
|
||||
// 高并发下就会出现 A 的文本配上 B 的图——而两者单独看都"成功"了,不报错。
|
||||
func TestStress_MediaInjectionConcurrentNoCrossTalk(t *testing.T) {
|
||||
const workers, perWorker = 32, 200
|
||||
|
||||
inj := &recordingInjector{}
|
||||
s := &PluginSDK{name: "stress"}
|
||||
s.SetIOInjector(inj)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < perWorker; i++ {
|
||||
// tag 唯一标识这次调用,文本与图片 URL 里都带上它。
|
||||
tag := fmt.Sprintf("w%d-i%d", w, i)
|
||||
switch i % 3 {
|
||||
case 0:
|
||||
s.InjectInputMedia("src", "ch", tag, []ContentBlock{imageBlock(tag)})
|
||||
case 1:
|
||||
if got := s.InjectInputMediaSync("src", "ch", tag, []ContentBlock{imageBlock(tag)}); got != "reply:"+tag {
|
||||
t.Errorf("同步注入回复错位: got %q want %q", got, "reply:"+tag)
|
||||
}
|
||||
default:
|
||||
s.InjectInterruptMedia("src", "ch", tag, []ContentBlock{imageBlock(tag)})
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
total := int64(workers * perWorker)
|
||||
if got := inj.nMedia.Load(); got != total {
|
||||
t.Fatalf("媒体注入调用数 = %d,期望 %d(有调用丢失)", got, total)
|
||||
}
|
||||
|
||||
// 逐条校验文本与媒体块配对:URL 必须含该次调用自己的 tag。
|
||||
seen := map[string]bool{}
|
||||
for _, c := range inj.snapshot() {
|
||||
if len(c.blocks) == 0 {
|
||||
continue
|
||||
}
|
||||
if c.blocks[0].ImageURL == nil {
|
||||
t.Fatalf("媒体块 ImageURL 丢失: %+v", c.blocks[0])
|
||||
}
|
||||
if !strings.HasSuffix(c.blocks[0].ImageURL.URL, c.text) {
|
||||
t.Fatalf("文本与媒体块错位: text=%q url=%q", c.text, c.blocks[0].ImageURL.URL)
|
||||
}
|
||||
if seen[c.text] {
|
||||
t.Fatalf("同一次调用被记录两次: %s", c.text)
|
||||
}
|
||||
seen[c.text] = true
|
||||
}
|
||||
if len(seen) != int(total) {
|
||||
t.Fatalf("去重后调用数 = %d,期望 %d", len(seen), total)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 2. 注入期间热替换 injector ----------
|
||||
|
||||
// 内核在插件运行期间可能重新注入 API(重载、恢复、子进程重连握手)。
|
||||
// 此时插件的后台 goroutine 仍在注入。这条路径若无同步就是对 s.io 的数据竞争,
|
||||
// 在 -race 下会被抓出;生产表现是偶发 nil 解引用崩溃。
|
||||
func TestStress_InjectorSwapDuringInjection(t *testing.T) {
|
||||
s := &PluginSDK{name: "stress"}
|
||||
s.SetIOInjector(&recordingInjector{})
|
||||
|
||||
stop := make(chan struct{})
|
||||
var injectors, swapper sync.WaitGroup
|
||||
|
||||
// 注入方:持续打直到 stop
|
||||
for w := 0; w < 8; w++ {
|
||||
injectors.Add(1)
|
||||
go func() {
|
||||
defer injectors.Done()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
s.InjectInputMedia("src", "ch", "x", []ContentBlock{imageBlock("x")})
|
||||
s.InjectText("src", "ch", "y")
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// 替换方:反复换 injector(含换成 nil——内核卸载 API 时的真实状态)
|
||||
swapper.Add(1)
|
||||
go func() {
|
||||
defer swapper.Done()
|
||||
for i := 0; i < 500; i++ {
|
||||
if i%7 == 0 {
|
||||
s.SetIOInjector(nil)
|
||||
} else {
|
||||
s.SetIOInjector(&recordingInjector{})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// 先等替换跑完,再告知注入方退出。
|
||||
// 顺序写反了就是死锁:注入方只依 close(stop) 退出。
|
||||
swapper.Wait()
|
||||
close(stop)
|
||||
injectors.Wait()
|
||||
// 断言就是「没崩、-race 没报」。nil injector 时必须静默跳过而非 panic。
|
||||
}
|
||||
|
||||
// ---------- 3. autoRestart 标志的并发读写 ----------
|
||||
|
||||
// SetAutoRestart 的文档用途是"插件有无法恢复的状态(如外部连接)时设为 false"——
|
||||
// 而连接建立本身通常是异步的,所以这个写入天然发生在后台 goroutine。
|
||||
// 内核侧 registry 在另一个 goroutine 读 AutoRestart() 决定崩溃后是否重启。
|
||||
// 这是一对跨 goroutine 的读写,必须同步。
|
||||
func TestStress_AutoRestartFlagConcurrent(t *testing.T) {
|
||||
s := &PluginSDK{name: "stress", autoRestart: true}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < 16; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 500; i++ {
|
||||
s.SetAutoRestart(i%2 == 0)
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
// 读方模拟内核 registry
|
||||
for r := 0; r < 8; r++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 500; i++ {
|
||||
_ = s.AutoRestart()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// ---------- 4. stop / onRemove handler 的"恰好一次" ----------
|
||||
|
||||
// RunStopHandlers 的契约是"执行后清空,幂等"。内核在停止插件时可能并发触发
|
||||
// (超时强杀与正常 Stop 竞争),handler 里往往是关连接、落盘——
|
||||
// 执行两次的后果从"重复写文件"到"close 已关闭的 channel 直接 panic"。
|
||||
func TestStress_StopHandlersExactlyOnce(t *testing.T) {
|
||||
const n = 300
|
||||
s := &PluginSDK{name: "stress"}
|
||||
|
||||
var counters [n]atomic.Int64
|
||||
for i := 0; i < n; i++ {
|
||||
i := i
|
||||
s.RegisterStopHandler(func() { counters[i].Add(1) })
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < 16; w++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
s.RunStopHandlers()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
if got := counters[i].Load(); got != 1 {
|
||||
t.Fatalf("handler %d 执行 %d 次,期望恰好 1 次", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 注册与执行并发:已注册的 handler 一次都不能多跑,未跑到的也不能被丢。
|
||||
// 断言用"每个 handler 的执行次数 <= 1"而非"总数相等"——
|
||||
// 与 RunStopHandlers 竞争的注册可能落在快照之后,那属于合法的未执行。
|
||||
func TestStress_StopHandlersRegisterWhileRunning(t *testing.T) {
|
||||
s := &PluginSDK{name: "stress"}
|
||||
const n = 500
|
||||
var counters [n]atomic.Int64
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < n; i++ {
|
||||
i := i
|
||||
s.RegisterStopHandler(func() { counters[i].Add(1) })
|
||||
}
|
||||
}()
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 50; i++ {
|
||||
s.RunStopHandlers()
|
||||
}
|
||||
}()
|
||||
wg.Wait()
|
||||
s.RunStopHandlers() // 收尾:把剩下的都跑掉
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
if got := counters[i].Load(); got > 1 {
|
||||
t.Fatalf("handler %d 被执行 %d 次(重复执行)", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStress_OnRemoveHandlersExactlyOnce(t *testing.T) {
|
||||
const n = 200
|
||||
s := &PluginSDK{name: "stress"}
|
||||
|
||||
var counters [n]atomic.Int64
|
||||
for i := 0; i < n; i++ {
|
||||
i := i
|
||||
s.RegisterOnRemoveHandler(func() { counters[i].Add(1) })
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < 12; w++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
s.RunOnRemoveHandlers()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
if got := counters[i].Load(); got != 1 {
|
||||
t.Fatalf("onRemove handler %d 执行 %d 次,期望恰好 1 次", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 5. StageContext 并发读改写 ----------
|
||||
|
||||
// StageContext 是全部 stage handler 共享的可变状态,字段全导出、靠调用方自觉
|
||||
// 持 Lock/RLock。媒体链路让 Extra 成为新热点(media_blocks 挂在这里),
|
||||
// 而 map 的并发写在 Go 里是直接 fatal,recover 都接不住。
|
||||
//
|
||||
// 这条测试锁定的不变量:按约定持锁的并发读改写不丢更新、不 fatal。
|
||||
func TestStress_StageContextConcurrentExtraAndFinalText(t *testing.T) {
|
||||
ctx := &StageContext{Extra: map[string]interface{}{}}
|
||||
|
||||
const workers, rounds = 16, 200
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < rounds; i++ {
|
||||
// 写:模拟插件往 Extra 塞媒体块并追加文本(读-改-写)
|
||||
ctx.Lock()
|
||||
ctx.Extra[fmt.Sprintf("k%d-%d", w, i)] = []ContentBlock{imageBlock("x")}
|
||||
ctx.FinalText += "."
|
||||
ctx.Unlock()
|
||||
|
||||
// 读:模拟另一个 handler 检查是否已被响应
|
||||
_ = ctx.IsResponded()
|
||||
ctx.RLock()
|
||||
_ = len(ctx.Extra)
|
||||
ctx.RUnlock()
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
ctx.RLock()
|
||||
defer ctx.RUnlock()
|
||||
if len(ctx.Extra) != workers*rounds {
|
||||
t.Fatalf("Extra 键数 = %d,期望 %d(出现 lost update)", len(ctx.Extra), workers*rounds)
|
||||
}
|
||||
if len(ctx.FinalText) != workers*rounds {
|
||||
t.Fatalf("FinalText 长度 = %d,期望 %d(出现 lost update)", len(ctx.FinalText), workers*rounds)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 6. OwnTools scope 包装器的并发正确性 ----------
|
||||
|
||||
// StageScopeOwnTools 的包装闭环里要读 ctx.ToolCalls 判断归属。
|
||||
// 并发下若判断与执行之间状态被改写,就会出现"别人的工具触发了我的 handler"——
|
||||
// 后果是插件对不属于自己的工具结果动手,且没有任何错误。
|
||||
func TestStress_OwnToolsScopeNoCrossPluginLeak(t *testing.T) {
|
||||
var registered StageHandler
|
||||
s := &PluginSDK{
|
||||
name: "mine",
|
||||
regStage: func(stage Stage, h StageHandler) { registered = h },
|
||||
}
|
||||
|
||||
var fired atomic.Int64
|
||||
s.RegisterStage(StageBeforeToolcall, func(ctx *StageContext) error {
|
||||
fired.Add(1)
|
||||
ctx.RLock()
|
||||
defer ctx.RUnlock()
|
||||
// 触发了就必须确实是自己的工具
|
||||
if len(ctx.ToolCalls) == 0 || ctx.ToolCalls[0].Plugin != "mine" {
|
||||
t.Errorf("handler 被别的插件的工具触发: %+v", ctx.ToolCalls)
|
||||
}
|
||||
return nil
|
||||
}, StageScopeOwnTools)
|
||||
|
||||
if registered == nil {
|
||||
t.Fatal("handler 未注册")
|
||||
}
|
||||
|
||||
const workers, rounds = 16, 100
|
||||
var wg sync.WaitGroup
|
||||
var mineCount atomic.Int64
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < rounds; i++ {
|
||||
// 每个 goroutine 用自己的 ctx——真实内核里 stage 扇出共享同一个
|
||||
// ctx,但那部分的并发由内核 host 仲裁;这里验证包装器本身。
|
||||
owner := "other"
|
||||
if (w+i)%2 == 0 {
|
||||
owner = "mine"
|
||||
mineCount.Add(1)
|
||||
}
|
||||
ctx := &StageContext{Extra: map[string]interface{}{}}
|
||||
ctx.ToolCalls = []ToolCall{{Plugin: owner, Name: "t"}}
|
||||
if err := registered(ctx); err != nil {
|
||||
t.Errorf("handler 返回错误: %v", err)
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if got, want := fired.Load(), mineCount.Load(); got != want {
|
||||
t.Fatalf("handler 触发 %d 次,期望 %d 次(漏触发或跨插件触发)", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 7. 媒体类型的 JSON 往返(跨进程边界的真实形态) ----------
|
||||
|
||||
// 媒体块与附件要经 JSON 过子进程边界。[]byte 在 JSON 里是 base64,
|
||||
// 往返不一致的后果是图片字节静默损坏——落进 CAS 后 digest 校验才会发现,
|
||||
// 而那时已经无从追查是谁改坏的。
|
||||
func TestStress_MediaTypesJSONRoundTripAtScale(t *testing.T) {
|
||||
// 覆盖真实会遇到的边界:空、单字节、含 0x00、全 0xFF、超过 base64 分组边界的长度
|
||||
sizes := []int{0, 1, 2, 3, 255, 256, 1023, 4096, 65537}
|
||||
for _, n := range sizes {
|
||||
data := make([]byte, n)
|
||||
for i := range data {
|
||||
data[i] = byte(i * 7 % 256)
|
||||
}
|
||||
att := MediaAttachment{
|
||||
Digest: strings.Repeat("a", 64),
|
||||
MIME: "image/png",
|
||||
Data: data,
|
||||
Name: "图片-名字 with space & 符号.png",
|
||||
Description: "一张紫蓝红三色带图,含 emoji 🎨 与换行\n第二行",
|
||||
}
|
||||
b, err := json.Marshal(att)
|
||||
if err != nil {
|
||||
t.Fatalf("size=%d marshal: %v", n, err)
|
||||
}
|
||||
var back MediaAttachment
|
||||
if err := json.Unmarshal(b, &back); err != nil {
|
||||
t.Fatalf("size=%d unmarshal: %v", n, err)
|
||||
}
|
||||
if len(back.Data) != n {
|
||||
t.Fatalf("size=%d 往返后长度 = %d", n, len(back.Data))
|
||||
}
|
||||
for i := range data {
|
||||
if back.Data[i] != data[i] {
|
||||
t.Fatalf("size=%d 第 %d 字节损坏: %02x != %02x", n, i, back.Data[i], data[i])
|
||||
}
|
||||
}
|
||||
if back.Name != att.Name || back.Description != att.Description || back.MIME != att.MIME || back.Digest != att.Digest {
|
||||
t.Fatalf("size=%d 元数据往返不一致: %+v", n, back)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// omitempty 必须真的生效:读路径上内核不回 Data,若序列化仍产出 "data":null
|
||||
// 之类的键,跨进程消息会凭空变大,且插件侧无法区分"没有字节"与"空字节"。
|
||||
func TestStress_MediaTypesOmitEmpty(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
v interface{}
|
||||
absent []string
|
||||
present []string
|
||||
}{
|
||||
{
|
||||
name: "Triple 无媒体",
|
||||
v: Triple{Subject: "甲方", Relation: "签署", Object: "合同"},
|
||||
absent: []string{"media_digests", "sentence_text", "confidence", "subject_type", "object_type"},
|
||||
present: []string{"subject", "relation", "object"},
|
||||
},
|
||||
{
|
||||
name: "Triple 带媒体",
|
||||
v: Triple{Subject: "甲方", Relation: "包含", Object: "图", MediaDigests: []string{"abc12345"}, SentenceText: "句子"},
|
||||
absent: []string{"confidence"},
|
||||
present: []string{"media_digests", "sentence_text"},
|
||||
},
|
||||
{
|
||||
name: "Doc 读路径无字节",
|
||||
v: Doc{ID: "d1", Title: "标题", Content: "正文", Attachments: []MediaAttachment{{Digest: "abc12345", MIME: "image/png"}}},
|
||||
absent: []string{"\"data\"", "media_digests", "score"},
|
||||
present: []string{"attachments", "digest", "mime"},
|
||||
},
|
||||
{
|
||||
name: "TextEvent 无附件",
|
||||
v: TextEvent{Role: "user", Content: "hi"},
|
||||
absent: []string{"attachments", "channel"},
|
||||
present: []string{"role", "content"},
|
||||
},
|
||||
{
|
||||
name: "ContentBlock 纯文本",
|
||||
v: ContentBlock{Type: "text", Text: "hi"},
|
||||
absent: []string{"image_url", "audio_url"},
|
||||
present: []string{"type", "text"},
|
||||
},
|
||||
{
|
||||
name: "ContentBlock 图片",
|
||||
v: imageBlock("AAA"),
|
||||
absent: []string{"audio_url", "\"text\""},
|
||||
present: []string{"image_url", "detail"},
|
||||
},
|
||||
}
|
||||
for _, c := range cases {
|
||||
b, err := json.Marshal(c.v)
|
||||
if err != nil {
|
||||
t.Fatalf("%s marshal: %v", c.name, err)
|
||||
}
|
||||
s := string(b)
|
||||
for _, k := range c.absent {
|
||||
if strings.Contains(s, k) {
|
||||
t.Errorf("%s: 不该出现的键 %s —— %s", c.name, k, s)
|
||||
}
|
||||
}
|
||||
for _, k := range c.present {
|
||||
if !strings.Contains(s, k) {
|
||||
t.Errorf("%s: 缺少键 %s —— %s", c.name, k, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 媒体块在并发序列化下必须各自独立:ImageURL/AudioURL 是指针,
|
||||
// 若某处复用同一个指针再改写,序列化结果会互相污染。
|
||||
func TestStress_ContentBlockConcurrentMarshal(t *testing.T) {
|
||||
const workers, rounds = 16, 300
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < rounds; i++ {
|
||||
tag := fmt.Sprintf("w%d-i%d", w, i)
|
||||
blocks := []ContentBlock{
|
||||
{Type: "text", Text: tag},
|
||||
imageBlock(tag),
|
||||
{Type: "audio_url", AudioURL: &AudioURL{URL: "data:audio/wav;base64," + tag}},
|
||||
}
|
||||
b, err := json.Marshal(blocks)
|
||||
if err != nil {
|
||||
t.Errorf("marshal: %v", err)
|
||||
return
|
||||
}
|
||||
var back []ContentBlock
|
||||
if err := json.Unmarshal(b, &back); err != nil {
|
||||
t.Errorf("unmarshal: %v", err)
|
||||
return
|
||||
}
|
||||
if len(back) != 3 {
|
||||
t.Errorf("块数 = %d", len(back))
|
||||
return
|
||||
}
|
||||
if back[0].ImageURL != nil || back[0].AudioURL != nil {
|
||||
t.Errorf("文本块被填了媒体指针: %+v", back[0])
|
||||
}
|
||||
if back[1].ImageURL == nil || !strings.HasSuffix(back[1].ImageURL.URL, tag) {
|
||||
t.Errorf("图片块 URL 错位: %+v", back[1].ImageURL)
|
||||
}
|
||||
if back[1].AudioURL != nil {
|
||||
t.Errorf("图片块被填了音频指针")
|
||||
}
|
||||
if back[2].AudioURL == nil || !strings.HasSuffix(back[2].AudioURL.URL, tag) {
|
||||
t.Errorf("音频块 URL 错位: %+v", back[2].AudioURL)
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// ---------- 8. 注册面的并发 ----------
|
||||
|
||||
// 插件在 Start() 里起多个 goroutine 分别注册工具是常见写法。
|
||||
// def.Plugin 的默认填充若不是每次调用独立的,就会出现工具归属错乱——
|
||||
// 表现是 OwnTools scope 失效、WebUI 里工具挂在别的插件名下。
|
||||
func TestStress_RegisterToolConcurrentPluginDefaulting(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
got := map[string]string{} // toolName -> def.Plugin
|
||||
|
||||
s := &PluginSDK{
|
||||
name: "mine",
|
||||
regTool: func(name string, def ToolDef, h ToolHandler) error {
|
||||
mu.Lock()
|
||||
got[name] = def.Plugin
|
||||
mu.Unlock()
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
const workers, perWorker = 16, 100
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < perWorker; i++ {
|
||||
name := fmt.Sprintf("tool_w%d_i%d", w, i)
|
||||
def := ToolDef{Description: "d", Parameters: map[string]interface{}{}}
|
||||
// 一半显式指定归属,一半靠 SDK 填默认值
|
||||
if i%2 == 0 {
|
||||
def.Plugin = "explicit"
|
||||
}
|
||||
if err := s.RegisterTool(name, def, func(map[string]interface{}) (interface{}, error) {
|
||||
return nil, nil
|
||||
}); err != nil {
|
||||
t.Errorf("RegisterTool: %v", err)
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if len(got) != workers*perWorker {
|
||||
t.Fatalf("注册工具数 = %d,期望 %d", len(got), workers*perWorker)
|
||||
}
|
||||
for name, owner := range got {
|
||||
want := "mine"
|
||||
if isEvenSuffix(name) {
|
||||
want = "explicit"
|
||||
}
|
||||
if owner != want {
|
||||
t.Fatalf("工具 %s 归属 = %q,期望 %q", name, owner, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// isEvenSuffix 判断 tool_wX_iY 里的 Y 是否为偶数。
|
||||
func isEvenSuffix(name string) bool {
|
||||
idx := strings.LastIndex(name, "_i")
|
||||
if idx < 0 {
|
||||
return false
|
||||
}
|
||||
n := 0
|
||||
if _, err := fmt.Sscanf(name[idx+2:], "%d", &n); err != nil {
|
||||
return false
|
||||
}
|
||||
return n%2 == 0
|
||||
}
|
||||
|
||||
// nil 依赖下所有便捷方法必须静默降级而非 panic。
|
||||
//
|
||||
// 这是"媒体存储可关闭"在 SDK 层的对应物:内核未注入某个 API 时
|
||||
// (精简部署、插件权限不足、子进程握手尚未完成),插件的调用不该崩。
|
||||
func TestStress_NilDependenciesDegradeSilently(t *testing.T) {
|
||||
s := &PluginSDK{name: "bare"}
|
||||
|
||||
const workers = 16
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 200; i++ {
|
||||
s.InjectText("s", "c", "t")
|
||||
s.InjectTextNoMemory("s", "c", "t")
|
||||
s.InjectInterruptText("s", "c", "t")
|
||||
if got := s.InjectInputSync("s", "c", "t"); got != "" {
|
||||
t.Errorf("无 injector 时同步注入应返回空串,got %q", got)
|
||||
}
|
||||
s.InjectInputMedia("s", "c", "t", []ContentBlock{imageBlock("x")})
|
||||
if got := s.InjectInputMediaSync("s", "c", "t", nil); got != "" {
|
||||
t.Errorf("无 injector 时媒体同步注入应返回空串,got %q", got)
|
||||
}
|
||||
s.InjectInterruptMedia("s", "c", "t", nil)
|
||||
|
||||
// getter 全部应返回 nil 而非 panic
|
||||
_ = s.Memory()
|
||||
_ = s.TextMemory()
|
||||
_ = s.DocMemory()
|
||||
_ = s.Knowledge()
|
||||
_ = s.LLM()
|
||||
_ = s.Social()
|
||||
_ = s.Events()
|
||||
_ = s.PluginMgr()
|
||||
_ = s.Settings()
|
||||
|
||||
// 注册面无 registrar 时应返回 nil error
|
||||
if err := s.RegisterTool("t", ToolDef{}, nil); err != nil {
|
||||
t.Errorf("无 registrar 时 RegisterTool 应返回 nil,got %v", err)
|
||||
}
|
||||
if err := s.RegisterPluginAPI("a"); err != nil {
|
||||
t.Errorf("无 registrar 时 RegisterPluginAPI 应返回 nil,got %v", err)
|
||||
}
|
||||
s.RegisterStage(StageOnInput, func(*StageContext) error { return nil })
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
Reference in New Issue
Block a user