AskaEth
eb1828ee7a
- 只翻注释/文档字符串：剥离注释后代码行逐字节不变（机械校验只剩允许的行尾注释形态） - 裁决编号与 design 节号（02.4.3 / 11.4.2 / design/05 §9 等）逐字保留 - 复核：go build 绿；orchestration 及其子包（audit/comm）测试绿；gofmt 干净
166 lines
6.2 KiB
Go
166 lines
6.2 KiB
Go
// Package comm 为 Cyrene v2 编排层实现 agent 间通信
|
||
// （DevDocs/design/05-子代理.md §4）。子代理与主 agent 经进程内总线（internal/bus）
|
||
// 交换消息，所用请求/应答协议建立在三种消息之上：ask、reply 与 event。
|
||
//
|
||
// # 协议
|
||
//
|
||
// 一个 Probe 绑定到一个 agent id。它可以向另一个 agent 提问并等待其回复
|
||
// （ask/reply），也可以向另一个 agent 或任何监听者发射事件（event）。消息以点分隔的
|
||
// subject 发布到总线上，因此总线类似 NATS 的通配匹配能把它们路由到正确的 agent；
|
||
// 见下方「Subject 约定」一节。
|
||
//
|
||
// # Subject 约定
|
||
//
|
||
// Subject 由经净化的 agent id 拼成（凡不在 [A-Za-z0-9_-] 之内的字符都被替换为 "_"，
|
||
// 因此 id 绝不会破坏 subject 词元）：
|
||
//
|
||
// ask: cyrene.agent.<from>.<to>.ask
|
||
// reply: cyrene.agent.<from>.<to>.reply.<askId>
|
||
// event (to): cyrene.agent.<from>.<to>.event
|
||
// event (bc): cyrene.agent.<from>.event
|
||
//
|
||
// 回复 subject 把 ask id 作为最后一个词元，给每个在途 ask 一个私有回复地址，因此同一
|
||
// agent 的并发 ask 的回复绝不串线。要接收发给自己的 ask，agent 订阅
|
||
// cyrene.agent.*.<me>.ask。要接收事件（广播与发给自己的事件），它订阅
|
||
// cyrene.agent.*.event 与 cyrene.agent.*.<me>.event。
|
||
//
|
||
// # 投递语义
|
||
//
|
||
// Ask 阻塞到收到回复、ask 超时到期，或传入的上下文被取消；前一种情况返回
|
||
// ErrTimeout，后一种情况返回包装后的上下文错误。Reply 与各事件发送方是即发即忘的
|
||
// 发布，返回总线错误（若有）。处理回调在总线的按订阅工作 goroutine 上调用，因此一个
|
||
// EventHandler 可能被两个订阅并发调用，必须可安全并发使用；而某个 agent 的
|
||
// AskHandler 由总线串行化。
|
||
package comm
|
||
|
||
import (
|
||
"errors"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// Kind 是 agent 间消息的线上类别。
|
||
type Kind string
|
||
|
||
const (
|
||
// KindAsk 是一个 agent 向另一个 agent 提的问题，由一条 reply 作答。
|
||
KindAsk Kind = "ask"
|
||
// KindReply 是对此前发布的 ask 的应答。
|
||
KindReply Kind = "reply"
|
||
// KindEvent 是单向通知，可为广播或定向。
|
||
KindEvent Kind = "event"
|
||
)
|
||
|
||
// subjectPrefix 是每个 agent 间 subject 固定的前两个词元。
|
||
const subjectPrefix = "cyrene.agent"
|
||
|
||
// DefaultAskTimeout 限制调用方上下文未携带更短期限时 Ask 等待回复的时长。它约束一个
|
||
// 请求/应答往返，使沉默的 agent 永不吊死其调用方。
|
||
const DefaultAskTimeout = 30 * time.Second
|
||
|
||
// 本包返回的哨兵错误。用 errors.Is 判断它们；它们被直接返回或以 %w 包装返回，因此底层
|
||
// 原因始终可得。
|
||
var (
|
||
// ErrTimeout 在 ask 期限（调用方上下文期限或 DefaultAskTimeout）到期前没有回复
|
||
// 到达时由 Ask 返回。
|
||
ErrTimeout = errors.New("comm: reply timeout")
|
||
|
||
// ErrNoReply 在 ask 无法投递（总线拒绝了发布/回复订阅）时由 Ask 返回，此时不应
|
||
// 期待任何回复。
|
||
ErrNoReply = errors.New("comm: no reply")
|
||
)
|
||
|
||
// Ask 是一个 agent 发给另一个 agent 的问题。
|
||
type Ask struct {
|
||
// ID 在该问答交互内唯一标识本次 ask。
|
||
ID string
|
||
// From 是提问 agent（发送方）的 id。
|
||
From string
|
||
// To 是应当作答的 agent（接收方）的 id。
|
||
To string
|
||
// Question 是人类可读的问题文本。
|
||
Question string
|
||
// Data 承载随问题一起发送的可选结构化载荷。
|
||
Data any
|
||
}
|
||
|
||
// Reply 是对此前发布的 ask 的应答。
|
||
type Reply struct {
|
||
// AskID 是本次回复所应答的 ask 的 ID。
|
||
AskID string
|
||
// From 是回复 agent（发送方）的 id。
|
||
From string
|
||
// To 是提问 agent（接收方）的 id。
|
||
To string
|
||
// Text 是人类可读的应答文本。
|
||
Text string
|
||
// Data 承载随应答一起发送的可选结构化载荷。
|
||
Data any
|
||
}
|
||
|
||
// Event 是单向通知，或广播，或定向给某个 agent。它没有回复。
|
||
type Event struct {
|
||
// From 是发射事件的 agent 的 id。
|
||
From string
|
||
// Name 是简短的事件名，例如 "task.progress" 或 "tool.used"。
|
||
Name string
|
||
// Data 承载可选的结构化载荷。
|
||
Data any
|
||
}
|
||
|
||
// token 净化一个 agent id，使其可用作单个 subject 词元：凡不在 [A-Za-z0-9_-] 之内的
|
||
// 字符都被替换为 "_"。这既让该 id 对路由仍然有用，又保证点号（以及其他 subject 元字符）
|
||
// 绝不会漏进词元。
|
||
func token(s string) string {
|
||
var b strings.Builder
|
||
for _, r := range s {
|
||
switch {
|
||
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
|
||
b.WriteRune(r)
|
||
default:
|
||
b.WriteRune('_')
|
||
}
|
||
}
|
||
return b.String()
|
||
}
|
||
|
||
// subjectFor 构建给定类别的 from/to 消息的基础 subject。
|
||
func subjectFor(from, to string, kind Kind) string {
|
||
return subjectPrefix + "." + token(from) + "." + token(to) + "." + string(kind)
|
||
}
|
||
|
||
// askSubject 是 from→to 的 ask 所发布到的 subject。
|
||
func askSubject(from, to string) string {
|
||
return subjectFor(from, to, KindAsk)
|
||
}
|
||
|
||
// replySubject 是应答 askID 的 from→to 回复所发布到的私有 subject。末尾的 askID 词元
|
||
// 给每个在途 ask 自己的地址，因此并发 ask 之间回复绝不串线。
|
||
func replySubject(from, to, askID string) string {
|
||
return subjectPrefix + "." + token(from) + "." + token(to) + "." + string(KindReply) + "." + token(askID)
|
||
}
|
||
|
||
// eventSubject 是 from→to 定向事件的 subject。
|
||
func eventSubject(from, to string) string {
|
||
return subjectFor(from, to, KindEvent)
|
||
}
|
||
|
||
// broadcastEventSubject 是 from 发射的广播事件的 subject。
|
||
func broadcastEventSubject(from string) string {
|
||
return subjectPrefix + "." + token(from) + "." + string(KindEvent)
|
||
}
|
||
|
||
// askReceivePattern 是名为 me 的 agent 用来接收任意发送方发给它的 ask 的订阅模式。
|
||
func askReceivePattern(me string) string {
|
||
return subjectPrefix + ".*." + token(me) + "." + string(KindAsk)
|
||
}
|
||
|
||
// eventReceivePatterns 是名为 me 的 agent 用来接收它应当看到的所有事件的订阅模式：
|
||
// 任意发送方的广播，以及专门定向给它的事件。
|
||
func eventReceivePatterns(me string) []string {
|
||
return []string{
|
||
subjectPrefix + ".*." + string(KindEvent),
|
||
subjectPrefix + ".*." + token(me) + "." + string(KindEvent),
|
||
}
|
||
}
|