Ask your sales calls anything

export CALLS_DIR=~/sales-calls
python 07_data_and_browsers/sales_call_search.py
python 07_data_and_browsers/sales_call_search.py "What did customers say about our pricing in August?"

import os
import re
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import noclick

EMBEDDING_MODEL = "openrouter/openai/text-embedding-3-small"
PASSAGE_SECONDS = 60
WORKERS = 4
AUDIO = {".mp3", ".m4a", ".wav", ".ogg", ".webm", ".mp4"}

recordings = noclick.Bucket.from_name("sales-calls", create_if_missing=True)
jobs = noclick.Queue.from_name("sales-call-indexing", create_if_missing=True, max_attempts=3)
calls_db = noclick.Database.from_name("sales-calls", create_if_missing=True, engine="postgres")

MIGRATIONS = {
    "0001_calls.sql": """
create table calls (
    id text primary key,
    etag text not null,
    customer text,
    rep text,
    recorded_on date,
    duration_s real,
    cost_usd real,
    indexed_at timestamptz not null default now()
);

create table passages (
    call_id text not null references calls (id) on delete cascade,
    position integer not null,
    start_s real not null,
    end_s real not null,
    text text not null,
    embedding halfvec(1536) not null,
    search tsvector generated always as (to_tsvector('english', text)) stored,
    primary key (call_id, position)
);

create index passages_embedding on passages using hnsw (embedding halfvec_cosine_ops);
create index passages_search on passages using gin (search);
create index calls_recorded_on on calls (recorded_on);
""",
}

def upload(folder: str) -> int:
    uploaded = 0
    for path in sorted(Path(folder).expanduser().iterdir()):
        named = re.fullmatch(r"(\d{4}-\d{2}-\d{2})_([^_]+)_(.+)", path.stem)
        if path.suffix.lower() not in AUDIO or not named:
            continue
        recorded_on, customer, rep = named.groups()
        key = f"calls/{recorded_on[:7]}/{path.name}"
        try:
            if recordings.head(key).size == path.stat().st_size:
                continue
        except noclick.NotFound:
            pass
        with open(path, "rb") as f:
            recordings.put(key, f, metadata={
                "customer": customer.replace("-", " ").title(), "rep": rep.title(), "recorded_on": recorded_on,
            })
        uploaded += 1
    return uploaded

def queue_unindexed() -> int:
    indexed = {row["id"]: row["etag"] for row in calls_db.query("select id, etag from calls")}
    keys = [obj.key for obj in recordings.objects(prefix="calls/") if indexed.get(obj.key) != obj.etag]
    for start in range(0, len(keys), 100):
        jobs.put_many([{"key": key} for key in keys[start:start + 100]])
    return len(keys)

def clock(seconds: float) -> str:
    return f"{int(seconds) // 60:02d}:{int(seconds) % 60:02d}"


def to_passages(segments: list[noclick.TranscriptSegment]) -> list[dict]:
    groups, current = [], []
    for segment in segments:
        current.append(segment)
        if segment.end_s - current[0].start_s >= PASSAGE_SECONDS:
            groups.append(current)
            current = []
    if current:
        groups.append(current)
    return [
        {
            "start_s": group[0].start_s,
            "end_s": group[-1].end_s,
            "text": "\n".join(f"[{clock(s.start_s)}] {s.speaker}: {s.text.strip()}" for s in group),
        }
        for group in groups
    ]


def index_call(key: str) -> str:
    info = recordings.head(key)
    done = calls_db.query("select etag from calls where id = ?", [key]).rows
    if done and done[0]["etag"] == info.etag:
        return f"{key} is already indexed"
    customer, rep, recorded_on = (info.metadata.get(k) for k in ("customer", "rep", "recorded_on"))

    words = noclick.AI.transcribe(recordings.signed_url(key, expires_in=3600), speakers=True)
    passages = to_passages(words.segments)
    context = f"Sales call with {customer} on {recorded_on}, rep {rep}."
    embedded = noclick.AI.embed([f"{context}\n{p['text']}" for p in passages], model=EMBEDDING_MODEL)

    statements = [
        ("delete from calls where id = ?", [key]),
        (
            "insert into calls (id, etag, customer, rep, recorded_on, duration_s, cost_usd) "
            "values (?, ?, ?, ?, cast(? as date), ?, ?)",
            [key, info.etag, customer, rep, recorded_on, words.duration_s,
             (words.cost_usd or 0) + (embedded.cost_usd or 0)],
        ),
    ]
    rows = [
        [key, position, p["start_s"], p["end_s"], p["text"], vector]
        for position, (p, vector) in enumerate(zip(passages, embedded.vectors))
    ]
    for start in range(0, len(rows), 20):
        chunk = rows[start:start + 20]
        values = ", ".join(["(?, ?, ?, ?, ?, cast(? as halfvec))"] * len(chunk))
        statements.append((
            f"insert into passages (call_id, position, start_s, end_s, text, embedding) values {values}",
            [value for row in chunk for value in row],
        ))
    calls_db.batch(statements)
    return f"{key}: {len(passages)} passages, {clock(words.duration_s or 0)} of audio"

def work() -> None:
    for msg in jobs.iterate(lease_s=1800, stop_when_empty=True):
        try:
            print(index_call(msg.body["key"]))
        except (noclick.PaymentRequired, noclick.SpendCapExceeded):
            msg.release()
            raise
        except Exception as err:
            msg.nack(delay_s=60, error=f"{type(err).__name__}: {err}")
        else:
            msg.ack()


def drain() -> None:
    with ThreadPoolExecutor(WORKERS) as pool:
        for future in [pool.submit(work) for _ in range(WORKERS)]:
            future.result()

INSTRUCTIONS = f"""You answer questions about our recorded sales calls by querying the sales-calls
database with SQL. You can only read it.

Tables:
- calls: one row per recording. id, customer, rep (our salesperson), recorded_on (a date),
  duration_s.
- passages: each call's transcript in passages of about a minute. call_id, start_s and end_s
  (seconds into the call), text (lines of "[mm:ss] speaker: words", with speakers labelled per
  call; the rep is the one presenting our product), embedding, and search, a full-text index of
  text.

Searching:
- For words people say exactly (a product, a competitor, "SSO"), use full text and include the
  other ways people say it: where search @@ websearch_to_tsquery('english',
  'sso OR "single sign-on" OR saml OR okta').
- For ideas phrased many ways ("worried about the price"), rank by meaning:
  order by embedding <=> noclick.embed('worried about the price', '{EMBEDDING_MODEL}') limit 20.
- Filter dates on calls.recorded_on. "Last month" is the previous calendar month.
- Read the passages before you answer: a match isn't always the customer asking.

Answer with who asked, what they asked in their own words, and when. Quote the customer, not our
rep. After each quote, give the call's date, the time into the call, and a link written exactly
as [listen](call:<call id>@<seconds>), with the passage's start_s. If nothing matches, say so."""


def define_analyst() -> noclick.Agent:
    return noclick.Agent.from_name(
        "call-analyst",
        create_if_missing=True,
        instructions=INSTRUCTIONS,
        tools=[calls_db.only("query", "schema")],
    )


def link_moments(text: str) -> str:
    def link(match: re.Match) -> str:
        url = recordings.signed_url(match["key"], expires_in=7 * 86400)
        return f"({url}#t={match['seconds']})"

    return re.sub(r"\(call:(?P<key>[^@\s)]+)@(?P<seconds>\d+)\)", link, text)


def ask(analyst: noclick.Agent, question: str) -> str:
    return link_moments(analyst.message(question).text)

def main() -> None:
    calls_db.migrate(MIGRATIONS)
    if os.environ.get("CALLS_DIR"):
        print(f"Uploaded {upload(os.environ['CALLS_DIR'])} new recordings")
    drain()
    print(f"Queued {queue_unindexed()} recordings to index")
    drain()
    for letter in jobs.dead_letters():
        print(f"Couldn't index {letter.body['key']} after {letter.attempts} tries: {letter.last_error}")
    question = " ".join(sys.argv[1:]) or "Which customers asked about SSO last month?"
    print(f"\n{question}\n")
    print(ask(define_analyst(), question))


if __name__ == "__main__":
    main()

noclick.Agent.from_name("call-analyst").delete()
noclick.Queue.from_name("sales-call-indexing").delete()
noclick.Database.from_name("sales-calls").delete()
noclick.Bucket.from_name("sales-calls").delete(force=True)