17 KiB
17 KiB
灵隙协议实现规范：传输模式定义
版本：1.0.0 | 层级：实现层
1. 传输模式枚举
传输模式（TransportMode）定义了任务通信的传输方式。Synapse 协议支持四种传输模式，适配不同的网络环境和业务场景。
1.1 模式定义
// TransportMode 定义任务通信的传输模式
type TransportMode string
const (
// TransportModeCallback - 异步回调模式
// 任务完成后通过 Webhook 回调通知结果
TransportModeCallback TransportMode = "callback"
// TransportModePolling - 异步轮询模式
// 调用方主动轮询获取任务状态和结果
TransportModePolling TransportMode = "polling"
// TransportModeSync - 同步模式
// 调用方阻塞等待任务完成，直接返回结果
TransportModeSync TransportMode = "sync"
// TransportModeHybrid - 混合模式
// 优先使用回调，回调失败时启用轮询作为兜底
TransportModeHybrid TransportMode = "hybrid"
)
1.2 模式对比
| 模式 | 枚举值 | 通信方向 | 实时性 | 适用场景 |
|---|---|---|---|---|
| 回调（Callback） | callback |
服务端 → 调用方 | 高 | 网络可达、期望实时通知 |
| 轮询（Polling） | polling |
调用方 → 服务端 | 中 | 调用方主动拉取、服务端不主动推送 |
| 同步（Sync） | sync |
双向阻塞 | 最高 | 短耗时任务、需要即时响应 |
| 混合（Hybrid） | hybrid |
双向 + 兜底 | 高 | 生产环境、容错要求高的场景 |
1.3 模式选择建议
| 条件 | 推荐模式 |
|---|---|
| 任务耗时 < 5s | sync — 直接同步等待 |
| 任务耗时 5s~5min，回调端点可达 | callback — 异步回调 |
| 任务耗时 5s~5min，回调端点不可达 | polling — 轮询拉取 |
| 生产环境，需要高可靠性 | hybrid — 回调 + 轮询兜底 |
| 任务耗时 > 5min | polling 或 hybrid — 长时间异步 |
2. 传输偏好配置
2.1 传输偏好结构
// TransportPreferences 定义任务的传输偏好配置
type TransportPreferences struct {
// Preferred - 首选传输模式
Preferred TransportMode `json:"preferred"`
// Fallback - 兜底传输模式，当首选模式不可用时使用
Fallback TransportMode `json:"fallback,omitempty"`
// CallbackEndpoint - 回调端点 URL
// 仅当 Preferred 或 Fallback 为 callback/hybrid 时生效
CallbackEndpoint string `json:"callback_endpoint,omitempty"`
// PollingInterval - 轮询间隔（秒）
// 仅当 Preferred 或 Fallback 为 polling/hybrid 时生效
PollingInterval int `json:"polling_interval,omitempty"`
// ProgressUpdates - 是否接收进度更新
// 为 true 时，服务端会在任务执行过程中推送进度事件
ProgressUpdates bool `json:"progress_updates"`
// ProgressInterval - 进度更新最小间隔（秒）
// 控制进度通知的频率，避免频繁推送
ProgressInterval int `json:"progress_interval,omitempty"`
}
2.2 字段说明
| 字段 | 类型 | 必需 | 默认值 | 说明 |
|---|---|---|---|---|
preferred |
TransportMode |
是 | callback |
首选传输模式 |
fallback |
TransportMode |
否 | polling |
兜底模式，首选不可用时回退 |
callback_endpoint |
string |
条件必填 | "" |
回调 URL，callback/hybrid 模式必需 |
polling_interval |
int |
否 | 5 |
轮询间隔（秒），最小 1 |
progress_updates |
bool |
否 | true |
是否启用进度推送 |
progress_interval |
int |
否 | 10 |
进度更新间隔（秒） |
2.3 默认配置
// DefaultTransportPreferences 返回默认传输偏好配置
//
// 默认值说明：
// - 首选回调模式（callback），兜底轮询（polling）
// - 轮询间隔 5 秒
// - 启用进度更新，最小间隔 10 秒
func DefaultTransportPreferences() *TransportPreferences {
return &TransportPreferences{
Preferred: TransportModeCallback,
Fallback: TransportModePolling,
PollingInterval: 5,
ProgressUpdates: true,
ProgressInterval: 10,
}
}
2.4 配置示例
纯回调模式：
prefs := &TransportPreferences{
Preferred: TransportModeCallback,
CallbackEndpoint: "https://my-service.com/callback",
ProgressUpdates: true,
ProgressInterval: 5,
}
纯轮询模式：
prefs := &TransportPreferences{
Preferred: TransportModePolling,
PollingInterval: 3,
ProgressUpdates: false,
}
混合模式（生产推荐）：
prefs := &TransportPreferences{
Preferred: TransportModeCallback,
Fallback: TransportModePolling,
CallbackEndpoint: "https://my-service.com/callback",
PollingInterval: 5,
ProgressUpdates: true,
ProgressInterval: 10,
}
同步模式：
prefs := &TransportPreferences{
Preferred: TransportModeSync,
}
3. 轮询 API 类型
轮询 API 为调用方提供主动拉取任务状态的能力。所有 Agent 端必须实现以下类型以支持轮询传输模式。
3.1 任务状态响应
// TaskStatusResponse 任务状态轮询响应
//
// 调用方通过此结构获取单个任务的当前状态。
// 成功时 Success 为 true，Task 字段包含任务详情；
// 失败时 Success 为 false，Error 字段包含错误信息。
type TaskStatusResponse struct {
Success bool `json:"success"`
Task *TaskInfo `json:"task,omitempty"`
Error *SynapseError `json:"error,omitempty"`
}
3.2 任务信息
// TaskInfo 任务详细信息
//
// 描述一个异步任务的完整状态，包括进度、结果和
// 错误信息。ETag 字段用于增量查询，避免重复传输
// 未变更的数据。
type TaskInfo struct {
// ID - 任务唯一标识
ID string `json:"id"`
// Type - 任务类型，如 "text_generation", "image_analysis"
Type string `json:"type"`
// Status - 任务状态
// 可选值: pending, running, completed, failed, cancelled
Status string `json:"status"`
// Progress - 任务进度百分比（0-100）
Progress int `json:"progress"`
// Message - 当前状态的人类可读描述
Message string `json:"message,omitempty"`
// CreatedAt - 任务创建时间
CreatedAt Timestamp `json:"created_at"`
// UpdatedAt - 任务最后更新时间
UpdatedAt Timestamp `json:"updated_at"`
// ETA - 预计完成时间（可选）
ETA *Timestamp `json:"eta,omitempty"`
// Result - 任务执行结果
// 当 Status 为 completed 时此字段有效
Result any `json:"result,omitempty"`
// Error - 任务执行错误
// 当 Status 为 failed 时此字段有效
Error *SynapseError `json:"error,omitempty"`
// ETag - 实体标签，用于增量查询和缓存
// 客户端可在后续请求中携带此值以获取变更数据
ETag string `json:"etag,omitempty"`
}
3.3 任务状态枚举
| 状态值 | 说明 | 可转换至 |
|---|---|---|
pending |
任务已提交，等待执行 | running |
running |
任务正在执行中 | completed, failed, cancelled |
completed |
任务执行成功，Result 可用 | —（终态） |
failed |
任务执行失败，Error 可用 | —（终态） |
cancelled |
任务已被取消 | —（终态） |
3.4 任务列表响应
// TaskListResponse 任务列表响应
//
// 支持分页查询，调用方可使用 Limit 和 Offset
// 参数控制返回结果的数量和偏移量。
type TaskListResponse struct {
Success bool `json:"success"`
Tasks []TaskInfo `json:"tasks"`
Total int `json:"total"`
Limit int `json:"limit"`
Offset int `json:"offset"`
HasMore bool `json:"has_more"`
}
分页参数说明：
| 字段 | 类型 | 说明 |
|---|---|---|
total |
int |
符合条件的任务总数 |
limit |
int |
本次请求的限制数量 |
offset |
int |
本次请求的偏移量 |
has_more |
bool |
是否还有更多数据 |
3.5 任务事件
// TaskEvent 任务事件
//
// 表示任务执行过程中产生的事件，用于事件流推送
// 和增量状态更新。
type TaskEvent struct {
// ID - 事件唯一标识
ID string `json:"id"`
// Type - 事件类型
// 可选值: status_update, progress, log, error, complete
Type string `json:"type"`
// Timestamp - 事件发生时间
Timestamp Timestamp `json:"timestamp"`
// Data - 事件数据
// 不同类型的事件携带不同的数据结构
Data any `json:"data"`
}
3.6 事件类型说明
| 事件类型 | 触发时机 | Data 内容 |
|---|---|---|
status_update |
任务状态变更 | 包含新状态和状态变更原因 |
progress |
进度更新 | 包含当前进度值（0-100）和描述 |
log |
任务产生日志输出 | 日志文本内容 |
error |
任务执行出错 | 错误代码和错误消息 |
complete |
任务完成 | 任务执行结果 |
3.7 任务事件流响应
// TaskEventsResponse 任务事件流响应
//
// 支持基于游标的分页，调用方可使用 Next 游标
// 持续拉取后续事件。
type TaskEventsResponse struct {
Success bool `json:"success"`
Since string `json:"since"`
Events []TaskEvent `json:"events"`
HasMore bool `json:"has_more"`
Next string `json:"next_cursor,omitempty"`
}
游标使用流程：
- 首次请求不带
since参数，获取当前所有事件 - 响应中返回
next_cursor作为下一页游标 - 后续请求携带
since=next_cursor获取后续事件 - 当
has_more为false时，表示已无新事件
3.8 轮询 API 端点约定
| 端点 | 方法 | 请求参数 | 响应类型 | 说明 |
|---|---|---|---|---|
/v1/tasks/{taskID} |
GET | — | TaskStatusResponse |
查询单个任务状态 |
/v1/tasks |
GET | limit, offset, status |
TaskListResponse |
查询任务列表 |
/v1/tasks/{taskID}/events |
GET | since |
TaskEventsResponse |
查询任务事件流 |
4. 流式传输类型
流式传输用于实时传输大文件或持续生成的数据，如音频流、视频流、长文本生成等场景。
4.1 流数据块
// StreamChunk 流式传输数据块
//
// 每次传输一个数据块，Sequence 字段保证数据块
// 的顺序。当 Final 为 true 时，表示流传输结束，
// 客户端可开始组装完整数据。
type StreamChunk struct {
// TaskID - 任务 ID，标识所属任务
TaskID string `json:"task_id"`
// Sequence - 块序列号，从 1 开始递增
// 用于客户端排序和校验丢包
Sequence int64 `json:"sequence"`
// ChunkType - 数据块类型
// 可选值: text, binary, audio, video
ChunkType string `json:"chunk_type"`
// Data - 数据块内容
// 二进制数据使用 base64 编码传输
Data string `json:"data"`
// Final - 是否为最后一个数据块
// 为 true 时表示流传输结束
Final bool `json:"final"`
// Timestamp - 数据块时间戳
Timestamp Timestamp `json:"timestamp"`
}
4.2 数据块类型说明
| ChunkType | 说明 | Data 编码 | 适用场景 |
|---|---|---|---|
text |
文本数据 | UTF-8 字符串 | 流式文本生成、日志输出 |
binary |
二进制数据 | Base64 | 文件传输、二进制协议数据 |
audio |
音频数据 | Base64 | 实时语音、音频流处理 |
video |
视频数据 | Base64 | 实时视频、视频流处理 |
4.3 流完成通知
// StreamComplete 流传输完成通知
//
// 当所有数据块传输完毕后发送，包含完整数据的
// 校验和，供客户端验证数据完整性。
type StreamComplete struct {
// TaskID - 任务 ID
TaskID string `json:"task_id"`
// TotalChunks - 总数据块数
// 客户端可据此验证是否收到所有数据块
TotalChunks int64 `json:"total_chunks"`
// FullContentChecksum - 完整数据校验和
// 格式: "sha256:<hex>"，用于完整性校验
FullContentChecksum string `json:"full_content_checksum"`
// Result - 任务执行结果（可选）
// 流传输完成后携带的最终结果
Result any `json:"result,omitempty"`
}
4.4 流式传输生命周期
调用方 服务端
| |
|-- POST /v1/tasks/submit --->|
| (TransportMode: sync) | 启动异步任务
| |
|<--- StreamChunk{seq:1} -----| 数据块 1
|<--- StreamChunk{seq:2} -----| 数据块 2
|<--- StreamChunk{seq:3} -----| 数据块 3
|<--- StreamChunk{seq:4} -----| 数据块 4（Final=true）
|<--- StreamComplete ---------| 完成通知
| |
客户端收到 StreamComplete 后：
1. 确认 TotalChunks == 实际收到的块数
2. 计算完整数据的 SHA256 校验和
3. 对比 FullContentChecksum 验证完整性
4. 读取 Result 获取任务执行结果
4.5 校验和验证
// ValidateStreamChecksum 验证流式传输数据的完整性
//
// 客户端在收到 StreamComplete 后应调用此函数
// 验证收到的数据是否完整。
func ValidateStreamChecksum(receivedChecksum, computedChecksum string) bool {
return receivedChecksum == computedChecksum
}
4.6 传输模式与流式传输的组合规则
| 传输模式 | 是否支持流式 | 说明 |
|---|---|---|
callback |
否 | 回调模式仅通知结果，不承载流数据 |
polling |
否 | 轮询模式通过分块获取，非实时流 |
sync |
是 | 同步模式下可实时流式返回数据 |
hybrid |
是（回调阶段） | 混合模式中回调路径支持流式 |
5. 传输模式交互流程
5.1 回调模式（Callback）
调用方 Agent
| |
|--- SubmitTask ------------>| 提交任务
| (callback_endpoint: URL) |
| | Agent 处理任务
| |---> HTTP POST --->| 调用方回调端点
| (异步回调) |<--- 200 OK -------| 回调响应
| |
| [处理完成] |
5.2 轮询模式（Polling）
调用方 Agent
| |
|--- SubmitTask ------------>| 提交任务
| |
|--- GET /v1/tasks/{id} ---->| 轮询状态（间隔 5s）
|<--- TaskInfo(status=running)|
|--- GET /v1/tasks/{id} ---->|
|<--- TaskInfo(status=running)|
|--- GET /v1/tasks/{id} ---->|
|<--- TaskInfo(status=completed, result=...)
| |
5.3 同步模式（Sync）
调用方 Agent
| |
|--- SubmitTask + Sync ----->| 提交任务
| | Agent 立即处理
|<-- Result (同步返回) -------| 直接返回结果
| |
5.4 混合模式（Hybrid）
调用方 Agent
| |
|--- SubmitTask ------------>| 提交任务
| (preferred: callback, |
| fallback: polling) |
| | Agent 处理任务
| |
| [回调成功] |
|<-- AsyncCallback -----------| 回调通知结果
| |
| [回调失败] |
|--- GET /v1/tasks/{id} ---->| 轮询兜底
|<-- TaskInfo(result=...) ----| 获取结果
| |
6. 错误处理
6.1 SynapseError 结构
// SynapseError 协议错误信息
type SynapseError struct {
Code int `json:"code"`
Message string `json:"message"`
}
6.2 传输模式相关错误码
| 错误码 | 名称 | 说明 |
|---|---|---|
4001 |
CALLBACK_FAILED |
回调端点不可达或返回非 2xx |
4002 |
POLLING_TIMEOUT |
轮询超时，任务未在预期时间内完成 |
4003 |
INVALID_CALLBACK_ENDPOINT |
回调端点 URL 格式无效 |
4004 |
MODE_NOT_SUPPORTED |
Agent 不支持请求的传输模式 |
4005 |
STREAM_SEQUENCE_GAP |
流式传输出现序列号跳变 |
附录：Go 类型汇总
// 传输模式枚举
type TransportMode string
const (
TransportModeCallback TransportMode = "callback"
TransportModePolling TransportMode = "polling"
TransportModeSync TransportMode = "sync"
TransportModeHybrid TransportMode = "hybrid"
)
// 传输偏好配置
type TransportPreferences struct {
Preferred TransportMode `json:"preferred"`
Fallback TransportMode `json:"fallback,omitempty"`
CallbackEndpoint string `json:"callback_endpoint,omitempty"`
PollingInterval int `json:"polling_interval,omitempty"`
ProgressUpdates bool `json:"progress_updates"`
ProgressInterval int `json:"progress_interval,omitempty"`
}
func DefaultTransportPreferences() *TransportPreferences { ... }
// 轮询 API 类型
type TaskStatusResponse struct { ... }
type TaskInfo struct { ... }
type TaskListResponse struct { ... }
type TaskEvent struct { ... }
type TaskEventsResponse struct { ... }
// 流式传输类型
type StreamChunk struct { ... }
type StreamComplete struct { ... }
本文档对应代码实现在 types/content.go 中。