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
[docs]class Step(BaseModel):
"""Step."""
value: str
"""The value."""
[docs]class Plan(BaseModel):
"""Plan."""
steps: List[Step]
"""The steps."""
[docs]class StepResponse(BaseModel):
"""Step response."""
response: str
"""The response."""
[docs]class BaseStepContainer(BaseModel):
"""Base step container."""
[docs] @abstractmethod
def add_step(self, step: Step, step_response: StepResponse) -> None:
"""Add step and step response to the container."""
[docs] @abstractmethod
def get_final_response(self) -> str:
"""Return the final response based on steps taken."""
[docs]class ListStepContainer(BaseStepContainer):
"""Container for List of steps."""
steps: List[Tuple[Step, StepResponse]] = Field(default_factory=list)
"""The steps."""
[docs] def add_step(self, step: Step, step_response: StepResponse) -> None:
self.steps.append((step, step_response))
[docs] def get_steps(self) -> List[Tuple[Step, StepResponse]]:
return self.steps
[docs] def get_final_response(self) -> str:
return self.steps[-1][1].response
[docs]class PlanOutputParser(BaseOutputParser):
"""Plan output parser."""
[docs] @abstractmethod
def parse(self, text: str) -> Plan:
"""Parse into a plan."""