Documentation
¶
Overview ¶
Package exp provides experimental Firebase integrations for Genkit's agent runtime (see github.com/firebase/genkit/go/ai/exp).
The FirestoreSessionStore persists agent session snapshots in Cloud Firestore. It resolves its Firestore client from the Firebase plugin registered with the Genkit instance, then wires into an agent:
g := genkit.Init(ctx, genkit.WithPlugins(&firebase.Firebase{ProjectId: "my-project"}))
store, err := exp.NewFirestoreSessionStore[MyState](ctx, g)
// handle err
agent := aix.DefineAgent(g, "assistant", run, aix.WithSessionStore(store))
APIs in this package are under active development and may change in any minor version release. Use with caution in production environments.
Index ¶
- Constants
- type CollectionOption
- type FirestoreSessionStore
- func (s *FirestoreSessionStore[State]) GetLatestSnapshot(ctx context.Context, sessionID string) (*aix.SessionSnapshot[State], error)
- func (s *FirestoreSessionStore[State]) GetSnapshot(ctx context.Context, snapshotID string) (*aix.SessionSnapshot[State], error)
- func (s *FirestoreSessionStore[State]) OnSnapshotStatusChange(ctx context.Context, snapshotID string) <-chan aix.SnapshotStatus
- func (s *FirestoreSessionStore[State]) SaveSnapshot(ctx context.Context, id string, ...) (*aix.SessionSnapshot[State], error)
- type FirestoreStreamManager
- type SessionStoreOption
- type StreamManagerOption
Constants ¶
const ( // DefaultTTL is the default time-to-live for stream documents. DefaultTTL = 5 * time.Minute )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CollectionOption ¶
type CollectionOption interface {
StreamManagerOption
SessionStoreOption
}
CollectionOption is an option valid for both Firestore services in this package (the stream manager and the session store). Only WithCollection returns one.
func WithCollection ¶
func WithCollection(collection string) CollectionOption
WithCollection sets the Firestore collection documents are stored under. For the stream manager this is the stream document collection (required); for the session store it is the root snapshot collection (two companion collections, "<collection>-shards" and "<collection>-pointers", are derived from it, and it defaults to "genkit-sessions" when omitted).
type FirestoreSessionStore ¶
type FirestoreSessionStore[State any] struct { // contains filtered or unexported fields }
FirestoreSessionStore is a Firestore-backed aix.SessionStore that persists session snapshots as incremental JSON Patch diffs anchored to periodic, sharded full-state checkpoints.
Storage layout (the <prefix> segment is the per-tenant prefix returned by WithSnapshotPathPrefix, or "global" when none is configured):
- <collection>/<prefix>/snapshots/<snapshotID> - one document per snapshot. A "diff" document holds the JSON Patch from its parent (statePatch); a "checkpoint" document holds a full-state materialization, sharded out of band.
- <collection>-shards/<prefix>/shards/<checkpointID>_<index> - the sharded full state for a checkpoint.
- <collection>-pointers/<prefix>/pointers/<sessionID> - one document per session pointing at its latest snapshot and the metadata needed to reconstruct it.
That default places every session under one shared "global" prefix, so pass WithSnapshotPathPrefix to scope them per tenant when identifiers could repeat across users (e.g. per-user session IDs).
Reconstruction uses only document-ID lookups (GetAll), so it needs no secondary indexes and is strongly consistent. No single document approaches the 1 MiB limit (state is sharded by shard size), and the number of diff documents touched per read or write is bounded by the checkpoint interval rather than total session length, so the store scales to arbitrarily long sessions. Checkpoints still store the full accumulated state, so checkpoint shard count grows with the state's size; tune WithCheckpointInterval to trade per-save diff reads against checkpoint write amplification.
It implements aix.SessionStore and aix.SnapshotSubscriber; the latter uses Firestore's native real-time listener, so an abort committed by one process is observed by the process running the detached turn even across instances.
func NewFirestoreSessionStore ¶
func NewFirestoreSessionStore[State any](ctx context.Context, g *genkit.Genkit, opts ...SessionStoreOption) (*FirestoreSessionStore[State], error)
NewFirestoreSessionStore creates a Firestore-backed snapshot store. It resolves the Firestore client from the Firebase plugin registered with g (the Firebase plugin must be passed to genkit.Init before calling this), mirroring github.com/firebase/genkit/go/plugins/firebase/exp.NewFirestoreStreamManager.
The State type parameter is the user-defined custom-state type carried in aix.SessionState.Custom; it must be JSON-serializable.
func (*FirestoreSessionStore[State]) GetLatestSnapshot ¶
func (s *FirestoreSessionStore[State]) GetLatestSnapshot(ctx context.Context, sessionID string) (*aix.SessionSnapshot[State], error)
GetLatestSnapshot returns the session's most recently created snapshot regardless of status, per the aix.SnapshotReader.GetLatestSnapshot contract. It reads the session's pointer document (which tracks the greatest-CreatedAt snapshot, ties broken by snapshot ID) and reconstructs from the pointer's cached checkpoint metadata.
func (*FirestoreSessionStore[State]) GetSnapshot ¶
func (s *FirestoreSessionStore[State]) GetSnapshot(ctx context.Context, snapshotID string) (*aix.SessionSnapshot[State], error)
GetSnapshot retrieves a snapshot by ID. Returns nil if not found. The reconstruction runs inside a read-only transaction so the snapshot's checkpoint shards and diff segment are read at one consistent point, never a mix of pre- and post-checkpoint-rewrite chunks.
func (*FirestoreSessionStore[State]) OnSnapshotStatusChange ¶
func (s *FirestoreSessionStore[State]) OnSnapshotStatusChange(ctx context.Context, snapshotID string) <-chan aix.SnapshotStatus
OnSnapshotStatusChange returns a channel that yields the snapshot's status at subscription time and on every subsequent change, until ctx is cancelled. It is backed by Firestore's native document listener, so a status change written by any process (e.g. an abort committed by the request handler while a detached worker watches) propagates across instances without polling. If the snapshot does not exist when the subscription is established, the channel is closed without yielding a value.
Values are level-triggered: the latest status is always delivered, but a slow reader may skip intermediate values. Treat a received value as "the status is now X", not "X happened once".
func (*FirestoreSessionStore[State]) SaveSnapshot ¶
func (s *FirestoreSessionStore[State]) SaveSnapshot( ctx context.Context, id string, fn func(existing *aix.SessionSnapshot[State]) (*aix.SessionSnapshot[State], error), ) (*aix.SessionSnapshot[State], error)
SaveSnapshot atomically reads the snapshot at id (if any), applies fn, and persists the result. See aix.SnapshotWriter for the full contract. The read-modify-write runs inside a Firestore transaction, which may re-run fn on contention; fn must therefore be free of side effects, as the contract requires.
type FirestoreStreamManager ¶
type FirestoreStreamManager struct {
// contains filtered or unexported fields
}
FirestoreStreamManager implements streaming.StreamManager using Firestore as the backend. Stream state is persisted in Firestore documents, allowing streams to survive server restarts and be accessible across multiple instances.
func NewFirestoreStreamManager ¶
func NewFirestoreStreamManager(ctx context.Context, g *genkit.Genkit, opts ...StreamManagerOption) (*FirestoreStreamManager, error)
NewFirestoreStreamManager creates a FirestoreStreamManager for durable streaming. Requires the Firebase plugin to be initialized in the Genkit instance.
func (*FirestoreStreamManager) Open ¶
func (m *FirestoreStreamManager) Open(ctx context.Context, streamID string) (streaming.StreamInput, error)
Open creates a new stream for writing. Returns ALREADY_EXISTS error if a stream with the given ID already exists.
func (*FirestoreStreamManager) Subscribe ¶
func (m *FirestoreStreamManager) Subscribe(ctx context.Context, streamID string) (<-chan streaming.StreamEvent, func(), error)
Subscribe subscribes to an existing stream.
type SessionStoreOption ¶
type SessionStoreOption interface {
// contains filtered or unexported methods
}
SessionStoreOption configures a FirestoreSessionStore.
func WithCheckpointInterval ¶
func WithCheckpointInterval(turns int) SessionStoreOption
WithCheckpointInterval sets the number of turns between full-state checkpoints. A larger value stores fewer (but reconstructs over more) diffs; a smaller value reconstructs faster at the cost of more frequent full-state writes. The number of diff documents read or written per turn is bounded by this value rather than by total session length. Must be at least 1; defaults to 25 when omitted.
func WithShardSize ¶
func WithShardSize(bytes int) SessionStoreOption
WithShardSize sets the maximum size in bytes of a single shard or diff document. Checkpoint state is split into chunks of this size, and any diff exceeding it is promoted to a (sharded) checkpoint, so no document approaches Firestore's 1 MiB limit. Must be positive; defaults to 512 KiB when omitted.
func WithSnapshotPathPrefix ¶
func WithSnapshotPathPrefix(fn func(ctx context.Context) string) SessionStoreOption
WithSnapshotPathPrefix derives a per-call tenant prefix from the operation's context. When set, all snapshot, shard, and pointer documents are nested under a tenant-scoped subcollection keyed by this prefix, so reads and writes are isolated per tenant: one tenant can never address another's snapshots, even holding a snapshot ID, because resolving it still requires the matching, auth-derived prefix. A typical fn pulls a stable identity (e.g. an authenticated user or org ID) out of ctx.
The value must be a valid Firestore document ID (no "/" separators) and stable for a given snapshot's lifetime, since every read recomputes it. It must be non-empty: an empty result is rejected at call time, since the way to request the default "global" prefix is to omit this option entirely, not to return an empty value.
type StreamManagerOption ¶
type StreamManagerOption interface {
// contains filtered or unexported methods
}
StreamManagerOption configures a FirestoreStreamManager.
func WithTTL ¶
func WithTTL(ttl time.Duration) StreamManagerOption
WithTTL sets how long stream documents are retained before Firestore auto-deletes them. Requires a TTL policy on the collection for the "expiresAt" field. Defaults to 5 minutes when omitted. See: https://firebase.google.com/docs/firestore/ttl
func WithTimeout ¶
func WithTimeout(timeout time.Duration) StreamManagerOption
WithTimeout sets how long a subscriber waits for new events before giving up. If no activity occurs within this duration, subscribers receive a DEADLINE_EXCEEDED error. Defaults to 60 seconds when omitted.
Source Files