apply() — langchain Function Reference

Architecture documentation for the apply() function in context_editing.py from the langchain codebase.

Function python LangChainCore Runnables calls 2 called by 1

Entity Profile

LangChainCore→ Runnables→ apply() — langchain Function Reference

Dependency Diagram

graph TD
  03942334_6db9_570a_7836_af54672f2984["apply()"]
  342553be_63ff_6f10_00db_f8b52bed863a["ClearToolUsesEdit"]
  03942334_6db9_570a_7836_af54672f2984 -->|defined in| 342553be_63ff_6f10_00db_f8b52bed863a
  eb1539c1_1bd9_cd97_dcd3_ca4a0fa011f2["apply()"]
  eb1539c1_1bd9_cd97_dcd3_ca4a0fa011f2 -->|calls| 03942334_6db9_570a_7836_af54672f2984
  ee42af87_4c5e_4b39_d0f1_983638d93ef0["_build_cleared_tool_input_message()"]
  03942334_6db9_570a_7836_af54672f2984 -->|calls| ee42af87_4c5e_4b39_d0f1_983638d93ef0
  eb1539c1_1bd9_cd97_dcd3_ca4a0fa011f2["apply()"]
  03942334_6db9_570a_7836_af54672f2984 -->|calls| eb1539c1_1bd9_cd97_dcd3_ca4a0fa011f2
  style 03942334_6db9_570a_7836_af54672f2984 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/context_editing.py lines 79–155

    def apply(
        self,
        messages: list[AnyMessage],
        *,
        count_tokens: TokenCounter,
    ) -> None:
        """Apply the clear-tool-uses strategy."""
        tokens = count_tokens(messages)

        if tokens <= self.trigger:
            return

        candidates = [
            (idx, msg) for idx, msg in enumerate(messages) if isinstance(msg, ToolMessage)
        ]

        if self.keep >= len(candidates):
            candidates = []
        elif self.keep:
            candidates = candidates[: -self.keep]

        cleared_tokens = 0
        excluded_tools = set(self.exclude_tools)

        for idx, tool_message in candidates:
            if tool_message.response_metadata.get("context_editing", {}).get("cleared"):
                continue

            ai_message = next(
                (m for m in reversed(messages[:idx]) if isinstance(m, AIMessage)), None
            )

            if ai_message is None:
                continue

            tool_call = next(
                (
                    call
                    for call in ai_message.tool_calls
                    if call.get("id") == tool_message.tool_call_id
                ),
                None,
            )

            if tool_call is None:
                continue

            if (tool_message.name or tool_call["name"]) in excluded_tools:
                continue

            messages[idx] = tool_message.model_copy(
                update={
                    "artifact": None,
                    "content": self.placeholder,
                    "response_metadata": {
                        **tool_message.response_metadata,
                        "context_editing": {
                            "cleared": True,
                            "strategy": "clear_tool_uses",
                        },
                    },
                }
            )

            if self.clear_tool_inputs:
                messages[messages.index(ai_message)] = self._build_cleared_tool_input_message(
                    ai_message,
                    tool_message.tool_call_id,
                )

            if self.clear_at_least > 0:
                new_token_count = count_tokens(messages)
                cleared_tokens = max(0, tokens - new_token_count)
                if cleared_tokens >= self.clear_at_least:
                    break

        return