Gracefully handling parsing errors for structured outputs (#966)

* feat: gracefully handling parsing errors for structured outputs

* chore: stringify parsed response

libs/superagent/app/api/agents.py

@@ -501,10 +501,14 @@ async def send_message(
                 from langchain.output_parsers.json import SimpleJsonOutputParser
 
                 parser = SimpleJsonOutputParser()
-                parsed_schema = str(parser.parse(schema_tokens))
+                try:
+                    parsed_res = parser.parse(schema_tokens)
+                except Exception as e:
+                    logger.error(f"Error parsing output: {e}")
+                    parsed_res = {}
 
                 # stream line by line to prevent streaming large data in one go
-                for line in parsed_schema.split("\n"):
+                for line in json.dumps(parsed_res).split("\n"):
                     async for val in stream_dict_keys(
                         {"event": "message", "data": line}
                     ):
@@ -609,8 +613,14 @@ async def send_message(
     if output_schema:
         from langchain.output_parsers.json import SimpleJsonOutputParser
 
-        json_parser = SimpleJsonOutputParser()
-        output["output"] = json_parser.parse(text=output["output"])
+        parser = SimpleJsonOutputParser()
+        try:
+            output["output"] = parser.parse(text=output["output"])
+        except Exception as e:
+            logger.error(f"Error parsing output: {e}")
+            output["output"] = {}
+
+        output = json.dumps(output)
 
     return {"success": True, "data": output}
 

libs/superagent/app/api/workflows.py

@@ -304,10 +304,15 @@ async def send_message() -> AsyncIterable[str]:
                         from langchain.output_parsers.json import SimpleJsonOutputParser
 
                         parser = SimpleJsonOutputParser()
-                        parsed_schema = str(parser.parse(schema_tokens))
+                        try:
+                            parsed_res = parser.parse(schema_tokens)
+                        except Exception as e:
+                            # TODO: stream schema parsing error as well
+                            logger.error(f"Error in parsing schema: {e}")
+                            parsed_res = {}
 
                         # stream line by line to prevent streaming large data in one go
-                        for line in parsed_schema.split("\n"):
+                        for line in json.dumps(parsed_res).split("\n"):
                             agent_name = workflow_step["agent_name"]
                             async for val in stream_dict_keys(
                                 {

libs/superagent/app/workflows/base.py

@@ -1,3 +1,5 @@
+import json
+import logging
 from typing import Any, List
 
 from agentops.langchain_callback_handler import (
@@ -9,6 +11,8 @@
 from app.agents.base import AgentBase
 from app.utils.callbacks import CustomAsyncIteratorCallbackHandler
 
+logger = logging.getLogger(__name__)
+
 
 class WorkflowBase:
     def __init__(
@@ -59,10 +63,15 @@ async def arun(self, input: Any):
             )
             if output_schema:
                 # TODO: throw error if output is not valid
-                json_parser = SimpleJsonOutputParser()
-                agent_response["output"] = json_parser.parse(
-                    text=agent_response["output"]
-                )
+                parser = SimpleJsonOutputParser()
+                try:
+                    agent_response["output"] = parser.parse(
+                        text=agent_response["output"]
+                    )
+                except Exception as e:
+                    logger.error(f"Error parsing output: {e}")
+                    agent_response["output"] = {}
+                agent_response["output"] = json.dumps(agent_response["output"])
 
             previous_output = agent_response.get("output")
             steps_output.append(agent_response)