An asynchronous generator lets a chat interface display an answer before the LLM finishes generating it. Instead of waiting for one complete response, an async query method yields structured updates as information becomes available. section 12.3
In a retrieval-augmented chat flow, yield the retrieved sources first so users can see the basis for the answer. Then iterate over the LLM’s `astream` output and yield each nonempty token for immediate display. An explicit completion signal tells the UI when generation has finished. section 12.3
Listing 12.15 Streaming query processing with real-time updates
async def process_query_stream(self, #1
query: str) #1
-> AsyncGenerator[Dict[str, Any], None]: #1
try:
docs = self.retriever.invoke(query)
sources = [...] # Source preparation code
yield {
"type": "sources", (C)#2
"content": None,
"sources": sources,
"doc_count": len(docs)
}
context = "\n\n".join([...])
prompt = f"""..."""
async for chunk in self.llm.astream( #3
prompt,
config={"callbacks": [self.langfuse_handler]}
):
if chunk.content:
yield {
"type": "token", #4
"content": chunk.content,
"sources": sources,
"doc_count": len(docs)
}
yield {
"type": "done", #5
"content": None,
"sources": sources,
"doc_count": len(docs)
}
except Exception as e:
yield {
"type": "error", #6
"content": str(e),
"sources": [],
"doc_count": 0
}
The streaming query method shows the `AsyncGenerator` return type, a sources event, and token events produced inside an `async for` loop. section 12.3
