Real-time AI — стриминг, WebRTC и разговорные агенты с минимальной задержкой

import anthropic

client = anthropic.Anthropic()

# Streaming через context manager
with client.messages.stream(
    model="claude-sonnet-5-5",   # актуальные модели: страница «Актуальное сейчас»
    max_tokens=1024,
    messages=[{"role": "user", "content": "Объясни квантовую запутанность простыми словами"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# Получить финальное сообщение после стрима
final_message = stream.get_final_message()
print(f"\n\nStop reason: {final_message.stop_reason}")
print(f"Input tokens: {final_message.usage.input_tokens}")
print(f"Output tokens: {final_message.usage.output_tokens}")

with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                yield event.delta.text
        elif event.type == "message_stop":
            break

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

async def generate_stream(prompt: str):
    """Генератор токенов для SSE"""
    with client.messages.stream(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            # SSE формат: data: <payload>\n\n
            yield f"data: {json.dumps({'text': text})}\n\n"
    
    # Сигнал окончания стрима
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream_response(prompt: str):
    return StreamingResponse(
        generate_stream(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Отключаем буферизацию nginx
        }
    )

const eventSource = new EventSource(`/stream?prompt=${encodeURIComponent(userInput)}`);
const output = document.getElementById('output');

eventSource.onmessage = (event) => {
    if (event.data === '[DONE]') {
        eventSource.close();
        return;
    }
    const { text } = JSON.parse(event.data);
    output.textContent += text;
};

eventSource.onerror = () => {
    eventSource.close();
    console.error('Stream ended or error occurred');
};

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    conversation_history = []
    
    try:
        while True:
            # Получаем сообщение от пользователя
            data = await websocket.receive_text()
            user_message = json.loads(data)
            
            conversation_history.append({
                "role": "user",
                "content": user_message["text"]
            })
            
            # Стримим ответ Claude обратно в WebSocket
            # (синхронный клиент блокирует цикл событий; в продакшне используй AsyncAnthropic и async with)
            full_response = ""
            with client.messages.stream(
                model="claude-sonnet-5-5",
                max_tokens=2048,
                system="Ты полезный ассистент. Отвечай по-русски.",
                messages=conversation_history,
            ) as stream:
                for text in stream.text_stream:
                    full_response += text
                    await websocket.send_json({
                        "type": "delta",
                        "text": text
                    })
            
            # Сигнал конца ответа
            await websocket.send_json({"type": "done"})
            
            # Добавляем ответ в историю для следующего хода
            conversation_history.append({
                "role": "assistant",
                "content": full_response
            })
    
    except WebSocketDisconnect:
        print("Client disconnected")

import time

start_time = time.time()
first_token_time = None
token_count = 0

with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        if first_token_time is None:
            first_token_time = time.time()
            ttft = first_token_time - start_time
            print(f"TTFT: {ttft * 1000:.0f}ms")
        token_count += 1

total_time = time.time() - start_time
tpot = (total_time - first_token_time) / token_count * 1000
print(f"TPOT: {tpot:.1f}ms/token")

Пользователь говорит → [Микрофон] → [WebRTC] → [STT: Whisper / Deepgram]
                                                        ↓
                                              [Claude (текст)]
                                                        ↓
                                    [TTS: ElevenLabs / OpenAI TTS]
                                                        ↓
                        [Пользователь слышит] ← [WebRTC] ← [Аудио]

from livekit import agents
from livekit.agents import AgentServer, AgentSession, Agent
from livekit.plugins import anthropic, openai, silero

class VoiceAssistant(Agent):
    def __init__(self):
        super().__init__(
            instructions="""Ты голосовой ассистент. 
            Отвечай по-русски, кратко и по делу.
            Используй образы и аналогии для объяснений."""
        )

server = AgentServer()

@server.rtc_session()
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        stt=openai.STT(language="ru"),           # Speech-to-Text
        llm=anthropic.LLM(model="claude-sonnet-5-5"),  # LLM; актуальные модели: страница «Актуальное сейчас»
        tts=openai.TTS(voice="alloy"),           # Text-to-Speech
        vad=silero.VAD.load(),                   # Voice Activity Detection
    )
    
    # Шумоподавление и другие параметры комнаты задаются через room_options
    # (см. документацию LiveKit Agents для вашей версии)
    await session.start(room=ctx.room, agent=VoiceAssistant())
    
    # Приветствие
    await session.generate_reply(
        instructions="Поприветствуй пользователя по-русски и предложи помощь"
    )

if __name__ == "__main__":
    agents.cli.run_app(server)

import asyncio
import json
import websockets
import base64

async def realtime_voice_session():
    # Имя модели в URL — пример: возьми актуальное из списка моделей в документации OpenAI
    url = "wss://api.openai.com/v1/realtime?model=gpt-realtime-2.1"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        # Заголовок OpenAI-Beta: realtime=v1 для текущей версии интерфейса не нужен
    }
    
    async with websockets.connect(url, additional_headers=headers) as ws:
        # Конфигурируем сессию. Формат аудио, голос и определение конца речи
        # задаются в session.audio: точные поля смотри в справочнике событий Realtime API
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "instructions": "Ты голосовой ассистент. Отвечай кратко.",
            }
        }))
        
        # Слушаем события
        async for message in ws:
            event = json.loads(message)
            
            if event["type"] == "response.output_audio.delta":
                # Получаем кусок аудио-ответа
                audio_data = base64.b64decode(event["delta"])
                # Воспроизводим аудио...
            
            elif event["type"] == "response.done":
                print("Ответ завершён")

messages = client.messages.create(
    model="claude-sonnet-5-5",
    system=[{
        "type": "text",
        "text": long_system_prompt,
        "cache_control": {"type": "ephemeral"}  # Кешируем
    }],
    messages=conversation,
    max_tokens=1024,
)

buffer = ""
async for text in stream.text_stream:
    buffer += text
    # Если накопили законченное предложение — сразу в TTS
    if any(buffer.endswith(p) for p in ['.', '!', '?', '...']):
        await tts_engine.synthesize_and_play(buffer.strip())
        buffer = ""

pip install fastapi uvicorn anthropic websockets python-dotenv

import os
import json
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import anthropic
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="Streaming Chat API")
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

SYSTEM_PROMPT = """Ты полезный AI-ассистент. Отвечай по-русски, 
используй структуру и образы для объяснений. Будь краток."""

# Простой HTML клиент для тестирования
HTML = """
<!DOCTYPE html>
<html>
<head><title>Streaming Chat</title></head>
<body>
<h1>Streaming Chat с Claude</h1>
<div id="chat" style="height:400px;overflow-y:auto;border:1px solid #ccc;padding:10px;"></div>
<div style="margin-top:10px">
    <input id="input" type="text" style="width:80%" placeholder="Введите сообщение..." />
    <button onclick="sendMessage()">Отправить</button>
</div>
<script>
    const ws = new WebSocket(`ws://${location.host}/ws/chat`);
    const chat = document.getElementById('chat');
    let currentBubble = null;
    
    ws.onmessage = function(event) {
        const data = JSON.parse(event.data);
        if (data.type === 'start') {
            currentBubble = document.createElement('div');
            currentBubble.style.cssText = 'margin:8px 0;padding:8px;background:#e8f4fd;border-radius:8px';
            currentBubble.innerHTML = '<strong>Claude:</strong> ';
            chat.appendChild(currentBubble);
        } else if (data.type === 'delta') {
            currentBubble.innerHTML += data.text;
            chat.scrollTop = chat.scrollHeight;
        } else if (data.type === 'done') {
            currentBubble = null;
        }
    };
    
    function sendMessage() {
        const input = document.getElementById('input');
        const text = input.value.trim();
        if (!text) return;
        
        const userBubble = document.createElement('div');
        userBubble.style.cssText = 'margin:8px 0;padding:8px;background:#f0f0f0;border-radius:8px;text-align:right';
        userBubble.innerHTML = `<strong>Вы:</strong> ${text}`;
        chat.appendChild(userBubble);
        
        ws.send(JSON.stringify({text}));
        input.value = '';
    }
    
    document.getElementById('input').addEventListener('keypress', (e) => {
        if (e.key === 'Enter') sendMessage();
    });
</script>
</body>
</html>
"""

@app.get("/")
async def get():
    return HTMLResponse(HTML)

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    history = []
    
    try:
        while True:
            data = await websocket.receive_text()
            user_msg = json.loads(data)
            
            history.append({
                "role": "user",
                "content": user_msg["text"]
            })
            
            # Сигнал начала ответа
            await websocket.send_json({"type": "start"})
            
            full_response = ""
            
            # Стримим ответ Claude
            with client.messages.stream(
                model="claude-haiku-4-5",  # Haiku для скорости (проверь, что модель ещё доступна в API)
                max_tokens=1024,
                system=SYSTEM_PROMPT,
                messages=history,
            ) as stream:
                for text in stream.text_stream:
                    full_response += text
                    await websocket.send_json({
                        "type": "delta",
                        "text": text
                    })
            
            # Завершение ответа
            await websocket.send_json({"type": "done"})
            
            # Сохраняем в историю
            history.append({
                "role": "assistant",
                "content": full_response
            })
            
            # Ограничиваем историю (последние 10 пар)
            if len(history) > 20:
                history = history[-20:]
    
    except WebSocketDisconnect:
        print(f"Client disconnected, history length: {len(history)}")
    except Exception as e:
        print(f"Error: {e}")
        await websocket.close()

# Убедитесь что ANTHROPIC_API_KEY в .env файле
uvicorn server:app --reload --port 8000

import time

start = time.time()
first_token = True
token_count = 0

with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        if first_token:
            ttft = (time.time() - start) * 1000
            print(f"TTFT: {ttft:.0f}ms")
            first_token = False
        token_count += 1
        # ... отправка клиенту

total = (time.time() - start) * 1000
tpot = total / token_count if token_count > 0 else 0
print(f"TPOT: {tpot:.1f}ms/token | Total tokens: {token_count}")

from fastapi.responses import StreamingResponse

@app.get("/stream")
async def stream_endpoint(prompt: str):
    async def generate():
        with client.messages.stream(
            model="claude-haiku-4-5",  # быстрая модель для real-time
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {json.dumps({'text': text})}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )