AI-система управления виртуальными мероприятиями
Организация онлайн-конференции на 500+ участников — это не только платформа для трансляции. Это регистрация с персонализацией, рассылки по сегментам, модерация Q&A в реальном времени, автоматические follow-up письма после секций, аналитика вовлечённости. AI-система закрывает операционную рутину: организатор занимается контентом, а не ответами на «а когда будут записи?».
Архитектура системы управления мероприятием
from anthropic import Anthropic
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from datetime import datetime, timedelta
import json
import asyncio
client = Anthropic()
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
class EventKnowledgeBase:
"""База знаний о мероприятии для AI-ассистента"""
def __init__(self, event_id: str):
self.vectorstore = Chroma(
collection_name=f"event_{event_id}",
embedding_function=embeddings,
)
def index_event_data(self, event_data: dict):
"""Индексирует данные о мероприятии"""
texts = []
metadatas = []
# Программа
for session in event_data.get("sessions", []):
text = f"""Секция: {session['title']}
Спикер: {session['speaker']} ({session['speaker_bio']})
Время: {session['start_time']} — {session['end_time']}
Описание: {session['description']}
Зал/трек: {session.get('track', 'общий')}"""
texts.append(text)
metadatas.append({"type": "session", "session_id": session["id"]})
# FAQ
for item in event_data.get("faq", []):
texts.append(f"Вопрос: {item['q']}\nОтвет: {item['a']}")
metadatas.append({"type": "faq"})
# Спикеры
for speaker in event_data.get("speakers", []):
text = f"""Спикер: {speaker['name']}
Должность: {speaker['title']} в {speaker['company']}
Биография: {speaker['bio']}
Темы: {', '.join(speaker.get('topics', []))}"""
texts.append(text)
metadatas.append({"type": "speaker", "speaker_name": speaker["name"]})
self.vectorstore.add_texts(texts=texts, metadatas=metadatas)
class EventAssistant:
"""AI-ассистент для участников мероприятия"""
def __init__(self, event_id: str, event_name: str):
self.event_name = event_name
self.kb = EventKnowledgeBase(event_id)
self.sessions: dict[str, list] = {} # история диалогов
def answer(self, participant_id: str, question: str, current_time: datetime = None) -> str:
results = self.kb.vectorstore.similarity_search_with_score(question, k=5)
context = "\n\n".join([doc.page_content for doc, score in results if (1 - score) > 0.5])
time_context = f"Текущее время: {current_time.strftime('%H:%M')}" if current_time else ""
history = self.sessions.get(participant_id, [])
messages = history + [{
"role": "user",
"content": f"{time_context}\n\nВопрос участника: {question}\n\nИнформация о мероприятии:\n{context}"
}]
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=512,
system=f"""Ты — ассистент участников мероприятия «{self.event_name}».
Отвечай кратко и конкретно. Если информации нет — скажи честно.""",
messages=messages,
)
answer = response.content[0].text
history.append({"role": "user", "content": question})
history.append({"role": "assistant", "content": answer})
self.sessions[participant_id] = history[-10:]
return answer
class QAModerator:
"""AI-модерация вопросов Q&A сессии"""
def __init__(self):
self.question_queue: list[dict] = []
self.answered: list[dict] = []
def moderate_question(self, question: str, participant: dict) -> dict:
"""Модерирует вопрос: релевантность, дубликаты, качество"""
# Проверяем на дубликат
if self.question_queue or self.answered:
existing = [q["question"] for q in self.question_queue + self.answered[-20:]]
existing_text = "\n".join(f"- {q}" for q in existing[-15:])
else:
existing_text = "нет"
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=256,
messages=[{
"role": "user",
"content": f"""Оцени вопрос для Q&A сессии:
Вопрос: "{question}"
Уже есть вопросы: {existing_text}
Верни JSON:
{{
"approve": true/false,
"is_duplicate": true/false,
"duplicate_of": "похожий вопрос или null",
"quality_score": 1-5,
"category": "technical|business|personal|off_topic",
"cleaned_question": "отредактированная версия (убрать грубость, опечатки)",
"reject_reason": "причина отклонения или null"
}}
Одобряй вопросы по теме, не спам и не грубость. Только JSON."""
}],
)
text = response.content[0].text
result = json.loads(text[text.find("{"):text.rfind("}") + 1])
if result.get("approve"):
self.question_queue.append({
"question": result.get("cleaned_question", question),
"original": question,
"participant": participant,
"quality": result.get("quality_score", 3),
"category": result.get("category"),
"submitted_at": datetime.now().isoformat(),
})
return result
def get_top_questions(self, n: int = 5) -> list[dict]:
"""Возвращает топ вопросов по оценке качества"""
sorted_queue = sorted(
self.question_queue,
key=lambda q: q.get("quality", 3),
reverse=True
)
return sorted_queue[:n]
def get_summary_for_speaker(self, session_topic: str) -> str:
"""Готовит спикеру сводку вопросов перед Q&A"""
if not self.question_queue:
return "Вопросов пока нет."
questions_text = "\n".join([
f"{i+1}. [{q['category']}] {q['question']}"
for i, q in enumerate(self.get_top_questions(10))
])
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=512,
messages=[{
"role": "user",
"content": f"""Тема сессии: {session_topic}
Вопросы от аудитории:
{questions_text}
Сгруппируй по темам, выдели наиболее частые — помоги спикеру подготовиться к Q&A.
Кратко, по делу."""
}],
)
return response.content[0].text
class EventEmailAutomation:
"""Автоматические рассылки для мероприятия"""
EMAIL_TYPES = {
"reminder_24h": "Напоминание за 24 часа",
"reminder_1h": "Напоминание за 1 час",
"welcome": "Приветствие при подключении",
"session_followup": "Follow-up после секции",
"post_event": "Итоговое письмо",
"recording_ready": "Запись доступна",
}
def __init__(self, event_data: dict):
self.event = event_data
def generate_email(
self,
email_type: str,
participant: dict,
session: dict = None,
extra_data: dict = {},
) -> dict:
"""Генерирует персонализированное письмо"""
context = f"""
Мероприятие: {self.event['name']}
Дата: {self.event['date']}
Участник: {participant['name']}, {participant.get('company', '')} ({participant.get('job_title', '')})
Интересы: {', '.join(participant.get('interests', []))}
"""
if session:
context += f"\nСекция: {session['title']}\nСпикер: {session['speaker']}"
context += f"\nДоп. данные: {json.dumps(extra_data, ensure_ascii=False)}"
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""Напиши {self.EMAIL_TYPES[email_type]} для участника конференции.
Контекст: {context}
Формат: {{"subject": "тема письма", "body_html": "HTML текст письма", "body_text": "plain text"}}
Тон: профессиональный, персонализированный.
Только JSON."""
}],
)
text = response.content[0].text
return json.loads(text[text.find("{"):text.rfind("}") + 1])
Аналитика вовлечённости в реальном времени
class EngagementAnalytics:
"""Аналитика вовлечённости участников"""
def __init__(self):
self.events_log: list[dict] = []
def log_event(self, event_type: str, participant_id: str, data: dict = {}):
self.events_log.append({
"type": event_type, # view, question, reaction, leave, join
"participant_id": participant_id,
"timestamp": datetime.now().isoformat(),
**data
})
def get_session_insights(self, session_id: str) -> str:
"""AI-анализ вовлечённости по сессии"""
session_events = [e for e in self.events_log if e.get("session_id") == session_id]
stats = {
"total_views": len([e for e in session_events if e["type"] == "view"]),
"questions_asked": len([e for e in session_events if e["type"] == "question"]),
"reactions": len([e for e in session_events if e["type"] == "reaction"]),
"avg_watch_time_min": sum(e.get("watch_time", 0) for e in session_events) / max(len(session_events), 1) / 60,
"drop_off_points": sorted(
[e.get("timestamp", "") for e in session_events if e["type"] == "leave"]
)[:5],
}
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=512,
messages=[{
"role": "user",
"content": f"Проанализируй вовлечённость аудитории по данным: {json.dumps(stats, ensure_ascii=False)}\nДай краткие выводы и рекомендации для следующего мероприятия."
}],
)
return response.content[0].text
Практический кейс: IT-конференция, 800 участников
Масштаб: 2-дневная онлайн-конференция, 22 секции, 800 зарегистрированных участников, 3 параллельных трека.
Что автоматизировали:
- Q&A чат-бот отвечал на вопросы о программе, спикерах и технических вопросах подключения (78% вопросов)
- Модерация Q&A: 340 вопросов за 2 дня, спикеры получали топ-5 самых качественных каждые 20 минут
- Персонализированные follow-up письма по 3 трекам (технический, бизнес, менеджмент)
- Напоминания за 24ч и 1ч: +18% к attendance rate vs предыдущей конференции без напоминаний
Результаты:
- Оргкоманда (3 человека): сэкономили ~40 часов операционной работы за 2 дня
- Время ответа на вопрос участника: 30 мин → 90 сек
- Отчёт по вовлечённости каждой секции: готов через 5 мин после окончания
Сроки
- AI-ассистент для участников + Q&A модерация: 1–2 недели
- Email-автоматизация (персонализированные рассылки): 1 неделя
- Аналитика вовлечённости: 3–5 дней
- Полная система для крупной конференции: 4–6 недель