IA en tiempo real: streaming, WebRTC y agentes de conversación con latencia mínima

import anthropic

client = anthropic.Anthropic()

# Streaming con un context manager
with client.messages.stream(
    model="claude-sonnet-5-5",   # modelos vigentes: página «Lo vigente»
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explica el entrelazamiento cuántico en palabras sencillas"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# Obtener el mensaje final después del stream
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):
    """Generador de tokens para 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:
            # Formato SSE: data: <payload>\n\n
            yield f"data: {json.dumps({'text': text})}\n\n"
    
    # Señal de fin del stream
    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",  # Desactivamos el búfer de 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:
            # Recibimos el mensaje del usuario
            data = await websocket.receive_text()
            user_message = json.loads(data)
            
            conversation_history.append({
                "role": "user",
                "content": user_message["text"]
            })
            
            # Transmitimos la respuesta de Claude de vuelta por el WebSocket
            # (el cliente síncrono bloquea el ciclo de eventos; en producción usa AsyncAnthropic y async with)
            full_response = ""
            with client.messages.stream(
                model="claude-sonnet-5-5",
                max_tokens=2048,
                system="Eres un asistente útil. Responde en español.",
                messages=conversation_history,
            ) as stream:
                for text in stream.text_stream:
                    full_response += text
                    await websocket.send_json({
                        "type": "delta",
                        "text": text
                    })
            
            # Señal de fin de la respuesta
            await websocket.send_json({"type": "done"})
            
            # Agregamos la respuesta al historial para el siguiente turno
            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")

El usuario habla → [Micrófono] → [WebRTC] → [STT: Whisper / Deepgram]
                                                        ↓
                                              [Claude (texto)]
                                                        ↓
                                    [TTS: ElevenLabs / OpenAI TTS]
                                                        ↓
                        [El usuario escucha] ← [WebRTC] ← [Audio]

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="""Eres un asistente de voz. 
            Responde en español, breve y al grano.
            Usa imágenes y analogías para explicar."""
        )

server = AgentServer()

@server.rtc_session()
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        stt=openai.STT(language="es"),           # Speech-to-Text
        llm=anthropic.LLM(model="claude-sonnet-5-5"),  # LLM; modelos vigentes: página «Lo vigente»
        tts=openai.TTS(voice="alloy"),           # Text-to-Speech
        vad=silero.VAD.load(),                   # Voice Activity Detection
    )
    
    # La supresión de ruido y otros parámetros de la sala se configuran con room_options
    # (ver la documentación de LiveKit Agents para tu versión)
    await session.start(room=ctx.room, agent=VoiceAssistant())
    
    # Saludo
    await session.generate_reply(
        instructions="Saluda al usuario en español y ofrécele ayuda"
    )

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

import asyncio
import json
import websockets
import base64

async def realtime_voice_session():
    # El nombre del modelo en la URL es un ejemplo: tómalo de la lista de modelos vigente en la documentación de OpenAI
    url = "wss://api.openai.com/v1/realtime?model=gpt-realtime-2.1"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        # El encabezado OpenAI-Beta: realtime=v1 no hace falta en la versión actual de la interfaz
    }
    
    async with websockets.connect(url, additional_headers=headers) as ws:
        # Configuramos la sesión. El formato de audio, la voz y la detección del fin del habla
        # se definen en session.audio: los campos exactos están en la referencia de eventos de la Realtime API
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "instructions": "Eres un asistente de voz. Responde breve.",
            }
        }))
        
        # Escuchamos los eventos
        async for message in ws:
            event = json.loads(message)
            
            if event["type"] == "response.output_audio.delta":
                # Recibimos un pedazo del audio de respuesta
                audio_data = base64.b64decode(event["delta"])
                # Reproducimos el audio...
            
            elif event["type"] == "response.done":
                print("Respuesta terminada")

messages = client.messages.create(
    model="claude-sonnet-5-5",
    system=[{
        "type": "text",
        "text": long_system_prompt,
        "cache_control": {"type": "ephemeral"}  # Lo guardamos en caché
    }],
    messages=conversation,
    max_tokens=1024,
)

buffer = ""
async for text in stream.text_stream:
    buffer += text
    # Si ya juntamos una oración completa, directo al 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 = """Eres un asistente de IA útil. Responde en español, 
usa estructura e imágenes para explicar. Sé breve."""

# Un cliente HTML sencillo para probar
HTML = """
<!DOCTYPE html>
<html>
<head><title>Streaming Chat</title></head>
<body>
<h1>Streaming Chat con 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="Escribe un mensaje..." />
    <button onclick="sendMessage()">Enviar</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>Tú:</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"]
            })
            
            # Señal de inicio de la respuesta
            await websocket.send_json({"type": "start"})
            
            full_response = ""
            
            # Transmitimos la respuesta de Claude
            with client.messages.stream(
                model="claude-haiku-4-5",  # Haiku por velocidad (revisa que el modelo siga disponible en la 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
                    })
            
            # Fin de la respuesta
            await websocket.send_json({"type": "done"})
            
            # Lo guardamos en el historial
            history.append({
                "role": "assistant",
                "content": full_response
            })
            
            # Limitamos el historial (los últimos 10 pares)
            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()

# Asegúrate de que ANTHROPIC_API_KEY esté en el archivo .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
        # ... envío al cliente

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",  # modelo rápido para tiempo real
            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"}
    )