Processor API

pydantic_ai_summarization.processor

Summarization history processor for managing conversation context.

DEFAULT_SUMMARY_PROMPT = "<role>\nContext Extraction Assistant\n</role>\n\n<primary_objective>\nExtract the most relevant context from the conversation history below.\n</primary_objective>\n\n<objective_information>\nYou're nearing the token limit and must extract key information. This context will overwrite the conversation history, so include only the most important information.\n</objective_information>\n\n<instructions>\nThe conversation history will be replaced with your extracted context. Extract and record the most important context. Focus on information relevant to the overall goal. Avoid repeating completed actions.\n</instructions>\n\nRead the message history carefully. Think about what is most important to preserve. Extract only essential context.\n\nRespond ONLY with the extracted context. No additional information.\n\n<messages>\nMessages to summarize:\n{messages}\n</messages>" module-attribute

Default prompt template used for generating summaries.

DEFAULT_CONTINUATION_PROMPT = 'Summary of previous conversation:\n\n' module-attribute

Default prefix prepended to the summary in the compressed message.

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.

Attributes:

Name	Type	Description
model	ModelType	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-based triggers).
trim_tokens_to_summarize	int | None	Maximum tokens to include when generating summary.

Example

Python

from pydantic_ai import Agent
from pydantic_ai_summarization import SummarizationProcessor

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("tokens", 100000),
    keep=("messages", 10),
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Source code in src/pydantic_ai_summarization/processor.py

@dataclass
class SummarizationProcessor:
    """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.

    Attributes:
        model: Model to use for generating summaries.
        trigger: Threshold(s) that trigger summarization.
        keep: How much context to keep after summarization.
        token_counter: Function to count tokens in messages.
        summary_prompt: Prompt template for generating summaries.
        max_input_tokens: Maximum input tokens (required for fraction-based triggers).
        trim_tokens_to_summarize: Maximum tokens to include when generating summary.

    Example:
        ```python
        from pydantic_ai import Agent
        from pydantic_ai_summarization import SummarizationProcessor

        processor = SummarizationProcessor(
            model="openai:gpt-4.1",
            trigger=("tokens", 100000),
            keep=("messages", 10),
        )

        agent = Agent(
            "openai:gpt-4.1",
            history_processors=[processor],
        )
        ```
    """

    model: ModelType
    """Model to use for generating summaries.

    Accepts a string model name (e.g., `"openai:gpt-4.1"`), a pydantic-ai
    :class:`~pydantic_ai.models.Model` instance, or a
    :data:`~pydantic_ai.models.KnownModelName` literal.
    """

    trigger: ContextSize | list[ContextSize] | None = None
    """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: ContextSize = ("messages", _DEFAULT_MESSAGES_TO_KEEP)
    """How much context to keep after summarization.

    Examples:
        - ("messages", 20) - keep last 20 messages
        - ("tokens", 10000) - keep last 10k tokens worth
        - ("messages", 0) - summarize everything except the in-flight request
    """

    token_counter: TokenCounter = field(default=count_tokens_approximately)
    """Function to count tokens in messages."""

    summary_prompt: str = DEFAULT_SUMMARY_PROMPT
    """Prompt template for generating summaries."""

    max_input_tokens: int | None = None
    """Maximum input tokens for the model (required for fraction-based triggers)."""

    trim_tokens_to_summarize: int | None = _DEFAULT_TRIM_TOKEN_LIMIT
    """Maximum tokens to include when generating summary. None to skip trimming."""

    _trigger_conditions: list[ContextSize] = field(default_factory=list, init=False)
    _summarization_agent: Agent[None, str] | None = field(default=None, init=False)

    def __post_init__(self) -> None:
        """Validate configuration and set up trigger conditions."""
        self._trigger_conditions, self.keep = _validate_trig_keep(
            self.trigger, self.keep, self.max_input_tokens
        )

    def _validate_context_size(self, context: ContextSize, parameter_name: str) -> ContextSize:
        """Validate context configuration tuples."""
        return _validate_ctx(context, parameter_name)

    def _should_summarize(self, messages: list[ModelMessage], total_tokens: int) -> bool:
        """Determine whether summarization should run."""
        return _should_trigger(
            self._trigger_conditions, messages, total_tokens, self.max_input_tokens
        )

    def _determine_cutoff_index(self, messages: list[ModelMessage]) -> int:
        """Choose cutoff index respecting retention configuration."""
        return _determine_cutoff(
            messages,
            self.keep,
            self.token_counter,
            self.max_input_tokens,
            _DEFAULT_MESSAGES_TO_KEEP,
            keep_in_flight_request=True,
        )

    def _find_token_based_cutoff(
        self, messages: list[ModelMessage], target_token_count: int
    ) -> int:
        """Find cutoff index based on target token retention."""
        return _find_token(messages, target_token_count, self.token_counter)

    def _find_safe_cutoff(self, messages: list[ModelMessage], messages_to_keep: int) -> int:
        """Find safe cutoff point that preserves AI/Tool message pairs."""
        return _find_safe(messages, messages_to_keep, keep_in_flight_request=True)

    def _is_safe_cutoff_point(self, messages: list[ModelMessage], cutoff_index: int) -> bool:
        """Check if cutting at index would separate AI/Tool message pairs."""
        return _is_safe(messages, cutoff_index)

    def _get_summarization_agent(self) -> Agent[None, str]:  # pragma: no cover
        """Get or create the summarization agent."""
        if self._summarization_agent is None:
            self._summarization_agent = Agent(
                self.model,
                instructions=(
                    "You are a context summarization assistant. "
                    "Extract the most important information from conversations."
                ),
            )
        return self._summarization_agent

    async def _create_summary(
        self,
        messages_to_summarize: list[ModelMessage],
        focus: str | None = None,
    ) -> str:  # pragma: no cover
        """Generate summary for the given messages."""
        if not messages_to_summarize:
            return "No previous conversation history."

        formatted = format_messages_for_summary(messages_to_summarize)

        # Trim if needed
        if self.trim_tokens_to_summarize and len(formatted) > self.trim_tokens_to_summarize * 4:
            formatted = formatted[-(self.trim_tokens_to_summarize * 4) :]

        prompt = self.summary_prompt.format(messages=formatted)
        if focus:
            prompt = (
                f"{prompt}\n\n<focus>\n"
                f"Prioritize information related to this focus topic: {focus}\n"
                "</focus>"
            )

        agent = self._get_summarization_agent()
        result = await agent.run(prompt)
        return result.output.strip()

    async def plan_compression(
        self,
        messages: list[ModelMessage],
        *,
        force: bool = False,
    ) -> CompressionPlan | None:
        """Decide whether and where to compress, without running the summary LLM.

        The capability can use the returned plan to fire `on_before_compress`
        with the real cutoff index before the LLM call runs. This is a pure
        decision step: no network calls, no message mutation.

        Args:
            messages: Current message history.
            force: When `True`, bypass trigger-condition checks. Use for manual
                compaction requests (the `compact_conversation` tool or
                `request_compact()`), which should always attempt compression.

        Returns:
            A `CompressionPlan` if compression should proceed, or `None` if no
            trigger matched (or the cutoff collapsed to 0).
        """
        total_tokens = await _async_count_tokens(self.token_counter, messages)

        if not force and not self._should_summarize(messages, total_tokens):
            return None

        cutoff_index = await _async_determine_cutoff(
            messages,
            self.keep,
            self.token_counter,
            self.max_input_tokens,
            _DEFAULT_MESSAGES_TO_KEEP,
            keep_in_flight_request=True,
        )

        if cutoff_index <= 0:
            return None

        system_parts = _extract_system_prompts(messages)
        return CompressionPlan(
            cutoff_index=cutoff_index,
            messages_to_summarize=messages[:cutoff_index],
            preserved_messages=messages[cutoff_index:],
            system_parts=system_parts,
        )

    async def execute_plan(
        self,
        plan: CompressionPlan,
        focus: str | None = None,
    ) -> SummarizationResult:
        """Execute a compression plan: run the summary LLM and build the result.

        The rebuilt history opens with a single `ModelRequest` holding the carried-over
        system prompts plus the summary as a `UserPromptPart`, followed by the preserved
        tail. The summary is deliberately user-visible so the result always maps to at
        least one provider message.

        On LLM failure, returns a `SummarizationResult` with `summarized=False`
        and `skip_reason="failed"`, with `messages` reconstructed from the plan
        (equivalent to the original input). The original history is preserved
        rather than partially mutated.
        """
        try:
            summary = await self._create_summary(plan.messages_to_summarize, focus)
        except Exception:
            # Keep the original history intact rather than discarding context or
            # injecting error text (which could leak sensitive details) into the
            # model-visible prompt. Reconstruct by concatenating the plan slices.
            logger.exception("Summarization failed; keeping original message history.")
            return SummarizationResult(
                messages=[*plan.messages_to_summarize, *plan.preserved_messages],
                summarized=False,
                skip_reason="failed",
            )

        # The summary is a user part, not a system one. Providers with a separate
        # system channel (Anthropic, Google) route SystemPromptParts into a top-level
        # parameter instead of the message list, so a request built only from system
        # parts maps to zero provider messages — a history no provider can accept
        # (#40). Keeping the summary out of the system channel also stops
        # `_extract_system_prompts` from carrying every past summary forward.
        summary_part = UserPromptPart(content=f"{DEFAULT_CONTINUATION_PROMPT}{summary}")
        summary_message = ModelRequest(parts=[*plan.system_parts, summary_part])
        return SummarizationResult(
            messages=[summary_message, *plan.preserved_messages],
            summarized=True,
            cutoff_index=plan.cutoff_index,
            summary=summary,
        )

    async def process(
        self,
        messages: list[ModelMessage],
        focus: str | None = None,
        *,
        force: bool = False,
    ) -> SummarizationResult:
        """One-shot plan + execute. Returns a structured result.

        Use this when you want both the resulting messages and diagnostic
        metadata (whether compression happened, cutoff index, summary text)
        in a single call.

        Args:
            messages: Current message history.
            focus: Optional focus topic for the summary.
            force: Bypass trigger checks (manual compaction).

        Returns:
            `SummarizationResult` describing the outcome.
        """
        plan = await self.plan_compression(messages, force=force)
        if plan is None:
            # plan_compression collapses "not triggered" and "cutoff_zero" into
            # None; re-check to give callers a precise skip_reason.
            total_tokens = await _async_count_tokens(self.token_counter, messages)
            if not force and not self._should_summarize(messages, total_tokens):
                skip_reason: SkipReason = "not_triggered"
            else:
                skip_reason = "cutoff_zero"
            return SummarizationResult(
                messages=messages,
                summarized=False,
                skip_reason=skip_reason,
            )
        return await self.execute_plan(plan, focus)

    async def __call__(
        self, messages: list[ModelMessage], focus: str | None = None
    ) -> list[ModelMessage]:
        """History-processor entry point. Returns the resulting messages only.

        Backwards-compatible with pydantic-ai's `history_processors` contract
        (`(messages) -> messages`). Loses the structured metadata — use
        `process()` directly if you need `summarized` / `cutoff_index` / `summary`.
        """
        result = await self.process(messages, focus=focus)
        return result.messages

model instance-attribute

Model to use for generating summaries.

Accepts a string model name (e.g., "openai:gpt-4.1"), a pydantic-ai :class:~pydantic_ai.models.Model instance, or a :data:~pydantic_ai.models.KnownModelName literal.

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
• ("messages", 0) - summarize everything except the in-flight request

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.

Source code in src/pydantic_ai_summarization/processor.py

def __post_init__(self) -> None:
    """Validate configuration and set up trigger conditions."""
    self._trigger_conditions, self.keep = _validate_trig_keep(
        self.trigger, self.keep, self.max_input_tokens
    )

plan_compression(messages, *, force=False) async

Decide whether and where to compress, without running the summary LLM.

The capability can use the returned plan to fire on_before_compress with the real cutoff index before the LLM call runs. This is a pure decision step: no network calls, no message mutation.

Parameters:

Name	Type	Description	Default
messages	list[ModelMessage]	Current message history.	required
force	bool	When True, bypass trigger-condition checks. Use for manual compaction requests (the compact_conversation tool or request_compact()), which should always attempt compression.	False

Returns:

Type	Description
CompressionPlan | None	A CompressionPlan if compression should proceed, or None if no
CompressionPlan | None	trigger matched (or the cutoff collapsed to 0).

Source code in src/pydantic_ai_summarization/processor.py

async def plan_compression(
    self,
    messages: list[ModelMessage],
    *,
    force: bool = False,
) -> CompressionPlan | None:
    """Decide whether and where to compress, without running the summary LLM.

    The capability can use the returned plan to fire `on_before_compress`
    with the real cutoff index before the LLM call runs. This is a pure
    decision step: no network calls, no message mutation.

    Args:
        messages: Current message history.
        force: When `True`, bypass trigger-condition checks. Use for manual
            compaction requests (the `compact_conversation` tool or
            `request_compact()`), which should always attempt compression.

    Returns:
        A `CompressionPlan` if compression should proceed, or `None` if no
        trigger matched (or the cutoff collapsed to 0).
    """
    total_tokens = await _async_count_tokens(self.token_counter, messages)

    if not force and not self._should_summarize(messages, total_tokens):
        return None

    cutoff_index = await _async_determine_cutoff(
        messages,
        self.keep,
        self.token_counter,
        self.max_input_tokens,
        _DEFAULT_MESSAGES_TO_KEEP,
        keep_in_flight_request=True,
    )

    if cutoff_index <= 0:
        return None

    system_parts = _extract_system_prompts(messages)
    return CompressionPlan(
        cutoff_index=cutoff_index,
        messages_to_summarize=messages[:cutoff_index],
        preserved_messages=messages[cutoff_index:],
        system_parts=system_parts,
    )

execute_plan(plan, focus=None) async

Execute a compression plan: run the summary LLM and build the result.

The rebuilt history opens with a single ModelRequest holding the carried-over system prompts plus the summary as a UserPromptPart, followed by the preserved tail. The summary is deliberately user-visible so the result always maps to at least one provider message.

On LLM failure, returns a SummarizationResult with summarized=False and skip_reason="failed", with messages reconstructed from the plan (equivalent to the original input). The original history is preserved rather than partially mutated.

Source code in src/pydantic_ai_summarization/processor.py

async def execute_plan(
    self,
    plan: CompressionPlan,
    focus: str | None = None,
) -> SummarizationResult:
    """Execute a compression plan: run the summary LLM and build the result.

    The rebuilt history opens with a single `ModelRequest` holding the carried-over
    system prompts plus the summary as a `UserPromptPart`, followed by the preserved
    tail. The summary is deliberately user-visible so the result always maps to at
    least one provider message.

    On LLM failure, returns a `SummarizationResult` with `summarized=False`
    and `skip_reason="failed"`, with `messages` reconstructed from the plan
    (equivalent to the original input). The original history is preserved
    rather than partially mutated.
    """
    try:
        summary = await self._create_summary(plan.messages_to_summarize, focus)
    except Exception:
        # Keep the original history intact rather than discarding context or
        # injecting error text (which could leak sensitive details) into the
        # model-visible prompt. Reconstruct by concatenating the plan slices.
        logger.exception("Summarization failed; keeping original message history.")
        return SummarizationResult(
            messages=[*plan.messages_to_summarize, *plan.preserved_messages],
            summarized=False,
            skip_reason="failed",
        )

    # The summary is a user part, not a system one. Providers with a separate
    # system channel (Anthropic, Google) route SystemPromptParts into a top-level
    # parameter instead of the message list, so a request built only from system
    # parts maps to zero provider messages — a history no provider can accept
    # (#40). Keeping the summary out of the system channel also stops
    # `_extract_system_prompts` from carrying every past summary forward.
    summary_part = UserPromptPart(content=f"{DEFAULT_CONTINUATION_PROMPT}{summary}")
    summary_message = ModelRequest(parts=[*plan.system_parts, summary_part])
    return SummarizationResult(
        messages=[summary_message, *plan.preserved_messages],
        summarized=True,
        cutoff_index=plan.cutoff_index,
        summary=summary,
    )

process(messages, focus=None, *, force=False) async

One-shot plan + execute. Returns a structured result.

Use this when you want both the resulting messages and diagnostic metadata (whether compression happened, cutoff index, summary text) in a single call.

Parameters:

Name	Type	Description	Default
messages	list[ModelMessage]	Current message history.	required
focus	str | None	Optional focus topic for the summary.	None
force	bool	Bypass trigger checks (manual compaction).	False

Returns:

Type	Description
SummarizationResult	SummarizationResult describing the outcome.

Source code in src/pydantic_ai_summarization/processor.py

async def process(
    self,
    messages: list[ModelMessage],
    focus: str | None = None,
    *,
    force: bool = False,
) -> SummarizationResult:
    """One-shot plan + execute. Returns a structured result.

    Use this when you want both the resulting messages and diagnostic
    metadata (whether compression happened, cutoff index, summary text)
    in a single call.

    Args:
        messages: Current message history.
        focus: Optional focus topic for the summary.
        force: Bypass trigger checks (manual compaction).

    Returns:
        `SummarizationResult` describing the outcome.
    """
    plan = await self.plan_compression(messages, force=force)
    if plan is None:
        # plan_compression collapses "not triggered" and "cutoff_zero" into
        # None; re-check to give callers a precise skip_reason.
        total_tokens = await _async_count_tokens(self.token_counter, messages)
        if not force and not self._should_summarize(messages, total_tokens):
            skip_reason: SkipReason = "not_triggered"
        else:
            skip_reason = "cutoff_zero"
        return SummarizationResult(
            messages=messages,
            summarized=False,
            skip_reason=skip_reason,
        )
    return await self.execute_plan(plan, focus)

__call__(messages, focus=None) async

History-processor entry point. Returns the resulting messages only.

Backwards-compatible with pydantic-ai's history_processors contract ((messages) -> messages). Loses the structured metadata — use process() directly if you need summarized / cutoff_index / summary.

Source code in src/pydantic_ai_summarization/processor.py

async def __call__(
    self, messages: list[ModelMessage], focus: str | None = None
) -> list[ModelMessage]:
    """History-processor entry point. Returns the resulting messages only.

    Backwards-compatible with pydantic-ai's `history_processors` contract
    (`(messages) -> messages`). Loses the structured metadata — use
    `process()` directly if you need `summarized` / `cutoff_index` / `summary`.
    """
    result = await self.process(messages, focus=focus)
    return result.messages

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.

This is a convenience factory function for creating SummarizationProcessor instances with sensible defaults.

Parameters:

Name	Type	Description	Default
model	ModelType	Model to use for generating summaries. Accepts a string name, a Model instance, or a KnownModelName. Defaults to "openai:gpt-4.1".	'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 Defaults to ("tokens", 170000).	('tokens', _DEFAULT_TRIGGER_TOKENS)
keep	ContextSize	How much context to keep after summarization. Defaults to ("messages", 20).	('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. Defaults to approximate counter.	None
summary_prompt	str | None	Custom prompt for summarization. Defaults to built-in prompt.	None

Returns:

Type	Description
SummarizationProcessor	Configured SummarizationProcessor.

Example

Python

from pydantic_ai import Agent
from pydantic_ai_summarization import create_summarization_processor

processor = create_summarization_processor(
    trigger=("messages", 50),
    keep=("messages", 10),
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Source code in src/pydantic_ai_summarization/processor.py

def create_summarization_processor(
    model: ModelType = "openai:gpt-4.1",
    trigger: ContextSize | list[ContextSize] | None = ("tokens", _DEFAULT_TRIGGER_TOKENS),
    keep: ContextSize = ("messages", _DEFAULT_MESSAGES_TO_KEEP),
    max_input_tokens: int | None = None,
    token_counter: TokenCounter | None = None,
    summary_prompt: str | None = None,
) -> SummarizationProcessor:
    """Create a summarization history processor.

    This is a convenience factory function for creating SummarizationProcessor
    instances with sensible defaults.

    Args:
        model: Model to use for generating summaries. Accepts a string name,
            a Model instance, or a KnownModelName. Defaults to "openai:gpt-4.1".
        trigger: 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
            Defaults to ("tokens", 170000).
        keep: How much context to keep after summarization. Defaults to ("messages", 20).
        max_input_tokens: Maximum input tokens (required for fraction-based triggers).
        token_counter: Custom token counting function. Defaults to approximate counter.
        summary_prompt: Custom prompt for summarization. Defaults to built-in prompt.

    Returns:
        Configured SummarizationProcessor.

    Example:
        ```python
        from pydantic_ai import Agent
        from pydantic_ai_summarization import create_summarization_processor

        processor = create_summarization_processor(
            trigger=("messages", 50),
            keep=("messages", 10),
        )

        agent = Agent(
            "openai:gpt-4.1",
            history_processors=[processor],
        )
        ```
    """
    kwargs: dict[str, Any] = {
        "model": model,
        "trigger": trigger,
        "keep": keep,
    }

    if max_input_tokens is not None:
        kwargs["max_input_tokens"] = max_input_tokens

    if token_counter is not None:
        kwargs["token_counter"] = token_counter

    if summary_prompt is not None:
        kwargs["summary_prompt"] = summary_prompt

    return SummarizationProcessor(**kwargs)

count_tokens_approximately(messages)

Approximate token count based on character length.

This is a simple heuristic: ~4 characters per token on average. For production use, consider using a proper tokenizer like tiktoken.

Parameters:

Name	Type	Description	Default
messages	Sequence[ModelMessage]	Sequence of messages to count tokens for.	required

Returns:

Type	Description
int	Approximate token count.

Example

Python

from pydantic_ai_summarization import count_tokens_approximately

messages = [...]  # Your message history
token_count = count_tokens_approximately(messages)
print(f"Approximately {token_count} tokens")

Source code in src/pydantic_ai_summarization/processor.py

def count_tokens_approximately(messages: Sequence[ModelMessage]) -> int:  # pragma: no branch
    """Approximate token count based on character length.

    This is a simple heuristic: ~4 characters per token on average.
    For production use, consider using a proper tokenizer like tiktoken.

    Args:
        messages: Sequence of messages to count tokens for.

    Returns:
        Approximate token count.

    Example:
        ```python
        from pydantic_ai_summarization import count_tokens_approximately

        messages = [...]  # Your message history
        token_count = count_tokens_approximately(messages)
        print(f"Approximately {token_count} tokens")
        ```
    """
    total_chars = 0
    for msg in messages:
        if isinstance(msg, ModelRequest):
            for part in msg.parts:
                if isinstance(part, UserPromptPart):
                    if isinstance(part.content, str):
                        total_chars += len(part.content)
                    else:
                        # List of content parts
                        for item in part.content:
                            if isinstance(item, dict) and "text" in item:
                                total_chars += len(str(item.get("text", "")))
                elif isinstance(part, SystemPromptPart):
                    total_chars += len(part.content)
                elif isinstance(part, ToolReturnPart):
                    total_chars += len(str(part.content))
        elif isinstance(msg, ModelResponse):
            for response_part in msg.parts:
                if isinstance(response_part, TextPart):
                    total_chars += len(response_part.content)
                elif isinstance(response_part, ToolCallPart):
                    total_chars += len(response_part.tool_name)
                    total_chars += len(str(response_part.args))

    return total_chars // 4

format_messages_for_summary(messages)

Format messages into a readable string for summarization.

This function converts a sequence of ModelMessage objects into a human-readable format suitable for passing to an LLM for summarization.

Parameters:

Name	Type	Description	Default
messages	Sequence[ModelMessage]	Sequence of messages to format.	required

Returns:

Type	Description
str	Formatted string representation of the messages.

Example

Python

from pydantic_ai_summarization import format_messages_for_summary

messages = [...]  # Your message history
formatted = format_messages_for_summary(messages)
print(formatted)
# User: Hello
# Assistant: Hi there!
# Tool Call [search]: {"query": "weather"}
# Tool [search]: Sunny, 72°F

Source code in src/pydantic_ai_summarization/processor.py

def format_messages_for_summary(messages: Sequence[ModelMessage]) -> str:  # pragma: no branch
    """Format messages into a readable string for summarization.

    This function converts a sequence of ModelMessage objects into a
    human-readable format suitable for passing to an LLM for summarization.

    Args:
        messages: Sequence of messages to format.

    Returns:
        Formatted string representation of the messages.

    Example:
        ```python
        from pydantic_ai_summarization import format_messages_for_summary

        messages = [...]  # Your message history
        formatted = format_messages_for_summary(messages)
        print(formatted)
        # User: Hello
        # Assistant: Hi there!
        # Tool Call [search]: {"query": "weather"}
        # Tool [search]: Sunny, 72°F
        ```
    """
    lines: list[str] = []

    for msg in messages:
        if isinstance(msg, ModelRequest):
            lines.extend(_format_request_parts(msg))
        elif isinstance(msg, ModelResponse):
            lines.extend(_format_response_parts(msg))

    return "\n".join(lines)

Async Token Counting

pydantic_ai_summarization._cutoff.async_count_tokens(token_counter, messages) async

Call a token counter, awaiting if it returns an awaitable.

Parameters:

Name	Type	Description	Default
token_counter	TokenCounter	Sync or async token counting function.	required
messages	Sequence[ModelMessage]	Messages to count tokens for.	required

Returns:

Type	Description
int	Token count.

Source code in src/pydantic_ai_summarization/_cutoff.py

async def async_count_tokens(token_counter: TokenCounter, messages: Sequence[ModelMessage]) -> int:
    """Call a token counter, awaiting if it returns an awaitable.

    Args:
        token_counter: Sync or async token counting function.
        messages: Messages to count tokens for.

    Returns:
        Token count.
    """
    result = token_counter(messages)
    if inspect.isawaitable(result):
        return await result
    return result