Processors API¶
History processors for managing conversation context.
create_summarization_processor¶
pydantic_deep.processors.create_summarization_processor(model='openai:gpt-4.1', trigger=('tokens', _DEFAULT_TRIGGER_TOKENS), keep=('messages', _DEFAULT_MESSAGES_TO_KEEP), max_input_tokens=None, token_counter=None, summary_prompt=None)
¶
Create a summarization history processor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model to use for generating summaries. |
'openai:gpt-4.1'
|
trigger
|
ContextSize | list[ContextSize] | None
|
When to trigger summarization. Can be: - ("messages", N) - trigger when N+ messages - ("tokens", N) - trigger when N+ tokens - ("fraction", F) - trigger at F fraction of max_input_tokens - List of tuples to trigger on any condition |
('tokens', _DEFAULT_TRIGGER_TOKENS)
|
keep
|
ContextSize
|
How much context to keep after summarization. |
('messages', _DEFAULT_MESSAGES_TO_KEEP)
|
max_input_tokens
|
int | None
|
Maximum input tokens (required for fraction-based triggers). |
None
|
token_counter
|
TokenCounter | None
|
Custom token counting function. |
None
|
summary_prompt
|
str | None
|
Custom prompt for summarization. |
None
|
Returns:
| Type | Description |
|---|---|
SummarizationProcessor
|
Configured SummarizationProcessor. |
Signature¶
def create_summarization_processor(
model: str = "openai:gpt-4.1",
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 |
"openai:gpt-4.1" |
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¶
from pydantic_deep import create_deep_agent
from pydantic_deep.processors import create_summarization_processor
processor = create_summarization_processor(
trigger=("tokens", 100000),
keep=("messages", 20),
)
agent = create_deep_agent(history_processors=[processor])
SummarizationProcessor¶
pydantic_deep.processors.SummarizationProcessor
dataclass
¶
History processor that summarizes conversation when limits are reached.
This processor monitors message token counts and automatically summarizes older messages when a threshold is reached, preserving recent messages and maintaining context continuity.
Example
model
instance-attribute
¶
Model to use for generating summaries.
trigger = None
class-attribute
instance-attribute
¶
Threshold(s) that trigger summarization.
Examples:
- ("messages", 50) - trigger when 50+ messages
- ("tokens", 100000) - trigger when 100k+ tokens
- ("fraction", 0.8) - trigger at 80% of max tokens (requires max_input_tokens)
keep = ('messages', _DEFAULT_MESSAGES_TO_KEEP)
class-attribute
instance-attribute
¶
How much context to keep after summarization.
Examples:
- ("messages", 20) - keep last 20 messages
- ("tokens", 10000) - keep last 10k tokens worth
token_counter = field(default=_count_tokens_approximately)
class-attribute
instance-attribute
¶
Function to count tokens in messages.
summary_prompt = DEFAULT_SUMMARY_PROMPT
class-attribute
instance-attribute
¶
Prompt template for generating summaries.
max_input_tokens = None
class-attribute
instance-attribute
¶
Maximum input tokens for the model (required for fraction-based triggers).
trim_tokens_to_summarize = _DEFAULT_TRIM_TOKEN_LIMIT
class-attribute
instance-attribute
¶
Maximum tokens to include when generating summary. None to skip trimming.
__post_init__()
¶
Validate configuration and set up trigger conditions.
__call__(messages)
async
¶
Process messages and summarize if needed.
This is the main entry point called by pydantic-ai's history processor mechanism.
Definition¶
@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__¶
Process messages and summarize if needed. This is called automatically by pydantic-ai's history processor mechanism.
Example¶
from pydantic_deep.processors import SummarizationProcessor
processor = SummarizationProcessor(
model="openai:gpt-4.1",
trigger=[
("messages", 50),
("tokens", 100000),
],
keep=("messages", 10),
trim_tokens_to_summarize=4000,
)
Type Aliases¶
ContextSize¶
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 ofmax_input_tokens(0 < F <= 1)
TokenCounter¶
Function type for custom token counting.
Constants¶
| Constant | Value | Description |
|---|---|---|
DEFAULT_SUMMARY_PROMPT |
(see source) | Default prompt template for summarization |