Skip to content

API reference

Generated automatically from docstrings via mkdocstrings.

Top-level

convopack: framework-agnostic context-window packer for LLM chat history.

PackEventKind module-attribute 

PackEventKind = Literal['kept', 'dropped', 'summarized', 'done']

Role module-attribute 

Role = Literal['system', 'user', 'assistant', 'tool']

ContentBlock module-attribute 

ContentBlock = TextBlock | ImageBlock | ToolUseBlock | ToolResultBlock

Packer

Coordinates a tokenizer, a strategy, and pinning rules.

Example

from convopack import Packer, Recency packer = Packer(budget=4000, tokenizer="approx", strategy=Recency()) packed = packer.pack(messages)

Source code in src/convopack/packer.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class Packer:
    """Coordinates a tokenizer, a strategy, and pinning rules.

    Example
    -------
    >>> from convopack import Packer, Recency
    >>> packer = Packer(budget=4000, tokenizer="approx", strategy=Recency())
    >>> packed = packer.pack(messages)
    """

    def __init__(
        self,
        *,
        budget: int,
        tokenizer: str | Tokenizer = "approx",
        strategy: Strategy | None = None,
        pin: Sequence[PinSpec] = ("system",),
        cache: bool = False,
    ) -> None:
        if budget <= 0:
            raise ValueError("budget must be positive")
        self.budget = budget
        self.tokenizer = get_tokenizer(tokenizer)
        self.strategy = strategy or Recency()
        self.pin = tuple(pin)
        self.cache = cache

    def pack(self, messages: Iterable[Message]) -> PackResult:
        """Pack a sequence of internal :class:`Message` objects."""
        msgs = list(messages)
        pinned = self._resolve_pinned(msgs)
        result = self.strategy.pack(
            msgs, budget=self.budget, tokenizer=self.tokenizer, pinned_indices=pinned
        )
        if self.cache:
            result.cache_markers = stable_marker_indices(result.kept, pin_specs=self.pin)
        return result

    async def pack_async(self, messages: Iterable[Message]) -> PackResult:
        """Async variant. Runs sync strategies in a thread; awaits async summarisers."""
        msgs = list(messages)
        pinned = self._resolve_pinned(msgs)

        strategy = self.strategy
        if hasattr(strategy, "pack_async"):
            return await strategy.pack_async(  # type: ignore[no-any-return]
                msgs,
                budget=self.budget,
                tokenizer=self.tokenizer,
                pinned_indices=pinned,
            )
        result = await asyncio.to_thread(
            strategy.pack,
            msgs,
            budget=self.budget,
            tokenizer=self.tokenizer,
            pinned_indices=pinned,
        )
        if self.cache:
            result.cache_markers = stable_marker_indices(result.kept, pin_specs=self.pin)
        return result

    def pack_stream(self, messages: Iterable[Message]) -> Iterator[PackEvent]:
        """Yield :class:`PackEvent` items reflecting the strategy's decisions.

        Useful for progress bars, audit logs, or piping pack telemetry to a
        debugger. The terminal ``done`` event carries the total token count.
        """
        msgs = list(messages)
        result = self.pack(msgs)
        positions = {id(m): i for i, m in enumerate(msgs)}
        summary_id = id(result.summary) if result.summary is not None else None
        for m in result.kept:
            cost = self.tokenizer.count_message(m)
            if summary_id is not None and id(m) == summary_id:
                yield PackEvent(kind="summarized", index=-1, message=m, token_cost=cost)
            else:
                yield PackEvent(
                    kind="kept",
                    index=positions.get(id(m), -1),
                    message=m,
                    token_cost=cost,
                )
        for m in result.dropped:
            cost = self.tokenizer.count_message(m)
            yield PackEvent(
                kind="dropped",
                index=positions.get(id(m), -1),
                message=m,
                token_cost=cost,
            )
        yield PackEvent(kind="done", index=-1, message=None, token_cost=result.token_count)

    def pack_openai(self, raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
        """Convenience: accept OpenAI Chat dicts, return packed OpenAI Chat dicts."""
        return to_openai(self.pack(from_openai(raw)).kept)

    def pack_anthropic(
        self, raw: Iterable[dict[str, Any]], *, system: str | None = None
    ) -> AnthropicPayload:
        """Convenience: accept Anthropic Messages dicts, return a packed payload.

        Cache markers (from ``Packer(cache=True)``) are forwarded so the
        resulting payload's content blocks carry the appropriate
        ``cache_control`` markers for Anthropic prompt caching.
        """
        result = self.pack(from_anthropic(raw, system=system))
        return to_anthropic(result.kept, cache_markers=result.cache_markers)

    def pack_gemini(
        self,
        raw: Iterable[dict[str, Any]],
        *,
        system_instruction: str | None = None,
    ) -> GeminiPayload:
        """Convenience: accept Gemini ``Content`` dicts, return a packed payload."""
        result = self.pack(from_gemini(raw, system_instruction=system_instruction))
        return to_gemini(result.kept)

    def cache_prefix_signature(self, messages: Iterable[Message]) -> str:
        """Return a sha256 of the *stable* prefix that should drive cache hits.

        For OpenAI's automatic prefix caching to work, the leading slice of
        every call must be byte-identical to prior calls. This signature
        covers exactly the kept messages flagged as cache-stable -- the same
        ones that would receive an Anthropic ``cache_control`` marker. If the
        signature differs between two calls, your cache prefix has drifted.
        """
        from convopack._types import history_hash

        msgs = list(messages)
        result = self.pack(msgs)
        markers = stable_marker_indices(result.kept, pin_specs=self.pin)
        if not markers:
            return history_hash([])
        prefix_end = max(markers) + 1
        return history_hash(result.kept[:prefix_end])

    def cache_info(self, messages: Iterable[Message]) -> dict[str, Any]:
        """Report what would be cached for a given history.

        Returns a dict with:

        * ``markers`` -- indices into ``kept`` that get a cache marker.
        * ``marked_messages`` -- count of those messages.
        * ``marked_tokens`` -- tokens covered by the marked prefix.
        * ``total_tokens`` -- tokens in the entire packed output.
        * ``hit_ratio`` -- ``marked_tokens / total_tokens`` (estimated hit
          ratio on the *next* call if the prefix doesn't drift).
        * ``prefix_signature`` -- ``cache_prefix_signature`` for convenience.
        """
        msgs = list(messages)
        result = self.pack(msgs)
        markers = stable_marker_indices(result.kept, pin_specs=self.pin)
        marked_tokens = sum(
            self.tokenizer.count_message(result.kept[i])
            for i in markers
            if 0 <= i < len(result.kept)
        )
        total = result.token_count or 1
        return {
            "markers": markers,
            "marked_messages": len(markers),
            "marked_tokens": marked_tokens,
            "total_tokens": result.token_count,
            "hit_ratio": marked_tokens / total,
            "prefix_signature": self.cache_prefix_signature(msgs),
        }

    def pack_litellm(self, raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
        """Convenience: accept LiteLLM messages (OpenAI Chat shape), return packed dicts."""
        from convopack.providers.litellm import from_litellm, to_litellm

        return to_litellm(self.pack(from_litellm(raw)).kept)

    def pack_dspy(self, raw: Any, *, as_history: bool = False) -> Any:
        """Convenience: accept dspy.History or message list, return packed equivalent."""
        from convopack.providers.dspy import from_dspy, to_dspy

        return to_dspy(self.pack(from_dspy(raw)).kept, as_history=as_history)

    def pack_anthropic_managed(
        self,
        raw: Iterable[dict[str, Any]],
        *,
        system: str | None = None,
        trigger_tokens: int = 30_000,
        keep_tool_uses: int = 3,
    ) -> tuple[AnthropicPayload, dict[str, Any]]:
        """Pack AND build an Anthropic ``context_management`` config in one call.

        The returned tuple is ``(payload, context_management_dict)``. Pass
        the dict as the ``context_management`` argument of
        ``client.messages.create``. ``trigger_tokens`` controls when the
        server compresses; ``keep_tool_uses`` is how many recent tool
        exchanges survive its compression.
        """
        from convopack.anthropic_managed import ContextManagementConfig

        payload = self.pack_anthropic(raw, system=system)
        config = ContextManagementConfig.clear_tool_uses(
            trigger_tokens=trigger_tokens, keep_n=keep_tool_uses
        )
        return payload, config.to_dict()

    def pack_langchain(self, raw: Iterable[Any]) -> list[Any]:
        """Convenience: accept LangChain ``BaseMessage`` objects, return packed objects.

        Drop-in replacement for ``langchain_core.messages.trim_messages`` with
        the added guarantee that ``tool_use`` / ``tool_result`` pairs stay
        atomic.
        """
        from convopack.providers.langchain import from_langchain, to_langchain

        result = self.pack(from_langchain(raw))
        return to_langchain(result.kept)

    def _resolve_pinned(self, msgs: list[Message]) -> set[int]:
        if not msgs:
            return set()
        pinned: set[int] = set()
        first_user: int | None = None
        last_user: int | None = None
        for idx, msg in enumerate(msgs):
            if msg.role == "user":
                if first_user is None:
                    first_user = idx
                last_user = idx
        for spec in self.pin:
            if isinstance(spec, int):
                if 0 <= spec < len(msgs) or -len(msgs) <= spec < 0:
                    pinned.add(spec % len(msgs))
                continue
            if spec == "system":
                for idx, msg in enumerate(msgs):
                    if msg.role == "system":
                        pinned.add(idx)
            elif spec == "first_user" and first_user is not None:
                pinned.add(first_user)
            elif spec == "last_user" and last_user is not None:
                pinned.add(last_user)
            elif spec == "tool_results":
                for idx, msg in enumerate(msgs):
                    if msg.has_tool_use() or msg.has_tool_result():
                        pinned.add(idx)
        return pinned

pack 

pack(messages: Iterable[Message]) -> PackResult

Pack a sequence of internal :class:Message objects.

Source code in src/convopack/packer.py 
53
54
55
56
57
58
59
60
61
62
def pack(self, messages: Iterable[Message]) -> PackResult:
    """Pack a sequence of internal :class:`Message` objects."""
    msgs = list(messages)
    pinned = self._resolve_pinned(msgs)
    result = self.strategy.pack(
        msgs, budget=self.budget, tokenizer=self.tokenizer, pinned_indices=pinned
    )
    if self.cache:
        result.cache_markers = stable_marker_indices(result.kept, pin_specs=self.pin)
    return result

pack_async async 

pack_async(messages: Iterable[Message]) -> PackResult

Async variant. Runs sync strategies in a thread; awaits async summarisers.

Source code in src/convopack/packer.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def pack_async(self, messages: Iterable[Message]) -> PackResult:
    """Async variant. Runs sync strategies in a thread; awaits async summarisers."""
    msgs = list(messages)
    pinned = self._resolve_pinned(msgs)

    strategy = self.strategy
    if hasattr(strategy, "pack_async"):
        return await strategy.pack_async(  # type: ignore[no-any-return]
            msgs,
            budget=self.budget,
            tokenizer=self.tokenizer,
            pinned_indices=pinned,
        )
    result = await asyncio.to_thread(
        strategy.pack,
        msgs,
        budget=self.budget,
        tokenizer=self.tokenizer,
        pinned_indices=pinned,
    )
    if self.cache:
        result.cache_markers = stable_marker_indices(result.kept, pin_specs=self.pin)
    return result

pack_stream 

pack_stream(messages: Iterable[Message]) -> Iterator[PackEvent]

Yield :class:PackEvent items reflecting the strategy's decisions.

Useful for progress bars, audit logs, or piping pack telemetry to a debugger. The terminal done event carries the total token count.

Source code in src/convopack/packer.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def pack_stream(self, messages: Iterable[Message]) -> Iterator[PackEvent]:
    """Yield :class:`PackEvent` items reflecting the strategy's decisions.

    Useful for progress bars, audit logs, or piping pack telemetry to a
    debugger. The terminal ``done`` event carries the total token count.
    """
    msgs = list(messages)
    result = self.pack(msgs)
    positions = {id(m): i for i, m in enumerate(msgs)}
    summary_id = id(result.summary) if result.summary is not None else None
    for m in result.kept:
        cost = self.tokenizer.count_message(m)
        if summary_id is not None and id(m) == summary_id:
            yield PackEvent(kind="summarized", index=-1, message=m, token_cost=cost)
        else:
            yield PackEvent(
                kind="kept",
                index=positions.get(id(m), -1),
                message=m,
                token_cost=cost,
            )
    for m in result.dropped:
        cost = self.tokenizer.count_message(m)
        yield PackEvent(
            kind="dropped",
            index=positions.get(id(m), -1),
            message=m,
            token_cost=cost,
        )
    yield PackEvent(kind="done", index=-1, message=None, token_cost=result.token_count)

pack_openai 

pack_openai(raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]

Convenience: accept OpenAI Chat dicts, return packed OpenAI Chat dicts.

Source code in src/convopack/packer.py 
119
120
121
def pack_openai(self, raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    """Convenience: accept OpenAI Chat dicts, return packed OpenAI Chat dicts."""
    return to_openai(self.pack(from_openai(raw)).kept)

pack_anthropic 

pack_anthropic(raw: Iterable[dict[str, Any]], *, system: str | None = None) -> AnthropicPayload

Convenience: accept Anthropic Messages dicts, return a packed payload.

Cache markers (from Packer(cache=True)) are forwarded so the resulting payload's content blocks carry the appropriate cache_control markers for Anthropic prompt caching.

Source code in src/convopack/packer.py 
123
124
125
126
127
128
129
130
131
132
133
def pack_anthropic(
    self, raw: Iterable[dict[str, Any]], *, system: str | None = None
) -> AnthropicPayload:
    """Convenience: accept Anthropic Messages dicts, return a packed payload.

    Cache markers (from ``Packer(cache=True)``) are forwarded so the
    resulting payload's content blocks carry the appropriate
    ``cache_control`` markers for Anthropic prompt caching.
    """
    result = self.pack(from_anthropic(raw, system=system))
    return to_anthropic(result.kept, cache_markers=result.cache_markers)

pack_gemini 

pack_gemini(raw: Iterable[dict[str, Any]], *, system_instruction: str | None = None) -> GeminiPayload

Convenience: accept Gemini Content dicts, return a packed payload.

Source code in src/convopack/packer.py 
135
136
137
138
139
140
141
142
143
def pack_gemini(
    self,
    raw: Iterable[dict[str, Any]],
    *,
    system_instruction: str | None = None,
) -> GeminiPayload:
    """Convenience: accept Gemini ``Content`` dicts, return a packed payload."""
    result = self.pack(from_gemini(raw, system_instruction=system_instruction))
    return to_gemini(result.kept)

cache_prefix_signature 

cache_prefix_signature(messages: Iterable[Message]) -> str

Return a sha256 of the stable prefix that should drive cache hits.

For OpenAI's automatic prefix caching to work, the leading slice of every call must be byte-identical to prior calls. This signature covers exactly the kept messages flagged as cache-stable -- the same ones that would receive an Anthropic cache_control marker. If the signature differs between two calls, your cache prefix has drifted.

Source code in src/convopack/packer.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def cache_prefix_signature(self, messages: Iterable[Message]) -> str:
    """Return a sha256 of the *stable* prefix that should drive cache hits.

    For OpenAI's automatic prefix caching to work, the leading slice of
    every call must be byte-identical to prior calls. This signature
    covers exactly the kept messages flagged as cache-stable -- the same
    ones that would receive an Anthropic ``cache_control`` marker. If the
    signature differs between two calls, your cache prefix has drifted.
    """
    from convopack._types import history_hash

    msgs = list(messages)
    result = self.pack(msgs)
    markers = stable_marker_indices(result.kept, pin_specs=self.pin)
    if not markers:
        return history_hash([])
    prefix_end = max(markers) + 1
    return history_hash(result.kept[:prefix_end])

cache_info 

cache_info(messages: Iterable[Message]) -> dict[str, Any]

Report what would be cached for a given history.

Returns a dict with:

  • markers -- indices into kept that get a cache marker.
  • marked_messages -- count of those messages.
  • marked_tokens -- tokens covered by the marked prefix.
  • total_tokens -- tokens in the entire packed output.
  • hit_ratio -- marked_tokens / total_tokens (estimated hit ratio on the next call if the prefix doesn't drift).
  • prefix_signature -- cache_prefix_signature for convenience.
Source code in src/convopack/packer.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def cache_info(self, messages: Iterable[Message]) -> dict[str, Any]:
    """Report what would be cached for a given history.

    Returns a dict with:

    * ``markers`` -- indices into ``kept`` that get a cache marker.
    * ``marked_messages`` -- count of those messages.
    * ``marked_tokens`` -- tokens covered by the marked prefix.
    * ``total_tokens`` -- tokens in the entire packed output.
    * ``hit_ratio`` -- ``marked_tokens / total_tokens`` (estimated hit
      ratio on the *next* call if the prefix doesn't drift).
    * ``prefix_signature`` -- ``cache_prefix_signature`` for convenience.
    """
    msgs = list(messages)
    result = self.pack(msgs)
    markers = stable_marker_indices(result.kept, pin_specs=self.pin)
    marked_tokens = sum(
        self.tokenizer.count_message(result.kept[i])
        for i in markers
        if 0 <= i < len(result.kept)
    )
    total = result.token_count or 1
    return {
        "markers": markers,
        "marked_messages": len(markers),
        "marked_tokens": marked_tokens,
        "total_tokens": result.token_count,
        "hit_ratio": marked_tokens / total,
        "prefix_signature": self.cache_prefix_signature(msgs),
    }

pack_litellm 

pack_litellm(raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]

Convenience: accept LiteLLM messages (OpenAI Chat shape), return packed dicts.

Source code in src/convopack/packer.py 
195
196
197
198
199
def pack_litellm(self, raw: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    """Convenience: accept LiteLLM messages (OpenAI Chat shape), return packed dicts."""
    from convopack.providers.litellm import from_litellm, to_litellm

    return to_litellm(self.pack(from_litellm(raw)).kept)

pack_dspy 

pack_dspy(raw: Any, *, as_history: bool = False) -> Any

Convenience: accept dspy.History or message list, return packed equivalent.

Source code in src/convopack/packer.py 
201
202
203
204
205
def pack_dspy(self, raw: Any, *, as_history: bool = False) -> Any:
    """Convenience: accept dspy.History or message list, return packed equivalent."""
    from convopack.providers.dspy import from_dspy, to_dspy

    return to_dspy(self.pack(from_dspy(raw)).kept, as_history=as_history)

pack_anthropic_managed 

pack_anthropic_managed(raw: Iterable[dict[str, Any]], *, system: str | None = None, trigger_tokens: int = 30000, keep_tool_uses: int = 3) -> tuple[AnthropicPayload, dict[str, Any]]

Pack AND build an Anthropic context_management config in one call.

The returned tuple is (payload, context_management_dict). Pass the dict as the context_management argument of client.messages.create. trigger_tokens controls when the server compresses; keep_tool_uses is how many recent tool exchanges survive its compression.

Source code in src/convopack/packer.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def pack_anthropic_managed(
    self,
    raw: Iterable[dict[str, Any]],
    *,
    system: str | None = None,
    trigger_tokens: int = 30_000,
    keep_tool_uses: int = 3,
) -> tuple[AnthropicPayload, dict[str, Any]]:
    """Pack AND build an Anthropic ``context_management`` config in one call.

    The returned tuple is ``(payload, context_management_dict)``. Pass
    the dict as the ``context_management`` argument of
    ``client.messages.create``. ``trigger_tokens`` controls when the
    server compresses; ``keep_tool_uses`` is how many recent tool
    exchanges survive its compression.
    """
    from convopack.anthropic_managed import ContextManagementConfig

    payload = self.pack_anthropic(raw, system=system)
    config = ContextManagementConfig.clear_tool_uses(
        trigger_tokens=trigger_tokens, keep_n=keep_tool_uses
    )
    return payload, config.to_dict()

pack_langchain 

pack_langchain(raw: Iterable[Any]) -> list[Any]

Convenience: accept LangChain BaseMessage objects, return packed objects.

Drop-in replacement for langchain_core.messages.trim_messages with the added guarantee that tool_use / tool_result pairs stay atomic.

Source code in src/convopack/packer.py 
231
232
233
234
235
236
237
238
239
240
241
def pack_langchain(self, raw: Iterable[Any]) -> list[Any]:
    """Convenience: accept LangChain ``BaseMessage`` objects, return packed objects.

    Drop-in replacement for ``langchain_core.messages.trim_messages`` with
    the added guarantee that ``tool_use`` / ``tool_result`` pairs stay
    atomic.
    """
    from convopack.providers.langchain import from_langchain, to_langchain

    result = self.pack(from_langchain(raw))
    return to_langchain(result.kept)

PackResult dataclass

Result of a single pack operation.

Source code in src/convopack/_types.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@dataclass(slots=True)
class PackResult:
    """Result of a single pack operation."""

    kept: list[Message]
    dropped: list[Message]
    summary: Message | None
    token_count: int
    budget: int
    cache_markers: list[int] = field(default_factory=list)

    @property
    def fits(self) -> bool:
        return self.token_count <= self.budget

    @property
    def headroom(self) -> int:
        return self.budget - self.token_count

PackEvent dataclass

One step in a pack operation.

Attributes:

Name Type Description
kind PackEventKind

kept -- message survived the pack. dropped -- message was evicted. summarized -- a synthetic summary message replaces some dropped block. done -- terminal event carrying the final token count.
index int

Original position in the input list, or -1 for synthetic events.
message Message | None

The message itself, or None for done.
token_cost int

Tokens contributed by this message. For done, the total kept tokens.
Source code in src/convopack/events.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(slots=True)
class PackEvent:
    """One step in a pack operation.

    Attributes
    ----------
    kind
        ``kept`` -- message survived the pack.
        ``dropped`` -- message was evicted.
        ``summarized`` -- a synthetic summary message replaces some dropped block.
        ``done`` -- terminal event carrying the final token count.
    index
        Original position in the input list, or ``-1`` for synthetic events.
    message
        The message itself, or ``None`` for ``done``.
    token_cost
        Tokens contributed by this message. For ``done``, the total kept tokens.
    """

    kind: PackEventKind
    index: int
    message: Message | None
    token_cost: int

Message dataclass

Provider-agnostic message.

content is always a list of blocks internally. Provider adapters convert to and from string / dict forms when crossing the public boundary.

Source code in src/convopack/_types.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@dataclass(slots=True)
class Message:
    """Provider-agnostic message.

    `content` is always a list of blocks internally. Provider adapters convert to
    and from string / dict forms when crossing the public boundary.
    """

    role: Role
    content: list[ContentBlock] = field(default_factory=list)
    name: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def has_tool_use(self) -> bool:
        return any(b.kind == "tool_use" for b in self.content)

    def has_tool_result(self) -> bool:
        return any(b.kind == "tool_result" for b in self.content)

    def tool_use_ids(self) -> list[str]:
        return [b.id for b in self.content if isinstance(b, ToolUseBlock)]

    def tool_result_ids(self) -> list[str]:
        return [b.tool_use_id for b in self.content if isinstance(b, ToolResultBlock)]

    def text(self) -> str:
        return "".join(b.text for b in self.content if isinstance(b, TextBlock))

    @property
    def content_hash(self) -> str:
        """Stable sha256 hex digest of this message's *semantic* content.

        The hash deliberately excludes runtime-generated tool-call IDs so that
        two messages with identical text, tool name, and input arguments hash
        the same regardless of which provider generated the wire IDs. Useful
        as a cache key and to detect prompt-prefix drift across packs.
        """
        return hashlib.sha256(
            json.dumps(_canonical(self), sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest()

content_hash property 

content_hash: str

Stable sha256 hex digest of this message's semantic content.

The hash deliberately excludes runtime-generated tool-call IDs so that two messages with identical text, tool name, and input arguments hash the same regardless of which provider generated the wire IDs. Useful as a cache key and to detect prompt-prefix drift across packs.

TextBlock dataclass

Source code in src/convopack/_types.py 
13
14
15
16
@dataclass(frozen=True, slots=True)
class TextBlock:
    text: str
    kind: Literal["text"] = "text"

ImageBlock dataclass

Reference to an image. source is opaque (URL, base64, file_id) and provider-specific.

Source code in src/convopack/_types.py 
19
20
21
22
23
24
25
@dataclass(frozen=True, slots=True)
class ImageBlock:
    """Reference to an image. `source` is opaque (URL, base64, file_id) and provider-specific."""

    source: str
    media_type: str | None = None
    kind: Literal["image"] = "image"

ToolUseBlock dataclass

Source code in src/convopack/_types.py 
28
29
30
31
32
33
@dataclass(frozen=True, slots=True)
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]
    kind: Literal["tool_use"] = "tool_use"

ToolResultBlock dataclass

Source code in src/convopack/_types.py 
36
37
38
39
40
41
@dataclass(frozen=True, slots=True)
class ToolResultBlock:
    tool_use_id: str
    content: str | list[TextBlock | ImageBlock]
    is_error: bool = False
    kind: Literal["tool_result"] = "tool_result"

Strategies

Packing strategies.

Strategy

Bases: Protocol

A packing strategy decides which messages survive a budget cut.

Implementations must:

  • keep pinned messages,
  • never split a tool_use / tool_result pair,
  • return a :class:PackResult whose kept list has the same relative order as the input.
Source code in src/convopack/strategies/base.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@runtime_checkable
class Strategy(Protocol):
    """A packing strategy decides which messages survive a budget cut.

    Implementations must:

      * keep pinned messages,
      * never split a tool_use / tool_result pair,
      * return a :class:`PackResult` whose ``kept`` list has the same
        relative order as the input.
    """

    name: str

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult: ...

Recency

Keep the most recent chunks that still fit under budget.

Pinned messages are always kept; if pinned messages alone exceed budget, they are still kept (the result will report fits=False).

Source code in src/convopack/strategies/recency.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class Recency:
    """Keep the most recent chunks that still fit under ``budget``.

    Pinned messages are always kept; if pinned messages alone exceed budget,
    they are still kept (the result will report ``fits=False``).
    """

    name = "recency"

    def __init__(self, *, min_keep: int = 1) -> None:
        if min_keep < 0:
            raise ValueError("min_keep must be >= 0")
        self._min_keep = min_keep

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult:
        if not messages:
            return PackResult(kept=[], dropped=[], summary=None, token_count=0, budget=budget)

        chunks = group_pairs(messages)
        pinned_chunk_ids: set[int] = set()
        for ci, chunk in enumerate(chunks):
            if pinned_indices.intersection(chunk.indices):
                chunk.pinned = True
                pinned_chunk_ids.add(ci)

        pinned_msgs: list[Message] = []
        for ci in pinned_chunk_ids:
            pinned_msgs.extend(chunks[ci].messages)
        used = tokenizer.count_messages(pinned_msgs) if pinned_msgs else 0

        kept_ids: set[int] = set(pinned_chunk_ids)
        non_pinned_kept = 0
        for ci in range(len(chunks) - 1, -1, -1):
            if ci in kept_ids:
                continue
            cost = tokenizer.count_messages(chunks[ci].messages)
            if used + cost <= budget or non_pinned_kept < self._min_keep:
                kept_ids.add(ci)
                used += cost
                non_pinned_kept += 1

        kept: list[Message] = []
        dropped: list[Message] = []
        for ci, chunk in enumerate(chunks):
            (kept if ci in kept_ids else dropped).extend(chunk.messages)

        return PackResult(
            kept=kept,
            dropped=dropped,
            summary=None,
            token_count=used,
            budget=budget,
        )

FirstFit

Counterpart to :class:Recency: keep the oldest chunks that fit.

Useful when the early turns carry the load — a long system prompt, a few high-value few-shot examples, or a question whose answer needs the setup far more than the chitchat that followed.

Source code in src/convopack/strategies/firstfit.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class FirstFit:
    """Counterpart to :class:`Recency`: keep the oldest chunks that fit.

    Useful when the early turns carry the load — a long system prompt, a few
    high-value few-shot examples, or a question whose answer needs the setup
    far more than the chitchat that followed.
    """

    name = "first_fit"

    def __init__(self, *, min_keep: int = 1) -> None:
        if min_keep < 0:
            raise ValueError("min_keep must be >= 0")
        self._min_keep = min_keep

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult:
        if not messages:
            return PackResult(kept=[], dropped=[], summary=None, token_count=0, budget=budget)

        chunks = group_pairs(messages)
        pinned_chunk_ids: set[int] = set()
        for ci, chunk in enumerate(chunks):
            if pinned_indices.intersection(chunk.indices):
                chunk.pinned = True
                pinned_chunk_ids.add(ci)

        pinned_msgs: list[Message] = []
        for ci in pinned_chunk_ids:
            pinned_msgs.extend(chunks[ci].messages)
        used = tokenizer.count_messages(pinned_msgs) if pinned_msgs else 0

        kept_ids: set[int] = set(pinned_chunk_ids)
        non_pinned_kept = 0
        for ci in range(len(chunks)):
            if ci in kept_ids:
                continue
            cost = tokenizer.count_messages(chunks[ci].messages)
            if used + cost <= budget or non_pinned_kept < self._min_keep:
                kept_ids.add(ci)
                used += cost
                non_pinned_kept += 1

        kept: list[Message] = []
        dropped: list[Message] = []
        for ci, chunk in enumerate(chunks):
            (kept if ci in kept_ids else dropped).extend(chunk.messages)

        return PackResult(
            kept=kept,
            dropped=dropped,
            summary=None,
            token_count=used,
            budget=budget,
        )

SummaryEvict

Drop oldest chunks like :class:Recency, but replace them with a summary message.

The summariser is any callable that takes a list of messages and returns a string (sync or async). The summary is inserted as a system message at the head of the kept list, with the prefix [summary] to make it identifiable.

Source code in src/convopack/strategies/summary.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class SummaryEvict:
    """Drop oldest chunks like :class:`Recency`, but replace them with a summary message.

    The summariser is any callable that takes a list of messages and returns a
    string (sync or async). The summary is inserted as a system message at the
    head of the kept list, with the prefix ``[summary]`` to make it
    identifiable.
    """

    name = "summary_evict"

    def __init__(
        self,
        summarizer: Summarizer,
        *,
        prefix: str = "[summary] ",
        reserve: int | None = None,
    ) -> None:
        self._summarizer = summarizer
        self._prefix = prefix
        self._reserve = reserve

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult:
        reserve = self._reserve if self._reserve is not None else max(budget // 10, 128)
        inner_budget = max(budget - reserve, 1)
        inner = Recency().pack(
            messages,
            budget=inner_budget,
            tokenizer=tokenizer,
            pinned_indices=pinned_indices,
        )

        if not inner.dropped:
            return PackResult(
                kept=inner.kept,
                dropped=inner.dropped,
                summary=None,
                token_count=inner.token_count,
                budget=budget,
            )

        summary_text = self._run_summarizer(inner.dropped)
        summary_msg = Message(
            role="system",
            content=[TextBlock(text=f"{self._prefix}{summary_text}")],
            metadata={"convopack.summary": True, "dropped_count": len(inner.dropped)},
        )
        summary_cost = tokenizer.count_message(summary_msg)
        kept = [summary_msg, *inner.kept]
        return PackResult(
            kept=kept,
            dropped=inner.dropped,
            summary=summary_msg,
            token_count=inner.token_count + summary_cost,
            budget=budget,
        )

    def _run_summarizer(self, messages: list[Message]) -> str:
        result = self._summarizer(messages)
        if asyncio.iscoroutine(result):
            try:
                asyncio.get_running_loop()
            except RuntimeError:
                return cast(str, asyncio.run(result))
            result.close()
            raise RuntimeError(
                "SummaryEvict.pack() called from inside a running event loop with an "
                "async summariser. Use `await packer.pack_async(...)` instead."
            )
        return cast(str, result)

Importance

Drop the lowest-scoring chunks until under budget.

A chunk's score is the maximum message score it contains, so a chunk that holds a tool_use/tool_result pair inherits the tool score.

Source code in src/convopack/strategies/importance.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class Importance:
    """Drop the lowest-scoring chunks until under budget.

    A chunk's score is the *maximum* message score it contains, so a chunk that
    holds a tool_use/tool_result pair inherits the tool score.
    """

    name = "importance"

    def __init__(self, scorer: Scorer | None = None) -> None:
        self._scorer = scorer or default_scorer

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult:
        if not messages:
            return PackResult(kept=[], dropped=[], summary=None, token_count=0, budget=budget)

        chunks = group_pairs(messages)
        scored: list[tuple[float, int, int]] = []
        for ci, chunk in enumerate(chunks):
            pinned = bool(pinned_indices.intersection(chunk.indices))
            chunk.pinned = pinned
            score = max((self._scorer(m) for m in chunk.messages), default=0.0)
            if pinned:
                score = float("inf")
            cost = tokenizer.count_messages(chunk.messages)
            scored.append((score, ci, cost))

        scored.sort(key=lambda t: (-t[0], -t[1]))

        kept_ids: set[int] = set()
        used = 0
        for score, ci, cost in scored:
            if score == float("inf") or used + cost <= budget:
                kept_ids.add(ci)
                used += cost

        kept: list[Message] = []
        dropped: list[Message] = []
        for ci, chunk in enumerate(chunks):
            (kept if ci in kept_ids else dropped).extend(chunk.messages)

        return PackResult(
            kept=kept,
            dropped=dropped,
            summary=None,
            token_count=used,
            budget=budget,
        )

SemanticDedup

Drop near-duplicate messages (by embedding similarity), then defer to a fallback strategy.

Chunks containing tool calls or pinned messages are never deduped — they're handed straight to the fallback. Among the remaining message-chunks, only the first occurrence of each near-duplicate group survives. Removing duplicates is order-preserving.

Parameters:

Name Type Description Default
embedder Embedder

An object with embed(text) -> list[float].

 required
threshold float

Cosine similarity above which two messages are considered duplicates.

 0.95
fallback Strategy | None

Strategy to enforce the budget after dedup. Defaults to :class:Recency.

 None
Source code in src/convopack/strategies/dedup.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class SemanticDedup:
    """Drop near-duplicate messages (by embedding similarity), then defer to a fallback strategy.

    Chunks containing tool calls or pinned messages are never deduped — they're
    handed straight to the fallback. Among the remaining message-chunks, only
    the *first* occurrence of each near-duplicate group survives. Removing
    duplicates is order-preserving.

    Parameters
    ----------
    embedder
        An object with ``embed(text) -> list[float]``.
    threshold
        Cosine similarity above which two messages are considered duplicates.
    fallback
        Strategy to enforce the budget after dedup. Defaults to :class:`Recency`.
    """

    name = "semantic_dedup"

    def __init__(
        self,
        embedder: Embedder,
        *,
        threshold: float = 0.95,
        fallback: Strategy | None = None,
    ) -> None:
        if not 0.0 < threshold <= 1.0:
            raise ValueError("threshold must be in (0, 1]")
        self._embedder = embedder
        self._threshold = threshold
        self._fallback = fallback or Recency()

    def pack(
        self,
        messages: list[Message],
        *,
        budget: int,
        tokenizer: Tokenizer,
        pinned_indices: set[int],
    ) -> PackResult:
        if not messages:
            return PackResult(kept=[], dropped=[], summary=None, token_count=0, budget=budget)

        chunks = group_pairs(messages)
        survivors: list[Message] = []
        dedup_dropped: list[Message] = []
        seen_embeddings: list[list[float]] = []

        for chunk in chunks:
            chunk_indices = set(chunk.indices)
            untouchable = bool(chunk_indices & pinned_indices) or any(
                m.has_tool_use() or m.has_tool_result() for m in chunk.messages
            )
            if untouchable:
                survivors.extend(chunk.messages)
                for m in chunk.messages:
                    text = m.text()
                    if text:
                        seen_embeddings.append(self._embedder.embed(text))
                continue

            for m in chunk.messages:
                text = m.text()
                if not text:
                    survivors.append(m)
                    continue
                vec = self._embedder.embed(text)
                if any(cosine(vec, prev) >= self._threshold for prev in seen_embeddings):
                    dedup_dropped.append(m)
                else:
                    survivors.append(m)
                    seen_embeddings.append(vec)

        fallback_pinned = self._reindex_pinned(messages, survivors, pinned_indices)
        fallback_result = self._fallback.pack(
            survivors,
            budget=budget,
            tokenizer=tokenizer,
            pinned_indices=fallback_pinned,
        )
        return PackResult(
            kept=fallback_result.kept,
            dropped=dedup_dropped + fallback_result.dropped,
            summary=fallback_result.summary,
            token_count=fallback_result.token_count,
            budget=budget,
        )

    @staticmethod
    def _reindex_pinned(
        original: list[Message],
        survivors: list[Message],
        pinned: set[int],
    ) -> set[int]:
        pinned_ids = {id(original[i]) for i in pinned if 0 <= i < len(original)}
        return {i for i, m in enumerate(survivors) if id(m) in pinned_ids}

Tokenizers

Tokenizer adapters.

Tokenizer

Bases: Protocol

Counts tokens for text and messages.

Implementations should be cheap to call repeatedly. Counting a message must include any per-message overhead the target provider charges (role tags, separators, tool framing).

Source code in src/convopack/tokenizers/base.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@runtime_checkable
class Tokenizer(Protocol):
    """Counts tokens for text and messages.

    Implementations should be cheap to call repeatedly. Counting a message must
    include any per-message overhead the target provider charges (role tags,
    separators, tool framing).
    """

    name: str

    def count(self, text: str) -> int: ...

    def count_message(self, message: Message) -> int: ...

    def count_messages(self, messages: Iterable[Message]) -> int: ...

ApproxTokenizer

Estimates tokens as ceil(char_count / 4).

Use when you don't want to pull in tiktoken or anthropic. Numbers will be off by 10-30% depending on language; fine for a soft budget but not for billing.

Source code in src/convopack/tokenizers/approx.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class ApproxTokenizer:
    """Estimates tokens as ``ceil(char_count / 4)``.

    Use when you don't want to pull in tiktoken or anthropic. Numbers will be off
    by 10-30% depending on language; fine for a soft budget but not for billing.
    """

    name = "approx"

    def count(self, text: str) -> int:
        if not text:
            return 0
        return (len(text) + _CHARS_PER_TOKEN - 1) // _CHARS_PER_TOKEN

    def count_message(self, message: Message) -> int:
        total = _PER_MESSAGE_OVERHEAD + self.count(message.role) + self.count(message.name or "")
        for block in message.content:
            total += self._block_tokens(block)
        return total

    def count_messages(self, messages: Iterable[Message]) -> int:
        return sum(self.count_message(m) for m in messages)

    def _block_tokens(self, block: object) -> int:
        if isinstance(block, TextBlock):
            return self.count(block.text)
        if isinstance(block, ImageBlock):
            return _IMAGE_TOKEN_ESTIMATE
        if isinstance(block, ToolUseBlock):
            payload = f"{block.id}{block.name}{block.input!r}"
            return self.count(payload) + _PER_MESSAGE_OVERHEAD
        if isinstance(block, ToolResultBlock):
            if isinstance(block.content, str):
                return self.count(block.content) + self.count(block.tool_use_id)
            inner = sum(self._block_tokens(b) for b in block.content)
            return inner + self.count(block.tool_use_id) + _PER_MESSAGE_OVERHEAD
        return 0

get_tokenizer 

get_tokenizer(spec: str | Tokenizer) -> Tokenizer

Resolve a tokenizer from a string spec or pass through an instance.

Spec formats: - "tiktoken:<encoding-or-model>" -- e.g. "tiktoken:gpt-4o". - "anthropic:<model>" -- e.g. "anthropic:claude-sonnet-4-6". - "huggingface:<model-id>" -- any HuggingFace tokenizer id. - "approx" -- char-length / 4 estimator (zero deps).

Source code in src/convopack/tokenizers/__init__.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def get_tokenizer(spec: str | Tokenizer) -> Tokenizer:
    """Resolve a tokenizer from a string spec or pass through an instance.

    Spec formats:
      - ``"tiktoken:<encoding-or-model>"`` -- e.g. ``"tiktoken:gpt-4o"``.
      - ``"anthropic:<model>"`` -- e.g. ``"anthropic:claude-sonnet-4-6"``.
      - ``"huggingface:<model-id>"`` -- any HuggingFace tokenizer id.
      - ``"approx"`` -- char-length / 4 estimator (zero deps).
    """
    if not isinstance(spec, str):
        return spec
    if spec == "approx":
        return ApproxTokenizer()
    if spec.startswith("tiktoken:"):
        from convopack.tokenizers.tiktoken_adapter import TiktokenAdapter

        return TiktokenAdapter(spec.split(":", 1)[1])
    if spec.startswith("anthropic:"):
        from convopack.tokenizers.anthropic_adapter import AnthropicAdapter

        return AnthropicAdapter(spec.split(":", 1)[1])
    if spec.startswith("huggingface:"):
        from convopack.tokenizers.huggingface_adapter import HFTokenizerAdapter

        return HFTokenizerAdapter(spec.split(":", 1)[1])
    raise ValueError(f"Unknown tokenizer spec: {spec!r}")

tiktoken-backed tokenizer (OpenAI BPE).

TiktokenAdapter

Wraps a tiktoken encoding. Specify either a model name or an encoding name.

Source code in src/convopack/tokenizers/tiktoken_adapter.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class TiktokenAdapter:
    """Wraps a tiktoken encoding. Specify either a model name or an encoding name."""

    name: str

    def __init__(self, model_or_encoding: str) -> None:
        try:
            import tiktoken
        except ImportError as exc:
            raise ImportError(
                "tiktoken is required for TiktokenAdapter; "
                "install with `pip install convopack[tiktoken]`."
            ) from exc

        self._model = model_or_encoding
        self.name = f"tiktoken:{model_or_encoding}"
        try:
            self._enc = tiktoken.encoding_for_model(model_or_encoding)
        except KeyError:
            self._enc = tiktoken.get_encoding(model_or_encoding)

    def count(self, text: str) -> int:
        if not text:
            return 0
        return len(self._enc.encode(text, disallowed_special=()))

    def count_message(self, message: Message) -> int:
        total = _PER_MESSAGE_OVERHEAD + self.count(message.role)
        if message.name:
            total += self.count(message.name)
        for block in message.content:
            total += self._block_tokens(block)
        return total

    def count_messages(self, messages: Iterable[Message]) -> int:
        return sum(self.count_message(m) for m in messages) + _PER_REPLY_OVERHEAD

    def _block_tokens(self, block: object) -> int:
        if isinstance(block, TextBlock):
            return self.count(block.text)
        if isinstance(block, ImageBlock):
            return _IMAGE_TOKEN_ESTIMATE
        if isinstance(block, ToolUseBlock):
            return self.count(f"{block.name}{block.input!r}") + self.count(block.id)
        if isinstance(block, ToolResultBlock):
            if isinstance(block.content, str):
                return self.count(block.content) + self.count(block.tool_use_id)
            return sum(self._block_tokens(b) for b in block.content) + self.count(block.tool_use_id)
        return 0

Anthropic tokenizer adapter.

Anthropic does not ship a local tokenizer; the official method is to call messages.count_tokens on the server. That is a network call, so we offer two modes:

  • offline=True (default) -- delegate to :class:ApproxTokenizer. Fast and free, off by ~10-15%.
  • offline=False -- if the anthropic SDK is installed and an API key is configured, call client.messages.count_tokens for exact counts. Slow.

HuggingFace transformers-backed tokenizer adapter.

HFTokenizerAdapter

Wraps a HuggingFace AutoTokenizer so any open-weights model can be used.

Pass either a HuggingFace model id (e.g. meta-llama/Llama-3.1-8B) or a local path. Tokeniser files are downloaded on first use unless local_files_only=True.

Source code in src/convopack/tokenizers/huggingface_adapter.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class HFTokenizerAdapter:
    """Wraps a HuggingFace ``AutoTokenizer`` so any open-weights model can be used.

    Pass either a HuggingFace model id (e.g. ``meta-llama/Llama-3.1-8B``) or a
    local path. Tokeniser files are downloaded on first use unless
    ``local_files_only=True``.
    """

    name: str

    def __init__(
        self,
        model: str,
        *,
        revision: str | None = None,
        local_files_only: bool = False,
    ) -> None:
        try:
            from transformers import AutoTokenizer
        except ImportError as exc:
            raise ImportError(
                "transformers is required for HFTokenizerAdapter; "
                "install with `pip install convopack[huggingface]`."
            ) from exc

        self._model = model
        self.name = f"huggingface:{model}"
        self._tok = AutoTokenizer.from_pretrained(
            model, revision=revision, local_files_only=local_files_only
        )

    def count(self, text: str) -> int:
        if not text:
            return 0
        return len(self._tok.encode(text, add_special_tokens=False))

    def count_message(self, message: Message) -> int:
        total = _PER_MESSAGE_OVERHEAD + self.count(message.role)
        if message.name:
            total += self.count(message.name)
        for block in message.content:
            total += self._block_tokens(block)
        return total

    def count_messages(self, messages: Iterable[Message]) -> int:
        return sum(self.count_message(m) for m in messages) + _PER_REPLY_OVERHEAD

    def _block_tokens(self, block: object) -> int:
        if isinstance(block, TextBlock):
            return self.count(block.text)
        if isinstance(block, ImageBlock):
            return _IMAGE_TOKEN_ESTIMATE
        if isinstance(block, ToolUseBlock):
            return self.count(f"{block.name}{block.input!r}") + self.count(block.id)
        if isinstance(block, ToolResultBlock):
            if isinstance(block.content, str):
                return self.count(block.content) + self.count(block.tool_use_id)
            return sum(self._block_tokens(b) for b in block.content) + self.count(block.tool_use_id)
        return 0

Embedders

Embedder protocol for semantic strategies.

Embedder

Bases: Protocol

Turns text into a dense vector.

Implementations may optionally provide :meth:embed_batch for efficiency; a default implementation that loops over embed is acceptable.

Source code in src/convopack/embedders/base.py 
10
11
12
13
14
15
16
17
18
19
20
@runtime_checkable
class Embedder(Protocol):
    """Turns text into a dense vector.

    Implementations may optionally provide :meth:`embed_batch` for efficiency;
    a default implementation that loops over ``embed`` is acceptable.
    """

    def embed(self, text: str) -> list[float]: ...

    def embed_batch(self, texts: Iterable[str]) -> list[list[float]]: ...

cosine 

cosine(a: list[float], b: list[float]) -> float

Cosine similarity. Returns 0 if either vector is zero.

Source code in src/convopack/embedders/base.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def cosine(a: list[float], b: list[float]) -> float:
    """Cosine similarity. Returns 0 if either vector is zero."""
    if len(a) != len(b):
        raise ValueError(f"Vector length mismatch: {len(a)} vs {len(b)}")
    dot = 0.0
    na = 0.0
    nb = 0.0
    for x, y in zip(a, b, strict=True):
        dot += x * y
        na += x * x
        nb += y * y
    denom = math.sqrt(na) * math.sqrt(nb)
    return dot / denom if denom else 0.0

Providers

Provider message-shape adapters.

GeminiPayload dataclass

What GenerativeModel.generate_content wants.

Source code in src/convopack/providers/gemini.py 
37
38
39
40
41
42
@dataclass(slots=True)
class GeminiPayload:
    """What ``GenerativeModel.generate_content`` wants."""

    system_instruction: str
    contents: list[dict[str, Any]] = field(default_factory=list)

from_openai 

from_openai(messages: Iterable[dict[str, Any]]) -> list[Message]

Convert OpenAI Chat Completions messages into :class:Message objects.

Source code in src/convopack/providers/openai.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def from_openai(messages: Iterable[dict[str, Any]]) -> list[Message]:
    """Convert OpenAI Chat Completions messages into :class:`Message` objects."""
    out: list[Message] = []
    for raw in messages:
        role = cast(Role, raw.get("role", "user"))
        if role == "tool":
            out.append(
                Message(
                    role="tool",
                    content=[
                        ToolResultBlock(
                            tool_use_id=raw.get("tool_call_id", ""),
                            content=str(raw.get("content", "")),
                        )
                    ],
                )
            )
            continue

        blocks: list[ContentBlock] = []
        content = raw.get("content")
        if isinstance(content, str):
            if content:
                blocks.append(TextBlock(text=content))
        elif isinstance(content, list):
            for part in content:
                ptype = part.get("type")
                if ptype == "text":
                    blocks.append(TextBlock(text=part.get("text", "")))
                elif ptype in {"image_url", "input_image"}:
                    url = part.get("image_url", {})
                    src = url.get("url") if isinstance(url, dict) else url
                    blocks.append(ImageBlock(source=src or part.get("image_url", "")))

        for call in raw.get("tool_calls", []) or []:
            fn = call.get("function", {})
            args_raw = fn.get("arguments", "{}")
            try:
                args = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
            except json.JSONDecodeError:
                args = {"_raw": args_raw}
            blocks.append(ToolUseBlock(id=call.get("id", ""), name=fn.get("name", ""), input=args))

        out.append(Message(role=role, content=blocks, name=raw.get("name")))
    return out

to_openai 

to_openai(messages: Iterable[Message]) -> list[dict[str, Any]]

Convert :class:Message objects into OpenAI Chat Completions shape.

Source code in src/convopack/providers/openai.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def to_openai(messages: Iterable[Message]) -> list[dict[str, Any]]:
    """Convert :class:`Message` objects into OpenAI Chat Completions shape."""
    out: list[dict[str, Any]] = []
    for msg in messages:
        if msg.role == "tool":
            tool_results = [b for b in msg.content if isinstance(b, ToolResultBlock)]
            for tr in tool_results:
                payload = (
                    tr.content
                    if isinstance(tr.content, str)
                    else "".join(b.text for b in tr.content if isinstance(b, TextBlock))
                )
                out.append({"role": "tool", "tool_call_id": tr.tool_use_id, "content": payload})
            continue

        text_parts: list[dict[str, Any]] = []
        tool_calls: list[dict[str, Any]] = []
        for block in msg.content:
            if isinstance(block, TextBlock):
                text_parts.append({"type": "text", "text": block.text})
            elif isinstance(block, ImageBlock):
                text_parts.append({"type": "image_url", "image_url": {"url": block.source}})
            elif isinstance(block, ToolUseBlock):
                tool_calls.append(
                    {
                        "id": block.id,
                        "type": "function",
                        "function": {
                            "name": block.name,
                            "arguments": json.dumps(block.input),
                        },
                    }
                )

        entry: dict[str, Any] = {"role": msg.role}
        if msg.name:
            entry["name"] = msg.name
        if len(text_parts) == 1 and text_parts[0]["type"] == "text":
            entry["content"] = text_parts[0]["text"]
        elif text_parts:
            entry["content"] = text_parts
        else:
            entry["content"] = None
        if tool_calls:
            entry["tool_calls"] = tool_calls
        out.append(entry)
    return out

from_anthropic 

from_anthropic(messages: Iterable[dict[str, Any]], system: str | None = None) -> list[Message]

Convert Anthropic-shape messages (and optional system prompt) to internal form.

Source code in src/convopack/providers/anthropic.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def from_anthropic(messages: Iterable[dict[str, Any]], system: str | None = None) -> list[Message]:
    """Convert Anthropic-shape messages (and optional system prompt) to internal form."""
    out: list[Message] = []
    if system:
        out.append(Message(role="system", content=[TextBlock(text=system)]))

    for raw in messages:
        role = cast(Role, raw.get("role", "user"))
        blocks: list[ContentBlock] = []
        content = raw.get("content")
        if isinstance(content, str):
            if content:
                blocks.append(TextBlock(text=content))
        elif isinstance(content, list):
            for part in content:
                ptype = part.get("type")
                if ptype == "text":
                    blocks.append(TextBlock(text=part.get("text", "")))
                elif ptype == "image":
                    src = part.get("source", {})
                    blocks.append(
                        ImageBlock(
                            source=src.get("data", src.get("url", "")),
                            media_type=src.get("media_type"),
                        )
                    )
                elif ptype == "tool_use":
                    blocks.append(
                        ToolUseBlock(
                            id=part.get("id", ""),
                            name=part.get("name", ""),
                            input=part.get("input", {}),
                        )
                    )
                elif ptype == "tool_result":
                    inner = part.get("content", "")
                    if isinstance(inner, list):
                        inner_blocks: list[TextBlock | ImageBlock] = []
                        for ip in inner:
                            if ip.get("type") == "text":
                                inner_blocks.append(TextBlock(text=ip.get("text", "")))
                            elif ip.get("type") == "image":
                                isrc = ip.get("source", {})
                                inner_blocks.append(
                                    ImageBlock(
                                        source=isrc.get("data", ""),
                                        media_type=isrc.get("media_type"),
                                    )
                                )
                        blocks.append(
                            ToolResultBlock(
                                tool_use_id=part.get("tool_use_id", ""),
                                content=inner_blocks,
                                is_error=bool(part.get("is_error", False)),
                            )
                        )
                    else:
                        blocks.append(
                            ToolResultBlock(
                                tool_use_id=part.get("tool_use_id", ""),
                                content=str(inner),
                                is_error=bool(part.get("is_error", False)),
                            )
                        )
        out.append(Message(role=role, content=blocks))
    return out

to_anthropic 

to_anthropic(messages: Iterable[Message], *, cache_markers: list[int] | None = None) -> AnthropicPayload

Convert internal messages to an :class:AnthropicPayload.

System messages are concatenated into a single system string. The remaining messages are emitted in Anthropic content-block shape.

If cache_markers is provided, it is interpreted as indices into the input messages list; the last content block of each marked message receives {"cache_control": {"type": "ephemeral"}}. Markers pointing at system messages are attached to the system string fragment instead, since Anthropic's system field is just a list of text blocks under the hood.

Source code in src/convopack/providers/anthropic.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def to_anthropic(
    messages: Iterable[Message],
    *,
    cache_markers: list[int] | None = None,
) -> AnthropicPayload:
    """Convert internal messages to an :class:`AnthropicPayload`.

    System messages are concatenated into a single ``system`` string. The
    remaining messages are emitted in Anthropic content-block shape.

    If ``cache_markers`` is provided, it is interpreted as indices into the
    *input* ``messages`` list; the **last content block** of each marked
    message receives ``{"cache_control": {"type": "ephemeral"}}``. Markers
    pointing at system messages are attached to the system string fragment
    instead, since Anthropic's system field is just a list of text blocks
    under the hood.
    """
    msgs = list(messages)
    markers = set(cache_markers or [])
    system_parts: list[dict[str, Any]] = []
    api_msgs: list[dict[str, Any]] = []
    for i, msg in enumerate(msgs):
        is_marked = i in markers
        if msg.role == "system":
            entry: dict[str, Any] = {"type": "text", "text": msg.text()}
            if is_marked:
                entry["cache_control"] = {"type": "ephemeral"}
            system_parts.append(entry)
            continue

        blocks = [_block_to_anthropic(b) for b in msg.content]
        if is_marked and blocks:
            blocks[-1] = {**blocks[-1], "cache_control": {"type": "ephemeral"}}

        if msg.role == "tool":
            api_msgs.append({"role": "user", "content": blocks})
            continue
        api_msgs.append({"role": msg.role, "content": blocks})

    return AnthropicPayload(system=_render_system(system_parts), messages=api_msgs)

from_gemini 

from_gemini(contents: Iterable[dict[str, Any]], system_instruction: str | None = None) -> list[Message]

Convert Gemini Content dicts (and optional system instruction).

Source code in src/convopack/providers/gemini.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def from_gemini(
    contents: Iterable[dict[str, Any]],
    system_instruction: str | None = None,
) -> list[Message]:
    """Convert Gemini ``Content`` dicts (and optional system instruction)."""
    out: list[Message] = []
    if system_instruction:
        out.append(Message(role="system", content=[TextBlock(text=system_instruction)]))

    pending_call_ids: list[str] = []
    synth_seq = 0
    for raw in contents:
        gem_role = raw.get("role", "user")
        role: Role = _GEMINI_TO_ROLE.get(gem_role, "user")
        blocks: list[ContentBlock] = []
        for part in raw.get("parts", []):
            if "text" in part:
                blocks.append(TextBlock(text=part["text"]))
            elif "inline_data" in part:
                data = part["inline_data"]
                blocks.append(
                    ImageBlock(
                        source=data.get("data", ""),
                        media_type=data.get("mime_type"),
                    )
                )
            elif "function_call" in part:
                fc = part["function_call"]
                synth_seq += 1
                synth_id = f"{_SYNTH_ID_PREFIX}{synth_seq}"
                pending_call_ids.append(synth_id)
                blocks.append(
                    ToolUseBlock(
                        id=synth_id,
                        name=fc.get("name", ""),
                        input=dict(fc.get("args", {})),
                    )
                )
            elif "function_response" in part:
                fr = part["function_response"]
                tool_id = (
                    pending_call_ids.pop(0) if pending_call_ids else f"{_SYNTH_ID_PREFIX}orphan"
                )
                response = fr.get("response", {})
                content = (
                    response.get("content")
                    if isinstance(response, dict) and "content" in response
                    else response
                )
                blocks.append(
                    ToolResultBlock(
                        tool_use_id=tool_id,
                        content=str(content) if not isinstance(content, str) else content,
                    )
                )
        out.append(Message(role=role, content=blocks))
    return out

to_gemini 

to_gemini(messages: Iterable[Message]) -> GeminiPayload

Convert internal messages to a :class:GeminiPayload.

System messages are concatenated into system_instruction. Tool calls become function_call parts; tool results become function_response parts on a synthetic user turn. Internal tool_use_id values are stripped because Gemini does not carry them.

Source code in src/convopack/providers/gemini.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def to_gemini(messages: Iterable[Message]) -> GeminiPayload:
    """Convert internal messages to a :class:`GeminiPayload`.

    System messages are concatenated into ``system_instruction``. Tool calls
    become ``function_call`` parts; tool results become ``function_response``
    parts on a synthetic ``user`` turn. Internal ``tool_use_id`` values are
    stripped because Gemini does not carry them.
    """
    msgs = list(messages)
    system_parts: list[str] = []
    contents: list[dict[str, Any]] = []
    pending_names: list[str] = []
    for msg in msgs:
        if msg.role == "system":
            system_parts.append(msg.text())
            continue
        parts: list[dict[str, Any]] = []
        for block in msg.content:
            if isinstance(block, TextBlock):
                parts.append({"text": block.text})
            elif isinstance(block, ImageBlock):
                parts.append(
                    {
                        "inline_data": {
                            "mime_type": block.media_type or "image/png",
                            "data": block.source,
                        }
                    }
                )
            elif isinstance(block, ToolUseBlock):
                pending_names.append(block.name)
                parts.append({"function_call": {"name": block.name, "args": dict(block.input)}})
            elif isinstance(block, ToolResultBlock):
                name = pending_names.pop(0) if pending_names else block.tool_use_id
                parts.append(
                    {
                        "function_response": {
                            "name": name,
                            "response": {"content": block.content}
                            if isinstance(block.content, str)
                            else {"content": _blocks_to_text(block.content)},
                        }
                    }
                )
        if parts:
            contents.append({"role": _ROLE_TO_GEMINI[msg.role], "parts": parts})
    return GeminiPayload(
        system_instruction="\n\n".join(p for p in system_parts if p),
        contents=contents,
    )