Spaces:
Sleeping
Sleeping
| import yaml | |
| from pocketflow import Node | |
| from pf_agent.utils import callLLM, callWebSearch, callWikipediaSearch | |
| class Decide(Node): | |
| def setLogger(self,logger): | |
| self.logger = logger | |
| def prep(self,shared): | |
| context = shared.get("context","No previous search") | |
| question = shared["question"] | |
| return question,context | |
| def exec(self,inputs): | |
| question,context = inputs | |
| prompt = f""" | |
| ### CONTEXT | |
| You are a research assistant and I need your help to answer the following question: | |
| Question: {question} | |
| Previous research: {context} | |
| ### ACTION SPACE | |
| [1] search | |
| Description: Look up more information on the web | |
| Parameters: | |
| - query (str): keywords to run the search on the web | |
| [2] wikipedia | |
| Description: Look up for more information in the wikipedia in English | |
| Parameters: | |
| - query (str): keywords to run the search in the English wikipedia | |
| [3] answer | |
| Description: Answer the question with current knowledge | |
| Parameters: | |
| - answer (str): final answer to the question | |
| ### NEXT ACTION | |
| Decide the next action based on the context and available actions. | |
| Return your response in the following format: | |
| ```yaml | |
| thinking: | | |
| <your step-by-step reasoning process> | |
| action: search OR answer OR wikipedia | |
| reason: <why you choose this action> | |
| parameters: <yaml containing the parameter or parameters required to execute the action selected in yaml format indicating parameter name and value> | |
| ``` | |
| IMPORTANT: Make sure to: | |
| 1. Use proper indentation (4 spaces) for all multi-line fields | |
| 2. Use the | character for multi-line text fields | |
| 3. Keep single-line fields without the | character | |
| """ | |
| self.logger.debug(f"=== Calling LLM to DECIDE") | |
| self.logger.debug(f"Context: {context}") | |
| response = callLLM(prompt) | |
| yaml_str = response.replace("|","") #.split("```yaml")[1].split("```")[0].strip() #.replace("|","") # | |
| decision = yaml.safe_load(yaml_str) | |
| self.logger.debug("LLM responded") | |
| self.logger.debug(f"Thinking: {decision['thinking']}") | |
| self.logger.debug(f"Action: {decision['action']} {decision['parameters']}") | |
| return decision | |
| def post(self,shared,prep_res,exec_res): | |
| if exec_res["action"] != "answer": | |
| shared["parameters"] = exec_res["parameters"] | |
| return exec_res["action"] | |
| class Search(Node): | |
| def setLogger(self,logger): | |
| self.logger = logger | |
| def prep(self,shared): | |
| if type(shared["parameters"]) == type([]): | |
| for p in shared["parameters"]: | |
| if 'query' in p: | |
| return p['query'] | |
| else: | |
| return shared["parameters"]["query"] | |
| return "invalid query" | |
| def exec(self,search_query): | |
| self.logger.debug("=== Searching the web") | |
| self.logger.debug(f"Search query: {search_query}") | |
| results = callWebSearch(search_query) | |
| self.logger.debug(f"Results: {results}") | |
| return results | |
| def post(self,shared,prep_res,exec_res): | |
| previous = shared.get("context","") | |
| shared["context"] = f"{previous}\n\nSEARCH: {shared['parameters']}\n\nRESULTS: {exec_res}" | |
| return "decide" | |
| class Wikipedia(Node): | |
| def setLogger(self,logger): | |
| self.logger = logger | |
| def prep(self,shared): | |
| if type(shared["parameters"]) == type([]): | |
| for p in shared["parameters"]: | |
| if 'query' in p: | |
| return shared["question"],p['query'] | |
| else: | |
| return shared["question"],shared["parameters"]["query"] | |
| return shared["question"],"invalid query" | |
| def exec(self,prepared): | |
| question,search_query = prepared | |
| self.logger.debug("=== Searching wikipedia") | |
| self.logger.debug(f"Search query: {search_query}") | |
| results = callWikipediaSearch(search_query) | |
| self.logger.debug(f"Results: {results}") | |
| prompt = f'''You are my research assistant and I need your help to answer the following question: {question} | |
| From my previous research I have found a Wikipedia page that could contain the answer or, at least, useful information for further research. | |
| Take a careful look at it. Ignore any HTML markup you may find. You will find sections, paragraphs, etc. formated as plain text and tables | |
| as comma-separated values. The answer or any useful information could be anywhere. | |
| If the text contains the answer, give it to me. | |
| If it is not, extract relevant information that could me help find it elsewere. | |
| Please don't include information not directly related to the question. | |
| It is very important that in any case, you are concise, brief and stick to the question being asked. Find the text below: | |
| {results}''' | |
| response = callLLM(prompt) | |
| return response | |
| def post(self,shared,prep_res,exec_res): | |
| previous = shared.get("context","") | |
| shared["context"] = f"{previous}\n\nWIKIPEDIA RESULTS: {shared['parameters']}\n\nRESULTS: {exec_res}" | |
| return "decide" | |
| class Answer(Node): | |
| def setLogger(self,logger): | |
| self.logger = logger | |
| def prep(self,shared): | |
| return shared["question"],shared.get("context","") | |
| def exec(self,inputs): | |
| question,context = inputs | |
| prompt = f''' | |
| ### CONTEXT | |
| Based on the following information, answer the question. | |
| Question: {question} | |
| Research: {context} | |
| ### YOUR ANSWER | |
| Provide a comprehensive answer using research results | |
| ''' | |
| response = callLLM(prompt) | |
| self.logger.debug(f"### CALLING LLM\n{prompt}\n### LLM RESPONSE\n{response}\n\n##########\n\n") | |
| return response | |
| def post(self,shared,prep_res,exec_res): | |
| shared["answer"] = exec_res |