Source code for langchain_core.prompts.pipeline

from typing import Any, Dict, List, Tuple

from langchain_core.prompt_values import PromptValue
from langchain_core.prompts.base import BasePromptTemplate
from langchain_core.prompts.chat import BaseChatPromptTemplate
from langchain_core.pydantic_v1 import root_validator


def _get_inputs(inputs: dict, input_variables: List[str]) -> dict:
    return {k: inputs[k] for k in input_variables}


class PipelinePromptTemplate(BasePromptTemplate):
    """用于将多个提示模板组合在一起的提示模板。

当您想要重用提示的部分时，这可能会很有用。
PipelinePrompt由两个主要部分组成：
    - final_prompt：这是返回的最终提示
    - pipeline_prompts：这是一个元组列表，包括
        一个字符串（`name`）和一个提示模板。
        每个PromptTemplate将被格式化，然后作为一个变量传递
        给未来的提示模板，变量名与`name`相同。
"""

    final_prompt: BasePromptTemplate
    """返回的最终提示。"""
    pipeline_prompts: List[Tuple[str, BasePromptTemplate]]
    """一个元组列表，包括一个字符串（`name`）和一个提示模板。"""

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """获取langchain对象的命名空间。"""
        return ["langchain", "prompts", "pipeline"]

    @root_validator(pre=True)
    def get_input_variables(cls, values: Dict) -> Dict:
        """获取输入变量。"""
        created_variables = set()
        all_variables = set()
        for k, prompt in values["pipeline_prompts"]:
            created_variables.add(k)
            all_variables.update(prompt.input_variables)
        values["input_variables"] = list(all_variables.difference(created_variables))
        return values

    def format_prompt(self, **kwargs: Any) -> PromptValue:
        for k, prompt in self.pipeline_prompts:
            _inputs = _get_inputs(kwargs, prompt.input_variables)
            if isinstance(prompt, BaseChatPromptTemplate):
                kwargs[k] = prompt.format_messages(**_inputs)
            else:
                kwargs[k] = prompt.format(**_inputs)
        _inputs = _get_inputs(kwargs, self.final_prompt.input_variables)
        return self.final_prompt.format_prompt(**_inputs)

    async def aformat_prompt(self, **kwargs: Any) -> PromptValue:
        for k, prompt in self.pipeline_prompts:
            _inputs = _get_inputs(kwargs, prompt.input_variables)
            if isinstance(prompt, BaseChatPromptTemplate):
                kwargs[k] = await prompt.aformat_messages(**_inputs)
            else:
                kwargs[k] = await prompt.aformat(**_inputs)
        _inputs = _get_inputs(kwargs, self.final_prompt.input_variables)
        return await self.final_prompt.aformat_prompt(**_inputs)

    def format(self, **kwargs: Any) -> str:
        return self.format_prompt(**kwargs).to_string()

    async def aformat(self, **kwargs: Any) -> str:
        return (await self.aformat_prompt(**kwargs)).to_string()

    @property
    def _prompt_type(self) -> str:
        raise ValueError