模型 - AgentScope

概述

模型层采用两层结构：上层是 Credential，下层是该提供商支持的若干模型族 —— Chat Model、TTS、Embedding 与 Realtime Model。

Credential

ChatModelBase

OpenAIChatModel

AnthropicChatModel

DashScopeChatModel

...

TTSModelBase

DashScopeTTSModel

DashScopeRealtimeTTSModel

EmbeddingModelBase

RealtimeModelBase (coming soon)

Credential 承载某个提供商的 API 认证字段（api_key、base_url 等）。从一个凭证出发，可以列出该提供商在每个模型族下支持的全部可用模型。这种分层与前端的自然交互流程一致 —— 先注册凭证，再从凭证下挑选模型 —— 让界面只需鉴权一次，就能展示该提供商支持的所有模型族。

Chat Model

Chat Model 是驱动 agent 对话与工具调用的 LLM，输入输出可以是文本之外的多模态内容。AgentScope 当前提供以下 Chat Model 类：

提供商	模型类	说明
OpenAI	`OpenAIChatModel`	Chat Completions API，兼容 vLLM 与 OpenAI 兼容端点
OpenAI (Responses)	`OpenAIResponseModel`	Responses API，原生支持推理（o3、o4-mini）
Anthropic	`AnthropicChatModel`	Claude 模型，支持 extended thinking 与 prompt 缓存
DashScope	`DashScopeChatModel`	Qwen 模型，多模态（视觉/音频/视频）、推理
DeepSeek	`DeepSeekChatModel`	OpenAI 兼容，带 prompt cache 命中 token
Gemini	`GeminiChatModel`	Google Gemini 模型，支持多模态
Moonshot	`MoonshotChatModel`	Kimi 模型（OpenAI 兼容）
xAI	`XAIChatModel`	Grok 模型，原生 reasoning effort
Ollama	`OllamaChatModel`	本地 LLM 托管，凭证可选

创建 Chat Model

每个 Chat Model 接收一个凭证、一个模型名，以及可选的提供商专属 Parameters 对象。下面三个 tab 分别展示流式、工具调用与推理三种典型初始化场景：

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,
)

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,
    ),
)

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）调用模型：

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，承载完整的累积内容。

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：

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 消息会被聚合，并用 `<history>` 标签包裹，标注发送者名字；工具调用 / 结果序列保持原 API 格式。

切换到多 agent 模式只需传入 MultiAgent 变体，无需修改 agent 代码：

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()：

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：

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：

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：

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：

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 由两层叠加而成：

基础 schema —— 由 Chat Model 的 Parameters 类通过 model_json_schema() 自动生成，列出全部可调参数（temperature、max_tokens、thinking_enable 等），并给出类型与 API 通用范围。
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。

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：

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 流式输出

创建 TTS Model

每个 TTS Model 接收一个凭证、一个模型名，以及可选的提供商专属 Parameters 对象。下面两个 tab 分别展示标准和实时两种初始化场景：

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,
)

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,
)

所有 TTS 模型通用的构造参数：

参数	类型	说明
`credential`	`CredentialBase`	提供商凭证
`model`	`str`	模型标识符（如 `"qwen3-tts-flash"`）
`parameters`	`Parameters \| None`	提供商专属参数，如 `voice`
`stream`	`bool`	是否流式输出音频

DashScopeRealtimeTTSModel 额外参数：

参数	类型	默认值	说明
`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() 方法合成语音：

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`	可选的提供商特定元数据

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）支持增量推送文本（典型场景：对接 LLM 的流式输出）。两者共享相同的 push() / synthesize() 接口。通过 async with 或手动调用 connect() / close() 管理生命周期：

DashScopeRealtimeTTSModel（Qwen3）以 token 级别粒度产出音频——每次 push() 通常都能返回音频数据。

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 集成 —— 自动拦截 agent 的文本输出并合成语音：

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 文件定义：

Qwen3 TTS

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: {}

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 模型卡片：

from agentscope.credential import DashScopeCredential

cards = DashScopeCredential.list_tts_models()
for card in cards:
    print(f"{card.name} (realtime={card.realtime}): {card.label}")

或直接通过模型类：

from agentscope.tts import DashScopeTTSModel

# Qwen3 TTS 模型
cards = DashScopeTTSModel.list_models()

自定义 TTS Provider

添加新的 TTS Provider，需实现 TTSModelBase 子类并在凭证上注册：

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，即输出向量的维度：

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),
)

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),
)

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),
)

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。

调用 Embedding Model

通过传入一组输入调用模型。纯文本模型接受 list[str]；多模态模型还接受 DataBlock 元素：

async def __call__(
    self,
    inputs: list[str | DataBlock],
    **kwargs: Any,
) -> EmbeddingResponse:

分批与重试由框架自动处理：

输入按模型的批大小切分（DashScope 文本为 10，OpenAI 为 2048，Gemini 为 100，Ollama 为 512）。
所有批次通过 asyncio.gather 并发发出。
每个批次在提供商专属的可重试错误下独立重试，最多 max_retries 次。
结果合并为单个 EmbeddingResponse，并保持输入顺序。

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：

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 哈希命名：

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”）：

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 声明提供商专属的瞬态错误：

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 卡片：

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() 发现模型类：

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。

快速开始

核心模块

部署

其他

模型

概述

Chat Model

创建 Chat Model

调用 Chat Model

生成结构化输出

Formatter

自定义 Provider

步骤 1：定义 Credential

步骤 2：实现 Chat Model

步骤 3：添加 Model Card（可选）

前端集成

什么是 ModelCard

参数 schema 与 override

获取 ModelCard

TTS

创建 TTS Model

调用 TTS Model

实时 TTS（流式输入）

与 Agent 集成

TTS Model Card

自定义 TTS Provider

Embedding

创建 Embedding Model

调用 Embedding Model

多模态 Embedding

Embedding 缓存

自定义 Embedding 提供商

第一步：关联凭证

第二步：实现 Embedding Model

第三步：添加 Model Card（可选）

EmbeddingModelCard

Realtime Model

​概述

​Chat Model

​创建 Chat Model

​调用 Chat Model

​生成结构化输出

​Formatter

​自定义 Provider

​步骤 1：定义 Credential

​步骤 2：实现 Chat Model

​步骤 3：添加 Model Card（可选）

​前端集成

​什么是 ModelCard

​参数 schema 与 override

​获取 ModelCard

​TTS

​创建 TTS Model

​调用 TTS Model

​实时 TTS（流式输入）

​与 Agent 集成

​TTS Model Card

​自定义 TTS Provider

​Embedding

​创建 Embedding Model

​调用 Embedding Model

​多模态 Embedding

​Embedding 缓存

​自定义 Embedding 提供商

​第一步：关联凭证

​第二步：实现 Embedding Model

​第三步：添加 Model Card（可选）

​EmbeddingModelCard

​Realtime Model

概述

Chat Model

创建 Chat Model

调用 Chat Model

生成结构化输出

Formatter

自定义 Provider

步骤 1：定义 Credential

步骤 2：实现 Chat Model

步骤 3：添加 Model Card（可选）

前端集成

什么是 ModelCard

参数 schema 与 override

获取 ModelCard

TTS

创建 TTS Model

调用 TTS Model

实时 TTS（流式输入）

与 Agent 集成

TTS Model Card

自定义 TTS Provider

Embedding

创建 Embedding Model

调用 Embedding Model

多模态 Embedding

Embedding 缓存

自定义 Embedding 提供商

第一步：关联凭证

第二步：实现 Embedding Model

第三步：添加 Model Card（可选）

EmbeddingModelCard

Realtime Model