Processors API¶

History processors for managing conversation context. These are re-exported from summarization-pydantic-ai.

create_summarization_processor¶

Factory function for creating a summarization processor with sensible defaults.

Signature¶

Python

def create_summarization_processor(
    model: str = "anthropic:claude-sonnet-4-6",
    trigger: ContextSize | list[ContextSize] | None = ("tokens", 170000),
    keep: ContextSize = ("messages", 20),
    max_input_tokens: int | None = None,
    token_counter: TokenCounter | None = None,
    summary_prompt: str | None = None,
) -> SummarizationProcessor

Parameters¶

Parameter	Type	Default	Description
`model`	`str`	`"anthropic:claude-sonnet-4-6"`	Model for generating summaries
`trigger`	`ContextSize \\| list[ContextSize] \\| None`	`("tokens", 170000)`	When to trigger summarization
`keep`	`ContextSize`	`("messages", 20)`	How much context to keep
`max_input_tokens`	`int \\| None`	`None`	Max tokens (required for fraction triggers)
`token_counter`	`TokenCounter \\| None`	`None`	Custom token counting function
`summary_prompt`	`str \\| None`	`None`	Custom summarization prompt

Returns¶

SummarizationProcessor - Configured processor instance.

Example¶

Python

from pydantic_deep import create_deep_agent, create_summarization_processor

processor = create_summarization_processor(
    trigger=("tokens", 100000),
    keep=("messages", 20),
)

agent = create_deep_agent(history_processors=[processor])

SummarizationProcessor¶

Dataclass for LLM-based conversation summarization.

Definition¶

Python

@dataclass
class SummarizationProcessor:
    model: str
    trigger: ContextSize | list[ContextSize] | None = None
    keep: ContextSize = ("messages", 20)
    token_counter: TokenCounter = count_tokens_approximately
    summary_prompt: str = DEFAULT_SUMMARY_PROMPT
    max_input_tokens: int | None = None
    trim_tokens_to_summarize: int | None = 4000

Attributes¶

Attribute	Type	Description
`model`	`str`	Model to use for generating summaries
`trigger`	`ContextSize \\| list[ContextSize] \\| None`	Threshold(s) that trigger summarization
`keep`	`ContextSize`	How much context to keep after summarization
`token_counter`	`TokenCounter`	Function to count tokens in messages
`summary_prompt`	`str`	Prompt template for generating summaries
`max_input_tokens`	`int \\| None`	Maximum input tokens (required for fraction triggers)
`trim_tokens_to_summarize`	`int \\| None`	Maximum tokens to include when generating summary

Methods¶

call¶

Python

async def __call__(self, messages: list[ModelMessage]) -> list[ModelMessage]

Process messages and summarize if needed. This is called automatically by pydantic-ai's history processor mechanism.

Example¶

Python

from pydantic_deep import SummarizationProcessor

processor = SummarizationProcessor(
    model="anthropic:claude-sonnet-4-6",
    trigger=[
        ("messages", 50),
        ("tokens", 100000),
    ],
    keep=("messages", 10),
    trim_tokens_to_summarize=4000,
)

create_sliding_window_processor¶

Factory function for creating a sliding window processor with sensible defaults.

Signature¶

Python

def create_sliding_window_processor(
    trigger: ContextSize | list[ContextSize] | None = ("messages", 100),
    keep: ContextSize = ("messages", 50),
    max_input_tokens: int | None = None,
    token_counter: TokenCounter | None = None,
) -> SlidingWindowProcessor

Parameters¶

Parameter	Type	Default	Description
`trigger`	`ContextSize \\| list[ContextSize] \\| None`	`("messages", 100)`	When to trigger trimming
`keep`	`ContextSize`	`("messages", 50)`	How much context to keep
`max_input_tokens`	`int \\| None`	`None`	Max tokens (required for fraction triggers)
`token_counter`	`TokenCounter \\| None`	`None`	Custom token counting function

Returns¶

SlidingWindowProcessor - Configured processor instance.

Example¶

Python

from pydantic_deep import create_deep_agent, create_sliding_window_processor

processor = create_sliding_window_processor(
    trigger=("messages", 100),
    keep=("messages", 50),
)

agent = create_deep_agent(history_processors=[processor])

SlidingWindowProcessor¶

Dataclass for zero-cost message trimming without LLM calls.

Definition¶

Python

@dataclass
class SlidingWindowProcessor:
    trigger: ContextSize | list[ContextSize] | None = None
    keep: ContextSize = ("messages", 50)
    token_counter: TokenCounter = count_tokens_approximately
    max_input_tokens: int | None = None

Attributes¶

Attribute	Type	Description
`trigger`	`ContextSize \\| list[ContextSize] \\| None`	Threshold(s) that trigger trimming
`keep`	`ContextSize`	How much context to keep after trimming
`token_counter`	`TokenCounter`	Function to count tokens in messages
`max_input_tokens`	`int \\| None`	Maximum input tokens (required for fraction triggers)

Methods¶

call¶

Python

def __call__(self, messages: list[ModelMessage]) -> list[ModelMessage]

Process messages and trim if needed. Note: This is a synchronous method (no LLM calls).

Example¶

Python

from pydantic_deep import SlidingWindowProcessor

processor = SlidingWindowProcessor(
    trigger=("tokens", 100000),
    keep=("messages", 50),
)

Type Aliases¶

ContextSize¶

Python

ContextFraction = tuple[Literal["fraction"], float]
ContextTokens = tuple[Literal["tokens"], int]
ContextMessages = tuple[Literal["messages"], int]

ContextSize = ContextFraction | ContextTokens | ContextMessages

Specifies context size thresholds:

("messages", N) - Number of messages
("tokens", N) - Number of tokens
("fraction", F) - Fraction of max_input_tokens (0 < F <= 1)

TokenCounter¶

Python

TokenCounter = Callable[[Sequence[ModelMessage]], int]

Function type for custom token counting.

EvictionProcessor¶

History processor that evicts large tool outputs to files. See Eviction.

Definition¶

Python

@dataclass
class EvictionProcessor:
    backend: BackendProtocol
    token_limit: int = 20_000
    eviction_path: str = "/large_tool_results"
    head_lines: int = 5
    tail_lines: int = 5

Factory¶

Python

from pydantic_deep import create_eviction_processor

processor = create_eviction_processor(
    backend=StateBackend(),
    token_limit=20000,
)

`pydantic_deep.processors.eviction.EvictionProcessor` `dataclass` ¶

History processor that evicts large tool outputs to files.

Scans ToolReturnPart elements in message history and saves content exceeding the token limit to the backend filesystem, replacing it with a preview + file reference.

Uses RunContext to access the runtime backend from deps, ensuring evicted files are written to the same backend that console tools (read_file, grep, execute) use. Falls back to self.backend for standalone usage without RunContext.

Attributes:

Name	Type	Description
`backend`	`BackendProtocol`	Fallback backend for standalone usage (without RunContext).
`token_limit`	`int`	Maximum tokens before eviction (default: 20K).
`eviction_path`	`str`	Directory path for evicted files.
`head_lines`	`int`	Lines to show from start in preview.
`tail_lines`	`int`	Lines to show from end in preview.

Example

Python

from pydantic_deep import create_deep_agent

# Via create_deep_agent (recommended - uses runtime deps.backend):
agent = create_deep_agent(eviction_token_limit=20000)

# Standalone with explicit backend:
from pydantic_ai import Agent
from pydantic_ai_backends import StateBackend
from pydantic_deep.processors.eviction import EvictionProcessor

backend = StateBackend()
processor = EvictionProcessor(backend=backend)
agent = Agent("anthropic:claude-sonnet-4-6", history_processors=[processor])

`backend` `instance-attribute` ¶

Fallback backend used when RunContext is not available or deps has no backend.

`token_limit = DEFAULT_TOKEN_LIMIT` `class-attribute` `instance-attribute` ¶

Maximum estimated tokens before a tool output is evicted.

`eviction_path = DEFAULT_EVICTION_PATH` `class-attribute` `instance-attribute` ¶

Directory path in the backend where evicted files are stored.

`head_lines = DEFAULT_HEAD_LINES` `class-attribute` `instance-attribute` ¶

Number of lines from the start to include in the preview.

`tail_lines = DEFAULT_TAIL_LINES` `class-attribute` `instance-attribute` ¶

Number of lines from the end to include in the preview.

`on_eviction = None` `class-attribute` `instance-attribute` ¶

Callback when tool output is evicted.

Called with (tool_name, file_path, original_chars, preview_chars).

`max_evicted_ids = DEFAULT_MAX_EVICTED_IDS` `class-attribute` `instance-attribute` ¶

Upper bound on remembered evicted IDs; oldest are dropped FIFO past this.

`call(ctx, messages)` `async` ¶

Process messages and evict large tool outputs.

This is the main entry point called by pydantic-ai's history processor mechanism before each model request. The RunContext parameter is detected by pydantic-ai's is_takes_ctx() and automatically provided.

Parameters:

Name	Type	Description	Default
`ctx`	`RunContext[Any]`	Run context providing access to runtime deps.	required
`messages`	`list[ModelMessage]`	Current message history.	required

Returns:

Type	Description
`list[ModelMessage]`	Message history with large tool outputs replaced by previews.

`pydantic_deep.processors.eviction.create_eviction_processor(backend, *, token_limit=DEFAULT_TOKEN_LIMIT, eviction_path=DEFAULT_EVICTION_PATH, head_lines=DEFAULT_HEAD_LINES, tail_lines=DEFAULT_TAIL_LINES, on_eviction=None)` ¶

Create an eviction processor for large tool outputs.

Factory function following the pattern of create_summarization_processor() from summarization-pydantic-ai.

Parameters:

Name	Type	Description	Default
`backend`	`BackendProtocol`	File storage backend for saving evicted content.	required
`token_limit`	`int`	Maximum tokens before eviction (default: 20K).	`DEFAULT_TOKEN_LIMIT`
`eviction_path`	`str`	Directory path for evicted files.	`DEFAULT_EVICTION_PATH`
`head_lines`	`int`	Lines to show from start in preview.	`DEFAULT_HEAD_LINES`
`tail_lines`	`int`	Lines to show from end in preview.	`DEFAULT_TAIL_LINES`
`on_eviction`	`Callable[[str, str, int, int], Any] \| None`	Callback when tool output is evicted. Called with (tool_name, file_path, original_chars, preview_chars).	`None`

Returns:

Type	Description
`EvictionProcessor`	Configured EvictionProcessor instance.

Example

Python

from pydantic_ai_backends import StateBackend
from pydantic_deep import create_eviction_processor

processor = create_eviction_processor(
    backend=StateBackend(),
    token_limit=20000,
)

`pydantic_deep.processors.eviction.create_content_preview(content, *, head_lines=5, tail_lines=5, max_chars=None)` ¶

Create a preview showing the head and tail of content with a truncation marker.

Parameters:

Name	Type	Description	Default
`content`	`str`	The full content string to preview.	required
`head_lines`	`int`	Number of lines to show from the start.	`5`
`tail_lines`	`int`	Number of lines to show from the end.	`5`
`max_chars`	`int \| None`	Optional hard character cap applied after the line-based trim. `None` (default) leaves the line-based result as-is. Pass a value to also bound single-line / few-line payloads (minified JSON, base64) that the line logic alone returns whole. Callers that trim by lines only (e.g. unified-diff rendering, which is always multi-line) leave this `None` so the line marker is preserved.	`None`

Returns:

Type	Description
`str`	Preview string with head, truncation marker, and tail.

patch_tool_calls_processor¶

History processor that fixes orphaned tool calls in message history.

Python

from pydantic_deep.processors.patch import patch_tool_calls_processor

# Use as history processor
agent = Agent("anthropic:claude-sonnet-4-6", history_processors=[patch_tool_calls_processor])

# Or via create_deep_agent
agent = create_deep_agent(patch_tool_calls=True)

ContextManagerCapability¶

Capability from summarization-pydantic-ai that tracks token usage and auto-compresses the conversation when approaching the token budget. It is enabled by default through create_deep_agent(context_manager=True).

Constructor¶

Python

from pydantic_ai_summarization import ContextManagerCapability

capability = ContextManagerCapability(
    max_tokens=200_000,
    compress_threshold=0.9,
    on_usage_update=lambda pct, cur, mx: print(f"{pct:.0%}"),
)

agent = create_deep_agent(capabilities=[capability], context_manager=False)

Most users do not construct it directly — set context_manager=True (default) and configure it via the context_manager_max_tokens, on_context_update, on_before_compress, and on_after_compress parameters of create_deep_agent.

See History Processors for details.

Next Steps¶

Agent API - Agent factory and configuration
Types API - Type definitions

Processors API¶

create_summarization_processor¶

Signature¶

Parameters¶

Returns¶

Example¶

SummarizationProcessor¶

Definition¶

Attributes¶

Methods¶

__call__¶

Example¶

create_sliding_window_processor¶

Signature¶

Parameters¶

Returns¶

Example¶

SlidingWindowProcessor¶

Definition¶

Attributes¶

Methods¶

__call__¶

Example¶

Type Aliases¶

ContextSize¶

TokenCounter¶

EvictionProcessor¶

Definition¶

Factory¶

pydantic_deep.processors.eviction.EvictionProcessor dataclass ¶

backend instance-attribute ¶

token_limit = DEFAULT_TOKEN_LIMIT class-attribute instance-attribute ¶

eviction_path = DEFAULT_EVICTION_PATH class-attribute instance-attribute ¶

head_lines = DEFAULT_HEAD_LINES class-attribute instance-attribute ¶

tail_lines = DEFAULT_TAIL_LINES class-attribute instance-attribute ¶

on_eviction = None class-attribute instance-attribute ¶

max_evicted_ids = DEFAULT_MAX_EVICTED_IDS class-attribute instance-attribute ¶

__call__(ctx, messages) async ¶

pydantic_deep.processors.eviction.create_eviction_processor(backend, *, token_limit=DEFAULT_TOKEN_LIMIT, eviction_path=DEFAULT_EVICTION_PATH, head_lines=DEFAULT_HEAD_LINES, tail_lines=DEFAULT_TAIL_LINES, on_eviction=None) ¶

pydantic_deep.processors.eviction.create_content_preview(content, *, head_lines=5, tail_lines=5, max_chars=None) ¶

patch_tool_calls_processor¶

ContextManagerCapability¶

Constructor¶

Next Steps¶

call¶

call¶

`pydantic_deep.processors.eviction.EvictionProcessor` `dataclass` ¶

`backend` `instance-attribute` ¶

`token_limit = DEFAULT_TOKEN_LIMIT` `class-attribute` `instance-attribute` ¶

`eviction_path = DEFAULT_EVICTION_PATH` `class-attribute` `instance-attribute` ¶

`head_lines = DEFAULT_HEAD_LINES` `class-attribute` `instance-attribute` ¶

`tail_lines = DEFAULT_TAIL_LINES` `class-attribute` `instance-attribute` ¶

`on_eviction = None` `class-attribute` `instance-attribute` ¶

`max_evicted_ids = DEFAULT_MAX_EVICTED_IDS` `class-attribute` `instance-attribute` ¶

`call(ctx, messages)` `async` ¶

`pydantic_deep.processors.eviction.create_eviction_processor(backend, *, token_limit=DEFAULT_TOKEN_LIMIT, eviction_path=DEFAULT_EVICTION_PATH, head_lines=DEFAULT_HEAD_LINES, tail_lines=DEFAULT_TAIL_LINES, on_eviction=None)` ¶

`pydantic_deep.processors.eviction.create_content_preview(content, *, head_lines=5, tail_lines=5, max_chars=None)` ¶