We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Streaming requires Generator where Iterator should suffice
**Describe the bug**
For the user code to provide streaming content, instead of the more generic Iterator protocol, Promptflow relies on the more specific `GeneratorType`. That is requiring the user to unnecessarily wrap their iterators in generators to stream content in more generic scenarios (such as streaming langchain results). Is there a reason PF depends on GeneratorType instead of the iterator protocol? Also, see this note in the [python source](https://github.com/python/cpython/blob/f65497fd252a4a4df960da04d68e8316b58624c0/Lib/types.py#L6-L10):
```
# Iterators in Python aren't a matter of type but of protocol.  A large
# and changing number of builtin types implement *some* flavor of
# iterator.  Don't check the type!  Use hasattr to check for both
# "__iter__" and "__next__" attributes instead.
```

**Concrete Issue:** 
When returning an iterator from my tool to enable streaming, I get the following error when running `pf flow test`
```
Flow test failed with UserErrorException: Exception: The output 'answer' for flow is incorrect. The output value is 
not JSON serializable. JSON dump failed: (TypeError) Object of type WordIterator is not JSON serializable. Please 
verify your flow output and make sure the value serializable.
```

**How To Reproduce the bug**
Here is my flow.dag.yaml:
```yaml
id: template_chat_flow
name: Template Chat Flow
inputs:
  chat_history:
    type: list
    default: []
  question:
    type: string
    default: What is the meaning of life?
outputs:
  answer:
    type: string
    is_chat_output: true
    reference: ${stream.output}
nodes:
- name: stream
  type: python
  source:
    type: code
    path: stream.py
  inputs:
    input: ${inputs.question}
```

And here is my stream.py
```python
from promptflow import tool

class WordIterator:
    def __init__(self, input: str):
        self.input = input

    def __iter__(self):
        return self

    def __next__(self):
        if self.input:
            word, *rest = self.input.split(" ")
            self.input = " ".join(rest)
            return f"{word} "
        else:
            raise StopIteration  

@tool
def my_python_tool(input: str):
    iterator = WordIterator(input)
    assert hasattr(iterator, "__iter__")
    assert hasattr(iterator, "__next__")
    return iterator
```
With the above PF run `pf flow test --flow .` -- I get this error:
```
2023-09-19 11:01:17 +0200   42558 execution          INFO     Start to run 1 nodes with concurrency level 16.
2023-09-19 11:01:17 +0200   42558 execution.flow     INFO     Executing node stream. node run id: c4ddaddd-6a38-44fd-9ab1-3258fb88bb37_stream_0
2023-09-19 11:01:17 +0200   42558 execution.flow     WARNING  Output of stream is not json serializable, use str to store it.
2023-09-19 11:01:17 +0200   42558 execution.flow     INFO     Node stream completes.
Flow test failed with UserErrorException: Exception: The output 'answer' for flow is incorrect. The output value is not JSON serializable. JSON dump failed: (TypeError) Object of type WordIterator is not JSON serializable. Please verify your flow output and make sure the value serializable.
```

**Expected behavior**
It should read out the iterator and stream the chunks to the caller -- `pf flow test --flow .` should look like this:
```
2023-09-19 10:34:05 +0200   40375 execution          INFO     Start to run 1 nodes with concurrency level 16.
2023-09-19 10:34:05 +0200   40375 execution.flow     INFO     Executing node stream. node run id: 24e60c4d-606a-4fc5-8e4c-cc4a5c41d6c8_stream_0
2023-09-19 10:34:05 +0200   40375 execution.flow     WARNING  Output of stream is not json serializable, use str to store it.
2023-09-19 10:34:05 +0200   40375 execution.flow     INFO     Node stream completes.
{
    "answer": "What is the meaning of life? "
}
```

**Running Information(please complete the following information):**
 - Promptflow Package Version using `pf -v`: 0.1.0b6
 - Operating System: 
```
ProductName:            macOS
ProductVersion:         13.5.2
BuildVersion:           22G91
```
 - Python Version using `python --version`: Python 3.10.12

**Additional context**
If I wrap the iterator in a generator, everything works as expected:

```python
from promptflow import tool

class WordIterator:
    def __init__(self, input: str):
        self.input = input

    def __iter__(self):
        return self

    def __next__(self):
        if self.input:
            word, *rest = self.input.split(" ")
            self.input = " ".join(rest)
            return f"{word} "
        else:
            raise StopIteration  
    
    def to_generator(self):
        try:
            while True:
                yield next(self)
        except StopIteration:
            pass

@tool
def my_python_tool(input: str):
    iterator = WordIterator(input).to_generator()
    assert hasattr(iterator, "__iter__")
    assert hasattr(iterator, "__next__")
    return iterator
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/promptflow/promptflow/_core/tracer.py`
Content:
```
1 # ---------------------------------------------------------
2 # Copyright (c) Microsoft Corporation. All rights reserved.
3 # ---------------------------------------------------------
4 
5 import inspect
6 import json
7 import logging
8 from contextvars import ContextVar
9 from datetime import datetime
10 from types import GeneratorType
11 from typing import Optional
12 
13 from promptflow._core.generator_proxy import GeneratorProxy, generate_from_proxy
14 from promptflow._utils.dataclass_serializer import serialize
15 from promptflow.contracts.tool import ConnectionType
16 from promptflow.contracts.trace import Trace, TraceType
17 
18 from .thread_local_singleton import ThreadLocalSingleton
19 
20 
21 class Tracer(ThreadLocalSingleton):
22     CONTEXT_VAR_NAME = "Tracer"
23     context_var = ContextVar(CONTEXT_VAR_NAME, default=None)
24 
25     def __init__(self, run_id):
26         self._run_id = run_id
27         self._traces = []
28         self._trace_stack = []
29 
30     @classmethod
31     def start_tracing(cls, run_id):
32         tracer = cls(run_id)
33         tracer._activate_in_context()
34 
35     @classmethod
36     def end_tracing(cls, raise_ex=False):
37         tracer = cls.active_instance()
38         if not tracer:
39             msg = "Try end tracing but no active tracer in current context."
40             if raise_ex:
41                 raise Exception(msg)
42             logging.warning(msg)
43             return []
44         tracer._deactivate_in_context()
45         return tracer.to_json()
46 
47     @classmethod
48     def push_tool(cls, f, args, kwargs):
49         obj = cls.active_instance()
50         sig = inspect.signature(f).parameters
51         all_kwargs = {**{k: v for k, v in zip(sig.keys(), args)}, **kwargs}
52         all_kwargs = {
53             k: ConnectionType.serialize_conn(v) if ConnectionType.is_connection_value(v) else v
54             for k, v in all_kwargs.items()
55         }
56         # TODO: put parameters in self to inputs for builtin tools
57         all_kwargs.pop("self", None)
58         trace = Trace(
59             name=f.__qualname__,
60             type=TraceType.TOOL,
61             start_time=datetime.utcnow().timestamp(),
62             inputs=all_kwargs,
63         )
64         obj._push(trace)
65         return trace
66 
67     @classmethod
68     def push(cls, trace: Trace):
69         obj = cls.active_instance()
70         if not obj:
71             logging.warning("Try to push trace but no active tracer in current context.")
72             return
73         obj._push(trace)
74 
75     @staticmethod
76     def to_serializable(obj):
77         if isinstance(obj, dict) and all(isinstance(k, str) for k in obj.keys()):
78             return {k: Tracer.to_serializable(v) for k, v in obj.items()}
79         if isinstance(obj, GeneratorProxy):
80             return obj
81         try:
82             obj = serialize(obj)
83             json.dumps(obj)
84         except Exception:
85             # We don't want to fail the whole function call because of a serialization error,
86             # so we simply convert it to str if it cannot be serialized.
87             obj = str(obj)
88         return obj
89 
90     def _push(self, trace: Trace):
91         if trace.inputs:
92             trace.inputs = self.to_serializable(trace.inputs)
93         if not trace.start_time:
94             trace.start_time = datetime.utcnow().timestamp()
95         if not self._trace_stack:
96             self._traces.append(trace)
97         else:
98             self._trace_stack[-1].children = self._trace_stack[-1].children or []
99             self._trace_stack[-1].children.append(trace)
100         self._trace_stack.append(trace)
101 
102     @classmethod
103     def pop(cls, output=None, error: Optional[Exception] = None):
104         obj = cls.active_instance()
105         return obj._pop(output, error)
106 
107     def _pop(self, output=None, error: Optional[Exception] = None):
108         last_trace = self._trace_stack[-1]
109         if isinstance(output, GeneratorType):
110             output = GeneratorProxy(output)
111         if output is not None:
112             last_trace.output = self.to_serializable(output)
113         if error is not None:
114             last_trace.error = self._format_error(error)
115         self._trace_stack[-1].end_time = datetime.utcnow().timestamp()
116         self._trace_stack.pop()
117 
118         if isinstance(output, GeneratorProxy):
119             return generate_from_proxy(output)
120         else:
121             return output
122 
123     def to_json(self) -> list:
124         return serialize(self._traces)
125 
126     @staticmethod
127     def _format_error(error: Exception) -> dict:
128         return {
129             "message": str(error),
130             "type": type(error).__qualname__,
131         }
132
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/promptflow/promptflow/_core/tracer.py b/src/promptflow/promptflow/_core/tracer.py
--- a/src/promptflow/promptflow/_core/tracer.py
+++ b/src/promptflow/promptflow/_core/tracer.py
@@ -5,9 +5,9 @@
 import inspect
 import json
 import logging
+from collections.abc import Iterator
 from contextvars import ContextVar
 from datetime import datetime
-from types import GeneratorType
 from typing import Optional
 
 from promptflow._core.generator_proxy import GeneratorProxy, generate_from_proxy
@@ -106,7 +106,7 @@
 
     def _pop(self, output=None, error: Optional[Exception] = None):
         last_trace = self._trace_stack[-1]
-        if isinstance(output, GeneratorType):
+        if isinstance(output, Iterator):
             output = GeneratorProxy(output)
         if output is not None:
             last_trace.output = self.to_serializable(output)

{"golden_diff": "diff --git a/src/promptflow/promptflow/_core/tracer.py b/src/promptflow/promptflow/_core/tracer.py\n--- a/src/promptflow/promptflow/_core/tracer.py\n+++ b/src/promptflow/promptflow/_core/tracer.py\n@@ -5,9 +5,9 @@\n import inspect\n import json\n import logging\n+from collections.abc import Iterator\n from contextvars import ContextVar\n from datetime import datetime\n-from types import GeneratorType\n from typing import Optional\n \n from promptflow._core.generator_proxy import GeneratorProxy, generate_from_proxy\n@@ -106,7 +106,7 @@\n \n     def _pop(self, output=None, error: Optional[Exception] = None):\n         last_trace = self._trace_stack[-1]\n-        if isinstance(output, GeneratorType):\n+        if isinstance(output, Iterator):\n             output = GeneratorProxy(output)\n         if output is not None:\n             last_trace.output = self.to_serializable(output)\n", "issue": "[BUG] Streaming requires Generator where Iterator should suffice\n**Describe the bug**\r\nFor the user code to provide streaming content, instead of the more generic Iterator protocol, Promptflow relies on the more specific `GeneratorType`. That is requiring the user to unnecessarily wrap their iterators in generators to stream content in more generic scenarios (such as streaming langchain results). Is there a reason PF depends on GeneratorType instead of the iterator protocol? Also, see this note in the [python source](https://github.com/python/cpython/blob/f65497fd252a4a4df960da04d68e8316b58624c0/Lib/types.py#L6-L10):\r\n```\r\n# Iterators in Python aren't a matter of type but of protocol.  A large\r\n# and changing number of builtin types implement *some* flavor of\r\n# iterator.  Don't check the type!  Use hasattr to check for both\r\n# \"__iter__\" and \"__next__\" attributes instead.\r\n```\r\n\r\n**Concrete Issue:** \r\nWhen returning an iterator from my tool to enable streaming, I get the following error when running `pf flow test`\r\n```\r\nFlow test failed with UserErrorException: Exception: The output 'answer' for flow is incorrect. The output value is \r\nnot JSON serializable. JSON dump failed: (TypeError) Object of type WordIterator is not JSON serializable. Please \r\nverify your flow output and make sure the value serializable.\r\n```\r\n\r\n**How To Reproduce the bug**\r\nHere is my flow.dag.yaml:\r\n```yaml\r\nid: template_chat_flow\r\nname: Template Chat Flow\r\ninputs:\r\n  chat_history:\r\n    type: list\r\n    default: []\r\n  question:\r\n    type: string\r\n    default: What is the meaning of life?\r\noutputs:\r\n  answer:\r\n    type: string\r\n    is_chat_output: true\r\n    reference: ${stream.output}\r\nnodes:\r\n- name: stream\r\n  type: python\r\n  source:\r\n    type: code\r\n    path: stream.py\r\n  inputs:\r\n    input: ${inputs.question}\r\n```\r\n\r\nAnd here is my stream.py\r\n```python\r\nfrom promptflow import tool\r\n\r\nclass WordIterator:\r\n    def __init__(self, input: str):\r\n        self.input = input\r\n\r\n    def __iter__(self):\r\n        return self\r\n\r\n    def __next__(self):\r\n        if self.input:\r\n            word, *rest = self.input.split(\" \")\r\n            self.input = \" \".join(rest)\r\n            return f\"{word} \"\r\n        else:\r\n            raise StopIteration  \r\n\r\n@tool\r\ndef my_python_tool(input: str):\r\n    iterator = WordIterator(input)\r\n    assert hasattr(iterator, \"__iter__\")\r\n    assert hasattr(iterator, \"__next__\")\r\n    return iterator\r\n```\r\nWith the above PF run `pf flow test --flow .` -- I get this error:\r\n```\r\n2023-09-19 11:01:17 +0200   42558 execution          INFO     Start to run 1 nodes with concurrency level 16.\r\n2023-09-19 11:01:17 +0200   42558 execution.flow     INFO     Executing node stream. node run id: c4ddaddd-6a38-44fd-9ab1-3258fb88bb37_stream_0\r\n2023-09-19 11:01:17 +0200   42558 execution.flow     WARNING  Output of stream is not json serializable, use str to store it.\r\n2023-09-19 11:01:17 +0200   42558 execution.flow     INFO     Node stream completes.\r\nFlow test failed with UserErrorException: Exception: The output 'answer' for flow is incorrect. The output value is not JSON serializable. JSON dump failed: (TypeError) Object of type WordIterator is not JSON serializable. Please verify your flow output and make sure the value serializable.\r\n```\r\n\r\n**Expected behavior**\r\nIt should read out the iterator and stream the chunks to the caller -- `pf flow test --flow .` should look like this:\r\n```\r\n2023-09-19 10:34:05 +0200   40375 execution          INFO     Start to run 1 nodes with concurrency level 16.\r\n2023-09-19 10:34:05 +0200   40375 execution.flow     INFO     Executing node stream. node run id: 24e60c4d-606a-4fc5-8e4c-cc4a5c41d6c8_stream_0\r\n2023-09-19 10:34:05 +0200   40375 execution.flow     WARNING  Output of stream is not json serializable, use str to store it.\r\n2023-09-19 10:34:05 +0200   40375 execution.flow     INFO     Node stream completes.\r\n{\r\n    \"answer\": \"What is the meaning of life? \"\r\n}\r\n```\r\n\r\n**Running Information(please complete the following information):**\r\n - Promptflow Package Version using `pf -v`: 0.1.0b6\r\n - Operating System: \r\n```\r\nProductName:            macOS\r\nProductVersion:         13.5.2\r\nBuildVersion:           22G91\r\n```\r\n - Python Version using `python --version`: Python 3.10.12\r\n\r\n**Additional context**\r\nIf I wrap the iterator in a generator, everything works as expected:\r\n\r\n```python\r\nfrom promptflow import tool\r\n\r\nclass WordIterator:\r\n    def __init__(self, input: str):\r\n        self.input = input\r\n\r\n    def __iter__(self):\r\n        return self\r\n\r\n    def __next__(self):\r\n        if self.input:\r\n            word, *rest = self.input.split(\" \")\r\n            self.input = \" \".join(rest)\r\n            return f\"{word} \"\r\n        else:\r\n            raise StopIteration  \r\n    \r\n    def to_generator(self):\r\n        try:\r\n            while True:\r\n                yield next(self)\r\n        except StopIteration:\r\n            pass\r\n\r\n@tool\r\ndef my_python_tool(input: str):\r\n    iterator = WordIterator(input).to_generator()\r\n    assert hasattr(iterator, \"__iter__\")\r\n    assert hasattr(iterator, \"__next__\")\r\n    return iterator\r\n```\r\n\n", "before_files": [{"content": "# ---------------------------------------------------------\n# Copyright (c) Microsoft Corporation. All rights reserved.\n# ---------------------------------------------------------\n\nimport inspect\nimport json\nimport logging\nfrom contextvars import ContextVar\nfrom datetime import datetime\nfrom types import GeneratorType\nfrom typing import Optional\n\nfrom promptflow._core.generator_proxy import GeneratorProxy, generate_from_proxy\nfrom promptflow._utils.dataclass_serializer import serialize\nfrom promptflow.contracts.tool import ConnectionType\nfrom promptflow.contracts.trace import Trace, TraceType\n\nfrom .thread_local_singleton import ThreadLocalSingleton\n\n\nclass Tracer(ThreadLocalSingleton):\n    CONTEXT_VAR_NAME = \"Tracer\"\n    context_var = ContextVar(CONTEXT_VAR_NAME, default=None)\n\n    def __init__(self, run_id):\n        self._run_id = run_id\n        self._traces = []\n        self._trace_stack = []\n\n    @classmethod\n    def start_tracing(cls, run_id):\n        tracer = cls(run_id)\n        tracer._activate_in_context()\n\n    @classmethod\n    def end_tracing(cls, raise_ex=False):\n        tracer = cls.active_instance()\n        if not tracer:\n            msg = \"Try end tracing but no active tracer in current context.\"\n            if raise_ex:\n                raise Exception(msg)\n            logging.warning(msg)\n            return []\n        tracer._deactivate_in_context()\n        return tracer.to_json()\n\n    @classmethod\n    def push_tool(cls, f, args, kwargs):\n        obj = cls.active_instance()\n        sig = inspect.signature(f).parameters\n        all_kwargs = {**{k: v for k, v in zip(sig.keys(), args)}, **kwargs}\n        all_kwargs = {\n            k: ConnectionType.serialize_conn(v) if ConnectionType.is_connection_value(v) else v\n            for k, v in all_kwargs.items()\n        }\n        # TODO: put parameters in self to inputs for builtin tools\n        all_kwargs.pop(\"self\", None)\n        trace = Trace(\n            name=f.__qualname__,\n            type=TraceType.TOOL,\n            start_time=datetime.utcnow().timestamp(),\n            inputs=all_kwargs,\n        )\n        obj._push(trace)\n        return trace\n\n    @classmethod\n    def push(cls, trace: Trace):\n        obj = cls.active_instance()\n        if not obj:\n            logging.warning(\"Try to push trace but no active tracer in current context.\")\n            return\n        obj._push(trace)\n\n    @staticmethod\n    def to_serializable(obj):\n        if isinstance(obj, dict) and all(isinstance(k, str) for k in obj.keys()):\n            return {k: Tracer.to_serializable(v) for k, v in obj.items()}\n        if isinstance(obj, GeneratorProxy):\n            return obj\n        try:\n            obj = serialize(obj)\n            json.dumps(obj)\n        except Exception:\n            # We don't want to fail the whole function call because of a serialization error,\n            # so we simply convert it to str if it cannot be serialized.\n            obj = str(obj)\n        return obj\n\n    def _push(self, trace: Trace):\n        if trace.inputs:\n            trace.inputs = self.to_serializable(trace.inputs)\n        if not trace.start_time:\n            trace.start_time = datetime.utcnow().timestamp()\n        if not self._trace_stack:\n            self._traces.append(trace)\n        else:\n            self._trace_stack[-1].children = self._trace_stack[-1].children or []\n            self._trace_stack[-1].children.append(trace)\n        self._trace_stack.append(trace)\n\n    @classmethod\n    def pop(cls, output=None, error: Optional[Exception] = None):\n        obj = cls.active_instance()\n        return obj._pop(output, error)\n\n    def _pop(self, output=None, error: Optional[Exception] = None):\n        last_trace = self._trace_stack[-1]\n        if isinstance(output, GeneratorType):\n            output = GeneratorProxy(output)\n        if output is not None:\n            last_trace.output = self.to_serializable(output)\n        if error is not None:\n            last_trace.error = self._format_error(error)\n        self._trace_stack[-1].end_time = datetime.utcnow().timestamp()\n        self._trace_stack.pop()\n\n        if isinstance(output, GeneratorProxy):\n            return generate_from_proxy(output)\n        else:\n            return output\n\n    def to_json(self) -> list:\n        return serialize(self._traces)\n\n    @staticmethod\n    def _format_error(error: Exception) -> dict:\n        return {\n            \"message\": str(error),\n            \"type\": type(error).__qualname__,\n        }\n", "path": "src/promptflow/promptflow/_core/tracer.py"}], "after_files": [{"content": "# ---------------------------------------------------------\n# Copyright (c) Microsoft Corporation. All rights reserved.\n# ---------------------------------------------------------\n\nimport inspect\nimport json\nimport logging\nfrom collections.abc import Iterator\nfrom contextvars import ContextVar\nfrom datetime import datetime\nfrom typing import Optional\n\nfrom promptflow._core.generator_proxy import GeneratorProxy, generate_from_proxy\nfrom promptflow._utils.dataclass_serializer import serialize\nfrom promptflow.contracts.tool import ConnectionType\nfrom promptflow.contracts.trace import Trace, TraceType\n\nfrom .thread_local_singleton import ThreadLocalSingleton\n\n\nclass Tracer(ThreadLocalSingleton):\n    CONTEXT_VAR_NAME = \"Tracer\"\n    context_var = ContextVar(CONTEXT_VAR_NAME, default=None)\n\n    def __init__(self, run_id):\n        self._run_id = run_id\n        self._traces = []\n        self._trace_stack = []\n\n    @classmethod\n    def start_tracing(cls, run_id):\n        tracer = cls(run_id)\n        tracer._activate_in_context()\n\n    @classmethod\n    def end_tracing(cls, raise_ex=False):\n        tracer = cls.active_instance()\n        if not tracer:\n            msg = \"Try end tracing but no active tracer in current context.\"\n            if raise_ex:\n                raise Exception(msg)\n            logging.warning(msg)\n            return []\n        tracer._deactivate_in_context()\n        return tracer.to_json()\n\n    @classmethod\n    def push_tool(cls, f, args, kwargs):\n        obj = cls.active_instance()\n        sig = inspect.signature(f).parameters\n        all_kwargs = {**{k: v for k, v in zip(sig.keys(), args)}, **kwargs}\n        all_kwargs = {\n            k: ConnectionType.serialize_conn(v) if ConnectionType.is_connection_value(v) else v\n            for k, v in all_kwargs.items()\n        }\n        # TODO: put parameters in self to inputs for builtin tools\n        all_kwargs.pop(\"self\", None)\n        trace = Trace(\n            name=f.__qualname__,\n            type=TraceType.TOOL,\n            start_time=datetime.utcnow().timestamp(),\n            inputs=all_kwargs,\n        )\n        obj._push(trace)\n        return trace\n\n    @classmethod\n    def push(cls, trace: Trace):\n        obj = cls.active_instance()\n        if not obj:\n            logging.warning(\"Try to push trace but no active tracer in current context.\")\n            return\n        obj._push(trace)\n\n    @staticmethod\n    def to_serializable(obj):\n        if isinstance(obj, dict) and all(isinstance(k, str) for k in obj.keys()):\n            return {k: Tracer.to_serializable(v) for k, v in obj.items()}\n        if isinstance(obj, GeneratorProxy):\n            return obj\n        try:\n            obj = serialize(obj)\n            json.dumps(obj)\n        except Exception:\n            # We don't want to fail the whole function call because of a serialization error,\n            # so we simply convert it to str if it cannot be serialized.\n            obj = str(obj)\n        return obj\n\n    def _push(self, trace: Trace):\n        if trace.inputs:\n            trace.inputs = self.to_serializable(trace.inputs)\n        if not trace.start_time:\n            trace.start_time = datetime.utcnow().timestamp()\n        if not self._trace_stack:\n            self._traces.append(trace)\n        else:\n            self._trace_stack[-1].children = self._trace_stack[-1].children or []\n            self._trace_stack[-1].children.append(trace)\n        self._trace_stack.append(trace)\n\n    @classmethod\n    def pop(cls, output=None, error: Optional[Exception] = None):\n        obj = cls.active_instance()\n        return obj._pop(output, error)\n\n    def _pop(self, output=None, error: Optional[Exception] = None):\n        last_trace = self._trace_stack[-1]\n        if isinstance(output, Iterator):\n            output = GeneratorProxy(output)\n        if output is not None:\n            last_trace.output = self.to_serializable(output)\n        if error is not None:\n            last_trace.error = self._format_error(error)\n        self._trace_stack[-1].end_time = datetime.utcnow().timestamp()\n        self._trace_stack.pop()\n\n        if isinstance(output, GeneratorProxy):\n            return generate_from_proxy(output)\n        else:\n            return output\n\n    def to_json(self) -> list:\n        return serialize(self._traces)\n\n    @staticmethod\n    def _format_error(error: Exception) -> dict:\n        return {\n            \"message\": str(error),\n            \"type\": type(error).__qualname__,\n        }\n", "path": "src/promptflow/promptflow/_core/tracer.py"}]}