jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-09-25 03:18:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

v4 architecture: pipeline stages, SDK, event bus, LLM-driven memory consolidation

Browse Source 
- SDK PluginAPI (internal/plugin/sdk/): RegisterTool/RegisterStage/Subscribe/Publish
- EventBus (internal/events/): system-level pub/sub with wildcard support
- StageHost (internal/agent/core/stages.go): 7-stage message pipeline
- Agent core: on_input/pre_action/post_action/before_toolcall/after_toolcall/before_output/after_output
- Plugin Registry: SDK plugin registration and tool routing
- GraphDB.MergeEntities: entity consolidation with relation redirection
- memory_merge tool: allows LLM to merge similar entities
- Consolidation task: heartbeat detects conflicts, enqueues via IO for LLM decision
- _consolidation_ internal channel for system-level memory maintenance
- Comprehensive documentation: ARCHITECTURE.md, PLAN.md, DESIGN.md, README.md
- 54 tests across all packages, all passing
This commit is contained in:
root
2026-07-03 08:04:39 +08:00
parent 304c3ae294
commit 3e3c6a24d2
20 changed files with 2318 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files

472
internal/agent/core/agent.go
View File

@ -11,10 +11,12 @@ import (
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
agentPkg "gitcode.com/JianFeeeee/HomeAgent/internal/agent"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
"gitcode.com/JianFeeeee/HomeAgent/internal/knowledge"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/document"
"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
"gitcode.com/JianFeeeee/HomeAgent/internal/skill"
"gitcode.com/JianFeeeee/HomeAgent/internal/tracker"
"gitcode.com/JianFeeeee/HomeAgent/pkg/types"
@ -59,6 +61,10 @@ type Agent struct {
// 当前请求的输出通道（mutex 保护，process() 内独占）
currentOutputChannel string
// 阶段管道：插件消息流编辑
stageHost *StageHost
eventBus *events.Bus
}
type AgentConfig struct {
@ -78,7 +84,10 @@ type AgentConfig struct {
PluginReg *plugin.Registry
PluginDir string
DistillInterval time.Duration
MaxContextSize int // 活跃上下文最大条数，超出按相关性裁剪
MaxContextSize int // 活跃上下文最大条数，超出按相关性裁剪
ContextSavePath string // 上下文持久化路径，空则不持久化
StageHost *StageHost
EventBus *events.Bus
}
func New(cfg AgentConfig) *Agent {
@ -100,7 +109,7 @@ func New(cfg AgentConfig) *Agent {
indexer: cfg.Indexer,
skills: cfg.Skills,
tracker: cfg.Tracker,
context: NewRelevanceContext(),
context: NewRelevanceContext(cfg.ContextSavePath),
systemPrompt: cfg.SystemPrompt,
ctx: ctx,
cancel: cancel,
@ -112,6 +121,8 @@ func New(cfg AgentConfig) *Agent {
pluginDir: cfg.PluginDir,
distillInterval: cfg.DistillInterval,
maxContextSize: cfg.MaxContextSize,
stageHost: cfg.StageHost,
eventBus: cfg.EventBus,
}
}
@ -168,13 +179,31 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) {
a.currentOutputChannel = evt.Source
}
// 记忆整理任务：不路由到外部输出通道
if evt.OutputChannel == "_consolidation_" {
a.processConsolidation(input)
return
}
// === Stage: on_input — 消息到达，插件可拦截 ===
stageCtx := a.stageCtxFromInput(input, evt.Source, "")
a.publishEvent(events.EventRawInput, map[string]interface{}{
"content": input,
"source": evt.Source,
})
if a.runStage(sdk.StageOnInput, stageCtx) {
a.emitResponse(evt, *stageCtx.Response)
return
}
input = stageCtx.RawMessage
a.context.Append(ContextEvent{
Timestamp: start,
Source: evt.Source,
Input: input,
})
response, toolsUsed, err := a.process(input)
response, toolsUsed, err := a.process(input, stageCtx)
if err != nil {
log.Printf("[agent] process error: %v", err)
resp := fmt.Sprintf("处理错误: %v", err)
@ -206,6 +235,14 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) {
}
func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
// === Stage: before_output — 最终文本就绪，插件可改写 ===
stageCtx := &sdk.StageContext{
FinalText: response,
Phase: sdk.StageBeforeOutput,
}
a.runStage(sdk.StageBeforeOutput, stageCtx)
response = stageCtx.FinalText
// 读取当前输出通道（可能已被 AI 通过 output_set_channel 切换）
ch := a.currentOutputChannel
if ch == "" {
@ -230,10 +267,19 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
OutputChannel: ch,
}
}
// === Stage: after_output — 输出完成，插件只读 ===
a.publishEvent(events.EventAgentOutput, map[string]interface{}{
"content": response,
"channel": ch,
"source": evt.Source,
})
stageCtx.Phase = sdk.StageAfterOutput
a.runStage(sdk.StageAfterOutput, stageCtx)
}
// process — 内部处理，带工具循环
func (a *Agent) process(input string) (response string, toolsUsed []string, err error) {
// process — 内部处理，带工具循环和阶段管道
func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, err error) {
a.mu.Lock()
defer a.mu.Unlock()
@ -248,6 +294,20 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err
a.personality != nil && a.personality.Content != "",
a.docStoreSize())
// === Stage: pre_action — 上下文就绪，即将调用 LLM ===
if a.runStage(sdk.StagePreAction, stageCtx) {
return *stageCtx.Response, toolsUsed, nil
}
if len(stageCtx.ContextMsgs) > 0 {
for _, m := range stageCtx.ContextMsgs {
role, _ := m["role"].(string)
content, _ := m["content"].(string)
if role != "" {
msgs = append(msgs, agentAPI.Message{Role: role, Content: content})
}
}
}
for turn := 0; turn < a.maxTurns; turn++ {
req := &agentAPI.CompletionRequest{
Messages: msgs,
@ -264,6 +324,15 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err
return "", toolsUsed, fmt.Errorf("provider: %w", err)
}
// === Stage: post_action — LLM 返回，插件可审查/修改 ===
stageCtx.LLMText = resp.Content
stageCtx.ToolCalls = convertToolCalls(resp.ToolCalls)
if a.runStage(sdk.StagePostAction, stageCtx) {
return *stageCtx.Response, toolsUsed, nil
}
resp.Content = stageCtx.LLMText
resp.ToolCalls = convertBackToolCalls(stageCtx.ToolCalls)
if len(resp.ToolCalls) == 0 {
return resp.Content, toolsUsed, nil
}
@ -271,16 +340,74 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err
for _, tc := range resp.ToolCalls {
toolsUsed = append(toolsUsed, tc.Name)
log.Printf("[agent] executing tool: %s (id=%s)", tc.Name, tc.ID)
// === Stage: before_toolcall — 插件可拒绝/改参 ===
sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments}
stageCtx.ToolCalls = []sdk.ToolCall{sdkTC}
stageCtx.ToolResults = nil
if a.runStage(sdk.StageBeforeToolcall, stageCtx) {
result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name)
msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{tc}})
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"args": tc.Arguments,
"result": result,
"status": "denied",
})
continue
}
tc.Arguments = stageCtx.ToolCalls[0].Arguments
result := a.executeToolCall(tc)
log.Printf("[agent] tool %s result: %s", tc.Name, truncateStr(result, 100))
// === Stage: after_toolcall — 插件可改结果 ===
stageCtx.ToolResults = []sdk.ToolResult{{CallID: tc.ID, Name: tc.Name, Success: true, Result: result}}
a.runStage(sdk.StageAfterToolcall, stageCtx)
if len(stageCtx.ToolResults) > 0 {
if r, ok := stageCtx.ToolResults[0].Result.(string); ok {
result = r
}
}
msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{tc}})
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"args": tc.Arguments,
"result": result,
"status": "ok",
})
}
}
return "", toolsUsed, fmt.Errorf("tool execution exceeded %d turns", a.maxTurns)
}
func convertToolCalls(tcs []agentAPI.ToolCall) []sdk.ToolCall {
if tcs == nil {
return nil
}
result := make([]sdk.ToolCall, len(tcs))
for i, tc := range tcs {
result[i] = sdk.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments}
}
return result
}
func convertBackToolCalls(tcs []sdk.ToolCall) []agentAPI.ToolCall {
if tcs == nil {
return nil
}
result := make([]agentAPI.ToolCall, len(tcs))
for i, tc := range tcs {
result[i] = agentAPI.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments}
}
return result
}
func (a *Agent) docStoreSize() int {
if a.docStore == nil {
return 0
@ -322,6 +449,15 @@ func (a *Agent) executeToolCall(tc agentAPI.ToolCall) string {
return a.executeOutputListChannels()
case tc.Name == "plgreload":
return a.executePluginReload()
case tc.Name == "spawn_child":
return a.executeSpawnChild(tc)
}
// 插件工具（通过 SDK RegisterTool 注册）
if a.stageHost != nil {
if result, err := a.stageHost.ExecuteTool(tc.Name, tc.Arguments); err == nil {
return fmt.Sprintf("%v", result)
}
}
if a.tracker != nil {
@ -414,7 +550,19 @@ func (a *Agent) executeMemoryTool(tc agentAPI.ToolCall) string {
return fmt.Sprintf("记忆统计: %v", stats)
case "memory_document_query":
return a.executeDocTool(tc)
return a.executeDocTool(tc)
case "memory_merge":
source, _ := tc.Arguments["source"].(string)
target, _ := tc.Arguments["target"].(string)
if source == "" || target == "" {
return "source 和 target 不能为空"
}
count, err := a.memory.MergeEntities(source, target)
if err != nil {
return fmt.Sprintf("合并失败: %v", err)
}
return fmt.Sprintf("已将「%s」合并到「%s」，%d 条关系已重定向", source, target, count)
default:
return fmt.Sprintf("未知的记忆工具: %s", tc.Name)
@ -600,12 +748,45 @@ func (a *Agent) buildToolDefs() []interface{} {
}
}
// 插件注册的工具（通过 SDK RegisterTool）
if a.stageHost != nil {
for _, td := range a.stageHost.GetToolDefs() {
tools = append(tools, map[string]interface{}{
"type": "function",
"function": map[string]interface{}{
"name": td.Name,
"description": td.Description,
"parameters": td.Parameters,
},
})
}
}
if a.indexer != nil {
for _, td := range a.indexer.GetToolDefinitions() {
tools = append(tools, td)
}
}
// 实体合并工具（心跳检测到冲突时 LLM 使用）
if a.memory != nil {
tools = append(tools, map[string]interface{}{
"type": "function",
"function": map[string]interface{}{
"name": "memory_merge",
"description": "合并两个同义实体：将所有关系从 source 重定向到 target，source 标记为 merged。仅在有明确证据时使用。",
"parameters": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"source": map[string]interface{}{"type": "string", "description": "被合并的实体名（合并后消失）"},
"target": map[string]interface{}{"type": "string", "description": "保留的实体名"},
},
"required": []string{"source", "target"},
},
},
})
}
// 知识库工具
if a.knowledge != nil {
tools = append(tools, map[string]interface{}{
@ -709,6 +890,25 @@ func (a *Agent) buildToolDefs() []interface{} {
})
}
// 子任务工具
tools = append(tools, map[string]interface{}{
"type": "function",
"function": map[string]interface{}{
"name": "spawn_child",
"description": "创建一个子 Agent 执行独立任务。子 Agent 使用传统上下文（无持久记忆），任务完成即销毁。适用于需要多步推理但不需要写入长期记忆的场景，例如：计算、分析、生成报告草稿等。",
"parameters": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"task": map[string]interface{}{
"type": "string",
"description": "要子 Agent 完成的任务描述。请描述清晰、完整，包含所有必要背景。",
},
},
"required": []string{"task"},
},
},
})
// 输出通道工具
tools = append(tools, map[string]interface{}{
"type": "function",
@ -764,6 +964,20 @@ func (a *Agent) buildToolDefs() []interface{} {
return tools
}
// ConsolidationTask 心跳检测到的记忆整理任务，通过 IO 发送给 Agent 让 LLM 决策
type ConsolidationTask struct {
Type string `json:"type"` // "entity_merge", "relation_conflict", "doc_archival"
Reason string `json:"reason"` // 人类可读的描述
Data interface{} `json:"data"` // 任务相关数据
}
// enqueueConsolidationTask 将记忆整理任务注入到 Agent 输入队列
func (a *Agent) enqueueConsolidationTask(task ConsolidationTask) {
msg := fmt.Sprintf("【记忆整理任务】\n类型: %s\n说明: %s", task.Type, task.Reason)
a.io.InjectTextTo("system", "_consolidation_", msg)
log.Printf("[agent] enqueued consolidation task: %s", task.Reason)
}
// distillLoop — 定期心跳：上下文→文档 + 图→文档 + 图重整
func (a *Agent) distillLoop() {
if a.docStore == nil && a.memory == nil {
@ -896,34 +1110,85 @@ func (a *Agent) reorgGraph() {
}
}
// 4. 实体向量同义合并
// 4. 实体同义冲突检测 → 交由 LLM 决策
result, err := a.memory.Recall(nil, nil, 1, "")
if err != nil || result == nil || len(result.Entities) < 2 {
return
}
merged := 0
candidates := 0
for i := 0; i < len(result.Entities); i++ {
for j := i + 1; j < len(result.Entities); j++ {
if isSimilarName(result.Entities[i].Name, result.Entities[j].Name) {
if result.Entities[i].MentionCount >= result.Entities[j].MentionCount {
log.Printf("[agent] reorg: merging '%s' → '%s'", result.Entities[j].Name, result.Entities[i].Name)
} else {
log.Printf("[agent] reorg: merging '%s' → '%s'", result.Entities[i].Name, result.Entities[j].Name)
}
merged++
sim := entitySimilarity(result.Entities[i].Name, result.Entities[j].Name)
if sim > 0.5 {
candidates++
a.enqueueConsolidationTask(ConsolidationTask{
Type: "entity_merge",
Reason: fmt.Sprintf(
"实体「%s」(类型:%s, 提及%d次) 与「%s」(类型:%s, 提及%d次) 相似度 %.0f%%，可能指代同一事物，请判断是否需要合并",
result.Entities[i].Name, result.Entities[i].Type, result.Entities[i].MentionCount,
result.Entities[j].Name, result.Entities[j].Type, result.Entities[j].MentionCount,
sim*100,
),
Data: map[string]interface{}{
"entity_a": result.Entities[i].Name,
"entity_a_type": result.Entities[i].Type,
"entity_a_mentions": result.Entities[i].MentionCount,
"entity_b": result.Entities[j].Name,
"entity_b_type": result.Entities[j].Type,
"entity_b_mentions": result.Entities[j].MentionCount,
"similarity": sim,
},
})
}
}
}
if merged > 0 {
log.Printf("[agent] graph reorg: merged %d similar entities", merged)
if candidates > 0 {
log.Printf("[agent] graph reorg: %d merge candidates sent for LLM decision", candidates)
} else {
log.Printf("[agent] graph reorg: no merges needed")
log.Printf("[agent] graph reorg: no similar entities found")
}
}
// isSimilarName — 使用字符 bigram Jaccard 相似度判断实体名是否同义
// entitySimilarity 计算两个实体名的相似度（字符 bigram Jaccard）
func entitySimilarity(a, b string) float64 {
if a == "" || b == "" {
return 0
}
if a == b {
return 1.0
}
runesA, runesB := []rune(a), []rune(b)
if len(runesA) < 2 || len(runesB) < 2 {
if len(runesA) == len(runesB) && len(runesA) == 1 {
if runesA[0] == runesB[0] {
return 1.0
}
}
return 0
}
setA := make(map[string]bool)
for i := 0; i < len(runesA)-1; i++ {
setA[string(runesA[i:i+2])] = true
}
intersect := 0
for i := 0; i < len(runesB)-1; i++ {
if setA[string(runesB[i:i+2])] {
intersect++
}
}
union := len(setA) + len(runesB) - 1 - intersect
if union <= 0 {
return 0
}
return float64(intersect) / float64(union)
}
// docToTriples 将文档转为图记忆三元组
func docToTriples(doc *document.Doc) []memory.Triple {
var triples []memory.Triple
@ -964,36 +1229,6 @@ func docToTriples(doc *document.Doc) []memory.Triple {
return triples
}
func isSimilarName(a, b string) bool {
if a == b {
return false // 自带跳过
}
runesA, runesB := []rune(a), []rune(b)
if len(runesA) < 2 || len(runesB) < 2 {
return false
}
setA := make(map[string]bool)
for i := 0; i < len(runesA)-1; i++ {
setA[string(runesA[i:i+2])] = true
}
intersect := 0
for i := 0; i < len(runesB)-1; i++ {
if setA[string(runesB[i:i+2])] {
intersect++
}
}
union := len(setA) + len(runesB) - 1 - intersect
if union <= 0 {
return false
}
jaccard := float64(intersect) / float64(union)
return jaccard > 0.5
}
func (a *Agent) emitMemoryCandidate(source, input, response string, toolsUsed []string) {
a.io.EmitOutput("memory", "memory_candidate", map[string]interface{}{
"source": source,
@ -1007,6 +1242,33 @@ func (a *Agent) emitMemoryCandidate(source, input, response string, toolsUsed []
// executeOutputChannelTool — AI 切换当前请求的输出通道
// 在 process() 内调用，mutex 保护，只有一个请求在执行
// processConsolidation 处理后台记忆整理任务（不发外部输出）
func (a *Agent) processConsolidation(input string) {
start := time.Now()
a.currentOutputChannel = "_consolidation_"
a.context.Append(ContextEvent{
Timestamp: start,
Source: "system",
Input: input,
})
response, toolsUsed, err := a.process(input, &sdk.StageContext{RawMessage: input})
if err != nil {
log.Printf("[agent] consolidation error: %v", err)
return
}
a.context.Append(ContextEvent{
Timestamp: time.Now(),
Source: "agent",
Input: input,
Response: response,
ToolsUsed: toolsUsed,
})
_ = a.context.Prune(response, a.maxContextSize, a.docStore)
// 只写入记忆，不发外部输出
a.emitMemoryCandidate("system", input, response, toolsUsed)
log.Printf("[agent] consolidation done (%dms, tools=%v)", time.Since(start).Milliseconds(), toolsUsed)
}
func (a *Agent) executeOutputChannelTool(tc agentAPI.ToolCall) string {
channel, _ := tc.Arguments["channel"].(string)
if channel == "" {
@ -1077,6 +1339,118 @@ func (a *Agent) executePluginReload() string {
return msg
}
// executeSpawnChild 创建子 Agent 执行独立任务
// 子 Agent 使用传统上下文（单轮对话），无持久记忆，任务完即销毁
func (a *Agent) executeSpawnChild(tc agentAPI.ToolCall) string {
task, _ := tc.Arguments["task"].(string)
if task == "" {
return "请提供 task 参数"
}
sysPrompt := fmt.Sprintf(`你是 HomeAgent 的子任务助手。
请完成以下任务。完成即可，无需保留记忆或查询历史。
任务: %s`, task)
msgs := []agentAPI.Message{
{Role: "system", Content: sysPrompt},
{Role: "user", Content: task},
}
// 子 Agent 无特殊工具，只保留基础 tool 定义（无记忆/知识/文档工具）
childTools := []interface{}{
map[string]interface{}{
"type": "function",
"function": map[string]interface{}{
"name": "output_send",
"description": "通过指定输出通道发送消息",
"parameters": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"channel": map[string]interface{}{"type": "string", "description": "输出通道"},
"content": map[string]interface{}{"type": "string", "description": "消息内容"},
},
"required": []string{"channel", "content"},
},
},
},
}
for turn := 0; turn < 5; turn++ {
req := &agentAPI.CompletionRequest{
Messages: msgs,
MaxTokens: 4096,
Tools: childTools,
ToolChoice: "auto",
ExtraBody: map[string]interface{}{
"thinking": map[string]interface{}{"type": "disabled"},
},
}
resp, err := a.provider.Chat(a.ctx, req)
if err != nil {
return fmt.Sprintf("子 Agent 执行失败: %v", err)
}
if len(resp.ToolCalls) == 0 {
return resp.Content
}
for _, ct := range resp.ToolCalls {
var result string
if ct.Name == "output_send" {
channel, _ := ct.Arguments["channel"].(string)
content, _ := ct.Arguments["content"].(string)
if channel != "" && content != "" {
a.io.EmitTextTo("child_agent", channel, content)
result = fmt.Sprintf("已通过 [%s] 通道发送", channel)
} else {
result = "channel 和 content 不能为空"
}
} else {
result = fmt.Sprintf("子 Agent 无法调用工具 %s", ct.Name)
}
msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{ct}})
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: ct.ID, Content: result})
}
}
return "子 Agent 执行超时（超过 5 轮）"
}
// runStage — 运行阶段管道，若插件 Response 被设置则返回 true（短路）
func (a *Agent) runStage(stage sdk.Stage, ctx *sdk.StageContext) bool {
if a.stageHost == nil {
return false
}
ctx.Phase = stage
a.stageHost.RunStage(stage, ctx)
return ctx.Response != nil
}
// publishEvent — 发布系统事件
func (a *Agent) publishEvent(evtType events.EventType, payload map[string]interface{}) {
if a.eventBus == nil {
return
}
a.eventBus.Publish(&events.Event{
Type: evtType,
Source: string(a.id),
Payload: payload,
Timestamp: time.Now().Unix(),
})
}
// stageCtxFromInput — 根据输入构建阶段上下文
func (a *Agent) stageCtxFromInput(input, userID, groupID string) *sdk.StageContext {
return &sdk.StageContext{
RawMessage: input,
UserID: userID,
GroupID: groupID,
Phase: sdk.StageOnInput,
Extra: make(map[string]interface{}),
}
}
func getFloat(m map[string]interface{}, key string) float64 {
if v, ok := m[key]; ok {
switch n := v.(type) {

18
internal/agent/core/agent_functions_test.go
View File

@ -6,21 +6,21 @@ import (
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/document"
)
func TestIsSimilarName(t *testing.T) {
func TestEntitySimilarity(t *testing.T) {
tests := []struct {
a, b string
want bool
want float64
}{
{"张三", "张三四", false},
{"", "", false},
{"a", "b", false},
{"张三", "李四", false},
{"张三", "张三", false},
{"", "", 0}, // empty → 0
{"a", "b", 0}, // single char → 0
{"张三", "张三", 1.0}, // identical → 1.0
{"张三", "李四", 0}, // no common bigrams
{"iPhone", "iPhone 15", 0.625}, // partial overlap
}
for _, tt := range tests {
got := isSimilarName(tt.a, tt.b)
got := entitySimilarity(tt.a, tt.b)
if got != tt.want {
t.Errorf("isSimilarName(%q, %q) = %v, want %v", tt.a, tt.b, got, tt.want)
t.Errorf("entitySimilarity(%q, %q) = %.3f, want %.3f", tt.a, tt.b, got, tt.want)
}
}
}

82
internal/agent/core/context.go
View File

@ -1,7 +1,10 @@
package core
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
@ -13,26 +16,63 @@ import (
// ContextEvent — 单条上下文事件
type ContextEvent struct {
Timestamp time.Time `json:"timestamp"`
Source string `json:"source"`
Input string `json:"input"`
Response string `json:"response,omitempty"`
ToolsUsed []string `json:"tools_used,omitempty"`
Vector vector.Vector `json:"-"` // 缓存向量，避免重复计算
Timestamp time.Time `json:"timestamp"`
Source string `json:"source"`
Input string `json:"input"`
Response string `json:"response,omitempty"`
ToolsUsed []string `json:"tools_used,omitempty"`
Vector vector.Vector `json:"-"` // 缓存向量，避免重复计算
}
// RelevanceContext — 基于相关性的上下文管理，非固定阈值
type RelevanceContext struct {
mu sync.Mutex
events []*ContextEvent
veczer *vector.TFIDFVectorizer
trained bool
mu sync.Mutex
events []*ContextEvent
veczer *vector.TFIDFVectorizer
trained bool
savePath string // 持久化路径，空则不持久化
}
func NewRelevanceContext() *RelevanceContext {
return &RelevanceContext{
veczer: vector.NewTFIDFVectorizer(2),
func NewRelevanceContext(savePath string) *RelevanceContext {
rc := &RelevanceContext{
veczer: vector.NewTFIDFVectorizer(2),
savePath: savePath,
}
if savePath != "" {
rc.load()
}
return rc
}
// load 从文件恢复上下文事件
func (c *RelevanceContext) load() {
data, err := os.ReadFile(c.savePath)
if err != nil {
return
}
var events []*ContextEvent
if err := json.Unmarshal(data, &events); err != nil {
return
}
for _, evt := range events {
evt.Vector = c.veczer.Vectorize(evt.Input + " " + evt.Response)
}
c.events = events
}
// Save 持久化上下文事件到文件
func (c *RelevanceContext) Save() error {
if c.savePath == "" {
return nil
}
if err := os.MkdirAll(filepath.Dir(c.savePath), 0755); err != nil {
return err
}
data, err := json.Marshal(c.events)
if err != nil {
return err
}
return os.WriteFile(c.savePath, data, 0644)
}
func (c *RelevanceContext) Append(evt ContextEvent) {
@ -44,6 +84,20 @@ func (c *RelevanceContext) Append(evt ContextEvent) {
// 增量训练向量化器
c.trained = false
c.save()
}
// save 无锁版本，Append/Prune 内部持有锁时调用
func (c *RelevanceContext) save() error {
if c.savePath == "" {
return nil
}
data, err := json.Marshal(c.events)
if err != nil {
return err
}
return os.WriteFile(c.savePath, data, 0644)
}
// Prune — 基于当前输入计算每条上下文的相关性，归档最不相关的
@ -114,6 +168,8 @@ func (c *RelevanceContext) Prune(currentInput string, topK int, docStore *docume
}
}
c.save()
return archived
}

12
internal/agent/core/context_test.go
View File

@ -6,7 +6,7 @@ import (
)
func TestContextAppendAndLen(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
if ctx.Len() != 0 {
t.Errorf("new context should be empty, got %d", ctx.Len())
}
@ -18,7 +18,7 @@ func TestContextAppendAndLen(t *testing.T) {
}
func TestContextRecent(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "a"})
ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "b"})
ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "c"})
@ -33,7 +33,7 @@ func TestContextRecent(t *testing.T) {
}
func TestContextFormat(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
f := ctx.Format()
if f != "" {
t.Errorf("empty context should format to empty string, got %q", f)
@ -54,7 +54,7 @@ func TestContextFormat(t *testing.T) {
}
func TestContextPruneKeepsTopK(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
for i := 0; i < 10; i++ {
ctx.Append(ContextEvent{
Timestamp: time.Now(),
@ -80,7 +80,7 @@ func TestContextPruneKeepsTopK(t *testing.T) {
}
func TestContextPruneWithDocStore(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
for i := 0; i < 15; i++ {
ctx.Append(ContextEvent{
Timestamp: time.Now(),
@ -97,7 +97,7 @@ func TestContextPruneWithDocStore(t *testing.T) {
}
func TestContextAppendAfterPrune(t *testing.T) {
ctx := NewRelevanceContext()
ctx := NewRelevanceContext("")
for i := 0; i < 10; i++ {
ctx.Append(ContextEvent{
Timestamp: time.Now(),

74
internal/agent/core/stages.go Normal file
View File

@ -0,0 +1,74 @@
package core
import (
"fmt"
"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
)
type StageHost struct {
plugins []*sdk.PluginAPI
toolDefs []sdk.ToolDef
tools map[string]sdk.ToolHandler
}
func NewStageHost() *StageHost {
return &StageHost{
tools: make(map[string]sdk.ToolHandler),
}
}
func (h *StageHost) RegisterPlugin(api *sdk.PluginAPI) {
h.plugins = append(h.plugins, api)
for name, handler := range api.Tools() {
h.tools[name] = handler
h.toolDefs = append(h.toolDefs, sdk.ToolDef{Name: name})
}
}
// SyncFromRegistry 从插件注册表同步 SDK 插件
func (h *StageHost) SyncFromRegistry(reg *plugin.Registry) {
if reg == nil {
return
}
for _, td := range reg.GetAllSDKToolDefs() {
h.toolDefs = append(h.toolDefs, td)
}
}
func (h *StageHost) GetToolDefs() []sdk.ToolDef {
return h.toolDefs
}
func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (interface{}, error) {
if handler, ok := h.tools[name]; ok {
return handler(args)
}
return nil, fmt.Errorf("tool %s not found in any plugin", name)
}
func (h *StageHost) RunStage(stage sdk.Stage, ctx *sdk.StageContext) {
for _, p := range h.plugins {
for _, handler := range p.StageHandlers(stage) {
if err := handler(ctx); err != nil {
return
}
if ctx.Response != nil {
return
}
}
}
}
func (h *StageHost) RunStageAll(stage sdk.Stage, ctx *sdk.StageContext) {
for _, p := range h.plugins {
for _, handler := range p.StageHandlers(stage) {
handler(ctx)
}
}
}
func (h *StageHost) PluginCount() int {
return len(h.plugins)
}

183
internal/agent/core/stages_test.go Normal file
View File

@ -0,0 +1,183 @@
package core
import (
"testing"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
)
func TestStageHostRegisterPlugin(t *testing.T) {
host := NewStageHost()
api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)
api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) {
return "ok", nil
})
host.RegisterPlugin(api)
if host.PluginCount() != 1 {
t.Errorf("expected 1 plugin, got %d", host.PluginCount())
}
defs := host.GetToolDefs()
if len(defs) != 1 {
t.Errorf("expected 1 tool def, got %d", len(defs))
}
if defs[0].Name != "test_tool" {
t.Errorf("expected test_tool, got %s", defs[0].Name)
}
}
func TestStageHostExecuteTool(t *testing.T) {
host := NewStageHost()
api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)
api.RegisterTool("hello", func(args map[string]interface{}) (interface{}, error) {
return "world", nil
})
host.RegisterPlugin(api)
result, err := host.ExecuteTool("hello", nil)
if err != nil {
t.Fatalf("execute: %v", err)
}
if result.(string) != "world" {
t.Errorf("expected world, got %v", result)
}
_, err = host.ExecuteTool("nonexistent", nil)
if err == nil {
t.Error("expected error for nonexistent tool")
}
}
func TestStageHostRunStage(t *testing.T) {
host := NewStageHost()
api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)
var called bool
api.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
called = true
return nil
})
host.RegisterPlugin(api)
ctx := &sdk.StageContext{RawMessage: "hello"}
host.RunStage(sdk.StageOnInput, ctx)
if !called {
t.Error("stage handler was not called")
}
}
func TestStageHostRunStageShortCircuit(t *testing.T) {
host := NewStageHost()
api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
api1.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
resp := "short-circuited"
ctx.Response = &resp
return nil
})
var api2called bool
api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
api2.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
api2called = true
return nil
})
host.RegisterPlugin(api1)
host.RegisterPlugin(api2)
ctx := &sdk.StageContext{RawMessage: "hello"}
host.RunStage(sdk.StageOnInput, ctx)
if ctx.Response == nil || *ctx.Response != "short-circuited" {
t.Errorf("expected short-circuited, got %v", ctx.Response)
}
if api2called {
t.Error("api2 should not have been called after short circuit")
}
}
func TestStageHostRunStageAll(t *testing.T) {
host := NewStageHost()
count := 0
api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
api1.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
count++
return nil
})
api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
api2.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
count++
return nil
})
host.RegisterPlugin(api1)
host.RegisterPlugin(api2)
host.RunStageAll(sdk.StageAfterOutput, &sdk.StageContext{})
if count != 2 {
t.Errorf("expected 2 handlers called, got %d", count)
}
}
func TestStageHostMultiplePlugins(t *testing.T) {
host := NewStageHost()
p1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
p1.RegisterTool("tool1", func(args map[string]interface{}) (interface{}, error) {
return "from_p1", nil
})
p2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
p2.RegisterTool("tool2", func(args map[string]interface{}) (interface{}, error) {
return "from_p2", nil
})
host.RegisterPlugin(p1)
host.RegisterPlugin(p2)
if host.PluginCount() != 2 {
t.Errorf("expected 2 plugins, got %d", host.PluginCount())
}
r1, _ := host.ExecuteTool("tool1", nil)
if r1.(string) != "from_p1" {
t.Errorf("expected from_p1, got %v", r1)
}
r2, _ := host.ExecuteTool("tool2", nil)
if r2.(string) != "from_p2" {
t.Errorf("expected from_p2, got %v", r2)
}
}
func TestStageHostEmpty(t *testing.T) {
host := NewStageHost()
if host.PluginCount() != 0 {
t.Errorf("expected 0 plugins, got %d", host.PluginCount())
}
defs := host.GetToolDefs()
if len(defs) != 0 {
t.Errorf("expected 0 tool defs, got %d", len(defs))
}
_, err := host.ExecuteTool("anything", nil)
if err == nil {
t.Error("expected error on empty host")
}
// RunStage on empty host should not panic
host.RunStage(sdk.StageOnInput, &sdk.StageContext{})
}

68
internal/api/plugin.go Normal file
View File

@ -0,0 +1,68 @@
package api
import (
"log"
"net/http"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
"gitcode.com/JianFeeeee/HomeAgent/internal/knowledge"
luaVM "gitcode.com/JianFeeeee/HomeAgent/internal/lua"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/text"
"gitcode.com/JianFeeeee/HomeAgent/internal/skill"
"gitcode.com/JianFeeeee/HomeAgent/internal/supervisor"
"gitcode.com/JianFeeeee/HomeAgent/internal/tracker"
"gitcode.com/JianFeeeee/HomeAgent/pkg/types"
)
// Plugin 将 HTTP API + WebUI 包装为 IO Device
// 作为 HomeAgent 自带的默认 IO 通道插件
type Plugin struct {
name string
handler *Handler
server *http.Server
addr string
mux *http.ServeMux
}
func NewWebUIPlugin(name, addr string, sup *supervisor.Daemon, mem *memory.GraphDB, sk *skill.Manager,
lua *luaVM.VM, cfg *types.Config, iom *agentIO.IOManager, tm *text.Memory, ks *knowledge.Store, tr *tracker.Tracker) *Plugin {
h := NewHandler(sup, mem, sk, lua, cfg, iom, tm, ks, tr)
mux := http.NewServeMux()
h.RegisterRoutes(mux)
return &Plugin{
name: name,
handler: h,
addr: addr,
mux: mux,
}
}
func (p *Plugin) Name() string { return p.name }
func (p *Plugin) Type() agentIO.DeviceType { return agentIO.DeviceIO }
func (p *Plugin) Description() string { return "HTTP API & Web Dashboard" }
func (p *Plugin) OutputCapabilities() agentIO.OutputCapability { return agentIO.CapText | agentIO.CapStructured }
func (p *Plugin) Tools() []agentIO.ToolDef { return nil }
func (p *Plugin) Execute(tool string, args map[string]interface{}) (interface{}, error) {
return nil, nil
}
func (p *Plugin) Start() error {
p.server = &http.Server{Addr: p.addr, Handler: p.mux}
go func() {
log.Printf("[webui] HTTP server listening on %s", p.addr)
if err := p.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Printf("[webui] server error: %v", err)
}
}()
return nil
}
func (p *Plugin) Stop() error {
if p.server != nil {
return p.server.Close()
}
return nil
}

69
internal/events/bus.go Normal file
View File

@ -0,0 +1,69 @@
package events
import (
"fmt"
"sync"
)
type EventType string
const (
EventRawInput EventType = "raw_input"
EventAgentOutput EventType = "agent_output"
EventToolCall EventType = "tool_call"
EventReasoning EventType = "reasoning"
EventSystem EventType = "system"
EventAll EventType = "*"
)
type Event struct {
Type EventType `json:"type"`
Source string `json:"source"`
Payload map[string]interface{} `json:"payload"`
Timestamp int64 `json:"timestamp"`
}
type Handler func(event *Event)
type Bus struct {
mu sync.RWMutex
subs map[EventType][]Handler
}
func NewBus() *Bus {
return &Bus{
subs: make(map[EventType][]Handler),
}
}
func (b *Bus) Publish(evt *Event) {
b.mu.RLock()
allHandlers := b.subs[EventAll]
typeHandlers := b.subs[evt.Type]
b.mu.RUnlock()
for _, h := range allHandlers {
h(evt)
}
for _, h := range typeHandlers {
h(evt)
}
}
func (b *Bus) Subscribe(eventType EventType, handler Handler) func() {
b.mu.Lock()
b.subs[eventType] = append(b.subs[eventType], handler)
b.mu.Unlock()
return func() {
b.mu.Lock()
defer b.mu.Unlock()
list := b.subs[eventType]
for i, h := range list {
if fmt.Sprintf("%p", h) == fmt.Sprintf("%p", handler) {
b.subs[eventType] = append(list[:i], list[i+1:]...)
break
}
}
}
}

102
internal/events/bus_test.go Normal file
View File

@ -0,0 +1,102 @@
package events
import (
"sync/atomic"
"testing"
"time"
)
func TestBusPublishSubscribe(t *testing.T) {
bus := NewBus()
var count int32
bus.Subscribe(EventRawInput, func(evt *Event) {
atomic.AddInt32(&count, 1)
})
bus.Publish(&Event{
Type: EventRawInput,
Source: "test",
Payload: map[string]interface{}{"content": "hello"},
})
if c := atomic.LoadInt32(&count); c != 1 {
t.Errorf("expected 1, got %d", c)
}
}
func TestBusWildcard(t *testing.T) {
bus := NewBus()
var count int32
bus.Subscribe(EventAll, func(evt *Event) {
atomic.AddInt32(&count, 1)
})
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
bus.Publish(&Event{Type: EventToolCall, Source: "test"})
if c := atomic.LoadInt32(&count); c != 2 {
t.Errorf("expected 2, got %d", c)
}
}
func TestBusUnsubscribe(t *testing.T) {
bus := NewBus()
var count int32
handler := func(evt *Event) {
atomic.AddInt32(&count, 1)
}
unsub := bus.Subscribe(EventRawInput, handler)
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
unsub()
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
if c := atomic.LoadInt32(&count); c != 1 {
t.Errorf("expected 1 after unsub, got %d", c)
}
}
func TestBusNoMatch(t *testing.T) {
bus := NewBus()
var count int32
bus.Subscribe(EventRawInput, func(evt *Event) {
atomic.AddInt32(&count, 1)
})
bus.Publish(&Event{Type: EventAgentOutput, Source: "test"})
if c := atomic.LoadInt32(&count); c != 0 {
t.Errorf("expected 0, got %d", c)
}
}
func TestBusConcurrent(t *testing.T) {
bus := NewBus()
var count int32
bus.Subscribe(EventAll, func(evt *Event) {
atomic.AddInt32(&count, 1)
})
done := make(chan struct{})
go func() {
for i := 0; i < 100; i++ {
bus.Publish(&Event{Type: EventRawInput, Source: "test"})
}
close(done)
}()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("timeout")
}
if c := atomic.LoadInt32(&count); c != 100 {
t.Errorf("expected 100, got %d", c)
}
}

92
internal/memory/graph.go
View File

@ -520,6 +520,98 @@ func (g *GraphDB) Introspect() (map[string]interface{}, error) {
}, nil
}
// MergeEntities 合并两个实体：将 sourceName 的所有信息合并到 targetName
// 1. sourceName 的所有关系重新指向 targetName
// 2. targetName 的 mention_count 增加 sourceName 的计数
// 3. sourceName 标记为 merged
// 返回 (关系的重定向数, error)
func (g *GraphDB) MergeEntities(sourceName, targetName string) (int, error) {
g.mu.Lock()
defer g.mu.Unlock()
tx, err := g.db.Begin()
if err != nil {
return 0, err
}
defer tx.Rollback()
var sourceID, targetID int64
var sourceCount, targetCount int
err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", sourceName).Scan(&sourceID, &sourceCount)
if err != nil {
return 0, fmt.Errorf("source entity '%s' not found: %w", sourceName, err)
}
err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", targetName).Scan(&targetID, &targetCount)
if err != nil {
return 0, fmt.Errorf("target entity '%s' not found: %w", targetName, err)
}
if sourceID == targetID {
return 0, fmt.Errorf("cannot merge entity with itself")
}
// 重定向 source → target 的关系（作为 source）
res, err := tx.Exec(
`UPDATE relations SET source_id = ?, updated_at = CURRENT_TIMESTAMP
WHERE source_id = ? AND status = 'active'`,
targetID, sourceID,
)
if err != nil {
return 0, err
}
redirectedSource, _ := res.RowsAffected()
// 重定向 source → target 的关系（作为 target）
res, err = tx.Exec(
`UPDATE relations SET target_id = ?, updated_at = CURRENT_TIMESTAMP
WHERE target_id = ? AND status = 'active'`,
targetID, sourceID,
)
if err != nil {
return 0, err
}
redirectedTarget, _ := res.RowsAffected()
// 删除可能产生的自引用关系
_, err = tx.Exec(
`DELETE FROM relations
WHERE source_id = target_id AND source_id = ?`,
targetID,
)
if err != nil {
return 0, err
}
// 更新 target 的 mention_count
_, err = tx.Exec(
`UPDATE entities SET mention_count = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
targetCount+sourceCount, targetID,
)
if err != nil {
return 0, err
}
// 标记 source 为 merged（改名避免 UNIQUE 冲突）
_, err = tx.Exec(
`UPDATE entities SET name = ? || '@merged_' || ?,
mention_count = 0,
updated_at = CURRENT_TIMESTAMP
WHERE id = ?`,
sourceName, time.Now().Format("20060102150405"), sourceID,
)
if err != nil {
return 0, err
}
if err := tx.Commit(); err != nil {
return 0, err
}
total := int(redirectedSource + redirectedTarget)
return total, nil
}
func (g *GraphDB) Archive(days int) (int, error) {
g.mu.Lock()
defer g.mu.Unlock()

89
internal/memory/graph_test.go
View File

@ -214,6 +214,95 @@ func TestIntrospectHotspots(t *testing.T) {
}
}
func TestMergeEntities(t *testing.T) {
g := newTestGraph(t)
defer os.Remove(g.dbPath)
defer g.Close()
g.Commit([]Triple{
{Subject: "张三", Relation: "喜欢", Object: "编程"},
{Subject: "张三", Relation: "居住", Object: "北京"},
}, "session", 0)
g.Commit([]Triple{
{Subject: "张先生", Relation: "工作", Object: "字节跳动"},
}, "session", 0)
// 合并前：两个实体各有关联
stats, _ := g.Introspect()
if stats["entity_count"].(int) != 5 {
t.Fatalf("expected 5 entities (张三, 编程, 北京, 张先生, 字节跳动), got %d", stats["entity_count"])
}
// 先增加张先生的 mention_count
g.Commit([]Triple{
{Subject: "张先生", Relation: "喜欢", Object: "Go"},
}, "session", 0)
n, err := g.MergeEntities("张先生", "张三")
if err != nil {
t.Fatal(err)
}
if n < 2 {
t.Errorf("expected at least 2 redirected relations, got %d", n)
}
// source 应被改名
result, err := g.Recall(nil, []string{"张先生"}, 1, "")
if err != nil {
t.Fatal(err)
}
if len(result.Entities) > 0 {
t.Error("张先生 should be merged and hidden")
}
// target 的 mention_count 应合并
// 验证 target 还存在（seedEntities 精确查找）
result2, err := g.Recall([]string{"张三"}, nil, 1, "")
if err != nil {
t.Fatal(err)
}
found := false
for _, e := range result2.Entities {
if e.Name == "张三" {
found = true
if e.MentionCount < 2 {
t.Errorf("expected 张三 mention_count >= 2 after merge, got %d", e.MentionCount)
}
break
}
}
if !found {
t.Error("张三 should still exist after merge")
}
}
func TestMergeEntitiesSelf(t *testing.T) {
g := newTestGraph(t)
defer os.Remove(g.dbPath)
defer g.Close()
g.Commit([]Triple{
{Subject: "张三", Relation: "喜欢", Object: "编程"},
}, "session", 0)
_, err := g.MergeEntities("张三", "张三")
if err == nil {
t.Error("expected error when merging entity with itself")
}
}
func TestMergeEntitiesNonexistent(t *testing.T) {
g := newTestGraph(t)
defer os.Remove(g.dbPath)
defer g.Close()
_, err := g.MergeEntities("不存在", "张三")
if err == nil {
t.Error("expected error for nonexistent source")
}
}
func TestPlaceholders(t *testing.T) {
if placeholders(0) != "NULL" {
t.Errorf("expected NULL for n=0, got %s", placeholders(0))

76
internal/plugin/plugin.go
View File

@ -12,6 +12,7 @@ import (
"time"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
)
type PluginType string
@ -147,15 +148,90 @@ type Registry struct {
plugins map[string]Plugin
ioMgr *agentIO.IOManager
factories map[string]NativeFactory // 名称匹配的插件使用原生实现
sdkAPIs map[string]*sdkAPI // SDK 插件 API 实例
}
type sdkAPI struct {
api *sdk.PluginAPI
tools map[string]sdk.ToolHandler
stages map[sdk.Stage][]sdk.StageHandler
}
func NewRegistry() *Registry {
return &Registry{
plugins: make(map[string]Plugin),
factories: make(map[string]NativeFactory),
sdkAPIs: make(map[string]*sdkAPI),
}
}
// RegisterPluginAPI 注册一个 SDK 插件 API 实例
func (r *Registry) RegisterPluginAPI(api *sdk.PluginAPI) error {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.sdkAPIs[api.Name]; ok {
return fmt.Errorf("sdk api %s already registered", api.Name)
}
r.sdkAPIs[api.Name] = &sdkAPI{
api: api,
tools: api.Tools(),
stages: make(map[sdk.Stage][]sdk.StageHandler),
}
for stage := range sdk.AllStages() {
if handlers := api.StageHandlers(stage); len(handlers) > 0 {
r.sdkAPIs[api.Name].stages[stage] = handlers
}
}
log.Printf("[plugin] registered SDK plugin: %s (tools=%d, stages=%d)",
api.Name, len(api.Tools()), len(r.sdkAPIs[api.Name].stages))
return nil
}
// GetAllSDKToolDefs 收集所有 SDK 插件的工具定义
func (r *Registry) GetAllSDKToolDefs() []sdk.ToolDef {
r.mu.RLock()
defer r.mu.RUnlock()
var defs []sdk.ToolDef
for _, sa := range r.sdkAPIs {
for name := range sa.tools {
defs = append(defs, sdk.ToolDef{Name: name})
}
}
return defs
}
// ExecuteSDKTool 执行 SDK 插件工具
func (r *Registry) ExecuteSDKTool(name string, args map[string]interface{}) (interface{}, error) {
r.mu.RLock()
defer r.mu.RUnlock()
for _, sa := range r.sdkAPIs {
if handler, ok := sa.tools[name]; ok {
return handler(args)
}
}
return nil, fmt.Errorf("sdk tool %s not found", name)
}
// GetStageHandlers 获取所有 SDK 插件在指定阶段的处理器
func (r *Registry) GetStageHandlers(stage sdk.Stage) []sdk.StageHandler {
r.mu.RLock()
defer r.mu.RUnlock()
var handlers []sdk.StageHandler
for _, sa := range r.sdkAPIs {
if h, ok := sa.stages[stage]; ok {
handlers = append(handlers, h...)
}
}
return handlers
}
// SDKPluginCount 返回已注册的 SDK 插件数量
func (r *Registry) SDKPluginCount() int {
r.mu.RLock()
defer r.mu.RUnlock()
return len(r.sdkAPIs)
}
// RegisterNative 注册内置原生插件工厂。当从 plugins/ 加载插件时，
// 如果插件名称匹配已注册的工厂，优先使用原生设备注册。
// 例如: r.RegisterNative("qq", onebot.NewDeviceFactory)

169
internal/plugin/sdk/api.go Normal file
View File

@ -0,0 +1,169 @@
package sdk
import "fmt"
type Stage string
const (
StageOnInput Stage = "on_input"
StagePreAction Stage = "pre_action"
StagePostAction Stage = "post_action"
StageBeforeToolcall Stage = "before_toolcall"
StageAfterToolcall Stage = "after_toolcall"
StageBeforeOutput Stage = "before_output"
StageAfterOutput Stage = "after_output"
)
type EventType string
const (
EventRawInput EventType = "raw_input"
EventAgentOutput EventType = "agent_output"
EventToolCall EventType = "tool_call"
EventReasoning EventType = "reasoning"
EventSystem EventType = "system"
EventAll EventType = "*"
)
type Event struct {
Type EventType `json:"type"`
Source string `json:"source"`
Payload map[string]interface{} `json:"payload"`
Timestamp int64 `json:"timestamp"`
}
type MemItem struct {
Content string `json:"content"`
Score float64 `json:"score"`
Source string `json:"source"`
}
type StageContext struct {
RawMessage string
UserID string
GroupID string
ContextMsgs []map[string]interface{}
LLMText string
ToolCalls []ToolCall
ToolResults []ToolResult
FinalText string
Response *string
Phase Stage
Memory []MemItem
Extra map[string]interface{}
}
type ToolCall struct {
ID string `json:"id"`
Name string `json:"name"`
Arguments map[string]interface{} `json:"arguments"`
}
type ToolResult struct {
CallID string `json:"call_id"`
Name string `json:"name"`
Success bool `json:"success"`
Result interface{} `json:"result"`
}
type ToolDef struct {
Name string `json:"name"`
Description string `json:"description"`
Parameters map[string]interface{} `json:"parameters"`
}
type MemoryAPI interface {
Recall(query string, topK int) ([]MemItem, error)
Commit(triples []map[string]string) error
Introspect() (map[string]interface{}, error)
}
type KnowledgeAPI interface {
Search(query string, topK int) ([]MemItem, error)
Create(name, content string) error
List() ([]string, error)
}
type EventHandler func(event *Event)
type StageHandler func(ctx *StageContext) error
type ToolHandler func(args map[string]interface{}) (interface{}, error)
type PluginAPI struct {
Name string
Version string
tools map[string]ToolHandler
stages map[Stage][]StageHandler
events map[EventType][]EventHandler
eventBus EventBus
memAPI MemoryAPI
knowAPI KnowledgeAPI
}
func NewPluginAPI(name, version string, bus EventBus, mem MemoryAPI, know KnowledgeAPI) *PluginAPI {
return &PluginAPI{
Name: name,
Version: version,
tools: make(map[string]ToolHandler),
stages: make(map[Stage][]StageHandler),
events: make(map[EventType][]EventHandler),
eventBus: bus,
memAPI: mem,
knowAPI: know,
}
}
func (p *PluginAPI) RegisterTool(name string, handler ToolHandler) error {
if _, ok := p.tools[name]; ok {
return fmt.Errorf("tool %s already registered by plugin %s", name, p.Name)
}
p.tools[name] = handler
if p.eventBus != nil {
p.eventBus.Publish(&Event{
Type: EventSystem,
Source: p.Name,
Payload: map[string]interface{}{"action": "register_tool", "tool": name},
})
}
return nil
}
func (p *PluginAPI) RegisterStage(stage Stage, handler StageHandler) {
p.stages[stage] = append(p.stages[stage], handler)
}
func (p *PluginAPI) Subscribe(eventType EventType, handler EventHandler) {
p.events[eventType] = append(p.events[eventType], handler)
if p.eventBus != nil {
p.eventBus.Subscribe(eventType, handler)
}
}
func (p *PluginAPI) Publish(evt *Event) {
if p.eventBus != nil {
p.eventBus.Publish(evt)
}
}
func (p *PluginAPI) Tools() map[string]ToolHandler {
return p.tools
}
func (p *PluginAPI) StageHandlers(stage Stage) []StageHandler {
return p.stages[stage]
}
func (p *PluginAPI) Memory() MemoryAPI { return p.memAPI }
func (p *PluginAPI) Knowledge() KnowledgeAPI { return p.knowAPI }
func AllStages() map[Stage]bool {
return map[Stage]bool{
StageOnInput: true,
StagePreAction: true,
StagePostAction: true,
StageBeforeToolcall: true,
StageAfterToolcall: true,
StageBeforeOutput: true,
StageAfterOutput: true,
}
}

42
internal/plugin/sdk/bus.go Normal file
View File

@ -0,0 +1,42 @@
package sdk
import "fmt"
type EventBus interface {
Publish(event *Event)
Subscribe(eventType EventType, handler EventHandler) func()
}
type InProcessBus struct {
subs map[EventType][]EventHandler
}
func NewInProcessBus() *InProcessBus {
return &InProcessBus{
subs: make(map[EventType][]EventHandler),
}
}
func (b *InProcessBus) Publish(evt *Event) {
for _, h := range b.subs[EventAll] {
h(evt)
}
if evt.Type != EventAll {
for _, h := range b.subs[evt.Type] {
h(evt)
}
}
}
func (b *InProcessBus) Subscribe(eventType EventType, handler EventHandler) func() {
b.subs[eventType] = append(b.subs[eventType], handler)
return func() {
list := b.subs[eventType]
for i, h := range list {
if fmt.Sprintf("%p", h) == fmt.Sprintf("%p", handler) {
b.subs[eventType] = append(list[:i], list[i+1:]...)
break
}
}
}
}

133
internal/plugin/sdk/bus_test.go Normal file
View File

@ -0,0 +1,133 @@
package sdk
import (
"testing"
)
func TestInProcessBus(t *testing.T) {
bus := NewInProcessBus()
var called bool
bus.Subscribe(EventRawInput, func(evt *Event) {
called = true
if evt.Source != "test" {
t.Errorf("expected source test, got %s", evt.Source)
}
})
bus.Publish(&Event{
Type: EventRawInput,
Source: "test",
})
if !called {
t.Error("handler was not called")
}
}
func TestInProcessBusWildcard(t *testing.T) {
bus := NewInProcessBus()
count := 0
bus.Subscribe(EventAll, func(evt *Event) {
count++
})
bus.Publish(&Event{Type: EventRawInput, Source: "s1"})
bus.Publish(&Event{Type: EventToolCall, Source: "s2"})
if count != 2 {
t.Errorf("expected 2, got %d", count)
}
}
func TestPluginAPI(t *testing.T) {
bus := NewInProcessBus()
api := NewPluginAPI("test", "1.0.0", bus, nil, nil)
if api.Name != "test" {
t.Errorf("expected test, got %s", api.Name)
}
if api.Version != "1.0.0" {
t.Errorf("expected 1.0.0, got %s", api.Version)
}
var stageCalled bool
api.RegisterStage(StageOnInput, func(ctx *StageContext) error {
stageCalled = true
if ctx.RawMessage != "hello" {
t.Errorf("expected hello, got %s", ctx.RawMessage)
}
return nil
})
ctx := &StageContext{RawMessage: "hello"}
for _, handler := range api.StageHandlers(StageOnInput) {
handler(ctx)
}
if !stageCalled {
t.Error("stage handler was not called")
}
}
func TestPluginAPITool(t *testing.T) {
api := NewPluginAPI("test", "1.0.0", nil, nil, nil)
err := api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) {
return "ok", nil
})
if err != nil {
t.Fatalf("register tool: %v", err)
}
if _, ok := api.Tools()["test_tool"]; !ok {
t.Error("tool not found")
}
// duplicate registration should fail
err = api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) {
return "ok", nil
})
if err == nil {
t.Error("expected error on duplicate tool registration")
}
}
func TestPluginAPIStageShortCircuit(t *testing.T) {
api := NewPluginAPI("test", "1.0.0", nil, nil, nil)
api.RegisterStage(StageOnInput, func(ctx *StageContext) error {
resp := "intercepted"
ctx.Response = &resp
return nil
})
ctx := &StageContext{RawMessage: "hello"}
handlers := api.StageHandlers(StageOnInput)
if len(handlers) != 1 {
t.Fatalf("expected 1 handler, got %d", len(handlers))
}
handlers[0](ctx)
if ctx.Response == nil || *ctx.Response != "intercepted" {
t.Errorf("expected intercepted, got %v", ctx.Response)
}
}
func TestAllStages(t *testing.T) {
stages := AllStages()
expected := []Stage{
StageOnInput, StagePreAction, StagePostAction,
StageBeforeToolcall, StageAfterToolcall,
StageBeforeOutput, StageAfterOutput,
}
for _, s := range expected {
if !stages[s] {
t.Errorf("missing stage: %s", s)
}
}
if len(stages) != len(expected) {
t.Errorf("expected %d stages, got %d", len(expected), len(stages))
}
}