From 81bfdfce1da0be82ee686e1f7bfd61532f76100f Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 26 Aug 2026 19:16:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(examples):=20a2a/acp=20=E5=9B=9E=E5=A4=8D?= =?UTF-8?q?=E9=97=AD=E7=8E=AF=20+=20=E4=BC=9A=E8=AF=9D=E5=BB=B6=E7=BB=AD?= =?UTF-8?q?=20+=20=E5=90=8C=E6=AD=A5=E6=B3=A8=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 入站请求从 InjectInterruptText(202 submitted) 改为 InjectInputSync 同步等待回复，直接返回回复文本；支持 params.session_id 延续多轮 上下文；注册为输出通道让回复有落点。详见 TrueAgent 仓库同名 commit。 --- example/a2a/plg.json | 2 +- example/a2a/plugin.go | 111 +++++++++++++++++++++++++++++++++++++++--- example/acp/plg.json | 8 ++- example/acp/plugin.go | 73 +++++++++++++++++++++++---- 4 files changed, 174 insertions(+), 20 deletions(-) diff --git a/example/a2a/plg.json b/example/a2a/plg.json index e1938ea..f1bf996 100644 --- a/example/a2a/plg.json +++ b/example/a2a/plg.json @@ -2,7 +2,7 @@ "name": "a2a", "name_zh": "A2A 代理通信", "name_en": "A2A Agent Communication", - "version": "1.1.0", + "version": "1.2.0", "description": "Agent-to-Agent 协议通信插件，支持双向 A2A 通信：可查询其他 Agent 并回复其请求。提供 HTTP 服务端暴露本 Agent 能力。", "author": "HomeAgent", "entry": "plugin.so", diff --git a/example/a2a/plugin.go b/example/a2a/plugin.go index 8732971..f6ca462 100644 --- a/example/a2a/plugin.go +++ b/example/a2a/plugin.go @@ -21,15 +21,47 @@ type Plugin struct { srvMu sync.Mutex server *http.Server serverAddr string + + // 会话表：session_id → 上下文前缀。A2A 无状态协议下由插件侧维护 + // 多轮上下文：同 session 的后续请求会把之前的对话拼进注入文本。 + sessMu sync.Mutex + sessions map[string]*a2aSession } +// a2aSession 记录一个会话的轮次历史，用于延续上下文。 +type a2aSession struct { + ID string + History []string // 轮次文本 [user1, agent1, user2, agent2, ...] + LastUsed time.Time +} + +// maxSessionTurns 单会话保留的最大轮次对数（防上下文无限膨胀）。 +const maxSessionTurns = 10 + +// sessionGCPeriod 会话过期清理周期；超过 2 小时未用的会话回收。 +const sessionGCPeriod = 30 * time.Minute + func (p *Plugin) Name() string { return p.name } func (p *Plugin) Start(s *sdk.PluginSDK) error { s.SetAutoRestart(true) p.sdk = s + p.sessions = make(map[string]*a2aSession) tp := p.name + "_" + // 注册自身为输出通道：agent 回复 emit 到本通道时有落点， + // 且 output_list_channels 可见（agent 能主动向 a2a 会话推送消息）。 + if err := s.RegisterOutputChannel(p.name, 1, "A2A Agent 互联通道（外部 agent 查询的回复由此返回）", sdk.ChannelDef{}, func(args map[string]interface{}) (interface{}, error) { + payload, _ := args["payload"].(string) + log.Printf("[%s] channel output: %s", p.name, truncateRunes(payload, 120)) + return map[string]interface{}{"status": "ok"}, nil + }); err != nil { + log.Printf("[%s] register output channel: %v", p.name, err) + } + + // 会话 GC：后台周期回收长期不用的会话 + go p.sessionGCLoop() + s.Settings().RegisterDef(sdk.ConfigDef{ Key: "listen", Default: "127.0.0.1:12000", Type: "string", DisplayName: "监听地址", @@ -114,6 +146,29 @@ func (p *Plugin) Stop() error { return nil } +// sessionGCLoop 周期清理超时会话。 +func (p *Plugin) sessionGCLoop() { + ticker := time.NewTicker(sessionGCPeriod) + defer ticker.Stop() + for range ticker.C { + p.sessMu.Lock() + for id, sess := range p.sessions { + if time.Since(sess.LastUsed) > 2*time.Hour { + delete(p.sessions, id) + } + } + p.sessMu.Unlock() + } +} + +func truncateRunes(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + "..." +} + func (p *Plugin) stopServer() { p.srvMu.Lock() defer p.srvMu.Unlock() @@ -191,7 +246,8 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) { ID string `json:"id"` Method string `json:"method"` Params struct { - Query string `json:"query,omitempty"` + Query string `json:"query,omitempty"` + SessionID string `json:"session_id,omitempty"` Message *struct { Role string `json:"role"` Parts []struct { @@ -215,19 +271,60 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) { } queryText = strings.TrimSpace(queryText) } - - // Inject into agent pipeline via interrupt (preempt current processing) or direct input - if queryText != "" { - p.sdk.InjectInterruptText("a2a", "webui", fmt.Sprintf("[来自A2A Agent的查询]\n%s", queryText)) + if queryText == "" { + http.Error(w, "query/message.text required", http.StatusBadRequest) + return } - // Respond with task accepted + // 会话：调用方可指定 session_id 延续多轮上下文；不指定则新建。 + sessionID := strings.TrimSpace(req.Params.SessionID) + injectText := queryText + p.sessMu.Lock() + if sessionID != "" { + sess := p.sessions[sessionID] + if sess == nil { + sess = &a2aSession{ID: sessionID, LastUsed: time.Now()} + p.sessions[sessionID] = sess + } + sess.LastUsed = time.Now() + // 有历史则把上下文拼在前面（截尾防爆量） + if len(sess.History) > 0 { + ctxText := strings.Join(sess.History, "\n") + injectText = "[对话上下文]\n" + ctxText + "\n[本轮输入]\n" + queryText + } + } else { + sessionID = fmt.Sprintf("a2a_%d", time.Now().UnixNano()) + p.sessions[sessionID] = &a2aSession{ID: sessionID, LastUsed: time.Now()} + } + p.sessMu.Unlock() + + // 同步注入：阻塞等待 agent 处理完成拿回复（不再抢占打断、 + // 也不再回 202 让请求方永远等不到结果）。HTTP 超时由调用方控制。 + reply := p.sdk.InjectInputSync(p.name, p.name, + fmt.Sprintf("[来自A2A Agent的查询 session=%s]\n%s\n[注意] 请直接以文本回复本查询，不要调用 output_send__%s——你的最终文本回复会被系统自动返回给请求方。", sessionID, injectText, p.name)) + + // 回复写回会话历史（下一轮作为上下文） + p.sessMu.Lock() + if sess := p.sessions[sessionID]; sess != nil { + sess.History = append(sess.History, "用户: "+queryText, "助手: "+reply) + if len(sess.History) > maxSessionTurns*2 { + sess.History = sess.History[len(sess.History)-maxSessionTurns*2 :] + } + sess.LastUsed = time.Now() + } + p.sessMu.Unlock() + resp := map[string]interface{}{ "jsonrpc": "2.0", "id": req.ID, "result": map[string]interface{}{ "id": fmt.Sprintf("task_%d", time.Now().UnixNano()), - "status": "submitted", + "status": "completed", + "session_id": sessionID, + "message": map[string]interface{}{ + "role": "agent", + "parts": []map[string]string{{"type": "text", "text": reply}}, + }, }, } w.Header().Set("Content-Type", "application/json") diff --git a/example/acp/plg.json b/example/acp/plg.json index 52df12c..f67c8b8 100644 --- a/example/acp/plg.json +++ b/example/acp/plg.json @@ -2,11 +2,15 @@ "name": "acp", "name_zh": "ACP 代理通信", "name_en": "ACP Agent Client Protocol", - "version": "1.0.0", + "version": "1.1.0", "description": "Agent Client Protocol 通信插件：充当 ACP 服务端接受其他 Agent 的任务请求，同时提供客户端工具向远程 ACP Agent（如 opencode）发起会话并读取回复", "author": "HomeAgent", "entry": "plugin.so", - "tags": ["acp", "agent", "interop"], + "tags": [ + "acp", + "agent", + "interop" + ], "targets": "linux/amd64", "outdir": "dist", "bundle": true, diff --git a/example/acp/plugin.go b/example/acp/plugin.go index 7fb7d69..7c91bc7 100644 --- a/example/acp/plugin.go +++ b/example/acp/plugin.go @@ -34,8 +34,13 @@ type Plugin struct { type sessionState struct { ID string Replying []map[string]interface{} + History []string // 轮次历史 [user, agent, user, agent...]，延续上下文用 + LastUsed time.Time } +// maxSessionTurns 单会话保留的最大轮次对数。 +const maxSessionTurns = 10 + func (p *Plugin) Name() string { return p.name } func (p *Plugin) Start(s *sdk.PluginSDK) error { @@ -44,6 +49,14 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { p.sessions = make(map[string]*sessionState) tp := p.name + "_" + // 注册自身为输出通道：agent 回复 emit 到本通道时有落点。 + // （回复主要走同步注入返回，此通道用于 agent 主动 output_send__acp） + s.RegisterOutputChannel(p.name, 1, "ACP Agent 互联通道（外部 agent 会话的回复由此返回）", sdk.ChannelDef{}, func(args map[string]interface{}) (interface{}, error) { + payload, _ := args["payload"].(string) + log.Printf("[%s] channel output: %s", p.name, truncateStr(payload, 120)) + return map[string]interface{}{"status": "ok"}, nil + }) + s.Settings().RegisterDef(sdk.ConfigDef{ Key: "listen", Default: "127.0.0.1:12001", Type: "string", DisplayName: "监听地址", @@ -190,21 +203,56 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) { return } - sid := fmt.Sprintf("session_%d", time.Now().UnixNano()) + // 会话：调用方可指定 session_id 延续多轮；不指定则新建。 + sid := strings.TrimSpace(req.Params.SessionID) p.mu.Lock() - p.sessions[sid] = &sessionState{ID: sid} + if sid != "" { + if _, exists := p.sessions[sid]; !exists { + p.sessions[sid] = &sessionState{ID: sid, LastUsed: time.Now()} + } + } else { + sid = fmt.Sprintf("session_%d", time.Now().UnixNano()) + p.sessions[sid] = &sessionState{ID: sid, LastUsed: time.Now()} + } + st := p.sessions[sid] p.mu.Unlock() - if p.sdk != nil { - p.sdk.InjectInterruptText(p.name, "acp", - fmt.Sprintf("[来自ACP Agent的请求请求 session %s]\n%s", sid, text)) + // 延续上下文 + injectText := text + p.mu.Lock() + if len(st.History) > 0 { + ctxText := strings.Join(st.History, "\n") + injectText = "[对话上下文]\n" + ctxText + "\n[本轮输入]\n" + text } + p.mu.Unlock() + + // 同步注入等待回复：不抢占打断，完整闭环返回文本。 + reply := "" + if p.sdk != nil { + reply = p.sdk.InjectInputSync(p.name, p.name, + fmt.Sprintf("[来自ACP Agent的请求 session %s]\n%s\n[注意] 请直接以文本回复本请求，不要调用 output_send__%s——你的最终文本回复会被系统自动返回给请求方。", sid, injectText, p.name)) + } + + // 写回历史 + 填充 Replying 供 SSE 消费 + p.mu.Lock() + st.History = append(st.History, "用户: "+text, "助手: "+reply) + if len(st.History) > maxSessionTurns*2 { + st.History = st.History[len(st.History)-maxSessionTurns*2:] + } + st.LastUsed = time.Now() + if reply != "" { + st.Replying = append(st.Replying, map[string]interface{}{ + "type": "reply", "text": reply, + }) + } + p.mu.Unlock() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ "jsonrpc": "2.0", "id": req.ID, "result": map[string]interface{}{ "session": map[string]interface{}{"id": sid}, + "reply": reply, }, }) @@ -212,12 +260,17 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) { sid := req.Params.SessionID p.mu.Lock() st := p.sessions[sid] - if st != nil && req.Params.Final { - st.Replying = append(st.Replying, map[string]interface{}{ - "type": "reply", "text": "done", - }) - } p.mu.Unlock() + if st == nil { + http.Error(w, "session not found", http.StatusNotFound) + return + } + if req.Params.Final { + // 客户端结束会话：标记并保留历史（后续可再 session/new 续） + p.mu.Lock() + st.LastUsed = time.Now() + p.mu.Unlock() + } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{