"""Heavily borrowed from langchain: https://github.com/langchain-ai/langchain/"""importrefromtypingimportAnyfromtypingimportListfromtypingimportUnionfromplanners.actionimportActionfromplanners.actionimportPlanFinishfromplanners.plannerimportBasePlanner
[docs]classReActPlanner(BasePlanner):""" **Description:** This class implements ReAct planner, which inherits from the BasePlanner base class. ReAct employs reasoning and action techniques to ascertain the essential actions to be undertaken. `Paper <https://arxiv.org/abs/2210.03629>`_ This code defines a base class called "BasePlanner" that inherits from the "BaseModel" class of the pydantic library. The BasePlanner class serves as a base for implementing specific planners. """classConfig:"""Configuration for this pydantic object."""arbitrary_types_allowed=True@propertydef_planner_type(self):return"zero-shot-react-planner"@propertydef_planner_model(self):returnself.llm_model@propertydef_stop(self)->List[str]:return["Observation"]@propertydef_planner_prompt(self):return"""You are very helpful empathetic health assistant and your goal is to help the user to get accurate information \about his/her health and well-being. Answer the following questions as best you can. Make sure you call all the needed tools before \reach to the Final Answer.Here are list of rules that you should follow:1- Avoid calling the same tool with the same inputs from PreviousActions.2- If you want to use the result of another tool, mention it in the execute tool action.3- Minimize the number of tool executions.4- You should try your best to pass datapipe data to other tools and avoid analysing data by yourself. Worst case read raw data from datapipe and \use it to answer the user's query.5- when your Thought is 'I now know the final answer' you should provide 'Final Answer:'6- When data in the datapipe requires analysis without numerical calculations. Please prioritize using tools for any \data-related tasks. Provide guidance on how to use the available tools effectively.7- you should be fully aware of what you are doing and based on the history and previous tools used, you should decide \what inputs should be provided to other tools. for example if you already fetched a user data in the next tools you should \provide data based on this knowledge.Use the following format. You should stick to the following format:MetaData: this contains the name of data files of different types like image, audio, video, and text. You can pass these files to tools when needed.History: the history of previous chats happened. You should use them to answer user's current question. If the answer is already in the history, \just return it.Question: the input question you must answerThought: you should always think about what to do. Describe what you want to do and then select actions.Action: the action to take, SHOULD be only the tool name selected from one of [{tool_names}]Action Inputs: the inputs should be seperated by $. Action inputs should be based on the input descriptions of the tool. \The examples for a two input tools are: input1$input2 or if datapipe is needed datapipe:key$input2Observation: the result of the action... (this Thought/Action/Action Inputs/Observation can repeat N times)Thought: Your final reasoning or 'I now know the final answer'. when you think you are done you should provide the 'Final Answer'.Final Answer: the final answer to the original input question. It should be based on the tools result.Begin!MetaData: {meta}History: {history}Question: {input}Thought: {agent_scratchpad}"""
[docs]defplan(self,query:str,history:str="",meta:str="",previous_actions:List[Action]=None,use_history:bool=False,**kwargs:Any,)->List[Union[Action,PlanFinish]]:""" Generate a plan using ReAct Args: query (str): Input query. history (str): History information. meta (str): meta information. previous_actions (List[Action]): List of previous actions. use_history (bool): Flag indicating whether to use history. **kwargs (Any): Additional keyword arguments. Return: Action: return action. """ifprevious_actionsisNone:previous_actions=[]agent_scratchpad=""iflen(previous_actions)>0:agent_scratchpad="\n".join([f"Action: {action.task}\nAction Inputs: {action.task_input}\nObservation: {action.task_response}\nThought:"foractioninprevious_actions])prompt=(self._planner_prompt.replace("{input}",query).replace("{meta}",", ".join(meta)).replace("{history}",historyifuse_historyelse"").replace("{agent_scratchpad}",agent_scratchpad).replace("{tool_names}",self.get_available_tasks()))print("prompt",prompt)# if len(previous_actions) > 0:# prompt += "\nThought:"kwargs["max_tokens"]=500kwargs["stop"]=self._stopresponse=self._planner_model.generate(query=prompt,**kwargs)index=min([response.find(text)fortextinself._stop])index1=response.find("\nAction:")ifindex1==-1:index1=0print("resp",response)response=response[index1:index]actions=self.parse(response)returnactions
[docs]defparse(self,query:str,**kwargs:Any,)->List[Union[Action,PlanFinish]]:""" Parse the output query into a list of actions or a final answer. It parses the output based on \ the following format: Thought: though\n Action: action\n Action Inputs: inputs or Thought: though\n Final Answer: final answer\n Args:\n query (str): The planner output query to extract actions. **kwargs (Any): Additional keyword arguments. Return: List[Union[Action, PlanFinish]]: List of parsed actions or a finishing signal. Raise: ValueError: If parsing encounters an invalid format or unexpected content. """FINAL_ANSWER_ACTION="Final Answer:"includes_answer=FINAL_ANSWER_ACTIONinquerystr_pattern=(r"(?:"+"|".join(self.get_available_tasks_list())+r")(?=.*Action\s*\d*\s*Inputs)")regex=(r"\s*Action\s*\d*\s*:[\s]*.*?("+str_pattern+r").*?[\s]*Action\s*\d*\s*Inputs\s*\d*\s*:[\s]*(.*)")action_match=re.search(regex,query,re.DOTALL)ifaction_matchandincludes_answer:ifquery.find(FINAL_ANSWER_ACTION)<query.find(action_match.group(0)):# if final answer is before the hallucination, return final answerstart_index=query.find(FINAL_ANSWER_ACTION)+len(FINAL_ANSWER_ACTION)end_index=query.find("\n\n",start_index)return[PlanFinish(query[start_index:end_index].strip())]else:raiseValueError("Parsing the output produced both a final answer and a parse-able action.")ifaction_match:action=action_match.group(1).strip()action_input=action_match.group(2)tool_input=action_input.strip(" ")# ensure if its a well formed SQL query we don't remove any trailing " charsiftool_input.startswith("SELECT ")isFalse:tool_input=tool_input.strip('"')return[Action(action,tool_input,"",query)]elifincludes_answer:return[PlanFinish(query.split(FINAL_ANSWER_ACTION)[-1].strip(),query,)]ifnotre.search(r"Action\s*\d*\s*:[\s]*.*?("+str_pattern+r").*?",query,re.DOTALL,):raiseValueError("Invalid Format: Missing 'Action:' or 'Final Answer' after 'Thought:'\n"# f"Or The tool name is wrong. The tool name should be one of: `{self.get_available_tasks_list()}`")elifnotre.search(r"[\s]*Action\s*\d*\s*Inputs\s*\d*\s*:[\s]*(.*)",query,re.DOTALL,):raiseValueError("Invalid Format: Missing 'Action Input:' after 'Action:'")else:raiseValueError("Wrong format.")