| 51 | |
| 52 | |
| 53 | class AgentOutputParser(BaseModel): |
| 54 | @staticmethod |
| 55 | def load_json_output(message: BaseMessage) -> Dict[str, Any]: |
| 56 | """If the message contains a json response, try to parse it into dictionary""" |
| 57 | text = message.content |
| 58 | clean_text = "" |
| 59 | |
| 60 | try: |
| 61 | clean_text = text[text.index("{") : text.rindex("}") + 1].strip() |
| 62 | response = json.loads(clean_text) |
| 63 | except Exception: |
| 64 | llm = ChatOpenAI(temperature=0) |
| 65 | message = [ |
| 66 | UserMessage( |
| 67 | content=f"""Fix the following json into correct format |
| 68 | ```json |
| 69 | {clean_text} |
| 70 | ``` |
| 71 | """ |
| 72 | ) |
| 73 | ] |
| 74 | full_output: Generation = llm.generate(message).generations[0] |
| 75 | response = json.loads(full_output.message.content) |
| 76 | |
| 77 | return response |
| 78 | |
| 79 | @abstractmethod |
| 80 | def parse(self, message: BaseMessage) -> Union[AgentAction, AgentFinish]: |
| 81 | """Parse text into agent action/finish.""" |
| 82 | |
| 83 | def parse_clarification( |
| 84 | self, message: BaseMessage, agent_action: AgentAction |
| 85 | ) -> Union[AgentAction, AgentFinish]: |
| 86 | """Parse clarification outputs""" |
| 87 | return agent_action |
| 88 | |
| 89 | def parse_estimated_confidence(self, message: BaseMessage) -> int: |
| 90 | """Parse estimated confidence from the message""" |
| 91 | return 1 |
nothing calls this directly
no outgoing calls
no test coverage detected