Diarización de audio con inteligencia de reuniones
Muchas organizaciones ya graban conversaciones importantes, pero una transcripción simple a menudo no es suficiente para un seguimiento confiable. La capa que falta es la atribución del orador: saber quién planteó una preocupación, quién hizo un compromiso y dónde aparece la evidencia en la grabación. Una transcripción con reconocimiento de orador te permite separar las necesidades del cliente del seguimiento del vendedor, mantener referencias de evidencia estructuradas junto a los elementos de acción y dirigir los compromisos sensibles a un flujo de trabajo de revisión antes de que lleguen a un CRM, sistema de tickets o base de conocimientos.
Este patrón es útil siempre que el flujo de trabajo posterior dependa de quién dijo qué:
- Descubrimiento de ventas y consultoría de soluciones: captura los requisitos del cliente, los compromisos del vendedor, los criterios de decisión, los bloqueadores y los próximos pasos con evidencia.
- Éxito del cliente y gestión de cuentas: convierte las QBR, las llamadas de renovación y las sesiones de incorporación en riesgos identificados, solicitudes de productos y planes de seguimiento.
- Escaladas de soporte y revisiones de incidentes: conserva la línea de tiempo, los síntomas reportados, los compromisos del propietario y las preguntas sin resolver antes de crear tickets o postmortems.
- Reclutamiento y ciclos de entrevistas: resume los comentarios del candidato o del entrevistador manteniendo las citas vinculadas al orador correcto.
- Revisiones reguladas o de alto riesgo: agrega redacción, verificaciones de evidencia y revisión humana antes de almacenar notas de conversaciones de atención médica, servicios financieros, legales o con mucha carga de cumplimiento.
Para los equipos de ingresos, el impacto suele ser menos sobre generar otro resumen y más sobre reducir la fuga entre la conversación y el sistema de registro. Un pipeline como este puede ayudar a los equipos a capturar los próximos pasos listos para CRM más rápido, identificar riesgos de renovación o expansión antes, preservar la evidencia detrás de las actualizaciones de pronóstico y capacitar a los representantes o equipos de soporte a partir de ejemplos documentados en lugar de notas anecdóticas. El objetivo no es automatizar el juicio; es hacer que las transferencias, revisiones y acciones de seguimiento sean más completas y auditables.
Este notebook muestra cómo construir un pipeline de inteligencia de reuniones post-llamada de estilo de producción con diarización de audio de OpenAI. Tú:
- Aceptarás un archivo de audio de una reunión grabada.
- Opcionalmente, mapearás oradores conocidos usando clips de referencia cortos.
- Llamarás a
gpt-4o-transcribe-diarizeconresponse_format="diarized_json". - Normalizarás los segmentos etiquetados por orador en JSON y Markdown con IDs de segmento estables.
- Usarás salidas estructuradas para extraer un resumen de la reunión, decisiones, riesgos, preguntas explícitas, seguimientos sugeridos, elementos de acción, referencias de evidencia y un borrador de correo electrónico de seguimiento.
- Escribirás artefactos revisables y un informe de protección local.
Las celdas predeterminadas se ejecutan sin una clave API utilizando una transcripción diarizada sintética. Las llamadas de audio reales son opcionales, por lo que el notebook es seguro de revisar de principio a fin.
Arquitectura
| Capa | Responsabilidad | Salida |
|---|---|---|
| Ingesta de audio | Acepta una grabación de llamada y clips opcionales de oradores conocidos. | meeting.wav, Agent=agent.wav |
| Ejecutor de pipeline | Valida entradas, codifica referencias como URLs de datos, llama a OpenAI y escribe artefactos. | Metadatos de ejecución y directorio de salida |
| Diarización | Llama a gpt-4o-transcribe-diarize con response_format="diarized_json" y chunking_strategy="auto". |
Segmentos etiquetados por orador |
| Normalización de transcripción | Convierte la salida de la API en JSON y Markdown consistentes con IDs de segmento estables. | transcript_segments.json, speaker_labeled_transcript.md |
| Inteligencia de reuniones | Extrae resumen, decisiones, acciones, riesgos, preguntas explícitas, seguimientos sugeridos, citas y correo electrónico de seguimiento con referencias de evidencia estructuradas. | meeting_intelligence.json, meeting_brief.md |
| Protecciones y puerta de revisión | Redacta campos sensibles, verifica referencias de evidencia, modera opcionalmente el contenido y dirige las salidas riesgosas para revisión. | guardrail_report.json |
Esto es intencionalmente basado en solicitudes. La API en tiempo real es más adecuada para UX de voz en vivo, captura de navegador o transmisión telefónica. Para la diarización duradera post-llamada, este patrón utiliza la API de transcripciones y luego ejecuta la extracción estructurada sobre la transcripción etiquetada por orador.
Por qué importan las transcripciones con reconocimiento de orador
La primera versión de la inteligencia de reuniones a menudo es "enviar una transcripción a un modelo y resumirla". Eso funciona para demostraciones, pero falla en los flujos de trabajo del cliente porque pierde quién dijo qué. Un cliente puede establecer un requisito, un vendedor puede hacer un compromiso y un gerente puede necesitar que la diferencia sea explícita.
La diarización con reconocimiento de orador le da al resto de la aplicación una mejor estructura:
- Los elementos de acción pueden incluir al orador que se comprometió con ellos.
- Los riesgos pueden citar la preocupación exacta del cliente.
- Los borradores de correo electrónico de seguimiento pueden evitar atribuir los compromisos del vendedor al cliente.
- Los revisores de QA pueden verificar la atribución del orador por ID de segmento y marca de tiempo.
- Los trabajos de sincronización de CRM pueden almacenar evidencia verificable mecánicamente en lugar de resúmenes opacos.
Seguridad y protecciones
La inteligencia de reuniones debe tratarse como un flujo de trabajo de datos sensibles, no solo como una tarea de transcripción o resumen. Las grabaciones sin procesar pueden contener nombres de clientes, términos comerciales, detalles de soporte, información de salud o financiera y estrategia interna. Los clips de referencia de orador también pueden ser sensibles porque están vinculados a la voz de una persona. Una vez que el pipeline convierte ese audio en salidas estructuradas, esas salidas pueden fluir a registros de CRM, tickets de soporte, planes de cuenta, paneles o colas de revisión.
La seguridad y las protecciones son más importantes cuando la salida puede influir en un proceso de negocio. Un resumen de una llamada de ventas puede capturar precios o compromisos contractuales. Una escalada de soporte puede incluir incidentes que afectan la producción o credenciales de clientes. Una sesión informativa de reclutamiento puede incluir comentarios de candidatos. Una reunión de la industria regulada puede contener datos que necesitan políticas de retención, control de acceso o redacción. Para estos flujos de trabajo, el patrón más seguro es minimizar la retención de audio sin procesar, redactar el contenido sensible cuando sea apropiado, requerir salidas respaldadas por evidencia y dirigir las salidas riesgosas o de baja confianza a través de una revisión humana antes de las escrituras posteriores.
La redacción de expresiones regulares incluida es intencionalmente ilustrativa: solo enmascara patrones básicos de correo electrónico y teléfono. No es un sistema completo de PII o DLP. Para nombres, direcciones, identificadores de cuenta, credenciales, datos de salud, datos financieros o flujos de trabajo regulados, usa un detector de PII/DLP aprobado por la política y mantén una puerta de revisión humana antes de las escrituras posteriores.
Para el paso de extracción estructurada, este notebook establece store=False en la llamada a la API de respuestas para que la respuesta de inteligencia de reuniones generada no se almacene como estado de la aplicación. store=False es un control útil a nivel de solicitud, pero no es lo mismo que habilitar la retención de datos cero para una organización o proyecto. Si tu flujo de trabajo requiere garantías de retención más estrictas, revisa la documentación de controles de datos de OpenAI y confirma la configuración de retención correcta para tu caso de uso.
| Riesgo | Protección |
|---|---|
| Mal uso de la grabación o referencia del orador | Requiere consentimiento y aprobación de la política antes de grabar, diarizar o usar clips de referencia. Trata las referencias de orador como datos sensibles adyacentes a la biometría. |
| Retención excesiva de audio sin procesar | No guardes la respuesta de transcripción sin procesar por defecto. Conserva el audio sin procesar y los clips de referencia solo el tiempo necesario. Cifra y restringe el acceso si se retienen. |
| Inyección de prompt dentro de las transcripciones | Trata el texto de la transcripción como evidencia no confiable. Mantén las instrucciones en el mensaje del sistema y exige que el modelo use solo hechos respaldados por la transcripción. |
| Elementos de acción o decisiones no respaldados | Usa salidas estructuradas estrictas y requiere referencias de evidencia que apunten a IDs de segmento y citas reales. |
| Contenido sensible en notas generadas | Ejecuta la redacción antes del resumen cuando sea posible, luego ejecuta verificaciones posteriores a la generación en la transcripción y el resumen. |
| Contenido dañino o sensible a la política | Opcionalmente, llama a la API de Moderación con omni-moderation-latest en el texto de la transcripción y el texto del resumen generado. La moderación detecta contenido dañino; no es un reemplazo para la revisión de privacidad. |
| Escrituras posteriores inseguras | No escribas directamente en sistemas de CRM, tickets o análisis desde la salida del modelo. Coloca una puerta de revisión humana frente a riesgos medios/altos, evidencia faltante, banderas de moderación o retención de respuesta sin procesar. |
| Deriva de calidad silenciosa | Registra las versiones del modelo, las versiones del prompt, las versiones del esquema, la duración del audio, el estado de redacción, el estado de moderación y las decisiones del revisor. Muestra llamadas para evaluaciones. |
Prerrequisitos
- Python 3.10 o posterior.
- Una clave API de OpenAI en
OPENAI_API_KEYpara ejecuciones de audio reales. - Una grabación de reunión en un formato de audio compatible para ejecuciones de audio reales.
- Las cargas de audio deben ser de 25 MB o menos. Los formatos de entrada compatibles son
mp3,mp4,mpeg,mpga,m4a,wavywebm. - Opcional: hasta cuatro clips de referencia cortos de un solo orador. La guía de voz a texto recomienda referencias de 2 a 10 segundos, codificadas como URLs de datos cuando se envían con datos de formulario multipart.
Ejecuta el notebook localmente
Desde un clon local del repositorio de Cookbook, crea un entorno virtual, instala Jupyter y el SDK de OpenAI, luego inicia este notebook:
git clone https://github.com/openai/openai-cookbook.git
cd openai-cookbook
python3 -m venv .venv
source .venv/bin/activate
python -m pip install jupyter "openai>=1.93.0"
export OPENAI_API_KEY="your-api-key"
jupyter notebook examples/audio/speaker_aware_meeting_intelligence/speaker_aware_meeting_intelligence.ipynb
La demostración sintética a continuación usa solo la biblioteca estándar de Python y no llama a la API. Para audio real, también puedes instalar el SDK de OpenAI desde dentro de un entorno de notebook existente:
%pip install "openai>=1.93.0"
Solicitud de diarización central
La solicitud central de la API es intencionalmente pequeña:
client = OpenAI(timeout=30 * 60)
with open("meeting.wav", "rb") as audio_file:
stream = client.audio.transcriptions.create(
model="gpt-4o-transcribe-diarize",
file=audio_file,
response_format="diarized_json",
chunking_strategy="auto",
stream=True,
extra_body={
"known_speaker_names": ["Agent"],
"known_speaker_references": [to_data_url(Path("agent_reference.wav"))],
},
)
for event in stream:
if event.type == "transcript.text.segment":
print(event.speaker, event.text, event.start, event.end)
Los detalles importantes son:
- Usa
response_format="diarized_json"cuando necesites metadatos de orador a nivel de segmento. - Usa
chunking_strategy="auto"para audio de más de 30 segundos. - Usa
stream=Truepara grabaciones completadas cuando quieras segmentos diarizados finalizados a medida que estén disponibles. - El SDK de Python tiene un tiempo de espera de lectura predeterminado de 10 minutos; el asistente a continuación usa 30 minutos para grabaciones más largas.
- Pasa los nombres de oradores conocidos y las referencias juntos, en el mismo orden.
- Mantén los clips de referencia cortos y de un solo orador.
Diarización vs. identificación de orador
La diarización responde "¿qué voz habló cada segmento?". Separa voces dentro de una grabación, pero no crea un perfil de identidad permanente ni recuerda que speaker_0 de una llamada es la misma persona que speaker_0 en una llamada posterior. Sin referencias, las etiquetas genéricas siguen siendo útiles porque preservan la atribución: el pipeline puede distinguir al orador que planteó un requisito del orador que hizo un compromiso.
Las referencias de oradores conocidos añaden una pista de identidad opcional para la solicitud actual:
| Entrada | Resultado |
|---|---|
| Solo audio de la reunión | El modelo separa las voces, generalmente con etiquetas genéricas como speaker_0 y speaker_1. |
| Audio de la reunión más un clip de referencia con nombre | Los segmentos coincidentes pueden usar el nombre proporcionado; los oradores no coincidentes pueden permanecer genéricos. |
| Una grabación posterior o histórica | Pasa el clip de referencia de nuevo. Las etiquetas no se transfieren automáticamente entre grabaciones. |
Pasa un clip de referencia con la grabación de la reunión
La grabación de la reunión y el clip de referencia son entradas separadas en una solicitud de transcripción. No concatenes el clip de referencia al audio de la reunión. La reunión se sube como file=...; cada clip de referencia se codifica como una URL de datos y se envía a través de known_speaker_references con un nombre en la misma posición en known_speaker_names.
El asistente a continuación envuelve esa forma de solicitud. Por ejemplo, esto pasa una grabación de reunión más un clip corto separado de un representante interno hablando:
meeting_audio = Path("customer_call.wav")
known_speakers = [
("Internal rep", Path("internal_rep_reference.wav")),
]
raw_transcription = transcribe_with_diarization(
audio_file=meeting_audio,
known_speakers=known_speakers,
)
Usa un clip limpio y consentido de 2 a 10 segundos con un solo orador y ruido de fondo mínimo. Para oradores internos recurrentes, una aplicación de producción puede mantener un registro de referencias con control de acceso y adjuntar el clip apropiado en cada solicitud. Para grabaciones históricas, ejecuta el mismo flujo por grabación; un clip de referencia puede provenir de una llamada anterior consentida si es limpio y de un solo orador. Trata las referencias como datos sensibles, evalúa la calidad de la coincidencia en audio representativo y mantén la revisión humana para escrituras posteriores de alto riesgo.
from __future__ import annotations
import base64
import json
import mimetypes
import os
import re
import tempfile
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
try:
from IPython.display import JSON, Markdown, display
except ImportError: # Makes this cell safe in non-notebook runners.
JSON = None
Markdown = None
def display(value):
print(value)
def show_markdown(text: str) -> None:
if Markdown:
display(Markdown(text))
else:
print(text)
def show_json(payload: Any, expanded: bool = False) -> None:
if JSON:
display(JSON(payload, expanded=expanded))
else:
print(json.dumps(payload, indent=2))
DEFAULT_TRANSCRIPTION_MODEL = "gpt-4o-transcribe-diarize"
DEFAULT_SUMMARY_MODEL = os.getenv("OPENAI_MEETING_INTELLIGENCE_MODEL", "gpt-4.1-mini")
DEFAULT_MODERATION_MODEL = "omni-moderation-latest"
SUPPORTED_REFERENCE_MIME_PREFIXES = ("audio/", "video/")
MAX_AUDIO_UPLOAD_BYTES = 25_000_000
DEFAULT_TRANSCRIPTION_TIMEOUT_SECONDS = 30 * 60
print("Notebook helpers loaded")
@dataclass(frozen=True)
class Segment:
segment_id: str
speaker: str
start: float
end: float
text: str
DEMO_SEGMENTS = [
Segment(
segment_id="seg_001",
speaker="Solutions Engineer",
start=0.0,
end=9.2,
text="Thanks for joining. I would like to understand where your support handoff breaks down today.",
),
Segment(
segment_id="seg_002",
speaker="Customer",
start=9.3,
end=22.4,
text="The biggest issue is that escalation notes are inconsistent. Managers spend Monday morning reconstructing what happened from call recordings.",
),
Segment(
segment_id="seg_003",
speaker="Solutions Engineer",
start=22.5,
end=38.1,
text="So the priority is reliable call summaries, who committed to what, and enough evidence that the team trusts the handoff.",
),
Segment(
segment_id="seg_004",
speaker="Customer",
start=38.2,
end=55.0,
text="Exactly. We also need risks called out, especially compliance-sensitive promises, and we need to push action items into our CRM.",
),
Segment(
segment_id="seg_005",
speaker="Solutions Engineer",
start=55.1,
end=70.3,
text="I will send a prototype that includes speaker-aware transcripts, action items with evidence, and a redaction pass before CRM sync.",
),
]
PII_PATTERNS = [
(re.compile(r"\b[\w.+-]+@[\w-]+(?:\.[\w-]+)+\b"), "[email]"),
(re.compile(r"\b(?:\+?1[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?)\d{3}[-.\s]?\d{4}\b"), "[phone]"),
]
print(f"Loaded {len(DEMO_SEGMENTS)} synthetic transcript segments")
Paso 1: Define el esquema de salida estructurada
La inteligencia de reuniones a menudo alimenta sistemas de registro. Usa salidas estructuradas estrictas para que el código posterior obtenga una forma estable y los campos no admitidos sean rechazados en lugar de aceptados silenciosamente. Este esquema también requiere evidence_refs estructuradas: cada elemento extraído debe citar un segment_id de transcripción y una cita de ese segmento, lo que permite que las protecciones verifiquen la fundamentación mecánicamente.
EVIDENCE_REF_SCHEMA: dict[str, Any] = {
"type": "object",
"additionalProperties": False,
"properties": {
"segment_id": {"type": "string"},
"quote": {"type": "string"},
},
"required": ["segment_id", "quote"],
}
EVIDENCE_REFS_SCHEMA: dict[str, Any] = {
"type": "array",
"items": EVIDENCE_REF_SCHEMA,
}
NULLABLE_STRING_SCHEMA: dict[str, Any] = {"type": ["string", "null"]}
MEETING_INTELLIGENCE_SCHEMA: dict[str, Any] = {
"name": "meeting_intelligence",
"strict": True,
"schema": {
"type": "object",
"additionalProperties": False,
"properties": {
"summary": {"type": "string"},
"participants": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"speaker": {"type": "string"},
"inferred_role": NULLABLE_STRING_SCHEMA,
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["speaker", "inferred_role", "evidence_refs"],
},
},
"customer_context": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"fact": {"type": "string"},
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["fact", "evidence_refs"],
},
},
"decisions": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"decision": {"type": "string"},
"speaker_or_group": NULLABLE_STRING_SCHEMA,
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["decision", "speaker_or_group", "evidence_refs"],
},
},
"action_items": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"owner_speaker": NULLABLE_STRING_SCHEMA,
"task": {"type": "string"},
"due_date_or_trigger": NULLABLE_STRING_SCHEMA,
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["owner_speaker", "task", "due_date_or_trigger", "evidence_refs"],
},
},
"risks": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"risk": {"type": "string"},
"severity": {"type": "string", "enum": ["low", "medium", "high"]},
"evidence_refs": EVIDENCE_REFS_SCHEMA,
"mitigation": {"type": "string"},
},
"required": ["risk", "severity", "evidence_refs", "mitigation"],
},
},
"explicit_questions": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"question": {"type": "string"},
"asked_by_speaker": {"type": "string"},
"directed_to_speaker": NULLABLE_STRING_SCHEMA,
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["question", "asked_by_speaker", "directed_to_speaker", "evidence_refs"],
},
},
"suggested_follow_ups": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"question": {"type": "string"},
"rationale": {"type": "string"},
"evidence_refs": EVIDENCE_REFS_SCHEMA,
},
"required": ["question", "rationale", "evidence_refs"],
},
},
"notable_quotes": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"speaker": {"type": "string"},
"quote": {"type": "string"},
"timestamp": {"type": "string"},
"segment_id": {"type": "string"},
},
"required": ["speaker", "quote", "timestamp", "segment_id"],
},
},
"follow_up_email": {
"type": "object",
"additionalProperties": False,
"properties": {
"subject": {"type": "string"},
"body": {"type": "string"},
},
"required": ["subject", "body"],
},
},
"required": [
"summary",
"participants",
"customer_context",
"decisions",
"action_items",
"risks",
"explicit_questions",
"suggested_follow_ups",
"notable_quotes",
"follow_up_email",
],
},
}
print("Structured output schema ready")
Paso 2: Construye ayudantes de audio y transcripción
Las referencias de oradores conocidos son opcionales. Sin ellas, la diarización aún puede separar a los oradores, pero las etiquetas pueden ser genéricas, como speaker_0 o speaker_1. Con las referencias, la API puede mapear segmentos a los nombres que tú proporcionas.
Usa clips de referencia cortos y limpios con un solo orador y ruido de fondo mínimo. Conserva los clips de referencia solo cuando tengas consentimiento y una necesidad comercial clara.
def to_data_url(path) -> str:
path = Path(path)
mime_type, _ = mimetypes.guess_type(path)
if mime_type is None:
mime_type = "audio/wav"
elif not mime_type.startswith(SUPPORTED_REFERENCE_MIME_PREFIXES):
raise ValueError(f"Reference clip must be an audio or video file, got {mime_type}: {path}")
encoded = base64.b64encode(path.read_bytes()).decode("utf-8")
return f"data:{mime_type};base64,{encoded}"
def transcribe_with_diarization(
audio_file: Path,
known_speakers: list[tuple[str, Path]],
model: str = DEFAULT_TRANSCRIPTION_MODEL,
request_timeout_seconds: float = DEFAULT_TRANSCRIPTION_TIMEOUT_SECONDS,
stream_transcription: bool = True,
) -> Any:
if not audio_file.is_file():
raise FileNotFoundError(f"Audio file does not exist or is not a regular file: {audio_file}")
if request_timeout_seconds <= 0:
raise ValueError("request_timeout_seconds must be positive.")
audio_size_bytes = audio_file.stat().st_size
if audio_size_bytes > MAX_AUDIO_UPLOAD_BYTES:
raise ValueError(
f"Audio file is {audio_size_bytes:,} bytes; "
"the Audio Transcriptions API accepts uploads up to 25 MB. "
"Compress or split the recording before retrying."
)
from openai import OpenAI
client = OpenAI(timeout=request_timeout_seconds)
params: dict[str, Any] = {
"model": model,
"response_format": "diarized_json",
"chunking_strategy": "auto",
"stream": stream_transcription,
}
if known_speakers:
if len(known_speakers) > 4:
raise ValueError("gpt-4o-transcribe-diarize accepts up to 4 known speaker references.")
params["extra_body"] = {
"known_speaker_names": [name for name, _ in known_speakers],
"known_speaker_references": [to_data_url(path) for _, path in known_speakers],
}
with audio_file.open("rb") as audio:
response = client.audio.transcriptions.create(file=audio, **params)
if stream_transcription:
return collect_streamed_transcription(response)
return response
def to_plain(value: Any) -> Any:
if hasattr(value, "model_dump"):
return value.model_dump()
if isinstance(value, dict):
return {key: to_plain(inner) for key, inner in value.items()}
if isinstance(value, list):
return [to_plain(item) for item in value]
return value
def collect_streamed_transcription(events: Any) -> dict[str, Any]:
segments: list[dict[str, Any]] = []
full_text = ""
usage: Any = None
for event in events:
data = to_plain(event)
if not isinstance(data, dict):
continue
event_type = data.get("type")
if event_type == "transcript.text.segment":
segments.append(data)
elif event_type == "transcript.text.done":
full_text = str(data.get("text") or "")
usage = data.get("usage")
if not segments:
raise ValueError("No diarized transcript segments were emitted by the transcription stream.")
return {"segments": segments, "text": full_text, "usage": usage}
def normalize_segments(transcription: Any) -> list[Segment]:
data = to_plain(transcription)
raw_segments = data.get("segments", []) if isinstance(data, dict) else []
segments: list[Segment] = []
for index, item in enumerate(raw_segments):
if hasattr(item, "model_dump"):
item = item.model_dump()
if not isinstance(item, dict):
continue
text = str(item.get("text", "")).strip()
if not text:
continue
segment_id = str(item.get("segment_id") or item.get("id") or f"seg_{len(segments) + 1:03d}")
segments.append(
Segment(
segment_id=segment_id,
speaker=str(item.get("speaker") or f"Speaker {index + 1}"),
start=float(item.get("start") or 0.0),
end=float(item.get("end") or 0.0),
text=text,
)
)
if not segments and isinstance(data, dict) and data.get("text"):
segments.append(Segment(segment_id="seg_001", speaker="Speaker 1", start=0.0, end=0.0, text=str(data["text"])))
if not segments:
raise ValueError("No transcript segments were found in the transcription response.")
return segments
print("Audio and transcript helpers ready")
Paso 3: Normaliza la transcripción
La transcripción normalizada es el contrato entre el procesamiento de audio y la inteligencia de reuniones. Te ayuda a volver a ejecutar el resumen sin retranscribir el audio, inspeccionar la calidad de la atribución y mantener la retención de audio sin procesar corta.
Cada segmento obtiene un segment_id estable como seg_005. Más tarde, el modelo debe citar esos IDs en evidence_refs, y el paso de protección verifica que cada cita citada aparezca en el segmento referenciado.
El ayudante de redacción de expresiones regulares a continuación es intencionalmente ilustrativo: solo enmascara patrones básicos de correo electrónico y teléfono. No es un sistema completo de PII o DLP; usa un detector aprobado por la política y revisión humana para flujos de trabajo sensibles o regulados.
def redact_text(text: str) -> str:
redacted = text
for pattern, replacement in PII_PATTERNS:
redacted = pattern.sub(replacement, redacted)
return redacted
def redact_segments(segments: list[Segment]) -> list[Segment]:
return [
Segment(
segment_id=segment.segment_id,
speaker=segment.speaker,
start=segment.start,
end=segment.end,
text=redact_text(segment.text),
)
for segment in segments
]
def pii_matches(text: str) -> list[str]:
matches: list[str] = []
for pattern, replacement in PII_PATTERNS:
if pattern.search(text):
matches.append(replacement.strip("[]"))
return sorted(set(matches))
def format_timestamp(seconds: float) -> str:
total_ms = max(0, int(round(seconds * 1000)))
minutes, remainder_ms = divmod(total_ms, 60_000)
secs, millis = divmod(remainder_ms, 1000)
return f"{minutes:02d}:{secs:02d}.{millis:03d}"
def transcript_as_markdown(segments: list[Segment]) -> str:
lines = ["# Speaker-Labeled Transcript", ""]
for segment in segments:
start = format_timestamp(segment.start)
end = format_timestamp(segment.end)
lines.append(f"**{segment.segment_id} | {segment.speaker} [{start}-{end}]**: {segment.text}")
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def transcript_for_model(segments: list[Segment]) -> str:
return "\n".join(
f"{segment.segment_id} | {segment.speaker} | {format_timestamp(segment.start)}-{format_timestamp(segment.end)} | {segment.text}"
for segment in segments
)
show_markdown(transcript_as_markdown(DEMO_SEGMENTS))
Paso 4: Extrae inteligencia de reuniones estructurada
El modelo recibe una transcripción etiquetada por orador y debe usar solo esa transcripción como evidencia. El valor predeterminado más seguro es producir arreglos vacíos en lugar de notas de CRM plausibles pero no respaldadas. El esquema también usa campos requeridos pero anulables, como due_date_or_trigger, inferred_role y directed_to_speaker, para que los valores desconocidos permanezcan null en lugar de llenarse con suposiciones.
Para cada hecho extraído, elemento de acción, riesgo, pregunta o recomendación, el modelo devuelve evidence_refs con un segment_id y una cita. Esto les da a los revisores un rastro de origen legible y al código algo concreto para validar.
def response_output_text_or_raise(response: Any) -> str:
data = to_plain(response)
status = data.get("status") if isinstance(data, dict) else getattr(response, "status", None)
if status and status != "completed":
details = data.get("incomplete_details") if isinstance(data, dict) else getattr(response, "incomplete_details", None)
raise RuntimeError(f"Responses API returned status={status!r}; incomplete_details={details!r}")
refusals: list[str] = []
if isinstance(data, dict):
for item in data.get("output", []):
if not isinstance(item, dict):
continue
for content in item.get("content", []):
if not isinstance(content, dict):
continue
if content.get("type") == "refusal" or content.get("refusal"):
refusals.append(str(content.get("refusal") or content.get("text") or content))
if refusals:
raise RuntimeError(f"Responses API returned a refusal: {refusals[0]}")
content = getattr(response, "output_text", None)
if content is None and isinstance(data, dict):
content = data.get("output_text")
content = str(content or "").strip()
if not content:
raise RuntimeError("The model returned an empty response.")
return content
def parse_meeting_intelligence_json(content: str) -> dict[str, Any]:
try:
parsed = json.loads(content)
except json.JSONDecodeError as exc:
raise RuntimeError(f"Responses API returned invalid JSON: {exc}") from exc
if not isinstance(parsed, dict):
raise RuntimeError("Responses API returned JSON, but the top-level value was not an object.")
return parsed
def generate_meeting_intelligence(segments: list[Segment], model: str = DEFAULT_SUMMARY_MODEL) -> dict[str, Any]:
from openai import OpenAI
client = OpenAI()
transcript = transcript_for_model(segments)
completion = client.responses.create(
model=model,
temperature=0,
store=False,
input=[
{
"role": "system",
"content": (
"You create meeting intelligence from speaker-labeled transcripts. "
"Use only the transcript as evidence. Do not invent names, dates, decisions, "
"commitments, or implementation details. If evidence is missing, leave the relevant array empty. "
"Use null for unknown roles, owners, due dates or triggers, directed-to speakers, or decision owners. "
"Put single-speaker commitments in action_items, not decisions. "
"Only include decisions when the transcript shows an explicit decision or agreement. "
"Put only questions actually asked in explicit_questions. "
"Put inferred next questions in suggested_follow_ups with rationale and evidence_refs. "
"Every extracted item must include evidence_refs with segment_id values copied from the transcript "
"and quote text copied from that same segment. Do not fabricate segment IDs or quotes. "
"Use empty arrays instead of unsupported items. "
"If the follow-up email signer is unknown, end with [Your name]."
),
},
{
"role": "user",
"content": (
"Extract a customer-safe meeting brief from this transcript. "
"Transcript rows use: segment_id | speaker | timestamp range | text.\n\n"
f"{transcript}"
),
},
],
text={
"format": {
"type": "json_schema",
"name": MEETING_INTELLIGENCE_SCHEMA["name"],
"strict": MEETING_INTELLIGENCE_SCHEMA["strict"],
"schema": MEETING_INTELLIGENCE_SCHEMA["schema"],
}
},
)
content = response_output_text_or_raise(completion)
return parse_meeting_intelligence_json(content)
def demo_meeting_intelligence() -> dict[str, Any]:
return {
"summary": (
"The customer needs a dependable post-call handoff process. Their main pain point is "
"inconsistent escalation notes, which forces managers to reconstruct calls manually. "
"The proposed path is a speaker-aware transcript, evidence-backed action items, risk "
"detection, redaction, and CRM sync."
),
"participants": [
{
"speaker": "Solutions Engineer",
"inferred_role": "OpenAI technical seller or solution owner",
"evidence_refs": [
{
"segment_id": "seg_001",
"quote": "I would like to understand where your support handoff breaks down today.",
},
{
"segment_id": "seg_005",
"quote": "I will send a prototype that includes speaker-aware transcripts, action items with evidence, and a redaction pass before CRM sync.",
},
],
},
{
"speaker": "Customer",
"inferred_role": "Customer stakeholder for support operations",
"evidence_refs": [
{
"segment_id": "seg_002",
"quote": "The biggest issue is that escalation notes are inconsistent.",
},
{
"segment_id": "seg_004",
"quote": "we need to push action items into our CRM.",
},
],
},
],
"customer_context": [
{
"fact": "Escalation notes are inconsistent today.",
"evidence_refs": [
{
"segment_id": "seg_002",
"quote": "The biggest issue is that escalation notes are inconsistent.",
}
],
},
{
"fact": "Managers spend time reconstructing calls from recordings.",
"evidence_refs": [
{
"segment_id": "seg_002",
"quote": "Managers spend Monday morning reconstructing what happened from call recordings.",
}
],
},
{
"fact": "The customer wants action items pushed into their CRM.",
"evidence_refs": [{"segment_id": "seg_004", "quote": "we need to push action items into our CRM."}],
},
],
"decisions": [],
"action_items": [
{
"owner_speaker": "Solutions Engineer",
"task": "Send a prototype that includes speaker-aware transcripts, action items with evidence, and a redaction pass before CRM sync.",
"due_date_or_trigger": None,
"evidence_refs": [
{
"segment_id": "seg_005",
"quote": "I will send a prototype that includes speaker-aware transcripts, action items with evidence, and a redaction pass before CRM sync.",
}
],
}
],
"risks": [
{
"risk": "Compliance-sensitive promises need to be identified in meeting notes.",
"severity": "medium",
"evidence_refs": [
{
"segment_id": "seg_004",
"quote": "We also need risks called out, especially compliance-sensitive promises",
}
],
"mitigation": "Route compliance-sensitive risks to human review before CRM sync.",
}
],
"explicit_questions": [
{
"question": "Where does your support handoff break down today?",
"asked_by_speaker": "Solutions Engineer",
"directed_to_speaker": "Customer",
"evidence_refs": [
{
"segment_id": "seg_001",
"quote": "I would like to understand where your support handoff breaks down today.",
}
],
}
],
"suggested_follow_ups": [
{
"question": "Which CRM object and fields should receive action items?",
"rationale": "The customer asked to push action items into their CRM but did not specify the target schema or workflow.",
"evidence_refs": [{"segment_id": "seg_004", "quote": "we need to push action items into our CRM."}],
}
],
"notable_quotes": [
{
"speaker": "Customer",
"quote": "Managers spend Monday morning reconstructing what happened from call recordings.",
"timestamp": "00:09.300",
"segment_id": "seg_002",
}
],
"follow_up_email": {
"subject": "Prototype for speaker-aware meeting handoffs",
"body": (
"Hi,\n\nThanks for the conversation. I heard that inconsistent escalation notes, "
"evidence-backed action items, compliance-sensitive risk detection, and CRM sync "
"are the core requirements. I will send a prototype with speaker-aware transcripts, "
"action items with evidence, and a redaction pass before CRM sync.\n\nBest,\n[Your name]"
),
},
}
show_json(demo_meeting_intelligence(), expanded=False)
Paso 5: Renderiza un resumen de reunión revisable
El artefacto de revisión mantiene al orador, el ID del segmento, la marca de tiempo y la evidencia de la cita junto a las decisiones, riesgos y elementos de acción para que los humanos puedan verificar antes de que se escriba algo en el sistema posterior.
def clean_markdown_cell(value: Any) -> str:
if value is None:
return "_Not specified._"
return str(value).replace("|", "\\|").replace("\n", "<br>")
def render_evidence_refs(refs: Any) -> str:
if not isinstance(refs, list) or not refs:
return "_No evidence refs._"
rendered = []
for ref in refs:
if not isinstance(ref, dict):
continue
segment_id = clean_markdown_cell(ref.get("segment_id", ""))
quote = clean_markdown_cell(ref.get("quote", ""))
rendered.append(f"`{segment_id}`: {quote}")
return "<br>".join(rendered) if rendered else "_No evidence refs._"
def markdown_table(rows: list[dict[str, Any]], columns: list[tuple[str, Any]]) -> str:
if not rows:
return "_None identified._"
header = "| " + " | ".join(title for title, _ in columns) + " |"
divider = "| " + " | ".join("---" for _ in columns) + " |"
body = []
for row in rows:
values = []
for _, key in columns:
value = key(row) if callable(key) else row.get(key, "")
values.append(clean_markdown_cell(value))
body.append("| " + " | ".join(values) + " |")
return "\n".join([header, divider, *body])
def render_meeting_brief(intelligence: dict[str, Any]) -> str:
follow_up = intelligence.get("follow_up_email", {})
evidence_column = ("Evidence", lambda row: render_evidence_refs(row.get("evidence_refs", [])))
lines = [
"# Meeting Brief",
"",
"## Summary",
"",
str(intelligence.get("summary", "")).strip() or "_No summary generated._",
"",
"## Participants",
"",
markdown_table(
intelligence.get("participants", []),
[("Speaker", "speaker"), ("Inferred role", "inferred_role"), evidence_column],
),
"",
"## Customer Context",
"",
markdown_table(intelligence.get("customer_context", []), [("Fact", "fact"), evidence_column]),
"",
"## Decisions",
"",
markdown_table(
intelligence.get("decisions", []),
[("Decision", "decision"), ("Owner", "speaker_or_group"), evidence_column],
),
"",
"## Action Items",
"",
markdown_table(
intelligence.get("action_items", []),
[
("Owner", "owner_speaker"),
("Task", "task"),
("Due date or trigger", "due_date_or_trigger"),
evidence_column,
],
),
"",
"## Risks",
"",
markdown_table(
intelligence.get("risks", []),
[("Risk", "risk"), ("Severity", "severity"), evidence_column, ("Mitigation", "mitigation")],
),
"",
"## Explicit Questions",
"",
markdown_table(
intelligence.get("explicit_questions", []),
[
("Question", "question"),
("Asked by", "asked_by_speaker"),
("Directed to", "directed_to_speaker"),
evidence_column,
],
),
"",
"## Suggested Follow-ups",
"",
markdown_table(
intelligence.get("suggested_follow_ups", []),
[("Question", "question"), ("Rationale", "rationale"), evidence_column],
),
"",
"## Notable Quotes",
"",
markdown_table(
intelligence.get("notable_quotes", []),
[("Speaker", "speaker"), ("Quote", "quote"), ("Timestamp", "timestamp"), ("Segment ID", "segment_id")],
),
"",
"## Follow-up Email Draft",
"",
f"**Subject:** {follow_up.get('subject', '')}",
"",
str(follow_up.get("body", "")).strip(),
"",
]
return "\n".join(lines).rstrip() + "\n"
demo_brief = render_meeting_brief(demo_meeting_intelligence())
show_markdown(demo_brief)
Paso 6: Agrega protecciones y escribe artefactos
La muestra escribe un guardrail_report.json con verificaciones locales para:
- segmentos de transcripción normalizados;
- patrones básicos de PII de correo electrónico y teléfono;
- referencias de evidencia que apuntan a IDs de segmento reales y citas coincidentes;
- salidas de riesgo medio/alto;
- banderas de moderación opcionales;
- almacenamiento de respuesta de transcripción sin procesar.
def summarize_moderation_response(response: Any) -> dict[str, Any]:
data = to_plain(response)
summaries: list[dict[str, Any]] = []
for result in data.get("results", []) if isinstance(data, dict) else []:
categories = result.get("categories", {}) if isinstance(result, dict) else {}
category_scores = result.get("category_scores", {}) if isinstance(result, dict) else {}
flagged_categories = sorted(key for key, value in categories.items() if bool(value))
top_scores = dict(sorted(category_scores.items(), key=lambda item: float(item[1] or 0.0), reverse=True)[:5])
summaries.append(
{
"flagged": bool(result.get("flagged")) if isinstance(result, dict) else False,
"flagged_categories": flagged_categories,
"top_category_scores": top_scores,
}
)
return {
"id": data.get("id") if isinstance(data, dict) else None,
"model": data.get("model") if isinstance(data, dict) else DEFAULT_MODERATION_MODEL,
"flagged": any(item["flagged"] for item in summaries),
"results": summaries,
}
def moderate_text(text: str, model: str = DEFAULT_MODERATION_MODEL) -> dict[str, Any]:
from openai import OpenAI
client = OpenAI()
response = client.moderations.create(model=model, input=text)
return summarize_moderation_response(response)
def iter_evidence_refs(value: Any, path: str = "$") -> list[tuple[str, Any]]:
found: list[tuple[str, Any]] = []
if isinstance(value, dict):
for key, inner in value.items():
next_path = f"{path}.{key}"
if key == "evidence_refs":
found.append((next_path, inner))
else:
found.extend(iter_evidence_refs(inner, next_path))
elif isinstance(value, list):
for index, item in enumerate(value):
found.extend(iter_evidence_refs(item, f"{path}[{index}]"))
return found
def normalize_for_quote_match(text: str) -> str:
return re.sub(r"\s+", " ", text).strip().casefold()
def validate_evidence_refs(intelligence: dict[str, Any], segments: list[Segment]) -> list[dict[str, Any]]:
segment_by_id = {segment.segment_id: segment for segment in segments}
problems: list[dict[str, Any]] = []
for path, refs in iter_evidence_refs(intelligence):
if not isinstance(refs, list) or not refs:
problems.append({"path": path, "issue": "missing_or_empty_evidence_refs"})
continue
for index, ref in enumerate(refs):
ref_path = f"{path}[{index}]"
if not isinstance(ref, dict):
problems.append({"path": ref_path, "issue": "evidence_ref_is_not_an_object"})
continue
segment_id = str(ref.get("segment_id", "")).strip()
quote = str(ref.get("quote", "")).strip()
if not segment_id or not quote:
problems.append({"path": ref_path, "issue": "missing_segment_id_or_quote", "segment_id": segment_id})
continue
segment = segment_by_id.get(segment_id)
if segment is None:
problems.append({"path": ref_path, "issue": "unknown_segment_id", "segment_id": segment_id})
continue
if normalize_for_quote_match(quote) not in normalize_for_quote_match(segment.text):
problems.append(
{
"path": ref_path,
"issue": "quote_not_found_in_segment",
"segment_id": segment_id,
"quote": quote,
}
)
for index, quote in enumerate(intelligence.get("notable_quotes", [])):
if not isinstance(quote, dict):
continue
segment_id = str(quote.get("segment_id", "")).strip()
quote_text = str(quote.get("quote", "")).strip()
segment = segment_by_id.get(segment_id)
if segment is None:
problems.append({"path": f"$.notable_quotes[{index}]", "issue": "unknown_segment_id", "segment_id": segment_id})
elif normalize_for_quote_match(quote_text) not in normalize_for_quote_match(segment.text):
problems.append(
{
"path": f"$.notable_quotes[{index}]",
"issue": "quote_not_found_in_segment",
"segment_id": segment_id,
"quote": quote_text,
}
)
return problems
def add_guardrail_check(checks: list[dict[str, Any]], name: str, status: str, detail: str, evidence = None) -> None:
check: dict[str, Any] = {"name": name, "status": status, "detail": detail}
if evidence is not None:
check["evidence"] = evidence
checks.append(check)
def build_guardrail_report(
segments: list[Segment],
intelligence: dict[str, Any],
meeting_brief: str,
redaction_enabled: bool,
raw_saved: bool,
moderation_results: dict[str, Any],
) -> dict[str, Any]:
checks: list[dict[str, Any]] = []
transcript_text = transcript_for_model(segments)
add_guardrail_check(
checks,
"transcript_segments_present",
"pass" if segments else "fail",
f"Found {len(segments)} normalized transcript segments.",
)
pii_found = pii_matches(transcript_text + "\n" + meeting_brief)
pii_detail = (
"Basic PII patterns remain after redaction."
if redaction_enabled
else "Basic PII patterns were detected; run with redaction or review before storage."
)
add_guardrail_check(
checks,
"basic_pii_scan",
"review" if pii_found else "pass",
pii_detail if pii_found else "No basic email or phone patterns detected.",
{"matches": pii_found, "redaction_enabled": redaction_enabled},
)
evidence_ref_problems = validate_evidence_refs(intelligence, segments)
add_guardrail_check(
checks,
"evidence_refs",
"review" if evidence_ref_problems else "pass",
(
"Some evidence references are missing, cite unknown segments, or quote text that is not present in the cited segment."
if evidence_ref_problems
else "All evidence references point to real segments with matching quote text."
),
{"problem_count": len(evidence_ref_problems), "examples": evidence_ref_problems[:5]},
)
risks = intelligence.get("risks", [])
review_risks = [risk for risk in risks if str(risk.get("severity", "")).lower() in {"medium", "high"}]
severity_counts = {
"low": sum(1 for risk in risks if str(risk.get("severity", "")).lower() == "low"),
"medium": sum(1 for risk in risks if str(risk.get("severity", "")).lower() == "medium"),
"high": sum(1 for risk in risks if str(risk.get("severity", "")).lower() == "high"),
}
add_guardrail_check(
checks,
"risk_outputs",
"review" if review_risks else "pass",
"Medium or high risks should be reviewed before downstream writes." if review_risks else "No medium or high risks identified.",
{"severity_counts": severity_counts},
)
moderation_flagged = [name for name, result in moderation_results.items() if isinstance(result, dict) and result.get("flagged")]
if moderation_results:
add_guardrail_check(
checks,
"moderation",
"review" if moderation_flagged else "pass",
"Moderation flagged content that should be reviewed." if moderation_flagged else "Moderation did not flag transcript or brief content.",
{"flagged_artifacts": moderation_flagged},
)
else:
add_guardrail_check(checks, "moderation", "not_run", "Moderation was not requested. Use moderation for content safety classification.")
add_guardrail_check(
checks,
"raw_response_storage",
"review" if raw_saved else "pass",
"Raw transcription response was saved; confirm retention and access controls." if raw_saved else "Raw transcription response was not saved.",
)
status = "review_required" if any(check["status"] in {"review", "fail"} for check in checks) else "pass"
if any(check["status"] == "fail" for check in checks):
status = "fail"
return {
"status": status,
"recommended_next_step": "Send artifacts to human review before downstream writes." if status != "pass" else "Artifacts passed local guardrail checks.",
"checks": checks,
"moderation": moderation_results,
}
print("Guardrail helpers ready")
def write_json(path: Path, payload: Any) -> None:
path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
def write_artifacts(
output_dir: Path,
segments: list[Segment],
intelligence: dict[str, Any],
guardrail_report: dict[str, Any],
raw_payload = None,
) -> None:
output_dir.mkdir(parents=True, exist_ok=True)
write_json(output_dir / "transcript_segments.json", [asdict(segment) for segment in segments])
(output_dir / "speaker_labeled_transcript.md").write_text(transcript_as_markdown(segments), encoding="utf-8")
write_json(output_dir / "meeting_intelligence.json", intelligence)
(output_dir / "meeting_brief.md").write_text(render_meeting_brief(intelligence), encoding="utf-8")
write_json(output_dir / "guardrail_report.json", guardrail_report)
if raw_payload is not None:
write_json(output_dir / "raw_transcription_response.json", to_plain(raw_payload))
def run_pipeline_from_segments(
segments: list[Segment],
output_dir: Path,
intelligence = None,
redaction_enabled: bool = False,
moderation_results = None,
raw_saved: bool = False,
raw_payload = None,
) -> dict[str, Any]:
if redaction_enabled:
segments = redact_segments(segments)
if intelligence is None:
intelligence = generate_meeting_intelligence(segments)
meeting_brief = render_meeting_brief(intelligence)
guardrail_report = build_guardrail_report(
segments=segments,
intelligence=intelligence,
meeting_brief=meeting_brief,
redaction_enabled=redaction_enabled,
raw_saved=raw_saved,
moderation_results=moderation_results or {},
)
write_artifacts(output_dir, segments, intelligence, guardrail_report, raw_payload=raw_payload if raw_saved else None)
return {
"segments": segments,
"intelligence": intelligence,
"meeting_brief": meeting_brief,
"guardrail_report": guardrail_report,
"output_dir": output_dir,
}
print("Artifact helpers ready")
Paso 7: Ejecuta el fixture de demostración determinista
Esta sección es una demostración determinista sin red, no una evaluación de la calidad del modelo. Utiliza una transcripción diarizada sintética fija y un objeto de inteligencia de reuniones esperado fijo para que los revisores puedan ejecutar el notebook sin una clave API.
Qué verifica este fixture
El fixture ejercita la misma ruta de artefactos y protecciones utilizada por el audio real: renderizado de transcripciones, escritura de JSON, renderizado de resúmenes de Markdown, ayudantes de redacción de PII, validación de referencias de evidencia, campos anulables y enrutamiento de revisión.
Qué no verifica este fixture
No mide la calidad de la transcripción, la precisión de la diarización o la calidad de la extracción del modelo en reuniones nuevas. Las secciones de evaluación a continuación añaden puntuación determinista y un patrón opcional de LLM como juez para esa capa.
output_dir = Path(tempfile.mkdtemp(prefix="meeting-intelligence-demo-"))
demo_run = run_pipeline_from_segments(
segments=DEMO_SEGMENTS,
output_dir=output_dir,
intelligence=demo_meeting_intelligence(),
)
print(f"Wrote meeting intelligence artifacts to {output_dir}")
for artifact_name in [
"transcript_segments.json",
"speaker_labeled_transcript.md",
"meeting_intelligence.json",
"meeting_brief.md",
"guardrail_report.json",
]:
artifact_path = output_dir / artifact_name
print(f"- {artifact_path} ({artifact_path.stat().st_size} bytes)")
segments = json.loads((output_dir / "transcript_segments.json").read_text())
segments[:2]
show_markdown((output_dir / "speaker_labeled_transcript.md").read_text())
meeting_intelligence_json = json.loads((output_dir / "meeting_intelligence.json").read_text())
show_json(meeting_intelligence_json, expanded=False)
show_markdown((output_dir / "meeting_brief.md").read_text())
guardrail_report = json.loads((output_dir / "guardrail_report.json").read_text())
show_json(guardrail_report, expanded=True)
Paso 8: Ejecuta con audio real
La siguiente celda es intencionalmente opcional. Establece RUN_REAL_AUDIO = True, proporciona tus rutas de audio locales y asegúrate de que OPENAI_API_KEY esté configurado.
La API de transcripciones acepta archivos de hasta 25 MB. chunking_strategy="auto" segmenta una carga válida; no divide un archivo de tamaño excesivo. Para reuniones más grandes, comprime a un formato de menor tasa de bits compatible o divide la grabación en archivos delimitados antes de la transcripción, luego conserva o compensa las marcas de tiempo al combinar los resultados.
Las grabaciones largas también pueden exceder el tiempo de espera de lectura predeterminado del SDK de Python. El asistente transmite segmentos diarizados finalizados por defecto y mantiene un tiempo de espera de 30 minutos como respaldo. Establece stream_transcription=False solo cuando necesites específicamente una respuesta no transmitida; divide grabaciones inusualmente largas cuando una solicitud no sea operativamente confiable.
Cómo los lectores suministran sus archivos
Este recetario es "notebook-first"; no hay un comando .py separado en el artefacto publicado. Coloca la grabación de la reunión y cualquier clip de referencia opcional en un lugar donde el kernel del notebook pueda leerlos. En Jupyter local, puede ser una carpeta junto al notebook o una ruta absoluta en el disco. En un notebook alojado, sube los archivos a la sesión del notebook primero.
Por ejemplo, un lector podría tener:
audio/
customer_call.mp3
internal_rep_reference.wav
Luego, apunta las variables de configuración a esos archivos:
AUDIO_FILE = Path("audio/customer_call.mp3")
KNOWN_SPEAKERS = {
"Internal rep": Path("audio/internal_rep_reference.wav"),
}
AUDIO_FILE es la grabación original de la reunión. KNOWN_SPEAKERS mapea la etiqueta que quieres en la salida a un clip de referencia separado de 2 a 10 segundos; el ayudante envía el clip con la solicitud en lugar de adjuntarlo al audio de la reunión. En una aplicación de producción, el mismo ayudante puede recibir un archivo temporal creado a partir de una carga o descargado del almacenamiento de objetos.
Para la primera ejecución de estilo de producción, mantén la configuración simple:
- Usa un archivo de audio de reunión.
- Usa
chunking_strategy="auto"para grabaciones más largas. - Agrega referencias de oradores conocidos solo cuando tengas consentimiento y una necesidad comercial clara.
- Ejecuta la redacción antes del almacenamiento.
- Ejecuta la moderación cuando la clasificación de contenido dañino sea parte de tu política de revisión.
RUN_REAL_AUDIO = False
AUDIO_FILE = Path("/path/to/meeting.wav")
# Keep each reference clip separate from AUDIO_FILE. Use a clean, consented 2-10 second sample of one speaker.
KNOWN_SPEAKERS = {
# "Internal rep": Path("/path/to/internal_rep_reference.wav"),
# "Customer": Path("/path/to/customer_reference.wav"),
}
# Demonstration only: masks basic email and phone patterns, not a complete PII/DLP solution.
REDACT_REAL_AUDIO = True
MODERATE_REAL_AUDIO = False
SAVE_RAW_RESPONSE = False
REAL_OUTPUT_DIR = Path(tempfile.mkdtemp(prefix="meeting-intelligence-real-"))
if RUN_REAL_AUDIO:
if not os.getenv("OPENAI_API_KEY"):
raise RuntimeError("Set OPENAI_API_KEY before running on real audio.")
if not AUDIO_FILE.is_file():
raise FileNotFoundError(AUDIO_FILE)
known_speakers = [(speaker_name, reference_path) for speaker_name, reference_path in KNOWN_SPEAKERS.items()]
raw_transcription = transcribe_with_diarization(AUDIO_FILE, known_speakers)
real_segments = normalize_segments(raw_transcription)
if REDACT_REAL_AUDIO:
real_segments = redact_segments(real_segments)
moderation_results: dict[str, Any] = {}
if MODERATE_REAL_AUDIO:
moderation_results["transcript"] = moderate_text(transcript_for_model(real_segments))
real_intelligence = generate_meeting_intelligence(real_segments)
real_brief = render_meeting_brief(real_intelligence)
if MODERATE_REAL_AUDIO:
moderation_results["meeting_brief"] = moderate_text(real_brief)
real_report = build_guardrail_report(
segments=real_segments,
intelligence=real_intelligence,
meeting_brief=real_brief,
redaction_enabled=REDACT_REAL_AUDIO,
raw_saved=SAVE_RAW_RESPONSE,
moderation_results=moderation_results,
)
write_artifacts(
REAL_OUTPUT_DIR,
real_segments,
real_intelligence,
real_report,
raw_payload=raw_transcription if SAVE_RAW_RESPONSE else None,
)
print(f"Wrote real-audio artifacts to {REAL_OUTPUT_DIR}")
else:
print("Skipped real audio run. Set RUN_REAL_AUDIO = True after configuring AUDIO_FILE and OPENAI_API_KEY.")
Paso 9: Ejecuta verificaciones deterministas de humo y regresión
Estas verificaciones son deterministas y no llaman a la API. Trátalas como una prueba de humo y un conjunto de regresión para la mecánica del notebook, no como una evaluación de la calidad del modelo.
Verificaciones de prueba de humo
Las primeras verificaciones confirman que el notebook escribe todos los artefactos esperados y dirige las salidas de riesgo medio a revisión.
Verificaciones de regresión
Las aserciones restantes detectan regresiones en la nulabilidad del esquema, las referencias de evidencia, las afirmaciones de demostración no respaldadas, los casos extremos de respuesta, la redacción y el formato de la marca de tiempo.
expected_files = {
"transcript_segments.json",
"speaker_labeled_transcript.md",
"meeting_intelligence.json",
"meeting_brief.md",
"guardrail_report.json",
}
assert expected_files.issubset({path.name for path in output_dir.iterdir()})
assert guardrail_report["status"] == "review_required"
assert any(check["name"] == "risk_outputs" and check["status"] == "review" for check in guardrail_report["checks"])
assert any(check["name"] == "evidence_refs" and check["status"] == "pass" for check in guardrail_report["checks"])
action_schema = MEETING_INTELLIGENCE_SCHEMA["schema"]["properties"]["action_items"]["items"]["properties"]
participant_schema = MEETING_INTELLIGENCE_SCHEMA["schema"]["properties"]["participants"]["items"]["properties"]
question_schema = MEETING_INTELLIGENCE_SCHEMA["schema"]["properties"]["explicit_questions"]["items"]["properties"]
assert "null" in action_schema["due_date_or_trigger"]["type"]
assert "null" in action_schema["owner_speaker"]["type"]
assert "null" in participant_schema["inferred_role"]["type"]
assert "null" in question_schema["directed_to_speaker"]["type"]
assert demo_run["segments"][0].segment_id == "seg_001"
assert demo_run["intelligence"]["decisions"] == []
assert demo_run["intelligence"]["action_items"][0]["due_date_or_trigger"] is None
assert demo_run["intelligence"]["action_items"][0]["evidence_refs"][0]["segment_id"] == "seg_005"
assert demo_run["intelligence"]["explicit_questions"]
assert demo_run["intelligence"]["suggested_follow_ups"]
assert validate_evidence_refs(demo_run["intelligence"], demo_run["segments"]) == []
assert "structured outputs" not in demo_run["intelligence"]["follow_up_email"]["body"].lower()
assert "`seg_005`" in demo_run["meeting_brief"]
assert "_Not specified._" in demo_run["meeting_brief"]
broken_intelligence = json.loads(json.dumps(demo_run["intelligence"]))
broken_intelligence["action_items"][0]["evidence_refs"] = [{"segment_id": "seg_999", "quote": "I will send a prototype"}]
assert validate_evidence_refs(broken_intelligence, demo_run["segments"])
broken_intelligence = json.loads(json.dumps(demo_run["intelligence"]))
broken_intelligence["action_items"][0]["evidence_refs"] = [{"segment_id": "seg_005", "quote": "I will send the contract tomorrow"}]
assert validate_evidence_refs(broken_intelligence, demo_run["segments"])
try:
response_output_text_or_raise({"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}})
raise AssertionError("Expected incomplete response to raise")
except RuntimeError as exc:
assert "incomplete" in str(exc)
try:
response_output_text_or_raise({"status": "completed", "output": [{"content": [{"type": "refusal", "refusal": "Cannot comply."}]}]})
raise AssertionError("Expected refusal response to raise")
except RuntimeError as exc:
assert "refusal" in str(exc).lower()
try:
response_output_text_or_raise({"status": "completed", "output_text": ""})
raise AssertionError("Expected empty response to raise")
except RuntimeError as exc:
assert "empty" in str(exc).lower()
try:
parse_meeting_intelligence_json("not json")
raise AssertionError("Expected invalid JSON to raise")
except RuntimeError as exc:
assert "invalid JSON" in str(exc)
redacted = redact_segments([
Segment("seg_test", "Customer", 0.0, 3.0, "Email me at [email protected] or call 415-555-0100.")
])
assert redacted[0].segment_id == "seg_test"
assert redacted[0].text == "Email me at [email] or call [phone]."
assert format_timestamp(6000) == "100:00.000"
with tempfile.NamedTemporaryFile(suffix=".wav") as oversized_audio:
oversized_audio.truncate(MAX_AUDIO_UPLOAD_BYTES + 1)
oversized_audio.flush()
try:
transcribe_with_diarization(Path(oversized_audio.name), [])
raise AssertionError("Expected oversized audio to raise")
except ValueError as exc:
assert "25 MB" in str(exc)
with tempfile.NamedTemporaryFile(suffix=".wav") as tiny_audio:
try:
transcribe_with_diarization(Path(tiny_audio.name), [], request_timeout_seconds=0)
raise AssertionError("Expected invalid timeout to raise")
except ValueError as exc:
assert "positive" in str(exc)
streamed_transcription = collect_streamed_transcription([
{"type": "transcript.text.segment", "id": "seg_stream", "speaker": "A", "start": 0.0, "end": 1.0, "text": "Hello"},
{"type": "transcript.text.done", "text": "Hello", "usage": {"total_tokens": 1}},
])
streamed_segments = normalize_segments(streamed_transcription)
assert streamed_segments[0].segment_id == "seg_stream"
assert streamed_segments[0].speaker == "A"
assert streamed_transcription["usage"]["total_tokens"] == 1
print("Notebook demo validation passed")
Paso 10: Ejecuta evaluaciones deterministas
Esta sección ejecuta una pequeña evaluación determinista contra el fixture de demostración etiquetado. Todavía no es una evaluación de producción amplia, pero muestra cómo calificar la calidad de la extracción con reglas reproducibles antes de agregar juicios calificados por el modelo.
Los calificadores a continuación miden la precisión/recuperación de elementos de acción, la precisión/recuperación de preguntas explícitas, las decisiones no respaldadas, los campos desconocidos anulables y la validez de la referencia de evidencia. Para producción, reemplaza GOLD_EVAL_LABELS con un conjunto de datos etiquetado más grande y ejecuta los mismos calificadores en cada ejemplo.
GOLD_EVAL_LABELS: dict[str, Any] = {
"action_items": [
{
"owner_speaker": "Solutions Engineer",
"task_contains": ["send a prototype", "speaker-aware transcripts", "redaction pass", "crm sync"],
"due_date_or_trigger": None,
"evidence_segment_ids": ["seg_005"],
}
],
"explicit_questions": [
{
"question_contains": ["support handoff", "break down"],
"asked_by_speaker": "Solutions Engineer",
"directed_to_speaker": "Customer",
"evidence_segment_ids": ["seg_001"],
}
],
"decisions": [],
}
def normalize_for_eval(text: Any) -> str:
return re.sub(r"\s+", " ", str(text or "")).strip().casefold()
def evidence_ref_segment_ids(item: dict[str, Any]) -> set[str]:
return {str(ref.get("segment_id", "")) for ref in item.get("evidence_refs", []) if isinstance(ref, dict)}
def contains_all_fragments(text: Any, fragments: list[str]) -> bool:
normalized = normalize_for_eval(text)
return all(normalize_for_eval(fragment) in normalized for fragment in fragments)
def action_item_matches_label(item: dict[str, Any], label: dict[str, Any]) -> bool:
return (
item.get("owner_speaker") == label.get("owner_speaker")
and item.get("due_date_or_trigger") == label.get("due_date_or_trigger")
and contains_all_fragments(item.get("task"), label.get("task_contains", []))
and set(label.get("evidence_segment_ids", [])).issubset(evidence_ref_segment_ids(item))
)
def explicit_question_matches_label(item: dict[str, Any], label: dict[str, Any]) -> bool:
return (
item.get("asked_by_speaker") == label.get("asked_by_speaker")
and item.get("directed_to_speaker") == label.get("directed_to_speaker")
and contains_all_fragments(item.get("question"), label.get("question_contains", []))
and set(label.get("evidence_segment_ids", [])).issubset(evidence_ref_segment_ids(item))
)
def precision_recall(predicted: list[dict[str, Any]], labels: list[dict[str, Any]], matcher) -> dict[str, Any]:
matched_label_indexes: set[int] = set()
matched_predictions = 0
for item in predicted:
for index, label in enumerate(labels):
if index in matched_label_indexes:
continue
if matcher(item, label):
matched_label_indexes.add(index)
matched_predictions += 1
break
precision = matched_predictions / len(predicted) if predicted else (1.0 if not labels else 0.0)
recall = len(matched_label_indexes) / len(labels) if labels else 1.0
return {
"precision": round(precision, 3),
"recall": round(recall, 3),
"matched_predictions": matched_predictions,
"predicted_count": len(predicted),
"label_count": len(labels),
}
def evidence_ref_count(intelligence: dict[str, Any]) -> int:
return sum(len(refs) for _, refs in iter_evidence_refs(intelligence) if isinstance(refs, list)) + len(intelligence.get("notable_quotes", []))
def run_deterministic_evals(
intelligence: dict[str, Any],
segments: list[Segment],
labels: dict[str, Any],
) -> dict[str, Any]:
action_item_scores = precision_recall(
intelligence.get("action_items", []),
labels.get("action_items", []),
action_item_matches_label,
)
explicit_question_scores = precision_recall(
intelligence.get("explicit_questions", []),
labels.get("explicit_questions", []),
explicit_question_matches_label,
)
evidence_problems = validate_evidence_refs(intelligence, segments)
total_evidence_refs = evidence_ref_count(intelligence)
valid_evidence_ref_rate = (
round((total_evidence_refs - len(evidence_problems)) / total_evidence_refs, 3)
if total_evidence_refs
else 1.0
)
action_items = intelligence.get("action_items", [])
nullable_due_date_rate = (
round(sum(1 for item in action_items if item.get("due_date_or_trigger") is None) / len(action_items), 3)
if action_items
else 1.0
)
unsupported_decision_count = len(intelligence.get("decisions", [])) if not labels.get("decisions") else 0
pass_conditions = [
action_item_scores["precision"] == 1.0,
action_item_scores["recall"] == 1.0,
explicit_question_scores["precision"] == 1.0,
explicit_question_scores["recall"] == 1.0,
valid_evidence_ref_rate == 1.0,
nullable_due_date_rate == 1.0,
unsupported_decision_count == 0,
]
return {
"status": "pass" if all(pass_conditions) else "review_required",
"action_items": action_item_scores,
"explicit_questions": explicit_question_scores,
"valid_evidence_ref_rate": valid_evidence_ref_rate,
"evidence_ref_problem_count": len(evidence_problems),
"unsupported_decision_count": unsupported_decision_count,
"nullable_due_date_rate": nullable_due_date_rate,
}
deterministic_eval_report = run_deterministic_evals(
demo_run["intelligence"],
demo_run["segments"],
GOLD_EVAL_LABELS,
)
show_json(deterministic_eval_report, expanded=True)
assert deterministic_eval_report["status"] == "pass"
Paso 11: Agrega evaluaciones opcionales de LLM como juez
Las evaluaciones de LLM como juez son útiles para calificar cualidades que los calificadores deterministas no pueden capturar completamente, como la utilidad del resumen, los seguimientos faltantes y si el resumen ayudaría a un revisor. Mantén esto opcional porque llama a la API y puede variar según el modelo del juez. Úsalo junto con los calificadores deterministas, no en su lugar.
El juez a continuación recibe la transcripción, la salida estructurada y una rúbrica. Devuelve puntuaciones y hallazgos de revisión. Deja RUN_LLM_JUDGE_EVAL = False para la ejecución predeterminada del notebook sin red.
RUN_LLM_JUDGE_EVAL = False
LLM_JUDGE_MODEL = os.getenv("OPENAI_MEETING_INTELLIGENCE_JUDGE_MODEL", DEFAULT_SUMMARY_MODEL)
LLM_JUDGE_SCHEMA: dict[str, Any] = {
"name": "meeting_intelligence_judge",
"strict": True,
"schema": {
"type": "object",
"additionalProperties": False,
"properties": {
"outcome": {"type": "string", "enum": ["pass", "review", "fail"]},
"overall_score": {"type": "number"},
"scores": {
"type": "object",
"additionalProperties": False,
"properties": {
"grounding": {"type": "number"},
"action_item_correctness": {"type": "number"},
"completeness": {"type": "number"},
"safety_review_readiness": {"type": "number"},
},
"required": ["grounding", "action_item_correctness", "completeness", "safety_review_readiness"],
},
"findings": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"properties": {
"area": {"type": "string"},
"severity": {"type": "string", "enum": ["low", "medium", "high"]},
"explanation": {"type": "string"},
},
"required": ["area", "severity", "explanation"],
},
},
},
"required": ["outcome", "overall_score", "scores", "findings"],
},
}
def run_llm_judge_eval(segments: list[Segment], intelligence: dict[str, Any], model: str = LLM_JUDGE_MODEL) -> dict[str, Any]:
from openai import OpenAI
client = OpenAI()
transcript = transcript_for_model(segments)
completion = client.responses.create(
model=model,
temperature=0,
store=False,
input=[
{
"role": "system",
"content": (
"You are judging a meeting-intelligence extraction. Grade only against the transcript. "
"Penalize unsupported claims, missing major action items, missing customer risks, incorrect speaker attribution, "
"and outputs that are not ready for human review. Return calibrated scores from 0 to 1."
),
},
{
"role": "user",
"content": (
"Transcript:\n"
f"{transcript}\n\n"
"Meeting intelligence JSON:\n"
f"{json.dumps(intelligence, indent=2)}"
),
},
],
text={
"format": {
"type": "json_schema",
"name": LLM_JUDGE_SCHEMA["name"],
"strict": LLM_JUDGE_SCHEMA["strict"],
"schema": LLM_JUDGE_SCHEMA["schema"],
}
},
)
return parse_meeting_intelligence_json(response_output_text_or_raise(completion))
if RUN_LLM_JUDGE_EVAL:
if not os.getenv("OPENAI_API_KEY"):
raise RuntimeError("Set OPENAI_API_KEY before running the LLM judge eval.")
llm_judge_report = run_llm_judge_eval(demo_run["segments"], demo_run["intelligence"])
show_json(llm_judge_report, expanded=True)
else:
print("Skipped LLM judge eval. Set RUN_LLM_JUDGE_EVAL = True after configuring OPENAI_API_KEY.")
Lista de verificación de endurecimiento de producción
Usa esta lista de verificación antes de convertir la muestra en un flujo de trabajo de cliente:
| Preocupación | Recomendación |
|---|---|
| Consentimiento | Asegúrate de que la grabación de llamadas, la diarización y las referencias de oradores conocidos estén permitidas en tu producto, política y región. |
| Retención de audio sin procesar | Almacena el audio sin procesar solo el tiempo necesario. Persiste los segmentos de transcripción normalizados cuando sea posible. |
| Grabaciones grandes | Rechaza archivos de más de 25 MB antes de la carga. Comprime o divide reuniones más largas, luego conserva los desfases de tiempo al combinar los resultados. |
| PII y DLP | Trata las expresiones regulares de correo electrónico y teléfono incluidas solo como ilustrativas. Usa un detector de PII/DLP aprobado por la política y revisión humana para flujos de trabajo sensibles o regulados. |
| Referencias de orador | Trata los clips de referencia como datos sensibles. Almacena mínimamente, cifra en reposo y rota/elimina cuando ya no sean necesarios. |
| Evidencia | Requiere referencias de evidencia estructuradas en decisiones, riesgos y elementos de acción. Valida que cada referencia apunte a un ID de segmento y una cita reales. |
| Revisión humana | Dirige resúmenes de alto riesgo, promesas de cumplimiento, reclamaciones de precios o términos contractuales para revisión. |
| Moderación | Usa la API de Moderación para la clasificación de contenido dañino cuando las notas puedan contener contenido inseguro. Mantén las verificaciones de privacidad y cumplimiento separadas. |
| Comportamiento de reintento | Reintenta errores transitorios de la API con retroceso exponencial. Evita duplicar escrituras posteriores en CRM usando claves de idempotencia. |
| Observabilidad | Registra nombres de modelos, versiones de prompts, versiones de esquemas, duración del audio, latencia, estado de redacción y decisiones del revisor. |
| Evaluación | Muestra llamadas semanalmente. Rastrea la precisión de la atribución del orador, la precisión de los elementos de acción y la tasa de reclamaciones no respaldadas. |
Guía de evaluación para producción
La evaluación determinista anterior es intencionalmente pequeña: prueba que el patrón de puntuación funciona en un conjunto de datos etiquetado. Los equipos de producción deben expandirla a un conjunto de evaluación representativo antes de escribir los resultados en sistemas posteriores. Comienza con un conjunto pequeño y consentido de grabaciones o transcripciones, crea etiquetas revisadas por humanos y guarda un conjunto de reserva para pruebas de regresión cuando cambien los prompts, esquemas o modelos.
| Área | Ejemplos de métricas |
|---|---|
| Atribución de orador | Precisión de la etiqueta del orador, tasa de error de diarización, precisión del límite de turno del orador, tasa de coincidencia de orador conocido. |
| Fundamentación de la transcripción | Exactitud de la cita, corrección de la marca de tiempo, validez de la referencia de evidencia, tasa de afirmaciones sin soporte. |
| Extracción estructurada | Precisión y exhaustividad para elementos de acción, decisiones, riesgos, preguntas explícitas, seguimientos sugeridos y requisitos del cliente. |
| Seguridad y privacidad | Exhaustividad de la redacción de PII, exhaustividad de la bandera de moderación, tasa de revisión de falsos positivos, cumplimiento de la retención de audio sin procesar. |
| Impacto en el flujo de trabajo | Tiempo hasta la actualización del CRM, tasa de anulación del revisor, tasa de finalización del seguimiento, latencia de detección de riesgo de renovación o escalada. |
Una primera evaluación útil es simple: pide a los revisores que marquen cada elemento de acción extraído como correcto, parcialmente correcto, sin soporte o faltante en el resultado. Haz un seguimiento de la precisión de los elementos generados y de la exhaustividad con respecto al conjunto de oro etiquetado por humanos. Para citas y evidencia, prefiere verificaciones de coincidencia exacta o casi exacta con el texto del segmento de la transcripción para que los resúmenes que suenan útiles pero sin soporte no pasen desapercibidos. LLM-as-judge puede ayudar a calificar la utilidad y la exhaustividad, pero mantén las verificaciones de fundamentación deterministas en el ciclo porque son más fáciles de reproducir.
Próximos pasos
Puedes adaptar el mismo pipeline para:
- Entregas de éxito del cliente después de revisiones comerciales trimestrales.
- Escaladas de soporte donde la responsabilidad y las citas exactas importan.
- Llamadas de descubrimiento de ventas que alimentan los próximos pasos del CRM.
- Debriefs de entrevistas de reclutamiento donde cada entrevistador necesita notas con fuentes.
- Flujos de trabajo de atención médica o servicios financieros con controles de revisión y retención más estrictos.
Para escenarios en vivo, usa Realtime para la experiencia durante la llamada y aún ejecuta este pipeline de diarización posterior a la llamada cuando necesites inteligencia de reuniones duradera y respaldada por evidencia.
Documentos útiles: