README
¶
runner — adk.Runner → session bridge
runner makes streaming and history persistence work out of the box for any
eino project, removing the per-project glue that consumes an adk.Runner
iterator, proxies tokens to the client, and persists the full answer through a
session.Turn.
It is the only place in components/memory that imports adk, keeping the
session package policy-free and adk-free.
What it does
Run consumes the adk.AsyncIterator returned by adk.Runner.Run and splits it
into two concurrent halves over a single duplicated stream:
- a proxy goroutine forwards the selected assistant tokens/messages to the
returned
StreamReader(token by token for streaming events) — this is what you stream back to the client; - a persistence goroutine drains the second copy, concatenates the full
assistant answer and commits it through the
Turn.
Guarantees
- no-dangling-user — if no assistant content is produced (or concatenation
fails), the
TurnisDiscard()ed, so the pending user message is never persisted alone and a retry cannot duplicate it; - incomplete — on an iterator error or a truncated stream, the committed
assistant message is tagged with
memory.MarkIncomplete; - ephemeral —
OnErrornotices (memory.NewEphemeralMessage) are streamed to the client but excluded from persistence, as are messages carrying tool calls.
The run is driven under context.Background() inside the bridge so a client
disconnection neither aborts generation nor persistence. Condensation stays the
caller's responsibility under the request context, before calling Run.
Usage
turn, err := sm.BeginTurn(userID, convID, schema.UserMessage(userInput))
if err != nil {
return err
}
defer turn.Discard() // no-op once the bridge commits/discards
if _, err := turn.Condense(ctx); err != nil { // under the request context
return err
}
iter := agentRunner.Run(context.Background(), turn.Window(0))
stream, err := runner.Run(runner.Config{
Turn: turn,
Iterator: iter,
Predicate: runner.AgentRole("supervisor", schema.Assistant), // nil => assistant-only
OnError: func(err error) *schema.Message {
return memory.NewEphemeralMessage(schema.Assistant, "an error occurred")
},
})
if err != nil {
return err
}
// Stream `stream` back to the client (e.g. over SSE). Persistence happens
// asynchronously and releases the session lock when done.
Predicates
Predicate selects which events are streamed and persisted, by emitting agent
name and message role. When nil, every assistant-role message is streamed and
persisted. Composable helpers are provided:
| Helper | Purpose |
|---|---|
AgentRole(name, role) |
Match a specific agent + role (e.g. supervisor + assistant). |
Role(role) |
Match a role regardless of the agent. |
And, Or, Not |
Combine predicates. |
API
| Type / function | Purpose |
|---|---|
Run(Config) |
Start the bridge; returns the client StreamReader. |
Config.Turn |
Required locked session turn (committed/discarded by the bridge). |
Config.Iterator |
Required adk.AsyncIterator from adk.Runner.Run. |
Config.Predicate |
Stream/persist selector; nil => assistant-only. |
Config.OnError |
Optional ephemeral error notice builder. |
Config.OnSkip |
Optional observer of filtered-out events. |
Config.BufferSize |
Pipe buffer size; <= 0 => DefaultBufferSize (1000). |
Documentation
¶
Overview ¶
Package runner bridges an eino adk.Runner run to the cross-request session lifecycle (session.Turn) so that any eino project gets streaming + history persistence out of the box, without re-implementing the glue per project.
Run consumes the adk.AsyncIterator returned by adk.Runner.Run and splits it into two concurrent halves over a single duplicated stream (schema Copy(2)):
- the proxy goroutine reads the iterator and forwards the selected assistant tokens/messages to the returned StreamReader, token by token for streaming events; it is what the caller streams back to the client;
- the persistence goroutine drains the second copy, concatenates the full assistant answer and commits it through the Turn.
Guarantees:
- no-dangling-user: if no assistant content is produced (or concatenation fails), the Turn is Discard()ed so the pending user message is never persisted alone, and a retry cannot duplicate it;
- incomplete: if the iterator reports an error or a stream is truncated, the committed assistant message is tagged via memory.MarkIncomplete;
- ephemeral: messages produced by Config.OnError (memory.NewEphemeralMessage) are streamed to the client but excluded from persistence, as are tool-call messages.
The run is driven under context.Background() inside the bridge so that a client disconnection neither aborts the generation nor the persistence. Any condensation must be performed by the caller (under the request context) before calling Run. This package owns the only adk import in components/memory so that the session package stays policy-free and adk-free.
Index ¶
Constants ¶
const DefaultBufferSize = 1000
DefaultBufferSize is the pipe buffer size used when Config.BufferSize <= 0.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Config ¶
type Config struct {
// Turn is the locked session turn that will persist the assistant answer.
// Required. The bridge takes ownership: it calls CommitAssistant or Discard
// exactly once, so the caller must NOT also release the turn (a plain
// `defer turn.Discard()` stays safe thanks to its idempotent release).
Turn *session.Turn `validate:"required" jsonschema:"description=Locked session turn that persists the assistant answer"`
// Iterator is the async iterator returned by adk.Runner.Run. Required.
Iterator *adk.AsyncIterator[*adk.AgentEvent] `validate:"required" jsonschema:"description=Async iterator returned by adk.Runner.Run"`
// Predicate selects which events are streamed and persisted, keyed by the
// emitting agent name and message role. When nil, the bridge streams and
// persists every assistant-role message (assistant-only default).
Predicate MessagePredicate `jsonschema:"description=Event filter keyed by agent name and message role, defaults to assistant-only"`
// OnError, when non-nil, is invoked with an iterator error to build an
// ephemeral notice streamed to the client (never persisted). When nil, the
// error is only forwarded on the stream and the answer marked incomplete.
OnError func(err error) *schema.Message `jsonschema:"description=Error handler that builds ephemeral notices streamed to client, never persisted"`
// OnSkip, when non-nil, observes events filtered out by Predicate (debug/trace).
OnSkip func(event *adk.AgentEvent) `jsonschema:"description=Debug/trace observer for events filtered out by Predicate"`
// BufferSize overrides the pipe buffer size. <= 0 uses DefaultBufferSize.
BufferSize int `validate:"gte=0" jsonschema:"description=Pipe buffer size, defaults to 1000 if zero"`
}
Config configures a single bridged run.
type MessagePredicate ¶
MessagePredicate decides whether an agent event (identified by its emitting agent name and message role) should be streamed to the client and persisted to the conversation history.
When a Config.Predicate is nil, the bridge defaults to "assistant-only": every message whose role is schema.Assistant is streamed and persisted (see runner.go).
func AgentRole ¶
func AgentRole(agentName string, role schema.RoleType) MessagePredicate
AgentRole matches events emitted by a specific agent with a specific role. It covers the common "supervisor + assistant" case where only the top-level agent's assistant output should reach the client and the history.
func And ¶
func And(preds ...MessagePredicate) MessagePredicate
And returns a predicate that is true only when all preds are true. With no arguments it always returns true.
func Or ¶
func Or(preds ...MessagePredicate) MessagePredicate
Or returns a predicate that is true when any of preds is true. With no arguments it always returns false.
func Role ¶
func Role(role schema.RoleType) MessagePredicate
Role matches events with the given role, regardless of the emitting agent.
Source Files