AI 驱动的会议效率提升：从语音转写到行动项提取的工程实践

一、会议的"时间税"：低效会议的隐性成本量化

知识工作者平均每周参加 15-20 场会议，其中 40% 被认为"低效或无效"。某科技公司对 200 名工程师的调研显示：每人每周在会议上花费 12 小时，其中 4.8 小时被判定为"可以不参加"。更严重的是，会议结论的执行率仅 35%——大量决策在会议结束后被遗忘或偏离。低效会议的隐性成本不仅在于时间浪费，更在于决策失真和行动延迟。

AI 驱动的会议效率提升不是用 AI 替代会议，而是用技术手段压缩低效环节：实时转写消除记录遗漏，智能摘要压缩回顾时间，行动项提取确保决策落地。

二、AI 会议助手的处理流水线

flowchart LR
    subgraph 输入层["实时输入"]
        A[语音流<br/>多声道采集]
        B[屏幕共享<br/>演示内容]
    end

    subgraph 处理层["AI 处理流水线"]
        C[语音识别 ASR<br/>说话人分离<br/>实时转写]
        D[语义理解 NLU<br/>议题识别<br/>决策点提取]
        E[行动项提取<br/>责任人与截止日期<br/>优先级判定]
    end

    subgraph 输出层["结构化输出"]
        F[会议纪要<br/>议题 + 结论 + 行动项]
        G[行动项追踪<br/>自动同步到项目管理工具]
        H[知识沉淀<br/>可检索的会议知识库]
    end

    A --> C --> D --> E
    B --> D
    E --> F
    E --> G
    D --> H

    style 输入层 fill:#eef,stroke:#333
    style 处理层 fill:#fee,stroke:#333
    style 输出层 fill:#efe,stroke:#333

三、AI 会议助手的工程化实现

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime
from enum import Enum
import re


class MeetingItemType(Enum):
    AGENDA = "agenda"           # 议题
    DECISION = "decision"       # 决策
    ACTION_ITEM = "action_item" # 行动项
    QUESTION = "question"       # 待解决问题
    INFO = "info"               # 信息同步


@dataclass
class SpeakerSegment:
    """说话人片段"""
    speaker_id: str
    start_time: float           # 秒
    end_time: float
    text: str


@dataclass
class ActionItem:
    """行动项"""
    item_id: str
    description: str
    assignee: Optional[str] = None
    due_date: Optional[str] = None
    priority: str = "medium"    # high / medium / low
    source_segment: Optional[int] = None  # 来源片段索引
    status: str = "pending"


@dataclass
class MeetingSummary:
    """会议纪要"""
    meeting_id: str
    title: str
    start_time: datetime
    end_time: datetime
    participants: List[str]
    agenda_items: List[str]
    decisions: List[str]
    action_items: List[ActionItem]
    open_questions: List[str]
    key_discussions: List[Dict]


class MeetingAssistant:
    """
    AI 会议助手：从实时语音到结构化纪要
    核心流程：转写 → 语义理解 → 行动项提取 → 纪要生成
    """

    def __init__(self):
        self._segments: List[SpeakerSegment] = []
        self._action_items: List[ActionItem] = []
        self._action_counter = 0

    # ============ 语音转写与说话人分离 ============

    def process_segment(self, speaker_id: str, start_time: float,
                        end_time: float, text: str):
        """处理一段语音转写结果"""
        segment = SpeakerSegment(
            speaker_id=speaker_id,
            start_time=start_time,
            end_time=end_time,
            text=text.strip()
        )
        self._segments.append(segment)

    # ============ 议题识别 ============

    def identify_agenda_items(self) -> List[str]:
        """从转写文本中识别议题"""
        agenda_keywords = [
            r"今天.*讨论", r"接下来.*看", r"第一个议题",
            r"第二.*话题", r"关于.*的方案", r"我们需要.*决定",
        ]
        agendas = []
        for seg in self._segments:
            for pattern in agenda_keywords:
                match = re.search(pattern, seg.text)
                if match:
                    # 提取议题描述（取匹配位置后的句子）
                    start = match.start()
                    sentence = seg.text[start:start + 80]
                    agendas.append(sentence.rstrip("，。、"))

        return agendas

    # ============ 决策点提取 ============

    def extract_decisions(self) -> List[str]:
        """从转写文本中提取决策结论"""
        decision_patterns = [
            r"决定(.{5,50})",
            r"就这么定了[，,]?(.{0,30})",
            r"确认(.{5,50})",
            r"方案[一二三四]?[：:]?(.{5,50})",
            r"最终选择(.{5,50})",
        ]
        decisions = []
        for seg in self._segments:
            for pattern in decision_patterns:
                matches = re.finditer(pattern, seg.text)
                for match in matches:
                    decision_text = match.group(0).rstrip("，。、")
                    decisions.append(f"[{seg.speaker_id}] {decision_text}")

        return decisions

    # ============ 行动项提取 ============

    def extract_action_items(self) -> List[ActionItem]:
        """从转写文本中提取行动项"""
        # 行动项触发词
        action_triggers = [
            r"(.+)负责(.+)",
            r"(.+)去(.+)",
            r"需要(.+)完成(.+)",
            r"(.+)跟进(.+)",
            r"分配给(.+)[，,]?(.+)",
            r"下周五之前(.+)",
            r"本周内(.+)",
        ]

        for seg_idx, seg in enumerate(self._segments):
            for pattern in action_triggers:
                match = re.search(pattern, seg.text)
                if match:
                    self._action_counter += 1
                    groups = match.groups()

                    # 尝试提取责任人
                    assignee = self._extract_assignee(groups, seg.speaker_id)

                    # 尝试提取截止日期
                    due_date = self._extract_due_date(seg.text)

                    # 判定优先级
                    priority = self._infer_priority(seg.text)

                    action = ActionItem(
                        item_id=f"ACTION-{self._action_counter:03d}",
                        description=seg.text[match.start():match.end()],
                        assignee=assignee,
                        due_date=due_date,
                        priority=priority,
                        source_segment=seg_idx,
                    )
                    self._action_items.append(action)

        return self._action_items

    def _extract_assignee(self, groups: Tuple[str, ...],
                          default_speaker: str) -> Optional[str]:
        """从匹配组中提取责任人"""
        # 优先使用匹配到的姓名
        for g in groups:
            if len(g) <= 6 and not any(kw in g for kw in ["完成", "跟进", "负责"]):
                return g
        return default_speaker

    def _extract_due_date(self, text: str) -> Optional[str]:
        """从文本中提取截止日期"""
        date_patterns = {
            r"今天": "today",
            r"明天": "tomorrow",
            r"本周五": "this_friday",
            r"下周一": "next_monday",
            r"下周五": "next_friday",
            r"(\d+)月(\d+)号": "specific_date",
            r"月底": "month_end",
        }
        for pattern, label in date_patterns.items():
            if re.search(pattern, text):
                return label
        return None

    def _infer_priority(self, text: str) -> str:
        """推断行动项优先级"""
        high_keywords = ["紧急", "立即", "尽快", "今天", "明天", "上线"]
        low_keywords = ["有空", "后续", "有时间", "慢慢"]

        for kw in high_keywords:
            if kw in text:
                return "high"
        for kw in low_keywords:
            if kw in text:
                return "low"
        return "medium"

    # ============ 会议纪要生成 ============

    def generate_summary(self, meeting_title: str,
                         participants: List[str]) -> MeetingSummary:
        """生成结构化会议纪要"""
        start = datetime.fromtimestamp(
            self._segments[0].start_time if self._segments else 0
        )
        end = datetime.fromtimestamp(
            self._segments[-1].end_time if self._segments else 0
        )

        agendas = self.identify_agenda_items()
        decisions = self.extract_decisions()
        actions = self.extract_action_items()
        questions = self._extract_open_questions()

        # 关键讨论摘要
        key_discussions = self._summarize_key_discussions()

        return MeetingSummary(
            meeting_id=f"MTG-{datetime.now().strftime('%Y%m%d%H%M')}",
            title=meeting_title,
            start_time=start,
            end_time=end,
            participants=participants,
            agenda_items=agendas,
            decisions=decisions,
            action_items=actions,
            open_questions=questions,
            key_discussions=key_discussions,
        )

    def _extract_open_questions(self) -> List[str]:
        """提取待解决问题"""
        question_patterns = [
            r"(.{5,30})\?",
            r"(.{5,30})还没定",
            r"(.{5,30})需要再讨论",
            r"(.{5,30})待确认",
        ]
        questions = []
        for seg in self._segments:
            for pattern in question_patterns:
                match = re.search(pattern, seg.text)
                if match:
                    questions.append(match.group(0).rstrip("？?，。、"))
        return questions

    def _summarize_key_discussions(self) -> List[Dict]:
        """提取关键讨论片段"""
        discussions = []
        # 基于讨论时长和说话人数量判断重要性
        current_topic = None
        topic_start = 0
        topic_speakers = set()

        for seg in self._segments:
            # 简单的议题切换检测
            if any(kw in seg.text for kw in ["接下来", "然后", "另一个"]):
                if current_topic and len(topic_speakers) > 1:
                    duration = seg.start_time - topic_start
                    if duration > 60:  # 超过 1 分钟的讨论才记录
                        discussions.append({
                            "topic": current_topic,
                            "duration_seconds": duration,
                            "participants": list(topic_speakers),
                        })
                current_topic = seg.text[:50]
                topic_start = seg.start_time
                topic_speakers = {seg.speaker_id}
            else:
                topic_speakers.add(seg.speaker_id)

        return discussions


# ============ 行动项追踪同步 ============

class ActionItemSync:
    """
    行动项同步器：将提取的行动项同步到项目管理工具
    """

    def __init__(self):
        self._synced_items: Dict[str, Dict] = {}

    def sync_to_task_board(self, action: ActionItem,
                           board_type: str = "jira") -> Dict:
        """同步行动项到任务看板"""
        task = {
            "title": action.description,
            "assignee": action.assignee or "unassigned",
            "due_date": action.due_date,
            "priority": action.priority,
            "labels": ["meeting-action"],
            "source": f"meeting-action-{action.item_id}",
        }

        # 根据看板类型适配字段
        if board_type == "jira":
            task["issuetype"] = "Task"
            task["story_points"] = 1 if action.priority == "low" else 2
        elif board_type == "linear":
            task["estimate"] = 1 if action.priority == "low" else 2

        self._synced_items[action.item_id] = task
        return task

    def check_duplicates(self, new_action: ActionItem) -> Optional[str]:
        """检查行动项是否与已有任务重复"""
        for item_id, task in self._synced_items.items():
            # 简单的文本相似度检查
            if self._text_similarity(new_action.description,
                                     task["title"]) > 0.7:
                return item_id
        return None

    @staticmethod
    def _text_similarity(a: str, b: str) -> float:
        """简单的文本相似度（Jaccard）"""
        set_a = set(a)
        set_b = set(b)
        intersection = len(set_a & set_b)
        union = len(set_a | set_b)
        return intersection / union if union > 0 else 0.0

四、AI 会议助手的 Trade-offs

实时转写的准确率与延迟矛盾。流式 ASR 的延迟越低，准确率越低（缺少上下文）。会议场景中，2 秒延迟的转写准确率约 95%，0.5 秒延迟降至 88%。关键决策点的转写错误可能导致行动项提取失败，需要在实时性和准确性之间取舍。

行动项提取的误报与漏报。基于规则的模式匹配容易产生误报（将普通讨论识别为行动项）和漏报（口语化表达未匹配到模板）。LLM 辅助提取可以提升准确率，但增加了推理延迟和成本。混合策略（规则初筛 + LLM 精排）是当前的工程折中方案。

说话人分离的准确性。多人会议中，说话人分离（Diarization）的错误率在 10-20%。将 A 的发言归到 B 名下，会导致行动项责任人错误。在重要会议中，需要人工校验说话人标注。

知识沉淀的维护成本。会议纪要归档后，如果缺乏持续维护，知识库会快速膨胀并失去检索价值。需要定期清理过期内容、合并重复议题、标注决策状态。这部分工作目前仍需人工介入。

五、总结

AI 驱动的会议效率提升覆盖从语音转写到行动项追踪的完整链路。核心工程实现包括：实时 ASR 与说话人分离、基于模式匹配的议题和决策提取、行动项的责任人与截止日期识别、结构化纪要生成与任务看板同步。关键权衡在于转写延迟与准确率、行动项提取的误报与漏报、说话人分离的错误率，以及知识沉淀的维护成本。AI 会议助手的价值不在于替代会议，而在于确保会议结论可追溯、行动项可追踪、决策过程可回溯。