(self,functions,process_id,key_pos=None,**args)
| 80 | print("end_print"+"*"*50) |
| 81 | |
| 82 | def parse(self,functions,process_id,key_pos=None,**args): |
| 83 | self.time = time.time() |
| 84 | conversation_history = self.conversation_history |
| 85 | for _ in range(self.TRY_TIME): |
| 86 | if _ != 0: |
| 87 | time.sleep(15) |
| 88 | if functions != []: |
| 89 | json_data = chat_completion_request( |
| 90 | self.openai_key, conversation_history, functions=functions,process_id=process_id, key_pos=key_pos,**args |
| 91 | ) |
| 92 | else: |
| 93 | json_data = chat_completion_request( |
| 94 | self.openai_key, conversation_history,process_id=process_id,key_pos=key_pos, **args |
| 95 | ) |
| 96 | try: |
| 97 | total_tokens = json_data['usage']['total_tokens'] |
| 98 | message = json_data["choices"][0]["message"] |
| 99 | if process_id == 0: |
| 100 | print(f"[process({process_id})]total tokens: {json_data['usage']['total_tokens']}") |
| 101 | |
| 102 | if "function_call" in message.keys() and "." in message["function_call"]["name"]: |
| 103 | message["function_call"]["name"] = message["function_call"]["name"].split(".")[-1] |
| 104 | |
| 105 | return message, 0, total_tokens |
| 106 | except BaseException as e: |
| 107 | print(f"[process({process_id})]Parsing Exception: {repr(e)}. Try again.") |
| 108 | if json_data is not None: |
| 109 | print(f"[process({process_id})]OpenAI return: {json_data}") |
| 110 | |
| 111 | |
| 112 | return {"role": "assistant", "content": str(json_data)}, -1, 0 |
| 113 | |
| 114 | if __name__ == "__main__": |
| 115 | llm = ChatGPTFunction() |
no test coverage detected