Affected by 
Documentation
¶
Index ¶
- Constants
- Variables
- func NewStreamManager() (interfaces.StreamManager, error)
- type MemoryStreamManager
- func (m *MemoryStreamManager) AppendEvent(ctx context.Context, sessionID, messageID string, event interfaces.StreamEvent) error
- func (m *MemoryStreamManager) AppendSteerEvents(_ context.Context, sessionID, messageID string, ...) error
- func (m *MemoryStreamManager) ClaimLiveRun(_ context.Context, sessionID, assistantMessageID, requestID string) error
- func (m *MemoryStreamManager) ClearLiveRun(_ context.Context, sessionID, assistantMessageID string) error
- func (m *MemoryStreamManager) DeleteSteerEvent(_ context.Context, sessionID, messageID, eventID string) (bool, error)
- func (m *MemoryStreamManager) DropMessageStreams(_ context.Context, sessionID string, messageIDs []string) error
- func (m *MemoryStreamManager) GetEvents(ctx context.Context, sessionID, messageID string, fromOffset int) ([]interfaces.StreamEvent, int, error)
- func (m *MemoryStreamManager) GetLiveRun(_ context.Context, sessionID string) (string, string, error)
- func (m *MemoryStreamManager) GetSteerEvents(_ context.Context, sessionID, messageID string, fromOffset int) ([]interfaces.StreamEvent, int, error)
- func (m *MemoryStreamManager) SetLiveRun(_ context.Context, sessionID, assistantMessageID, requestID string) error
- func (m *MemoryStreamManager) UpdateSteerEventData(_ context.Context, sessionID, messageID, eventID string, ...) (bool, error)
- type RedisStreamManager
- func (r *RedisStreamManager) AppendEvent(ctx context.Context, sessionID, messageID string, event interfaces.StreamEvent) error
- func (r *RedisStreamManager) AppendSteerEvents(ctx context.Context, sessionID, messageID string, ...) error
- func (r *RedisStreamManager) ClaimLiveRun(ctx context.Context, sessionID, assistantMessageID, requestID string) error
- func (r *RedisStreamManager) ClearLiveRun(ctx context.Context, sessionID, assistantMessageID string) error
- func (r *RedisStreamManager) Close() error
- func (r *RedisStreamManager) DeleteSteerEvent(ctx context.Context, sessionID, messageID, eventID string) (bool, error)
- func (r *RedisStreamManager) DropMessageStreams(ctx context.Context, sessionID string, messageIDs []string) error
- func (r *RedisStreamManager) GetEvents(ctx context.Context, sessionID, messageID string, fromOffset int) ([]interfaces.StreamEvent, int, error)
- func (r *RedisStreamManager) GetLiveRun(ctx context.Context, sessionID string) (string, string, error)
- func (r *RedisStreamManager) GetSteerEvents(ctx context.Context, sessionID, messageID string, fromOffset int) ([]interfaces.StreamEvent, int, error)
- func (r *RedisStreamManager) SetLiveRun(ctx context.Context, sessionID, assistantMessageID, requestID string) error
- func (r *RedisStreamManager) UpdateSteerEventData(ctx context.Context, sessionID, messageID, eventID string, ...) (bool, error)
Constants ¶
const ( TypeMemory = "memory" TypeRedis = "redis" )
流管理器类型
Variables ¶
var ErrLiveRunExists = errors.New("session already has a live run")
ErrLiveRunExists is returned by SetLiveRun when another assistant already holds the session. Follow-up handoff uses ClaimLiveRun to overwrite.
Functions ¶
func NewStreamManager ¶
func NewStreamManager() (interfaces.StreamManager, error)
NewStreamManager 创建流管理器
Types ¶
type MemoryStreamManager ¶
type MemoryStreamManager struct {
// contains filtered or unexported fields
}
MemoryStreamManager implements StreamManager using in-memory storage
func NewMemoryStreamManager ¶
func NewMemoryStreamManager() *MemoryStreamManager
NewMemoryStreamManager creates a new in-memory stream manager
func (*MemoryStreamManager) AppendEvent ¶
func (m *MemoryStreamManager) AppendEvent( ctx context.Context, sessionID, messageID string, event interfaces.StreamEvent, ) error
AppendEvent appends a single event to the stream
func (*MemoryStreamManager) AppendSteerEvents ¶
func (m *MemoryStreamManager) AppendSteerEvents( _ context.Context, sessionID, messageID string, events []interfaces.StreamEvent, ) error
AppendSteerEvents appends control events to the dedicated steer sub-list.
func (*MemoryStreamManager) ClaimLiveRun ¶
func (m *MemoryStreamManager) ClaimLiveRun( _ context.Context, sessionID, assistantMessageID, requestID string, ) error
ClaimLiveRun overwrites the live-run marker for follow-up handoff.
func (*MemoryStreamManager) ClearLiveRun ¶
func (m *MemoryStreamManager) ClearLiveRun( _ context.Context, sessionID, assistantMessageID string, ) error
ClearLiveRun drops the marker only when it still points at assistantMessageID.
func (*MemoryStreamManager) DeleteSteerEvent ¶
func (m *MemoryStreamManager) DeleteSteerEvent( _ context.Context, sessionID, messageID, eventID string, ) (bool, error)
DeleteSteerEvent removes a queued steer event by ID.
func (*MemoryStreamManager) DropMessageStreams ¶
func (m *MemoryStreamManager) DropMessageStreams( _ context.Context, sessionID string, messageIDs []string, ) error
DropMessageStreams removes event and steer buffers for deleted messages so a later rewind cannot replay or steer a turn that no longer exists.
func (*MemoryStreamManager) GetEvents ¶
func (m *MemoryStreamManager) GetEvents( ctx context.Context, sessionID, messageID string, fromOffset int, ) ([]interfaces.StreamEvent, int, error)
GetEvents gets events starting from offset Returns: events slice, next offset, error
func (*MemoryStreamManager) GetLiveRun ¶
func (m *MemoryStreamManager) GetLiveRun( _ context.Context, sessionID string, ) (string, string, error)
GetLiveRun returns the session's generating assistant message, if any.
func (*MemoryStreamManager) GetSteerEvents ¶
func (m *MemoryStreamManager) GetSteerEvents( _ context.Context, sessionID, messageID string, fromOffset int, ) ([]interfaces.StreamEvent, int, error)
GetSteerEvents drains the steer sub-list starting fromOffset.
func (*MemoryStreamManager) SetLiveRun ¶
func (m *MemoryStreamManager) SetLiveRun( _ context.Context, sessionID, assistantMessageID, requestID string, ) error
SetLiveRun records the session's currently generating assistant message.
func (*MemoryStreamManager) UpdateSteerEventData ¶
func (m *MemoryStreamManager) UpdateSteerEventData( _ context.Context, sessionID, messageID, eventID string, data map[string]interface{}, ) (bool, error)
UpdateSteerEventData merges keys into a queued steer event's Data map.
type RedisStreamManager ¶
type RedisStreamManager struct {
// contains filtered or unexported fields
}
RedisStreamManager implements StreamManager using Redis Lists for append-only event streaming
func NewRedisStreamManager ¶
func NewRedisStreamManager(redisAddr, redisUsername, redisPassword string, redisDB int, prefix string, ttl time.Duration, ) (*RedisStreamManager, error)
NewRedisStreamManager creates a new Redis-based stream manager
func (*RedisStreamManager) AppendEvent ¶
func (r *RedisStreamManager) AppendEvent( ctx context.Context, sessionID, messageID string, event interfaces.StreamEvent, ) error
AppendEvent appends a single event to the stream using Redis RPush
func (*RedisStreamManager) AppendSteerEvents ¶
func (r *RedisStreamManager) AppendSteerEvents( ctx context.Context, sessionID, messageID string, events []interfaces.StreamEvent, ) error
AppendSteerEvents appends control events to the per-run steer sub-list.
func (*RedisStreamManager) ClaimLiveRun ¶
func (r *RedisStreamManager) ClaimLiveRun( ctx context.Context, sessionID, assistantMessageID, requestID string, ) error
ClaimLiveRun overwrites the live-run marker for follow-up handoff.
func (*RedisStreamManager) ClearLiveRun ¶
func (r *RedisStreamManager) ClearLiveRun( ctx context.Context, sessionID, assistantMessageID string, ) error
ClearLiveRun drops the marker only when it still points at assistantMessageID.
func (*RedisStreamManager) Close ¶
func (r *RedisStreamManager) Close() error
Close closes the Redis connection
func (*RedisStreamManager) DeleteSteerEvent ¶
func (r *RedisStreamManager) DeleteSteerEvent( ctx context.Context, sessionID, messageID, eventID string, ) (bool, error)
DeleteSteerEvent removes a queued steer event by ID from Redis.
func (*RedisStreamManager) DropMessageStreams ¶
func (r *RedisStreamManager) DropMessageStreams( ctx context.Context, sessionID string, messageIDs []string, ) error
DropMessageStreams removes event and steer lists for deleted messages, and clears a live-run marker that still names one of them.
func (*RedisStreamManager) GetEvents ¶
func (r *RedisStreamManager) GetEvents( ctx context.Context, sessionID, messageID string, fromOffset int, ) ([]interfaces.StreamEvent, int, error)
GetEvents gets events starting from offset using Redis LRange Returns: events slice, next offset, error
func (*RedisStreamManager) GetLiveRun ¶
func (r *RedisStreamManager) GetLiveRun( ctx context.Context, sessionID string, ) (string, string, error)
GetLiveRun returns the session's generating assistant message, if any.
func (*RedisStreamManager) GetSteerEvents ¶
func (r *RedisStreamManager) GetSteerEvents( ctx context.Context, sessionID, messageID string, fromOffset int, ) ([]interfaces.StreamEvent, int, error)
GetSteerEvents drains the steer sub-list starting fromOffset.
func (*RedisStreamManager) SetLiveRun ¶
func (r *RedisStreamManager) SetLiveRun( ctx context.Context, sessionID, assistantMessageID, requestID string, ) error
SetLiveRun records the session's currently generating assistant message so every replica routes steer requests to the same run. It refuses to replace a different live assistant; ClaimLiveRun is the overwrite path.
func (*RedisStreamManager) UpdateSteerEventData ¶
func (r *RedisStreamManager) UpdateSteerEventData( ctx context.Context, sessionID, messageID, eventID string, data map[string]interface{}, ) (bool, error)
UpdateSteerEventData merges keys into a queued steer event's Data map.
Source Files