_first_max_tokens() — langchain Function Reference

Architecture documentation for the _first_max_tokens() function in utils.py from the langchain codebase.

Function python CoreAbstractions MessageSchema calls 1 called by 2

Entity Profile

CoreAbstractions→ MessageSchema→ _first_max_tokens() — langchain Function Reference

Dependency Diagram

graph TD
  83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf["_first_max_tokens()"]
  0b528c80_0ce7_1c74_8932_bc433bcb03c6["utils.py"]
  83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf -->|defined in| 0b528c80_0ce7_1c74_8932_bc433bcb03c6
  d6e12d35_63a7_f8dc_cd86_45811dae362e["trim_messages()"]
  d6e12d35_63a7_f8dc_cd86_45811dae362e -->|calls| 83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf
  9e133c6c_250c_1c48_07a7_cf5acf5d50d3["_last_max_tokens()"]
  9e133c6c_250c_1c48_07a7_cf5acf5d50d3 -->|calls| 83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf
  f7cf2143_3a02_fa39_15a7_05fb127258ed["_is_message_type()"]
  83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf -->|calls| f7cf2143_3a02_fa39_15a7_05fb127258ed
  style 83dc7aaa_e1bf_befe_66c0_fdea2b3bccdf fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/messages/utils.py lines 1912–2025

def _first_max_tokens(
    messages: Sequence[BaseMessage],
    *,
    max_tokens: int,
    token_counter: Callable[[list[BaseMessage]], int],
    text_splitter: Callable[[str], list[str]],
    partial_strategy: Literal["first", "last"] | None = None,
    end_on: str | type[BaseMessage] | Sequence[str | type[BaseMessage]] | None = None,
) -> list[BaseMessage]:
    messages = list(messages)
    if not messages:
        return messages

    # Check if all messages already fit within token limit
    if token_counter(messages) <= max_tokens:
        # When all messages fit, only apply end_on filtering if needed
        if end_on:
            for _ in range(len(messages)):
                if not _is_message_type(messages[-1], end_on):
                    messages.pop()
                else:
                    break
        return messages

    # Use binary search to find the maximum number of messages within token limit
    left, right = 0, len(messages)
    max_iterations = len(messages).bit_length()
    for _ in range(max_iterations):
        if left >= right:
            break
        mid = (left + right + 1) // 2
        if token_counter(messages[:mid]) <= max_tokens:
            left = mid
            idx = mid
        else:
            right = mid - 1

    # idx now contains the maximum number of complete messages we can include
    idx = left

    if partial_strategy and idx < len(messages):
        included_partial = False
        copied = False
        if isinstance(messages[idx].content, list):
            excluded = messages[idx].model_copy(deep=True)
            copied = True
            num_block = len(excluded.content)
            if partial_strategy == "last":
                excluded.content = list(reversed(excluded.content))
            for _ in range(1, num_block):
                excluded.content = excluded.content[:-1]
                if token_counter([*messages[:idx], excluded]) <= max_tokens:
                    messages = [*messages[:idx], excluded]
                    idx += 1
                    included_partial = True
                    break
            if included_partial and partial_strategy == "last":
                excluded.content = list(reversed(excluded.content))
        if not included_partial:
            if not copied:
                excluded = messages[idx].model_copy(deep=True)
                copied = True

            # Extract text content efficiently
            text = None
            if isinstance(excluded.content, str):
                text = excluded.content
            elif isinstance(excluded.content, list) and excluded.content:
                for block in excluded.content:
                    if isinstance(block, str):
                        text = block
                        break
                    if isinstance(block, dict) and block.get("type") == "text":
                        text = block.get("text")
                        break

            if text:
                if not copied:
                    excluded = excluded.model_copy(deep=True)

                split_texts = text_splitter(text)