codekingpro/portable-devtools

1"""Structured prompt template for a language model."""2 3from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence4from typing import (5    Any,6)7 8from pydantic import BaseModel, Field9from typing_extensions import override10 11from langchain_core._api.beta_decorator import beta12from langchain_core.language_models.base import BaseLanguageModel13from langchain_core.prompts.chat import (14    ChatPromptTemplate,15    MessageLikeRepresentation,16)17from langchain_core.prompts.string import PromptTemplateFormat18from langchain_core.runnables.base import (19    Other,20    Runnable,21    RunnableSequence,22    RunnableSerializable,23)24from langchain_core.utils import get_pydantic_field_names25 26 27@beta()28class StructuredPrompt(ChatPromptTemplate):29    """Structured prompt template for a language model."""30 31    schema_: dict | type32    """Schema for the structured prompt."""33 34    structured_output_kwargs: dict[str, Any] = Field(default_factory=dict)35 36    def __init__(37        self,38        messages: Sequence[MessageLikeRepresentation],39        schema_: dict | type[BaseModel] | None = None,40        *,41        structured_output_kwargs: dict[str, Any] | None = None,42        template_format: PromptTemplateFormat = "f-string",43        **kwargs: Any,44    ) -> None:45        """Create a structured prompt template.46 47        Args:48            messages: Sequence of messages.49            schema_: Schema for the structured prompt.50            structured_output_kwargs: Additional kwargs for structured output.51            template_format: Template format for the prompt.52 53        Raises:54            ValueError: If schema is not provided.55        """56        schema_ = schema_ or kwargs.pop("schema", None)57        if not schema_:58            err_msg = (59                "Must pass in a non-empty structured output schema. Received: "60                f"{schema_}"61            )62            raise ValueError(err_msg)63        structured_output_kwargs = structured_output_kwargs or {}64        for k in set(kwargs).difference(get_pydantic_field_names(self.__class__)):65            structured_output_kwargs[k] = kwargs.pop(k)66        super().__init__(67            messages=messages,68            schema_=schema_,69            structured_output_kwargs=structured_output_kwargs,70            template_format=template_format,71            **kwargs,72        )73 74    @classmethod75    def get_lc_namespace(cls) -> list[str]:76        """Get the namespace of the LangChain object.77 78        For example, if the class is `langchain.llms.openai.OpenAI`, then the namespace79        is `["langchain", "llms", "openai"]`80 81        Returns:82            The namespace of the LangChain object.83        """84        return cls.__module__.split(".")85 86    @classmethod87    def from_messages_and_schema(88        cls,89        messages: Sequence[MessageLikeRepresentation],90        schema: dict | type,91        **kwargs: Any,92    ) -> ChatPromptTemplate:93        """Create a chat prompt template from a variety of message formats.94 95        Examples:96            Instantiation from a list of message templates:97 98            ```python99            from langchain_core.prompts import StructuredPrompt100 101 102            class OutputSchema(BaseModel):103                name: str104                value: int105 106 107            template = StructuredPrompt(108                [109                    ("human", "Hello, how are you?"),110                    ("ai", "I'm doing well, thanks!"),111                    ("human", "That's good to hear."),112                ],113                OutputSchema,114            )115            ```116 117        Args:118            messages: Sequence of message representations.119 120                A message can be represented using the following formats:121 122                1. `BaseMessagePromptTemplate`123                2. `BaseMessage`124                3. 2-tuple of `(message type, template)`; e.g.,125                    `("human", "{user_input}")`126                4. 2-tuple of `(message class, template)`127                5. A string which is shorthand for `("human", template)`; e.g.,128                    `"{user_input}"`129            schema: A dictionary representation of function call, or a Pydantic model.130            **kwargs: Any additional kwargs to pass through to131                `ChatModel.with_structured_output(schema, **kwargs)`.132 133        Returns:134            A structured prompt template135        """136        return cls(messages, schema, **kwargs)137 138    @override139    def __or__(140        self,141        other: Runnable[Any, Other]142        | Callable[[Iterator[Any]], Iterator[Other]]143        | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]144        | Callable[[Any], Other]145        | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],146    ) -> RunnableSerializable[dict, Other]:147        return self.pipe(other)148 149    def pipe(150        self,151        *others: Runnable[Any, Other]152        | Callable[[Iterator[Any]], Iterator[Other]]153        | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]154        | Callable[[Any], Other]155        | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],156        name: str | None = None,157    ) -> RunnableSerializable[dict, Other]:158        """Pipe the structured prompt to a language model.159 160        Args:161            others: The language model to pipe the structured prompt to.162            name: The name of the pipeline.163 164        Returns:165            A `RunnableSequence` object.166 167        Raises:168            NotImplementedError: If the first element of `others` is not a language169                model.170        """171        if (others and isinstance(others[0], BaseLanguageModel)) or hasattr(172            others[0], "with_structured_output"173        ):174            return RunnableSequence(175                self,176                others[0].with_structured_output(177                    self.schema_, **self.structured_output_kwargs178                ),179                *others[1:],180                name=name,181            )182        msg = "Structured prompts need to be piped to a language model."183        raise NotImplementedError(msg)184