Source code for langchain_experimental.plan_and_execute.schema

from abc import abstractmethod
from typing import List, Tuple

from langchain_core.output_parsers import BaseOutputParser

from langchain_experimental.pydantic_v1 import BaseModel, Field


class Step(BaseModel):
    """步骤。"""

    value: str
    """该数值。"""


class Plan(BaseModel):
    """计划。"""

    steps: List[Step]
    """这些步骤。"""


class StepResponse(BaseModel):
    """阶跃响应。"""

    response: str
    """响应。"""


class BaseStepContainer(BaseModel):
    """基本步骤容器。"""

    @abstractmethod
    def add_step(self, step: Step, step_response: StepResponse) -> None:
        """向容器中添加步骤和步骤响应。"""

    @abstractmethod
    def get_final_response(self) -> str:
        """根据所采取的步骤返回最终响应。"""


class ListStepContainer(BaseStepContainer):
    """步骤列表的容器。"""

    steps: List[Tuple[Step, StepResponse]] = Field(default_factory=list)
    """这些步骤。"""

    def add_step(self, step: Step, step_response: StepResponse) -> None:
        self.steps.append((step, step_response))

    def get_steps(self) -> List[Tuple[Step, StepResponse]]:
        return self.steps

    def get_final_response(self) -> str:
        return self.steps[-1][1].response


class PlanOutputParser(BaseOutputParser):
    """计划输出解析器。"""

    @abstractmethod
    def parse(self, text: str) -> Plan:
        """解析为一个计划。"""