Module agents.llm.context_window

Backward-compatibility shim. ContextWindow now lives in the videosdk.agents.llm.context package.

Classes

class ContextWindow (*,
max_tokens: int | None = None,
max_context_items: int | None = None,
keep_recent_turns: int = 3,
max_tool_calls_per_turn: int = 10,
summary_llm: "'LLM | None'" = None)

Expand source code

class ContextWindow:
    """Policy layer that decides *when* to compress/truncate a ChatContext.

    Compression and truncation primitives live on ChatContext; this class
    holds the budgets and delegates to them.

    Args:
        max_tokens: Maximum estimated token budget.
        max_context_items: Maximum number of items.
        keep_recent_turns: Recent user-assistant exchanges kept verbatim
            during compression.
        max_tool_calls_per_turn: Retained for compatibility with callers.
        summary_llm: Optional dedicated LLM for summaries; falls back to the
            agent's main LLM.
    """

    def __init__(
        self,
        *,
        max_tokens: int | None = None,
        max_context_items: int | None = None,
        keep_recent_turns: int = 3,
        max_tool_calls_per_turn: int = 10,
        summary_llm: "LLM | None" = None,
    ) -> None:
        self.max_tokens = max_tokens
        self.max_context_items = max_context_items
        self.max_tool_calls_per_turn = max_tool_calls_per_turn
        self._keep_recent = keep_recent_turns
        self._summary_llm = summary_llm

    async def manage(self, ctx: ChatContext, llm: "LLM") -> None:
        """Run the full context management cycle: compress then truncate."""
        if self._needs_compression(ctx):
            try:
                active_llm = self._summary_llm or llm
                if active_llm is not None:
                    logger.info("Compressing context via summary generation")
                    await ctx.summarize(active_llm, keep_recent_turns=self._keep_recent)
                    logger.info(
                        f"Context compression complete. Size: {len(ctx.items)} items"
                    )
                else:
                    logger.warning("No LLM available for context compression")
            except Exception as e:
                logger.error(f"Error during context compression: {e}", exc_info=True)

        if self.max_tokens is not None or self.max_context_items is not None:
            before = len(ctx.items)
            ctx.truncate(max_items=self.max_context_items, max_tokens=self.max_tokens)
            after = len(ctx.items)
            if after < before:
                logger.info(f"Truncated context from {before} to {after} items")

    def _needs_compression(self, ctx: ChatContext) -> bool:
        """Check if context exceeds budget and has enough old turns to compress."""
        est = ctx.estimated_tokens()
        items = len(ctx.items)
        turns = ctx.turn_count()
        exceeds_tokens = self.max_tokens is not None and est > self.max_tokens
        exceeds_items = (
            self.max_context_items is not None and items > self.max_context_items
        )
        enough_turns = turns > self._keep_recent + 1
        needed = (exceeds_tokens or exceeds_items) and enough_turns
        logger.debug(
            f"Compression check: {est} tokens (max={self.max_tokens}), "
            f"{items} items (max={self.max_context_items}), {turns} turns, needed={needed}"
        )
        return needed

Policy layer that decides when to compress/truncate a ChatContext.

Compression and truncation primitives live on ChatContext; this class holds the budgets and delegates to them.

Args

max_tokens

Maximum estimated token budget.

max_context_items

Maximum number of items.

keep_recent_turns

Recent user-assistant exchanges kept verbatim during compression.

max_tool_calls_per_turn

Retained for compatibility with callers.

summary_llm

Optional dedicated LLM for summaries; falls back to the agent's main LLM.

Methods

async def manage(self, ctx: ChatContext, llm: "'LLM'") ‑> None

Expand source code

async def manage(self, ctx: ChatContext, llm: "LLM") -> None:
    """Run the full context management cycle: compress then truncate."""
    if self._needs_compression(ctx):
        try:
            active_llm = self._summary_llm or llm
            if active_llm is not None:
                logger.info("Compressing context via summary generation")
                await ctx.summarize(active_llm, keep_recent_turns=self._keep_recent)
                logger.info(
                    f"Context compression complete. Size: {len(ctx.items)} items"
                )
            else:
                logger.warning("No LLM available for context compression")
        except Exception as e:
            logger.error(f"Error during context compression: {e}", exc_info=True)

    if self.max_tokens is not None or self.max_context_items is not None:
        before = len(ctx.items)
        ctx.truncate(max_items=self.max_context_items, max_tokens=self.max_tokens)
        after = len(ctx.items)
        if after < before:
            logger.info(f"Truncated context from {before} to {after} items")

Run the full context management cycle: compress then truncate.