Documentation
¶
Overview ¶
Package eventstream 提供 Agent 执行事件流
为 Agent 运行过程提供实时事件推送，支持：
- 多订阅者并发接收
- 非阻塞发布（缓冲满时丢弃）
- 同步发布（等待所有订阅者消费）
- 便捷的事件发射方法
使用示例：
stream := eventstream.New(eventstream.WithBufferSize(200))
ch, unsub := stream.Subscribe()
defer unsub()
stream.Emit(eventstream.EventAgentStart, "agent-1", map[string]any{"model": "gpt-4"})
event := <-ch
Index ¶
- type Event
- type EventType
- type Option
- type Stream
- func (s *Stream) BridgeTo(bus *event.Bus) func()
- func (s *Stream) Close()
- func (s *Stream) Emit(eventType EventType, agentID string, data map[string]any)
- func (s *Stream) EmitWithTrace(eventType EventType, agentID, traceID, spanID string, data map[string]any)
- func (s *Stream) Publish(event Event)
- func (s *Stream) PublishSync(ctx context.Context, event Event) error
- func (s *Stream) Subscribe() (<-chan Event, func())
- func (s *Stream) SubscriberCount() int
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event struct {
// Type 事件类型
Type EventType `json:"type"`
// AgentID Agent 标识
AgentID string `json:"agent_id"`
// Timestamp 事件时间
Timestamp time.Time `json:"timestamp"`
// Data 事件数据
Data map[string]any `json:"data,omitempty"`
// TraceID 追踪 ID
TraceID string `json:"trace_id,omitempty"`
// SpanID 跨度 ID
SpanID string `json:"span_id,omitempty"`
}
Event Agent 执行事件
type EventType ¶
type EventType string
EventType 事件类型
const ( // EventAgentStart Agent 开始执行 EventAgentStart EventType = "agent.start" // EventAgentEnd Agent 执行完成 EventAgentEnd EventType = "agent.end" // EventAgentError Agent 执行出错 EventAgentError EventType = "agent.error" // EventToolCall 工具调用开始 EventToolCall EventType = "tool.call" // EventToolResult 工具调用结果 EventToolResult EventType = "tool.result" // EventLLMRequest LLM 请求发出 EventLLMRequest EventType = "llm.request" // EventLLMResponse LLM 响应接收 EventLLMResponse EventType = "llm.response" // EventStateChange 状态变更 EventStateChange EventType = "state.change" // EventCheckpoint 检查点保存 EventCheckpoint EventType = "checkpoint" // EventMessage 消息事件 EventMessage EventType = "message" )
type Option ¶
type Option func(*Stream)
Option 配置选项
type Stream ¶
type Stream struct {
// contains filtered or unexported fields
}
Stream Agent 事件流
支持多个订阅者同时接收事件。 线程安全。
func (*Stream) BridgeTo ¶
BridgeTo 将事件流桥接到 toolkit 事件总线
所有通过 Stream 发布的事件会同时发送到指定的 toolkit event.Bus。 事件类型映射为 string(EventType)，Payload 为 Event 结构体。 返回取消桥接的函数。
func (*Stream) EmitWithTrace ¶
func (s *Stream) EmitWithTrace(eventType EventType, agentID, traceID, spanID string, data map[string]any)
EmitWithTrace 带追踪信息的便捷发布
func (*Stream) Publish ¶
Publish 非阻塞发布事件
向所有订阅者发送事件。若某个订阅者的缓冲区满，该订阅者会丢失此事件。 若订阅者 channel 被外部意外关闭，会安全地移除该订阅者而非 panic。
func (*Stream) PublishSync ¶
PublishSync 同步发布事件（阻塞直到所有订阅者消费或 ctx 取消）
Source Files
Click to show internal directories.
Click to hide internal directories.