Spaces:
Running
Running
File size: 2,182 Bytes
eb10851 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
"""
SSE (Server-Sent Events) Utilities
Format streaming responses for real-time chat
"""
import json
from typing import Dict, Any, AsyncGenerator
import asyncio
def format_sse(event: str, data: Any) -> str:
"""
Format data as SSE message
Args:
event: Event type (token, status, done, error)
data: Data payload (string or dict)
Returns:
Formatted SSE string
Example:
format_sse("token", "Hello")
# "event: token\ndata: Hello\n\n"
"""
if isinstance(data, dict):
data_str = json.dumps(data, ensure_ascii=False)
else:
data_str = str(data)
return f"event: {event}\ndata: {data_str}\n\n"
async def simulate_typing(
text: str,
chars_per_chunk: int = 3,
delay_ms: float = 20
) -> AsyncGenerator[str, None]:
"""
Simulate typing effect by yielding text in chunks
Args:
text: Full text to stream
chars_per_chunk: Characters per chunk
delay_ms: Milliseconds delay between chunks
Yields:
Text chunks
Example:
async for chunk in simulate_typing("Hello world", chars_per_chunk=2):
yield format_sse("token", chunk)
"""
for i in range(0, len(text), chars_per_chunk):
chunk = text[i:i + chars_per_chunk]
yield chunk
await asyncio.sleep(delay_ms / 1000)
async def stream_text_slowly(
text: str,
event_type: str = "token",
chars_per_chunk: int = 3,
delay_ms: float = 20
) -> AsyncGenerator[str, None]:
"""
Stream text with typing effect in SSE format
Args:
text: Text to stream
event_type: SSE event type
chars_per_chunk: Characters per chunk
delay_ms: Delay between chunks
Yields:
SSE formatted chunks
"""
async for chunk in simulate_typing(text, chars_per_chunk, delay_ms):
yield format_sse(event_type, chunk)
# Event type constants
EVENT_STATUS = "status"
EVENT_TOKEN = "token"
EVENT_DONE = "done"
EVENT_ERROR = "error"
EVENT_METADATA = "metadata"
|