> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.
# 模型
> 驱动智能体的 LLM、TTS、Embedding 与 Realtime 模型
## 概述
AgentScope 的模型层由 API 凭证（Credential）和模型族组成。一个凭证对应一个 API（例如 OpenAI，Gemini，DashScope等），并对应该 API 下的若干模型族（LLM，TTS，Embedding，Realtime）。
**Credential** 承载某个提供商的 API 认证字段（`api_key`、`base_url` 等）。从一个凭证出发，可以列出该提供商在每个模型族下支持的全部可用模型。
这种分层与前端的自然交互流程一致 —— 先注册凭证，再从凭证下挑选模型 —— 让界面只需鉴权一次，就能展示该提供商支持的所有模型族。
## Chat Model
**Chat Model** 是驱动 agent 对话与工具调用的 LLM，输入输出可以是文本之外的多模态内容。AgentScope 当前提供以下 Chat Model 类：
| 提供商 | 模型类 |
| ------------------ | --------------------- |
| OpenAI | `OpenAIChatModel` |
| OpenAI (Responses) | `OpenAIResponseModel` |
| Anthropic | `AnthropicChatModel` |
| DashScope | `DashScopeChatModel` |
| DeepSeek | `DeepSeekChatModel` |
| Gemini | `GeminiChatModel` |
| Moonshot | `MoonshotChatModel` |
| xAI | `XAIChatModel` |
| Ollama | `OllamaChatModel` |
### 创建 Chat Model
每个 Chat Model 接收一个 API 凭证、一个模型名，以及可选的提供商专属 `Parameters` 对象。下面分别展示流式、工具调用与推理三种典型初始化场景：
```python Streaming theme={null}
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen-plus",
stream=True,
)
```
```python Tools theme={null}
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen-plus",
stream=False,
parameters=DashScopeChatModel.Parameters(
parallel_tool_calls=False,
),
)
```
```python Reasoning theme={null}
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-235b-a22b-thinking-2507",
parameters=DashScopeChatModel.Parameters(
thinking_enable=True,
thinking_budget=2048,
),
)
```
所有 Chat Model 共享的构造参数：
| 参数 | 类型 | 说明 |
| -------------- | ----------------------- | ---------------------------------------------------------------- |
| `credential` | `CredentialBase` | 提供商专属凭证 |
| `model` | `str` | 模型标识符（例如 `"qwen-plus"`） |
| `parameters` | `Parameters \| None` | 提供商专属参数，例如 `temperature`、`thinking_enable`、`parallel_tool_calls` |
| `stream` | `bool` | 是否流式输出 |
| `max_retries` | `int` | API 失败时的最大重试次数 |
| `context_size` | `int` | 上下文窗口大小，用于上下文压缩 |
| `formatter` | `FormatterBase \| None` | 覆盖默认的消息 formatter |
### 调用 Chat Model
通过传入一组 `Msg` 对象（以及可选的 `tools` 和 `tool_choice`）调用模型：
```python theme={null}
async def __call__(
self,
messages: list[Msg],
tools: list[dict] | None = None,
tool_choice: ToolChoice | None = None,
**kwargs: Any,
) -> ChatResponse | AsyncGenerator[ChatResponse, None]:
```
返回类型取决于模型的 `stream` 设置：
* **`stream=False`** —— 返回单个 `ChatResponse`，承载完整输出。
* **`stream=True`** —— 返回 `AsyncGenerator[ChatResponse, None]`。中间 chunk（`is_last=False`）只携带**增量**内容。为了让开发者无需自行累积增量，AgentScope 会在末尾追加一个 `is_last=True` 的 chunk，承载**完整的累积内容**。
```python theme={null}
import asyncio
import os
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg
async def main():
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen-plus",
stream=True,
)
msgs = [UserMsg(name="user", content="Count from 1 to 5.")]
async for chunk in await model(msgs):
if chunk.is_last:
print("Final:", chunk.content) # 完整累积内容
else:
print("Delta:", chunk.content) # 仅增量
asyncio.run(main())
```
一段典型的流式输出示例，展示「增量 → 累积」的模式：
```
Delta: [TextBlock(text='1')]
Delta: [TextBlock(text=', 2,')]
Delta: [TextBlock(text=' 3, ')]
Delta: [TextBlock(text='4, 5')]
Final: [TextBlock(text='1, 2, 3, 4, 5')]
```
每个 `ChatResponse` 包含若干 content block（`TextBlock`、`ThinkingBlock`、`ToolCallBlock`、`DataBlock`）、一个 `is_last` 标志，以及记录 token 数与耗时的 `ChatUsage`。
### 生成结构化输出
当需要返回符合 Pydantic 模型或 JSON schema 的结构化结果时，调用 `generate_structured_output` 而非 `__call__`。它返回一个 `StructuredResponse`，其 `content` 是经过 schema 校验的 dict：
```python theme={null}
import asyncio
import os
from pydantic import BaseModel
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg
class WeatherInfo(BaseModel):
city: str
temperature: float
unit: str
async def main():
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen-plus",
stream=False,
)
response = await model.generate_structured_output(
messages=[UserMsg(name="user", content="What's the weather in Shanghai?")],
structured_model=WeatherInfo,
)
print(response.content) # 符合 WeatherInfo 的 dict
asyncio.run(main())
```
`generate_structured_output` 会基于 schema 合成一个强制工具调用，再对模型输出做校验与修复。
### Formatter
**Formatter** 负责把 AgentScope 的 `Msg` 对象转换为各提供商 API 期望的 `list[dict]` 载荷。它通过 Chat Model 构造函数中可选的 `formatter` 参数配置。每个提供商内置两种 formatter：
| 类型 | 适用场景 |
| ----------------------- | ----------------------------------------------------------------------------------------------- |
| **ChatFormatter**（默认） | 标准的单 agent 对话。每条 `Msg` 1:1 映射为一条 API 消息，保留原始角色（`user`、`assistant`、`system`）。 |
| **MultiAgentFormatter** | 多 agent 场景，例如辩论、moderator 等。连续的 agent 消息会被聚合，并用 `` 标签包裹，标注发送者名字；工具调用 / 结果序列保持原 API 格式。 |
切换到多 agent 模式只需传入 MultiAgent 变体，无需修改 agent 代码：
```python theme={null}
import os
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
from agentscope.formatter import OpenAIMultiAgentFormatter
model = OpenAIChatModel(
credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
model="gpt-4.1",
formatter=OpenAIMultiAgentFormatter(),
)
```
如果提供商的载荷格式不属于 OpenAI 或 Anthropic 风格，开发者可以继承 `FormatterBase` 实现自定义 formatter，并通过同一个 `formatter` 参数传入。
### 自定义 Provider
开发者可以通过实现一个 credential 与一个 chat model，并注册该 credential，把自定义模型提供商接入 AgentScope。
#### 步骤 1：定义 Credential
继承 `CredentialBase`，使用唯一的 `type` 判别字段，并实现 `get_chat_model_class()`：
```python theme={null}
from typing import Literal, Type, TYPE_CHECKING
from pydantic import ConfigDict, Field, SecretStr
from agentscope.credential import CredentialBase
if TYPE_CHECKING:
from agentscope.model import ChatModelBase
class MyProviderCredential(CredentialBase):
model_config = ConfigDict(title="My Provider API")
type: Literal["my_provider_credential"] = "my_provider_credential"
api_key: SecretStr = Field(description="API key for My Provider.")
base_url: str = Field(default="https://api.myprovider.com/v1")
@classmethod
def get_chat_model_class(cls) -> Type["ChatModelBase"]:
from .my_model import MyProviderChatModel
return MyProviderChatModel
```
#### 步骤 2：实现 Chat Model
继承 `ChatModelBase`，定义内部 `Parameters` 类，并实现 `_call_api`：
```python theme={null}
from typing import Literal, Any, AsyncGenerator
from pydantic import BaseModel, Field
from agentscope.model import ChatModelBase, ChatResponse
from agentscope.message import Msg
from agentscope.tool import ToolChoice
from agentscope.formatter import FormatterBase, OpenAIChatFormatter
class MyProviderChatModel(ChatModelBase):
class Parameters(BaseModel):
max_tokens: int | None = Field(default=None, gt=0)
temperature: float | None = Field(default=None, ge=0, le=2)
type: Literal["my_provider_chat"] = "my_provider_chat"
def __init__(
self,
credential: "MyProviderCredential",
model: str,
parameters: Parameters | None = None,
stream: bool = True,
max_retries: int = 3,
context_size: int = 128000,
formatter: FormatterBase | None = None,
) -> None:
super().__init__(
credential=credential,
model=model,
parameters=parameters or self.Parameters(),
stream=stream,
max_retries=max_retries,
context_size=context_size,
)
# 如果 API 兼容 OpenAI 格式，复用 OpenAIChatFormatter；
# 否则自行实现 FormatterBase 子类。
self.formatter = formatter or OpenAIChatFormatter()
async def _call_api(
self,
model_name: str,
messages: list[Msg],
tools: list[dict] | None = None,
tool_choice: ToolChoice | None = None,
**kwargs: Any,
) -> ChatResponse | AsyncGenerator[ChatResponse, None]:
formatted_messages = await self.formatter.format(messages)
# 用 self.credential.api_key 等调用提供商 API
...
```
#### 步骤 3：添加 Model Card（可选）
把 YAML 文件放在模型实现旁边的 `_models/` 目录里。每个文件描述一个模型 —— 它的能力（`input_types`、`output_types`）、上限（`context_size`、`output_size`），以及该模型专属的 `parameter_overrides`：
```yaml theme={null}
name: my-model-v1
label: My Model V1
status: active
input_types:
- text/plain
output_types:
- text/plain
context_size: 128000
output_size: 4096
parameter_overrides:
max_tokens: {"maximum": 4096}
```
`MyProviderChatModel.list_models()` 会加载该目录下的所有 YAML。如果想从其他位置（例如应用自己维护的模型注册表）拉取 Model Card，传入 `custom_yaml_dir`：
```python theme={null}
cards = MyProviderChatModel.list_models(custom_yaml_dir="/path/to/cards")
```
## 前端集成
### 什么是 ModelCard
`ModelCard` 是对模型能力与约束的声明式描述，用于驱动前端 —— 模型选择器、参数表单、能力开关都可以基于它动态渲染，无需在前端硬编码任何提供商相关的逻辑。
每个 `ModelCard` 包含以下字段：
| 字段 | 类型 | 说明 |
| ---------------------- | -------------------------------------- | -------------------------------------------------------------------- |
| `name` | `str` | 模型标识符（例如 `"claude-sonnet-4-6"`） |
| `label` | `str` | 用于展示的可读名称（例如 `"Claude Sonnet 4.6"`） |
| `status` | `"active" \| "deprecated" \| "sunset"` | 模型生命周期状态 |
| `input_types` | `list[str]` | 接受的输入 MIME 类型 —— 前端据此过滤附件上传组件（例如仅在支持 `image/*` 时显示图片按钮） |
| `output_types` | `list[str]` | 模型可输出的 MIME 类型 —— 用于标注模型能力（例如出现 `application/x-thinking` 时启用思考开关） |
| `context_size` | `int` | 最大上下文窗口（token 数） |
| `output_size` | `int` | 最大输出 token 数 |
| `parameter_schema` | `dict` | 最终用于渲染参数表单的 JSON Schema —— 由基础 schema 与 per-model override 合并而来（见下文） |
| `parameters_overrides` | `dict[str, dict]` | 合并前的原始 per-model override |
`input_types` 与 `output_types` 都用 MIME 类型描述模态，常见取值如下：
| MIME 类型 | 含义 |
| ------------------------------------- | -------- |
| `text/plain` | 文本 |
| `application/x-thinking` | 推理 / 思考链 |
| `image/*`（如 `image/png`、`image/jpeg`） | 图片 |
| `audio/*`（如 `audio/wav`、`audio/mp3`） | 音频 |
| `video/*`（如 `video/mp4`） | 视频 |
`claude-sonnet-4-6` 的典型 YAML：
```yaml theme={null}
name: claude-sonnet-4-6
label: Claude Sonnet 4.6
status: active
input_types:
- text/plain
- image/jpeg
- image/png
- image/gif
- image/webp
output_types:
- text/plain
- application/x-thinking
context_size: 1000000
output_size: 65536
parameter_overrides:
max_tokens: {"maximum": 65536}
```
#### 参数 schema 与 override
暴露给前端的 `parameter_schema` 由两层叠加而成：
1. **基础 schema** —— 由 Chat Model 的 `Parameters` 类通过 `model_json_schema()` 自动生成，列出全部可调参数（`temperature`、`max_tokens`、`thinking_enable` 等），并给出类型与 API 通用范围。
2. **per-model override** —— YAML 中的 `parameter_overrides` 块会按字段叠加在基础 schema 之上。
override 之所以重要，是因为同一个 API 下不同模型的可调范围并不一致：每个 Qwen 模型都接受 `max_tokens`，但上限各不相同。借助 override，Model Card 可以收紧某个范围、固定默认值，或隐藏某个不适用的参数。
| Override 写法 | 效果 |
| ------------------------- | ------------------------------------------- |
| `param: { ... }` | 浅合并到基础字段（例如 `max_tokens: {maximum: 16384}`） |
| `param: { hidden: true }` | 在前端隐藏该参数 |
| `param: null` | 完全移除该参数 |
### 获取 ModelCard
通过 credential 类或 model 类调用 `list_models()` 获取 Model Card。`CredentialBase.list_models()` 内部会委托到与之关联的 `ChatModelBase` 子类（通过 `get_chat_model_class()` 拿到），后者会从其 `_models/` 目录加载 YAML。
```python theme={null}
from agentscope.credential import DashScopeCredential
from agentscope.model import AnthropicChatModel
# 通过 credential 类
cards = DashScopeCredential.list_models()
# 也可以直接通过 model 类
cards = AnthropicChatModel.list_models()
for card in cards:
print(f"{card.name}: context={card.context_size}, inputs={card.input_types}")
```
`get_chat_model_class()` 返回对应的 `ChatModelBase` 子类，该子类知道如何定位它的 Model Card YAML：
```python theme={null}
model_cls = DashScopeCredential.get_chat_model_class() # -> DashScopeChatModel
cards = model_cls.list_models() # -> list[ModelCard]
```
这种设计让前端只需一个 credential，就能发现所有可用模型、它们的能力与合法参数范围 —— 无需任何硬编码的提供商逻辑。
## TTS
**TTS Model** 将文本转换为合成语音音频，支持标准模式和实时（流式输入）合成模式。AgentScope 当前提供以下 TTS 模型类：
| 提供商 | 模型类 | 说明 |
| ------------------------------ | ------------------------------------ | ---------------------------------------------------- |
| DashScope | `DashScopeTTSModel` | Qwen3-TTS，多种音色，流式输出 |
| DashScope (Realtime) | `DashScopeRealtimeTTSModel` | Qwen3-TTS WebSocket 流式输入，适合对接 LLM 流式输出 |
| DashScope (CosyVoice Realtime) | `DashScopeCosyVoiceRealtimeTTSModel` | CosyVoice-v3 流式输入，支持 cosyvoice-v3-plus/flash/sambert |
### 创建 TTS Model
每个 TTS Model 接收一个凭证、一个模型名，以及可选的提供商专属 `Parameters` 对象。下面两个 tab 分别展示标准和实时两种初始化场景：
```python 标准模式 theme={null}
import os
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential
tts = DashScopeTTSModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-tts-flash",
parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
stream=True,
)
```
```python 实时模式（Qwen3 流式输入） theme={null}
import os
from agentscope.tts import DashScopeRealtimeTTSModel
from agentscope.credential import DashScopeCredential
tts = DashScopeRealtimeTTSModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-tts-flash-realtime",
parameters=DashScopeRealtimeTTSModel.Parameters(voice="Serena"),
stream=True,
)
```
```python 实时模式（CosyVoice 流式输入） theme={null}
import os
from agentscope.tts import DashScopeCosyVoiceRealtimeTTSModel
from agentscope.credential import DashScopeCredential
tts = DashScopeCosyVoiceRealtimeTTSModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="cosyvoice-v3-plus",
parameters=DashScopeCosyVoiceRealtimeTTSModel.Parameters(voice="longanyang"),
stream=True,
)
```
所有 TTS 模型通用的构造参数：
| 参数 | 类型 | 说明 |
| ------------ | -------------------- | ---------------------------- |
| `credential` | `CredentialBase` | 提供商凭证 |
| `model` | `str` | 模型标识符（如 `"qwen3-tts-flash"`） |
| `parameters` | `Parameters \| None` | 提供商专属参数，如 `voice` |
| `stream` | `bool` | 是否流式输出音频 |
`DashScopeRealtimeTTSModel` 和 `DashScopeCosyVoiceRealtimeTTSModel` 额外参数：
| 参数 | 类型 | 默认值 | 说明 |
| ------------------- | ------------- | ------ | ------------------ |
| `cold_start_length` | `int \| None` | `None` | 首次发送前的最小字符数 |
| `cold_start_words` | `int \| None` | `None` | 首次发送前的最小词数 |
| `max_retries` | `int` | `3` | WebSocket 失败最大重试次数 |
| `retry_delay` | `float` | `5.0` | 初始重试延迟（秒，指数退避） |
### 调用 TTS Model
通过 `synthesize()` 方法合成语音：
```python theme={null}
async def synthesize(
self,
text: str | None = None,
**kwargs: Any,
) -> TTSResponse | AsyncGenerator[TTSResponse, None]:
```
返回类型取决于 `stream` 设置：
* **`stream=False`** — 返回单个 `TTSResponse`，包含完整音频。
* **`stream=True`** — 返回 `AsyncGenerator[TTSResponse, None]`，每个 chunk 携带增量音频 delta；最后一个 chunk 的 `is_last=True`。
每个 `TTSResponse` 包含：
| 字段 | 类型 | 说明 |
| ---------- | ------------------- | -------------------------------------------------------- |
| `content` | `DataBlock \| None` | 音频数据，格式由 `content.source.media_type` 指示（如 `"audio/wav"`） |
| `is_last` | `bool` | 是否为流式最后一个 chunk |
| `usage` | `TTSUsage \| None` | token 统计（`input_tokens`、`output_tokens`）和耗时 `time`（秒） |
| `id` | `str` | 自动生成的唯一标识 |
| `metadata` | `dict \| None` | 可选的提供商特定元数据 |
```python theme={null}
import asyncio
import os
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential
async def main():
tts = DashScopeTTSModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-tts-flash",
parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
stream=True,
)
# 流式合成
async for chunk in await tts.synthesize("你好，世界！"):
if chunk.content:
print(f"音频 chunk: {len(chunk.content.source.data)} bytes")
asyncio.run(main())
```
### 实时 TTS（流式输入）
实时模型（`DashScopeRealtimeTTSModel` 和 `DashScopeCosyVoiceRealtimeTTSModel`）支持增量推送文本（典型场景：对接 LLM 的流式输出）。两者共享相同的 `push()` / `synthesize()` 接口。通过 `async with` 或手动调用 `connect()` / `close()` 管理生命周期：
`DashScopeRealtimeTTSModel`（Qwen3）以 token 级别粒度产出音频——每次 `push()` 通常都能返回音频数据。而 `DashScopeCosyVoiceRealtimeTTSModel` 依赖 CosyVoice 服务端自动分句后才进行合成，音频只在检测到完整句子边界后才返回，因此 `push()` 对不完整的句子可能返回空响应。调用 `synthesize()` 会强制合成所有剩余文本（包括未完成的句子）。
```python theme={null}
import asyncio
import os
from agentscope.tts import DashScopeRealtimeTTSModel
from agentscope.credential import DashScopeCredential
async def main():
tts = DashScopeRealtimeTTSModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-tts-flash-realtime",
parameters=DashScopeRealtimeTTSModel.Parameters(voice="Cherry"),
stream=True,
)
async with tts:
# 增量推送文本，每次 push() 返回当前可用的音频
resp1 = await tts.push("你好，")
if resp1.content:
print("第一次 push 后有音频")
resp2 = await tts.push("今天怎么样？")
if resp2.content:
print("第二次 push 后有音频")
# 结束：刷新剩余缓冲文本并收集最终音频
# text= 可选，传入则追加后再结束；不传则直接结束之前 push 的文本
response = await tts.synthesize()
asyncio.run(main())
```
| 方法 | 说明 |
| -------------- | --------------------- |
| `connect()` | 打开 WebSocket 连接 |
| `push(text)` | 增量追加文本（非阻塞），返回当前可用音频 |
| `synthesize()` | 结束当前 utterance，返回剩余音频 |
| `close()` | 断开连接 |
### 与 Agent 集成
在 agent 层，TTS 通过 [`TTSMiddleware`](/versions/2.0.4dev/zh/building-blocks/middleware#ttsmiddleware) 集成 —— 自动拦截 agent 的文本输出并合成语音：
```python theme={null}
from agentscope.agent import Agent
from agentscope.middleware import TTSMiddleware
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential
agent = Agent(
name="assistant",
model=chat_model,
middlewares=[
TTSMiddleware(
DashScopeTTSModel(
credential=DashScopeCredential(api_key="..."),
model="qwen3-tts-flash",
parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
stream=True,
),
),
],
)
# agent 的回复流中同时包含文本和音频事件
async for event in agent.reply_stream(user_msg):
# TextBlockDeltaEvent — 文本内容
# DataBlockDeltaEvent — 音频内容
...
```
中间件根据 TTS 模式自动选择最优策略：
| TTS 模式 | 中间件行为 |
| ------ | ------------------------- |
| 标准模式 | 等待完整文本生成后一次性合成 |
| 实时模式 | 随文本 delta 到达实时推送，并发流式返回音频 |
### TTS Model Card
`TTSModelCard` 描述 TTS 模型的能力（可用音色、流式支持、参数范围），用于驱动前端模型选择器。每个 card 由模型实现旁的 YAML 文件定义：
```yaml Qwen3 TTS theme={null}
name: qwen3-tts-flash
label: Qwen3-TTS-Flash
status: active
input_types:
- text/plain
output_types:
- audio/wav
voices:
- Cherry
- Serena
- Ethan
- Chelsie
parameter_overrides: {}
```
```yaml CosyVoice Realtime theme={null}
name: cosyvoice-v3-plus
label: CosyVoice-v3-Plus
status: active
realtime: true
input_types:
- text/plain
output_types:
- audio/wav
voices:
- longanyang
- longxiaochun
- longshuo
parameter_overrides: {}
```
`voices` 列表会自动注入 `parameter_schema` 中 `voice` 字段的 enum 约束，前端据此渲染下拉选择器。
| 字段 | 类型 | 说明 |
| ---------------------- | ----------- | -------------------------------------- |
| `name` | `str` | 模型标识符（如 `"qwen3-tts-flash"`） |
| `label` | `str` | 显示名称（如 `"Qwen3-TTS-Flash"`） |
| `status` | `str` | `"active"`、`"deprecated"` 或 `"sunset"` |
| `realtime` | `bool` | 是否支持流式输入 |
| `input_types` | `list[str]` | 接受的输入 MIME 类型（固定 `["text/plain"]`） |
| `output_types` | `list[str]` | 输出 MIME 类型（通常 `["audio/wav"]`） |
| `parameter_schema` | `dict` | 合并后的 JSON Schema，用于前端参数表单 |
| `parameters_overrides` | `dict` | 每模型覆盖（语法同 Chat Model Card） |
获取 TTS 模型卡片：
```python theme={null}
from agentscope.credential import DashScopeCredential
cards = DashScopeCredential.list_tts_models()
for card in cards:
print(f"{card.name} (realtime={card.realtime}): {card.label}")
```
或直接通过模型类：
```python theme={null}
from agentscope.tts import DashScopeTTSModel, DashScopeCosyVoiceRealtimeTTSModel
# Qwen3 TTS 模型
cards = DashScopeTTSModel.list_models()
# CosyVoice Realtime 模型
cosyvoice_cards = DashScopeCosyVoiceRealtimeTTSModel.list_models()
```
### 自定义 TTS Provider
添加新的 TTS Provider，需实现 `TTSModelBase` 子类并在凭证上注册：
```python theme={null}
from typing import Literal, Type, TYPE_CHECKING, AsyncGenerator, Any
from pydantic import BaseModel, Field
from agentscope.tts import TTSModelBase, TTSResponse
from agentscope.credential import CredentialBase
if TYPE_CHECKING:
from agentscope.tts import TTSModelBase as TTSBase
class MyTTSModel(TTSModelBase):
class Parameters(BaseModel):
voice: str = Field(default="default", title="Voice")
type: Literal["my_tts"] = "my_tts"
async def synthesize(
self, text: str | None = None, **kwargs: Any
) -> TTSResponse | AsyncGenerator[TTSResponse, None]:
# 调用你的提供商 API
...
# 在凭证上注册
class MyCredential(CredentialBase):
@classmethod
def get_tts_model_classes(cls) -> list[Type["TTSBase"]]:
return [MyTTSModel]
```
## Embedding
**Embedding Model** 将文本（多模态模型还包括图片、视频等媒体）转换为稠密向量，支撑语义检索、RAG 与记忆召回。AgentScope 当前提供以下 Embedding 模型类：
| 提供商 | 模型类 | 说明 |
| --------- | ------------------------- | ----------------------------------------------------------------------- |
| DashScope | `DashScopeEmbeddingModel` | 文本 + 多模态统一 API（`text-embedding-v4`、`qwen3-vl-embedding` 等），内容感知分批 |
| OpenAI | `OpenAIEmbeddingModel` | `text-embedding-3-small/large`，兼容 OpenAI 兼容端点 |
| Gemini | `GeminiEmbeddingModel` | 文本（`gemini-embedding-001`）与多模态（`gemini-embedding-2`，图片 / 视频 / 音频 / PDF） |
| Ollama | `OllamaEmbeddingModel` | 本地 Embedding 模型（`nomic-embed-text` 等），凭证承载 host URL |
### 创建 Embedding Model
每个 Embedding 模型接收一个凭证、一个模型名，以及可选的 `Parameters` 对象 —— 与 Chat Model 的模式完全一致。`Parameters` 承载 `dimensions`，即输出向量的维度：
```python DashScope theme={null}
import os
from agentscope.embedding import DashScopeEmbeddingModel
from agentscope.credential import DashScopeCredential
model = DashScopeEmbeddingModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="text-embedding-v4",
parameters=DashScopeEmbeddingModel.Parameters(dimensions=1024),
)
```
```python OpenAI theme={null}
import os
from agentscope.embedding import OpenAIEmbeddingModel
from agentscope.credential import OpenAICredential
model = OpenAIEmbeddingModel(
credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
model="text-embedding-3-small",
parameters=OpenAIEmbeddingModel.Parameters(dimensions=1536),
)
```
```python Gemini theme={null}
import os
from agentscope.embedding import GeminiEmbeddingModel
from agentscope.credential import GeminiCredential
model = GeminiEmbeddingModel(
credential=GeminiCredential(api_key=os.environ["GEMINI_API_KEY"]),
model="gemini-embedding-001",
parameters=GeminiEmbeddingModel.Parameters(dimensions=768),
)
```
```python Ollama theme={null}
from agentscope.embedding import OllamaEmbeddingModel
from agentscope.credential import OllamaCredential
model = OllamaEmbeddingModel(
credential=OllamaCredential(host="http://localhost:11434"),
model="nomic-embed-text",
)
```
所有 Embedding 模型共享的构造参数：
| 参数 | 类型 | 说明 |
| ----------------- | ---------------------------- | -------------------------------- |
| `credential` | `CredentialBase` | 提供商专属凭证 |
| `model` | `str` | 模型标识符（例如 `"text-embedding-v4"`） |
| `parameters` | `Parameters \| None` | `dimensions` —— 输出向量维度（默认 `512`） |
| `embedding_cache` | `EmbeddingCacheBase \| None` | 可选缓存，跳过重复的 API 调用（见下文） |
| `context_size` | `int` | 单条输入的最大 token 数 |
| `max_retries` | `int` | 每个批次在可重试错误下的最大重试次数 |
| `retry_delay` | `float` | 两次重试之间的间隔秒数 |
`dimensions` 的合法取值因模型而异 —— 每个 Model Card 通过 `parameter_overrides` 固定支持的 `enum` 与默认值（例如 `text-embedding-v4` 接受 2048 / 1536 / 1024 / ... / 64）。参见 [EmbeddingModelCard](#embeddingmodelcard)。
### 调用 Embedding Model
通过传入一组输入调用模型。纯文本模型接受 `list[str]`；多模态模型还接受 `DataBlock` 元素：
```python theme={null}
async def __call__(
self,
inputs: list[str | DataBlock],
**kwargs: Any,
) -> EmbeddingResponse:
```
分批与重试由框架自动处理：
1. 输入按模型的批大小切分（DashScope 文本为 10，OpenAI 为 2048，Gemini 为 100，Ollama 为 512）。
2. 所有批次通过 `asyncio.gather` **并发**发出。
3. 每个批次在提供商专属的可重试错误下独立重试，最多 `max_retries` 次。
4. 结果合并为单个 `EmbeddingResponse`，并保持输入顺序。
```python theme={null}
import asyncio
import os
from agentscope.embedding import DashScopeEmbeddingModel
from agentscope.credential import DashScopeCredential
async def main():
model = DashScopeEmbeddingModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="text-embedding-v4",
parameters=DashScopeEmbeddingModel.Parameters(dimensions=1024),
)
response = await model(
["What is AgentScope?", "A multi-agent framework."],
)
print(len(response.embeddings)) # 2 —— 每条输入一个向量
print(len(response.embeddings[0])) # 1024
print(response.usage.tokens) # 消耗的总 token 数
print(response.source) # "api" 或 "cache"
asyncio.run(main())
```
每个 `EmbeddingResponse` 包含：
| 字段 | 类型 | 说明 |
| ---------------------------- | ------------------------ | -------------------------------- |
| `embeddings` | `list[Embedding]` | 每条输入一个向量，按输入顺序排列 |
| `usage` | `EmbeddingUsage \| None` | 消耗的 `tokens` 与耗时 `time`（秒） |
| `source` | `"api" \| "cache"` | 结果来自 API 还是缓存 |
| `id` / `created_at` / `type` | `str` | 响应标识与时间戳；`type` 恒为 `"embedding"` |
#### 多模态 Embedding
多模态模型（`DashScopeEmbeddingModel` 搭配 `qwen3-vl-embedding` 等，`GeminiEmbeddingModel` 搭配 `gemini-embedding-2`）在字符串之外还接受 `DataBlock` 输入 —— 图片支持 URL 或 base64，视频仅支持 URL：
```python theme={null}
import asyncio
import os
from agentscope.embedding import DashScopeEmbeddingModel
from agentscope.credential import DashScopeCredential
from agentscope.message import DataBlock, URLSource
async def main():
model = DashScopeEmbeddingModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen3-vl-embedding",
)
response = await model([
"A cat sitting on a windowsill",
DataBlock(
source=URLSource(
url="https://example.com/cat.png",
media_type="image/png",
),
),
])
print(len(response.embeddings)) # 2 —— 每条输入一个向量
asyncio.run(main())
```
多模态模型用**内容感知分批**取代简单的批大小切分：输入会按贪心策略打包，确保每个批次满足模型对总元素数、图片数、视频数的单次请求限制（例如 `qwen3-vl-embedding` 单次允许 20 个元素 / 5 张图片 / 1 个视频，`tongyi-embedding-vision-plus` 允许 20 / 64 / 8）。你无需自行拆分输入。
### Embedding 缓存
通过 `embedding_cache` 参数传入一个 `EmbeddingCacheBase` 实现，即可复用已计算过的向量。内置的 `FileEmbeddingCache` 将每次结果存为 `.npy` 文件，以请求的 SHA-256 哈希命名：
```python theme={null}
import asyncio
import os
from agentscope.embedding import DashScopeEmbeddingModel, FileEmbeddingCache
from agentscope.credential import DashScopeCredential
async def main():
model = DashScopeEmbeddingModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="text-embedding-v4",
embedding_cache=FileEmbeddingCache(
cache_dir="./.cache/embeddings",
max_file_number=1000,
max_cache_size=100, # MB
),
)
r1 = await model(["What is AgentScope?"])
print(r1.source) # "api" —— 首次调用走 API
r2 = await model(["What is AgentScope?"])
print(r2.source) # "cache" —— 相同请求由本地缓存返回
asyncio.run(main())
```
当超过 `max_file_number` 或 `max_cache_size` 时，最旧的文件会被优先淘汰。如需其他后端（Redis、SQLite 等），继承 `EmbeddingCacheBase` 并实现它的四个方法：`store`、`retrieve`、`remove`、`clear`。
### 自定义 Embedding 提供商
新增 Embedding 提供商的步骤与 Chat 提供商一致。
#### 第一步：关联凭证
在你的凭证上重写 `get_embedding_model_class()`（基类默认返回 `None`，表示"不支持 Embedding"）：
```python theme={null}
from typing import Type, TYPE_CHECKING
from agentscope.credential import CredentialBase
if TYPE_CHECKING:
from agentscope.embedding import EmbeddingModelBase
class MyProviderCredential(CredentialBase):
# ... 字段与 get_chat_model_class() 同前 ...
@classmethod
def get_embedding_model_class(cls) -> Type["EmbeddingModelBase"]:
from .my_embedding import MyProviderEmbeddingModel
return MyProviderEmbeddingModel
```
#### 第二步：实现 Embedding Model
继承 `EmbeddingModelBase` 并实现 `_call_api` 处理**单个批次** —— 分批、并发与重试逻辑全部继承自基类。通过 `_get_retryable_exceptions` 声明提供商专属的瞬态错误：
```python theme={null}
from typing import Any, Type
from agentscope.embedding import EmbeddingModelBase, EmbeddingResponse, EmbeddingUsage
class MyProviderEmbeddingModel(EmbeddingModelBase[str]):
def __init__(
self,
credential: "MyProviderCredential",
model: str,
parameters: "MyProviderEmbeddingModel.Parameters | None" = None,
context_size: int = 8192,
max_retries: int = 3,
retry_delay: float = 1.0,
) -> None:
super().__init__(
credential=credential,
model=model,
parameters=parameters,
context_size=context_size,
batch_size=100, # 单次 API 调用的最大条目数
max_retries=max_retries,
retry_delay=retry_delay,
)
@classmethod
def _get_retryable_exceptions(cls) -> tuple[Type[Exception], ...]:
return (TimeoutError,) # 最多重试 max_retries 次
async def _call_api(
self,
inputs: list[str],
**kwargs: Any,
) -> EmbeddingResponse:
# 框架保证 len(inputs) <= self.batch_size。
# 调用你的提供商 API 并返回向量。
...
```
按提供商支持的输入类型绑定泛型参数：纯文本用 `EmbeddingModelBase[str]`，多模态用 `EmbeddingModelBase[str | DataBlock]` —— IDE 会据此为调用方提示正确的 `inputs` 类型。
#### 第三步：添加 Model Card（可选）
将 YAML 文件放入实现旁边的 `_models/` 目录，`MyProviderEmbeddingModel.list_models()` 即可加载 —— 与 Chat Model Card 完全一致。
### EmbeddingModelCard
`EmbeddingModelCard` 是面向前端的 `ModelCard` 对应物，带有 Embedding 专属的默认值 —— 输出类型 `application/x-embedding` 表示该模型产出稠密向量：
| 字段 | 与 `ModelCard` 的差异 |
| ------------------ | ----------------------------------------------------------------------------------------------------- |
| `type` | 恒为 `"embedding_model"` |
| `input_types` | 默认 `["text/plain"]`；多模态卡片追加 `image/*`、`video/*` 等 |
| `output_types` | 默认 `["application/x-embedding"]` |
| `parameter_schema` | 由 Embedding 的 `Parameters` 类（`dimensions`）与 YAML `parameter_overrides` 合并生成 —— override 语义与 Chat 卡片一致 |
| `output_size` | 不存在 —— Embedding 模型没有输出 token 上限 |
典型的 YAML 卡片：
```yaml theme={null}
name: text-embedding-v4
label: Text Embedding v4
status: active
input_types:
- text/plain
output_types:
- application/x-embedding
context_size: 8192
parameter_overrides:
dimensions:
default: 1024
enum: [2048, 1536, 1024, 768, 512, 256, 128, 64]
```
可以直接在模型类上获取卡片，也可以通过凭证的 `get_embedding_model_class()` 发现模型类：
```python theme={null}
from agentscope.credential import DashScopeCredential
from agentscope.embedding import OpenAIEmbeddingModel
# 直接通过模型类
cards = OpenAIEmbeddingModel.list_models()
# 或从凭证发现模型类
embed_cls = DashScopeCredential.get_embedding_model_class()
cards = embed_cls.list_models()
for card in cards:
print(f"{card.name}: context={card.context_size}, inputs={card.input_types}")
```
## Realtime Model
即将上线 —— Realtime Model 支持正在从 v1.0 迁移到 v2.0。