Streams¶
If you haven't already, we recommend first reading the section on Calls
Streaming is a powerful feature when using LLMs that allows you to process chunks of an LLM response in real-time as they are generated. This can be particularly useful for long-running tasks, providing immediate feedback to users, or implementing more responsive applications.
Diagram illustrating standard vs. streaming responses
sequenceDiagram
participant User
participant App
participant LLM
User->>App: Request
App->>LLM: Query
Note right of LLM: Standard Response
LLM-->>App: Complete Response
App-->>User: Display Result
User->>App: Request
App->>LLM: Query (Stream)
Note right of LLM: Streaming Response
loop For each chunk
LLM-->>App: Response Chunk
App-->>User: Display Chunk
endThis approach offers several benefits:
- Immediate feedback: Users can see responses as they're being generated, creating a more interactive experience.
- Reduced latency: For long responses, users don't have to wait for the entire generation to complete before seeing results.
- Incremental processing: Applications can process and act on partial results as they arrive.
- Efficient resource use: Memory usage can be optimized by processing chunks instead of storing the entire response.
- Early termination: If the desired information is found early in the response, processing can be stopped without waiting for the full generation.
API Documentation
Basic Usage and Syntax¶
To use streaming, simply set the stream parameter to True in your call decorator:
from mirascope.core import Messages, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import bedrock, prompt_template
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
In this example:
- We use the
calldecorator withstream=Trueto enable streaming. - The
recommend_bookfunction now returns a generator that yields(chunk, tool)tuples of the response. - We iterate over the chunks, printing each one as it's received.
- We use
end=""andflush=Trueparameters in the print function to ensure that the output is displayed in real-time without line breaks.
Handling Streamed Responses¶
API Documentation
mirascope.core.base.call_response_chunk
mirascope.core.openai.call_response_chunk
mirascope.core.anthropic.call_response_chunk
mirascope.core.mistral.call_response_chunk
mirascope.core.gemini.call_response_chunk
mirascope.core.groq.call_response_chunk
mirascope.core.cohere.call_response_chunk
mirascope.core.litellm.call_response_chunk
mirascope.core.azure.call_response_chunk
When streaming, the initial response will be a provider-specific BaseStream instance (e.g. OpenAIStream), which is a generator that yields tuples (chunk, tool) where chunk is a provider-specific BaseCallResponseChunk (e.g. OpenAICallResponseChunk) that wraps the original chunk in the provider's response. These objects provide a consistent interface across providers while still allowing access to provider-specific details.
Streaming Tools
You'll notice in the above example that we ignore the tool in each tuple. If no tools are set in the call, then tool will always be None and can be safely ignored. For more details, check out the documentation on streaming tools
Common Chunk Properties and Methods¶
All BaseCallResponseChunk objects share these common properties:
content: The main text content of the response. If no content is present, this will be the empty string.finish_reasons: A list of reasons why the generation finished (e.g., "stop", "length"). These will be typed specifically for the provider used. If no finish reasons are present, this will beNone.model: The name of the model used for generation.id: A unique identifier for the response if available. Otherwise this will beNone.usage: Information about token usage for the call if available. Otherwise this will beNone.input_tokens: The number of input tokens used if available. Otherwise this will beNone.output_tokens: The number of output tokens generated if available. Otherwise this will beNone.
Common Stream Properties and Methods¶
Must Exhaust Stream
To access these properties, you must first exhaust the stream by iterating through it.
Once exhausted, all BaseStream objects share the same common properties and methods as BaseCallResponse, except for usage, tools, tool, and __str__.
from mirascope.core import anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import Messages, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import openai, prompt_template
@openai.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import anthropic, prompt_template
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import mistral, prompt_template
@mistral.call("mistral-large-latest", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import gemini, prompt_template
@gemini.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import groq, prompt_template
@groq.call("llama-3.1-70b-versatile", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import cohere, prompt_template
@cohere.call("command-r-plus", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import litellm, prompt_template
@litellm.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import azure, prompt_template
@azure.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import prompt_template, vertex
@vertex.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import bedrock, prompt_template
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
from mirascope.core import BaseMessageParam, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
You can access the additional missing properties by using the method construct_call_response to reconstruct a provider-specific BaseCallResponse instance:
from mirascope.core import openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import Messages, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import openai, prompt_template
@openai.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import anthropic, prompt_template
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import mistral, prompt_template
@mistral.call("mistral-large-latest", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import gemini, prompt_template
@gemini.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import groq, prompt_template
@groq.call("llama-3.1-70b-versatile", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import cohere, prompt_template
@cohere.call("command-r-plus", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import litellm, prompt_template
@litellm.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import azure, prompt_template
@azure.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import prompt_template, vertex
@vertex.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import bedrock, prompt_template
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
from mirascope.core import BaseMessageParam, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(chunk.content, end="", flush=True)
print(f"Content: {stream.content}")
call_response = stream.construct_call_response()
print(f"Usage: {call_response.usage}")
Reconstructed Response Limitations
While we try our best to reconstruct the BaseCallResponse instance from the stream, there's always a chance that some information present in a standard call might be missing from the stream.
Provider-Specific Response Details¶
While Mirascope provides a consistent interface, you can always access the full, provider-specific response object if needed. This is available through the chunk property of the BaseCallResponseChunk object:
from mirascope.core import anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import Messages, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import openai, prompt_template
@openai.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import anthropic, prompt_template
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import mistral, prompt_template
@mistral.call("mistral-large-latest", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import gemini, prompt_template
@gemini.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import groq, prompt_template
@groq.call("llama-3.1-70b-versatile", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import cohere, prompt_template
@cohere.call("command-r-plus", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import litellm, prompt_template
@litellm.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import azure, prompt_template
@azure.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import prompt_template, vertex
@vertex.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import bedrock, prompt_template
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, openai
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, mistral
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
from mirascope.core import BaseMessageParam, bedrock
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
stream = recommend_book("fantasy")
for chunk, _ in stream:
print(f"Original chunk: {chunk.chunk}")
print(chunk.content, end="", flush=True)
Reasoning For Provider-Specific BaseCallResponseChunk Objects
The reason that we have provider-specific response objects (e.g. OpenAICallResponseChunk) is to provide proper type hints and safety when accessing the original response chunk.
Multi-Modal Outputs¶
While most LLM providers focus on text streaming, some providers support streaming additional output modalities like audio. The availability of multi-modal streaming varies among providers:
| Provider | Text | Audio | Image |
|---|---|---|---|
| OpenAI | ✓ | ✓ | - |
| Anthropic | ✓ | - | - |
| Mistral | ✓ | - | - |
| Gemini | ✓ | - | - |
| Groq | ✓ | - | - |
| Cohere | ✓ | - | - |
| LiteLLM | ✓ | - | - |
| Azure AI | ✓ | - | - |
| Vertex AI | ✓ | - | - |
Legend: ✓ (Supported), - (Not Supported)
Audio Streaming¶
For providers that support audio outputs, you can stream both text and audio responses simultaneously:
import io
from pydub.playback import play
from pydub import AudioSegment
from mirascope.core import openai
SAMPLE_WIDTH = 2
FRAME_RATE = 24000
CHANNELS = 1
@openai.call(
"gpt-4o-audio-preview",
call_params={
"audio": {"voice": "alloy", "format": "pcm16"},
"modalities": ["text", "audio"],
},
stream=True,
)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
audio_chunk = b""
audio_transcript_chunk = ""
stream = recommend_book("fantasy")
for chunk, _ in stream:
if chunk.audio:
audio_chunk += chunk.audio
if chunk.audio_transcript:
audio_transcript_chunk += chunk.audio_transcript
print(audio_transcript_chunk)
audio_segment = AudioSegment.from_raw(
io.BytesIO(audio_chunk),
sample_width=SAMPLE_WIDTH,
frame_rate=FRAME_RATE,
channels=CHANNELS,
)
play(audio_segment)
import io
from pydub.playback import play
from pydub import AudioSegment
from mirascope.core import openai, Messages
SAMPLE_WIDTH = 2
FRAME_RATE = 24000
CHANNELS = 1
@openai.call(
"gpt-4o-audio-preview",
call_params={
"audio": {"voice": "alloy", "format": "pcm16"},
"modalities": ["text", "audio"],
},
stream=True,
)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
audio_chunk = b""
audio_transcript_chunk = ""
stream = recommend_book("fantasy")
for chunk, _ in stream:
if chunk.audio:
audio_chunk += chunk.audio
if chunk.audio_transcript:
audio_transcript_chunk += chunk.audio_transcript
print(audio_transcript_chunk)
audio_segment = AudioSegment.from_raw(
io.BytesIO(audio_chunk),
sample_width=SAMPLE_WIDTH,
frame_rate=FRAME_RATE,
channels=CHANNELS,
)
play(audio_segment)
import io
from pydub.playback import play
from pydub import AudioSegment
from mirascope.core import openai, prompt_template
SAMPLE_WIDTH = 2
FRAME_RATE = 24000
CHANNELS = 1
@openai.call(
"gpt-4o-audio-preview",
call_params={
"audio": {"voice": "alloy", "format": "pcm16"},
"modalities": ["text", "audio"],
},
stream=True,
)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
audio_chunk = b""
audio_transcript_chunk = ""
stream = recommend_book("fantasy")
for chunk, _ in stream:
if chunk.audio:
audio_chunk += chunk.audio
if chunk.audio_transcript:
audio_transcript_chunk += chunk.audio_transcript
print(audio_transcript_chunk)
audio_segment = AudioSegment.from_raw(
io.BytesIO(audio_chunk),
sample_width=SAMPLE_WIDTH,
frame_rate=FRAME_RATE,
channels=CHANNELS,
)
play(audio_segment)
import io
from pydub.playback import play
from pydub import AudioSegment
from mirascope.core import openai, BaseMessageParam
SAMPLE_WIDTH = 2
FRAME_RATE = 24000
CHANNELS = 1
@openai.call(
"gpt-4o-audio-preview",
call_params={
"audio": {"voice": "alloy", "format": "pcm16"},
"modalities": ["text", "audio"],
},
stream=True,
)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
audio_chunk = b""
audio_transcript_chunk = ""
stream = recommend_book("fantasy")
for chunk, _ in stream:
if chunk.audio:
audio_chunk += chunk.audio
if chunk.audio_transcript:
audio_transcript_chunk += chunk.audio_transcript
print(audio_transcript_chunk)
audio_segment = AudioSegment.from_raw(
io.BytesIO(audio_chunk),
sample_width=SAMPLE_WIDTH,
frame_rate=FRAME_RATE,
channels=CHANNELS,
)
play(audio_segment)
Each stream chunk provides access to:
chunk.audio: Raw audio data in bytes formatchunk.audio_transcript: The transcript of the audio
This allows you to process both text and audio streams concurrently. Since audio data is received in chunks, you could technically begin playback before receiving the complete response.
Audio Playback Requirements
The example above uses pydub and ffmpeg for audio playback, but you can use any audio processing libraries or media players that can handle WAV format audio data. Choose the tools that best fit your needs and environment.
If you decide to use pydub:
- Install pydub: pip install pydub
- Install ffmpeg: Available from ffmpeg.org or through system package managers
Voice Options
For providers that support audio outputs, refer to their documentation for available voice options and configurations:
- OpenAI: Text to Speech Guide
Error Handling¶
Error handling in streams is similar to standard non-streaming calls. However, it's important to note that errors may occur during iteration rather than at the initial function call:
from mirascope.core import openai
from openai import OpenAIError
@openai.call(model="gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except OpenAIError as e:
print(f"Error: {str(e)}")
from anthropic import AnthropicError
from mirascope.core import anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except AnthropicError as e:
print(f"Error: {str(e)}")
from mirascope.core import mistral
from mistralai import models
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except models.HTTPValidationError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle e.data: models.HTTPValidationErrorData
raise (e)
except models.SDKError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle exception
raise (e)
from groq import GroqError
from mirascope.core import groq
@groq.call(model="llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except GroqError as e:
print(f"Error: {str(e)}")
from cohere.errors import BadRequestError
from mirascope.core import cohere
@cohere.call(model="command-r-plus", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from litellm.exceptions import BadRequestError
from mirascope.core import litellm
@litellm.call(model="gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from mirascope.core import bedrock
from botocore.exceptions import ClientError
@bedrock.call(model="anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> str:
return f"Recommend a {genre} book"
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except ClientError as e:
print(f"Error: {str(e)}")
from mirascope.core import Messages, openai
from openai import OpenAIError
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except OpenAIError as e:
print(f"Error: {str(e)}")
from anthropic import AnthropicError
from mirascope.core import Messages, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except AnthropicError as e:
print(f"Error: {str(e)}")
from mirascope.core import Messages, mistral
from mistralai import models
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except models.HTTPValidationError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle e.data: models.HTTPValidationErrorData
raise (e)
except models.SDKError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle exception
raise (e)
from mirascope.core import Messages, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from groq import GroqError
from mirascope.core import Messages, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except GroqError as e:
print(f"Error: {str(e)}")
from cohere.errors import BadRequestError
from mirascope.core import Messages, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from litellm.exceptions import BadRequestError
from mirascope.core import Messages, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from mirascope.core import Messages, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import Messages, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import Messages, bedrock
from botocore.exceptions import ClientError
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> Messages.Type:
return Messages.User(f"Recommend a {genre} book")
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except ClientError as e:
print(f"Error: {str(e)}")
from mirascope.core import openai, prompt_template
from openai import OpenAIError
@openai.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except OpenAIError as e:
print(f"Error: {str(e)}")
from anthropic import AnthropicError
from mirascope.core import anthropic, prompt_template
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except AnthropicError as e:
print(f"Error: {str(e)}")
from mirascope.core import mistral, prompt_template
from mistralai import models
@mistral.call("mistral-large-latest", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except models.HTTPValidationError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle e.data: models.HTTPValidationErrorData
raise (e)
except models.SDKError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle exception
raise (e)
from mirascope.core import gemini, prompt_template
@gemini.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from groq import GroqError
from mirascope.core import groq, prompt_template
@groq.call("llama-3.1-70b-versatile", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except GroqError as e:
print(f"Error: {str(e)}")
from cohere.errors import BadRequestError
from mirascope.core import cohere, prompt_template
@cohere.call("command-r-plus", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from litellm.exceptions import BadRequestError
from mirascope.core import litellm, prompt_template
@litellm.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from mirascope.core import azure, prompt_template
@azure.call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import prompt_template, vertex
@vertex.call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import bedrock, prompt_template
from botocore.exceptions import ClientError
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str): ...
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except ClientError as e:
print(f"Error: {str(e)}")
from mirascope.core import BaseMessageParam, openai
from openai import OpenAIError
@openai.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except OpenAIError as e:
print(f"Error: {str(e)}")
from anthropic import AnthropicError
from mirascope.core import BaseMessageParam, anthropic
@anthropic.call("claude-3-5-sonnet-20240620", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except AnthropicError as e:
print(f"Error: {str(e)}")
from mirascope.core import BaseMessageParam, mistral
from mistralai import models
@mistral.call("mistral-large-latest", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except models.HTTPValidationError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle e.data: models.HTTPValidationErrorData
raise (e)
except models.SDKError as e: # pyright: ignore [reportAttributeAccessIssue]
# handle exception
raise (e)
from mirascope.core import BaseMessageParam, gemini
@gemini.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from groq import GroqError
from mirascope.core import BaseMessageParam, groq
@groq.call("llama-3.1-70b-versatile", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except GroqError as e:
print(f"Error: {str(e)}")
from cohere.errors import BadRequestError
from mirascope.core import BaseMessageParam, cohere
@cohere.call("command-r-plus", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from litellm.exceptions import BadRequestError
from mirascope.core import BaseMessageParam, litellm
@litellm.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except BadRequestError as e:
print(f"Error: {str(e)}")
from mirascope.core import BaseMessageParam, azure
@azure.call("gpt-4o-mini", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import BaseMessageParam, vertex
@vertex.call("gemini-1.5-flash", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except Exception as e:
print(f"Error: {str(e)}")
from mirascope.core import BaseMessageParam, bedrock
from botocore.exceptions import ClientError
@bedrock.call("anthropic.claude-3-haiku-20240307-v1:0", stream=True)
def recommend_book(genre: str) -> list[BaseMessageParam]:
return [BaseMessageParam(role="user", content=f"Recommend a {genre} book")]
try:
for chunk, _ in recommend_book("fantasy"):
print(chunk.content, end="", flush=True)
except ClientError as e:
print(f"Error: {str(e)}")
In these examples, we wrap the iteration loop in a try/except block to catch any errors that might occur during streaming.
When Errors Occur
The initial response when calling an LLM function with stream=True will return a generator. Any errors that may occur during streaming will not happen until you actually iterate through the generator. This is why we wrap the generation loop in the try/except block and not just the call to recommend_book.
Next Steps¶
By leveraging streaming effectively, you can create more responsive and efficient LLM-powered applications with Mirascope's streaming capabilities.
Next, we recommend taking a look at the Streams documentation, which shows you how to break tasks down into smaller, more directed calls and chain them togethder.