Base interface that all chains should implement. Chain will standardize inputs and outputs, the main entry pointy is the run function.
| 19 | |
| 20 | |
| 21 | class BaseChain(BaseModel, ABC): |
| 22 | """ |
| 23 | Base interface that all chains should implement. |
| 24 | Chain will standardize inputs and outputs, the main entry pointy is the run function. |
| 25 | """ |
| 26 | |
| 27 | agent: Optional[BaseAgent] = None |
| 28 | memory: Optional[BaseMemory] = None |
| 29 | last_query: str = "" |
| 30 | max_iterations: Optional[int] = 15 |
| 31 | max_execution_time: Optional[float] = None |
| 32 | |
| 33 | def prep_inputs(self, user_query: str) -> Dict[str, str]: |
| 34 | """Load conversation history from memory and prep inputs.""" |
| 35 | inputs = { |
| 36 | constants.CONVERSATION_HISTORY: ChatMessageHistory(), |
| 37 | constants.INTERMEDIATE_STEPS: [], |
| 38 | } |
| 39 | if self.memory is not None: |
| 40 | intermediate_steps = self.memory.load_memory( |
| 41 | constants.INTERMEDIATE_STEPS, [] |
| 42 | ) |
| 43 | self.memory.save_conversation( |
| 44 | message=user_query, message_type=MessageType.UserMessage |
| 45 | ) |
| 46 | |
| 47 | inputs[constants.CONVERSATION_HISTORY] = deepcopy( |
| 48 | self.memory.load_conversation() |
| 49 | ) |
| 50 | inputs[constants.INTERMEDIATE_STEPS] = deepcopy(intermediate_steps) |
| 51 | |
| 52 | return inputs |
| 53 | |
| 54 | def prep_output( |
| 55 | self, |
| 56 | inputs: Dict[str, str], |
| 57 | output: AgentFinish, |
| 58 | return_only_outputs: bool = False, |
| 59 | ) -> Dict[str, Any]: |
| 60 | """Save conversation into memory and prep outputs.""" |
| 61 | output_dict = output.format_output() |
| 62 | if self.memory is not None: |
| 63 | self.memory.save_conversation( |
| 64 | message=output.message, message_type=MessageType.AIMessage |
| 65 | ) |
| 66 | self.memory.save_memory( |
| 67 | key=constants.INTERMEDIATE_STEPS, value=output.intermediate_steps |
| 68 | ) |
| 69 | |
| 70 | if return_only_outputs: |
| 71 | return output_dict |
| 72 | else: |
| 73 | return {**inputs, **output_dict} |
| 74 | |
| 75 | def run( |
| 76 | self, |
| 77 | user_query: str, |
| 78 | return_only_outputs: bool = False, |
nothing calls this directly
no outgoing calls
no test coverage detected