Skip to content

teff.harness.context

teff.harness.context

Context management — token estimation and message trimming.

Classes:

Name Description
ContextLimitError

Raised when a conversation cannot fit the configured context limits.

Functions:

Name Description
trim_messages

Trim messages down to fit context limits.

ContextLimitError

Bases: WorkflowError

Raised when a conversation cannot fit the configured context limits.

Source code in teff/harness/context.py 
89
90
class ContextLimitError(WorkflowError):
    """Raised when a conversation cannot fit the configured context limits."""

trim_messages 

trim_messages(messages, max_tokens=None, max_messages=None)

Trim messages down to fit context limits.

The leading system message (if any) is always preserved; older messages are dropped from the front of the conversation until the estimated token count and message count fit the limits.

A limit <= 0 keeps only the system message(s). When the system message alone cannot fit max_tokens (and dropping it is not allowed) a :class:ContextLimitError is raised.

Parameters:

Name Type Description Default
messages list[dict]

The conversation history.

 required
max_tokens int | None

Maximum estimated tokens to keep.

 None
max_messages int | None

Maximum number of messages to keep.

 None

Returns:

Type Description
list[dict]

A new list of messages, trimmed from the front (system kept).

Raises:

Type Description
ContextLimitError

When even the system message alone would exceed max_tokens (system is never dropped).
Source code in teff/harness/context.py 
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
def trim_messages(
    messages: list[dict],
    max_tokens: int | None = None,
    max_messages: int | None = None,
) -> list[dict]:
    """Trim *messages* down to fit context limits.

    The leading ``system`` message (if any) is always preserved; older
    messages are dropped from the front of the conversation until the
    estimated token count and message count fit the limits.

    A limit ``<= 0`` keeps only the system message(s).  When the system
    message alone cannot fit *max_tokens* (and dropping it is not allowed)
    a :class:`ContextLimitError` is raised.

    Args:
        messages: The conversation history.
        max_tokens: Maximum estimated tokens to keep.
        max_messages: Maximum number of messages to keep.

    Returns:
        A new list of messages, trimmed from the front (system kept).

    Raises:
        ContextLimitError: When even the system message alone would exceed
            *max_tokens* (system is never dropped).
    """
    if not messages:
        return []

    system: list[dict] = []
    body: list[dict] = []
    for msg in messages:
        if msg.get("role") == "system":
            system.append(msg)
        else:
            body.append(msg)

    if max_messages is not None and max_messages <= 0:
        return system
    if max_tokens is not None and max_tokens <= 0:
        return system

    if max_messages is not None and len(body) > max_messages:
        body = body[-max_messages:]
    if max_tokens is not None and _estimate_tokens(messages) > max_tokens:
        while body and _estimate_tokens(system + body) > max_tokens:
            body.pop(0)
        if not body and _estimate_tokens(system) > max_tokens:
            raise ContextLimitError(
                "conversation system message exceeds max_context_tokens"
            )
    return system + body