Build a Meeting Audio Action-Item Extractor with Thinking Machines Inkling

pip install openai pydantic fastapi uvicorn pydub
export FIREWORKS_API_KEY="your-key-here"

import base64
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.fireworks.ai/inference/v1",
api_key=os.environ["FIREWORKS_API_KEY"],
)
MODEL = "accounts/fireworks/models/inkling" # check your provider's catalog for the exact id
with open("standup_monday.mp3", "rb") as f:
audio_b64 = base64.b64encode(f.read()).decode()
response = client.chat.completions.create(
model=MODEL,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "List every action item that was agreed in this meeting, with who owns i
{"type": "input_audio", "input_audio": {"data": audio_b64, "format": "mp3"}},
],
}],
)
print(response.choices[0].message.content)

import json
from pydantic import BaseModel, Field, ValidationError
class ActionItem(BaseModel):
task: str = Field(description="What needs to be done, one sentence")
owner: str = Field(description="Person responsible, exactly as named in the audio")
deadline: str | None = Field(default=None, description="Deadline as stated, null if none was said")
quote: str = Field(description="Short verbatim quote containing the commitment")
class ActionItems(BaseModel):
items: list[ActionItem]
EXTRACT_PROMPT = f"""Listen to this meeting recording. Extract every action item that a specific person commi
Return ONLY valid JSON matching this schema, no markdown fences, no commentary:
{json.dumps(ActionItems.model_json_schema(), indent=2)}
Rules: an action item needs a named owner. General discussion is not an action item.
If no deadline was spoken, use null. Never invent one."""
def extract_items(audio_b64: str, fmt: str) -> ActionItems:
for attempt in range(2):
response = client.chat.completions.create(
model=MODEL,
temperature=0,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": EXTRACT_PROMPT},
{"type": "input_audio", "input_audio": {"data": audio_b64, "format": fmt}},
],
}],
)
raw = response.choices[0].message.content.strip()
raw = raw.removeprefix("```json").removesuffix("```").strip()
try:
return ActionItems.model_validate_json(raw)
except ValidationError:
if attempt == 1:
raise
raise RuntimeError("unreachable")

from difflib import get_close_matches
ROSTER = ["Sarah Chen", "Tom Okafor", "Danielle Ruiz", "Marcus Webb"]
FIRST_NAMES = {name.split()[0].lower(): name for name in ROSTER}
def resolve_owner(spoken: str) -> str:
first = spoken.strip().split()[0].lower()
if first in FIRST_NAMES:
return FIRST_NAMES[first]
match = get_close_matches(first, FIRST_NAMES.keys(), n=1, cutoff=0.75)
return FIRST_NAMES[match[0]] if match else f"UNRESOLVED: {spoken}"

from fastapi import FastAPI, UploadFile, HTTPException
app = FastAPI()
ALLOWED = {"mp3", "wav", "m4a"}
@app.post("/extract")
async def extract(file: UploadFile):
fmt = (file.filename or "").rsplit(".", 1)[-1].lower()
if fmt not in ALLOWED:
raise HTTPException(422, f"unsupported format: {fmt}")
audio_b64 = base64.b64encode(await file.read()).decode()
result = extract_items(audio_b64, fmt)
for item in result.items:
item.owner = resolve_owner(item.owner)
return result.model_dump()

import io
from pydub import AudioSegment # requires ffmpeg installed
SEGMENT_MINUTES = 20
def extract_long(path: str) -> ActionItems:
audio = AudioSegment.from_file(path)
step = SEGMENT_MINUTES * 60 * 1000
merged: list[ActionItem] = []
for start in range(0, len(audio), step):
buf = io.BytesIO()
audio[start:start + step].export(buf, format="mp3")
chunk_b64 = base64.b64encode(buf.getvalue()).decode()
merged.extend(extract_items(chunk_b64, "mp3").items)
seen: dict[str, ActionItem] = {}
for item in merged:
seen.setdefault(item.task.lower()[:60], item)
return ActionItems(items=list(seen.values()))

import base64
import io
import json
import os
from difflib import get_close_matches
from fastapi import FastAPI, HTTPException, UploadFile
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError
from pydub import AudioSegment
client = OpenAI(
base_url="https://api.fireworks.ai/inference/v1",
api_key=os.environ["FIREWORKS_API_KEY"],
)
MODEL = "accounts/fireworks/models/inkling"
ROSTER = ["Sarah Chen", "Tom Okafor", "Danielle Ruiz", "Marcus Webb"]
FIRST_NAMES = {name.split()[0].lower(): name for name in ROSTER}
SEGMENT_MINUTES = 20
ALLOWED = {"mp3", "wav", "m4a"}
class ActionItem(BaseModel):
task: str = Field(description="What needs to be done, one sentence")
owner: str = Field(description="Person responsible, exactly as named in the audio")
deadline: str | None = Field(default=None, description="Deadline as stated, null if none was said")
quote: str = Field(description="Short verbatim quote containing the commitment")
class ActionItems(BaseModel):
items: list[ActionItem]
EXTRACT_PROMPT = f"""Listen to this meeting recording. Extract every action item that a specific person commi
Return ONLY valid JSON matching this schema, no markdown fences, no commentary:
{json.dumps(ActionItems.model_json_schema(), indent=2)}
Rules: an action item needs a named owner. General discussion is not an action item.
If no deadline was spoken, use null. Never invent one."""
def extract_items(audio_b64: str, fmt: str) -> ActionItems:
for attempt in range(2):
response = client.chat.completions.create(
model=MODEL,
temperature=0,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": EXTRACT_PROMPT},
{"type": "input_audio", "input_audio": {"data": audio_b64, "format": fmt}},
],
}],
)
raw = response.choices[0].message.content.strip()
raw = raw.removeprefix("```json").removesuffix("```").strip()
try:
return ActionItems.model_validate_json(raw)
except ValidationError:
if attempt == 1:
raise
raise RuntimeError("unreachable")
def resolve_owner(spoken: str) -> str:
first = spoken.strip().split()[0].lower()
if first in FIRST_NAMES:
return FIRST_NAMES[first]
match = get_close_matches(first, FIRST_NAMES.keys(), n=1, cutoff=0.75)
return FIRST_NAMES[match[0]] if match else f"UNRESOLVED: {spoken}"
def segments(data: bytes, fmt: str):
audio = AudioSegment.from_file(io.BytesIO(data), format=fmt)
step = SEGMENT_MINUTES * 60 * 1000
for start in range(0, len(audio), step):
buf = io.BytesIO()
audio[start:start + step].export(buf, format="mp3")
yield base64.b64encode(buf.getvalue()).decode()
app = FastAPI()
@app.post("/extract")
async def extract(file: UploadFile):
fmt = (file.filename or "").rsplit(".", 1)[-1].lower()
if fmt not in ALLOWED:
raise HTTPException(422, f"unsupported format: {fmt}")
data = await file.read()
merged: list[ActionItem] = []
for chunk_b64 in segments(data, fmt):
merged.extend(extract_items(chunk_b64, "mp3").items)
seen: dict[str, ActionItem] = {}
for item in merged:
item.owner = resolve_owner(item.owner)
seen.setdefault(item.task.lower()[:60], item)
return ActionItems(items=list(seen.values())).model_dump()