Coverage for haystack/hooks/compaction/types/protocol.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any, Protocol 

6 

7from haystack.core.serialization import default_from_dict 

8from haystack.dataclasses import ChatMessage 

9from haystack.token_counters import TokenCounter 

10 

11 

12class Compactor(Protocol): 

13 """ 

14 Rewrites an Agent's conversation into a shorter one that carries the same working context. 

15 

16 A compactor is the *how* of context compaction; deciding *when* to compact is the caller's job, which 

17 `CompactionHook` does by comparing the context size against a fraction of the model's window. Strategies 

18 differ widely in cost and fidelity, from dropping the oldest messages outright to condensing them with an LLM. 

19 

20 Implementations must honor three rules: 

21 

22 1. **Return `None` unless the conversation actually gets smaller.** Callers apply whatever else is returned, so 

23 judging whether compacting was worthwhile is the compactor's job. 

24 2. **Return a new list; leave `messages` as it is.** The caller owns that list and writes the returned one back. 

25 3. **Keep tool calls and their results together.** Do not retain a tool result after removing the assistant message 

26 that contains its originating call, or retain a tool call without all of its results. Chat-completion APIs reject 

27 these incomplete tool-call exchanges. 

28 

29 `target_tokens` is a goal, not a guarantee: a compactor that cannot reach it should get as close as it can rather 

30 than strip the conversation past what the Agent needs to keep working. 

31 

32 Implement `to_dict` so the compactor's settings survive serialization. The default `from_dict` passes them straight 

33 back to the constructor, which is enough for plain values; override it when `to_dict` emitted something that has to 

34 be rebuilt first, such as a `Secret` or a nested component. 

35 """ 

36 

37 def compact( 

38 self, messages: list[ChatMessage], target_tokens: int, token_counter: TokenCounter 

39 ) -> list[ChatMessage] | None: 

40 """ 

41 Return a shorter replacement for `messages`, or None to leave it unchanged. 

42 

43 :param messages: The conversation to compact, oldest to newest. 

44 :param target_tokens: The size the compacted conversation should come in under. 

45 :param token_counter: The `TokenCounter` to measure messages with. The same one the caller sized the context 

46 with, so a compactor's measurements are consistent with the decision to compact. 

47 :returns: The replacement conversation, or None when this compactor has nothing to change. 

48 """ 

49 ... 

50 

51 async def compact_async( 

52 self, messages: list[ChatMessage], target_tokens: int, token_counter: TokenCounter 

53 ) -> list[ChatMessage] | None: 

54 """ 

55 Asynchronously return a shorter replacement for `messages`, or None to leave it unchanged. 

56 

57 The default implementation calls `compact` directly. Override it when compaction does I/O, so the event loop is 

58 not blocked. 

59 

60 :param messages: The conversation to compact, oldest to newest. 

61 :param target_tokens: The size the compacted conversation should come in under. 

62 :param token_counter: The `TokenCounter` to measure messages with. The same one the caller sized the context 

63 with, so a compactor's measurements are consistent with the decision to compact. 

64 :returns: The replacement conversation, or None when this compactor has nothing to change. 

65 """ 

66 return self.compact(messages, target_tokens, token_counter) 

67 

68 def to_dict(self) -> dict[str, Any]: 

69 """Serialize the compactor to a dictionary.""" 

70 ... 

71 

72 @classmethod 

73 def from_dict(cls, data: dict[str, Any]) -> "Compactor": 

74 """Deserialize the compactor from a dictionary.""" 

75 return default_from_dict(cls, data)