diff --git a/apps/docs/content/docs/framework/events/compaction.mdx b/apps/docs/content/docs/framework/events/compaction.mdx index 82a519054..0c6cb5867 100644 --- a/apps/docs/content/docs/framework/events/compaction.mdx +++ b/apps/docs/content/docs/framework/events/compaction.mdx @@ -1,341 +1,152 @@ --- title: Event Compaction -description: Manage long session histories with automatic event summarization and compaction +description: Automatically summarise long ADK-TS session histories with LLM-generated compaction events so context windows stay manageable without losing key information. --- import { Callout } from "fumadocs-ui/components/callout"; -import { Tab, Tabs } from "fumadocs-ui/components/tabs"; -import { Step, Steps } from "fumadocs-ui/components/steps"; +import { Tabs, Tab } from "fumadocs-ui/components/tabs"; +import { Cards, Card } from "fumadocs-ui/components/card"; +import { Mermaid } from "@/components/mdx/mermaid"; -Event Compaction automatically manages long conversation histories by periodically replacing ranges of events with LLM-generated summaries. This keeps context windows manageable while preserving important information. +As a conversation grows, sending the full event history to the LLM on every turn becomes expensive and eventually hits the model's context limit. Event compaction solves this by periodically replacing a window of older events with a single LLM-generated summary event. Your agent continues to receive relevant context; the session history stays compact. -## Overview +## How it works -As conversations grow longer, sending the full history to the LLM becomes inefficient and costly. Event Compaction solves this by: +After each agent turn, ADK-TS checks how many new invocations have completed since the last compaction. When that count reaches `compactionInterval`, it selects a sliding window of events to summarise, calls the configured summariser, and appends the result as a special compaction event. Future LLM requests skip the original events inside the window and use the summary instead. -- **Reducing token usage** through concise summaries -- **Improving performance** with smaller context windows -- **Maintaining context** by preserving key information -- **Operating transparently** without code changes + - -Event Compaction uses a sliding window algorithm that triggers after a -configured number of new invocations, ensuring consistent behavior and -predictable costs. - - - -## How It Works - -### The Compaction Process - - - - - ### Trigger Condition After each agent turn completes, ADK-TS checks if enough - new invocations have occurred since the last compaction. - - - - ### Window Selection When triggered, ADK-TS selects a window of invocations to - compact, including some overlap from the previous compaction for continuity. - - - - ### Summarization The selected events are sent to an LLM (by default, the - agent's own model) to generate a concise summary. - - - - ### Storage The summary is stored as a special compaction event with - `actions.compaction` containing the timestamp range and summarized content. - - - - ### Consumption When building LLM requests, compaction events are converted to - normal model events and original covered events are filtered out. - + R->>C: Check after turn N + C->>C: Count new invocations since last compaction + alt threshold not reached + C-->>R: No-op + else threshold reached + C->>Sess: Read events in sliding window + C->>S: maybeSummarizeEvents(events) + S-->>C: Compaction Event + C->>Sess: appendEvent(compactionEvent) + end - +`} +/> -### Sliding Window Algorithm +The sliding window includes a configurable `overlapSize` — a few invocations from the previously-compacted range — so there is no abrupt context gap between successive summaries. -```mermaid -timeline - title Event Compaction Timeline - Invocations 1-5 : Original events - After Inv 5 : First compaction (Inv 1-5) - Invocations 6-10 : New original events - After Inv 10 : Second compaction (Inv 4-10) - : Overlap = Inv 4-5 - Invocations 11-15 : New original events -``` - -The algorithm maintains continuity by including overlap from the previous range, ensuring no context is lost between compactions. - -## Configuration - -### Basic Setup +## Quick start - - +Pass `eventsCompactionConfig` when building your runner. Compaction then happens automatically with no further code changes: ```typescript import { AgentBuilder } from "@iqai/adk"; -// Simple usage with AgentBuilder const { runner } = await AgentBuilder.create("assistant") .withModel("gemini-2.5-flash") + .withInstruction("You are a helpful assistant.") .withEventsCompaction({ - compactionInterval: 10, // Compact every 10 invocations - overlapSize: 2, // Include 2 prior invocations + compactionInterval: 10, // compact after every 10 new invocations + overlapSize: 2, // include 2 prior invocations for continuity }) .build(); -// Now compaction happens automatically +// Use runner normally — compaction is transparent const response = await runner.ask("Hello!"); ``` - - +## Configuration -```typescript -import { Runner, EventsCompactionConfig } from "@iqai/adk"; +`EventsCompactionConfig` has three fields: -const compactionConfig: EventsCompactionConfig = { - // Trigger compaction every 10 new invocations - compactionInterval: 10, +| Field | Type | Required | Description | +| -------------------- | ------------------ | -------- | --------------------------------------------------------------------------------- | +| `compactionInterval` | `number` | Yes | Number of new invocations needed to trigger compaction | +| `overlapSize` | `number` | Yes | Prior invocations to include from the last compacted range for context continuity | +| `summarizer` | `EventsSummarizer` | No | Custom summariser — defaults to `LlmEventSummarizer` using the agent's own model | - // Include 2 prior invocations for continuity - overlapSize: 2, -}; + + A higher `compactionInterval` retains more history before compacting. A lower + value compacts more aggressively — useful for very long sessions. Start with + 10 / 2 and tune based on your model's context budget. + -const runner = new Runner({ - appName: "MyApp", - agent: myAgent, - sessionService: mySessionService, - eventsCompactionConfig: compactionConfig, -}); -``` +### Configuring via Runner directly - - +If you construct the `Runner` manually rather than through `AgentBuilder`: ```typescript -import { - EventsSummarizer, - Event, - EventActions, - LlmEventSummarizer, -} from "@iqai/adk"; - -// Option 1: Use default with custom model -const customModel = LLMRegistry.newLLM("gpt-4o-mini"); -const summarizer = new LlmEventSummarizer(customModel); - -// Option 2: Implement custom logic -class CustomSummarizer implements EventsSummarizer { - async maybeSummarizeEvents(events: Event[]): Promise { - // Your custom summarization logic - const summary = await this.generateSummary(events); - - if (!summary) return undefined; - - return new Event({ - invocationId: Event.newId(), - author: "user", - actions: new EventActions({ - compaction: { - startTimestamp: events[0].timestamp, - endTimestamp: events[events.length - 1].timestamp, - compactedContent: { - role: "model", - parts: [{ text: summary }], - }, - }, - }), - }); - } -} +import { Runner } from "@iqai/adk"; +import type { EventsCompactionConfig } from "@iqai/adk"; const compactionConfig: EventsCompactionConfig = { - summarizer: new CustomSummarizer(), compactionInterval: 10, overlapSize: 2, }; -``` - - - - -### Configuration Options - -| Option | Type | Default | Description | -| -------------------- | ------------------ | ------------ | ------------------------------------------------------ | -| `compactionInterval` | `number` | Required | Number of new invocations needed to trigger compaction | -| `overlapSize` | `number` | Required | Number of prior invocations to include for continuity | -| `summarizer` | `EventsSummarizer` | Auto-created | Custom summarizer (defaults to LLM-based) | - - - -Choose your `compactionInterval` based on your use case: - **Higher values -(20+)**: Less frequent compaction, more detailed history retained - **Lower -values (5-10)**: More aggressive compaction, better for very long sessions - - - -## Understanding the Components -### EventCompaction Interface - -The compaction data structure that's stored in event actions: - -```typescript -interface EventCompaction { - startTimestamp: number; // Start of compacted range - endTimestamp: number; // End of compacted range - compactedContent: Content; // The summarized content -} -``` - -### EventsSummarizer Interface - -The interface for custom summarizers: - -```typescript -interface EventsSummarizer { - maybeSummarizeEvents(events: Event[]): Promise; -} +const runner = new Runner({ + appName: "my-app", + agent: myAgent, + sessionService: mySessionService, + eventsCompactionConfig: compactionConfig, +}); ``` -### LlmEventSummarizer +## LlmEventSummarizer -The default implementation that uses an LLM to generate summaries: +When no `summarizer` is provided, ADK-TS uses `LlmEventSummarizer` with the agent's own model. You can also instantiate it explicitly to use a different model — for example, a smaller, cheaper model dedicated to summarisation: ```typescript -import { LlmEventSummarizer } from "@iqai/adk"; - -// Use with custom prompt -const summarizer = new LlmEventSummarizer( - myModel, - `Summarize these conversation events focusing on: - - Key decisions made - - Important information exchanged - - Action items identified - - Events: {events}`, -); -``` - -## Examples +import { AgentBuilder, LlmEventSummarizer, LLMRegistry } from "@iqai/adk"; -### Basic Usage with AgentBuilder - -```typescript -import { AgentBuilder } from "@iqai/adk"; +const cheapModel = LLMRegistry.newLLM("gemini-2.0-flash-lite"); +const summarizer = new LlmEventSummarizer(cheapModel); -// Create agent with compaction using AgentBuilder const { runner } = await AgentBuilder.create("assistant") .withModel("gemini-2.5-flash") - .withInstruction("You are a helpful assistant.") .withEventsCompaction({ - compactionInterval: 5, - overlapSize: 1, + summarizer, + compactionInterval: 10, + overlapSize: 2, }) .build(); - -// Use normally - compaction happens automatically -const response = await runner.ask("Hello!"); -console.log(response); ``` -### Basic Usage with Runner +`LlmEventSummarizer` accepts an optional second argument — a custom prompt template. Use `{events}` as the placeholder where the formatted event list will be inserted: ```typescript -import { InMemoryRunner, LlmAgent } from "@iqai/adk"; - -const agent = new LlmAgent({ - name: "Assistant", - model: "gemini-2.5-flash", - instruction: "You are a helpful assistant.", -}); - -const runner = new InMemoryRunner(agent, { - appName: "ChatApp", -}); - -// Enable compaction -runner.eventsCompactionConfig = { - compactionInterval: 5, - overlapSize: 1, -}; +const summarizer = new LlmEventSummarizer( + myModel, + `Summarise this conversation history concisely, preserving: +- Key decisions and conclusions +- Important facts and named entities +- Any open questions or action items -// Use normally - compaction happens automatically -for await (const event of runner.runAsync({ - userId: "user123", - sessionId: "session456", - newMessage: { role: "user", parts: [{ text: "Hello!" }] }, -})) { - console.log(event); -} +Conversation: +{events}`, +); ``` -### Checking Compaction Status - -```typescript -// Check if a session has compacted events -function hasCompactions(session: Session): boolean { - return session.events.some(event => event.actions?.compaction); -} - -// Get compaction statistics -function getCompactionStats(session: Session) { - const compactions = session.events.filter(e => e.actions?.compaction); - - return { - totalCompactions: compactions.length, - totalEventsCompacted: compactions.reduce((sum, e) => { - const compact = e.actions.compaction!; - const covered = session.events.filter( - ev => - ev.timestamp >= compact.startTimestamp && - ev.timestamp <= compact.endTimestamp, - ); - return sum + covered.length; - }, 0), - }; -} -``` +## Custom summariser -### Custom Summarization Strategy +For full control, implement the `EventsSummarizer` interface. Return an `Event` with `actions.compaction` set, or `undefined` to skip compaction for this batch: ```typescript -class BulletPointSummarizer implements EventsSummarizer { - constructor(private model: BaseLlm) {} +import { Event, EventActions } from "@iqai/adk"; +import type { EventsSummarizer } from "@iqai/adk"; +class BulletPointSummarizer implements EventsSummarizer { async maybeSummarizeEvents(events: Event[]): Promise { - if (events.length < 3) return undefined; // Skip small batches + if (events.length < 5) return undefined; // skip small batches - const eventTexts = events + const lines = events .filter(e => e.content?.parts?.[0]?.text) - .map(e => `${e.author}: ${e.content.parts[0].text}`); + .map(e => `${e.author}: ${e.content!.parts![0].text}`); - const prompt = `Create a bullet-point summary of this conversation: - -${eventTexts.join("\n")} - -Format as: -• Key point 1 -• Key point 2 -• Key point 3`; - - let summary = ""; - for await (const response of this.model.generateContentAsync({ - contents: [{ role: "user", parts: [{ text: prompt }] }], - } as any)) { - summary += response.content?.parts?.map(p => p.text || "").join(""); - } + const summary = await generateBulletSummary(lines.join("\n")); return new Event({ invocationId: Event.newId(), @@ -346,7 +157,7 @@ Format as: endTimestamp: events[events.length - 1].timestamp, compactedContent: { role: "model", - parts: [{ text: summary.trim() }], + parts: [{ text: summary }], }, }, }), @@ -355,239 +166,64 @@ Format as: } ``` -## Best Practices - -### Choosing Compaction Interval - -```typescript -// For customer support (high detail retention) -const supportConfig: EventsCompactionConfig = { - compactionInterval: 20, // Compact less frequently - overlapSize: 5, // More overlap -}; - -// For general chat (aggressive compaction) -const chatConfig: EventsCompactionConfig = { - compactionInterval: 5, // Compact frequently - overlapSize: 1, // Minimal overlap -}; - -// For long research sessions (balanced) -const researchConfig: EventsCompactionConfig = { - compactionInterval: 10, // Moderate compaction - overlapSize: 2, // Standard overlap -}; -``` - -### Monitoring Compaction - -```typescript -class CompactionMonitor { - private compactionCount = 0; - private eventsCompacted = 0; - - async runWithMonitoring(runner: Runner, query: string) { - const sessionBefore = await runner.sessionService.getSession(...); - - // Run agent - for await (const event of runner.runAsync(...)) { - if (event.actions?.compaction) { - this.compactionCount++; - console.log('Compaction occurred:', { - range: [ - event.actions.compaction.startTimestamp, - event.actions.compaction.endTimestamp, - ], - summary: event.actions.compaction.compactedContent, - }); - } - } - - const sessionAfter = await runner.sessionService.getSession(...); - console.log('Session stats:', { - totalEvents: sessionAfter.events.length, - compactions: this.compactionCount, - }); - } -} -``` - -### Testing with Compaction - -```typescript -import { describe, it, expect } from "vitest"; - -describe("Agent with Compaction", () => { - it("should compact after threshold", async () => { - const runner = new InMemoryRunner(agent, { - appName: "TestApp", - }); - - runner.eventsCompactionConfig = { - compactionInterval: 3, - overlapSize: 1, - }; - - // Generate 3 invocations - for (let i = 0; i < 3; i++) { - for await (const event of runner.runAsync({ - userId: "test", - sessionId: "test", - newMessage: { role: "user", parts: [{ text: `Message ${i}` }] }, - })) { - // Process events - } - } - - const session = await runner.sessionService.getSession( - "TestApp", - "test", - "test", - ); - const hasCompaction = session?.events.some(e => e.actions?.compaction); - - expect(hasCompaction).toBe(true); - }); -}); -``` - -## Performance Considerations - -### Token Savings - -Event compaction can significantly reduce token usage: +Pass the custom summariser in the config: ```typescript -// Example: 10 invocations with 200 tokens each = 2000 tokens -// After compaction: 1 summary (300 tokens) + 5 recent (1000 tokens) = 1300 tokens -// Savings: 35% -``` - -### Latency Impact - -- **First compaction**: Adds summarization latency (~1-2s) -- **Subsequent requests**: Faster due to smaller context -- **Net effect**: Improved performance for long sessions - -### Cost Optimization - -```typescript -// Use cheaper model for summarization -import { LlmEventSummarizer, LLMRegistry } from "@iqai/adk"; - -const cheapModel = LLMRegistry.newLLM("gemini-2.0-flash-lite"); -const costEffectiveSummarizer = new LlmEventSummarizer(cheapModel); - -const config: EventsCompactionConfig = { - summarizer: costEffectiveSummarizer, - compactionInterval: 8, - overlapSize: 2, -}; -``` - -## Troubleshooting - -### Compaction Not Triggering - - - -If compaction isn't occurring, check: 1. `eventsCompactionConfig` is set on -the Runner 2. `compactionInterval` threshold has been reached 3. Summarizer is -properly configured 4. Session has enough invocations - - - -```typescript -// Debug compaction -runner.eventsCompactionConfig = { - compactionInterval: 2, // Lower threshold for testing - overlapSize: 1, -}; - -// Check after each invocation -const session = await runner.sessionService.getSession(...); -console.log('Invocation count:', new Set( - session.events.map(e => e.invocationId) -).size); -console.log('Has compaction:', session.events.some( - e => e.actions?.compaction -)); -``` - -### Summary Quality Issues - -```typescript -// Improve summary quality with custom prompt -const detailedSummarizer = new LlmEventSummarizer( - model, - `You are summarizing a conversation for context retention. - -Include: -- All important facts, decisions, and conclusions -- Named entities and specific details -- User preferences and requirements -- Open questions or unresolved issues - -Exclude: -- Greetings and pleasantries -- Redundant information -- Off-topic tangents - -Events to summarize: -{events} - -Provide a detailed yet concise summary:`, -); -``` - -## Advanced Topics - -### Compaction in Multi-Agent Systems - -```typescript -// Compaction works with agent hierarchies -const rootAgent = new LlmAgent({ - name: "Coordinator", - model: "gemini-2.5-pro", - subAgents: [assistantAgent, researchAgent], -}); - -const runner = new Runner({ - agent: rootAgent, - sessionService: mySessionService, - eventsCompactionConfig: { +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withEventsCompaction({ + summarizer: new BulletPointSummarizer(), compactionInterval: 10, overlapSize: 2, - }, -}); - -// All agent events are compacted together + }) + .build(); ``` -### Selective Compaction +## Inspecting compaction events + +A compaction event has no text content of its own — its payload is entirely in `event.actions.compaction`. You can detect and inspect compactions in your event loop: ```typescript -// Only compact certain event types -class SelectiveSummarizer implements EventsSummarizer { - async maybeSummarizeEvents(events: Event[]): Promise { - // Filter events to compact - const textEvents = events.filter( - e => e.content?.parts?.some(p => p.text) && !e.actions?.stateDelta, // Preserve state changes +for await (const event of runner.runAsync(request)) { + if (event.actions.compaction) { + const { startTimestamp, endTimestamp, compactedContent } = + event.actions.compaction; + console.log( + `Compacted events from ${new Date(startTimestamp * 1000).toISOString()} ` + + `to ${new Date(endTimestamp * 1000).toISOString()}`, ); - - if (textEvents.length < 5) return undefined; - - // Summarize only text events - // ... implementation + console.log("Summary:", compactedContent.parts?.[0]?.text); } } ``` -## Related Topics - -- [Event Actions](/docs/framework/events/event-actions) - Understanding event metadata and control flow -- [Session](/docs/framework/session-state-memory/session) - Managing conversation state -- [Memory](/docs/framework/session-state-memory/memory) - Long-term storage patterns -- [Run Config](/docs/framework/runtime/run-config) - Runner configuration options +To check after a run whether a session has ever been compacted: -Event Compaction is a powerful feature for managing long-running conversations efficiently. Start with the defaults and tune based on your specific use case and performance requirements. +```typescript +const session = await runner.sessionService.getSession( + "my-app", + "user-1", + "session-1", +); +const hasCompaction = session?.events.some(e => e.actions?.compaction) ?? false; +``` + +## Next steps + + + + + + diff --git a/apps/docs/content/docs/framework/events/event-actions.mdx b/apps/docs/content/docs/framework/events/event-actions.mdx index fc0300239..3436b9452 100644 --- a/apps/docs/content/docs/framework/events/event-actions.mdx +++ b/apps/docs/content/docs/framework/events/event-actions.mdx @@ -1,257 +1,213 @@ --- title: Event Actions -description: Understanding the EventActions system for state management and control flow +description: EventActions carries side-effect signals on every ADK-TS event — state changes, artifact versions, agent transfers, escalation, and compaction metadata. --- import { Callout } from "fumadocs-ui/components/callout"; -import { Tabs, Tab } from "fumadocs-ui/components/tabs"; +import { Cards, Card } from "fumadocs-ui/components/card"; -Event Actions are the mechanism by which ADK-TS components signal side effects, state changes, and control flow instructions. They enable powerful features like state management, agent transfers, and artifact tracking. +`EventActions` is the side-effect envelope attached to every `Event`. Rather than letting agents and tools mutate state directly, ADK-TS routes all changes through actions so the `Runner` can apply them atomically and record them in the session history. -## What Event Actions Are - -The `EventActions` class represents the actions attached to an event. It contains instructions for: - -- **State Changes**: Updates to session state via `stateDelta` -- **Artifact Management**: Tracking file versions via `artifactDelta` -- **Control Flow**: Agent transfers and escalation signals -- **Tool Behavior**: Skipping summarization and authentication requests - -## EventActions Structure +## Full reference ```typescript import { EventActions } from "@iqai/adk"; -class EventActions { - // Control tool response processing - skipSummarization?: boolean; - - // State management - stateDelta: Record = {}; +const actions = new EventActions({ + // Prevent the LLM from summarising this tool response — + // used when the raw result should be shown directly. + skipSummarization: true, - // Artifact versioning - artifactDelta: Record = {}; + // Key-value pairs to merge into session.state. + stateDelta: { task_status: "done", result_count: 3 }, - // Flow control - transferToAgent?: string; - escalate?: boolean; + // Artifact filenames → new version numbers saved this turn. + artifactDelta: { "report.pdf": 1 }, - // Authentication - requestedAuthConfigs?: Record; -} -``` + // Route the next turn to a different agent. + transferToAgent: "BillingAgent", -## Action Types + // Signal that a LoopAgent or parent agent should stop iterating. + escalate: true, -### State Management + // OAuth / API-key auth configs requested by tools. + // Keys are function call IDs, not tool names. + requestedAuthConfigs: { "call-abc123": { type: "oauth" } }, -State deltas track changes to session state: + // Set by the compaction system — marks this event as a summary + // that replaces a range of earlier events. + compaction: { + startTimestamp: 1700000000, + endTimestamp: 1700001000, + compactedContent: { role: "model", parts: [{ text: "Summary…" }] }, + }, - - -```typescript -const actions = new EventActions({ - stateDelta: { - 'user_preference': 'dark_mode', - 'last_activity': Date.now(), - 'temp_calculation': 42 // Will be filtered out (temp_ prefix) - } + // Rewind the session to just before this invocation ID. + rewindBeforeInvocationId: "abc123", }); +``` -const event = new Event({ -author: 'MyAgent', -actions: actions, -content: { parts: [{ text: 'Updated preferences' }] } -}); +All fields are optional and default to empty / undefined. You only set the ones relevant to what just happened. -```` - +## stateDelta - -```typescript -// In a tool or agent callback -context.state['user_preference'] = 'dark_mode'; -context.state['app:global_setting'] = 'enabled'; -context.state['temp_intermediate_result'] = calculation; +State changes travel inside `stateDelta` as a plain key-value map. The `Runner` merges each delta into `session.state` after appending the event, so every downstream event in the same run sees the updated values. -// These changes automatically populate the next event's stateDelta -```` - - - +```typescript +import { AgentBuilder } from "@iqai/adk"; +import type { CallbackContext } from "@iqai/adk"; + +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withAfterAgentCallback(async (ctx: CallbackContext) => { + // Writing to ctx.state is the normal way to update state — + // the framework populates stateDelta from these writes automatically. + ctx.state["task_status"] = "complete"; + ctx.state["app:theme"] = "dark"; // app: prefix → persisted app-wide + ctx.state["user:lang"] = "en"; // user: prefix → persisted per user + ctx.state["temp_scratch"] = "xyz"; // temp_ prefix → not persisted + return undefined; + }) + .build(); +``` - +When you read `event.actions.stateDelta` in your event loop, you see exactly which keys changed during that event — useful for driving reactive UI updates. -State keys with `temp_` prefix are excluded from persistence to avoid -cluttering permanent session state. +### State key prefixes - +| Prefix | Scope | Persisted | +| -------- | ------------------------- | --------- | +| `app:` | All users of this app | Yes | +| `user:` | This user across sessions | Yes | +| `temp_` | Current turn only | No | +| _(none)_ | This session | Yes | -### Artifact Management +## artifactDelta -Artifact deltas track file saves and versions: +When a tool or callback saves an artifact, the framework records the filename and version in `artifactDelta`. Your event loop can use this to refresh UI file viewers or trigger downstream processing without polling. ```typescript -const actions = new EventActions({ - artifactDelta: { - "report.pdf": 1, - "analysis.json": 3, - "diagram.png": 1, - }, -}); +for await (const event of runner.runAsync(request)) { + if (Object.keys(event.actions.artifactDelta).length > 0) { + for (const [filename, version] of Object.entries( + event.actions.artifactDelta, + )) { + console.log(`${filename} saved at version ${version}`); + } + } +} ``` -### Control Flow Actions - -#### Agent Transfer +## transferToAgent -Transfer control to another agent: +Setting `transferToAgent` tells the `Runner` to hand the next turn to a different named agent. This is the mechanism behind multi-agent routing: a routing agent yields an event with this field set, and the framework looks up the target agent and continues from there. ```typescript -const actions = new EventActions({ - transferToAgent: "BillingAgent", -}); - -// The framework will route the next turn to BillingAgent -``` - -#### Escalation - -Signal that a loop or process should terminate: +// Inside a custom BaseAgent.runAsyncImpl() +import { Event, EventActions } from "@iqai/adk"; -```typescript -const actions = new EventActions({ - escalate: true, +yield new Event({ + author: this.name, + invocationId: ctx.invocationId, + content: { parts: [{ text: "Routing you to billing support." }] }, + actions: new EventActions({ transferToAgent: "BillingAgent" }), }); - -// Useful in loop agents or error handling ``` -### Tool Behavior Control - -#### Skip Summarization +## escalate -Prevent the LLM from summarizing tool results: +`escalate: true` signals that the current loop or sub-agent chain should terminate. `LoopAgent` checks this flag after each iteration — when it sees it, the loop stops and control returns to the caller. ```typescript -const actions = new EventActions({ - skipSummarization: true, +// Stop a LoopAgent when the task is complete +yield new Event({ + author: this.name, + invocationId: ctx.invocationId, + actions: new EventActions({ escalate: true }), }); - -// Tool result will be displayed directly without LLM processing ``` -#### Authentication Requests +## skipSummarization -Signal that tools need authentication: +By default, when a tool returns a result, the LLM generates a human-readable summary of that result before the agent responds. Set `skipSummarization: true` on the function-response event to bypass that step and use the raw tool output directly. ```typescript -const actions = new EventActions({ - requestedAuthConfigs: { - "gmail-tool": { - type: "oauth", - provider: "gmail", - }, - }, -}); +// The framework sets this automatically when you use certain tool patterns. +// You can also set it manually in custom tool implementations. +actions: new EventActions({ skipSummarization: true }); ``` -## Working with Actions in Events +## requestedAuthConfigs -### Detecting Actions - -Check for actions in received events: +Tools that need OAuth or API-key credentials populate `requestedAuthConfigs` with one entry per pending tool invocation. The key is the **function call ID** (not the tool name) — the framework uses it to match the auth response back to the specific call that triggered it. ```typescript -for await (const event of runner.runAsync(query, session)) { - if (event.actions) { - // Check for state changes - if (Object.keys(event.actions.stateDelta).length > 0) { - console.log("State updated:", event.actions.stateDelta); - } - - // Check for artifact updates - if (Object.keys(event.actions.artifactDelta).length > 0) { - console.log("Artifacts saved:", event.actions.artifactDelta); - } - - // Check for control signals - if (event.actions.transferToAgent) { - console.log(`Transferring to: ${event.actions.transferToAgent}`); - } - - if (event.actions.escalate) { - console.log("Escalation signal received"); +// Reading auth requests from events +for await (const event of runner.runAsync(request)) { + if (event.actions.requestedAuthConfigs) { + for (const [functionCallId, config] of Object.entries( + event.actions.requestedAuthConfigs, + )) { + console.log(`Function call "${functionCallId}" needs auth:`, config); } } } ``` -### Creating Events with Actions +## compaction -When building custom agents: +When [event compaction](/docs/framework/events/compaction) runs, it produces an event whose entire purpose is to carry a summary of earlier events. That summary lives in `actions.compaction`. You do not set this field manually — the compaction system manages it. ```typescript -import { Event, EventActions } from "@iqai/adk"; - -export class CustomAgent extends BaseAgent { - async *runAsyncImpl(context: InvocationContext): AsyncGenerator { - // Perform some logic... - - const actions = new EventActions({ - stateDelta: { - task_status: "completed", - result_count: 5, - }, - transferToAgent: "ReviewAgent", - }); - - yield new Event({ - author: this.name, - invocationId: context.invocationId, - content: { - parts: [{ text: "Task completed, transferring for review" }], - }, - actions: actions, - }); - } +interface EventCompaction { + startTimestamp: number; // earliest event covered by this summary + endTimestamp: number; // latest event covered by this summary + compactedContent: Content; // { role: "model", parts: [{ text: "…" }] } } ``` -## State Scope Patterns +## rewindBeforeInvocationId -Use prefixes to indicate state scope: +The [Rewind](/docs/framework/runtime/rewind) feature sets this field to mark the invocation that should be rolled back. Like `compaction`, this is managed by the framework — you will see it when iterating events on a rewound session, but you do not set it yourself. -- **`app:`** - Application-wide settings -- **`user:`** - User-specific preferences across sessions -- **`temp_`** - Temporary values (filtered from persistence) -- **No prefix** - Session-specific state +## Creating events with actions + +When building a custom `BaseAgent`, construct `EventActions` in the same `new Event(…)` call: ```typescript -context.state["app:theme"] = "dark"; // App setting -context.state["user:language"] = "en"; // User preference -context.state["temp_calculation"] = result; // Temporary (not persisted) -context.state["current_task"] = "processing"; // Session state +import { Event, EventActions } from "@iqai/adk"; +import type { InvocationContext } from "@iqai/adk"; + +async *runAsyncImpl(ctx: InvocationContext): AsyncGenerator { + const result = await doWork(); + + yield new Event({ + author: this.name, + invocationId: ctx.invocationId, + content: { parts: [{ text: `Done. Found ${result.count} items.` }] }, + actions: new EventActions({ + stateDelta: { result_count: result.count, task_status: "complete" }, + }), + }); +} ``` -## Best Practices - -### Action Design - -- **Use specific actions**: Be explicit about what each action signals -- **Combine logically**: Group related state changes in one event -- **Document side effects**: Clear comments for complex action combinations - -### State Management - -- **Minimize state**: Only store what's needed across turns -- **Use appropriate prefixes**: Indicate scope and persistence intent -- **Avoid large objects**: Keep state values reasonably sized - -### Control Flow - -- **Single responsibility**: One control action per event when possible -- **Clear transfers**: Use descriptive agent names for transfers -- **Document escalation**: Explain why escalation occurs - -Event Actions provide the foundation for sophisticated agent behaviors and seamless component communication in ADK-TS applications. +## Next steps + + + + + + diff --git a/apps/docs/content/docs/framework/events/index.mdx b/apps/docs/content/docs/framework/events/index.mdx index 15caf7612..88622df73 100644 --- a/apps/docs/content/docs/framework/events/index.mdx +++ b/apps/docs/content/docs/framework/events/index.mdx @@ -1,177 +1,118 @@ --- -title: Overview -description: Understanding ADK-TS's event system for communication and control flow between components +title: Events +description: Events are the fundamental unit of communication in ADK-TS — every message, tool call, and state change flows through the event system as your agent runs. --- +import { Cards, Card } from "fumadocs-ui/components/card"; import { Callout } from "fumadocs-ui/components/callout"; -import { Card, Cards } from "fumadocs-ui/components/card"; +import { Mermaid } from "@/components/mdx/mermaid"; -Events are the fundamental units of information flow within ADK-TS. They represent every significant occurrence during an agent's interaction lifecycle, from initial user input to the final response and all the steps in between. +Every interaction with an ADK-TS agent produces a stream of `Event` objects. An event is an immutable record of one thing that happened during a run: a user message, an agent reply, a tool call, a tool result, or a state update. Your application consumes this stream to display responses, react to tool activity, and read state changes. -## What Events Are +## Event structure -An `Event` in ADK-TS is an immutable record representing a specific point in the agent's execution. It captures: +`Event` extends `LlmResponse` and adds ADK-TS-specific metadata: -- **User messages** - Direct input from end users -- **Agent replies** - Responses and actions from agents -- **Tool interactions** - Function calls and their results -- **State changes** - Updates to session state and artifacts -- **Control signals** - Flow control and error conditions - - - -Events serve as the standard message format between the user interface, the -`Runner`, agents, the LLM, and tools. Everything flows as an `Event`. - - +```typescript +import { Event, EventActions } from "@iqai/adk"; -## Event Structure +// Fields on every Event +event.id; // unique 8-char identifier +event.invocationId; // ID shared by all events in one run +event.author; // "user" or the agent's name +event.timestamp; // seconds since Unix epoch +event.content; // { role, parts } payload — text, function calls, results +event.partial; // true while the LLM is still streaming this event +event.actions; // EventActions — state deltas, artifact updates, control signals +event.longRunningToolIds; // present when a tool is running in the background +event.branch; // dotted agent path for multi-agent hierarchies +``` -Events build upon the basic `LlmResponse` structure by adding essential ADK-TS-specific metadata: +Two helper methods let you inspect content without parsing `parts` manually: ```typescript -import { Event, EventActions } from "@iqai/adk"; - -class Event extends LlmResponse { - // Core identification - id: string; // Unique event identifier - invocationId: string; // ID for the whole interaction run - author: string; // 'user' or agent name - timestamp: number; // Creation time (seconds since epoch) - - // Content and flow control - content?: any; // Event payload (text, function calls, etc.) - partial?: boolean; // True for streaming chunks - turnComplete?: boolean; // True when agent's turn is finished - - // ADK-TS-specific features - actions: EventActions; // Side-effects & control signals - longRunningToolIds?: Set; // Background tool execution - branch?: string; // Hierarchy path for multi-agent scenarios - - // Helper methods - isFinalResponse(): boolean; - getFunctionCalls(): any[]; - getFunctionResponses(): any[]; -} +event.getFunctionCalls(); // { name, args }[] — tool calls the agent wants to make +event.getFunctionResponses(); // { name, response }[] — results returned from tools +event.isFinalResponse(); // true when this event is ready to display ``` -## Event Types +## How events flow -### User Events + R[Runner] + R --> A[Agent] + A -->|text chunk| E1[Event partial=true] + A -->|tool call| E2[Event functionCall] + E2 --> T[Tool executes] + T -->|result| E3[Event functionResponse] + E3 --> A + A -->|final text| E4[Event isFinalResponse=true] + E1 & E2 & E3 & E4 --> S[Session history] +`} +/> -Events with `author: 'user'` represent direct input from the end-user. +The `Runner` drives this loop. It calls `agent.runAsync()`, collects every yielded event, appends it to the session, and re-emits it to your `for await` loop. When the agent is finished, the stream ends. -### Agent Events +## Quick start -Events with `author: 'AgentName'` represent output or actions from a specific agent. +`runner.runAsync()` returns an async generator. Iterate it to process events as they arrive: + +```typescript +import { AgentBuilder } from "@iqai/adk"; + +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withInstruction("You are a helpful assistant.") + .build(); + +const events = runner.runAsync({ + userId: "user-1", + sessionId: "session-1", + newMessage: { role: "user", parts: [{ text: "What is 2 + 2?" }] }, +}); + +for await (const event of events) { + if (event.isFinalResponse() && event.content?.parts?.[0]?.text) { + console.log(event.content.parts[0].text); + } +} +``` -### Tool Events + + `isFinalResponse()` returns `true` when the event is not a streaming chunk, + not a function call, and not a function response — meaning it is safe to + display to the user. + -Events containing function calls or responses for tool execution. +## Session history -### Control Events +Every event is appended to `session.events` in chronological order. Reading `session.events` after a run gives you a full, auditable trace of everything that happened — useful for debugging, replaying state, and building audit logs. -Events carrying state changes, control flow signals, or configuration updates. +State changes travel inside `event.actions.stateDelta`. The `Runner` applies each delta to `session.state` as it processes events, so your agent's state is always up to date by the time you receive the event in your loop. -## Core Concepts +## Next steps - - -## Quick Example - -Here's a basic example of processing events from an agent: - -```typescript -import { LlmAgent, Event } from "@iqai/adk"; - -// Process events from an agent interaction -for await (const event of runner.runAsync(query, session)) { - console.log(`Event from: ${event.author}`); - - // Handle different event types - if (event.content?.parts?.[0]?.text) { - console.log("Text:", event.content.parts[0].text); - } - - if (event.getFunctionCalls().length > 0) { - console.log("Tool calls:", event.getFunctionCalls()); - } - - if (event.actions?.stateDelta) { - console.log("State changes:", event.actions.stateDelta); - } - - if (event.isFinalResponse()) { - console.log("Final response detected"); - } -} -``` - -## Event Flow in ADK-TS - -Events enable communication throughout the ADK-TS system: - -1. **User Input** → Creates user event -2. **Agent Processing** → Generates agent events with text or tool calls -3. **Tool Execution** → Creates tool response events -4. **State Updates** → Tracked via event actions -5. **Final Response** → Delivered as final event ready for display - - - -The sequence of events recorded in `session.events` provides a complete, -chronological history of an interaction, invaluable for debugging, auditing, -and understanding agent behavior. - - - -## Key Features - -### State Management - -Events carry state changes through `actions.stateDelta`, allowing components to update session state reactively. - -### Artifact Tracking - -File operations and artifact versions are tracked through `actions.artifactDelta`. - -### Control Flow - -Events can signal agent transfers (`transferToAgent`), loop termination (`escalate`), and other control flow changes. - -### Streaming Support - -Events support real-time streaming with `partial` flags for progressive content delivery. - -### Error Handling - -Events inherit error properties (`errorCode`, `errorMessage`) for comprehensive error tracking. - -Events are the backbone of the ADK-TS communication system, enabling powerful features like state management, tool orchestration, and complex agent workflows. Explore the detailed guides above to learn how to leverage events effectively in your applications. diff --git a/apps/docs/content/docs/framework/events/meta.json b/apps/docs/content/docs/framework/events/meta.json index a516f8f35..5ea1fc666 100644 --- a/apps/docs/content/docs/framework/events/meta.json +++ b/apps/docs/content/docs/framework/events/meta.json @@ -2,7 +2,6 @@ "title": "Events", "icon": "Bell", "pages": [ - "index", "event-actions", "working-with-events", "compaction", diff --git a/apps/docs/content/docs/framework/events/patterns.mdx b/apps/docs/content/docs/framework/events/patterns.mdx index 2a0b68363..c51742f35 100644 --- a/apps/docs/content/docs/framework/events/patterns.mdx +++ b/apps/docs/content/docs/framework/events/patterns.mdx @@ -1,691 +1,261 @@ --- title: Event Patterns -description: Common architectural patterns and best practices for event-driven ADK-TS applications +description: Practical ADK-TS event patterns — chat loops, tool activity indicators, agent transfer tracking, state replay, and audit logs built on the real framework APIs. --- import { Callout } from "fumadocs-ui/components/callout"; import { Tabs, Tab } from "fumadocs-ui/components/tabs"; +import { Cards, Card } from "fumadocs-ui/components/card"; -This guide covers common architectural patterns for building robust, scalable event-driven applications with ADK-TS. +These patterns address concrete problems you encounter when building with ADK-TS events. Each one is grounded in the actual APIs — `runner.runAsync()`, `event.actions`, `session.events`, and the helper methods on `Event`. -## Architectural Patterns +## Multi-turn chat loop -### Event Sourcing - -Store all application state changes as a sequence of events: +A production chat loop needs to handle streaming chunks, tool activity, agent transfers, and the final reply in a single pass. This pattern separates each concern without nested conditionals: ```typescript -class EventStore { - private events: Event[] = []; - private snapshots = new Map(); - - append(event: Event) { - this.events.push(event); - this.updateSnapshot(event); - } - - getEventHistory(sessionId: string): Event[] { - return this.events.filter( - e => e.invocationId === sessionId || e.content?.sessionId === sessionId, - ); - } - - rebuildState(sessionId: string): Record { - const state: Record = {}; - const events = this.getEventHistory(sessionId); - - for (const event of events) { - if (event.actions?.stateDelta) { - Object.assign(state, event.actions.stateDelta); - } +import { AgentBuilder } from "@iqai/adk"; + +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withInstruction("You are a helpful assistant.") + .build(); + +async function chat(userId: string, sessionId: string, message: string) { + let streamBuffer = ""; + + for await (const event of runner.runAsync({ + userId, + sessionId, + newMessage: { role: "user", parts: [{ text: message }] }, + })) { + // 1. Accumulate streaming chunks for real-time display + if (event.partial && event.content?.parts?.[0]?.text) { + streamBuffer += event.content.parts[0].text; + onStreamChunk(streamBuffer); + continue; } - return state; - } - - private updateSnapshot(event: Event) { - if (event.actions?.stateDelta) { - const sessionId = event.invocationId; - const currentSnapshot = this.snapshots.get(sessionId) || {}; - this.snapshots.set(sessionId, { - ...currentSnapshot, - ...event.actions.stateDelta, - }); + // 2. Show tool activity while the agent is working + const calls = event.getFunctionCalls(); + if (calls.length > 0) { + onToolStart(calls.map(c => c.name)); + continue; } - } -} -``` - -### Command Query Responsibility Segregation (CQRS) -Separate event writing from reading: - - - -```typescript -interface Command { - type: string; - payload: any; - sessionId: string; -} - -class CommandHandler { -constructor(private eventStore: EventStore) {} - -async execute(command: Command): Promise { -const events: Event[] = []; - - switch (command.type) { - case 'USER_MESSAGE': - events.push(this.createUserMessageEvent(command)); - break; - case 'AGENT_RESPONSE': - events.push(this.createAgentResponseEvent(command)); - break; - case 'STATE_UPDATE': - events.push(this.createStateUpdateEvent(command)); - break; + const responses = event.getFunctionResponses(); + if (responses.length > 0) { + onToolEnd(responses.map(r => r.name)); + continue; } - for (const event of events) { - this.eventStore.append(event); + // 3. React to state changes as they happen + if (Object.keys(event.actions.stateDelta).length > 0) { + onStateChange(event.actions.stateDelta); } - return events; - + // 4. Display the completed reply + if (event.isFinalResponse()) { + const text = event.content?.parts?.[0]?.text ?? streamBuffer; + onFinalReply(text.trim()); + streamBuffer = ""; + } + } } +``` -private createUserMessageEvent(command: Command): Event { -return new Event({ -author: 'user', -invocationId: command.sessionId, -content: { parts: [{ text: command.payload.message }] } -}); -} -} +## Tool activity indicator -```` - +When an agent makes tool calls, users need feedback that something is happening. Function-call and function-response events bracket the tool execution window — use them to drive a loading state: - ```typescript -class QueryHandler { - constructor(private eventStore: EventStore) {} - - getConversationHistory(sessionId: string): Event[] { - return this.eventStore.getEventHistory(sessionId) - .filter(event => event.isFinalResponse()); - } - - getSessionState(sessionId: string): Record { - return this.eventStore.rebuildState(sessionId); - } - - getEventsByTimeRange(start: Date, end: Date): Event[] { - const startTimestamp = start.getTime() / 1000; - const endTimestamp = end.getTime() / 1000; - - return this.eventStore.getEventHistory('') - .filter(event => - event.timestamp >= startTimestamp && - event.timestamp <= endTimestamp - ); +for await (const event of runner.runAsync(request)) { + const calls = event.getFunctionCalls(); + const responses = event.getFunctionResponses(); + + if (calls.length > 0) { + const names = calls.map(c => c.name).join(", "); + setStatus(`Using tools: ${names}…`); + } else if (responses.length > 0) { + setStatus("Processing results…"); + } else if (event.isFinalResponse()) { + setStatus("idle"); } - - getEventStatistics(sessionId: string) { - const events = this.eventStore.getEventHistory(sessionId); - - return { - totalEvents: events.length, - messageCount: events.filter(e => e.content?.parts?.[0]?.text).length, - toolCalls: events.reduce((sum, e) => sum + e.getFunctionCalls().length, 0), - averageResponseTime: this.calculateAverageResponseTime(events) - }; - } -} -```` - - - - -```typescript -class EventBus { - private subscribers = new Map void>>(); - -subscribe(eventType: string, handler: (event: Event) => void) { -if (!this.subscribers.has(eventType)) { -this.subscribers.set(eventType, []); -} -this.subscribers.get(eventType)!.push(handler); } +``` -publish(event: Event) { -const eventType = this.getEventType(event); -const handlers = this.subscribers.get(eventType) || []; - - handlers.forEach(handler => { - try { - handler(event); - } catch (error) { - console.error('Event handler error:', error); - } - }); - -} - -private getEventType(event: Event): string { -if (event.author === 'user') return 'USER_EVENT'; -if (event.getFunctionCalls().length > 0) return 'TOOL_REQUEST'; -if (event.getFunctionResponses().length > 0) return 'TOOL_RESPONSE'; -if (event.isFinalResponse()) return 'FINAL_RESPONSE'; -return 'GENERAL_EVENT'; -} -} - -```` - - - -## State Management Patterns - -### State Machine - -Manage complex state transitions: +For long-running tools specifically, the `longRunningToolIds` field appears on a final-response event before the tool finishes. This is the signal to show a persistent background indicator rather than a brief spinner: ```typescript -interface StateTransition { - from: string; - to: string; - event: string; - condition?: (event: Event) => boolean; -} - -class StateMachine { - private currentState: string; - private transitions: StateTransition[] = []; - - constructor(initialState: string) { - this.currentState = initialState; - this.defineTransitions(); - } - - private defineTransitions() { - this.transitions = [ - { from: 'idle', to: 'processing', event: 'USER_MESSAGE' }, - { from: 'processing', to: 'waiting_tool', event: 'TOOL_REQUEST' }, - { from: 'waiting_tool', to: 'processing', event: 'TOOL_RESPONSE' }, - { from: 'processing', to: 'complete', event: 'FINAL_RESPONSE' }, - { from: 'complete', to: 'idle', event: 'USER_MESSAGE' }, - { - from: 'processing', - to: 'error', - event: 'ERROR', - condition: (event) => !!event.errorCode - } - ]; - } - - processEvent(event: Event): string { - const eventType = this.getEventType(event); - const transition = this.transitions.find(t => - t.from === this.currentState && - t.event === eventType && - (!t.condition || t.condition(event)) +for await (const event of runner.runAsync(request)) { + if (event.longRunningToolIds && event.isFinalResponse()) { + showBackgroundProcessingBanner( + `Running in background: ${[...event.longRunningToolIds].join(", ")}`, ); - - if (transition) { - console.log(`State transition: ${this.currentState} → ${transition.to}`); - this.currentState = transition.to; - this.onStateChange(transition.to, event); - } - - return this.currentState; } - private onStateChange(newState: string, event: Event) { - // Handle state-specific logic - switch (newState) { - case 'processing': - this.startProcessingIndicator(); - break; - case 'waiting_tool': - this.showToolExecutionStatus(); - break; - case 'complete': - this.hideLoadingIndicators(); - break; - case 'error': - this.showErrorMessage(event.errorMessage || 'Unknown error'); - break; - } + if (event.isFinalResponse() && event.content?.parts?.[0]?.text) { + displayReply(event.content.parts[0].text); } } -```` +``` -### Saga Pattern +## Agent transfer tracking -Manage complex, long-running workflows: +In multi-agent systems, `event.actions.transferToAgent` appears the moment the active agent hands off to another. Watch for it to update a "currently talking to" indicator in your UI: ```typescript -class ConversationSaga { - private steps: SagaStep[] = []; - private currentStep = 0; - private compensations: Array<() => Promise> = []; - - async execute(events: AsyncGenerator): Promise { - try { - for await (const event of events) { - await this.processStep(event); - } - } catch (error) { - await this.compensate(); - throw error; - } - } - - private async processStep(event: Event) { - const step = this.steps[this.currentStep]; - if (!step) return; - - try { - const result = await step.execute(event); - if (result.shouldProceed) { - this.currentStep++; - } +let activeAgent = "assistant"; - // Add compensation for this step - if (step.compensate) { - this.compensations.push(() => step.compensate!(result)); - } - } catch (error) { - throw new SagaError(`Step ${this.currentStep} failed`, error); - } +for await (const event of runner.runAsync(request)) { + if (event.actions.transferToAgent) { + activeAgent = event.actions.transferToAgent; + updateAgentLabel(activeAgent); } - private async compensate() { - console.log("Executing saga compensation"); - - // Execute compensations in reverse order - for (const compensation of this.compensations.reverse()) { - try { - await compensation(); - } catch (error) { - console.error("Compensation failed:", error); - } - } + if (event.isFinalResponse() && event.content?.parts?.[0]?.text) { + displayMessage({ author: event.author, text: event.content.parts[0].text }); } } ``` -## Observer Patterns - -### Event Listeners - -Implement reactive behaviors: +`event.author` identifies which agent produced each event. In a multi-agent conversation this lets you render a distinct name or avatar for each agent: ```typescript -class EventObserver { - private listeners = new Map>(); - - interface EventListener { - predicate: (event: Event) => boolean; - handler: (event: Event) => void | Promise; - once?: boolean; - } - - addListener( - name: string, - predicate: (event: Event) => boolean, - handler: (event: Event) => void | Promise, - once = false - ) { - if (!this.listeners.has(name)) { - this.listeners.set(name, new Set()); - } - - this.listeners.get(name)!.add({ - predicate, - handler, - once - }); - } - - async notify(event: Event) { - for (const [name, listenerSet] of this.listeners) { - const listenersToRemove: EventListener[] = []; - - for (const listener of listenerSet) { - if (listener.predicate(event)) { - try { - await listener.handler(event); - } catch (error) { - console.error(`Listener ${name} error:`, error); - } - - if (listener.once) { - listenersToRemove.push(listener); - } - } - } - - // Remove one-time listeners - listenersToRemove.forEach(listener => listenerSet.delete(listener)); - } - } +for await (const event of runner.runAsync(request)) { + if (!event.isFinalResponse()) continue; + const text = event.content?.parts?.[0]?.text; + if (!text) continue; + + appendMessage({ + author: event.author === "user" ? "You" : event.author, + text, + // event.branch is the dotted path, e.g. "coordinator.billing" + agentPath: event.branch, + }); } - -// Usage example -const observer = new EventObserver(); - -observer.addListener( - 'userMessages', - (event) => event.author === 'user', - (event) => console.log('User said:', event.content?.parts?.[0]?.text) -); - -observer.addListener( - 'toolCalls', - (event) => event.getFunctionCalls().length > 0, - (event) => logToolUsage(event.getFunctionCalls()) -); ``` -### Reactive Streams +## Reactive state updates -Create reactive event processing pipelines: +Rather than re-fetching `session.state` after each turn, read `event.actions.stateDelta` to learn exactly what changed. This is efficient for driving UI components that depend on individual state keys: ```typescript -class EventStream { - private transforms: Array<(event: Event) => Event | null> = []; - private filters: Array<(event: Event) => boolean> = []; - private subscribers: Array<(event: Event) => void> = []; - - filter(predicate: (event: Event) => boolean): EventStream { - this.filters.push(predicate); - return this; - } - - map(transform: (event: Event) => Event): EventStream { - this.transforms.push(transform); - return this; - } - - subscribe(handler: (event: Event) => void): () => void { - this.subscribers.push(handler); - - // Return unsubscribe function - return () => { - const index = this.subscribers.indexOf(handler); - if (index > -1) { - this.subscribers.splice(index, 1); - } - }; - } - - push(event: Event) { - // Apply filters - for (const filter of this.filters) { - if (!filter(event)) { - return; // Event filtered out - } - } +for await (const event of runner.runAsync(request)) { + const delta = event.actions.stateDelta; - // Apply transforms - let transformedEvent = event; - for (const transform of this.transforms) { - const result = transform(transformedEvent); - if (result === null) return; // Transform filtered out event - transformedEvent = result; - } - - // Notify subscribers - this.subscribers.forEach(subscriber => { - try { - subscriber(transformedEvent); - } catch (error) { - console.error("Stream subscriber error:", error); - } - }); - } + // Only update the parts of the UI that changed + if ("task_status" in delta) setTaskStatus(delta.task_status); + if ("result_count" in delta) setResultBadge(delta.result_count); + if ("app:theme" in delta) applyTheme(delta["app:theme"]); } - -// Usage example -const userMessageStream = new EventStream() - .filter(event => event.author === "user") - .map(event => ({ - ...event, - content: { - ...event.content, - timestamp: Date.now(), - }, - })) - .subscribe(event => processUserMessage(event)); ``` -## Error Handling Patterns + + Keys prefixed with `temp_` are ephemeral and not persisted. Skip them if you + are mirroring state to durable storage. + -### Circuit Breaker +## Artifact save notifications -Prevent cascade failures: +`event.actions.artifactDelta` records every file saved during a turn. Use it to refresh previews or notify users without polling: ```typescript -class EventCircuitBreaker { - private failures = 0; - private lastFailure = 0; - private state: "closed" | "open" | "half-open" = "closed"; - - constructor( - private maxFailures = 5, - private timeout = 60000, // 1 minute - ) {} - - async execute(event: Event, handler: (event: Event) => Promise) { - if (this.state === "open") { - if (Date.now() - this.lastFailure > this.timeout) { - this.state = "half-open"; - } else { - throw new Error("Circuit breaker is open"); - } - } - - try { - const result = await handler(event); - this.onSuccess(); - return result; - } catch (error) { - this.onFailure(); - throw error; - } - } - - private onSuccess() { - this.failures = 0; - this.state = "closed"; - } - - private onFailure() { - this.failures++; - this.lastFailure = Date.now(); - - if (this.failures >= this.maxFailures) { - this.state = "open"; +for await (const event of runner.runAsync(request)) { + for (const [filename, version] of Object.entries( + event.actions.artifactDelta, + )) { + if (filename.endsWith(".pdf")) { + refreshPdfPreview(filename, version); + } else if (filename.endsWith(".csv")) { + refreshDataTable(filename, version); } } } ``` -### Retry with Backoff +## Rebuild state from session history -Handle transient failures: +`session.events` is the full chronological record of every event in a session. Walk the `stateDelta` fields in order to reconstruct what `session.state` looked like at any point — useful for debugging, auditing, or rewinding to a past turn: ```typescript -class RetryHandler { - async withRetry( - operation: () => Promise, - maxRetries = 3, - baseDelay = 1000, - ): Promise { - let lastError: Error; - - for (let attempt = 0; attempt <= maxRetries; attempt++) { - try { - return await operation(); - } catch (error) { - lastError = error as Error; - - if (attempt === maxRetries) { - break; // Don't delay on last attempt - } - - const delay = this.calculateDelay(attempt, baseDelay); - console.warn( - `Attempt ${attempt + 1} failed, retrying in ${delay}ms:`, - error, - ); - await this.delay(delay); - } - } +import type { Session } from "@iqai/adk"; - throw new Error( - `Operation failed after ${maxRetries + 1} attempts: ${ - lastError!.message - }`, - ); - } - - private calculateDelay(attempt: number, baseDelay: number): number { - // Exponential backoff with jitter - const exponentialDelay = baseDelay * Math.pow(2, attempt); - const jitter = Math.random() * 0.1 * exponentialDelay; - return exponentialDelay + jitter; - } - - private delay(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)); - } -} -``` - -## Performance Patterns - -### Event Batching +function rebuildStateAt( + session: Session, + beforeTimestamp: number, +): Record { + const state: Record = {}; -Process events in batches for efficiency: + for (const event of session.events) { + if (event.timestamp >= beforeTimestamp) break; + if (event.actions?.compaction) continue; // skip summary events -```typescript -class EventBatcher { - private batch: Event[] = []; - private timer: NodeJS.Timeout | null = null; - - constructor( - private batchSize = 10, - private flushInterval = 100, // ms - private processor: (events: Event[]) => Promise, - ) {} - - add(event: Event) { - this.batch.push(event); - - if (this.batch.length >= this.batchSize) { - this.flush(); - } else if (!this.timer) { - this.timer = setTimeout(() => this.flush(), this.flushInterval); + for (const [key, value] of Object.entries(event.actions.stateDelta)) { + if (!key.startsWith("temp_")) { + state[key] = value; + } } } - private async flush() { - if (this.timer) { - clearTimeout(this.timer); - this.timer = null; - } - - if (this.batch.length === 0) return; - - const eventsToProcess = [...this.batch]; - this.batch = []; - - try { - await this.processor(eventsToProcess); - } catch (error) { - console.error("Batch processing error:", error); - // Optionally re-queue failed events - } - } + return state; } ``` -### Event Deduplication +## Session audit log -Prevent duplicate event processing: +`session.events` gives you a complete, ordered trace of everything that happened — who said what, which tools were called, and what state changed. This pattern builds a human-readable log from that history: ```typescript -class EventDeduplicator { - private processedEvents = new Set(); - private cleanupInterval: NodeJS.Timeout; - - constructor(private ttl = 300000) { - // 5 minutes - // Periodic cleanup of old event IDs - this.cleanupInterval = setInterval(() => { - this.cleanup(); - }, ttl); - } +import type { Session } from "@iqai/adk"; - isDuplicate(event: Event): boolean { - const key = this.getEventKey(event); - return this.processedEvents.has(key); - } +function buildAuditLog(session: Session): string[] { + const lines: string[] = []; - markProcessed(event: Event) { - const key = this.getEventKey(event); - this.processedEvents.add(key); - } + for (const event of session.events) { + const ts = new Date(event.timestamp * 1000).toISOString(); - private getEventKey(event: Event): string { - // Create unique key based on content and context - return `${event.id}-${event.timestamp}-${event.author}`; - } + const calls = event.getFunctionCalls(); + const responses = event.getFunctionResponses(); - private cleanup() { - // Simple cleanup - in production, use time-based expiry - if (this.processedEvents.size > 10000) { - this.processedEvents.clear(); + if (calls.length > 0) { + for (const call of calls) { + lines.push(`[${ts}] ${event.author} called tool "${call.name}"`); + } + } else if (responses.length > 0) { + for (const r of responses) { + lines.push(`[${ts}] Tool "${r.name}" returned a result`); + } + } else if (event.content?.parts?.[0]?.text && !event.partial) { + lines.push(`[${ts}] ${event.author}: ${event.content.parts[0].text}`); } - } - destroy() { - clearInterval(this.cleanupInterval); + const stateKeys = Object.keys(event.actions.stateDelta); + if (stateKeys.length > 0) { + lines.push(`[${ts}] State updated: ${stateKeys.join(", ")}`); + } } + + return lines; } ``` -## Best Practices - -### Event Design - -- **Use past tense**: Events represent things that happened -- **Include context**: Provide enough information for processing -- **Keep events immutable**: Never modify events after creation -- **Version events**: Plan for schema evolution - -### Error Handling - -- **Graceful degradation**: Continue processing other events when one fails -- **Idempotent handlers**: Design handlers to be safely retried -- **Dead letter queues**: Store failed events for analysis - -### Performance - -- **Filter early**: Apply filters before expensive operations -- **Batch when possible**: Group related operations -- **Use appropriate data structures**: Choose efficient storage for your access patterns - - - -These patterns provide building blocks for sophisticated event-driven -architectures. Choose patterns that fit your specific use case and complexity -requirements. - - - -Event patterns enable you to build maintainable, scalable agent applications that can handle complex workflows and provide excellent user experiences. +## Next steps + + + + + + diff --git a/apps/docs/content/docs/framework/events/streaming.mdx b/apps/docs/content/docs/framework/events/streaming.mdx index 61be2f911..045522c5c 100644 --- a/apps/docs/content/docs/framework/events/streaming.mdx +++ b/apps/docs/content/docs/framework/events/streaming.mdx @@ -1,531 +1,159 @@ --- -title: Event Streaming -description: Real-time event processing and streaming patterns in ADK-TS +title: Streaming +description: Stream text responses in real time from ADK-TS agents using textStreamFrom, collectTextFrom, and streamTextWithFinalEvent — purpose-built utilities for partial event handling. --- import { Callout } from "fumadocs-ui/components/callout"; import { Tabs, Tab } from "fumadocs-ui/components/tabs"; +import { Cards, Card } from "fumadocs-ui/components/card"; -ADK-TS event streaming enables real-time interaction with agents, allowing your applications to process and display responses as they're generated rather than waiting for complete responses. +ADK-TS agents emit text incrementally as the LLM generates it. Rather than waiting for the complete response, your application can display each fragment the moment it arrives. ADK-TS ships three utility functions in `@iqai/adk/utils` that abstract the `partial` flag and event iteration so you can work with a clean string stream. -## Understanding Streaming Events +## How streaming events work -Streaming events allow for progressive content delivery, improving user experience by showing partial results immediately. +When the LLM is mid-generation, the runner emits `Event` objects with `partial: true`. Each partial event contains a text fragment in `event.content.parts`. Once generation finishes, a final event arrives with `partial` unset (or `false`) — this is the event `isFinalResponse()` returns `true` for. -### Event Properties for Streaming - -Key properties that control streaming behavior: - -```typescript -interface Event { - partial?: boolean; // True for incomplete streaming chunks - turnComplete?: boolean; // True when the agent's turn is finished - content?: any; // Contains the partial or complete content - // ... other properties -} ``` - -## Basic Streaming Patterns - -### Simple Text Streaming - -Handle streaming text responses: - -```typescript -async function handleStreamingResponse(runner, query, session) { - let accumulatedText = ""; - - for await (const event of runner.runAsync(query, session)) { - if (event.content?.parts?.[0]?.text) { - if (event.partial) { - // Partial chunk - accumulate and display - accumulatedText += event.content.parts[0].text; - displayPartialText(accumulatedText); - } else { - // Complete chunk - accumulatedText += event.content.parts[0].text; - if (event.isFinalResponse()) { - displayFinalText(accumulatedText); - accumulatedText = ""; // Reset for next response - } - } - } - } -} +Event { partial: true, content: { parts: [{ text: "The answer" }] } } ← delta +Event { partial: true, content: { parts: [{ text: " is 42." }] } } ← delta +Event { partial: false, content: { parts: [{ text: "The answer is 42." }] } } ← final (full text) ``` -### Progressive UI Updates - - - -```typescript -import { useState, useEffect } from 'react'; - -function useEventStream(runner, query, session) { -const [streamingText, setStreamingText] = useState(''); -const [events, setEvents] = useState([]); -const [isStreaming, setIsStreaming] = useState(false); - -useEffect(() => { -let currentText = ''; - - async function processStream() { - setIsStreaming(true); - - try { - for await (const event of runner.runAsync(query, session)) { - setEvents(prev => [...prev, event]); - - if (event.content?.parts?.[0]?.text) { - if (event.partial) { - currentText += event.content.parts[0].text; - setStreamingText(currentText); - } else if (event.isFinalResponse()) { - currentText += event.content.parts[0].text; - setStreamingText(currentText); - currentText = ''; // Reset for next response - } - } - } - } finally { - setIsStreaming(false); - } - } - - processStream(); - -}, [query]); - -return { streamingText, events, isStreaming }; -} - -```` - - - -```typescript -class StreamingRenderer { - private container: HTMLElement; - private currentElement: HTMLElement | null = null; - - constructor(containerId: string) { - this.container = document.getElementById(containerId)!; - } - - async processStream(runner, query, session) { - let currentText = ''; - - for await (const event of runner.runAsync(query, session)) { - if (event.content?.parts?.[0]?.text) { - if (event.partial) { - currentText += event.content.parts[0].text; - this.updateCurrentElement(currentText); - } else if (event.isFinalResponse()) { - currentText += event.content.parts[0].text; - this.finalizeCurrentElement(currentText); - this.createNewElement(); - currentText = ''; - } - } - } - } - - private updateCurrentElement(text: string) { - if (!this.currentElement) { - this.createNewElement(); - } - this.currentElement!.textContent = text; - } - - private createNewElement() { - this.currentElement = document.createElement('div'); - this.currentElement.className = 'streaming-text'; - this.container.appendChild(this.currentElement); - } - - private finalizeCurrentElement(text: string) { - if (this.currentElement) { - this.currentElement.textContent = text; - this.currentElement.className = 'final-text'; - } - } -} -```` - - - - -```typescript -import { ref, onMounted } from 'vue'; - -export function useEventStream(runner, query, session) { - const streamingText = ref(''); - const events = ref([]); - const isStreaming = ref(false); - -const processStream = async () => { -let currentText = ''; -isStreaming.value = true; - - try { - for await (const event of runner.runAsync(query, session)) { - events.value.push(event); - - if (event.content?.parts?.[0]?.text) { - if (event.partial) { - currentText += event.content.parts[0].text; - streamingText.value = currentText; - } else if (event.isFinalResponse()) { - currentText += event.content.parts[0].text; - streamingText.value = currentText; - currentText = ''; - } - } - } - } finally { - isStreaming.value = false; - } - -}; - -onMounted(processStream); - -return { streamingText, events, isStreaming }; -} +You can handle `partial` events manually, but the streaming utilities make the common cases much simpler. -```` - - +## textStreamFrom -## Advanced Streaming Patterns - -### Multi-Stream Handling - -Handle multiple concurrent streams: +`textStreamFrom` converts a `runner.runAsync()` generator into a generator that yields only the raw text deltas. Use it when you need to print or display each word as it arrives and don't need any other event metadata: ```typescript -class MultiStreamManager { - private streams = new Map(); - - interface StreamState { - text: string; - isActive: boolean; - lastUpdate: number; - } - - async processEvent(event: Event) { - const streamId = event.invocationId; - - if (!this.streams.has(streamId)) { - this.streams.set(streamId, { - text: '', - isActive: true, - lastUpdate: Date.now() - }); - } - - const stream = this.streams.get(streamId)!; - - if (event.content?.parts?.[0]?.text) { - if (event.partial) { - stream.text += event.content.parts[0].text; - stream.lastUpdate = Date.now(); - this.onStreamUpdate(streamId, stream.text); - } else if (event.isFinalResponse()) { - stream.text += event.content.parts[0].text; - stream.isActive = false; - this.onStreamComplete(streamId, stream.text); - } - } - } - - private onStreamUpdate(streamId: string, text: string) { - // Update UI for specific stream - console.log(`Stream ${streamId} updated:`, text); - } - - private onStreamComplete(streamId: string, text: string) { - // Finalize stream display - console.log(`Stream ${streamId} completed:`, text); - this.streams.delete(streamId); - } -} -```` - -### Buffered Streaming - -Implement buffering for smoother display: - -```typescript -class BufferedStreamRenderer { - private buffer: string[] = []; - private isDisplaying = false; - private displayInterval = 50; // ms - - addChunk(text: string) { - this.buffer.push(text); - this.startDisplayLoop(); - } - - private startDisplayLoop() { - if (this.isDisplaying) return; - this.isDisplaying = true; - - const displayNext = () => { - if (this.buffer.length > 0) { - const chunk = this.buffer.shift()!; - this.displayChunk(chunk); - setTimeout(displayNext, this.displayInterval); - } else { - this.isDisplaying = false; - } - }; - - displayNext(); - } - - private displayChunk(chunk: string) { - // Implement smooth character-by-character display - const element = document.getElementById("output"); - if (element) { - element.textContent += chunk; - } - } +import { AgentBuilder } from "@iqai/adk"; +import { textStreamFrom } from "@iqai/adk/utils"; + +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withInstruction("You are a helpful assistant.") + .build(); + +const events = runner.runAsync({ + userId: "user-1", + sessionId: "session-1", + newMessage: { role: "user", parts: [{ text: "Tell me about black holes." }] }, +}); + +for await (const text of textStreamFrom(events)) { + process.stdout.write(text); // prints each chunk immediately } ``` -## Event Queue Management +`textStreamFrom` only yields text from `partial: true` events, so it silently skips function calls, function responses, and the non-streaming final event. -### Priority Queue +## collectTextFrom -Handle events with different priorities: +`collectTextFrom` accumulates all text deltas and resolves to the complete response string once the stream ends. Use it when you want the streaming benefit of low first-byte latency from the model but only need the final string in your code: ```typescript -class PriorityEventQueue { - private queues = { - high: [] as Event[], - normal: [] as Event[], - low: [] as Event[], - }; - - enqueue(event: Event, priority: "high" | "normal" | "low" = "normal") { - this.queues[priority].push(event); - this.processNext(); - } +import { collectTextFrom } from "@iqai/adk/utils"; - private processNext() { - const event = this.dequeue(); - if (event) { - this.handleEvent(event); - // Schedule next processing - setImmediate(() => this.processNext()); - } - } +const events = runner.runAsync({ + userId: "user-1", + sessionId: "session-1", + newMessage: { role: "user", parts: [{ text: "What is TypeScript?" }] }, +}); - private dequeue(): Event | null { - if (this.queues.high.length > 0) { - return this.queues.high.shift()!; - } - if (this.queues.normal.length > 0) { - return this.queues.normal.shift()!; - } - if (this.queues.low.length > 0) { - return this.queues.low.shift()!; - } - return null; - } - - private handleEvent(event: Event) { - // Process the event based on its type - if (event.partial) { - this.handleStreamingEvent(event); - } else if (event.isFinalResponse()) { - this.handleFinalEvent(event); - } - } -} +const fullText = await collectTextFrom(events); +console.log(fullText); // complete response ``` -## Real-Time Features - -### Live Status Updates +## streamTextWithFinalEvent -Show real-time processing status: +`streamTextWithFinalEvent` combines streaming and full event access in one call. It returns a `textStream` async generator for progressive display and a `finalEvent` promise that resolves with the last event once streaming completes — giving you both the text stream and the metadata (usage stats, tool calls, state deltas) from the final event: ```typescript -class LiveStatusManager { - private statusElement: HTMLElement; - private currentStatus = "idle"; +import { streamTextWithFinalEvent } from "@iqai/adk/utils"; - constructor(elementId: string) { - this.statusElement = document.getElementById(elementId)!; - } +const events = runner.runAsync({ + userId: "user-1", + sessionId: "session-1", + newMessage: { role: "user", parts: [{ text: "Calculate 17 × 23." }] }, +}); - updateFromEvent(event: Event) { - if (event.partial) { - this.setStatus("streaming", "Receiving response..."); - } else if (event.getFunctionCalls().length > 0) { - this.setStatus("processing", "Using tools..."); - } else if (event.getFunctionResponses().length > 0) { - this.setStatus("processing", "Processing results..."); - } else if (event.isFinalResponse()) { - this.setStatus("complete", "Response complete"); - setTimeout(() => this.setStatus("idle", "Ready"), 2000); - } - } +const { textStream, finalEvent } = streamTextWithFinalEvent(events); - private setStatus(status: string, message: string) { - this.currentStatus = status; - this.statusElement.textContent = message; - this.statusElement.className = `status ${status}`; - } +// Stream the text as it arrives +for await (const text of textStream) { + process.stdout.write(text); } -``` - -### Typing Indicators - -Show when agents are "thinking": - -```typescript -class TypingIndicator { - private element: HTMLElement; - private timer: NodeJS.Timeout | null = null; - - constructor(elementId: string) { - this.element = document.getElementById(elementId)!; - } - - show(agentName: string) { - this.element.textContent = `${agentName} is thinking...`; - this.element.style.display = "block"; - this.startAnimation(); - } - - hide() { - this.element.style.display = "none"; - this.stopAnimation(); - } - - private startAnimation() { - let dots = ""; - this.timer = setInterval(() => { - dots = dots.length >= 3 ? "" : dots + "."; - this.element.textContent = - this.element.textContent?.replace(/\.+$/, "") + dots; - }, 500); - } - private stopAnimation() { - if (this.timer) { - clearInterval(this.timer); - this.timer = null; - } - } +// Inspect the final event for metadata +const final = await finalEvent; +if (final) { + console.log("\nState changes:", final.actions.stateDelta); + console.log("Tool calls:", final.getFunctionCalls()); } ``` -## Performance Optimization - -### Debounced Updates - -Reduce UI update frequency: - -```typescript -function debounce void>( - func: T, - delay: number, -): (...args: Parameters) => void { - let timeoutId: NodeJS.Timeout; - - return (...args: Parameters) => { - clearTimeout(timeoutId); - timeoutId = setTimeout(() => func(...args), delay); - }; -} - -class OptimizedStreamRenderer { - private debouncedUpdate = debounce((text: string) => { - this.updateUI(text); - }, 16); // ~60fps + + `finalEvent` resolves after the `textStream` generator completes, so always + `await` the `finalEvent` after finishing the `for await` loop — not before. + - handleStreamingEvent(event: Event) { - if (event.partial && event.content?.parts?.[0]?.text) { - this.debouncedUpdate(event.content.parts[0].text); - } - } +## Choosing the right utility - private updateUI(text: string) { - // Expensive UI update operation - document.getElementById("output")!.textContent = text; - } -} -``` +| Utility | Use when | +| -------------------------- | -------------------------------------------------------------------- | +| `textStreamFrom` | You need progressive display and nothing else | +| `collectTextFrom` | You need the complete string; don't need per-chunk display | +| `streamTextWithFinalEvent` | You need progressive display AND metadata from the final event | +| Raw `runner.runAsync()` | You need to handle tool calls, state changes, or non-text events too | -### Memory Management +## Manual streaming loop -Handle large streams efficiently: +When you need finer control — for example, to handle both streaming text and tool events in the same loop — work with `event.partial` directly: ```typescript -class MemoryEfficientStream { - private maxEvents = 1000; - private events: Event[] = []; - - addEvent(event: Event) { - this.events.push(event); +let buffer = ""; - // Keep only recent events - if (this.events.length > this.maxEvents) { - this.events = this.events.slice(-this.maxEvents); - } +for await (const event of runner.runAsync(request)) { + // Accumulate streaming chunks + if (event.partial && event.content?.parts?.[0]?.text) { + buffer += event.content.parts[0].text; + updateUI(buffer); // show partial text in real time + continue; } - getRecentEvents(count: number = 100): Event[] { - return this.events.slice(-count); + // Handle function calls while streaming + if (event.getFunctionCalls().length > 0) { + showToolActivity(event.getFunctionCalls()); + continue; } - cleanup() { - this.events = []; + // Final response — the final event carries the full accumulated text, + // so use it directly rather than appending to the buffer (which already + // holds all the deltas and would duplicate the content). + if (event.isFinalResponse()) { + const text = event.content?.parts?.[0]?.text ?? buffer; + finalizeUI(text); + buffer = ""; } } ``` -## Error Handling in Streams - -### Stream Recovery - -Handle stream interruptions gracefully: - -```typescript -async function robustStreamProcessing(runner, query, session) { - let retries = 0; - const maxRetries = 3; - - while (retries < maxRetries) { - try { - for await (const event of runner.runAsync(query, session)) { - await processEvent(event); - } - break; // Success, exit retry loop - } catch (error) { - retries++; - console.warn(`Stream interrupted (attempt ${retries}):`, error); - - if (retries < maxRetries) { - await new Promise(resolve => setTimeout(resolve, 1000 * retries)); - } else { - console.error("Max retries exceeded, stream failed"); - throw error; - } - } - } -} -``` - - - -When implementing streaming, always handle network interruptions and provide -fallback mechanisms for critical functionality. - - - -Event streaming enables responsive, real-time agent interactions that significantly improve user experience in ADK-TS applications. +## Next steps + + + + + + diff --git a/apps/docs/content/docs/framework/events/working-with-events.mdx b/apps/docs/content/docs/framework/events/working-with-events.mdx index 9cbbc9ccf..3b924233b 100644 --- a/apps/docs/content/docs/framework/events/working-with-events.mdx +++ b/apps/docs/content/docs/framework/events/working-with-events.mdx @@ -1,401 +1,214 @@ --- title: Working with Events -description: Practical patterns and examples for handling events in ADK-TS applications +description: Read text content, detect final responses, inspect tool calls, track state changes, and filter event streams from ADK-TS agents. --- import { Callout } from "fumadocs-ui/components/callout"; import { Tabs, Tab } from "fumadocs-ui/components/tabs"; +import { Cards, Card } from "fumadocs-ui/components/card"; -This guide provides practical patterns for handling events in your ADK-TS applications, from basic event processing to advanced patterns for complex workflows. +The `runner.runAsync()` method returns an async generator of `Event` objects. This page shows the practical patterns for working with that stream: how to read different types of content, when an event is ready to display, and how to inspect side-effect signals like state changes and artifact saves. -## Event Processing Basics +## Reading text content -### Event Identification - -First, understand what type of event you're dealing with: +Text lives in `event.content.parts` as an array of `Part` objects. Each text part has a `text` string: ```typescript -import { Event } from "@iqai/adk"; - -function processEvent(event: Event) { - console.log(`Event from: ${event.author}`); - console.log(`Event ID: ${event.id}`); - console.log(`Invocation ID: ${event.invocationId}`); - - // Check event type - if (event.content?.parts) { - const functionCalls = event.getFunctionCalls(); - const functionResponses = event.getFunctionResponses(); - - if (functionCalls.length > 0) { - console.log("Type: Tool Call Request"); - functionCalls.forEach(call => { - console.log(` Tool: ${call.name}`, call.args); - }); - } else if (functionResponses.length > 0) { - console.log("Type: Tool Result"); - functionResponses.forEach(response => { - console.log(` Result from: ${response.name}`, response.response); - }); - } else if (event.content.parts[0]?.text) { - console.log( - event.partial ? "Type: Streaming Text" : "Type: Complete Text", - ); - } +for await (const event of runner.runAsync(request)) { + // Only write streaming deltas — the final non-partial event contains + // the full accumulated text, so writing both would duplicate output. + if (event.partial && event.content?.parts?.[0]?.text) { + process.stdout.write(event.content.parts[0].text); } } ``` -### Extracting Content - -Access different types of content from events: - - - -```typescript -function extractText(event: Event): string | null { - if (event.content?.parts?.[0]?.text) { - return event.content.parts[0].text; - } - return null; -} +During streaming, the LLM emits many events with `partial: true`, each carrying a text delta. Once generation finishes, a final non-partial event is emitted containing the full accumulated text. To avoid duplicating output, only process one of the two — use the [streaming utilities](/docs/framework/events/streaming) which handle this automatically, or filter by `event.partial` yourself. -// Usage -const text = extractText(event); -if (text) { -console.log("Message:", text); -} +## Detecting the final response -```` - +`event.isFinalResponse()` returns `true` when an event is safe to display as the agent's completed reply. Internally it checks that the event is not a streaming chunk, not a function call, and not a function response: - ```typescript -function processFunctionCalls(event: Event) { - const calls = event.getFunctionCalls(); - - for (const call of calls) { - console.log(`Tool requested: ${call.name}`); - console.log(`Arguments:`, call.args); - - // Dispatch based on tool name - switch (call.name) { - case 'search': - handleSearchRequest(call.args); - break; - case 'sendEmail': - handleEmailRequest(call.args); - break; - default: - console.log(`Unknown tool: ${call.name}`); - } +import { AgentBuilder } from "@iqai/adk"; + +const { runner } = await AgentBuilder.create("assistant") + .withModel("gemini-2.5-flash") + .withInstruction("You are a helpful assistant.") + .build(); + +for await (const event of runner.runAsync({ + userId: "user-1", + sessionId: "session-1", + newMessage: { + role: "user", + parts: [{ text: "Summarise quantum computing." }], + }, +})) { + if (event.isFinalResponse()) { + const text = event.content?.parts?.[0]?.text; + if (text) console.log("Final answer:", text); } } -```` - - - - -```typescript -function processFunctionResponses(event: Event) { - const responses = event.getFunctionResponses(); - -for (const response of responses) { -console.log(`Tool ${response.name} returned:`, response.response); - - // Handle specific tool responses - if (response.name === 'search') { - displaySearchResults(response.response); - } else if (response.name === 'sendEmail') { - showEmailConfirmation(response.response); - } +``` -} -} +`isFinalResponse()` also returns `true` in two special cases: -```` - - +- `event.actions.skipSummarization` is set — the raw tool result should be shown directly without an LLM summary +- `event.longRunningToolIds` is set — a background tool has started; signal a loading state to the user -## Event Filtering Patterns +## Inspecting tool calls and results -### Final Response Detection +Tool interactions produce two distinct event shapes. A function-call event carries what the agent wants to invoke; a function-response event carries what the tool returned. -Use the built-in helper to identify display-ready events: + + + `getFunctionCalls()` returns an array of `{ name, args }` objects — one entry per tool the agent is invoking in this turn: -```typescript -async function processAgentConversation(runner, query, session) { - let streamingText = ''; - - for await (const event of runner.runAsync(query, session)) { - // Handle streaming content - if (event.partial && event.content?.parts?.[0]?.text) { - streamingText += event.content.parts[0].text; - updateUI(streamingText); // Real-time updates + ```typescript + for await (const event of runner.runAsync(request)) { + const calls = event.getFunctionCalls(); + if (calls.length > 0) { + for (const call of calls) { + console.log(`Tool requested: ${call.name}`); + console.log("Args:", call.args); + } + } } - - // Handle final responses - if (event.isFinalResponse()) { - if (event.content?.parts?.[0]?.text) { - // Complete text response - const finalText = streamingText + (event.content.parts[0].text || ''); - displayFinalMessage(finalText.trim()); - streamingText = ''; // Reset - } else if (event.actions?.skipSummarization) { - // Raw tool result - const toolResponse = event.getFunctionResponses()[0]; - displayToolResult(toolResponse.response); - } else if (event.longRunningToolIds) { - // Background process - showProcessingIndicator(); + ``` + + + + `getFunctionResponses()` returns an array of `{ name, response }` objects — one entry per completed tool result: + + ```typescript + for await (const event of runner.runAsync(request)) { + const responses = event.getFunctionResponses(); + if (responses.length > 0) { + for (const result of responses) { + console.log(`Tool "${result.name}" returned:`, result.response); + } } } + ``` - // Handle actions - processEventActions(event); - } -} -```` - -### Custom Event Filters - -Create filters for specific event types: - -```typescript -// Filter by author -const agentEvents = events.filter(event => event.author !== "user"); -const userEvents = events.filter(event => event.author === "user"); - -// Filter by content type -const textEvents = events.filter( - event => event.content?.parts?.[0]?.text && !event.partial, -); - -const toolEvents = events.filter( - event => - event.getFunctionCalls().length > 0 || - event.getFunctionResponses().length > 0, -); - -// Filter by time range -const recentEvents = events.filter( - event => event.timestamp > Date.now() / 1000 - 3600, // Last hour -); -``` - -## State and Action Handling + + -### Processing State Changes +## Tracking state changes -Monitor and react to state updates: +State changes travel in `event.actions.stateDelta`. Each entry is a key-value pair the agent or a tool wrote to `context.state` during this turn. Reading these deltas lets you update your UI reactively without re-fetching the full session: ```typescript -function handleStateChanges(event: Event) { - if (!event.actions?.stateDelta) return; - - const changes = event.actions.stateDelta; - - for (const [key, value] of Object.entries(changes)) { - console.log(`State changed: ${key} = ${value}`); - - // React to specific state changes - switch (key) { - case "user_preference": - updateUserInterface(value); - break; - case "task_status": - updateTaskProgress(value); - break; - case "app:theme": - applyTheme(value); - break; +for await (const event of runner.runAsync(request)) { + const delta = event.actions.stateDelta; + if (Object.keys(delta).length > 0) { + for (const [key, value] of Object.entries(delta)) { + console.log(`State: ${key} = ${JSON.stringify(value)}`); } } } ``` -### Artifact Tracking - -Track file operations: - -```typescript -function handleArtifactChanges(event: Event) { - if (!event.actions?.artifactDelta) return; +Keys prefixed with `temp_` are ephemeral and are not persisted to the session. Keys prefixed with `app:` or `user:` have broader scope. See [State](/docs/framework/session-state-memory/state) for the full scoping rules. - const artifacts = event.actions.artifactDelta; +## Tracking artifact saves - for (const [filename, version] of Object.entries(artifacts)) { - console.log(`File updated: ${filename} (version ${version})`); +When an agent or tool saves a file, `event.actions.artifactDelta` records the filename and the version number that was written: - // Refresh file displays - if (filename.endsWith(".pdf")) { - refreshPdfViewer(filename); - } else if (filename.endsWith(".json")) { - refreshDataView(filename); +```typescript +for await (const event of runner.runAsync(request)) { + const saved = event.actions.artifactDelta; + if (Object.keys(saved).length > 0) { + for (const [filename, version] of Object.entries(saved)) { + console.log(`Saved: ${filename} (v${version})`); } } } ``` -## Advanced Event Patterns - -### Event Aggregation +## Filtering events -Combine related events for analysis: +A few common filters worth applying in your loop: ```typescript -class EventAggregator { - private events: Event[] = []; +for await (const event of runner.runAsync(request)) { + // Only events from agents (not the user message) + if (event.author === "user") continue; - addEvent(event: Event) { - this.events.push(event); - } + // Only complete text (skip streaming chunks) + if (event.partial) continue; - getConversationSummary(invocationId: string) { - const conversationEvents = this.events.filter( - event => event.invocationId === invocationId, - ); - - return { - totalEvents: conversationEvents.length, - userMessages: conversationEvents.filter(e => e.author === "user").length, - agentResponses: conversationEvents.filter(e => e.author !== "user") - .length, - toolCalls: conversationEvents.reduce( - (sum, e) => sum + e.getFunctionCalls().length, - 0, - ), - duration: this.getConversationDuration(conversationEvents), - }; - } + // Only events with text content + const text = event.content?.parts?.[0]?.text; + if (!text) continue; - private getConversationDuration(events: Event[]): number { - if (events.length < 2) return 0; - const sorted = events.sort((a, b) => a.timestamp - b.timestamp); - return sorted[sorted.length - 1].timestamp - sorted[0].timestamp; - } + console.log(`[${event.author}] ${text}`); } ``` -### Event Replay - -Recreate conversation state from event history: +For post-run analysis, filter `session.events` by invocation ID to isolate a single interaction: ```typescript -function replayEvents(events: Event[]): Record { - const state: Record = {}; - - for (const event of events.sort((a, b) => a.timestamp - b.timestamp)) { - if (event.actions?.stateDelta) { - for (const [key, value] of Object.entries(event.actions.stateDelta)) { - if (!key.startsWith("temp_")) { - state[key] = value; - } - } - } - } - - return state; -} +const run = session.events.filter(e => e.invocationId === myInvocationId); +const agentReplies = run.filter( + e => e.author !== "user" && e.isFinalResponse(), +); ``` -### Error Event Handling +## Handling errors -Handle errors gracefully: +`Event` inherits `errorCode` and `errorMessage` from `LlmResponse`. When the LLM or a safety filter rejects a request, the resulting event carries these fields: ```typescript -function processEventSafely(event: Event) { - try { - // Check for error conditions - if (event.errorCode) { - handleEventError(event); - return; - } - - // Normal processing - processEvent(event); - } catch (error) { - console.error("Error processing event:", error); - logEventError(event, error); +for await (const event of runner.runAsync(request)) { + if (event.errorCode) { + console.error(`Error ${event.errorCode}: ${event.errorMessage}`); + break; } -} -function handleEventError(event: Event) { - console.error(`Event error: ${event.errorCode}`); - console.error(`Message: ${event.errorMessage}`); - - // Show user-friendly error message - switch (event.errorCode) { - case "SAFETY_FILTER_TRIGGERED": - showSafetyWarning(); - break; - case "RATE_LIMIT_EXCEEDED": - showRateLimitMessage(); - break; - default: - showGenericError(event.errorMessage); + if (event.isFinalResponse()) { + // normal processing } } ``` -## Building Event-Driven UIs +## Identifying the agent that spoke -### React Integration +In multi-agent systems, `event.author` tells you which agent produced the event. Compare it against known agent names to route display logic: ```typescript -import { useState, useEffect } from "react"; - -function ConversationComponent({ runner, query, session }) { - const [events, setEvents] = useState([]); - const [isLoading, setIsLoading] = useState(false); - - useEffect(() => { - async function runConversation() { - setIsLoading(true); - const newEvents: Event[] = []; - - try { - for await (const event of runner.runAsync(query, session)) { - newEvents.push(event); - setEvents([...newEvents]); // Update UI with each event - } - } finally { - setIsLoading(false); - } - } - - runConversation(); - }, [query]); - - return ( -
- {events.map(event => ( - - ))} - {isLoading && } -
- ); +for await (const event of runner.runAsync(request)) { + if (!event.isFinalResponse()) continue; + const text = event.content?.parts?.[0]?.text; + if (!text) continue; + + if (event.author === "ResearchAgent") { + displayResearchResult(text); + } else if (event.author === "SummaryAgent") { + displaySummary(text); + } else { + displayGenericReply(text); + } } ``` -## Best Practices - -### Performance - -- **Filter early**: Apply filters before processing heavy operations -- **Batch updates**: Group related state changes when possible -- **Cleanup**: Remove event listeners and clear large event arrays - -### Error Handling - -- **Graceful degradation**: Continue processing other events if one fails -- **User feedback**: Provide clear error messages for failed operations -- **Logging**: Record detailed error information for debugging - -### Memory Management - -- **Event retention**: Only keep necessary events in memory -- **Pagination**: Use pagination for large event histories -- **Cleanup**: Clear temporary state and cached data - -Events provide the foundation for building responsive, interactive agent applications with ADK-TS. +## Next steps + + + + + +