构建带说话人识别的会议智能流水线：音频说话人分离实战指南

本教程面向有编程基础的开发者，深入讲解如何利用OpenAI的gpt-4o-transcribe-diarize模型构建生产级的会议智能流水线。文章从架构设计、核心原理到代码实现，全面覆盖了音频输入、说话人分离、转录规范化、结构化信息提取、护栏检查等关键环节。通过合成数据演示和真实音频可选运行，读者可以掌握说话人识别、证据引用验证、PII脱敏等工程实践，并了解如何将提取的会议情报安全地集成到CRM等下游系统。

技术背景

在会议记录场景中，单纯的转录文本无法满足业务需求，因为缺少说话人归属信息，导致无法准确追踪谁说了什么。带说话人标签的转录文本能够支持销售发现、客户成功、支持升级、招聘面试等场景，帮助团队减少信息流失，提高交接和审查的完整性。传统方法仅将转录文本发送给模型总结，会丢失关键上下文，因此需要专门的说话人分离技术。

核心原理

核心原理是利用OpenAI的gpt-4o-transcribe-diarize模型，通过Transcriptions API实现说话人分离。该模型接受音频文件，返回带说话人标签的转录片段。通过设置response_format为diarized_json，可以获取结构化输出。chunking_strategy=auto用于处理长音频，stream=True用于流式获取最终结果。已知说话人参考片段通过数据URL传递，帮助模型识别特定说话人。

架构分析

系统架构分为六层：音频输入层接受会议录音和可选参考片段；流水线运行器负责验证输入、调用API并写入产物；说话人分离层调用模型获取带标签的转录；转录规范化层将API输出转换为统一的JSON和Markdown格式；会议智能层使用结构化输出提取摘要、决策、行动项等；护栏和审查门层进行PII脱敏、证据验证和内容审核。

完整实战教程

许多组织已经录制了重要的对话，但单纯的转录文本往往不足以支持可靠的后续跟进。缺失的一层是说话人归属：知道谁提出了担忧，谁做出了承诺，以及证据在录音中的位置。带说话人标签的转录文本可以让你将客户需求与销售跟进分开，将结构化证据引用与行动项放在一起，并在敏感承诺进入CRM、工单系统或知识库之前，将其路由到审查流程中。

当下游工作流依赖于谁说了什么时，这种模式非常有用：

对于收入团队来说，影响通常不在于生成另一个摘要，而在于减少对话与记录系统之间的信息流失。这样的流水线可以帮助团队更快地捕获CRM就绪的下一步行动，更早地识别续约或扩展风险，保留预测更新背后的证据，并利用有来源的示例而不是轶事笔记来指导销售代表或支持团队。目标不是自动化判断，而是使交接、审查和后续行动更加完整和可审计。

本笔记本展示了如何使用OpenAI音频说话人分离构建生产风格的呼叫后会议智能流水线。你将：

  1. 接受录制的会议音频文件。
  2. 可选地使用短参考片段映射已知说话人。
  3. 调用gpt-4o-transcribe-diarize，设置response_format="diarized_json"。
  4. 将带说话人标签的片段规范化为JSON和Markdown，并具有稳定的片段ID。
  5. 使用结构化输出提取会议简报、决策、风险、明确问题、建议的后续行动、行动项、证据引用和后续邮件草稿。
  6. 写入可审查的工件和本地护栏报告。

默认单元格使用合成说话人分离转录文本运行，无需API密钥。真实音频调用是可选加入的，因此笔记本可以安全地从头到尾审查。

架构

Architecture diagram| 层 | 职责 | 输出 |

| --- | --- | --- |

| 音频输入 | 接受通话录音和可选的已知说话人片段。 | meeting.wav, Agent=agent.wav |

| 流水线运行器 | 验证输入，将参考编码为数据 URL，调用 OpenAI，并写入产物。 | 运行元数据和输出目录 |

| 说话人分离 | 调用 gpt-4o-transcribe-diarize，使用 response_format="diarized_json" 和 chunking_strategy="auto"。 | 带说话人标签的片段 |

| 转录规范化 | 将 API 输出转换为一致的 JSON 和 Markdown，并带有稳定的片段 ID。 | transcript_segments.json, speaker_labeled_transcript.md |

| 会议智能 | 提取摘要、决策、行动项、风险、明确问题、建议后续、引文和后续邮件，并带有结构化证据引用。 | meeting_intelligence.json, meeting_brief.md |

| 护栏和审查门 | 编辑敏感字段，验证证据引用，可选地审核内容，并将有风险的输出路由到审查。 | guardrail_report.json |

这是有意基于请求的。Realtime API 更适合实时语音体验、浏览器捕获或电话流。对于持久的通话后说话人分离，此模式使用 Transcriptions API，然后在带说话人标签的转录上运行结构化提取。

为什么说话人感知的转录很重要

会议智能的第一版通常是“将转录发送给模型并总结”。这在演示中有效，但在客户工作流中会崩溃，因为它丢失了谁说了什么。客户可能陈述需求，销售可能做出承诺，而经理可能需要明确区分这些。

说话人感知的说话人分离为应用程序的其余部分提供了更好的结构：

安全与防护

会议智能应被视为敏感数据工作流，而不仅仅是转录或摘要任务。原始录音可能包含客户姓名、商业条款、支持细节、健康或财务信息以及内部战略。说话人参考片段也可能很敏感，因为它们与个人声音相关联。一旦管道将音频转换为结构化输出，这些输出可能会流入 CRM 记录、支持工单、客户计划、仪表板或审查队列。

当输出可能影响业务流程时，安全与防护最为重要。销售电话摘要可能包含定价或合同承诺。支持升级可能包含影响生产的事件或客户凭据。招聘汇报可能包含候选人反馈。受监管行业的会议可能包含需要保留、访问控制或编辑策略的数据。对于这些工作流，最安全的模式是尽量减少原始音频保留，在适当情况下编辑敏感内容，要求有证据支持的输出，并在下游写入之前将高风险或低置信度的输出路由到人工审查。

包含的正则表达式编辑仅是示例性的：它仅屏蔽基本的电子邮件和电话模式。它不是完整的 PII 或 DLP 系统。对于姓名、地址、账户标识符、凭据、健康数据、财务数据或受监管的工作流，请使用政策批准的 PII/DLP 检测器，并在下游写入之前保留人工审查关口。

对于结构化提取步骤，本笔记本在 Responses API 调用中设置了 store=False，以便生成的会议智能响应不会作为应用程序状态存储。store=False 是一个有用的请求级控制，但它不等同于为组织或项目启用零数据保留。如果您的工作流需要更严格的保留保证，请查看 OpenAI 的数据控制文档，并确认适合您用例的保留配置。

| 风险 | 防护措施 |

| --- | --- |

| 录音或说话人引用滥用 | 在录音、说话人分离或引用片段使用前，要求获得同意和政策批准。将说话人引用视为敏感的生物特征相关数据。 |

| 原始音频过度保留 | 默认不保存原始转录响应。原始音频和引用片段仅保留所需时间。若保留，需加密并限制访问。 |

| 转录文本中的提示注入 | 将转录文本视为不可信证据。将指令保留在系统消息中，并要求模型仅使用基于转录的事实。 |

| 不支持的行动项或决策 | 使用严格的结构化输出，并要求证据引用指向真实的片段ID和引文。 |

| 生成的笔记中包含敏感内容 | 尽可能在摘要前进行编辑，然后在转录文本和简报上运行生成后检查。 |

| 有害或政策敏感内容 | 可选地对转录文本和生成的简报文本调用 omni-moderation-latest 审核API。审核可检测有害内容，但不能替代隐私审查。 |

| 不安全的后续写入 | 不要直接从模型输出写入CRM、工单或分析系统。在中等/高风险、缺失证据、审核标记或原始响应保留之前设置人工审查门。 |

| 静默质量下降 | 记录模型版本、提示版本、模式版本、音频时长、编辑状态、审核状态和审查者决策。抽样调用以进行评估。 |

先决条件

在本地运行笔记本

从 Cookbook 仓库的本地克隆中，创建虚拟环境，安装 Jupyter 和 OpenAI SDK，然后启动此笔记本：

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

下面的合成演示仅使用 Python 标准库，不调用 API。对于真实音频，你也可以在现有笔记本环境中安装 OpenAI SDK：

%pip install "openai>=1.93.0"

核心说话人分离请求

核心 API 请求有意保持精简：

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)

重要细节如下：

说话人分离与说话人识别

说话人分离回答的是“每个片段是哪位说话人说的？”它在一段录音中区分不同的声音，但不会创建永久的身份档案，也不会记住某次通话中的 speaker_0 与之后通话中的 speaker_0 是同一个人。在没有参考信息的情况下，通用标签仍然有用，因为它们保留了归属关系：流水线可以区分提出需求的说话人和做出承诺的说话人。

已知说话人参考为当前请求提供了可选的身份提示：

| 输入 | 结果 |

| --- | --- |

| 仅会议音频 | 模型区分声音，通常使用通用标签，如 speaker_0 和 speaker_1。 |

| 会议音频加上命名参考片段 | 匹配的片段可以使用提供的名称；未匹配的说话人可以保持通用标签。 |

| 后续或历史录音 | 再次传递参考片段。标签不会自动跨录音传递。 |

将参考片段与会议录音一起传递

会议录音和参考片段是同一转录请求中的两个独立输入。不要将参考片段拼接到会议音频上。会议以 file=... 上传；每个参考片段编码为数据 URL，并通过 known_speaker_references 发送，同时在 known_speaker_names 的相同位置提供名称。

下面的辅助函数封装了该请求格式。例如，以下代码传递了一个会议录音以及一个内部代表发言的简短独立片段：

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,
)

使用一段干净、经同意的 2-10 秒片段，包含单一说话人且背景噪音最小。对于经常出现的内部说话人，生产应用可以维护一个受访问控制的参考注册表，并在每次请求时附加适当的片段。对于历史录音，对每段录音运行相同的流程；参考片段可以来自较早的经同意的通话，只要它干净且为单一说话人。将参考数据视为敏感数据，在代表性音频上评估匹配质量，并对高风险的下游写入保留人工审查。

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")

步骤 1：定义结构化输出模式

会议智能通常服务于记录系统。使用严格的结构化输出，以便下游代码获得稳定的形状，并拒绝不支持的字段，而不是静默接受。此模式还要求结构化的 evidence_refs：每个提取项必须引用转录文本的 segment_id 和该片段中的引用，这使护栏能够机械地验证接地性。

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")

步骤 2：构建音频和转录辅助工具

已知说话人参考是可选的。没有它们，说话人分离仍然可以区分说话人，但标签可能是通用的，例如 speaker_0 或 speaker_1。有了参考，API 可以将片段映射到您提供的名称。

使用简短、清晰的参考片段，包含一个说话人且背景噪音最小。仅在您获得同意并有明确业务需求时保留参考片段。

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")

步骤 3：规范化转录文本

规范化后的转录文本是音频处理与会议智能之间的契约。它帮助你无需重新转录音频即可重新运行摘要生成、检查归因质量，并保持原始音频的短期留存。

每个片段都有一个稳定的 segment_id，例如 seg_005。之后，模型必须在 evidence_refs 中引用这些 ID，而护栏步骤会验证每个被引用的引文确实出现在所引用的片段中。

下面的正则表达式脱敏辅助函数仅为示意：它仅屏蔽基本的电子邮件和电话号码模式。它不是一个完整的 PII 或 DLP 系统；对于敏感或受监管的工作流程，请使用经政策批准的检测器和人工审核。

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))

第四步：提取结构化会议情报

模型获得带有说话人标签的转录文本，并且必须仅使用该转录文本作为证据。最安全的默认做法是生成空数组，而不是看似合理但无依据的 CRM 备注。模式还使用了必需但可空的字段，例如 due_date_or_trigger、inferred_role 和 directed_to_speaker，因此未知值保持为 null，而不是用猜测填充。

对于提取的每个事实、行动项、风险、问题或建议，模型返回带有 segment_id 和引文的 evidence_refs。这为审阅者提供了可读的来源追踪，并为代码提供了可验证的具体内容。

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)

步骤 5：渲染可审阅的会议简报

审阅工件将发言人、片段 ID、时间戳和引用证据与决策、风险和行动项并列展示，以便人类在下游写入任何内容之前进行抽查。

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)

步骤 6：添加护栏并写入产物

该示例写入一个 guardrail_report.json，其中包含针对以下内容的本地检查：

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")

第7步：运行确定性演示夹具

本节是一个确定性的无网络演示，并非模型质量评估。它使用固定的合成对话转录和固定的预期会议智能对象，以便审阅者无需API密钥即可运行笔记本。

此夹具检查的内容

该夹具演练了与真实音频相同的工件和防护栏路径：转录渲染、JSON写入、Markdown简报渲染、PII脱敏辅助、证据引用验证、可空字段以及审阅路由。

此夹具不检查的内容

它不衡量新会议上的转录质量、说话人分离准确性或模型提取质量。下面的评估部分增加了确定性评分和可选的LLM作为评判者的模式，用于该层。

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)

步骤 8：使用真实音频运行

下一个单元格是特意选择加入的。设置 RUN_REAL_AUDIO = True，提供您的本地音频路径，并确保已设置 OPENAI_API_KEY。

Transcriptions API 接受最大 25 MB 的文件。chunking_strategy="auto" 会分段有效的上传；它不会拆分过大的文件。对于较大的会议，请压缩为受支持的低比特率格式，或在转录前将录音拆分为有界文件，然后在合并结果时保留或偏移时间戳。

长录音也可能超出 Python SDK 默认的读取超时时间。辅助函数默认流式传输最终的分段结果，并保留 30 分钟的超时作为后备。仅在您特别需要单个非流式响应时设置 stream_transcription=False；当单个请求在操作上不可靠时，请拆分异常长的录音。

读者如何提供文件

本食谱以笔记本为先；发布的工件中没有单独的 .py 命令。将会议录音和任何可选的参考片段放在笔记本内核可以读取的位置。在本地 Jupyter 中，这可以是笔记本旁边的文件夹或磁盘上的绝对路径。在托管笔记本中，请先将文件上传到笔记本会话中。

例如，读者可能有：

audio/
customer_call.mp3
internal_rep_reference.wav

然后将配置变量指向这些文件：

AUDIO_FILE = Path("audio/customer_call.mp3")
KNOWN_SPEAKERS = {
    "Internal rep": Path("audio/internal_rep_reference.wav"),
}

AUDIO_FILE 是原始会议录音。KNOWN_SPEAKERS 将你希望在输出中使用的标签映射到单独的 2-10 秒参考片段；辅助函数随请求发送该片段，而不是将其附加到会议音频中。在生产应用中，同一个辅助函数可以接收从上传或对象存储下载的临时文件。

对于首次生产级运行，保持设置简单：

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.")

步骤 9：运行确定性的冒烟测试和回归检查

这些检查是确定性的，不会调用 API。将它们视为笔记本机制的冒烟测试和回归套件，而不是对模型质量的评估。

冒烟测试检查

第一项检查确认笔记本写入所有预期的工件，并将中等风险的输出路由到审查。

回归检查

其余的断言捕获模式可空性、证据引用、不支持的演示声明、响应边缘情况、编辑和时间戳格式方面的回归。

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 alex@example.com 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")

步骤 10：运行确定性评估

本节针对带标签的演示夹具运行一个小型确定性评估。这仍然不是广泛的生产评估，但它展示了如何在添加模型评分判断之前，使用可复现的规则对提取质量进行评分。

下面的评分器衡量行动项精确率/召回率、明确问题精确率/召回率、不支持的决策、可空未知字段以及证据引用有效性。对于生产环境，请将 GOLD_EVAL_LABELS 替换为更大的带标签数据集，并对每个示例运行相同的评分器。

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"

第 11 步：添加可选的 LLM 作为评判者的评估

LLM 作为评判者的评估对于评定确定性评分器无法完全捕捉的质量非常有用，例如摘要的有用性、缺失的后续问题，以及简报是否有助于审阅者。请保持此步骤为可选，因为它会调用 API，并且可能因评判模型而异。将其与确定性评分器一起使用，而不是替代它们。

下面的评判器接收转录文本、结构化输出和评分标准。它返回分数和审阅发现。将 RUN_LLM_JUDGE_EVAL = False 保留为默认的无网络笔记本运行。

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.")

生产环境加固清单

在将示例转化为客户工作流之前，请使用此清单：

| 关注点 | 建议 |

| --- | --- |

| 同意 | 确保您的产品、政策和地区允许通话录音、说话人分离和已知说话人引用。 |

| 原始音频保留 | 仅按需存储原始音频。尽可能持久化规范化后的转录片段。 |

| 大型录音 | 上传前拒绝超过 25 MB 的文件。压缩或拆分较长的会议，并在合并结果时保留时间戳偏移。 |

| PII 和 DLP | 将包含的电子邮件和电话正则表达式仅视为示例。对于敏感或受监管的工作流，请使用经政策批准的 PII/DLP 检测器和人工审核。 |

| 说话人引用 | 将参考片段视为敏感数据。最小化存储，静态加密，并在不再需要时轮换/删除。 |

| 证据 | 要求对决策、风险和行动项提供结构化证据引用。验证每个引用都指向真实的片段 ID 和引用文本。 |

| 人工审核 | 将高风险摘要、合规承诺、定价声明或合同条款路由至人工审核。 |

| 审核 | 当笔记可能包含不安全内容时，使用审核 API 进行有害内容分类。保持隐私和合规检查分离。 |

| 重试行为 | 使用退避重试瞬时 API 错误。通过使用幂等键避免重复写入下游 CRM。 |

| 可观测性 | 记录模型名称、提示版本、模式版本、音频时长、延迟、脱敏状态和审核人决策。 |

| 评估 | 每周抽样通话。跟踪说话人归属准确性、行动项精确度和不支持声明率。 |

生产环境的评估指南

上述确定性评估刻意保持较小规模：它证明了评分模式在标记的测试数据上有效。生产团队在将输出写入下游系统之前，应将其扩展为具有代表性的评估集。从一小部分经同意使用的录音或转录文本开始，创建人工审核的标签，并保留一个留出集，以便在提示、模式或模型发生变化时进行回归测试。

| 领域 | 示例指标 |

| --- | --- |

| 说话人归属 | 说话人标签准确率、说话人分离错误率、说话人轮次边界准确率、已知说话人匹配率。 |

| 转录文本接地 | 引用精确度、时间戳正确性、证据引用有效性、无依据声明率。 |

| 结构化提取 | 行动项、决策、风险、明确问题、建议后续行动和客户需求的精确率和召回率。 |

| 安全与隐私 | PII 编辑召回率、审核标记召回率、误报审查率、原始音频保留合规性。 |

| 工作流影响 | CRM 更新时间、审核者覆盖率、后续完成率、续订或升级风险检测延迟。 |

一个有用的初步评估很简单：要求审核者将每个提取的行动项标记为正确、部分正确、无依据或输出中缺失。跟踪生成项目的精确率，并与人工标记的金标准集进行召回率对比。对于引用和证据，优先使用与转录文本片段的精确匹配或近似精确匹配检查，以便听起来有帮助但无依据的摘要不会在不知不觉中通过。LLM 作为评判者可以帮助评估有用性和完整性，但请保持确定性接地检查在循环中，因为它们更容易复现。

后续步骤

你可以将相同的流程应用于：

对于实时场景，使用 Realtime 进行通话中体验，并在需要持久、有证据支持的会议情报时，仍然运行此通话后说话人分离流程。

有用的文档：

常见问题

常见错误包括：音频文件超过25MB限制，需压缩或拆分；请求超时，需调整timeout参数；参考片段格式不正确，需确保为音频或视频文件；证据引用无效，需验证segment_id和引文匹配；PII脱敏不完整，需使用更全面的检测器。排查方法包括检查API返回状态、验证输入文件、使用日志记录模型版本和参数。

企业应用扩展

生产环境优化建议：实施同意管理，确保录音和说话人分离合规；最小化原始音频保留，优先存储转录文本；使用政策批准的PII/DLP检测器；对高风险输出设置人工审查门；记录模型版本和评估指标；定期抽样评估说话人归属准确率和行动项精确率。

总结

本教程面向有编程基础的开发者，深入讲解如何利用OpenAI的gpt-4o-transcribe-diarize模型构建生产级的会议智能流水线。文章从架构设计、核心原理到代码实现，全面覆盖了音频输入、说话人分离、转录规范化、结构化信息提取、护栏检查等关键环节。通过合成数据演示和真实音频可选运行，读者可以掌握说话人识别、证据引用验证、PII脱敏等工程实践，并了解如何将提取的会议情报安全地集成到CRM等下游系统。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部语音与实时 API教程