Default chain with take_next_step implemented It handles a few common error cases with agent, such as taking repeated action with same inputs and whether agent should continue the conversation
| 12 | |
| 13 | |
| 14 | class Chain(BaseChain): |
| 15 | """ |
| 16 | Default chain with take_next_step implemented |
| 17 | It handles a few common error cases with agent, such as taking repeated action with same |
| 18 | inputs and whether agent should continue the conversation |
| 19 | """ |
| 20 | |
| 21 | return_intermediate_steps: bool = False |
| 22 | handle_parsing_errors = True |
| 23 | graceful_exit_tool: Tool = HandOffToAgent() |
| 24 | |
| 25 | def handle_repeated_action(self, agent_action: AgentAction) -> AgentFinish: |
| 26 | print( |
| 27 | f"Action taken before: {agent_action.tool}, " |
| 28 | f"input: {agent_action.tool_input}" |
| 29 | ) |
| 30 | if agent_action.model_response: |
| 31 | return AgentFinish( |
| 32 | message=agent_action.response, |
| 33 | log=f"Action taken before: {agent_action.tool}, " |
| 34 | f"input: {agent_action.tool_input}", |
| 35 | ) |
| 36 | else: |
| 37 | print("No response from agent. Gracefully exit due to repeated action") |
| 38 | return AgentFinish( |
| 39 | message=self.graceful_exit_tool.run(), |
| 40 | log="Gracefully exit due to repeated action", |
| 41 | ) |
| 42 | |
| 43 | def take_next_step( |
| 44 | self, |
| 45 | name_to_tool_map: Dict[str, Tool], |
| 46 | inputs: Dict[str, str], |
| 47 | ) -> (AgentFinish, AgentAction): |
| 48 | """ |
| 49 | How agent determines the next step after observing the inputs and intermediate steps |
| 50 | Args: |
| 51 | name_to_tool_map: map of tool name to the actual tool object |
| 52 | inputs: a dictionary of all inputs, such as user query, past conversation and |
| 53 | tools outputs |
| 54 | |
| 55 | Returns: |
| 56 | Either AgentFinish to respond to user or AgentAction to take the next action |
| 57 | """ |
| 58 | |
| 59 | try: |
| 60 | # Call the LLM to see what to do. |
| 61 | output = self.agent.plan( |
| 62 | **inputs, |
| 63 | ) |
| 64 | except Exception as e: |
| 65 | if not self.handle_parsing_errors: |
| 66 | raise e |
| 67 | tool_output = f"Invalid or incomplete response due to {e}" |
| 68 | print(tool_output) |
| 69 | output = AgentFinish(message=self.graceful_exit_tool.run(), log=tool_output) |
| 70 | return output |
| 71 |