Class LlmCodec

python/nemo_relay/codecs.py:60–123

Protocol for request codecs used by annotated LLM intercepts. ``decode()`` converts a provider-specific ``LLMRequest`` into an ``AnnotatedLLMRequest`` so request intercepts can work with a normalized structure. ``encode()`` merges any annotated edits back into the original raw paylo

@runtime_checkable
class LlmCodec(Protocol):
    """Protocol for request codecs used by annotated LLM intercepts.

    ``decode()`` converts a provider-specific ``LLMRequest`` into an
    ``AnnotatedLLMRequest`` so request intercepts can work with a normalized
    structure. ``encode()`` merges any annotated edits back into the original
    raw payload before the provider callback is invoked.

    Notes:
        ``encode()`` should preserve unknown provider-specific fields whenever
        possible instead of rebuilding the payload from scratch. That keeps
        transport-specific settings intact even when an intercept edits the
        normalized representation.

    Example::

        from nemo_relay import AnnotatedLLMRequest, LLMRequest
        from nemo_relay.codecs import LlmCodec

        class DemoCodec(LlmCodec):
            def decode(self, request: LLMRequest) -> AnnotatedLLMRequest:
                return AnnotatedLLMRequest(
                    request.content.get("messages", []),
                    model=request.content.get("model"),
                )

            def encode(
                self,
                annotated: AnnotatedLLMRequest,
                original: LLMRequest,
            ) -> LLMRequest:
                content = {**original.content, "messages": annotated.messages}
                if annotated.model is not None:
                    content["model"] = annotated.model
                return LLMRequest(original.headers, content)
    """

    def decode(self, request: LLMRequest) -> AnnotatedLLMRequest:
        """Decode a raw provider request into ``AnnotatedLLMRequest``.

        Args:
            request: The provider-specific request payload received by
                ``nemo_relay.llm.execute()`` or ``nemo_relay.llm.stream_execute()``.

        Returns:
            AnnotatedLLMRequest: The normalized request consumed by annotated
            intercepts.
        """
        ...

    def encode(self, annotated: AnnotatedLLMRequest, original: LLMRequest) -> LLMRequest:
        """Merge annotated edits back into the original raw request.

        Args:
            annotated: The normalized request after intercepts have applied any
                edits.
            original: The original provider-specific request passed into the
                runtime before normalization.