Пошаговое руководство по созданию AI-агента, работающего в режиме реального времени через WebSocket-соединение. Вы научитесь стримить ответы LLM, управлять сессиями и оркестрировать вызовы инструментов без задержек, характерных для REST API.
Традиционные REST API-вызовы к LLM страдают от накопительной задержки: каждый запрос требует полного раунд-трипа, а ответ возвращается только после генерации всех токенов. Для AI-агента, который может выполнять 5-10 последовательных вызовов LLM (reasoning → tool choice → tool result → final answer), задержка в 2-3 секунды на каждый этап делает взаимодействие мучительно медленным. WebSocket решает эту проблему: одно постоянное соединение, дуплексная передача сообщений и стриминг токенов в реальном времени.
Ключевое преимущество агента на WebSocket — пользователь видит, как агент «думает» и «действует». Токены появляются на экране по мере генерации, вызовы инструментов видны как промежуточные события. Это создаёт эффект живого диалога и на порядок улучшает UX по сравнению с ожиданием полного ответа.
# Сравнение: REST vs WebSocket для AI-агента # REST подход — каждый шаг отдельный запрос import requests response = requests.post("https://api.openai.com/v1/chat/completions", json={ "model": "gpt-4", "messages": messages, "stream": False }) # Ждём 3-5 секунд полного ответа → обрабатываем → следующий запрос # WebSocket подход — одно соединение, стриминг всего import asyncio import websockets async def agent_loop(websocket): async for message in websocket: # Получаем сообщение, думаем, стримим ответ — всё в одном контексте async for token in stream_llm_response(message): await websocket.send(token) # мгновенная доставка
Ядро real-time агента — асинхронный event loop, который обрабатывает входящие сообщения, управляет состоянием сессии и оркестрирует вызовы LLM и инструментов. Каждое WebSocket-соединение получает собственную сессию с историей диалога, контекстом и активными tool-call цепочками. Ключевые компоненты:
Connection Manager — отслеживает активные соединения, обрабатывает переподключения, управляет heartbeat'ами. Session Store — хранит историю сообщений и состояние агента (Redis для масштабирования, in-memory dict для разработки). LLM Adapter — унифицирует доступ к разным провайдерам (OpenAI, Anthropic, локальные модели) с поддержкой стриминга. Tool Executor — выполняет вызовы инструментов асинхронно, возвращает результаты обратно в LLM.
# connection_manager.py — ядро управления WebSocket-соединениями import asyncio import json from typing import Dict, Set from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections: Dict[str, WebSocket] = {} self.sessions: Dict[str, list] = {} # session_id → message history async def connect(self, session_id: str, ws: WebSocket): await ws.accept() self.active_connections[session_id] = ws if session_id not in self.sessions: self.sessions[session_id] = [] async def send_event(self, session_id: str, event: str, data: dict): ws = self.active_connections.get(session_id) if ws: await ws.send_text(json.dumps({"event": event, "data": data})) def disconnect(self, session_id: str): self.active_connections.pop(session_id, None)
Самая интересная часть — передача токенов LLM в реальном времени и перехват tool calls прямо в потоке. Когда модель хочет вызвать инструмент, мы отправляем клиенту событие «tool_call_start», приостанавливаем стриминг текста, выполняем функцию и отправляем результат обратно модели. Клиент видит весь процесс: модель думает → вызывает поиск → получает результат → формирует финальный ответ.
# streaming_agent.py — стриминг LLM + tool calls через WebSocket import json from openai import AsyncOpenAI client = AsyncOpenAI() async def stream_agent_response(session_id: str, messages: list, manager): # Отправляем системный промпт агенту с доступными инструментами tools = [ {"type": "function", "function": { "name": "search_web", "description": "Поиск информации в интернете", "parameters": {"type": "object", "properties": { "query": {"type": "string"} }, "required": ["query"]}} }} ] stream = await client.chat.completions.create( model="gpt-4", messages=messages, tools=tools, tool_choice="auto", stream=True ) tool_calls_buffer = {} async for chunk in stream: delta = chunk.choices[0].delta if delta.tool_calls: for tc in delta.tool_calls: idx = tc.index if idx not in tool_calls_buffer: tool_calls_buffer[idx] = {"id": tc.id, "name": "", "args": ""} # Уведомляем клиент о начале tool-call await manager.send_event(session_id, "tool_call_start", {"name": tc.function.name or "processing"}) if tc.function.name: tool_calls_buffer[idx]["name"] += tc.function.name if tc.function.arguments: tool_calls_buffer[idx]["args"] += tc.function.arguments elif delta.content: # Стримим текстовый токен клиенту await manager.send_event(session_id, "token", {"content": delta.content})
FastAPI отлично поддерживает WebSocket из коробки благодаря звездной интеграции с asyncio. Создаём эндпоинт, который принимает соединение, инициализирует сессию, загружает историю из Redis (если есть) и запускает основной цикл обработки сообщений. Важно обрабатывать разрыв соединения (graceful shutdown), таймауты и переподключения с восстановлением контекста.
# main.py — FastAPI приложение с WebSocket-эндпоинтом from fastapi import FastAPI, WebSocket, WebSocketDisconnect from contextlib import asynccontextmanager import redis.asyncio as aioredis app = FastAPI() manager = ConnectionManager() redis = None @asynccontextmanager async def lifespan(app: FastAPI): global redis redis = await aioredis.from_url("redis://localhost:6379", decode_responses=True) yield await redis.close() app.router.lifespan_context = lifespan @app.websocket("/ws/agent/{session_id}") async def agent_websocket(websocket: WebSocket, session_id: str): await manager.connect(session_id, websocket) # Восстанавливаем историю из Redis history_json = await redis.get(f"session:{session_id}:messages") messages = json.loads(history_json) if history_json else [ {"role": "system", "content": "Ты AI-агент. Используй инструменты."} ] try: while True: raw = await websocket.receive_text() payload = json.loads(raw) user_msg = payload.get("content", "") messages.append({"role": "user", "content": user_msg}) # Стримим ответ асинхронно await stream_agent_response(session_id, messages, manager) # Сохраняем обновлённую историю в Redis await redis.setex( f"session:{session_id}:messages", 3600, json.dumps(messages) ) except WebSocketDisconnect: manager.disconnect(session_id) print(f"Session {session_id} disconnected") except Exception as e: print(f"Error in session {session_id}: {e}") manager.disconnect(session_id)
В продакшене WebSocket-агент сталкивается с сетевыми проблемами: обрывы соединения, медленные клиенты, «залипание» соединений из-за фаерволов и прокси. Три обязательных паттерна: Heartbeat (ping/pong каждые 30 секунд для поддержания соединения), Reconnect with State Recovery (клиент переподключается и получает недоставленные события через курсор в Redis), Backpressure (если клиент не успевает читать — буферизуем с лимитом, при переполнении дропаем старые сообщения).
# production_ws.py — heartbeat и backpressure для WebSocket агента import asyncio from collections import deque class RobustAgentSession: def __init__(self, ws, session_id, max_buffer=256): self.ws = ws self.sid = session_id self.outbox = deque(maxlen=max_buffer) # кольцевой буфер self.cursor = 0 self._heartbeat_task = None async def start_heartbeat(self, interval=30): async def _ping(): while True: await asyncio.sleep(interval) try: await self.ws.send_json({"type": "ping"}) except: break self._heartbeat_task = asyncio.create_task(_ping()) async def send_with_backpressure(self, data): self.cursor += 1 self.outbox.append({"cursor": self.cursor, "data": data}) await self.ws.send_json(data) async def replay_from(self, cursor: int): # После реконнекта: отправляем пропущенные события for msg in self.outbox: if msg["cursor"] > cursor: await self.ws.send_json(msg["data"])
На фронтенде подключаемся к WebSocket-эндпоинту и обрабатываем поток событий. Токены собираются в реальном времени в DOM-элемент, tool calls отображаются как индикаторы загрузки, отключения обрабатываются с exponential backoff реконнектом. Код ниже показывает минимальный рабочий клиент с обработкой всех типов событий.
// agent-client.js — WebSocket клиент для real-time AI агента class AgentClient { constructor(url, sessionId) { this.url = `${url}/ws/agent/${sessionId}`; this.cursor = 0; this.retries = 0; this.outputEl = document.getElementById('agent-output'); this.connect(); } connect() { this.ws = new WebSocket(this.url); this.ws.onopen = () => { console.log('Connected to AI agent'); if (this.cursor > 0) { this.ws.send(JSON.stringify({ type: 'replay', cursor: this.cursor })); } }; this.ws.onmessage = (event) => { const msg = JSON.parse(event.data); this.cursor = msg.cursor || this.cursor; switch (msg.event) { case 'token': this.outputEl.textContent += msg.data.content; break; case 'tool_call_start': this.showToolIndicator(msg.data.name); break; case 'tool_result': this.hideToolIndicator(); break; case 'done': this.onComplete(msg.data); break; } }; this.ws.onclose = () => { const backoff = Math.min(1000 * 2 ** this.retries, 30000); setTimeout(() => { this.retries++; this.connect(); }, backoff); }; } send(content) { this.ws.send(JSON.stringify({ type: 'message', content })); } }
Вы построили полноценного real-time AI-агента на WebSocket: асинхронный сервер на FastAPI, стриминг ответов LLM с перехватом tool calls, управление сессиями через Redis и надёжный клиент с automatic reconnect. Следующий шаг — добавление мульти-агентной оркестрации, где несколько агентов общаются через общий WebSocket-канал, и интеграция голосового ввода (WebRTC + STT), чтобы превратить текстового агента в голосового ассистента реального времени. Также рассмотрите Server-Sent Events (SSE) как более лёгкую альтернативу для сценариев, где достаточно однонаправленного стриминга от сервера к клиенту.