response_generator = self.agent.stream_chat(message=messages[-1].content, chat_history=messages[:-1]).chat_stream
for token in response_generator:
yield tokenValueError: generator already executing
response_gen instead of chat_stream it's working flawless. However, i truly need that ChatResponse objectsresp = agent.stream_chat(...) for r in resp.response_gen: print(r.delta, end="", flush=True)
thread = Thread(
target=agent_response_stream.write_response_to_history,
args=(task.extra_state["new_memory"],),
kwargs={"on_stream_end_fn": partial(self.finalize_task, task)},
)
thread.start()agent_response_stream.write_response_to_history is already consuming chat_stream generator def write_response_to_history(
self,
memory: BaseMemory,
on_stream_end_fn: Optional[Callable] = None,
) -> None:
if self.chat_stream is None:
raise ValueError(
"chat_stream is None. Cannot write to history without chat_stream."
)
# try/except to prevent hanging on error
dispatcher.event(StreamChatStartEvent())
try:
final_text = ""
for chat in self.chat_stream:
self.is_function = is_function(chat.message)
if chat.delta:
dispatcher.event(
StreamChatDeltaReceivedEvent(
delta=chat.delta,
)
)
self.put_in_queue(chat.delta)
[...]