Overview
StreamedChatResponse is a large discriminated union of nine event types representing every possible event emitted during a streaming chat API response.
Description
The StreamedChatResponse union is returned when calling the Cohere Chat API with stream=True. It is discriminated on the event_type field and contains the following variants:
- StreamStartStreamedChatResponse (
event_type="stream-start") -- The initial event emitted when a stream begins. Contains the generation_id.
- SearchQueriesGenerationStreamedChatResponse (
event_type="search-queries-generation") -- Emitted when the model generates search queries for RAG. Contains a list of ChatSearchQuery objects.
- SearchResultsStreamedChatResponse (
event_type="search-results") -- Emitted when search results are retrieved. Contains optional lists of ChatSearchResult and ChatDocument objects.
- TextGenerationStreamedChatResponse (
event_type="text-generation") -- Emitted for each text chunk as the model generates its response. Contains a text string.
- CitationGenerationStreamedChatResponse (
event_type="citation-generation") -- Emitted when citations are generated. Contains a list of ChatCitation objects.
- ToolCallsGenerationStreamedChatResponse (
event_type="tool-calls-generation") -- Emitted when the model generates complete tool calls. Contains a list of ToolCall objects and optional text.
- StreamEndStreamedChatResponse (
event_type="stream-end") -- The final event in a stream. Contains the finish_reason and the complete NonStreamedChatResponse.
- ToolCallsChunkStreamedChatResponse (
event_type="tool-calls-chunk") -- Emitted during streaming tool call generation. Contains a ToolCallDelta with partial tool call data and optional text.
- DebugStreamedChatResponse (
event_type="debug") -- Emitted for debug information. Contains an optional prompt string.
All variants extend UncheckedBaseModel and allow extra fields.
Usage
Use StreamedChatResponse when consuming events from a streaming chat call. Iterate over the stream and dispatch on event_type to handle each event type appropriately -- accumulating text chunks, processing citations, executing tool calls, or finalizing the response.
Code Reference
Source Location
Signature
class StreamStartStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["stream-start"] = "stream-start"
generation_id: str
class SearchQueriesGenerationStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["search-queries-generation"] = "search-queries-generation"
search_queries: typing.List[ChatSearchQuery]
class SearchResultsStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["search-results"] = "search-results"
search_results: typing.Optional[typing.List[ChatSearchResult]] = None
documents: typing.Optional[typing.List[ChatDocument]] = None
class TextGenerationStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["text-generation"] = "text-generation"
text: str
class CitationGenerationStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["citation-generation"] = "citation-generation"
citations: typing.List[ChatCitation]
class ToolCallsGenerationStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["tool-calls-generation"] = "tool-calls-generation"
text: typing.Optional[str] = None
tool_calls: typing.List[ToolCall]
class StreamEndStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["stream-end"] = "stream-end"
finish_reason: ChatStreamEndEventFinishReason
response: NonStreamedChatResponse
class ToolCallsChunkStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["tool-calls-chunk"] = "tool-calls-chunk"
tool_call_delta: ToolCallDelta
text: typing.Optional[str] = None
class DebugStreamedChatResponse(UncheckedBaseModel):
event_type: typing.Literal["debug"] = "debug"
prompt: typing.Optional[str] = None
StreamedChatResponse = typing_extensions.Annotated[
typing.Union[
StreamStartStreamedChatResponse,
SearchQueriesGenerationStreamedChatResponse,
SearchResultsStreamedChatResponse,
TextGenerationStreamedChatResponse,
CitationGenerationStreamedChatResponse,
ToolCallsGenerationStreamedChatResponse,
StreamEndStreamedChatResponse,
ToolCallsChunkStreamedChatResponse,
DebugStreamedChatResponse,
],
UnionMetadata(discriminant="event_type"),
]
Import
from cohere.types import (
StreamedChatResponse,
StreamStartStreamedChatResponse,
SearchQueriesGenerationStreamedChatResponse,
SearchResultsStreamedChatResponse,
TextGenerationStreamedChatResponse,
CitationGenerationStreamedChatResponse,
ToolCallsGenerationStreamedChatResponse,
StreamEndStreamedChatResponse,
ToolCallsChunkStreamedChatResponse,
DebugStreamedChatResponse,
)
I/O Contract
StreamStartStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["stream-start"] |
Yes |
Discriminant for stream start event.
|
generation_id |
str |
Yes |
Unique identifier for this generation stream.
|
SearchQueriesGenerationStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["search-queries-generation"] |
Yes |
Discriminant for search query generation event.
|
search_queries |
List[ChatSearchQuery] |
Yes |
The search queries generated by the model.
|
SearchResultsStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["search-results"] |
Yes |
Discriminant for search results event.
|
search_results |
Optional[List[ChatSearchResult]] |
No |
The search results retrieved.
|
documents |
Optional[List[ChatDocument]] |
No |
The documents from search results.
|
TextGenerationStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["text-generation"] |
Yes |
Discriminant for text generation event.
|
text |
str |
Yes |
A chunk of generated text.
|
CitationGenerationStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["citation-generation"] |
Yes |
Discriminant for citation generation event.
|
citations |
List[ChatCitation] |
Yes |
The citations generated for the response.
|
ToolCallsGenerationStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["tool-calls-generation"] |
Yes |
Discriminant for tool calls generation event.
|
text |
Optional[str] |
No |
Optional text accompanying the tool calls.
|
tool_calls |
List[ToolCall] |
Yes |
The complete tool calls generated by the model.
|
StreamEndStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["stream-end"] |
Yes |
Discriminant for stream end event.
|
finish_reason |
ChatStreamEndEventFinishReason |
Yes |
The reason the stream ended (e.g., COMPLETE, MAX_TOKENS).
|
response |
NonStreamedChatResponse |
Yes |
The complete aggregated response object.
|
ToolCallsChunkStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["tool-calls-chunk"] |
Yes |
Discriminant for tool calls chunk event.
|
tool_call_delta |
ToolCallDelta |
Yes |
Partial tool call data for streaming.
|
text |
Optional[str] |
No |
Optional text chunk accompanying the tool call delta.
|
DebugStreamedChatResponse Fields
| Field |
Type |
Required |
Description
|
event_type |
Literal["debug"] |
Yes |
Discriminant for debug event.
|
prompt |
Optional[str] |
No |
The prompt sent to the model (for debugging).
|
Usage Examples
from cohere.types import (
StreamStartStreamedChatResponse,
TextGenerationStreamedChatResponse,
ToolCallsGenerationStreamedChatResponse,
StreamEndStreamedChatResponse,
)
# Consume a streaming chat response
stream = client.chat_stream(
message="Tell me about machine learning",
connectors=[{"id": "web-search"}]
)
full_text = ""
for event in stream:
if event.event_type == "stream-start":
print(f"Stream started: {event.generation_id}")
elif event.event_type == "text-generation":
full_text += event.text
print(event.text, end="")
elif event.event_type == "search-queries-generation":
for query in event.search_queries:
print(f"Search query: {query.text}")
elif event.event_type == "citation-generation":
for citation in event.citations:
print(f"Citation: {citation}")
elif event.event_type == "tool-calls-generation":
for tool_call in event.tool_calls:
print(f"Tool call: {tool_call.name}({tool_call.parameters})")
elif event.event_type == "tool-calls-chunk":
delta = event.tool_call_delta
if delta.name:
print(f"Tool name chunk: {delta.name}")
if delta.parameters:
print(f"Params chunk: {delta.parameters}")
elif event.event_type == "stream-end":
print(f"\nFinish reason: {event.finish_reason}")
final_response = event.response
Related Pages
