OpenAIModerationMiddleware Class — langchain Architecture

Source Code

libs/partners/openai/langchain_openai/middleware/openai_moderation.py lines 49–478

class OpenAIModerationMiddleware(AgentMiddleware[AgentState[Any], Any]):
    """Moderate agent traffic using OpenAI's moderation endpoint."""

    def __init__(
        self,
        *,
        model: ModerationModel = "omni-moderation-latest",
        check_input: bool = True,
        check_output: bool = True,
        check_tool_results: bool = False,
        exit_behavior: Literal["error", "end", "replace"] = "end",
        violation_message: str | None = None,
        client: OpenAI | None = None,
        async_client: AsyncOpenAI | None = None,
    ) -> None:
        """Create the middleware instance.

        Args:
            model: OpenAI moderation model to use.
            check_input: Whether to check user input messages.
            check_output: Whether to check model output messages.
            check_tool_results: Whether to check tool result messages.
            exit_behavior: How to handle violations
                (`'error'`, `'end'`, or `'replace'`).
            violation_message: Custom template for violation messages.
            client: Optional pre-configured OpenAI client to reuse.
                If not provided, a new client will be created.
            async_client: Optional pre-configured AsyncOpenAI client to reuse.
                If not provided, a new async client will be created.
        """
        super().__init__()
        self.model = model
        self.check_input = check_input
        self.check_output = check_output
        self.check_tool_results = check_tool_results
        self.exit_behavior = exit_behavior
        self.violation_message = violation_message

        self._client = client
        self._async_client = async_client

    @hook_config(can_jump_to=["end"])
    def before_model(
        self, state: AgentState[Any], runtime: Runtime[Any]
    ) -> dict[str, Any] | None:  # type: ignore[override]
        """Moderate user input and tool results before the model is called.

        Args:
            state: Current agent state containing messages.
            runtime: Agent runtime context.

        Returns:
            Updated state with moderated messages, or `None` if no changes.
        """
        if not self.check_input and not self.check_tool_results:
            return None

        messages = list(state.get("messages", []))
        if not messages:
            return None

        return self._moderate_inputs(messages)

    @hook_config(can_jump_to=["end"])
    def after_model(
        self, state: AgentState[Any], runtime: Runtime[Any]
    ) -> dict[str, Any] | None:  # type: ignore[override]
        """Moderate model output after the model is called.

        Args:
            state: Current agent state containing messages.
            runtime: Agent runtime context.

        Returns:
            Updated state with moderated messages, or `None` if no changes.
        """
        if not self.check_output:
            return None

        messages = list(state.get("messages", []))
        if not messages: