오디오 리서치 노트북 만들기

pip install requests
export VENICE_API_KEY="your-key-here"

import io
import json
import os
import re
import wave
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests

BASE_URL = "https://api.venice.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VENICE_API_KEY']}"}

EMBED_MODEL = "text-embedding-bge-m3"
TTS_MODEL = "tts-xai-v1"
HOSTS = {"Ana": "luna", "Marco": "orion"}

sources = []
chunks = []

def default_text_model():
    response = requests.get(
        f"{BASE_URL}/models/traits", headers=HEADERS, params={"type": "text"}, timeout=60
    )
    response.raise_for_status()
    return response.json()["data"]["default"]


CHAT_MODEL = default_text_model()


def chat(messages, **options):
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=HEADERS,
        json={"model": CHAT_MODEL, "messages": messages, **options},
        timeout=300,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

def read_url(url):
    response = requests.post(
        f"{BASE_URL}/augment/scrape", headers=HEADERS, json={"url": url}, timeout=180
    )
    response.raise_for_status()
    return response.json()["content"]


def read_file(path):
    with open(path, "rb") as handle:
        response = requests.post(
            f"{BASE_URL}/augment/text-parser",
            headers=HEADERS,
            files={"file": (Path(path).name, handle)},
            timeout=180,
        )
    response.raise_for_status()
    return response.json()["text"]

def split(text, limit=1200):
    """문단을 반으로 자르지 않고 청크로 묶어 담는다."""
    packed, current = [], ""
    for para in re.split(r"\n\s*\n", text):
        para = para.strip()
        if not para:
            continue
        if current and len(current) + len(para) + 2 > limit:
            packed.append(current)
            current = para
        else:
            current = f"{current}\n\n{para}" if current else para
    if current:
        packed.append(current)
    return packed


def embed(texts):
    vectors = []
    for start in range(0, len(texts), 64):
        response = requests.post(
            f"{BASE_URL}/embeddings",
            headers=HEADERS,
            json={"model": EMBED_MODEL, "input": texts[start : start + 64]},
            timeout=180,
        )
        response.raise_for_status()
        vectors.extend(row["embedding"] for row in response.json()["data"])
    return vectors

def add_source(title, ref):
    text = read_url(ref) if ref.startswith("http") else read_file(ref)
    number = len(sources) + 1
    sources.append({"number": number, "title": title, "ref": ref})

    pieces = split(text)
    for piece, vector in zip(pieces, embed(pieces)):
        magnitude = sum(x * x for x in vector) ** 0.5
        chunks.append(
            {"source": number, "title": title, "text": piece,
             "vector": vector, "magnitude": magnitude}
        )
    print(f"[{number}] {title}: {len(text)} characters, {len(pieces)} chunks")

def retrieve(question, k=6):
    query = embed([question])[0]
    query_magnitude = sum(x * x for x in query) ** 0.5

    def similarity(chunk):
        dot = sum(a * b for a, b in zip(query, chunk["vector"]))
        return dot / (query_magnitude * chunk["magnitude"])

    return sorted(chunks, key=similarity, reverse=True)[:k]

def ask(question, k=6):
    hits = retrieve(question, k)
    notes = "\n\n".join(f"[{h['source']}] {h['title']}\n{h['text']}" for h in hits)
    answer = chat(
        [
            {"role": "system", "content": (
                "Answer only from the numbered notes. Cite every claim with the bracket number "
                "of the note it came from. If the notes do not answer the question, say so "
                "instead of filling the gap.")},
            {"role": "user", "content": f"Notes:\n\n{notes}\n\nQuestion: {question}"},
        ],
        temperature=0.2,
    )
    cited = sorted({int(n) for n in re.findall(r"\[(\d+)\]", answer)})
    return answer, [s for s in sources if s["number"] in cited]

DIALOGUE_SCHEMA = {
    "type": "json_schema",
    "json_schema": {
        "name": "dialogue",
        "strict": True,
        "schema": {
            "type": "object",
            "additionalProperties": False,
            "required": ["turns"],
            "properties": {
                "turns": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "additionalProperties": False,
                        "required": ["speaker", "text"],
                        "properties": {
                            "speaker": {"type": "string", "enum": list(HOSTS)},
                            "text": {"type": "string"},
                        },
                    },
                }
            },
        },
    },
}


def write_script(turns=16):
    """자료에 근거한 두 진행자의 대화를 chat 모델에 요청한다."""
    spread = chunks[:: max(1, len(chunks) // 12)][:12]
    notes = "\n\n".join(f"{c['title']}\n{c['text']}" for c in spread)
    hosts = " and ".join(HOSTS)
    raw = chat(
        [
            {"role": "system", "content": (
                f"You write podcast dialogue for two hosts, {hosts}. Ground every statement in "
                "the supplied notes. Write for the ear: no markdown, no URLs, no bracket "
                "citations, no stage directions. Spell out abbreviations the first time they "
                "appear. Vary the length of turns. Open with a hook and close with a takeaway.")},
            {"role": "user", "content": f"Notes:\n\n{notes}\n\nWrite about {turns} turns."},
        ],
        temperature=0.7,
        response_format=DIALOGUE_SCHEMA,
    )
    return json.loads(raw)["turns"]

def speak(turn):
    response = requests.post(
        f"{BASE_URL}/audio/speech",
        headers=HEADERS,
        json={"model": TTS_MODEL, "voice": HOSTS[turn["speaker"]],
              "input": turn["text"], "response_format": "wav"},
        timeout=300,
    )
    response.raise_for_status()
    with wave.open(io.BytesIO(response.content)) as clip:
        return clip.getparams(), clip.readframes(clip.getnframes())

def audio_overview(turns, path="overview.wav", pause_seconds=0.25):
    with ThreadPoolExecutor(max_workers=4) as pool:
        rendered = list(pool.map(speak, turns))

    params = rendered[0][0]
    silence = b"\x00" * int(params.framerate * params.sampwidth * params.nchannels * pause_seconds)
    with wave.open(path, "wb") as out:
        out.setnchannels(params.nchannels)
        out.setsampwidth(params.sampwidth)
        out.setframerate(params.framerate)
        for position, (_, frames) in enumerate(rendered):
            if position:
                out.writeframes(silence)
            out.writeframes(frames)
    return path

if __name__ == "__main__":
    add_source("Venice Privacy", "https://docs.venice.ai/overview/privacy")
    add_source("TEE and E2EE Models", "https://docs.venice.ai/guides/features/tee-e2ee-models")
    add_source("VVV and DIEM", "https://docs.venice.ai/overview/vvv-diem")

    answer, cited = ask("How does Venice keep my prompts private, and what do I give up?")
    print(answer)
    print("\nSources:", ", ".join(f"[{s['number']}] {s['title']}" for s in cited))

    turns = write_script()
    print(f"\nWriting {len(turns)} turns to overview.wav")
    audio_overview(turns)

python notebook.py

[1] Venice Privacy: 7507 characters, 7 chunks
[2] TEE and E2EE Models: 43859 characters, 40 chunks
[3] VVV and DIEM: 9609 characters, 11 chunks

Venice's privacy architecture is built around a proxy foundation. All requests pass
through Venice over HTTPS and are relayed to the model provider without Venice storing
your prompt or response content [1]. On top of that proxy, each model offers one of four
progressively stronger privacy modes [1]...

Sources: [1] Venice Privacy

Writing 23 turns to overview.wav