before_model() — langchain Function Reference

Architecture documentation for the before_model() function in pii.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 1 called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ before_model() — langchain Function Reference

Dependency Diagram

graph TD
  e81931fa_c61c_8dac_e9d4_8b63fbff471d["before_model()"]
  dc527b75_da40_4d1c_e472_82d1252ba70b["PIIMiddleware"]
  e81931fa_c61c_8dac_e9d4_8b63fbff471d -->|defined in| dc527b75_da40_4d1c_e472_82d1252ba70b
  9c796f5d_d259_f87b_dbac_56cc6e575e35["abefore_model()"]
  9c796f5d_d259_f87b_dbac_56cc6e575e35 -->|calls| e81931fa_c61c_8dac_e9d4_8b63fbff471d
  ffe55823_7c55_8b73_2be6_a2b281bb64cf["_process_content()"]
  e81931fa_c61c_8dac_e9d4_8b63fbff471d -->|calls| ffe55823_7c55_8b73_2be6_a2b281bb64cf
  style e81931fa_c61c_8dac_e9d4_8b63fbff471d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/pii.py lines 171–263

    def before_model(
        self,
        state: AgentState[Any],
        runtime: Runtime[ContextT],
    ) -> dict[str, Any] | None:
        """Check user messages and tool results for PII before model invocation.

        Args:
            state: The current agent state.
            runtime: The langgraph runtime.

        Returns:
            Updated state with PII handled according to strategy, or `None` if no PII
                detected.

        Raises:
            PIIDetectionError: If PII is detected and strategy is `'block'`.
        """
        if not self.apply_to_input and not self.apply_to_tool_results:
            return None

        messages = state["messages"]
        if not messages:
            return None

        new_messages = list(messages)
        any_modified = False

        # Check user input if enabled
        if self.apply_to_input:
            # Get last user message
            last_user_msg = None
            last_user_idx = None
            for i in range(len(messages) - 1, -1, -1):
                if isinstance(messages[i], HumanMessage):
                    last_user_msg = messages[i]
                    last_user_idx = i
                    break

            if last_user_idx is not None and last_user_msg and last_user_msg.content:
                # Detect PII in message content
                content = str(last_user_msg.content)
                new_content, matches = self._process_content(content)

                if matches:
                    updated_message: AnyMessage = HumanMessage(
                        content=new_content,
                        id=last_user_msg.id,
                        name=last_user_msg.name,
                    )

                    new_messages[last_user_idx] = updated_message
                    any_modified = True

        # Check tool results if enabled
        if self.apply_to_tool_results:
            # Find the last AIMessage, then process all `ToolMessage` objects after it
            last_ai_idx = None
            for i in range(len(messages) - 1, -1, -1):
                if isinstance(messages[i], AIMessage):
                    last_ai_idx = i
                    break

            if last_ai_idx is not None:
                # Get all tool messages after the last AI message
                for i in range(last_ai_idx + 1, len(messages)):
                    msg = messages[i]
                    if isinstance(msg, ToolMessage):
                        tool_msg = msg
                        if not tool_msg.content:
                            continue

                        content = str(tool_msg.content)
                        new_content, matches = self._process_content(content)

                        if not matches:
                            continue

                        # Create updated tool message
                        updated_message = ToolMessage(
                            content=new_content,