Integrations 與 Plugins

Osprey 可以在不修改核心 codebase 的情況下擴充。平台可以透過 Osprey 啟動時自動發現的 Plugin packages，接入偵測函式、output destinations、Entity state storage 與 ML models 等自訂邏輯。Plugin package 可以實作可用 hooks表格中的任意子集。本頁說明採用團隊最常詢問的 integration points。

可執行的參考 package 請見 example_plugins/ directory。

Plugins 載入方式

Osprey 使用 pluggy 發現 Plugins。Plugin package 會在 pyproject.toml 宣告下列一個或兩個 entry-point groups。

osprey_plugin　由標準 gevent Worker 載入
osprey_async_plugin　由 experimental asyncio Worker 載入

例如下列設定。

[project.entry-points.osprey_plugin]
register_plugins = "register_plugins"

[project.entry-points.osprey_async_plugin]
register_async_plugins = "register_async_plugins"

每個 entry point 都會解析到包含 hook functions 的 module，並以 @hookimpl_osprey 或 @hookimpl_osprey_async 裝飾。Osprey 啟動時會呼叫各 hook 收集註冊項目。詳情請見 example_plugins/src/register_plugins.py 與 register_async_plugins.py。

撰寫 UDFs

User-defined function，UDF，是可從 Rules 呼叫的 Python class。UDFs 將文字比對、DNS lookups、hash comparisons 或 ML inference 等可重複使用的偵測邏輯，包裝成規則語言中的具名函式。語言層級的說明目前請見英文官方文件 Writing Rules § User Defined Functions。

UDF 的結構

UDFs 需要兩個部分。

Arguments class　繼承 ArgumentsBase，以 Types 宣告 UDF 接受的 parameters
UDF class　繼承 UDFBase[Arguments, ReturnType]，並以 execute method 實作邏輯

例如下列程式碼。

# example_plugins/src/udfs/text_contains.py
import re

from osprey.engine.executor.execution_context import ExecutionContext
from osprey.engine.udf.arguments import ArgumentsBase
from osprey.engine.udf.base import UDFBase


class TextContainsArguments(ArgumentsBase):
    text: str
    phrase: str
    case_sensitive = False


class TextContains(UDFBase[TextContainsArguments, bool]):
    def execute(self, execution_context: ExecutionContext, arguments: TextContainsArguments) -> bool:
        escaped = re.escape(arguments.phrase)

        pattern = rf'\b{escaped}\b'

        flags = 0 if arguments.case_sensitive else re.IGNORECASE
        regex = re.compile(pattern, flags)

        return bool(regex.search(arguments.text))

完成註冊後，可以在 Rules 中呼叫 TextContains。

TextContains(text=SomeFeature, phrase="spam")

具有 side effects 的 UDFs

UDFs 也能產生 Effects，也就是供 downstream systems 執行的 structured outputs，例如停用使用者或標示內容。Effects 使用 EffectBase 作為 return type。範例請見 example_plugins/src/udfs/ban_user.py。

Async UDFs

執行 network calls 或 database reads 等 I/O 的 UDFs，在 async Worker 中使用時應繼承 AsyncUDFBase。範例請見 osprey_async_worker/src/osprey/async_worker/stdlib_udfs/async_mx_lookup.py。TextContains 等只進行運算的 UDFs，可以不修改而同時用於兩種 Workers。

註冊 UDFs

從 register_udfs hook 回傳 UDF classes。

from osprey.worker.adaptor.plugin_manager import hookimpl_osprey

@hookimpl_osprey
def register_udfs():
    return [TextContains, BanUser]

使用 UdfCategories 為每個 UDF 指定 category。定義位於 osprey_worker/src/osprey/engine/stdlib/udfs/categories.py，例如 STRING、HASH、ENTITY、HTTP，讓 UDF 在 UDF Registry 中依合理類別分組。

Hash-based lookups

Osprey standard library 內建 Hash* UDF family，包括 HashMd5、HashSha1、HashSha256 與 HashSha512，歸類於 HASH 且不需額外註冊。這些 UDF 接受 string input 並回傳 hex digest。搭配 SML 的 in operator 或 HasLabel，可以在不保存 raw data 的情況下比對 known-bad sets。

# Check a hashed value against a small inline set
IsKnownBadHash = HashSha256(input=SomeValue) in ['abc123...', 'def456...']

# Or check membership via a label that was set by some other process
IsKnownBadActor = HasLabel(entity=SomeEntity, label='KnownBad')

Inline sets 適合小型且穩定的清單。Osprey 沒有 bulk-import 或 lookup-table primitive；若外部清單包含數百萬筆 hashes 且經常更新，應撰寫自訂 UDF 查詢自己的 store。

設定 input sinks

Input sink 是 Events 進入 Osprey 的位置。Osprey 內建 Kafka、Google Pub/Sub、Osprey Coordinator，以及供本機測試使用的 synthetic generator，並以 InputStreamSource config value 選擇。若都不符合平台需求，可以註冊 custom input stream Plugin。概念請見資料流的輸入資料。

內建 sources

Worker 依 InputStreamSource 選擇 input stream。

Source	Config	Use case
`KAFKA`	`OSPREY_KAFKA_INPUT_STREAM_TOPIC`, `OSPREY_KAFKA_BOOTSTRAP_SERVERS`	從 Kafka topic 消費 Action Events
`PUBSUB`	`PUBSUB_OSPREY_PROJECT_ID`, `PUBSUB_OSPREY_RULES_SINK_SUBSCRIPTION`	從 Google Pub/Sub 消費
`OSPREY_COORDINATOR`	`OSPREY_COORDINATOR_SERVICE_NAME`	從 Osprey Coordinator service 取得工作
`SYNTHETIC`		產生隨機假 Events，適合沒有 upstream system 的本機開發
`PLUGIN`		委派給註冊的 `register_input_stream` hook

若 Events 已經由 Kafka 傳送，設定 InputStreamSource.KAFKA；使用 Google Pub/Sub 時則設定 InputStreamSource.PUBSUB。其他情況可以實作 custom input stream，並在 config 設定 InputStreamSource.PLUGIN。若 Events 使用 protobuf 而非 JSON，另可透過 register_action_proto_deserializer hook 提供 deserializer。

撰寫 custom input stream

若 event source 不是 Kafka 或 Pub/Sub，例如 webhook receiver、其他 message Queue 或 polling API，可以繼承 BaseInputStream 並實作 _gen generator。每個 event 會 yield 一個包在 AckingContext 中的 Action。

from collections.abc import Iterator

from osprey.engine.executor.execution_context import Action
from osprey.worker.sinks.sink.input_stream import BaseInputStream
from osprey.worker.sinks.utils.acking_contexts import BaseAckingContext, NoopAckingContext


class MyInputStream(BaseInputStream[BaseAckingContext[Action]]):
    def __init__(self, my_client):
        super().__init__()
        self._client = my_client

    def _gen(self) -> Iterator[BaseAckingContext[Action]]:
        while True:
            raw_event = self._client.poll()  # block until the next event
            action = Action(
                action_id=int(raw_event['id']),
                action_name=raw_event['type'],
                data=raw_event['payload'],
                timestamp=raw_event['timestamp'],
            )
            yield NoopAckingContext(item=action)

_gen 只會呼叫一次並重複使用。它應持續 block 與 yield，不應 return。若 source 不需要明確 ack 或 nack，使用 NoopAckingContext；若 Queue 提供 at-least-once delivery 等機制，則實作 custom BaseAckingContext，在成功時 ack。

從 hook 註冊 input stream，並在 config 設定 InputStreamSource.PLUGIN。

@hookimpl_osprey
def register_input_stream(config):
    return MyInputStream(my_client=build_client(config))

設定 output sinks

Rule 評估後，每個 ExecutionResult 都會傳給 output sink，由 sink 決定如何處理，例如寫入 log、轉送至 Queue、呼叫 webhook 或寫入 database。概念請見資料流的輸出資料。若要將 Results 保存至 BigTable、GCS、MinIO、Postgres 以外的 backend，應改用 register_execution_result_store hook。

Sync output sink

繼承 BaseOutputSink 並實作 methods。

from osprey.worker.sinks.sink.output_sink import BaseOutputSink
from osprey.engine.executor.execution_context import ExecutionResult


class MyOutputSink(BaseOutputSink):
    def will_do_work(self, result: ExecutionResult) -> bool:
        # Return False to skip this result early (e.g. filter by rule hit)
        return True

    def push(self, result: ExecutionResult) -> None:
        # Do something with the result—send to a queue, call an API, etc.
        pass

    def stop(self) -> None:
        # Clean up connections, flush buffers
        pass

從 hook 註冊。

@hookimpl_osprey
def register_output_sinks(config):
    return [MyOutputSink()]

Async output sink

使用 async Worker 時，繼承 AsyncBaseOutputSink，並將 push 與 stop 實作為 coroutines。範例請見 example_plugins/src/async_sinks/example_async_output_sink.py。

from osprey.async_worker.adaptor.interfaces import AsyncBaseOutputSink
import logging

logger = logging.getLogger(__name__)

class ExampleAsyncOutputSink(AsyncBaseOutputSink):
    def will_do_work(self, result: ExecutionResult) -> bool:
        return True

    async def push(self, result: ExecutionResult) -> None:
        logger.info(
            'example async output sink: features=%s verdicts=%s',
            result.extracted_features_json,
            result.verdicts,
        )

    async def stop(self) -> None:
        pass

使用 @hookimpl_osprey_async，在 register_async_output_sinks hook 下註冊。這與 sync register_output_sinks 是不同 hook，應放在連接至 osprey_async_plugin entry point 的 register_async_plugins.py module。

from osprey.async_worker.adaptor.plugin_manager import hookimpl_osprey_async

@hookimpl_osprey_async
def register_async_output_sinks(config):
    return [ExampleAsyncOutputSink()]

Labels service

Osprey 透過 Entity Labels 在 Events 之間追蹤狀態。Labels 是附加於使用者、帳號或其他 Entities 的任意 tags，例如「這位使用者先前有三次違規」。Rule 評估時會讀取 Labels，Rules 也能透過 Label Effects 寫入。若要讓 Labels 在 process restart 後繼續存在，並供多個 Workers 共用，需要透過 register_labels_service_or_provider hook 提供以自有 storage 為 backend 的 LabelsServiceBase 實作。

example_plugins/src/services/labels_service.py 的範例使用 PostgreSQL。

from osprey.worker.lib.storage.labels import LabelsServiceBase

class PostgresLabelsService(LabelsServiceBase):
    def initialize(self) -> None:
        # Called once at startup—open connections here
        ...

    def read_labels(self, entity) -> EntityLabels:
        # Return labels for this entity from your store
        ...

    @contextmanager
    def read_modify_write_labels_atomically(self, entity):
        # Yield the current labels; caller mutates them in place;
        # persist the result before the context manager exits
        ...

從 hook 註冊 service。

@hookimpl_osprey
def register_labels_service_or_provider(config):
    return PostgresLabelsService()

連接審查工具

Osprey 目前沒有直接整合審查工具，但可利用下列 extension points 完成整合。

register_output_sinks　在產生 Execution Results 時將資料傳入 review Queue
register_label_output_sink　專門處理 Label mutations 並取代預設 LabelOutputSink 的 sink
以上一節所述、使用既有 datastore 的 labels service。Rule 可以將 Entity 標記為 flagged，再由 review Queue 查詢自有 store 的 Label

接入自有 ML model

ML models 可以實作成 UDFs。使用 in-process model 時，在 execute 中包裝 model 的 predict call。UDF 的 __init__ 會從 framework 收到 validation_context 與 arguments，因此 override 時須接受並轉交兩者，再於 super().__init__() 後載入 model。

class Arguments(ArgumentsBase):
    text: str

class MySpamClassifier(UDFBase[Arguments, float]):
    def __init__(self, validation_context, arguments):
        super().__init__(validation_context, arguments)
        self._model = load_model("/path/to/model.pkl")

    def execute(self, execution_context: ExecutionContext, arguments: Arguments) -> float:
        return self._model.predict_proba([arguments.text])[0][1]

回傳的 score 可以在 Rules 中使用。

MySpamClassifier(text=MessageContent) > 0.85

Rules compile 時，Osprey 會為每個 call site 建立一個 UDF instance，而非為每個 event 建立，因此 model 不會在每個 event 重新載入。Instance 是每個 call site 一份，不是每個 class 一份。若多個 Rules 呼叫同一個 UDF，每個 call site 都會建立 instance 並載入各自的 model copy。大型 model 應集中從單一 Rule 呼叫 UDF，或使用 module-level cache 共用 loaded weights，避免從多個位置呼叫。

若 model 由遠端提供，同樣在 execute() 中透過 HTTP、gRPC 或 model server SDK 呼叫。遠端 model calls 通常較慢或有成本，應使用英文官方文件 Writing Rules 的 Require(..., require_if=...) pattern，只在相關情況執行。

Require(rule='ai_services/my_ai_service.sml', require_if=ActionName == 'register')

安全與資料提醒

UDF、input stream、output sink、Labels service 與遠端 ML model 都可能接觸平台 Events 或完整 Execution Results。正式導入時應限制 Plugin 來源與權限、管理 secrets、設定 timeout 與 failure handling，並確認傳往第三方服務的資料範圍、保存及刪除條件。Hashing 也不必然等同匿名化，仍應依資料可連結性與使用情境評估。

封裝 Plugin

Plugin package 需要在 pyproject.toml 宣告 entry points。

[project]
name = "my-osprey-plugins"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["pluggy==1.5.0"]

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

[project.entry-points.osprey_plugin]
register_plugins = "register_plugins"

將 package 安裝到與 Osprey 相同的 environment，Osprey 下次啟動時就會自動發現。

另請參考英文官方文件撰寫規則。

Keyboard shortcuts

Osprey 台灣繁體中文指南