Code Examples

1
from typing import Any
2
3
import nemo_relay
4
5
class HeaderPlugin:
6
    def validate(self, plugin_config: dict[str, Any]) -> list[dict[str, str]]:
7
        if "header_name" not in plugin_config or "value" not in plugin_config:
8
            return [{
9
                "level": "error",
10
                "code": "header-plugin.invalid_config",
11
                "message": "header_name and value are required",
12
            }]
13
        return []
14
15
    def register(self, plugin_config: dict[str, Any], context: nemo_relay.plugin.PluginContext):
16
        def add_header(
17
            name: str,
18
            request: nemo_relay.LLMRequest,
19
            annotated: nemo_relay.AnnotatedLLMRequest | None
20
        ) -> tuple[nemo_relay.LLMRequest, nemo_relay.AnnotatedLLMRequest | None]:
21
            # The request object is immutable, however we can return a new instance with updated headers.
22
            headers = request.headers.copy()
23
            headers[plugin_config["header_name"]] = plugin_config["value"]
24
            return nemo_relay.LLMRequest(headers=headers, content=request.content), annotated
25
26
        context.register_llm_request_intercept("inject-header", 100, False, add_header)
27
28
nemo_relay.plugin.register("header-plugin", HeaderPlugin())

1
import nemo_relay
2
3
class OpenInferencePlugin:
4
    def validate(self, plugin_config):
5
        if "endpoint" not in plugin_config:
6
            return [{
7
                "level": "error",
8
                "code": "openinference-export.invalid_config",
9
                "message": "endpoint is required",
10
            }]
11
        return []
12
13
    def register(self, plugin_config, context):
14
        endpoint = plugin_config["endpoint"]
15
16
        def on_event(event):
17
            print("export", endpoint, event.kind, event.name)
18
19
        context.register_subscriber("openinference-export", on_event)
20
21
nemo_relay.plugin.register("openinference-export", OpenInferencePlugin())