from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
from llama_index.schema import NodeWithScore
from llama_index.types import TokenGen
from llama_index.utils import truncate_text
@dataclass
class StreamingResponse:
"""StreamingResponse object."""
response_gen: TokenGen
source_nodes: List[NodeWithScore] = field(default_factory=list)
metadata: Optional[Dict[str, Any]] = None
response_txt: Optional[str] = None
def __str__(self) -> str:
"""Convert to string representation."""
if self.response_txt is None and self.response_gen is not None:
response_txt = ""
for text in self.response_gen:
response_txt += text
self.response_txt = response_txt
return self.response_txt or "None"
def get_response(self) -> Response:
"""Get a standard response object."""
if self.response_txt is None and self.response_gen is not None:
response_txt = ""
for text in self.response_gen:
response_txt += text
self.response_txt = response_txt
return Response(self.response_txt, self.source_nodes, self.metadata)
def print_response_stream(self) -> None:
"""Print the response stream."""
if self.response_txt is None and self.response_gen is not None:
response_txt = ""
for text in self.response_gen:
print(text, end="", flush=True)
response_txt += text
self.response_txt = response_txt
else:
print(self.response_txt)
I want to use the inner code of the above def print_response_stream(self) -> None: method but change it in a way so I can use it for the streamlit framework where I can replace the print() method with message_placeholder.markdown() so that it prints the response_txt and text in my chat instead of the console.
instance_of_schema = schema.StreamingResponse()
if instance_of_schema.response_txt is None and instance_of_schema.response_gen is not None:
response_txt = ""
for text in instance_of_schema.response_gen:
message_placeholder.markdown(text, end="", flush=True)
response_txt += text
instance_of_schema.response_txt = response_txt
else:
message_placeholder.markdown(instance_of_schema.response_txt)
full_response = instance_of_schema.response_txt
But then it gives me this Exception:
instance_of_schema = schema.StreamingResponse()
^^^^^^^^^^^^^^^^^^^^^^^^^^ TypeError: StreamingResponse.init() missing 1 required positional argument: 'response_gen'
so then I tried it this way:
if self.response_txt is None and self.response_gen is not None:
response_txt = ""
for text in self.response_gen:
message_placeholder.markdown(text, end="", flush=True)
response_txt += text
self.response_txt = response_txt
else:
message_placeholder.markdown(self.response_txt)
full_response = self.response_txt
it gives me this Ecxeption:
if self.response_txt is None and self.response_gen is not None:
^^^^
NameError: name 'self' is not defined
I tried it this way:
if schema.StreamingResponse.response_txt is None and schema.StreamingResponse.response_gen is not None:
response_txt = ""
for text in schema.StreamingResponse.response_gen:
message_placeholder.markdown(text, end="", flush=True)
response_txt += text
schema.StreamingResponse.response_txt = response_txt
else:
message_placeholder.markdown(schema.StreamingResponse.response_txt)
full_response = schema.StreamingResponse.response_txt
Then it throws this Exception:
if schema.StreamingResponse.response_txt is None and schema.StreamingResponse.response_gen is not None:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: type object 'StreamingResponse' has no attribute 'response_gen'
Sorry I think there is probably a pretty easy answer, but I don't get it, I'm new with Python and what I found so far didn't help me out..