基于微软Foundry构建会议音频RAG pipeline

你们组织曾经召开过的每一次会议录音本身就是一个知识库。只不过它恰好以最难查询的格式存储——一堆 MP4 文件躺在存储账户里，几乎没人会打开第二次。好消息是，从这堆文件到一个可用的问答智能体的距离，现在比以往短得多，因为 Microsoft Foundry 把你需要的两半整合到了一个地方。快速转录在几秒钟内将音频转换为带说话人标记的文本（而非实时转录），Foundry IQ 则将这些文本转化为支持权限控制的知识库，任何智能体都可以通过单一端点查询。

本教程将从头到尾构建整套系统。到最后，你将得到一条流水线：监视 Blob 容器中的新录音、用说话人标签转录、将内容分块为说话人轮次并附加足够的元数据以使引用真正有用、将其索引为 Foundry IQ 知识源，并暴露一个 Foundry 智能体，能够回答类似"我们在 Q2 关于定价迁移做了什么决定，谁提出了反对"这样的问题，并附带回到录音中那一刻的真实引用。

在开始之前先说明一下命名，因为这块领域已经发生了变化。Ignite 2025 上，微软将 Azure AI Foundry 更名为 Microsoft Foundry，并于 2026 年 1 月的产品条款中正式确定。平台还是那个平台，但现在有两个门户体验和两代 SDK。azure-ai-projects 的 2.x 预览版面向新的 Foundry 门户和 API，而 1.x GA 版则面向文档中所说的 Foundry 经典版。本文中所有内容均使用 2.x 系列和基于 Responses 的智能体界面。

我们要构建什么，以及数据流的形态

这条流水线有两个独立的半部分，它们在经过整理的转录文本 Blob 容器处汇合。引入端是批处理和事件驱动的，关注吞吐量和文件不丢失。检索端是同步的、面向用户的，关注延迟和 grounded 质量。通过存储将两者解耦，意味着你可以重新索引、重新分块或更换检索策略，而无需再碰任何字节的音频。

这个流程值得从左到右读一遍。录音到达 raw-recordings。Event Grid 捕获 Blob Created 事件，并向队列放入一条消息，这样你就免费获得了重试语义和死信路径。队列触发的 Function 拉取消息，将音频 POST 到 Foundry Speech 快速转录端点，得到一个包含带 diarized 短语的同步响应。第二阶段将这些短语分组为说话人轮次，附加时间戳和会议元数据，并将 JSONL 写入 curated-transcripts。Foundry IQ 按计划对该容器建立索引。

为什么在 Event Grid 和 Function 之间放一个队列而不是直接触发？因为快速转录是同步的，而且音频文件很大。直接 blob 触发几乎无法控制并发，一旦有人批量上传六个月的存档录音，你就会耗尽 Speech 资源并开始收到 429。队列让你可以在 host.json 中限制 batchSize 并调整负载。

搭建 Foundry 项目和 Speech 资源

首先创建一个 Foundry 项目。在门户中，确保 New Foundry 开关是打开的，然后创建或选择一个项目。你需要从门户获取的是项目端点，形式为 https://<resource-name>.services.ai.azure.com/api/projects/<project-name>。

安装预览版包。

pip install "azure-ai-projects>=2.4.0" azure-identity openai azure-storage-blob requests
az login

Entra ID 是项目客户端唯一支持的认证方式，所以这里没有基于密钥的后备方案。在开发工作中给自己分配项目资源上的 Azure AI User 角色。对于流水线本身，使用用户分配的托管标识，并授予它 Azure AI User 加 Storage Blob Data Contributor 角色。

两个环境变量承载了本文其余部分的内容。

export FOUNDRY_PROJECT_ENDPOINT="https://your-account.services.ai.azure.com/api/projects/meetings"
export SPEECH_RESOURCE_NAME="your-speech-resource"

在一切构建之前，先确认项目客户端能正常与服务通信。

import os
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

with (
    DefaultAzureCredential() as credential,
    AIProjectClient(
        endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        credential=credential,
    ) as project,
):
    openai = project.get_openai_client()
    r = openai.responses.create(
        model="gpt-5-mini",
        input="Reply with the single word ready.",
    )
    print(r.output_text)

get_openai_client() 返回一个来自 openai 包的可认证客户端，配置为对你的 Foundry 项目端点执行 Responses 操作。这是需要内化的模式。你使用项目客户端进行设置、配置、智能体和评估，而 OpenAI 兼容客户端用于实际的模型调用。

将一小时的音频转换为带说话人标记的轮次

快速转录是处理已录制会议的正确工具。它同步返回结果，且速度远快于实时，这正是处理一个已存在的文件时你所需要的权衡。批转录是替代方案，在非常长的存档和高级定制场景下更有优势，但对于一小时的标准格式录音，快速转录在少量秒数内即可得到结果，延迟可预测。

端点是 /speechtotext/transcriptions:transcribe，当前普遍可用的 API 版本是 2025-10-15。它接受 multipart/form-data，音频在一部分，JSON 定义在另一部分。 diarization 通过携带 maxSpeakers 的 diarization 对象配置，服务可以在单一通道中分离最多 35 个不同说话人，之后才会报错。

以下是完整的 worker 代码，带有你绝对会需要的重试行为。

import json
import os
import time
import requests
from azure.identity import DefaultAzureCredential

SPEECH_ENDPOINT = (
    f"https://{os.environ['SPEECH_RESOURCE_NAME']}"
    ".cognitiveservices.azure.com/speechtotext/transcriptions:transcribe"
    "?api-version=2025-10-15"
)
SCOPE = "https://cognitiveservices.azure.com/.default"
RETRYABLE = {408, 429, 500, 502, 503, 504}

def transcribe(audio_path, locales=("en-US",), max_speakers=8, max_attempts=5):
    """Fast transcription with diarization and bounded exponential backoff."""
    credential = DefaultAzureCredential()
    definition = {
        "locales": list(locales),
        "diarization": {"enabled": True, "maxSpeakers": max_speakers},
        "profanityFilterMode": "None",
    }

for attempt in range(max_attempts):
        token = credential.get_token(SCOPE).token
        with open(audio_path, "rb") as fh:
            response = requests.post(
                SPEECH_ENDPOINT,
                headers={"Authorization": f"Bearer {token}"},
                files={"audio": (os.path.basename(audio_path), fh)},
                data={"definition": json.dumps(definition)},
                timeout=600,
            )

if response.status_code == 200:
            return response.json()

if response.status_code not in RETRYABLE:
            raise RuntimeError(
                f"Fast transcription failed {response.status_code} {response.text[:400]}"
            )

wait = float(response.headers.get("Retry-After", 2 ** attempt))
        time.sleep(min(wait, 60))

raise RuntimeError(f"Giving up on {audio_path} after {max_attempts} attempts")

其中有几处值得注意。服务发送 Retry-After 头时会遵循它，这在限流情况下非常重要，因为在共享 Speech 资源上盲目使用指数退避只会导致所有 worker 步调一致地重试。脏话过滤设为 None，因为默认是 Masked，而转录文本中被屏蔽的词会悄悄损害检索效果，因为星号会成为不匹配任何内容的 token。600 秒的超时是刻意设得宽松的，因为一个大文件在受限的出口路径上上传，在服务开始工作之前可能会花费很长时间。

响应包含一个 phrases 数组，每个条目携带 speaker、offsetMilliseconds、durationMilliseconds 和 text。Phrases 不是适合检索的分块粒度。它们通常只有一两句话，这意味着一个 phrase 的 embedding 几乎不带任何上下文，而对 phrase 的引用会把读者丢到一句话的中间。所以要把它们分组为说话人轮次。

from dataclasses import dataclass, asdict

@dataclass
class Turn:
    meeting_id: str
    meeting_title: str
    meeting_date: str
    speaker: str
    start_ms: int
    end_ms: int
    text: str

@property
    def chunk_id(self):
        return f"{self.meeting_id}-{self.start_ms:09d}"

def to_turns(result, meta, max_chars=2400, gap_ms=4000):
    """将说话人识别后的短语合并为说话轮次，拆分过长的轮次。"""
    turns, current = [], None

    for p in result.get("phrases", []):
        speaker = f"Speaker {p.get('speaker', 'unknown')}"
        start = p["offsetMilliseconds"]
        end = start + p["durationMilliseconds"]

        same_speaker = current and current.speaker == speaker
        contiguous = current and (start - current.end_ms) < gap_ms
        room = current and (len(current.text) + len(p["text"])) < max_chars

        if same_speaker and contiguous and room:
            current.text += " " + p["text"]
            current.end_ms = end
            continue

        if current:
            turns.append(current)
        current = Turn(
            meeting_id=meta["meeting_id"],
            meeting_title=meta["title"],
            meeting_date=meta["date"],
            speaker=speaker,
            start_ms=start,
            end_ms=end,
            text=p["text"],
        )

    if current:
        turns.append(current)
    return turns

gap_ms 守卫条件是人们容易遗漏的部分。没有它，一个在第 3 分钟和第 40 分钟都发言的说话人会被合并为一个 chunk（如果中间没有其他人发言，这种情况虽少见但会产生一个时间戳范围毫无意义的 chunk）。4 秒的静默是会议音频中合理的轮次边界。

制作值得引用的 chunks

会议转录的检索质量取决于原始文本周围的内容。一个孤零零的说话轮次，比如"Yeah, I think that's fine, let's go with option two"，几乎是无法检索的，因为它没有名词。解决方案是在每条记录中写入少量生成上下文，让混合搜索能够匹配到它。

def contextualize(openai, turn, neighbors):
    """前置一句场景摘要，使短轮次保持可检索性。"""
    window = "\n".join(f"{n.speaker}: {n.text}" for n in neighbors)
    r = openai.responses.create(
        model="gpt-4.1-mini",
        input=(
            "Write one sentence, under 25 words, situating the final utterance "
            "inside this meeting excerpt. Name the topic and any decision. "
            "Do not editorialize.\n\n"
            f"Meeting: {turn.meeting_title} ({turn.meeting_date})\n\n"
            f"{window}\n\nFinal utterance: {turn.speaker}: {turn.text}"
        ),
    )
    return r.output_text.strip()

def to_records(openai, turns):
    for i, turn in enumerate(turns):
        neighbors = turns[max(0, i - 3): i + 1]
        context = contextualize(openai, turn, neighbors)
        yield {
            **asdict(turn),
            "chunk_id": turn.chunk_id,
            "context": context,
            "content": f"{context}\n\n{turn.speaker}: {turn.text}",
            "timecode": f"{turn.start_ms // 60000:02d}:{(turn.start_ms // 1000) % 60:02d}",
        }

每个轮次调用一次小模型，一小时会议大约几百次调用，每次几百个 token。用信号量控制并发执行，而不是串行执行。timecode 字段是让引用成为产品功能而非脚注的关键，因为它可以渲染为视频播放器的深度链接。

将记录以 JSONL 格式写入 curated-transcripts，每个会议一个文件，到此音频处理就结束了。

将转录文本接入 Foundry IQ 知识库

Foundry IQ 是构建在 Azure AI Search 上的知识和检索层。思维模型是两个嵌套对象。知识源（knowledge source）指向可搜索内容，知识库（knowledge base）将一个或多个知识源包装在单个端点后，供 AI 智能体查询。对于索引源，Foundry IQ 管理整个索引管道，因此内容会被摄取、分块、向量化，并准备好进行混合检索，无需手动搭建 skillset。

AI 智能体检索功能在 REST API 2026-04-01 版本中正式可用。2026-05-01-preview 版本则暴露了更完整的功能集，包括预览知识源类型以及向非 Web 源附加 LLM 的能力。Blob Storage 是正式可用的索引源类型，这正是我们需要的。

将知识源指向经过整理的容器。

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    KnowledgeBase,
    KnowledgeSourceReference,
    AzureBlobKnowledgeSource,
    AzureBlobKnowledgeSourceParameters,
)
from azure.identity import DefaultAzureCredential

index_client = SearchIndexClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    credential=DefaultAzureCredential(),
)

source = AzureBlobKnowledgeSource(
    name="meeting-transcripts",
    description=(
        "Diarized speaker turns from recorded internal meetings, 2024 onward. "
        "Each chunk carries meeting title, date, speaker label, and timecode."
    ),
    azure_blob_parameters=AzureBlobKnowledgeSourceParameters(
        connection_string=os.environ["BLOB_CONNECTION"],
        container_name="curated-transcripts",
        embedding_model=..., # your deployed text embedding model
        chat_completion_model=..., # optional, enables verbalization
    ),
)
index_client.create_or_update_knowledge_source(source)

那个 description 字段不是装饰。当一个知识库包含多个源时，检索引擎会规划查询哪些源，而 description 是它用来路由的主要信号。写的时候要像在给一个从未见过你数据的同事做简报。

现在创建知识库。

kb = KnowledgeBase(
    name="meetings-kb",
    knowledge_sources=[
        KnowledgeSourceReference(name="meeting-transcripts", always_query_source=False),
    ],
    retrieval_instructions=(
        "Meeting transcripts. When the user asks who said or decided something, "
        "return the speaker turns that contain the statement plus the surrounding turns. "
        "Prefer recent meetings when the question is about current state."
    ),
)
index_client.create_or_update_knowledge_base(kb)

检索引擎规划查询哪些源，并在第一轮检索未达到相关性阈值时执行迭代搜索。迭代搜索依赖于设置中等检索推理强度（reasoning effort），可以在知识库级别设置，也可以在每次请求时设置。这个旋钮也是影响延迟和成本的最大杠杆，所以要把它当作调参参数而不是设置后就忘了的值。

给 AI 智能体配备知识库和人格

知识库就位后，AI 智能体就很简洁了。2.x SDK 中的 AI 智能体操作构建在 Responses 协议之上，AI 智能体是通过 create_version 创建的版本化对象。

from azure.ai.projects.models import PromptAgentDefinition

INSTRUCTIONS = """You answer questions about internal meetings using only the
meeting transcript knowledge base.

Rules you follow without exception.
1. Every factual claim carries a citation naming the meeting title, date, and timecode.
2. When you cannot find support in the transcripts, say so plainly and stop.
3. Attribute statements to the speaker label exactly as it appears. Never guess a real name.
4. When speakers disagreed, surface the disagreement rather than flattening it into consensus.
5. Distinguish a decision from a suggestion. Quote the language that makes it one or the other.
"""

agent = project.agents.create_version(
    agent_name="meeting-analyst",
    definition=PromptAgentDefinition(
        model="gpt-5-mini",
        instructions=INSTRUCTIONS,
        tools=[{"type": "knowledge_base", "knowledge_base": {"name": "meetings-kb"}}],
    ),
)
print(agent.id, agent.version)

规则三起了实际作用。说话人识别给你的是录音内稳定的说话人标识符，而不是真实身份，所以你得到的是通用标签而非姓名。如果指令不禁止它，有能力的模型会愉快地推断 Speaker 2 就是会议标题中出现姓名的人，而它出错和正确的概率大致相当。如果你需要真实姓名，在分块阶段从日历元数据或多声道采集自行映射，并将解析后的姓名写入记录。

调用 AI 智能体就像调用任何 Responses 一样。

def ask(openai, agent_name, question, previous_response_id=None):
    return openai.responses.create(
        extra_body={"agent": {"name": agent_name, "type": "agent_reference"}},
        input=question,
        previous_response_id=previous_response_id,
    )

first = ask(openai, "meeting-analyst",
            "What did we decide about the pricing migration, and did anyone object?")
print(first.output_text)

follow_up = ask(openai, "meeting-analyst",
                "Which of those objections were ever resolved?",
                previous_response_id=first.id)
print(follow_up.output_text)

通过 previous_response_id 串联对话可以保持在服务端处理，这意味着你无需在每次交互时发送一份不断增长的聊天记录，也无需自己实现历史存储。

当检索或模型不配合时如何优雅降级

生产环境中有两类失败需要区别对待。瞬时服务端错误需要重试；而空检索或弱检索则不应重试，因为用同样的查询再次检索同一个索引只会返回同样的空结果。

import random
from openai import APIStatusError, APITimeoutError

TRANSIENT = {408, 409, 429, 500, 502, 503, 504}

def ask_resilient(openai, agent_name, question, attempts=4, **kwargs):
    last = None
    for i in range(attempts):
        try:
            return ask(openai, agent_name, question, **kwargs)
        except APITimeoutError as exc:
            last = exc
        except APIStatusError as exc:
            if exc.status_code not in TRANSIENT:
                raise
            retry_after = exc.response.headers.get("retry-after")
            last = exc
            if retry_after:
                time.sleep(min(float(retry_after), 30))
                continue
        time.sleep(min(2 ** i + random.random(), 30))
    raise last

在退避策略中加入全抖动（full jitter）在任何真实并发场景下都是必选项。没有它，你的重试请求会同步成一个雷鸣般的群体效应，把一个短暂的限流变成持续的拥塞。

对于检索端，答案是让智能体的失败可见而非静默。上文的第二条指令告诉模型「找到就直说找到了」，你应该在评估集中对此做断言。一个承认无知的有据可查系统，远比一个从三个无关片段中拼凑出自信满满文字的系统更有价值——而且第二种失败模式在生产环境中更难被发现，因为它的输出看起来完全正常。

衡量系统是否真正有效

这条流水线中存在两个独立的质量问题，需要分别测量。转录层的准确率问题以词错误率（WER）衡量。检索和生成层有一个 groundedness 问题，由裁判模型测量。从外部看，这两种退化表现一模一样，这正说明将两者分开测量是必要的。

首先构建一个黄金集。用你实际听过的会议写出大约一百个问题，这比一千个合成问题更有价值——因为价值在于预期答案，而只有亲历会议的人才能写出那些答案。要有意识地覆盖那些别扭的边缘情况。包含一些答案确实不存在的问题，这样你可以测量拒绝行为。包含跨越两场会议的问题。包含有两人产生分歧的问题。

{"question": "Who owned the migration rollback plan after the March review?",
 "expected": "Speaker 3 accepted ownership at 41:12 in Platform Review 2026-03-04.",
 "must_cite": "Platform Review 2026-03-04",
 "kind": "attribution"}
{"question": "What was the agreed SLA for the batch job?",
 "expected": "Not discussed in any recorded meeting.",
 "must_cite": null,
 "kind": "refusal"}

评估操作位于 2.x SDK 的项目客户端中，通过 evaluators、evaluation_rules 和 schedules 等属性访问。对于 groundedness 和相关性，你可以使用内置的裁判评估器。对于词错误率，你需要注册一个自定义评估器，因为那个是算术计算而非判断。

import jiwer

def transcript_wer(reference_text, hypothesis_text):
    transform = jiwer.Compose([
        jiwer.ToLowerCase(),
        jiwer.RemovePunctuation(),
        jiwer.RemoveMultipleSpaces(),
        jiwer.Strip(),
        jiwer.ReduceToListOfListOfWords(),
    ])
    return jiwer.wer(reference_text, hypothesis_text,
                     truth_transform=transform, hypothesis_transform=transform)

手动纠正三到四段录音中的二十分钟音频，并将其作为参考基准。二十分钟听起来很少，确实也是，但它能捕捉到那些真正重要的失败——即领域词汇和缩写词变成语音糊状的问题。如果你的产品名称 WER 很差，解决方案是短语列表而不是更好的模型。短语列表让你向识别器传入一组可能出现的词汇，它们对专有名词和内部术语的效果非常显著。

值得在部署前把关的指标有以下四个。

引用有效性是最便宜但每个人都跳过的那个。你已经有 chunk 元数据了，所以从答案中解析出引用，然后断言每个会议标题都存在、每个时间码都在该录音时长范围内，大概三十行代码就能搞定。它能捕捉到裁判模型意外宽容的一种特定而尴尬的错误。

投入生产而不留遗憾

定时重新索引并预期内容波动。Foundry IQ 会为已索引数据源自动触发索引和数据同步，但你精心管理的容器才是契约。如果你改变了分块策略，就等于在重写每一条记录，而对大型语料库的全量重索引并非瞬时完成。对你的分块逻辑做版本控制，并将版本号写入每条记录，这样在迁移过程中你就能区分混合一代的内容。

在任何内容索引之前先决定权限模型。会议录音是组织内最敏感的内容之一。