diff --git a/README.md b/README.md index 03e12c9..28c039e 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,10 @@ Manifest (parquet / jsonl / csv) doc_id, s3_url │ ▼ Stage 1 — audio/workflow/analyze.py - │ Daft: read manifest → download audio bytes → duration filter + │ Daft: read manifest → download audio bytes → duration gate + │ (rows are never dropped: failures are kept with a status + │ column — download_failed / decode_failed / duration_filtered + │ / llm_failed — and null analysis fields) │ → SenseVoice ASR (transcript + acoustic_emotion) │ → PII redaction (ID card, phone numbers) │ → DeepSeek LLM (downgrade_related, bad_tone, emotion_score …) @@ -40,7 +43,7 @@ Manifest (parquet / jsonl / csv) │ → validate blob v2 │ ▼ Lance asset table (blob v2, local or S3) - │ columns: doc_id, s3_url, audio_blob, + │ columns: doc_id, s3_url, status, audio_blob, │ transcript, acoustic_emotion, │ downgrade_related, primary_reason, │ secondary_reason, summary, confidence, @@ -103,6 +106,11 @@ EMBED_BACKEND=signal # signal (128-dim RMS+ZCR) or wav2vec2 # Daft runner USE_RAY=0 # set to 1 to use Ray for Daft-backed steps RAY_ADDRESS= # leave empty to start/join local Ray + +# Daft's native S3 writes honor HTTP_PROXY/HTTPS_PROXY; when a proxy is set, +# keep local MinIO traffic off it or multipart uploads fail with 400 +NO_PROXY=127.0.0.1,localhost +no_proxy=127.0.0.1,localhost ``` ### Tuning for large Ray runs @@ -440,7 +448,10 @@ create the unified asset table once, then local and LLM batches can share it. The default `--num-partitions 16` requires at least 4096 rows. For tables with fewer rows, pass `--num-partitions 1` (or skip `--embedding` and rely on scalar queries only). **DeepSeek key absent → LLM columns are null.** -If `DEEPSEEK_API_KEY` is not set, `downgrade_related`, `bad_tone`, `primary_reason`, `summary`, `confidence`, `text_emotion`, and `emotion_score` are all `null`. ASR and acoustic embeddings still run normally. +If `DEEPSEEK_API_KEY` is not set, `downgrade_related`, `bad_tone`, `primary_reason`, `summary`, `confidence`, `text_emotion`, and `emotion_score` are all `null` (`status` stays `ok`). ASR and acoustic embeddings still run normally. When the key **is** set but a call fails or returns unparsable JSON, the row is kept with `status = 'llm_failed'` and null LLM columns — a null verdict always means "not judged", never "judged no". + +**Audio rows are never dropped.** +Like the image pipeline, every audio manifest entry produces exactly one output row. Failures are marked in the `status` column (`ok` / `download_failed` / `decode_failed` / `duration_filtered` / `llm_failed`, plus `blob_download_failed` stamped at ingest when a Stage-1-ok object has disappeared before Stage 2), with null transcript/verdict columns. Rows outside `[MIN_DURATION_S, MAX_DURATION_S]` keep their measured `duration_s` but skip ASR/LLM/embedding. **Blob v2 is validated after every ingest.** `validate_blob_v2` raises immediately if Lance silently downgraded `audio_blob` to `large_binary`. Never skip this check when testing new library versions. diff --git a/multimodal_toolkit/audio/llm.py b/multimodal_toolkit/audio/llm.py new file mode 100644 index 0000000..ceecb68 --- /dev/null +++ b/multimodal_toolkit/audio/llm.py @@ -0,0 +1,162 @@ +"""音频 LLM 打标的响应校验与 OpenAI-compatible provider 适配层。 + +与 image/vlm.py 的适配器解决同一个 Daft 0.7.15 问题（媒体隔离约定下各自 +持有一份）： + +1. Daft 的 prompt UDF 会把 UDF 级选项（on_error / concurrency / ...）原样 + 留在 OpenAIPrompter.generation_config 里，OpenAI SDK 收到未知 kwarg 直接 + 抛 TypeError——每一行都失败，整批打标全军覆没。descriptor 适配器在 + 实例化 prompter 前剔除这些 UDF-only 选项，其余请求参数原样保留。 +2. 非 ok 行（下载/解码失败、时长过滤）的 prompt 为 null，不能进入 Daft 的 + 消息分发器（0.7.15 里一个 None 能带崩整个 morsel），也不该浪费一次 API + 调用；prompter 对含 null 的输入直接返回 null。 +3. JSON mode 不提供字段类型和业务约束保证，而 Daft 0.7.15 的 + try_deserialize 遇到合法 JSON 中的错类型会终止整个批次。因此本模块在 + Python batch UDF 中逐行严格校验，坏响应降级为全 null，由 workflow 标记 + llm_failed，确保单行模型异常不会破坏“行不丢弃”语义。 + +历史教训：llm_prompt 的 concurrency/on_error 参数是 #18 引入的，引入即 +触发泄漏、每行失败，但当时结论列有 fill_null 缺省值兜底——输出全是 +downgrade_related=False，看起来像"LLM 判定为否"，回归完全不可见。 +结论列保持 null + status=llm_failed 之后它才第一次暴露出来。 +""" +from __future__ import annotations + +import json +import math +from dataclasses import dataclass + +import daft +from daft.ai.openai.protocols.prompter import OpenAIPrompter, OpenAIPrompterDescriptor +from daft.ai.openai.provider import OpenAIProvider +from daft.ai.typing import UDFOptions + +from . import config +from .udfs import ANALYSIS_DTYPE + + +# 校验失败时返回字段齐全、值全为 null 的 struct。不能直接依赖 Daft 的 +# try_deserialize：0.7.15 遇到“JSON 语法合法但字段类型错误”的响应会抛异常， +# 一个坏响应就能终止整个批次；缺字段对象又会被解析成非 null struct，无法 +# 正确标记 llm_failed。 +_NULL_ANALYSIS = { + "downgrade_related": None, + "primary_reason": None, + "secondary_reason": None, + "summary": None, + "confidence": None, + "text_emotion": None, + "bad_tone": None, + "emotion_score": None, +} +_EXPECTED_FIELDS = set(_NULL_ANALYSIS) + + +def _valid_score(value) -> bool: + """只接受 0 到 1 的有限实数；bool 虽是 int 子类，但不是合法分数。""" + return ( + type(value) in (int, float) + and math.isfinite(value) + and 0.0 <= value <= 1.0 + ) + + +def validate_llm_response(raw: str | None) -> dict: + """把单条模型响应校验为规范结论，任何异常都降级为全 null。 + + JSON mode 只能保证响应大概率是 JSON，不能保证字段完整、类型正确或枚举/ + 分数满足业务约束。因此这里严格要求字段集合与 prompt 契约一致，避免半条 + 结论被当成成功，也确保坏响应不会逃逸成 Daft 的批次级异常。 + """ + try: + payload = json.loads(raw) if raw is not None else None + if not isinstance(payload, dict) or set(payload) != _EXPECTED_FIELDS: + return dict(_NULL_ANALYSIS) + + if type(payload["downgrade_related"]) is not bool: + return dict(_NULL_ANALYSIS) + if type(payload["bad_tone"]) is not bool: + return dict(_NULL_ANALYSIS) + if payload["primary_reason"] not in config.PRIMARY_REASONS: + return dict(_NULL_ANALYSIS) + if payload["text_emotion"] not in config.TEXT_EMOTIONS: + return dict(_NULL_ANALYSIS) + + secondary_reason = payload["secondary_reason"] + summary = payload["summary"] + if not isinstance(secondary_reason, str) or not isinstance(summary, str): + return dict(_NULL_ANALYSIS) + if not summary.strip(): + return dict(_NULL_ANALYSIS) + if not _valid_score(payload["confidence"]): + return dict(_NULL_ANALYSIS) + if not _valid_score(payload["emotion_score"]): + return dict(_NULL_ANALYSIS) + + return { + "downgrade_related": payload["downgrade_related"], + "primary_reason": payload["primary_reason"], + "secondary_reason": secondary_reason.strip(), + "summary": summary.strip(), + "confidence": float(payload["confidence"]), + "text_emotion": payload["text_emotion"], + "bad_tone": payload["bad_tone"], + "emotion_score": float(payload["emotion_score"]), + } + except (TypeError, ValueError, OverflowError): + return dict(_NULL_ANALYSIS) + + +@daft.func.batch(return_dtype=ANALYSIS_DTYPE) +def validate_llm_responses(raw_responses): + """逐行校验响应，保持输入输出行数和 null 对齐。""" + return [validate_llm_response(raw) for raw in raw_responses.to_pylist()] + + +@dataclass +class _OnErrorSafePrompterDescriptor(OpenAIPrompterDescriptor): + def instantiate(self): + prompt_options = dict(self.prompt_options) + for name in UDFOptions.__dataclass_fields__: + prompt_options.pop(name, None) + return _NullSafeOpenAIPrompter( + provider_name=self.provider_name, + provider_options=self.provider_options, + model=self.model_name, + system_message=self.system_message, + return_format=self.return_format, + prompt_options=prompt_options, + ) + + +class _NullSafeOpenAIPrompter(OpenAIPrompter): + async def prompt(self, messages): + if any(message is None for message in messages): + return None + return await super().prompt(messages) + + +class AudioLLMProvider(OpenAIProvider): + def get_prompter(self, model=None, return_format=None, system_message=None, **options): + descriptor = super().get_prompter( + model, + return_format=return_format, + system_message=system_message, + **options, + ) + return _OnErrorSafePrompterDescriptor( + provider_name=descriptor.provider_name, + provider_options=descriptor.provider_options, + model_name=descriptor.model_name, + prompt_options=descriptor.prompt_options, + system_message=descriptor.system_message, + return_format=descriptor.return_format, + ) + + +def get_audio_llm_provider() -> AudioLLMProvider: + return AudioLLMProvider( + name="audio-llm", + base_url=config.DEEPSEEK_BASE_URL, + api_key=config.DEEPSEEK_API_KEY, + ) diff --git a/multimodal_toolkit/audio/schema.py b/multimodal_toolkit/audio/schema.py index 18a3af8..8049d89 100644 --- a/multimodal_toolkit/audio/schema.py +++ b/multimodal_toolkit/audio/schema.py @@ -8,6 +8,9 @@ [ pa.field("doc_id", pa.utf8()), pa.field("s3_url", pa.utf8()), + # 处理状态：ok / download_failed / decode_failed / duration_filtered / + # llm_failed / blob_download_failed。非 ok 行的转写和打标列为 null。 + pa.field("status", pa.utf8()), pa.field("audio_blob", pa.large_binary()), # written as lance blob v2 at ingest time pa.field("ingest_time", pa.timestamp("us", tz="UTC")), pa.field("audio_embedding", pa.list_(pa.float32(), config.EMBED_DIM)), diff --git a/multimodal_toolkit/audio/udfs.py b/multimodal_toolkit/audio/udfs.py index 311c1a0..294a437 100644 --- a/multimodal_toolkit/audio/udfs.py +++ b/multimodal_toolkit/audio/udfs.py @@ -19,6 +19,10 @@ @daft.func.batch(return_dtype=daft.DataType.float64()) def duration_udf(audio_bytes_col): + """音频时长（秒）。字节缺失或解码失败返回 null 而不是 0.0—— + 下游靠 duration 是否为 null 判定 decode_failed，0.0 会和 + "合法的零长音频"混在一起，无法区分"坏文件"和"空录音"。 + """ import io as _io import soundfile as sf @@ -26,22 +30,26 @@ def duration_udf(audio_bytes_col): results = [] for b in audio_bytes_col.to_pylist(): if not b: - results.append(0.0) + results.append(None) continue try: info = sf.info(_io.BytesIO(b)) - results.append(float(info.frames) / info.samplerate if info.samplerate else 0.0) + results.append(float(info.frames) / info.samplerate if info.samplerate else None) except Exception: - results.append(0.0) + results.append(None) return results @daft.func.batch(return_dtype=daft.DataType.string()) def prompt_udf(transcripts, acoustic_emotions): + """构造 LLM 打标提示词。transcript 为 null 说明该行下载/解码/ASR + 没有成功，返回 null 让 LLM 跳过——空字符串转写（真实的无语音录音） + 仍然正常构造提示词。 + """ from multimodal_toolkit.audio.prompt import build_prompt return [ - build_prompt(t or "", e or "NEUTRAL") + build_prompt(t, e or "NEUTRAL") if t is not None else None for t, e in zip(transcripts.to_pylist(), acoustic_emotions.to_pylist()) ] @@ -66,6 +74,11 @@ def __call__(self, audio_bytes_col, doc_ids): results = [] for audio_bytes, doc_id in zip(audio_bytes_col.to_pylist(), doc_ids.to_pylist()): + # 字节为 null（下载/解码失败或被时长过滤的行）→ 整个 struct 置 + # null，transcript 保持"没处理"语义，不能落成空字符串。 + if not audio_bytes: + results.append(None) + continue suffix = Path(doc_id).suffix if doc_id else ".wav" if not suffix: suffix = ".wav" diff --git a/multimodal_toolkit/audio/workflow/analyze.py b/multimodal_toolkit/audio/workflow/analyze.py index 5468088..3980277 100644 --- a/multimodal_toolkit/audio/workflow/analyze.py +++ b/multimodal_toolkit/audio/workflow/analyze.py @@ -11,6 +11,11 @@ default (no embed) → JSONL on S3 (scalar fields, embeddable in downstream JSON pipelines) --embed → Lance staging table on S3 (fixed-size vector cannot be stored in JSON) +manifest 里的每个条目都对应输出中的一行——下载失败、解码失败、被时长过滤、 +LLM 打标失败的通话不会被丢弃，而是通过 status 列标记（ok / download_failed / +decode_failed / duration_filtered / llm_failed），转写和打标结论为 null。 +合规场景必须能区分"通话有问题"和"根本没处理"。 + The output always includes s3_url so Stage 2 (ingest.py) can download the audio blob. """ from __future__ import annotations @@ -19,12 +24,12 @@ import daft from daft import col -from daft.functions import download, regexp_replace +from daft.functions import download, regexp_replace, when from daft.functions.ai import prompt as llm_prompt from .. import config +from ..llm import validate_llm_responses from ..udfs import ( - ANALYSIS_DTYPE, AsrUDF, duration_udf, prompt_udf, @@ -46,6 +51,7 @@ _BASE_OUTPUT_COLS = [ "doc_id", "s3_url", + "status", "duration_s", "transcript", "acoustic_emotion", @@ -93,19 +99,34 @@ def _build_analysis_df(manifest: str, io_config) -> daft.DataFrame: df = df.with_column( "audio_bytes", download(col("s3_url"), on_error="null", io_config=io_config) ) - df = df.where(~col("audio_bytes").is_null()) - # Duration gate: drop clips too short to analyse or too long to afford. + # manifest 里的每个条目都对应输出中的一行——下载失败、解码失败、被时长 + # 过滤的音频不会被丢弃，而是通过 status 列标记，转写和打标结论为 null。 + # 合规场景必须能区分"通话有问题"和"根本没处理"。duration_udf 对坏字节 + # 返回 null，所以 duration 为 null 即 decode_failed；时长超出 + # [MIN_DURATION_S, MAX_DURATION_S] 的行标 duration_filtered（太短没有 + # 分析价值，太长成本不可控），duration_s 本身保留。 df = df.with_column("duration_s", duration_udf(col("audio_bytes"))) - df = df.where( - (col("duration_s") >= config.MIN_DURATION_S) - & (col("duration_s") <= config.MAX_DURATION_S) + df = df.with_column( + "status", + when(col("audio_bytes").is_null(), "download_failed") + .when(col("duration_s").is_null(), "decode_failed") + .when( + (col("duration_s") < config.MIN_DURATION_S) + | (col("duration_s") > config.MAX_DURATION_S), + "duration_filtered", + ) + .otherwise("ok"), ) # ASR = automatic speech recognition. SenseVoice returns the transcript # plus an acoustic emotion label (angry/sad/... inferred from tone, not words). + # 非 ok 行传 null 字节进 UDF，ASR 返回 null struct，昂贵的模型推理只花在 + # 真正要分析的行上。 asr = AsrUDF() - df = df.with_column("asr", asr(col("audio_bytes"), col("doc_id"))) + df = df.with_column( + "asr", asr(when(col("status") == "ok", col("audio_bytes")), col("doc_id")) + ) df = df.with_column("transcript_raw", col("asr")["transcript"]) df = df.with_column("acoustic_emotion", col("asr")["acoustic_emotion"]) @@ -122,20 +143,19 @@ def _build_analysis_df(manifest: str, io_config) -> daft.DataFrame: # LLM tagging: build the instruction (audio/prompt.py), ask DeepSeek for a # JSON object with the business fields. Skipped when no API key is set — - # those columns then fall back to the fill_null defaults below. + # those columns then stay null（"没打标"，不是"判定为否"）。 + # 非 ok 行 transcript 为 null，prompt_udf 返回 null，LLM 不会收到请求。 df = df.with_column("prompt", prompt_udf(col("transcript"), col("acoustic_emotion"))) if config.DEEPSEEK_API_KEY: - from daft.ai.openai.provider import OpenAIProvider + # 不直接用 OpenAIProvider：Daft 0.7.15 会把 on_error/concurrency 泄漏进 + # OpenAI 请求 kwargs 导致每行 TypeError，适配层见 audio/llm.py。 + from ..llm import get_audio_llm_provider - provider = OpenAIProvider( - base_url=config.DEEPSEEK_BASE_URL, - api_key=config.DEEPSEEK_API_KEY, - ) df = df.with_column( "analysis_json", llm_prompt( col("prompt"), - provider=provider, + provider=get_audio_llm_provider(), model=config.DEEPSEEK_MODEL, use_chat_completions=True, response_format={"type": "json_object"}, @@ -147,23 +167,50 @@ def _build_analysis_df(manifest: str, io_config) -> daft.DataFrame: else: df = df.with_column("analysis_json", daft.lit(None).cast(daft.DataType.string())) - # Parse the LLM's JSON into typed columns. try_deserialize yields null on - # malformed JSON, and fill_null supplies a safe default per field, so one - # bad LLM response never fails the batch. - df = df.with_column("analysis", col("analysis_json").try_deserialize("json", ANALYSIS_DTYPE)) + # 在 Python batch UDF 里严格校验 LLM JSON，再生成 typed struct。不能用 + # try_deserialize：Daft 0.7.15 遇到合法 JSON 中的类型错误会抛批次级异常， + # 而缺字段对象会生成非 null struct，两个行为都会破坏 llm_failed 语义。 + # 结论列不做 fill_null：null 就是"LLM 没给出结论"（没配 key、请求失败、 + # JSON 不合法），填 False/'其他'/0.0 会把"没处理"伪装成"判定为否"， + # 真实降档通话会被静默漏报。 + df = df.with_column("analysis", validate_llm_responses(col("analysis_json"))) df = ( - df.with_column("downgrade_related", col("analysis")["downgrade_related"].fill_null(False)) - .with_column("primary_reason", col("analysis")["primary_reason"].fill_null("其他")) - .with_column("secondary_reason", col("analysis")["secondary_reason"].fill_null("")) - .with_column("summary", col("analysis")["summary"].fill_null("")) - .with_column("confidence", col("analysis")["confidence"].fill_null(0.0)) - .with_column("text_emotion", col("analysis")["text_emotion"].fill_null("未知")) - .with_column("bad_tone", col("analysis")["bad_tone"].fill_null(False)) - .with_column("emotion_score", col("analysis")["emotion_score"].fill_null(0.0)) + df.with_column("downgrade_related", col("analysis")["downgrade_related"]) + .with_column("primary_reason", col("analysis")["primary_reason"]) + .with_column("secondary_reason", col("analysis")["secondary_reason"]) + .with_column("summary", col("analysis")["summary"]) + .with_column("confidence", col("analysis")["confidence"]) + .with_column("text_emotion", col("analysis")["text_emotion"]) + .with_column("bad_tone", col("analysis")["bad_tone"]) + .with_column("emotion_score", col("analysis")["emotion_score"]) ) + + # 配了 key 却没有得到可解析结果的 ok 行标 llm_failed——区别于"没配 key + # 整批跳过"（status 保持 ok、结论列全 null）。只升级 ok 行：download/ + # decode/duration 状态是更早的失败原因，不能被覆盖。 + if config.DEEPSEEK_API_KEY: + df = df.with_column( + "status", + when( + (col("status") == "ok") + & col("analysis")["downgrade_related"].is_null(), + "llm_failed", + ).otherwise(col("status")), + ) return df +def _embedding_input(audio_bytes, status): + """只按音频本身是否可分析决定 embedding 输入。 + + llm_failed 表示音频下载、解码和时长门控均已通过，只是外部 LLM 没有给出 + 合法结论；声学 embedding 与 LLM 独立，必须继续计算，避免可用音频从 ANN + 索引中静默消失。 + """ + media_ready = (status == "ok") | (status == "llm_failed") + return when(media_ready, audio_bytes) + + def run(manifest: str, out_path: str, embed: bool = False) -> None: configure_daft_runner() io_config = daft_io_config() @@ -172,7 +219,12 @@ def run(manifest: str, out_path: str, embed: bool = False) -> None: if embed: embed_udf = _EmbedUDF() - df = df.with_column("audio_embedding", embed_udf(col("audio_bytes"))) + # 下载/解码失败和时长过滤行不做 embedding；llm_failed 的音频本身 + # 已通过媒体门控，声学 embedding 不依赖 LLM，仍应正常计算。 + df = df.with_column( + "audio_embedding", + embed_udf(_embedding_input(col("audio_bytes"), col("status"))), + ) output = df.select(*_BASE_OUTPUT_COLS, "audio_embedding") output.write_lance( out_path, diff --git a/multimodal_toolkit/audio/workflow/ingest.py b/multimodal_toolkit/audio/workflow/ingest.py index 5d7885a..700d7aa 100644 --- a/multimodal_toolkit/audio/workflow/ingest.py +++ b/multimodal_toolkit/audio/workflow/ingest.py @@ -11,12 +11,42 @@ import daft from daft import col, lit -from daft.functions import download +from daft.functions import download, when from ... import config from ...storage.blob import validate_blob_v2 from ...storage.io import configure_daft_runner, daft_io_config, lance_write_mode, read_analysis_output +# JSONL 分析结果靠类型推断读入：某列在整个批次里全为 null 时（比如整批 +# 没配 DEEPSEEK_API_KEY 导致打标列全空，或全批次 download_failed），推断 +# 出来的是 null 类型。直接落 Lance 会建出 null 类型的列，后续正常批次 +# append 时类型冲突、永远写不进去。这里记录每列的规范类型（即正常批次 +# JSON 推断会得到的类型），把全 null 的列显式 cast 回去。 +_ANALYSIS_COLUMN_DTYPES = { + "doc_id": daft.DataType.string(), + "s3_url": daft.DataType.string(), + "status": daft.DataType.string(), + "duration_s": daft.DataType.float64(), + "transcript": daft.DataType.string(), + "acoustic_emotion": daft.DataType.string(), + "downgrade_related": daft.DataType.bool(), + "primary_reason": daft.DataType.string(), + "secondary_reason": daft.DataType.string(), + "summary": daft.DataType.string(), + "confidence": daft.DataType.float64(), + "text_emotion": daft.DataType.string(), + "bad_tone": daft.DataType.bool(), + "emotion_score": daft.DataType.float64(), +} + + +def _cast_all_null_columns(df: daft.DataFrame) -> daft.DataFrame: + for field in df.schema(): + dtype = _ANALYSIS_COLUMN_DTYPES.get(field.name) + if dtype is not None and field.dtype == daft.DataType.null(): + df = df.with_column(field.name, col(field.name).cast(dtype)) + return df + def run(analysis_path: str, lance_uri: str) -> None: configure_daft_runner() @@ -25,11 +55,25 @@ def run(analysis_path: str, lance_uri: str) -> None: now = datetime.now(timezone.utc) df = read_analysis_output(analysis_path, io_config) + df = _cast_all_null_columns(df) + # 重新按 s3_url 下载音频字节作为 blob 列（Stage 1 输出不带原始字节）。 + # 下载失败的行（含 Stage 1 标记为 download_failed 的）blob 为 null， + # 但行本身保留——Lance 表是完整台账，status 列记录了失败原因。 df = df.with_column( "audio_blob", download(col("s3_url"), on_error="null", io_config=io_config) ) - df = df.where(~col("audio_blob").is_null()) + # Stage 1 到 Stage 2 之间对象可能被删或失效：Stage 1 判为 ok 的行这次 + # 下载不到字节时，改标 blob_download_failed，不能让台账里出现 + # "status=ok 但没有 blob"的行。转写和打标列保持 Stage 1 的值—— + # 分析本身是成功的，缺的只是归档字节。 + df = df.with_column( + "status", + when( + (col("status") == "ok") & col("audio_blob").is_null(), + "blob_download_failed", + ).otherwise(col("status")), + ) df = df.with_column( "ingest_time", lit(now).cast(daft.DataType.timestamp("us", "UTC")), diff --git a/tests/audio/test_analyze_udfs.py b/tests/audio/test_analyze_udfs.py new file mode 100644 index 0000000..c52b569 --- /dev/null +++ b/tests/audio/test_analyze_udfs.py @@ -0,0 +1,67 @@ +"""音频 Stage 1 UDF 的空值语义测试（不加载 ASR/嵌入模型，纯本地）。 + +行不丢弃约定的基石：duration_udf 对坏字节返回 null（下游据此判定 +decode_failed），prompt_udf 对 null 转写返回 null（LLM 跳过没处理的行）。 +""" +from __future__ import annotations + +import io + +import daft +import soundfile as sf + + +def _wav_bytes(seconds: float = 1.0, samplerate: int = 16000) -> bytes: + import numpy as np + + buf = io.BytesIO() + samples = np.zeros(int(seconds * samplerate), dtype="float32") + sf.write(buf, samples, samplerate, format="WAV") + return buf.getvalue() + + +def test_duration_udf_null_for_bad_bytes(): + """字节缺失或解码失败必须返回 null 而不是 0.0，否则无法区分坏文件和空录音。""" + from multimodal_toolkit.audio.udfs import duration_udf + + df = daft.from_pydict({"b": [None, b"not-audio-at-all", _wav_bytes(1.0)]}) + out = df.select(duration_udf(df["b"]).alias("d")).to_pydict()["d"] + assert out[0] is None + assert out[1] is None + assert out[2] is not None and abs(out[2] - 1.0) < 0.01 + + +def test_prompt_udf_null_transcript_skips_llm(): + """null 转写（没处理的行）→ null 提示词；空字符串转写（无语音）仍构造提示词。""" + from multimodal_toolkit.audio.udfs import prompt_udf + + df = daft.from_pydict( + {"t": [None, "", "想改成8元套餐"], "e": [None, "NEUTRAL", "ANGRY"]} + ) + out = df.select(prompt_udf(df["t"], df["e"]).alias("p")).to_pydict()["p"] + assert out[0] is None + assert out[1] is not None + assert out[2] is not None and "8元套餐" in out[2] + + +def test_embedding_input_keeps_llm_failed_audio(): + """LLM 失败不影响独立的声学 embedding，只跳过媒体本身不可用的行。""" + from multimodal_toolkit.audio.workflow.analyze import _embedding_input + + df = daft.from_pydict( + { + "audio_bytes": [b"ok", b"llm-failed", b"missing", b"bad", b"filtered"], + "status": [ + "ok", + "llm_failed", + "download_failed", + "decode_failed", + "duration_filtered", + ], + } + ) + out = df.select( + _embedding_input(df["audio_bytes"], df["status"]).alias("embedding_input") + ).to_pydict()["embedding_input"] + + assert out == [b"ok", b"llm-failed", None, None, None] diff --git a/tests/audio/test_ingest.py b/tests/audio/test_ingest.py new file mode 100644 index 0000000..3420f65 --- /dev/null +++ b/tests/audio/test_ingest.py @@ -0,0 +1,106 @@ +"""音频 Stage 2 ingest 的台账语义测试（不下载模型，全本地）。 + +与图片侧 tests/image/test_ingest.py 对应：blob 缺失的行改标 +blob_download_failed 而不是丢弃；全 null 批次（没配 DEEPSEEK_API_KEY 时 +打标列整批为 null）不能把 Lance 表的 schema 毒化成 null 类型。 +""" +from __future__ import annotations + +import json +from pathlib import Path + +import lance + +# 没配 DEEPSEEK_API_KEY 的典型 Stage 1 输出：ASR 字段有值，LLM 打标字段全 null。 +_ANALYSIS_OK = { + "duration_s": 12.5, + "transcript": "想把128元套餐改成8元套餐", + "acoustic_emotion": "NEUTRAL", + "downgrade_related": None, + "primary_reason": None, + "secondary_reason": None, + "summary": None, + "confidence": None, + "text_emotion": None, + "bad_tone": None, + "emotion_score": None, +} + +_ANALYSIS_NULLS = {key: None for key in _ANALYSIS_OK} + + +def _analysis_row(doc_id: str, s3_url: str, status: str = "ok") -> dict: + fields = _ANALYSIS_OK if status == "ok" else _ANALYSIS_NULLS + return {"doc_id": doc_id, "s3_url": s3_url, "status": status, **fields} + + +def _write_analysis_jsonl(path: Path, rows: list[dict]) -> str: + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + return str(path) + + +def _table_by_doc_id(lance_uri: str, columns: list[str]) -> dict[str, dict]: + tbl = lance.dataset(lance_uri).to_table(columns=["doc_id", *columns]).to_pydict() + return { + doc_id: {c: tbl[c][i] for c in columns} for i, doc_id in enumerate(tbl["doc_id"]) + } + + +def test_ingest_marks_ok_rows_with_missing_blob(tmp_path): + """Stage 1 判 ok 但 Stage 2 下载不到字节的行必须改标 blob_download_failed。""" + from multimodal_toolkit.audio.workflow.ingest import run as ingest_run + + present = tmp_path / "present.wav" + present.write_bytes(b"RIFFfake-wav-bytes") + gone = tmp_path / "gone.wav" # 从不落盘：模拟两阶段之间对象被删 + + analysis = _write_analysis_jsonl( + tmp_path / "analysis.jsonl", + [ + _analysis_row("present.wav", str(present)), + _analysis_row("gone.wav", str(gone)), + _analysis_row("missing.wav", str(tmp_path / "missing.wav"), status="download_failed"), + ], + ) + calls_uri = str(tmp_path / "calls.lance") + ingest_run(analysis, calls_uri) + + rows = _table_by_doc_id(calls_uri, ["status", "transcript"]) + assert len(rows) == 3 + assert rows["present.wav"]["status"] == "ok" + assert rows["gone.wav"]["status"] == "blob_download_failed" + # Stage 1 的失败状态原样保留，不被覆盖 + assert rows["missing.wav"]["status"] == "download_failed" + # 转写是 Stage 1 算出来的，blob 缺失不应抹掉它 + assert rows["gone.wav"]["transcript"] == _ANALYSIS_OK["transcript"] + + +def test_ingest_all_null_llm_batch_does_not_poison_schema(tmp_path): + """整批打标列全 null（没配 key）建表后，正常批次仍能 append。""" + from multimodal_toolkit.audio.workflow.ingest import run as ingest_run + + calls_uri = str(tmp_path / "calls.lance") + + batch1 = _write_analysis_jsonl( + tmp_path / "batch1.jsonl", + [ + _analysis_row("a.wav", str(tmp_path / "a.wav"), status="download_failed"), + _analysis_row("b.wav", str(tmp_path / "b.wav"), status="download_failed"), + ], + ) + ingest_run(batch1, calls_uri) + + audio = tmp_path / "c.wav" + audio.write_bytes(b"RIFFfake-wav-bytes") + batch2 = _write_analysis_jsonl( + tmp_path / "batch2.jsonl", [_analysis_row("c.wav", str(audio))] + ) + ingest_run(batch2, calls_uri) + + rows = _table_by_doc_id(calls_uri, ["status", "transcript", "downgrade_related"]) + assert len(rows) == 3 + assert rows["c.wav"]["status"] == "ok" + assert rows["c.wav"]["transcript"] == _ANALYSIS_OK["transcript"] + # 没配 key 时打标结论必须保持 null（"没打标"），不能被填成 False + assert rows["c.wav"]["downgrade_related"] is None + assert rows["a.wav"]["transcript"] is None diff --git a/tests/audio/test_llm_provider.py b/tests/audio/test_llm_provider.py new file mode 100644 index 0000000..821569d --- /dev/null +++ b/tests/audio/test_llm_provider.py @@ -0,0 +1,97 @@ +"""audio/llm.py 适配层测试（不发真实请求），与 tests/image/test_llm_analysis.py +的适配器测试对应：UDF 选项不得泄漏进 OpenAI 请求参数，null prompt 不得 +触发 API 调用。 +""" +from __future__ import annotations + +import asyncio +import json + +import daft + + +def _make_descriptor(): + from multimodal_toolkit.audio.llm import AudioLLMProvider + + provider = AudioLLMProvider( + name="audio-llm-test", base_url="https://example.invalid/v1", api_key="test-key" + ) + return provider.get_prompter( + "test-chat-model", + use_chat_completions=True, + concurrency=8, + on_error="ignore", + response_format={"type": "json_object"}, + temperature=0, + ) + + +def test_udf_options_kept_out_of_openai_request(): + descriptor = _make_descriptor() + # UDF 层的选项 Daft 仍要读到 + assert descriptor.get_udf_options().on_error == "ignore" + assert descriptor.get_udf_options().concurrency == 8 + # 但不能进入发给 OpenAI SDK 的请求参数 + prompter = descriptor.instantiate() + assert "on_error" not in prompter.generation_config + assert "concurrency" not in prompter.generation_config + assert prompter.generation_config["response_format"] == {"type": "json_object"} + assert prompter.generation_config["temperature"] == 0 + + +def test_null_prompt_skipped_without_api_call(): + prompter = _make_descriptor().instantiate() + assert asyncio.run(prompter.prompt((None,))) is None + + +def _valid_analysis_payload() -> dict: + return { + "downgrade_related": True, + "primary_reason": "价格敏感", + "secondary_reason": "套餐费用过高", + "summary": "客户希望降低套餐费用", + "confidence": 0.92, + "text_emotion": "不满", + "bad_tone": True, + "emotion_score": 0.75, + } + + +def test_llm_validator_accepts_complete_typed_response(): + """完整且满足业务约束的响应必须保留类型和值。""" + from multimodal_toolkit.audio.llm import validate_llm_response + + expected = _valid_analysis_payload() + assert validate_llm_response(json.dumps(expected)) == expected + + +def test_llm_validator_rejects_bad_envelopes_without_batch_failure(): + """合法 JSON 中的错类型/缺字段也只能让该行失败，不能终止整个 morsel。""" + from multimodal_toolkit.audio.llm import validate_llm_responses + + wrong_type = _valid_analysis_payload() + wrong_type["downgrade_related"] = "yes" + missing_field = _valid_analysis_payload() + missing_field.pop("summary") + invalid_enum = _valid_analysis_payload() + invalid_enum["primary_reason"] = "模型自造原因" + invalid_score = _valid_analysis_payload() + invalid_score["confidence"] = 1.5 + + raw = [ + "not-json", + "[]", + "{}", + json.dumps(wrong_type), + json.dumps(missing_field), + json.dumps(invalid_enum), + json.dumps(invalid_score), + ] + df = daft.from_pydict({"raw": raw}) + rows = df.select(validate_llm_responses(df["raw"]).alias("analysis")).to_pydict() + + assert len(rows["analysis"]) == len(raw) + assert all( + analysis is not None and all(value is None for value in analysis.values()) + for analysis in rows["analysis"] + )