Context Window

Automatically manage history for the Pipeline. Configure a ContextWindow with token and item budgets. Before each LLM call it summarizes older turns and truncates excess items so your agent preserves key memory without exceeding limits.

Context Window compress and truncate cycle

Context Window replaces manual context management. All token budgeting, history compression, and truncation is handled automatically through a single configuration object.

How Context Window Works

Context Window is configured on a Pipeline instance via the context_window parameter. It runs a two-step management cycle before every LLM call:

Step	Action	Purpose
1. Compress	Summarize old conversation turns via LLM	Preserve long-term memory without keeping every message
2. Truncate	Remove oldest non-protected items	Enforce hard token and item count limits

Three item types are always protected and never removed:

Protected Item	Reason
System message	Agent instructions must persist
Summary message	Compressed history is the agent’s long-term memory
Last user message	LLMs require the conversation to end with a user turn

from zrt import Pipeline, ContextWindow
from zrt.plugins import DeepgramSTT, OpenAILLM, CartesiaTTS, SileroVAD, TurnDetector
from zrt.inference import AICousticsDenoise

pipeline = Pipeline(
    stt=DeepgramSTT(),
    llm=OpenAILLM(),
    tts=CartesiaTTS(),
    vad=SileroVAD(),
    turn_detector=TurnDetector(model="echo-large"),

    # Configure context window management
    context_window=ContextWindow(
        max_tokens=4000,
        max_context_items=20,
        keep_recent_turns=3,
        max_tool_calls_per_turn=10,
    ),
    denoise=AICousticsDenoise(model_id="quail-vf-2.2-l-16khz"),
)

Configuration Parameters

Adjust these parameters to match your application’s token limits, cost targets, and expected tool-call patterns.

Parameter	Type	Default	Description
`max_tokens`	`int \| None`	`None`	• Token budget for the full history. • On overflow, old turns compress, then truncate.
`max_context_items`	`int \| None`	`None`	• Max items (messages + tool calls + results). • Either limit triggers compression/truncation.
`keep_recent_turns`	`int`	`3`	• Recent exchanges kept verbatim. • Older ones get summarized.
`max_tool_calls_per_turn`	`int`	`10`	• Max tool calls per user turn. • Prevents infinite tool-call loops.
`summary_llm`	`LLM \| None`	`None`	• Optional LLM for summaries. • Falls back to the main LLM. • Use a cheaper model to cut costs.

Processing Cycle

Before each LLM call, an internal process is automatically executed. This process performs two steps in sequence.

Step 1: Compress

When the context exceeds the token or item budget and there are enough old turns to compress (more user turns than keep_recent_turns), compression kicks in:

Split - Separate items into old turns and recent turns (keeping the last N user exchanges).
Render - Convert old items into human-readable text for the summarization prompt.
Summarize - Call the LLM (or summary_llm) to generate a concise summary.
Replace - Remove all old items and insert the summary as a system message tagged with the internal Summary source (rendered as [Conversation Summary]).

The summary preserves:

Key facts, names, and numbers
Decisions made and their reasoning
Tool/function call results and outcomes
Commitments or promises the assistant made
User objectives, preferences, and unresolved tasks

Step 2: Truncate

After compression (or if compression wasn’t needed), truncation enforces hard limits:

Remove the oldest non-protected items one at a time.
Function call/output pairs are removed together to avoid orphaned tool calls.
Continue until both max_tokens and max_context_items are satisfied.
If only protected items remain, stop even if still over budget.

How Tool Chaining Works

Context Window works with tool chaining. Here’s the lifecycle of a multi-tool turn:

User request

User says “Plan for Dubai” → LLM returns get_weather(Dubai).

First tool runs

Tool executes → result added to context → LLM called again.

Chained tool call

LLM returns get_clothing_advice(22°C) → execute → call LLM again.

Next tool call

LLM returns get_activity_suggestion(22°C, "jacket") → execute → call LLM.

Final response

LLM returns text “Dubai is 22°C, wear a jacket, go hiking!” → spoken by TTS.

That’s 3 tool calls + 1 text response = 4 rounds, well within max_tool_calls_per_turn=10.

Some LLMs (Anthropic Claude, OpenAI GPT-4o) can return multiple tool calls in a single response. These are collected and executed in parallel using asyncio.gather, then all results are added to context before the next LLM call. Google Gemini sends one tool call at a time (always sequential).

Example

A full example combining Context Window with tool chaining for a production-ready travel assistant:

import aiohttp
import zrt
from zrt import Agent, Pipeline, function_tool, ContextWindow
from zrt.plugins import DeepgramSTT, CartesiaTTS, OpenAILLM, SileroVAD, TurnDetector
from zrt.inference import AICousticsDenoise

AGENT_ID = "travel-assistant"

@function_tool
async def get_weather(city: str) -> dict:
    """Get the current weather temperature for a given city."""
    city_coords = {
        "dubai": (25.2048, 55.2708),
        "mumbai": (19.0760, 72.8777),
        "new york": (40.7128, -74.0060),
    }
    coords = city_coords.get(city.lower(), (25.2048, 55.2708))
    lat, lon = coords
    url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                data = await response.json()
                temp = data["current"]["temperature_2m"]
                return {"city": city, "temperature": temp, "unit": "Celsius"}
            else:
                return {"city": city, "temperature": 25, "unit": "Celsius", "note": "fallback"}

@function_tool
async def get_clothing_advice(temperature: float) -> dict:
    """Get clothing recommendation based on temperature."""
    if temperature > 35:
        advice = "Very light breathable clothes, hat, and sunscreen."
    elif temperature > 25:
        advice = "Light clothes like t-shirt and shorts."
    elif temperature > 15:
        advice = "Light jacket or sweater with comfortable pants."
    elif temperature > 5:
        advice = "Warm coat, scarf, and layered clothing."
    else:
        advice = "Heavy winter coat, gloves, hat, and thermal layers."
    return {"temperature": temperature, "clothing_advice": advice}

pipeline = Pipeline(
    stt=DeepgramSTT(),
    llm=OpenAILLM(),
    tts=CartesiaTTS(),
    vad=SileroVAD(),
    turn_detector=TurnDetector(model="echo-large"),

    # ── Context Window Configuration ───────────────────────────
    # max_tokens: token budget for the conversation (~8 city plans + chat).
    # max_context_items: max messages + tool calls before compression/truncation.
    # keep_recent_turns: recent exchanges kept verbatim; older ones summarized.
    # max_tool_calls_per_turn: safety limit to prevent infinite tool-call loops.
    context_window=ContextWindow(
        max_tokens=4000,
        max_context_items=20,
        keep_recent_turns=3,
        max_tool_calls_per_turn=10,
    ),
    denoise=AICousticsDenoise(model_id="quail-vf-2.2-l-16khz"),
)

class TravelAgent(Agent):
    def __init__(self):
        super().__init__(
            agent_id=AGENT_ID,
            instructions=(
                "You are a helpful travel assistant. When a user asks what to do in a city:\n"
                "1. FIRST call get_weather to get the temperature\n"
                "2. THEN call get_clothing_advice with that temperature\n"
                "3. Combine results into a natural spoken response (2-3 sentences max)."
            ),
            pipeline=pipeline,
            tools=[get_weather, get_clothing_advice],
        )

    async def on_enter(self) -> None:
        await self.session.say("Hi! I'm your travel assistant. Ask me about any city!")

    async def on_exit(self) -> None:
        pass

if __name__ == "__main__":
    # Pass the class itself (not an instance): serve() builds a fresh TravelAgent +
    # pipeline per call, which is required for correct per-call state under concurrent calls.
    zrt.serve(TravelAgent, on_ready=lambda: zrt.invoke(AGENT_ID, room=zrt.Room(playground=True)))

What’s Next

Context Management

How ChatContext records conversation memory.

Agent Handoffs

Share context across specialized agents.

References

Python

Examples

Chat Context

Compress and truncate long conversations.

SDK Reference

Context Window in the Python API reference.

Chat Context

ChatContext in the Python API reference.

Get started

Modalities

Configure Pipeline

Tools and Capabilities

Turn Detection and Interruptions

Telephony Agents

Context Management

Observability and Monitoring

Context Window

How Context Window Works

Configuration Parameters

Processing Cycle

Step 1: Compress

Step 2: Truncate

How Tool Chaining Works

Example

What’s Next

Context Management

Agent Handoffs

References

Examples

Chat Context

SDK Reference

Context Window

Chat Context

​How Context Window Works

​Configuration Parameters

​Processing Cycle

​Step 1: Compress

​Step 2: Truncate

​How Tool Chaining Works

​Example

​What’s Next

Context Management

Agent Handoffs

​References

​Examples

Chat Context

​SDK Reference

Context Window

Chat Context

How Context Window Works

Configuration Parameters

Processing Cycle

Step 1: Compress

Step 2: Truncate

How Tool Chaining Works

Example

What’s Next

References

Examples

SDK Reference