Every recorded meeting your organization has ever held is already a knowledge base. It just happens to be stored in the least queryable format imaginable, which is a wall of MP4 files sitting in a storage account that nobody opens twice. The good news is that the gap between that wall of files and a working question-answering agent is now much shorter than it used to be, because Microsoft Foundry ships the two halves you need in one place. Fast transcription turns the audio into diarized text in seconds rather than in real time, and Foundry IQ turns that text into a permission-aware knowledge base that any agent can query through a single endpoint.
This walkthrough builds the whole thing end to end. By the end you will have a pipeline that watches a blob container for new recordings, transcribes them with speaker labels, chunks them into speaker turns with enough metadata to make citations useful, indexes them as a Foundry IQ knowledge source, and exposes a Foundry agent that answers questions like "what did we decide about the pricing migration in Q2 and who pushed back" with real references back to the moment in the recording.
A quick naming note before we start, because the ground has moved. At Ignite 2025 Microsoft renamed Azure AI Foundry to Microsoft Foundry, and the rename was formalized in the January 2026 Product Terms. The platform is the same platform, but there are now two portal experiences and two generations of SDK. The 2.x preview of azure-ai-projects
targets the new Foundry portal and API, and the 1.x GA line targets what the docs call Foundry classic. Everything in this article uses the 2.x line and the Responses-based agent surface.
The pipeline has two independent halves that meet at a blob container of curated transcripts. The ingestion half is batch and event-driven. It cares about throughput and about not losing files. The retrieval half is synchronous and user-facing. It cares about latency and about grounding quality. Keeping them decoupled through storage means you can reindex, re-chunk, or swap the retrieval strategy without touching a byte of audio again.
The flow is worth reading left to right once. A recording lands in raw-recordings
. Event Grid picks up the Blob Created event and drops a message on a queue, which gives you retry semantics and a dead letter path for free. A queue-triggered Function pulls the message, POSTs the audio to the Foundry Speech fast transcription endpoint, and gets back a synchronous response containing diarized phrases. A second stage groups those phrases into speaker turns, attaches timestamps and meeting metadata, and writes JSONL into curated-transcripts
. Foundry IQ indexes that container on a schedule.
Why a queue between Event Grid and the Function rather than a direct trigger? Because fast transcription is synchronous and the audio files are large. A direct blob trigger gives you very little control over concurrency, and the first time somebody bulk-uploads six months of archived recordings you will saturate your Speech resource and start collecting 429s. The queue lets you cap batchSize
in host.json
and shape the load.
Create a Foundry project first. In the portal, make sure the New Foundry toggle is on, then create or select a project. The thing you need out of the portal is the project endpoint, which has the form https://<resource-name>.services.ai.azure.com/api/projects/<project-name>
.
Install the preview packages.
pip install "azure-ai-projects>=2.4.0" azure-identity openai azure-storage-blob requests
az login
Entra ID is the only authentication method the projects client supports, so there is no key-based escape hatch here. Give yourself the Azure AI User role on the project resource for development work. For the pipeline itself, use a user-assigned managed identity and grant it Azure AI User plus Storage Blob Data Contributor.
Two environment variables carry the rest of the article.
export FOUNDRY_PROJECT_ENDPOINT="https://your-account.services.ai.azure.com/api/projects/meetings"
export SPEECH_RESOURCE_NAME="your-speech-resource"
Confirm the project client talks to the service before you build anything on top of it.
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()
returns an authenticated client from the openai
package configured to run Responses operations against your Foundry project endpoint. That is the pattern to internalize. You use the project client for setup, configuration, agents, and evaluations, and the OpenAI-compatible client for the actual model calls.
Fast transcription is the right tool for recorded meetings. It returns results synchronously and much faster than real time, which is exactly the tradeoff you want for a file that already exists. Batch transcription is the alternative and it wins on very long archives and on advanced customization, but for a one-hour standard-format recording, fast transcription gets you a result in a small number of seconds with predictable latency.
The endpoint is /speechtotext/transcriptions:transcribe
and the current generally available API version is 2025-10-15
. It takes multipart/form-data
with the audio in one part and a JSON definition in another. Diarization is configured with a diarization
object carrying maxSpeakers
, and the service can separate up to 35 distinct speakers in a single channel before it errors out.
Here is the worker in full, with the retry behavior that you will absolutely need.
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")
A few things in there earn their place. The Retry-After
header is honored when the service sends one, which matters a lot under throttling because blind exponential backoff on a shared Speech resource just means every worker retries in lockstep. Profanity filtering is set to None
because the default is Masked
and masked words in a transcript quietly damage retrieval, since the asterisks become tokens that match nothing. The 600 second timeout is generous on purpose, because a large file up over a constrained egress path can spend a long while before the service even starts work.
The response contains a phrases
array where each entry carries speaker
, offsetMilliseconds
, durationMilliseconds
, and text
. Phrases are the wrong chunk size for retrieval. They are usually a sentence or two, which means an embedding of a phrase carries almost no context and a citation to a phrase drops the reader into the middle of a thought. Group them into speaker turns instead.
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):
"""Collapse diarized phrases into speaker turns, splitting very long ones."""
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
The gap_ms
guard is the part people leave out. Without it, a speaker who talks at minute three and again at minute forty gets merged into one chunk if nobody else spoke in between, which is rare but produces a chunk whose timestamp range is meaningless. Four seconds of silence is a reasonable turn boundary for meeting audio.
Retrieval quality on meeting transcripts lives or dies on what surrounds the raw text. A bare speaker turn like "yeah I think that's fine, let's go with option two" is nearly unretrievable, because it contains no nouns. The fix is to write a small amount of generated context into each record and let the hybrid search match on that.
def contextualize(openai, turn, neighbors):
"""Prepend a one-line situating summary so short turns stay retrievable."""
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}",
}
This costs one small model call per turn, which on a one-hour meeting is a few hundred calls of a couple hundred tokens each. Run it concurrently with a semaphore rather than serially. The timecode
field is what makes citations feel like a product feature rather than a footnote, because you can render it as a deep link into your video player.
Write the records as JSONL to curated-transcripts
, one file per meeting, and you are done with audio forever.
Foundry IQ is the knowledge and retrieval layer built on Azure AI Search. The mental model is two nested objects. A knowledge source points at searchable content, and a knowledge base wraps one or more knowledge sources behind a single endpoint that agents query. For indexed sources, Foundry IQ manages the whole indexing pipeline, so content gets ingested, chunked, vectorized, and prepared for hybrid retrieval without you standing up a skillset by hand.
Agentic retrieval features are generally available in the 2026-04-01
REST API. The 2026-05-01-preview
version exposes the fuller feature set, including preview knowledge source kinds and the ability to attach an LLM to non-web sources. Blob Storage is a generally available indexed source kind, which is exactly what we need.
Point a knowledge source at the curated container.
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)
That description
field is not decoration. When a knowledge base holds several sources, the retrieval engine plans which sources to query, and the description is the primary signal it uses to route. Write it like you are briefing a colleague who has never seen your data.
Now the knowledge base.
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)
The retrieval engine plans which sources to query and performs iterative search when the first pass does not clear its relevance bar. Iterative search depends on setting a medium retrieval reasoning effort, either on the knowledge base or per request. That single knob is also the biggest lever on both latency and spend, so treat it as a tuning parameter rather than a set-and-forget value.
| Reasoning effort | What the engine does | Good fit for |
|---|---|---|
| Minimal | Single pass, extractive results, no query planning | Lookup-style questions where the user names the meeting |
| Low | Light query decomposition across sources | Most interactive chat traffic |
| Medium | Iterative search plus richer planning over sources | Analytical questions spanning many meetings |
With the knowledge base in place, the agent is short. Agent operations in the 2.x SDK are built on the Responses protocol, and agents are versioned objects created with 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)
Rule three is doing real work. Diarization gives you stable speaker identifiers within a recording, not identities, so you get generic labels rather than names. If the instructions do not forbid it, a capable model will cheerfully infer that Speaker 2 is the person whose name appears in the meeting title, and it will be wrong roughly as often as it is right. If you need real names, map them yourself in the chunking stage from calendar metadata or from multichannel capture, and write the resolved name into the record.
Calling the agent looks like any Responses call.
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)
Threading through previous_response_id
keeps the conversation server-side, which means you are not shipping a growing transcript of the chat on every turn and you are not writing your own history store.
Two failure classes matter in production and they want different handling. Transient service errors want retries. Empty or weak retrieval wants a different answer, not a retry, because running the same query again against the same index returns the same nothing.
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 on the backoff is not optional at any real concurrency. Without it your retries synchronize into a thundering herd and you turn a brief throttle into a sustained one.
For the retrieval side, the answer is to make the agent's failure visible rather than silent. Instruction two above tells the model to say it found nothing, and you should assert on that in your evaluation set. A grounded system that admits ignorance is far more valuable than one that produces confident prose from three irrelevant chunks, and the second failure mode is much harder to notice in production because the output looks fine.
Two separate quality questions live in this pipeline and they need separate measurement. The transcription layer has an accuracy problem measured in word error rate. The retrieval and generation layer has a groundedness problem measured by a judge model. A regression in either one looks identical from the outside, which is a good argument for measuring them apart.
Build a golden set first. A hundred or so questions written against meetings you have actually listened to is worth more than a thousand synthetic ones, because the value is in the expected answers and only a human who sat through the meeting can write those. Cover the awkward shapes deliberately. Include questions whose answer is genuinely absent so you can measure refusal behavior. Include questions that span two meetings. Include questions where two people disagreed.
{"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"}
The evaluation operations live on the project client in the 2.x SDK, under properties such as evaluators
, evaluation_rules
, and schedules
. For groundedness and relevance you use built-in judge evaluators. For word error rate you register a custom evaluator, because that one is arithmetic rather than judgment.
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)
Hand-correct twenty minutes of audio across three or four recordings and keep it as your reference. Twenty minutes sounds thin and it is, but it catches the failures that matter, which are domain vocabulary and acronyms coming back as phonetic mush. If your WER on product names is bad, the fix is a phrase list rather than a better model. Phrase lists let you hand the recognizer a set of words likely to appear, and they move the needle hard on proper nouns and internal jargon.
The metrics worth gating a deploy on are these four.
| Metric | What it catches | Where it comes from |
|---|---|---|
| Word error rate on domain terms | Vocabulary drift, new product names, bad audio | Custom evaluator against hand-corrected reference |
| Groundedness | Answers not supported by retrieved chunks | Built-in judge evaluator |
| Citation validity | Fabricated meeting titles, timecodes outside the recording | Deterministic check against chunk metadata |
| Refusal rate on absent answers | Confident invention when nothing was retrieved | Golden set questions with no supporting content |
Citation validity is the cheap one everyone skips. You already have the chunk metadata, so parsing the citations out of the answer and asserting that each meeting title exists and each timecode falls inside that recording's duration is maybe thirty lines of code. It catches a specific and embarrassing failure that judge models are surprisingly forgiving of.
Reindex on a schedule and expect churn. Foundry IQ triggers indexing and data synchronization automatically for indexed sources, but your curated container is the contract. If you change chunking strategy, you are rewriting every record, and a full reindex of a large corpus is not instant. Version your chunking logic and write the version into each record so you can tell mixed-generation content apart during a migration.
Decide the permission model before you index anything. Meeting recordings are among the most sensitive content an organization has. Retrieval in Foundry IQ respects user permissions for supported knowledge source types, and for the remote SharePoint source, Purview sensitivity labels and data classifications flow through the indexing and retrieval pipeline. Blob-backed sources do not give you that for free. If access control per meeting matters, either enforce it with security filters at query time using a field on each chunk, or keep recordings in SharePoint and use the remote source, where content never leaves SharePoint and SharePoint enforces permissions. Retrofitting this later means reindexing everything and auditing every conversation that already happened.
Instrument with tracing from day one. The projects SDK ships GenAI tracing instrumentation, currently an experimental preview where spans and attributes may change between versions. Turn it on anyway. When a user says the agent gave a bad answer, you want the retrieved chunk IDs and the query plan from that exact response, and reconstructing them after the fact from logs you did not write is miserable.
Watch the two meters. Retrieval bills token usage for subquery execution and semantic reranking, and the model you attach for query planning and answer synthesis bills separately on the model side. Reasoning effort, source count, and how much content you route into synthesis are the levers, in that order.
Plan the migration if you are on the old pattern. If you are still using Azure OpenAI On Your Data, the "Add your data" flow in the classic chat playground, it is deprecated and retires on October 14, 2026. The official migration target is exactly the stack in this article, which is Foundry Agent Service plus Foundry IQ.
The obvious alternative is a hand-built stack. Whisper for transcription behind your own GPU or an inference endpoint, pyannote for diarization, your own chunker, a vector database, and LangChain or a custom orchestrator on top. That stack is genuinely good and it is genuinely more work. The honest comparison looks like this.
| Concern | Foundry with fast transcription and Foundry IQ | Self-hosted Whisper plus pyannote plus a vector DB | Amazon Transcribe plus Bedrock Knowledge Bases | Google Speech-to-Text plus Vertex AI Search |
|---|---|---|---|---|
| Diarization | Built into the same call, up to 35 speakers | Separate model, separate tuning, best-in-class quality achievable | Built into the transcription job | Built into the recognizer |
| Time to first working answer | Hours | Days to weeks | Hours | Hours |
| Retrieval planning | Agentic, multi-query, iterative at higher effort | Whatever you write | Managed retrieval, less query planning | Managed retrieval with good semantic ranking |
| Permission-aware retrieval | Native for supported sources, Purview labels honored for remote SharePoint | You build it | IAM-scoped, coarser at the chunk level | IAM-scoped |
| Where the audio goes | Your Azure region | Wherever you run it, including fully on-premises | Your AWS region | Your GCP region |
| Escape hatch | Knowledge bases callable from any app through the Search APIs | Total control | Bedrock APIs | Vertex APIs |
The self-hosted path wins on two things and they are not small. One is cost at very high volume, because at some point per-minute transcription pricing loses to a GPU you already own. The other is data residency in the strict sense, meaning audio that legally cannot leave your premises. If neither applies to you, the managed path buys back weeks of work you would otherwise spend on chunking heuristics and retry logic.
Within Azure there is also a smaller decision, which is fast transcription against batch transcription. Fast wins on latency and simplicity for files under the size limit. Batch wins when you need to process very large archives asynchronously, when you want webhook notifications on completion, or when you want to bring your own storage account for the outputs.
The pipeline above is the spine. The interesting extensions hang off the chunking stage, because that is where you decide what the retrieval layer is even capable of answering. Extracting action items into a structured field lets you answer "what did I commit to last month" without any retrieval creativity. Writing a sentiment or disagreement flag onto each turn lets the agent find contested moments directly. Adding a second knowledge source pointed at your specs and design docs turns "what did we decide" into "what did we decide and does the shipped code match", and because a knowledge base fronts multiple sources behind one endpoint, that is a configuration change rather than an architecture change.
The part worth protecting as you extend is the evaluation loop. Meeting corpora grow continuously and unevenly, and a retrieval strategy tuned on six months of transcripts behaves differently on three years. The golden set is what tells you when that has happened.