Qwen3.8-Omni-Flash 接入实战

1# 确认 Python 版本
2python --version        # >= 3.10
3
4# 安装官方 SDK
5pip install -U dashscope
6
7# 或者如果你打算走 OpenAI 兼容协议（推荐，切换模型成本最低）
8pip install -U openai

1# Linux / macOS
2export DASHSCOPE_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
3
4# Windows PowerShell
5$env:DASHSCOPE_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxx"
6
7# Windows CMD
8set DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

1echo 'export DASHSCOPE_API_KEY="sk-xxxx"' >> ~/.bashrc
2source ~/.bashrc

1# ❌ 错误示范
2client = OpenAI(api_key="sk-1234567890abcdef")
3
4# ✅ 正确做法
5import os
6from openai import OpenAI
7
8api_key = os.getenv("DASHSCOPE_API_KEY")
9if not api_key:
10    raise RuntimeError("未设置 DASHSCOPE_API_KEY 环境变量")
11
12client = OpenAI(
13    api_key=api_key,
14    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
15)

1import os
2from openai import OpenAI
3
4client = OpenAI(
5    api_key=os.getenv("DASHSCOPE_API_KEY"),
6    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
7)
8
9resp = client.chat.completions.create(
10    model="qwen3.8-omni-flash",
11    messages=[
12        {
13            "role": "user",
14            "content": [
15                {"type": "text", "text": "这段音频在讲什么？用 3 句话总结。"},
16                {"type": "input_audio", "input_audio": {"data": "", "format": "mp3"}},
17            ],
18        }
19    ],
20)
21print(resp.choices[0].message.content)

1import os
2import dashscope
3from dashscope import MultiModalConversation
4
5dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
6
7messages = [
8    {
9        "role": "user",
10        "content": [
11            {"audio": "https://example.com/podcast.mp3"},
12            {"text": "提取这段播客的行动项，输出 JSON。"},
13        ],
14    }
15]
16
17resp = MultiModalConversation.call(
18    model="qwen3.8-omni-flash",
19    messages=messages,
20    stream=False,
21)
22print(resp.output.choices[0].message.content)

1messages = [
2    {
3        "role": "user",
4        "content": [
5            {"type": "input_audio", "input_audio": {"data": "https://cdn.example.com/meeting.mp3"}},
6            {"type": "text", "text": "总结会议结论。"},
7        ],
8    }
9]

1import base64
2from pathlib import Path
3
4def to_base64(path: str) -> str:
5    return base64.b64encode(Path(path).read_bytes()).decode("utf-8")
6
7audio_b64 = to_base64("./meeting.mp3")
8
9messages = [
10    {
11        "role": "user",
12        "content": [
13            {
14                "type": "input_audio",
15                "input_audio": {"data": f"data:;base64,{audio_b64}", "format": "mp3"},
16            },
17            {"type": "text", "text": "总结会议结论。"},
18        ],
19    }
20]

1from dashscope import Files
2
3# 1. 上传一次
4f = Files.create(file=Path("./2h-conference.mp4"), purpose="file-extract")
5file_id = f.id
6print("file_id:", file_id)
7
8# 2. 后续多次引用，不再重复传文件
9messages = [
10    {
11        "role": "user",
12        "content": [
13            {"type": "file", "file": {"file_id": file_id}},
14            {"type": "text", "text": "第 37 分钟提到的预算数字是多少？"},
15        ],
16    }
17]

1# ❌ 等长切片：会把一句话切成两半，两侧都丢失上下文
2chunks = [audio[i:i+600] for i in range(0, len(audio), 600)]
3
4# ✅ 语义切片：按静音段 / 话题边界切

1import subprocess
2import json
3from pathlib import Path
4
5def detect_silence(audio_path: str, min_silence_len: float = 0.8, silence_thresh: int = -35):
6    """用 ffmpeg 检测静音段，返回 [(start, end), ...]"""
7    cmd = [
8        "ffmpeg", "-i", audio_path,
9        "-af", f"silencedetect=noise={silence_thresh}dB:d={min_silence_len}",
10        "-f", "null", "-",
11    ]
12    proc = subprocess.run(cmd, capture_output=True, text=True)
13    lines = proc.stderr.splitlines()
14
15    starts, ends = [], []
16    for ln in lines:
17        if "silence_start:" in ln:
18            starts.append(float(ln.split("silence_start:")[1].strip()))
19        elif "silence_end:" in ln:
20            ends.append(float(ln.split("silence_end:")[1].strip().split("|")[0].strip()))
21    return list(zip(starts, ends))
22
23def semantic_segments(audio_path: str, total_duration: float, target_len: float = 300.0):
24    """在静音段附近切分，每片目标 300 秒"""
25    silences = detect_silence(audio_path)
26    bounds = [0.0]
27    for s, e in silences:
28        mid = (s + e) / 2
29        if mid - bounds[-1] >= target_len * 0.6:   # 至少到目标长度的 60% 才切
30            bounds.append(mid)
31    bounds.append(total_duration)
32    return [(bounds[i], bounds[i + 1]) for i in range(len(bounds) - 1)]

1import os
2import json
3from concurrent.futures import ThreadPoolExecutor, as_completed
4from openai import OpenAI
5from pathlib import Path
6
7client = OpenAI(
8    api_key=os.getenv("DASHSCOPE_API_KEY"),
9    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
10)
11
12MODEL = "qwen3.8-omni-flash"
13
14EXTRACT_PROMPT = """你是会议记录提取器。请从这段音频片段中提取：
151. 讨论的话题（1 句话）
162. 做出的决策（列表，没有就写「无」）
173. 分配的行动项（含负责人和截止时间，没有就写「无」）
184. 提到的关键数字（预算 / 时间 / 指标）
19
20输出严格 JSON，不要 Markdown 代码块包裹：
21{"topics": "...", "decisions": [], "actions": [{"owner":"","task":"","due":""}], "numbers": []}
22"""
23
24REDUCE_PROMPT = """以下是同一场会议 {n} 个片段的提取结果（按时间顺序）。
25请合并为一份完整会议纪要，要求：
261. 决策去重合并，冲突时以时间靠后的为准
272. 行动项按负责人分组
283. 关键数字汇总成表
294. 输出 Markdown 格式
30
31片段结果：
32{chunks}
33"""
34
35def process_chunk(args):
36    idx, (start, end), audio_path = args
37    # 用 ffmpeg 切出片段
38    seg_path = f"./tmp/seg_{idx:03d}.mp3"
39    subprocess.run(
40        ["ffmpeg", "-y", "-i", audio_path, "-ss", str(start), "-to", str(end),
41         "-ar", "16000", "-ac", "1", seg_path],
42        capture_output=True,
43    )
44    b64 = base64.b64encode(Path(seg_path).read_bytes()).decode()
45
46    resp = client.chat.completions.create(
47        model=MODEL,
48        messages=[{
49            "role": "user",
50            "content": [
51                {"type": "input_audio", "input_audio": {"data": f"data:;base64,{b64}", "format": "mp3"}},
52                {"type": "text", "text": EXTRACT_PROMPT},
53            ],
54        }],
55        temperature=0.1,
56    )
57    return idx, resp.choices[0].message.content
58
59def summarize_long_audio(audio_path: str, duration: float, max_workers: int = 6):
60    segs = semantic_segments(audio_path, duration)
61    print(f"切成 {len(segs)} 段：{[f'{s:.0f}-{e:.0f}s' for s, e in segs]}")
62
63    results = [None] * len(segs)
64    with ThreadPoolExecutor(max_workers=max_workers) as ex:
65        futs = [ex.submit(process_chunk, (i, seg, audio_path)) for i, seg in enumerate(segs)]
66        for fut in as_completed(futs):
67            idx, out = fut.result()
68            results[idx] = out
69            print(f"  ✓ 片段 {idx + 1}/{len(segs)} 完成")
70
71    joined = "\n\n".join(f"[片段 {i + 1}]\n{r}" for i, r in enumerate(results))
72    final = client.chat.completions.create(
73        model=MODEL,
74        messages=[{
75            "role": "user",
76            "content": [{"type": "text", "text": REDUCE_PROMPT.format(n=len(segs), chunks=joined)}],
77        }],
78        temperature=0.2,
79    )
80    return final.choices[0].message.content

1import time
2import random
3
4def call_with_retry(fn, max_retries: int = 5):
5    delay = 1.0
6    for attempt in range(max_retries):
7        try:
8            return fn()
9        except Exception as e:
10            msg = str(e)
11            if "429" in msg or "throttl" in msg.lower() or "529" in msg:
12                if attempt == max_retries - 1:
13                    raise
14                sleep_s = delay + random.uniform(0, 0.5)
15                print(f"  限流，{sleep_s:.1f}s 后重试（第 {attempt + 1} 次）")
16                time.sleep(sleep_s)
17                delay = min(delay * 2, 32)
18            else:
19                raise

1ACTION_EXTRACT_PROMPT = """你是会议纪要助手。从音频中提取：
2
3## 输出格式（严格 JSON）
4{
5  "meeting_title": "",
6  "participants": [],
7  "decisions": [
8    {"content": "", "owner": "", "timestamp": "MM:SS"}
9  ],
10  "action_items": [
11    {"task": "", "owner": "", "due": "", "priority": "P0|P1|P2", "source_quote": ""}
12  ],
13  "open_questions": [],
14  "next_meeting": ""
15}
16
17## 规则
181. `owner` 必须是音频里出现的具体人名，不确定就写 "TBD"
192. `due` 只写音频里明确提到的日期，格式 YYYY-MM-DD，没提就写 null
203. `priority` 判断依据：提到「阻塞」「上线前」「必须」→ P0；「本周」「尽快」→ P1；其余 P2
214. `source_quote` 是触发这条行动项的原始发言（原文，不要改写），没有明确发言就写 null
225. 提取不到就写空数组，不要编造
23"""
24
25def meeting_to_tickets(audio_path: str) -> dict:
26    resp = client.chat.completions.create(
27        model=MODEL,
28        messages=[{
29            "role": "user",
30            "content": [
31                {"type": "input_audio", "input_audio": {"data": f"data:;base64,{to_base64(audio_path)}", "format": "mp3"}},
32                {"type": "text", "text": ACTION_EXTRACT_PROMPT},
33            ],
34        }],
35        temperature=0.0,
36        response_format={"type": "json_object"},
37    )
38    return json.loads(resp.choices[0].message.content)

1def push_to_jira(items: list, dry_run: bool = True):
2    """dry_run=True 时只打印不创建，确认无误后再置 False"""
3    for it in items:
4        payload = {
5            "fields": {
6                "project": {"key": "PROD"},
7                "summary": it["task"],
8                "priority": {"name": {"P0": "Highest", "P1": "High", "P2": "Medium"}[it["priority"]]},
9                "assignee": resolve_user(it["owner"]),   # 人名 → Jira 账号
10                "description": f"来源：会议音频\n\n原始发言：{it.get('source_quote') or '（无明确发言）'}",
11            }
12        }
13        if dry_run:
14            print(f"[DRY] 将创建：{it['priority']} | {it['owner']} | {it['task'][:40]}")
15        else:
16            requests.post(JIRA_API, json=payload, auth=(JIRA_USER, JIRA_TOKEN))

1CHAPTER_PROMPT = """你是视频内容分析师。为这段视频片段生成解说稿：
2
3## 输出格式（严格 JSON）
4{
5  "start_time": "MM:SS",
6  "end_time": "MM:SS",
7  "chapter_title": "",
8  "summary": "（80-120 字）",
9  "key_points": ["", "", ""],
10  "code_or_formula_shown": ["（画面上出现的代码/公式原文，没有就空数组）"],
11  "visual_description": "（画面内容描述，用于后续配图定位）"
12}
13
14## 规则
151. `chapter_title` 要具体到内容，不要写「第一部分」这种
162. `code_or_formula_shown` 必须逐字抄录画面上的代码，不要改写、不要省略
173. 如果画面主要是 PPT，描述 PPT 的结构而不是念字
18"""
19
20def video_to_chapters(video_path: str, duration: float) -> list:
21    segs = semantic_segments(video_path, duration, target_len=420.0)
22    chapters = []
23    for i, (s, e) in enumerate(segs):
24        seg = cut_video(video_path, s, e)
25        resp = client.chat.completions.create(
26            model=MODEL,
27            messages=[{
28                "role": "user",
29                "content": [
30                    {"type": "video", "video": f"data:;base64,{to_base64(seg)}"},
31                    {"type": "text", "text": CHAPTER_PROMPT},
32                ],
33            }],
34            temperature=0.3,
35            response_format={"type": "json_object"},
36        )
37        chapters.append(json.loads(resp.choices[0].message.content))
38    return chapters

1SUBTITLE_PROMPT = """你是字幕翻译专家。把这段视频的语音翻译成{target_lang}字幕。
2
3## 输出格式（严格 JSON 数组）
4[
5  {{"start": "MM:SS.mmm", "end": "MM:SS.mmm", "original": "", "translated": ""}}
6]
7
8## 规则
91. 每条字幕不超过 2 行、每行不超过 18 个字符（中文）/ 42 个字符（英文）
102. 单条时长 0.8s - 6s，太短合并、太长切分
113. `original` 必须与听到的语音逐字一致，不要润色
124. 翻译要口语化，符合{target_lang}字幕习惯，不要直译
135. 专有名词（人名/品牌/术语）保留原文不翻译
146. 不要输出任何 JSON 之外的内容
15"""

1def align_offsets(chunks: list, seg_bounds: list) -> list:
2    """把片段内相对时间转成全片绝对时间"""
3    out = []
4    for (seg_start, _), items in zip(seg_bounds, chunks):
5        for it in items:
6            out.append({
7                "start": shift(it["start"], seg_start),
8                "end": shift(it["end"], seg_start),
9                "original": it["original"],
10                "translated": it["translated"],
11            })
12    return out
13
14def shift(t: str, offset_s: float) -> str:
15    mm, ss = t.split(":")
16    total = int(mm) * 60 + float(ss) + offset_s
17    return f"{int(total // 60):02d}:{total % 60:06.3f}"

1MV_PROMPT = """你是 MV 导演。基于这段音乐生成分镜脚本。
2
3## 分析维度
41. 音乐结构：前奏 / 主歌 / 预副歌 / 副歌 / 间奏 / 桥段 / 尾奏，给出各自时间段
52. 情绪曲线：每段的情绪强度（1-10）和关键词
63. BPM 与节拍点
7
8## 分镜输出（严格 JSON）
9{
10  "structure": [{"section": "", "start": "MM:SS", "end": "MM:SS", "intensity": 0, "mood": ""}],
11  "shots": [
12    {
13      "start": "MM:SS", "end": "MM:SS",
14      "shot_size": "远景|全景|中景|近景|特写",
15      "camera_move": "固定|推|拉|摇|移|跟|升降",
16      "scene": "",
17      "action": "",
18      "transition": "硬切|叠化|闪白|黑场"
19    }
20  ]
21}
22
23## 规则
241. 副歌段落镜头切换频率提高（2-3s/镜），主歌段落放缓（4-6s/镜）
252. 卡点：镜头切换点尽量落在节拍上
263. 情绪强度 ≥8 的段落必须有运动镜头
27"""

1import librosa
2
3def get_beats(audio_path: str):
4    y, sr = librosa.load(audio_path)
5    tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
6    beat_times = librosa.frames_to_time(beat_frames, sr=sr)
7    return float(tempo), beat_times.tolist()
8
9def snap_to_beat(shot_start: float, beat_times: list, tolerance: float = 0.4) -> float:
10    """把分镜起点吸附到最近的节拍"""
11    nearest = min(beat_times, key=lambda b: abs(b - shot_start))
12    return nearest if abs(nearest - shot_start) <= tolerance else shot_start

1resp = client.chat.completions.create(
2    model=MODEL,
3    messages=messages,
4    stream=True,
5)
6for chunk in resp:
7    if chunk.choices[0].delta.content:
8        print(chunk.choices[0].delta.content, end="", flush=True)

1responses = MultiModalConversation.call(
2    model="qwen3.8-omni-flash",
3    messages=messages,
4    stream=True,
5    incremental_output=True,   # 关键：只返回增量，否则每次返回全量
6)
7for chunk in responses:
8    print(chunk.output.choices[0].message.content, end="", flush=True)

1# OpenAI 兼容协议
2resp = client.chat.completions.create(
3    model=MODEL,
4    messages=messages,
5    response_format={"type": "json_object"},
6)

1client = OpenAI(
2    api_key=api_key,
3    base_url=BASE_URL,
4    timeout=600.0,
5    max_retries=2,
6)

1import hashlib
2from functools import lru_cache
3
4def file_hash(path: str) -> str:
5    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
6
7# L1：进程内缓存（同一素材 + 同一 prompt）
8@lru_cache(maxsize=128)
9def cached_call(file_hash_str: str, prompt: str) -> str:
10    ...
11
12# L2：文件级缓存（upload 后复用 file_id）
13FILE_ID_CACHE = {}   # {file_hash: file_id}
14
15def get_or_upload(path: str) -> str:
16    h = file_hash(path)
17    if h not in FILE_ID_CACHE:
18        FILE_ID_CACHE[h] = Files.create(file=Path(path), purpose="file-extract").id
19    return FILE_ID_CACHE[h]

1你的素材是什么语言？
2├─ 中文为主 ────────────────► Qwen3.8-Omni-Flash
3├─ 英文为主 ────────────────► Gemini 3.8 Flash 或 GPT-5o
4└─ 多语种混合
5   ├─ 成本敏感 ─────────────► Qwen3.8-Omni-Flash（英文略弱但便宜 20 倍）
6   └─ 质量优先 ─────────────► 两者都跑，取并集
7
8你的任务类型？
9├─ 纯理解（问答 / 摘要）────► 任一都可
10├─ 要执行动作（建单 / 通知）► Qwen3.8-Omni-Flash（Agentic 能力最强）
11└─ 要生成视频 ──────────────► 都不行，换生成模型
12
13你的合规要求？
14├─ 数据可上公网 ────────────► 直接 API
15└─ 数据必须不出内网 ────────► Qwen2.5-Omni 开源自建（精度降一档）

1ffmpeg -i meeting.mp4 -vn -ar 16000 -ac 1 meeting.mp3

1def process_all(segs):
2    for i, seg in enumerate(segs):
3        out_path = Path(f"./cache/seg_{i:03d}.json")
4        if out_path.exists():          # 已完成，跳过
5            continue
6        result = process_chunk((i, seg, AUDIO))
7        out_path.write_text(json.dumps(result, ensure_ascii=False))