We are building a low-latency meeting insight pipeline that chunks long audio transcripts, analyzes them in parallel, and merges the results into a structured report. This helps engineering teams turn raw meeting recordings into actionable summaries without waiting for a single massive LLM call to finish.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai - A sample transcript to analyze. You can generate one with Oxlo.ai's Whisper endpoint or paste your own text.
Step 1: Chunk the transcript
I split transcripts on sentence boundaries with a small overlap window. This keeps each chunk large enough to retain speaker context but small enough to avoid long processing delays. Overlap prevents action items that span sentence boundaries from getting dropped.
import re
def chunk_transcript(text: str, chunk_size: int = 800, overlap: int = 100):
sentences = re.split(r'(?<=[.!?])\s+', text)
chunks = []
current_chunk = []
current_len = 0
for sentence in sentences:
sentence_len = len(sentence.split())
if current_len + sentence_len > chunk_size and current_chunk:
chunks.append(" ".join(current_chunk))
# carry over overlap sentences
overlap_sentences = []
overlap_len = 0
for s in reversed(current_chunk):
if overlap_len + len(s.split()) > overlap:
break
overlap_sentences.insert(0, s)
overlap_len += len(s.split())
current_chunk = overlap_sentences
current_len = overlap_len
current_chunk.append(sentence)
current_len += sentence_len
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
Step 2: Write the system prompt
The system prompt forces valid JSON output for every chunk. Structured output makes the merge step deterministic and avoids regex parsing later.
SYSTEM_PROMPT = """You are an audio transcript analyst. Analyze the provided transcript chunk and return strict JSON with no markdown formatting.
Required fields:
- "summary": string, 1-2 sentences of what happened
- "action_items": list of strings, tasks assigned with owner if mentioned
- "sentiment": string, one of positive, neutral, negative
- "key_topics": list of strings, main discussion topics
- "speaker_issues": list of strings, any problems or blockers raised
Rules:
- Return only valid JSON.
- If a field has no data, use an empty string or empty list.
- Do not guess information outside the provided text."""Step 3: Analyze chunks in parallel
Sequential chunk analysis adds up quickly on long recordings. I fire all chunk requests concurrently through Oxlo.ai. Because Oxlo.ai charges per request rather than per token, parallelizing does not inflate cost, and the flat per-request pricing makes this predictable even when chunks vary in length.
import json
import concurrent.futures
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def analyze_chunk(chunk: str, chunk_id: int):
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Transcript chunk {chunk_id}:\n{chunk}"},
],
temperature=0.1,
)
raw = response.choices[0].message.content
return json.loads(raw)
def analyze_all_chunks(chunks: list[str]):
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(analyze_chunk, c, i): i
for i, c in enumerate(chunks)
}
results = [None] * len(chunks)
for future in concurrent.futures.as_completed(futures):
idx = futures[future]
results[idx] = future.result()
return results
Step 4: Merge the results
Parallel chunks can duplicate action items or split context. A second LLM pass reconciles the partial analyses into one coherent report. I use kimi-k2.6 here because its advanced reasoning and long context window handle the merged JSON array cleanly.
MERGE_PROMPT = """You are a senior editor. Merge the following JSON analysis chunks into one coherent report.
Input: a JSON array of chunk analyses.
Output: strict JSON with:
- "executive_summary": string
- "all_action_items": list of unique strings
- "overall_sentiment": string
- "all_key_topics": list of unique strings
- "critical_issues": list of strings
Remove duplicates and resolve contradictions."""
def merge_results(chunk_results: list[dict]):
payload = json.dumps(chunk_results, indent=2)
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": MERGE_PROMPT},
{"role": "user", "content": payload},
],
temperature=0.1,
)
raw = response.choices[0].message.content
clean = raw.strip()
if clean.startswith("
```json"):
clean = clean[7:]
elif clean.startswith("```
"):
clean = clean[3:]
if clean.endswith("
```
"):
clean = clean[:-3]
return json.loads(clean.strip())
Step 5: Stream the final report
For real-time dashboards, I stream the final narrative so users see output immediately. Oxlo.ai serves these requests with no cold starts on popular models, so the first token arrives fast.
def stream_final_report(merged: dict):
prompt = f"Turn the following structured analysis into a concise narrative report.\n\n{json.dumps(merged, indent=2)}"
stream = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": "You write concise executive summaries from structured data."},
{"role": "user", "content": prompt},
],
stream=True,
)
print("Final Report:")
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
print()
Run it
Here is the full integration. I use a short synthetic transcript, but the same code handles hour-long recordings if you increase the worker count.
if __name__ == "__main__":
raw_transcript = """
Alice: We need to ship the auth refactor by Friday. Bob, can you handle the OAuth tests?
Bob: Sure, but I am blocked on the dev environment. The new staging box is down.
Alice: I will open a ticket with IT right now. Carol, please review the PR once Bob pushes.
Carol: Will do. Also, the latency on the new endpoint is still at 400ms. We should profile it before release.
Alice: Good catch. Let us add profiling to the sprint. I am feeling confident about the timeline if we fix the staging issue today.
""".strip()
chunks = chunk_transcript(raw_transcript, chunk_size=50, overlap=10)
print(f"Split into {len(chunks)} chunks")
analyses = analyze_all_chunks(chunks)
final = merge_results(analyses)
stream_final_report(final)
Example output:
Split into 2 chunks
Final Report:
The team discussed shipping the auth refactor by Friday. Bob agreed to handle OAuth testing but is blocked by a down staging environment, which Alice will escalate to IT. Carol committed to PR review and raised concerns about 400ms endpoint latency, prompting the team to add profiling to the sprint. Overall sentiment is positive contingent on resolving the staging issue today. Action items include Bob completing OAuth tests, Alice opening an IT ticket, Carol reviewing the PR, and the team profiling endpoint latency before release.
Next steps
Swap the transcript input for audio files by piping Oxlo.ai's Whisper endpoint output directly into the chunking function. If you need to process many recordings daily, review the Oxlo.ai pricing page to pick a plan that matches your request volume. For even lower latency on the merge step, try switching the merge model to qwen-3-32b, which handles multilingual reasoning efficiently if your transcripts include non-English segments.
Top comments (0)