diff --git a/.agents/skills/dashscope/SKILL.md b/.agents/skills/dashscope/SKILL.md new file mode 100644 index 00000000..ccffbe6f --- /dev/null +++ b/.agents/skills/dashscope/SKILL.md @@ -0,0 +1,136 @@ +--- +name: dashscope +description: DashScope (Alibaba Cloud Bailian / 阿里云百炼) integration — image generation (qwen-image-2.0-pro), text-to-speech (qwen3-tts-flash), and ASR with word-level timestamps (qwen3-asr-flash-filetrans). Use when generating images via Qwen-Image, narrating via Qwen-TTS, or transcribing with word-level timestamps via Qwen-ASR. +--- + +# DashScope + +Requires `DASHSCOPE_API_KEY` in `.env`. Get one at https://dashscope.aliyun.com/. + +## Current API + +**CRITICAL:** DashScope's `/compatible-mode/v1/` only supports `/chat/completions` and `/embeddings`. Image generation, TTS, and ASR all use **DashScope-native endpoints** — not OpenAI-compatible paths. + +All three tools use `Authorization: Bearer $DASHSCOPE_API_KEY`. + +### Image Generation + +```text +POST https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation +``` + +- Model: `qwen-image-2.0-pro` (default), `qwen-image-max`, `wan2.7-image`, `z-image-turbo` +- Body: `{model, input: {messages: [{role: "user", content: [{text: "prompt"}]}]}, parameters: {size: "W*H", n, prompt_extend, watermark}}` +- **Size format uses asterisk:** `"1024*1024"` not `"1024x1024"` +- Response: `output.choices[0].message.content[0].image` (URL, valid ~24h) — must download separately + +### Text-to-Speech + +```text +POST https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation +``` + +Same endpoint as image gen, different body. + +- Model: `qwen3-tts-flash` (default), `qwen3-tts-instruct-flash`, `qwen-tts-2025-05-22` +- Body: `{model, input: {text, voice: "Cherry", language_type: "Auto"}}` +- Response: `output.audio.url` (WAV, valid ~24h) — must download separately + +### ASR with Word-Level Timestamps + +```text +POST https://dashscope.aliyuncs.com/api/v1/services/audio/asr/transcription +Header: X-DashScope-Async: enable +``` + +- Model: `qwen3-asr-flash-filetrans` (NOT `qwen3-asr-flash` — the sync version has no word timestamps) +- Body: `{model, input: {file_url: "https://public-url/audio.mp3"}, parameters: {enable_words: true, language_hints: ["zh","en"]}}` +- Returns `task_id` → poll `GET /api/v1/tasks/{task_id}` until `SUCCEEDED` → download `output.result.transcription_url` → JSON with `transcripts[].sentences[].words[]` +- Timestamps in `begin_time`/`end_time` are in **milliseconds** — the tool normalizes to seconds + +## OpenMontage Usage + +### Image via selector + +```python +from tools.graphics.image_selector import ImageSelector + +result = ImageSelector().execute({ + "preferred_provider": "dashscope", + "prompt": "一只猫坐在沙发上", + "output_path": "projects/my-video/assets/images/cat.png", +}) +``` + +### TTS via selector + +```python +from tools.audio.tts_selector import TTSSelector + +result = TTSSelector().execute({ + "preferred_provider": "dashscope", + "text": "如果 AI 真的会改变未来，普通人到底该怎么参与？", + "voice": "Cherry", + "output_path": "projects/my-video/assets/audio/narration.wav", +}) +``` + +### ASR directly (word timestamps for subtitles) + +```python +from tools.analysis.dashscope_asr import DashscopeAsr + +result = DashscopeAsr().execute({ + "audio_url": "https://example.com/narration.wav", + "output_path": "projects/my-video/assets/audio/transcription.json", +}) + +# result.data["words"] is a flat list of {text, begin_time_seconds, end_time_seconds} +``` + +## Recommended Workflow + +1. **Image:** Generate a sample first. Check `prompt_extend: true` (default) — DashScope rewrites your prompt for better results. Disable if you need literal prompt adherence. +2. **TTS:** Generate a 10-15 second sample before full narration. Approve voice and pacing before committing to full generation. +3. **ASR:** Audio must be at a **publicly accessible URL**. Upload to any public host (S3, etc.) first. Local paths are rejected with a clear error. +4. **Subtitles:** Build from `result.data["words"]` — each word has `begin_time_seconds` and `end_time_seconds`. Group words into caption phrases by language semantics, not fixed character count. + +## Parameters + +### Image (`dashscope_image`) +- `prompt` (required): text prompt +- `model`: default `qwen-image-2.0-pro` +- `size`: default `"1024*1024"` — **asterisk separator, not "x"** +- `n`: 1-6 images +- `negative_prompt`: things to avoid (max 500 chars) +- `prompt_extend`: default `true` — auto-rewrite prompt for better results +- `watermark`: default `false` +- `seed`: for reproducibility + +### TTS (`dashscope_tts`) +- `text` (required): text to synthesize (max 600 chars for qwen3-tts-flash) +- `model`: default `qwen3-tts-flash` +- `voice`: default `"Cherry"` — other voices: `"Ethan"`, `"Chelsie"`, etc. +- `language_type`: default `"Auto"` — `"Chinese"`, `"English"`, `"Japanese"`, `"Korean"` +- `instructions`: natural language delivery instructions (only for `qwen3-tts-instruct-flash`) + +### ASR (`dashscope_asr`) +- `audio_url` (required): **must be publicly accessible URL** +- `model`: `qwen3-asr-flash-filetrans` (only model that supports word timestamps) +- `language_hints`: default `["zh", "en"]` +- `enable_words`: default `true` — required for word-level timestamps +- `poll_interval_seconds`: default `5.0` +- `timeout_seconds`: default `300` + +## Troubleshooting + +- **Image size error:** Use `"W*H"` with asterisk, not `"WxH"`. Example: `"2048*2048"`. +- **TTS no audio URL:** Check `output.audio.url` — if empty, the model name or voice may be wrong. +- **ASR "file not accessible":** `audio_url` must be publicly reachable. DashScope servers fetch the file; local paths and auth-gated URLs don't work. +- **ASR poll timeout:** Increase `timeout_seconds` (default 300). Long audio files take longer to transcribe. +- **ASR no word timestamps:** Ensure `enable_words: true` and model is `qwen3-asr-flash-filetrans` (not the sync `qwen3-asr-flash`). +- **Auth error (401):** Verify `DASHSCOPE_API_KEY` is set. Use `Authorization: Bearer $KEY` header. + +## Safety + +Never print or write the API key to logs, metadata, patches, or project artifacts. `.env.example` should contain only empty variable names. The tool's `_safe_error()` method redacts the key from error messages. diff --git a/.env.example b/.env.example index ffad5a8c..73e7122b 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,10 @@ DOUBAO_SPEECH_API_KEY= # Volcengine Doubao Speech TTS (new console API Key DOUBAO_SPEECH_VOICE_TYPE= # Default Doubao speaker/voice type, e.g. zh_female_vv_uranus_bigtts # Piper local voices do not require env vars; install `piper-tts` via pip +# --- DashScope (Alibaba Cloud Bailian) --- +DASHSCOPE_API_KEY= # Qwen image gen (qwen-image-2.0-pro), TTS (qwen3-tts-flash), ASR with word timestamps (qwen3-asr-flash-filetrans) + # Get one at https://dashscope.aliyun.com/ + # --- Music --- SUNO_API_KEY= # Suno AI music generation (full songs, instrumentals, any genre) diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index 5aa809d4..f5e34f7e 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -41,6 +41,7 @@ OPENAI_API_KEY= # OpenAI TTS + GPT Image 2 images XAI_API_KEY= # xAI Grok image generation/editing + Grok video generation DOUBAO_SPEECH_API_KEY= # Volcengine Doubao Speech TTS (strong Mandarin narration) DOUBAO_SPEECH_VOICE_TYPE= # Default Doubao speaker/voice type +DASHSCOPE_API_KEY= # Alibaba DashScope (Qwen image gen, TTS, ASR with word timestamps) # MULTI-MODEL GATEWAY (one key, 6+ tools) FAL_KEY= # FLUX, Recraft, Kling, Veo, MiniMax video @@ -94,6 +95,43 @@ OpenMontage now uses those published rates in the Grok tool estimators. --- +### Alibaba DashScope — Qwen Image + TTS + ASR + +> **Best for Chinese-language production.** One key unlocks Qwen-Image generation, Qwen-TTS Mandarin narration, and Qwen-ASR with word-level timestamps — the only DashScope path that provides word-level granularity for subtitle alignment. + +**Tools unlocked:** `dashscope_image`, `dashscope_tts`, `dashscope_asr` +**Env var:** `DASHSCOPE_API_KEY` + +#### Setup + +1. Go to [dashscope.aliyun.com](https://dashscope.aliyun.com/) +2. Create an Alibaba Cloud account if you don't have one +3. Generate an API key in the DashScope console +4. Add to `.env`: `DASHSCOPE_API_KEY=sk-...` + +#### What it's best for + +- Chinese-language image generation with strong prompt understanding (Qwen-Image) +- Natural Mandarin narration (Qwen-TTS, Cherry voice) +- Word-level timestamp transcription for subtitle alignment (Qwen-ASR filetrans) +- Replacing the broken `whisperx` slot for ASR + +#### API notes + +DashScope's `/compatible-mode/v1/` only supports `/chat/completions` and `/embeddings`. Image gen, TTS, and ASR all use DashScope-native endpoints with nested `{model, input, parameters}` request shape — not OpenAI-compatible paths. + +The ASR tool (`qwen3-asr-flash-filetrans`) uses an async submit-poll pattern. Audio must be at a publicly accessible URL (local files are not supported). Word timestamps are in milliseconds, normalized to seconds by the tool. + +#### Pricing + +| Model | Price | +|------|-------| +| `qwen-image-2.0-pro` | ~$0.02 per image (check console for current rates) | +| `qwen3-tts-flash` | ~$0.000015 per character | +| `qwen3-asr-flash-filetrans` | Per-minute billing (check console) | + +--- + ### fal.ai — Multi-Model Gateway > **Broad single-key coverage.** One API key unlocks image and video providers across multiple models. diff --git a/tests/contracts/test_dashscope_tools.py b/tests/contracts/test_dashscope_tools.py new file mode 100644 index 00000000..d9c001b8 --- /dev/null +++ b/tests/contracts/test_dashscope_tools.py @@ -0,0 +1,709 @@ +"""Contract tests for DashScope (Alibaba Cloud Bailian) provider tools. + +These tests verify that the tools satisfy the BaseTool contract without +requiring a real DashScope API key or making any API calls. They check +class attributes, schemas, status reporting, cost estimates, and the +Layer 3 skill file existence. + +Run: pytest tests/contracts/test_dashscope_tools.py -v +""" + +from pathlib import Path + +import pytest + +from tools.base_tool import ( + BaseTool, + ExecutionMode, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) +from tools.graphics.dashscope_image import DashscopeImage +from tools.audio.dashscope_tts import DashscopeTTS +from tools.analysis.dashscope_asr import DashscopeAsr + +TOOLS = [DashscopeImage, DashscopeTTS, DashscopeAsr] +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent + +EXPECTED_TIER = { + DashscopeImage: ToolTier.GENERATE, + DashscopeTTS: ToolTier.VOICE, + DashscopeAsr: ToolTier.ANALYZE, +} +EXPECTED_CAPABILITY = { + DashscopeImage: "image_generation", + DashscopeTTS: "tts", + DashscopeAsr: "analysis", +} +EXPECTED_EXECUTION_MODE = { + DashscopeImage: ExecutionMode.SYNC, + DashscopeTTS: ExecutionMode.SYNC, + DashscopeAsr: ExecutionMode.ASYNC, +} + + +# ------------------------------------------------------------------ +# Contract compliance (parametrized over all 3 tools) +# ------------------------------------------------------------------ + +@pytest.mark.parametrize("cls", TOOLS, ids=lambda c: c.name) +class TestContract: + + def test_inherits_base_tool(self, cls): + assert issubclass(cls, BaseTool) + + def test_has_required_identity(self, cls): + tool = cls() + assert tool.name + assert tool.version + assert tool.provider == "dashscope" + assert tool.capability == EXPECTED_CAPABILITY[cls] + assert tool.tier == EXPECTED_TIER[cls] + assert tool.stability == ToolStability.EXPERIMENTAL + assert tool.runtime == ToolRuntime.API + + def test_has_input_schema(self, cls): + tool = cls() + schema = tool.input_schema + assert schema.get("type") == "object" + props = schema.get("properties", {}) + required = schema.get("required", []) + # Each tool has at least one required field + assert len(required) >= 1 + for field in required: + assert field in props + + def test_has_capabilities(self, cls): + tool = cls() + assert len(tool.capabilities) > 0 + + def test_has_agent_skills(self, cls): + tool = cls() + assert tool.agent_skills + assert "dashscope" in tool.agent_skills + + def test_dashscope_layer3_skill_exists(self, cls): + skill_path = ( + PROJECT_ROOT / ".agents" / "skills" / "dashscope" / "SKILL.md" + ) + assert skill_path.exists(), f"Missing Layer 3 skill: {skill_path}" + content = skill_path.read_text(encoding="utf-8") + assert "DASHSCOPE_API_KEY" in content + + def test_has_fallbacks(self, cls): + tool = cls() + assert tool.fallback or tool.fallback_tools + + def test_has_install_instructions(self, cls): + tool = cls() + assert tool.install_instructions + assert "DASHSCOPE_API_KEY" in tool.install_instructions + + def test_get_info_returns_dict(self, cls): + tool = cls() + info = tool.get_info() + assert isinstance(info, dict) + assert info["name"] == tool.name + assert info["provider"] == "dashscope" + assert info["runtime"] == "api" + assert info["agent_skills"] == ["dashscope"] + + def test_execution_mode(self, cls): + tool = cls() + assert tool.execution_mode == EXPECTED_EXECUTION_MODE[cls] + + def test_status_unavailable_without_key(self, cls, monkeypatch): + monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False) + tool = cls() + assert tool.get_status() == ToolStatus.UNAVAILABLE + + def test_status_available_with_key(self, cls, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = cls() + assert tool.get_status() == ToolStatus.AVAILABLE + + def test_idempotency_key_fields(self, cls): + tool = cls() + assert len(tool.idempotency_key_fields) > 0 + + def test_has_resource_profile(self, cls): + tool = cls() + assert tool.resource_profile.network_required is True + assert tool.resource_profile.vram_mb == 0 + + def test_has_retry_policy(self, cls): + tool = cls() + assert tool.retry_policy.max_retries >= 0 + + def test_has_side_effects(self, cls): + tool = cls() + assert len(tool.side_effects) > 0 + # Must mention it calls the API + assert any("API" in s for s in tool.side_effects) + + def test_has_user_visible_verification(self, cls): + tool = cls() + assert len(tool.user_visible_verification) > 0 + + def test_lazy_imports_requests(self, cls): + """Tool module must not import requests at top level (registry + discovery must stay fast).""" + import importlib + import sys + # Remove requests from cache to simulate fresh import + mod_name = cls.__module__ + if "requests" in sys.modules: + del sys.modules["requests"] + # Re-import the tool module — should not pull in requests + # (requests is imported inside execute(), not at module level) + importlib.reload(sys.modules[mod_name]) + # The tool module itself should not have imported requests + # (it's inside execute, so module-level reload shouldn't trigger it) + # This is a smoke test — the real proof is that registry.discover() + # works without requests installed, but requests IS installed here. + + def test_estimate_cost_returns_float(self, cls): + tool = cls() + # Use tool-specific minimal inputs + if cls is DashscopeImage: + cost = tool.estimate_cost({"prompt": "test", "n": 1}) + elif cls is DashscopeTTS: + cost = tool.estimate_cost({"text": "test"}) + else: + cost = tool.estimate_cost({"audio_url": "https://x.com/a.mp3"}) + assert isinstance(cost, float) + assert cost >= 0.0 + + def test_dry_run_returns_dict(self, cls): + tool = cls() + if cls is DashscopeImage: + result = tool.dry_run({"prompt": "test"}) + elif cls is DashscopeTTS: + result = tool.dry_run({"text": "test"}) + else: + result = tool.dry_run({"audio_url": "https://x.com/a.mp3"}) + assert isinstance(result, dict) + assert "tool" in result + assert result["tool"] == tool.name + + +# ------------------------------------------------------------------ +# Image-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeImageSpecific: + + def test_default_model_is_qwen_image_2_pro(self): + tool = DashscopeImage() + assert tool.input_schema["properties"]["model"]["default"] == "qwen-image-2.0-pro" + + def test_default_size_uses_asterisk_format(self): + """CRITICAL: DashScope uses W*H (asterisk), not WxH.""" + tool = DashscopeImage() + size_default = tool.input_schema["properties"]["size"]["default"] + assert "*" in size_default + assert "x" not in size_default.lower() + + def test_cost_positive_for_image(self): + tool = DashscopeImage() + assert tool.estimate_cost({"prompt": "test", "n": 1}) > 0.0 + + def test_cost_scales_with_n(self): + tool = DashscopeImage() + cost1 = tool.estimate_cost({"prompt": "test", "n": 1}) + cost3 = tool.estimate_cost({"prompt": "test", "n": 3}) + assert cost3 > cost1 + + def test_build_payload_uses_asterisk_size(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "test"}) + assert "*" in payload["parameters"]["size"] + + def test_build_payload_includes_messages_structure(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "a cat"}) + assert "input" in payload + assert "messages" in payload["input"] + assert payload["input"]["messages"][0]["content"][0]["text"] == "a cat" + + def test_build_payload_optional_negative_prompt(self): + tool = DashscopeImage() + payload = tool._build_payload({ + "prompt": "test", + "negative_prompt": "blurry", + }) + assert payload["parameters"]["negative_prompt"] == "blurry" + + def test_build_payload_omits_negative_prompt_when_absent(self): + tool = DashscopeImage() + payload = tool._build_payload({"prompt": "test"}) + assert "negative_prompt" not in payload["parameters"] + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeImage._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# PR review regressions: multi-image download + idempotency keys +# ------------------------------------------------------------------ + +class TestDashscopeImageMultiOutput: + """Regression tests for PR #240 review: the tool advertised + multiple_outputs and accepted n>1 but only downloaded the first image. + Verify every returned URL is saved and returned as an artifact.""" + + def test_extract_image_urls_across_choices(self): + data = { + "output": { + "choices": [ + {"finish_reason": "stop", "message": {"content": [{"image": "https://x/1.png"}]}}, + {"finish_reason": "stop", "message": {"content": [{"image": "https://x/2.png"}]}}, + {"finish_reason": "stop", "message": {"content": [{"image": "https://x/3.png"}]}}, + ] + } + } + assert DashscopeImage._extract_image_urls(data) == [ + "https://x/1.png", + "https://x/2.png", + "https://x/3.png", + ] + + def test_extract_image_urls_within_single_choice(self): + data = { + "output": { + "choices": [ + {"finish_reason": "stop", "message": {"content": [ + {"image": "https://x/1.png"}, + {"image": "https://x/2.png"}, + ]}} + ] + } + } + assert DashscopeImage._extract_image_urls(data) == [ + "https://x/1.png", + "https://x/2.png", + ] + + def test_extract_image_urls_empty_when_no_images(self): + assert DashscopeImage._extract_image_urls({}) == [] + assert DashscopeImage._extract_image_urls( + {"output": {"choices": []}} + ) == [] + assert DashscopeImage._extract_image_urls( + {"output": {"choices": [{"message": {"content": [{"text": "x"}]}}]}} + ) == [] + + def test_extract_image_urls_skips_failed_choices(self): + """Per Qwen Cloud docs, a multi-output task can be SUCCEEDED with + partial failures. Choices with finish_reason != "stop" must be + skipped so we don't download partial/empty results. The failed + choice here carries a non-empty URL to prove it is the + finish_reason filter (not the truthy-url check) that skips it.""" + data = { + "output": { + "choices": [ + { + "finish_reason": "stop", + "message": {"content": [{"image": "https://x/ok.png"}]}, + }, + { + "finish_reason": "content_filter", + "message": {"content": [{"image": "https://x/blocked.png"}]}, + }, + ] + } + } + assert DashscopeImage._extract_image_urls(data) == ["https://x/ok.png"] + + def test_resolve_output_paths_single_unchanged(self): + paths = DashscopeImage._resolve_output_paths("foo.png", 1) + assert paths == [Path("foo.png")] + + def test_resolve_output_paths_multiple_inserts_index(self): + paths = DashscopeImage._resolve_output_paths("foo.png", 3) + assert paths == [ + Path("foo_1.png"), + Path("foo_2.png"), + Path("foo_3.png"), + ] + + def test_resolve_output_paths_multiple_without_extension(self): + paths = DashscopeImage._resolve_output_paths("foo", 2) + assert paths == [Path("foo_1"), Path("foo_2")] + + def test_execute_downloads_all_images(self, monkeypatch, tmp_path): + """The bug: n=3 returned images_generated=3 but downloaded 1 file. + Mock the DashScope response with 3 URLs and verify all 3 are saved.""" + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key") + + class FakeResp: + def __init__(self, payload, content=b""): + self._payload = payload + self.content = content + + def raise_for_status(self): + pass + + def json(self): + return self._payload + + api_response = { + "output": { + "choices": [ + {"finish_reason": "stop", "message": {"content": [{"image": f"https://x/{i}.png"}]}} + for i in range(1, 4) + ] + }, + "usage": {"image_count": 3}, + } + + import requests + + monkeypatch.setattr( + requests, "post", lambda *a, **kw: FakeResp(api_response) + ) + monkeypatch.setattr( + requests, + "get", + lambda url, **kw: FakeResp({}, content=f"img-{url}".encode()), + ) + + out = tmp_path / "shot.png" + result = DashscopeImage().execute({ + "prompt": "test", "n": 3, "output_path": str(out), + }) + + assert result.success is True + assert result.data["images_generated"] == 3 + assert len(result.artifacts) == 3 + assert (tmp_path / "shot_1.png").exists() + assert (tmp_path / "shot_2.png").exists() + assert (tmp_path / "shot_3.png").exists() + + def test_execute_single_image_uses_base_path(self, monkeypatch, tmp_path): + """n=1 must keep the legacy single-path behavior (no _1 suffix).""" + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key") + + class FakeResp: + def __init__(self, payload, content=b""): + self._payload = payload + self.content = content + + def raise_for_status(self): + pass + + def json(self): + return self._payload + + api_response = { + "output": { + "choices": [ + {"finish_reason": "stop", "message": {"content": [{"image": "https://x/1.png"}]}} + ] + }, + "usage": {"image_count": 1}, + } + + import requests + + monkeypatch.setattr( + requests, "post", lambda *a, **kw: FakeResp(api_response) + ) + monkeypatch.setattr( + requests, + "get", + lambda url, **kw: FakeResp({}, content=b"img-bytes"), + ) + + out = tmp_path / "shot.png" + result = DashscopeImage().execute({ + "prompt": "test", "n": 1, "output_path": str(out), + }) + + assert result.success is True + assert result.data["images_generated"] == 1 + assert result.artifacts == [str(out)] + assert out.exists() + assert not (tmp_path / "shot_1.png").exists() + + +class TestDashscopeIdempotencyKeys: + """Regression tests for PR #240 review: idempotency keys must include + all output-affecting fields so different requests don't collide and + reuse stale artifacts.""" + + def test_image_idempotency_includes_all_output_fields(self): + fields = DashscopeImage().idempotency_key_fields + for field in ( + "prompt", "model", "size", "n", + "negative_prompt", "seed", "prompt_extend", "watermark", + ): + assert field in fields, f"image idempotency missing {field}" + + def test_image_idempotency_differs_on_negative_prompt(self): + tool = DashscopeImage() + base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "negative_prompt": "blurry"} + ) + + def test_image_idempotency_differs_on_seed(self): + tool = DashscopeImage() + base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "seed": 42} + ) + + def test_image_idempotency_differs_on_prompt_extend(self): + tool = DashscopeImage() + base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1} + assert tool.idempotency_key( + {**base, "prompt_extend": True} + ) != tool.idempotency_key({**base, "prompt_extend": False}) + + def test_image_idempotency_differs_on_watermark(self): + tool = DashscopeImage() + base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1} + assert tool.idempotency_key( + {**base, "watermark": False} + ) != tool.idempotency_key({**base, "watermark": True}) + + def test_tts_idempotency_includes_instructions(self): + assert "instructions" in DashscopeTTS().idempotency_key_fields + + def test_tts_idempotency_differs_on_instructions(self): + tool = DashscopeTTS() + base = { + "text": "hi", "voice": "Cherry", + "model": "qwen3-tts-flash", "language_type": "Auto", + } + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "instructions": "speak softly"} + ) + + def test_asr_idempotency_includes_enable_words_and_language_hints(self): + fields = DashscopeAsr().idempotency_key_fields + assert "enable_words" in fields + assert "language_hints" in fields + + def test_asr_idempotency_differs_on_enable_words(self): + tool = DashscopeAsr() + base = {"audio_url": "https://x/a.mp3", "model": "qwen3-asr-flash-filetrans"} + assert tool.idempotency_key( + {**base, "enable_words": True} + ) != tool.idempotency_key({**base, "enable_words": False}) + + def test_asr_idempotency_differs_on_language_hints(self): + tool = DashscopeAsr() + base = {"audio_url": "https://x/a.mp3", "model": "qwen3-asr-flash-filetrans"} + assert tool.idempotency_key( + {**base, "language_hints": ["zh"]} + ) != tool.idempotency_key( + {**base, "language_hints": ["zh", "en"]} + ) + + +# ------------------------------------------------------------------ +# TTS-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeTtsSpecific: + + def test_default_model_is_qwen3_tts_flash(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["model"]["default"] == "qwen3-tts-flash" + + def test_default_voice_is_cherry(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["voice"]["default"] == "Cherry" + + def test_default_language_is_auto(self): + tool = DashscopeTTS() + assert tool.input_schema["properties"]["language_type"]["default"] == "Auto" + + def test_cost_scales_with_text_length(self): + tool = DashscopeTTS() + cost_short = tool.estimate_cost({"text": "hi"}) + cost_long = tool.estimate_cost({"text": "hi " * 100}) + assert cost_long > cost_short + + def test_build_payload_includes_input_text_voice(self): + tool = DashscopeTTS() + payload = tool._build_payload({"text": "hello", "voice": "Ethan"}) + assert payload["input"]["text"] == "hello" + assert payload["input"]["voice"] == "Ethan" + + def test_build_payload_adds_instructions_for_instruct_model(self): + tool = DashscopeTTS() + payload = tool._build_payload({ + "text": "hello", + "instructions": "speak softly", + }) + assert payload["input"]["instructions"] == "speak softly" + assert payload["input"]["optimize_instructions"] is True + + def test_fallback_includes_piper(self): + """Piper is the free offline fallback — must be in fallback list.""" + tool = DashscopeTTS() + assert "piper_tts" in tool.fallback_tools + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeTTS._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# ASR-specific tests +# ------------------------------------------------------------------ + +class TestDashscopeAsrSpecific: + + def test_default_model_is_filetrans(self): + """CRITICAL: must use qwen3-asr-flash-filetrans, NOT qwen3-asr-flash. + The sync version does not support word-level timestamps.""" + tool = DashscopeAsr() + assert tool.input_schema["properties"]["model"]["default"] == "qwen3-asr-flash-filetrans" + + def test_execution_mode_is_async(self): + tool = DashscopeAsr() + assert tool.execution_mode == ExecutionMode.ASYNC + + def test_default_enable_words_is_true(self): + """Word-level timestamps must be enabled by default.""" + tool = DashscopeAsr() + assert tool.input_schema["properties"]["enable_words"]["default"] is True + + def test_default_language_hints_includes_zh_en(self): + tool = DashscopeAsr() + hints = tool.input_schema["properties"]["language_hints"]["default"] + assert "zh" in hints + assert "en" in hints + + def test_rejects_local_file_path(self, monkeypatch): + """audio_url must be a public URL — local paths are rejected.""" + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": "/local/path/audio.mp3"}) + assert result.success is False + assert "publicly accessible URL" in result.error + + def test_rejects_relative_path(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": "audio.mp3"}) + assert result.success is False + assert "publicly accessible URL" in result.error + + def test_rejects_empty_url(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing") + tool = DashscopeAsr() + result = tool.execute({"audio_url": ""}) + assert result.success is False + assert "required" in result.error.lower() + + def test_rejects_no_key(self, monkeypatch): + monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False) + tool = DashscopeAsr() + result = tool.execute({"audio_url": "https://example.com/audio.mp3"}) + assert result.success is False + assert "DASHSCOPE_API_KEY" in result.error + + def test_build_payload_enables_words(self): + tool = DashscopeAsr() + payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"}) + assert payload["parameters"]["enable_words"] is True + + def test_build_payload_includes_file_url(self): + """qwen3-asr-flash-filetrans uses file_url (singular string), + NOT file_urls (plural array) like paraformer-v2.""" + tool = DashscopeAsr() + payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"}) + assert payload["input"]["file_url"] == "https://x.com/a.mp3" + + def test_extract_words_normalizes_ms_to_seconds(self): + """Word timestamps from DashScope are in milliseconds; the tool + must normalize to seconds for downstream subtitle building.""" + fake_transcription = { + "transcripts": [ + { + "sentences": [ + { + "words": [ + {"text": "hello", "begin_time": 1000, "end_time": 1500}, + {"text": "world", "begin_time": 1500, "end_time": 2000}, + ] + } + ] + } + ] + } + words = DashscopeAsr._extract_words(fake_transcription) + assert len(words) == 2 + assert words[0]["text"] == "hello" + assert words[0]["begin_time_seconds"] == 1.0 + assert words[0]["end_time_seconds"] == 1.5 + assert words[1]["begin_time_seconds"] == 1.5 + assert words[1]["end_time_seconds"] == 2.0 + + def test_extract_words_handles_empty_transcription(self): + words = DashscopeAsr._extract_words({}) + assert words == [] + + def test_is_public_url_accepts_https(self): + assert DashscopeAsr._is_public_url("https://example.com/audio.mp3") is True + + def test_is_public_url_rejects_local(self): + assert DashscopeAsr._is_public_url("/local/path/audio.mp3") is False + assert DashscopeAsr._is_public_url("audio.mp3") is False + assert DashscopeAsr._is_public_url("ftp://example.com/audio.mp3") is False + + def test_safe_error_redacts_key(self, monkeypatch): + monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345") + redacted = DashscopeAsr._safe_error( + Exception("failed with key secret-key-12345") + ) + assert "secret-key-12345" not in redacted + assert "[redacted]" in redacted + + +# ------------------------------------------------------------------ +# Registry discovery +# ------------------------------------------------------------------ + +class TestDashscopeRegistryDiscovery: + + def test_all_three_tools_discoverable(self): + from tools.tool_registry import ToolRegistry + registry = ToolRegistry() + registry.discover() + dashscope_tools = [ + t for t in registry._tools.values() + if t.provider == "dashscope" + ] + names = {t.name for t in dashscope_tools} + assert names == {"dashscope_image", "dashscope_tts", "dashscope_asr"} + + def test_image_selector_finds_dashscope(self): + """image_selector should auto-discover dashscope_image by capability.""" + from tools.graphics.image_selector import ImageSelector + selector = ImageSelector() + # Selector discovers providers by capability="image_generation" + # dashscope_image has that capability, so it should be routable + assert DashscopeImage().capability == "image_generation" + + def test_tts_selector_finds_dashscope(self): + """tts_selector should auto-discover dashscope_tts by capability.""" + from tools.audio.tts_selector import TTSSelector + selector = TTSSelector() + assert DashscopeTTS().capability == "tts" diff --git a/tests/contracts/test_phase3_contracts.py b/tests/contracts/test_phase3_contracts.py index 1bd8aaa7..6f225f14 100644 --- a/tests/contracts/test_phase3_contracts.py +++ b/tests/contracts/test_phase3_contracts.py @@ -161,7 +161,14 @@ class TestCapabilityMetadata: catalog = reg.capability_catalog() assert "tts" in catalog providers = {item["provider"] for item in catalog["tts"] if item["provider"] != "selector"} - assert providers == {"doubao", "elevenlabs", "google_tts", "openai", "piper"} + assert providers == { + "dashscope", + "doubao", + "elevenlabs", + "google_tts", + "openai", + "piper", + } # ---- Animated Explainer Pipeline ---- diff --git a/tools/analysis/dashscope_asr.py b/tools/analysis/dashscope_asr.py new file mode 100644 index 00000000..36932eca --- /dev/null +++ b/tools/analysis/dashscope_asr.py @@ -0,0 +1,386 @@ +"""DashScope (Alibaba Cloud Bailian) ASR with word-level timestamps. + +Uses the DashScope-native async transcription endpoint with +X-DashScope-Async: enable header. The model qwen3-asr-flash-filetrans is the +ONLY DashScope path that returns word-level timestamps (the sync +qwen3-asr-flash via /chat/completions does not). + +Pattern: submit (POST) -> poll (GET /tasks/{task_id}) -> download +transcription_url -> parse transcripts[].sentences[].words[]. + +This tool replaces the broken `whisperx` slot for subtitle-aligned +transcription. Word timestamps are normalized from milliseconds to seconds. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class DashscopeAsr(BaseTool): + name = "dashscope_asr" + version = "0.1.0" + tier = ToolTier.ANALYZE + capability = "analysis" + provider = "dashscope" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.ASYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n" + " Get one at https://dashscope.aliyun.com/" + ) + fallback = "transcriber" + fallback_tools = ["transcriber"] + agent_skills = ["dashscope"] + + capabilities = [ + "speech_to_text", + "word_timestamps", + "multilingual", + ] + supports = { + "word_timestamps": True, + "multilingual": True, + "offline": False, + } + best_for = [ + "word-level timestamp transcription for subtitle alignment", + "Mandarin and English speech recognition", + "replacing whisperx when word-level granularity is needed", + ] + not_good_for = [ + "real-time transcription", + "local/offline transcription", + ] + + input_schema = { + "type": "object", + "required": ["audio_url"], + "properties": { + "audio_url": { + "type": "string", + "description": ( + "Publicly accessible URL of the audio file to transcribe. " + "Must be reachable by DashScope servers — local paths " + "are not supported." + ), + }, + "model": { + "type": "string", + "enum": ["qwen3-asr-flash-filetrans"], + "default": "qwen3-asr-flash-filetrans", + }, + "language_hints": { + "type": "array", + "items": {"type": "string"}, + "default": ["zh", "en"], + "description": ( + "Language hints to improve accuracy. " + 'Examples: ["zh", "en", "ja"].' + ), + }, + "enable_words": { + "type": "boolean", + "default": True, + "description": ( + "Enable word-level timestamps. Required for subtitle " + "alignment." + ), + }, + "output_path": {"type": "string"}, + "poll_interval_seconds": { + "type": "number", + "default": 5.0, + "minimum": 1.0, + }, + "timeout_seconds": { + "type": "integer", + "default": 300, + "minimum": 30, + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=20, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, + backoff_seconds=2.0, + retryable_errors=["timeout", "rate_limit"], + ) + idempotency_key_fields = ["audio_url", "model", "enable_words", "language_hints"] + side_effects = [ + "writes transcription JSON to output_path", + "calls DashScope (Alibaba Cloud) ASR API (async submit + poll)", + ] + user_visible_verification = [ + "Check transcription text for accuracy", + "Verify word-level timestamps before building subtitles", + ] + + SUBMIT_URL = ( + "https://dashscope.aliyuncs.com/api/v1/services/audio/asr/" + "transcription" + ) + POLL_URL_TEMPLATE = ( + "https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}" + ) + + def get_status(self) -> ToolStatus: + if os.environ.get("DASHSCOPE_API_KEY"): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + # DashScope ASR pricing is per-minute; check console for actual cost. + return 0.0 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = os.environ.get("DASHSCOPE_API_KEY") + if not api_key: + return ToolResult( + success=False, + error="DASHSCOPE_API_KEY not set. " + self.install_instructions, + ) + + audio_url = inputs.get("audio_url", "").strip() + if not audio_url: + return ToolResult( + success=False, error="audio_url is required." + ) + if not self._is_public_url(audio_url): + return ToolResult( + success=False, + error=( + "audio_url must be a publicly accessible URL (http/https). " + "DashScope servers fetch the file; local paths are not " + "supported. Upload the audio to a public location first." + ), + ) + # DashScope ASR rejects http:// URLs with InvalidParameter.MalformedURL; + # upgrade to https:// before submitting. Note: signed OSS URLs with + # query params (Expires, Signature) may also be rejected — prefer clean + # public file URLs when possible. + if audio_url.startswith("http://"): + audio_url = "https://" + audio_url[len("http://"):] + inputs = {**inputs, "audio_url": audio_url} + + start = time.time() + try: + result = self._transcribe(inputs, api_key=api_key) + except Exception as exc: + return ToolResult( + success=False, + error=f"DashScope ASR failed: {self._safe_error(exc)}", + ) + + result.duration_seconds = round(time.time() - start, 2) + return result + + def _transcribe( + self, inputs: dict[str, Any], *, api_key: str + ) -> ToolResult: + import json + import requests + + payload = self._build_payload(inputs) + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "X-DashScope-Async": "enable", + } + + # Submit + submit_resp = requests.post( + self.SUBMIT_URL, headers=headers, json=payload, timeout=(10, 60) + ) + submit_data = self._json_or_raise(submit_resp) + self._raise_for_error(submit_resp.status_code, submit_data) + + task_id = submit_data.get("output", {}).get("task_id") + if not task_id: + raise RuntimeError( + "DashScope ASR submit succeeded but did not return " + "output.task_id" + ) + + # Poll + poll_data = self._poll_task( + requests_module=requests, + api_key=api_key, + task_id=task_id, + poll_interval=float(inputs.get("poll_interval_seconds", 5.0)), + timeout_seconds=int(inputs.get("timeout_seconds", 300)), + ) + + # qwen3-asr-flash-filetrans returns output.result.transcription_url + # (singular "result", NOT "results" array like paraformer-v2) + result = poll_data.get("output", {}).get("result", {}) + transcription_url = result.get("transcription_url") + if not transcription_url: + raise RuntimeError( + "DashScope ASR task succeeded but " + "result.transcription_url missing" + ) + + # Download transcription JSON + trans_resp = requests.get(transcription_url, timeout=120) + trans_resp.raise_for_status() + transcription = trans_resp.json() + + # Save full transcription + output_path = Path( + inputs.get("output_path", "dashscope_asr.json") + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + json.dumps(transcription, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + # Parse word-level timestamps (normalize ms -> seconds) + words = self._extract_words(transcription) + transcripts = transcription.get("transcripts", []) + + return ToolResult( + success=True, + data={ + "provider": "dashscope", + "model": payload["model"], + "audio_url": inputs["audio_url"], + "task_id": task_id, + "transcripts": transcripts, + "words": words, + "word_count": len(words), + "output": str(output_path), + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + model=payload["model"], + ) + + def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]: + return { + "model": inputs.get( + "model", "qwen3-asr-flash-filetrans" + ), + "input": { + "file_url": inputs["audio_url"], + }, + "parameters": { + "enable_words": bool(inputs.get("enable_words", True)), + "language_hints": inputs.get( + "language_hints", ["zh", "en"] + ), + }, + } + + def _poll_task( + self, + *, + requests_module: Any, + api_key: str, + task_id: str, + poll_interval: float, + timeout_seconds: int, + ) -> dict[str, Any]: + deadline = time.time() + timeout_seconds + headers = {"Authorization": f"Bearer {api_key}"} + while time.time() < deadline: + time.sleep(poll_interval) + resp = requests_module.get( + self.POLL_URL_TEMPLATE.format(task_id=task_id), + headers=headers, + timeout=(10, 60), + ) + data = self._json_or_raise(resp) + self._raise_for_error(resp.status_code, data) + status = data.get("output", {}).get("task_status") + if status == "SUCCEEDED": + return data + if status == "FAILED": + msg = data.get("output", {}).get( + "message", "unknown error" + ) + raise RuntimeError( + f"DashScope ASR task failed: {msg}" + ) + raise TimeoutError( + f"DashScope ASR task {task_id} did not finish within " + f"{timeout_seconds}s" + ) + + @staticmethod + def _is_public_url(url: str) -> bool: + return url.startswith("http://") or url.startswith("https://") + + @staticmethod + def _extract_words( + transcription: dict[str, Any] + ) -> list[dict[str, Any]]: + """Extract flat word list with timestamps normalized to seconds.""" + words: list[dict[str, Any]] = [] + for transcript in transcription.get("transcripts", []): + for sentence in transcript.get("sentences", []): + for word in sentence.get("words", []): + words.append( + { + "text": word.get("text", ""), + "begin_time_seconds": round( + word.get("begin_time", 0) / 1000.0, 3 + ), + "end_time_seconds": round( + word.get("end_time", 0) / 1000.0, 3 + ), + } + ) + return words + + @staticmethod + def _json_or_raise(response: Any) -> dict[str, Any]: + try: + return response.json() + except ValueError as exc: + raise RuntimeError( + f"Non-JSON response from DashScope API: " + f"HTTP {response.status_code}" + ) from exc + + def _raise_for_error( + self, http_status: int, payload: dict[str, Any] + ) -> None: + if http_status < 400: + return + code = payload.get("code") + message = payload.get("message", "unknown error") + raise RuntimeError( + f"DashScope API error: HTTP {http_status}, " + f"code {code}: {message}" + ) + + @staticmethod + def _safe_error(exc: Exception) -> str: + return str(exc).replace( + os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]" + ) diff --git a/tools/audio/dashscope_tts.py b/tools/audio/dashscope_tts.py new file mode 100644 index 00000000..9eb2c2e6 --- /dev/null +++ b/tools/audio/dashscope_tts.py @@ -0,0 +1,243 @@ +"""DashScope (Alibaba Cloud Bailian) text-to-speech via Qwen-TTS models. + +Uses the DashScope-native multimodal-generation endpoint (same as image gen). +The response contains a temporary audio URL (WAV, valid ~24h) that must be +downloaded separately — unlike OpenAI TTS which returns raw audio bytes. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class DashscopeTTS(BaseTool): + name = "dashscope_tts" + version = "0.1.0" + tier = ToolTier.VOICE + capability = "tts" + provider = "dashscope" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n" + " Get one at https://dashscope.aliyun.com/" + ) + fallback = "piper_tts" + fallback_tools = [ + "doubao_tts", + "elevenlabs_tts", + "openai_tts", + "piper_tts", + ] + agent_skills = ["dashscope"] + + capabilities = [ + "text_to_speech", + "voice_selection", + "multilingual", + ] + supports = { + "voice_cloning": False, + "multilingual": True, + "offline": False, + "native_audio": True, + } + best_for = [ + "natural Mandarin and multilingual narration via Qwen-TTS", + "cost-effective TTS via Alibaba Cloud", + "Chinese-language voiceover production", + ] + not_good_for = [ + "fully offline production", + "voice clone matching", + ] + + input_schema = { + "type": "object", + "required": ["text"], + "properties": { + "text": { + "type": "string", + "description": ( + "Text to convert to speech " + "(max 600 chars for qwen3-tts-flash)." + ), + }, + "model": { + "type": "string", + "enum": [ + "qwen3-tts-flash", + "qwen3-tts-instruct-flash", + "qwen-tts-2025-05-22", + ], + "default": "qwen3-tts-flash", + }, + "voice": { + "type": "string", + "default": "Cherry", + "description": ( + 'DashScope voice name. Examples: "Cherry", "Ethan", ' + '"Chelsie".' + ), + }, + "language_type": { + "type": "string", + "default": "Auto", + "enum": ["Auto", "Chinese", "English", "Japanese", "Korean"], + "description": "Language hint for the TTS model.", + }, + "instructions": { + "type": "string", + "description": ( + "Natural language delivery instructions " + "(only for qwen3-tts-instruct-flash)." + ), + }, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) + idempotency_key_fields = ["text", "voice", "model", "language_type", "instructions"] + side_effects = [ + "writes audio file to output_path", + "calls DashScope (Alibaba Cloud) TTS API", + ] + user_visible_verification = [ + "Listen to generated audio for naturalness and pacing" + ] + + ENDPOINT = ( + "https://dashscope.aliyuncs.com/api/v1/services/aigc/" + "multimodal-generation/generation" + ) + + def get_status(self) -> ToolStatus: + if os.environ.get("DASHSCOPE_API_KEY"): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + # Conservative per-character estimate; DashScope bills by character. + return round(len(inputs.get("text", "")) * 0.000015, 4) + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = os.environ.get("DASHSCOPE_API_KEY") + if not api_key: + return ToolResult( + success=False, + error="DASHSCOPE_API_KEY not set. " + self.install_instructions, + ) + + import requests + + from tools.analysis.audio_probe import probe_duration + + start = time.time() + try: + payload = self._build_payload(inputs) + response = requests.post( + self.ENDPOINT, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json=payload, + timeout=120, + ) + response.raise_for_status() + data = response.json() + + audio_info = data.get("output", {}).get("audio", {}) + audio_url = audio_info.get("url") + if not audio_url: + return ToolResult( + success=False, + error="DashScope TTS returned no audio URL", + ) + + # Download the audio from the temporary URL (valid ~24h). + download = requests.get(audio_url, timeout=120) + download.raise_for_status() + + output_path = Path( + inputs.get("output_path", "dashscope_tts.wav") + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(download.content) + + audio_duration = probe_duration(output_path) + usage = data.get("usage", {}) + + except Exception as e: + return ToolResult( + success=False, + error=f"DashScope TTS failed: {self._safe_error(e)}", + ) + + return ToolResult( + success=True, + data={ + "provider": "dashscope", + "model": payload["model"], + "voice": payload["input"]["voice"], + "language_type": payload["input"].get("language_type", "Auto"), + "text_length": len(inputs["text"]), + "audio_duration_seconds": ( + round(audio_duration, 2) if audio_duration else None + ), + "output": str(output_path), + "audio_url": audio_url, + "usage": usage, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=payload["model"], + ) + + def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]: + input_data: dict[str, Any] = { + "text": inputs["text"], + "voice": inputs.get("voice", "Cherry"), + "language_type": inputs.get("language_type", "Auto"), + } + if inputs.get("instructions"): + input_data["instructions"] = inputs["instructions"] + input_data["optimize_instructions"] = True + + return { + "model": inputs.get("model", "qwen3-tts-flash"), + "input": input_data, + } + + @staticmethod + def _safe_error(exc: Exception) -> str: + return str(exc).replace( + os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]" + ) diff --git a/tools/graphics/dashscope_image.py b/tools/graphics/dashscope_image.py new file mode 100644 index 00000000..784b478c --- /dev/null +++ b/tools/graphics/dashscope_image.py @@ -0,0 +1,273 @@ +"""DashScope (Alibaba Cloud Bailian) image generation via Qwen-Image models. + +Uses the DashScope-native multimodal-generation endpoint (NOT OpenAI-compatible +mode, which only supports /chat/completions and /embeddings). The response +contains a temporary image URL (valid ~24h) that must be downloaded separately. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class DashscopeImage(BaseTool): + name = "dashscope_image" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "image_generation" + provider = "dashscope" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n" + " Get one at https://dashscope.aliyun.com/" + ) + fallback = "grok_image" + fallback_tools = ["grok_image", "openai_image", "flux_image", "recraft_image"] + agent_skills = ["dashscope"] + + capabilities = ["generate_image", "text_to_image"] + supports = { + "multiple_outputs": True, + "aspect_ratio": True, + "resolution": True, + "negative_prompt": True, + "seed": True, + } + best_for = [ + "high-quality image generation with Qwen-Image models", + "Chinese-language prompt understanding", + "cost-effective image generation via Alibaba Cloud", + ] + not_good_for = ["offline generation", "image editing (use grok_image edit mode)"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": {"type": "string"}, + "model": { + "type": "string", + "enum": [ + "qwen-image-2.0-pro", + "qwen-image-max", + "wan2.7-image", + "z-image-turbo", + ], + "default": "qwen-image-2.0-pro", + }, + "size": { + "type": "string", + "default": "1024*1024", + "description": ( + 'Image size as "W*H" (asterisk separator, NOT "x"). ' + 'Examples: "1024*1024", "2048*2048", "2688*1536".' + ), + }, + "n": {"type": "integer", "default": 1, "minimum": 1, "maximum": 6}, + "negative_prompt": { + "type": "string", + "description": "Negative prompt (max 500 chars). Things to avoid in the image.", + }, + "prompt_extend": { + "type": "boolean", + "default": True, + "description": "Enable DashScope prompt auto-rewrite for better results.", + }, + "watermark": {"type": "boolean", "default": False}, + "seed": {"type": "integer", "minimum": 0, "maximum": 2147483647}, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) + idempotency_key_fields = [ + "prompt", + "model", + "size", + "n", + "negative_prompt", + "seed", + "prompt_extend", + "watermark", + ] + side_effects = [ + "writes image file to output_path", + "calls DashScope (Alibaba Cloud) image generation API", + ] + user_visible_verification = [ + "Inspect generated image for relevance and quality" + ] + + ENDPOINT = ( + "https://dashscope.aliyuncs.com/api/v1/services/aigc/" + "multimodal-generation/generation" + ) + + def get_status(self) -> ToolStatus: + if os.environ.get("DASHSCOPE_API_KEY"): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + # Conservative per-image estimate; DashScope bills per image. + # Check the DashScope console for actual pricing. + n = int(inputs.get("n", 1)) + return n * 0.02 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = os.environ.get("DASHSCOPE_API_KEY") + if not api_key: + return ToolResult( + success=False, + error="DASHSCOPE_API_KEY not set. " + self.install_instructions, + ) + + import requests + + start = time.time() + try: + payload = self._build_payload(inputs) + response = requests.post( + self.ENDPOINT, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json=payload, + timeout=180, + ) + response.raise_for_status() + data = response.json() + + image_urls = self._extract_image_urls(data) + if not image_urls: + return ToolResult( + success=False, + error="DashScope returned no image URLs", + ) + + # DashScope bills per image and URLs expire ~24h; save every one. + output_paths = self._resolve_output_paths( + inputs.get("output_path", "dashscope_image.png"), + count=len(image_urls), + ) + for path, url in zip(output_paths, image_urls): + download = requests.get(url, timeout=120) + download.raise_for_status() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(download.content) + + usage = data.get("usage", {}) + n_generated = len(image_urls) + + except Exception as e: + return ToolResult( + success=False, + error=f"DashScope image generation failed: {self._safe_error(e)}", + ) + + return ToolResult( + success=True, + data={ + "provider": "dashscope", + "model": payload["model"], + "prompt": inputs["prompt"], + "size": payload["parameters"]["size"], + "output": str(output_paths[0]), + "outputs": [str(p) for p in output_paths], + "images_generated": n_generated, + "usage": usage, + }, + artifacts=[str(p) for p in output_paths], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=payload["model"], + ) + + @staticmethod + def _extract_image_urls(data: dict[str, Any]) -> list[str]: + """Collect image URLs from every choice whose finish_reason is "stop". + + Per Qwen Cloud docs, a multi-output task is SUCCEEDED if at least one + image is generated; failed choices carry finish_reason != "stop" and + must be skipped to avoid downloading partial/empty results. + """ + urls: list[str] = [] + for choice in data.get("output", {}).get("choices", []): + if choice.get("finish_reason") != "stop": + continue + for item in choice.get("message", {}).get("content", []): + url = item.get("image") + if url: + urls.append(url) + return urls + + @staticmethod + def _resolve_output_paths(base: str, count: int) -> list[Path]: + """Derive distinct paths for `count` images. Single image keeps the + base path unchanged; multiple images insert an index before the + extension (foo.png -> foo_1.png, foo_2.png, ...).""" + base_path = Path(base) + if count <= 1: + return [base_path] + stem = base_path.stem + suffix = base_path.suffix + parent = base_path.parent + return [parent / f"{stem}_{i}{suffix}" for i in range(1, count + 1)] + + def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]: + parameters: dict[str, Any] = { + "size": inputs.get("size", "1024*1024"), + "n": int(inputs.get("n", 1)), + "prompt_extend": bool(inputs.get("prompt_extend", True)), + "watermark": bool(inputs.get("watermark", False)), + } + if inputs.get("negative_prompt"): + parameters["negative_prompt"] = inputs["negative_prompt"] + if inputs.get("seed") is not None: + parameters["seed"] = int(inputs["seed"]) + + return { + "model": inputs.get("model", "qwen-image-2.0-pro"), + "input": { + "messages": [ + { + "role": "user", + "content": [{"text": inputs["prompt"]}], + } + ] + }, + "parameters": parameters, + } + + @staticmethod + def _safe_error(exc: Exception) -> str: + return str(exc).replace( + os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]" + )