模型与格式化器

AgentScope 2.x 把"调用哪个模型"和"用什么凭据"拆成了两个独立模块：model 负责调用与解析响应，credential 负责持有密钥与端点；formatter 则负责把多 Agent 的 Msg 列表序列化成具体 provider API 需要的字典结构。三者 通过模型构造参数装配，Agent 本身不感知这些细节。

版本基线：本文示例以 agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。所有导入与 构造签名均核对自 src/agentscope/model/、src/agentscope/credential/、src/agentscope/formatter/ 的 __init__.py 与具体实现文件。1.x 的 Trinity model wrapper 已在 2.x 移除，本页不出现。

Model 与 Credential 解耦

2.x 的一次重要破坏性变更是把凭据从 Model 内部拆到了独立的 credential 模块。ChatModelBase 只持有 credential、model（模型名）、parameters（provider 专属参数）、stream、formatter 等字段；密钥、 base_url 这类敏感配置全部由 CredentialBase 子类承载。这样做的好处是：同一个凭据可以同时驱动 chat、 embedding、TTS 三类模型，而模型类只关心 API 调用与响应解析。

CredentialBase 还提供 get_chat_model_class() 与 list_models() 两个类方法，把"凭据 → 对应模型类 → 候选模型列表"这条链路串起来，方便在 Agent Service 等 UI 场景里动态枚举可用模型。

前端类比

如果你来自前端：

• 把 ChatModelBase 类比为 Axios 的 request adapter——它接收标准化的输入（list[Msg]），输出标准化的 结果（ChatResponse），内部把 provider 专属的请求/响应格式藏在 adapter 里。
• 把 Credential 类比为 API client config（baseURL + 拦截器注入的 Authorization）——它和请求逻辑 分离，可以单独传递、序列化或轮换。
• 把 Formatter 类比为 请求拦截器里的 payload transform——把框架统一的 Msg 结构翻译成 OpenAI / Anthropic / Gemini 各自要求的 messages 数组。

AgentScope 原生语义：ChatModelBase.__call__ 是协程，签名是 async def __call__(self, messages, tools=None, tool_choice=None, **kwargs) -> ChatResponse | AsyncGenerator。 当 stream=False 时返回单个 ChatResponse；当 stream=True（默认）时返回一个异步生成器，逐个产出增量 ChatResponse。ChatResponse.content 是 list[TextBlock | ToolCallBlock | ThinkingBlock]，和 Msg.content 共享同一套 ContentBlock 体系。Formatter.format 同样是协程，返回 list[dict]，在每次 _call_api 内部被 await self.formatter.format(messages) 调用。这套语义和前端的"拦截器返回新 payload"不同：formatter 是 模型层的必备组件，不注入也会用 provider 默认的 XxxChatFormatter。

9 个 Provider

下表列出 2.x 内置的全部 9 个 chat model provider，类名、凭据类与对应环境变量均核对自 model/__init__.py、credential/__init__.py 与各 provider 实现。

Provider	模型类	凭据类	环境变量	备注
DashScope（通义千问）	DashScopeChatModel	DashScopeCredential	DASHSCOPE_API_KEY	OpenAI 兼容端点，支持思考模式与多模态
OpenAI（Chat）	OpenAIChatModel	OpenAICredential	OPENAI_API_KEY	标准 Chat Completions API
OpenAI（Responses）	OpenAIResponseModel	OpenAICredential	OPENAI_API_KEY	Responses API，与 Chat 共用同一凭据
Anthropic	AnthropicChatModel	AnthropicCredential	ANTHROPIC_API_KEY	Claude 系列
Gemini	GeminiChatModel	GeminiCredential	GEMINI_API_KEY / GOOGLE_API_KEY	Google 模型
DeepSeek	DeepSeekChatModel	DeepSeekCredential	DEEPSEEK_API_KEY
Moonshot	MoonshotChatModel	MoonshotCredential	MOONSHOT_API_KEY	Kimi
xAI	XAIChatModel	XAICredential	XAI_API_KEY	Grok
Ollama	OllamaChatModel	OllamaCredential	无（本地推理）	本地服务，无需 API key

所有模型类都继承 ChatModelBase，构造方式一致：credential + model 必填，parameters、stream、 formatter、client_kwargs 等可选。Ollama 比较特殊，可以不传 credential，仅传 model 即可连本地服务。

构造与调用

下面以 DashScope 为例，展示模型层的最小调用。密钥通过 os.environ 读取，不要硬编码进源码。

import asyncio
import os

from agentscope.credential import DashScopeCredential
from agentscope.message import TextBlock, UserMsg
from agentscope.model import ChatResponse, DashScopeChatModel

model = DashScopeChatModel(
    credential=DashScopeCredential(
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    model="qwen-plus",
    stream=False,  # 非流式：await 一次返回完整 ChatResponse
)


async def main():
    messages = [UserMsg(name="user", content="用一句话介绍 AgentScope")]
    response: ChatResponse = await model(messages)
    # ChatResponse.content 是 ContentBlock 列表，取出文本需要过滤 TextBlock
    text = "".join(
        block.text for block in response.content if isinstance(block, TextBlock)
    )
    print(text)


asyncio.run(main())

stream 默认为 True。非流式调用把 stream=False 显式传入，await model(messages) 就会跑完整个生成过程 后返回单个 ChatResponse，其 is_last=True、content 包含完整结果。

流式调用

流式是默认行为，await model(messages) 返回一个异步生成器，逐个产出增量 ChatResponse，最后一个 chunk 的 is_last=True 并携带完整 usage：

import asyncio
import os

from agentscope.credential import DashScopeCredential
from agentscope.message import TextBlock, UserMsg
from agentscope.model import DashScopeChatModel

model = DashScopeChatModel(
    credential=DashScopeCredential(
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    model="qwen-plus",
    # stream=True 是默认值，这里显式写出便于说明
    stream=True,
)


async def main():
    messages = [UserMsg(name="user", content="用一句话介绍 AgentScope")]
    stream = await model(messages)
    async for chunk in stream:
        for block in chunk.content:
            if isinstance(block, TextBlock):
                print(block.text, end="", flush=True)
    print()  # 收尾换行


asyncio.run(main())

注意 await model(messages) 本身是一次 await：它返回的是生成器对象，不是单个响应。直接对它做 async for 之前必须先 await。

带工具的调用

模型层也接受 tools 与 tool_choice，但通常你不需要手动构造——Agent 会在 ReAct 循环里自动从 Toolkit 生成并传入。如果要在模型层直接测试工具调用，可参考 Agent 核心 与 工具与 MCP 两页。

Formatter

Formatter 的职责是把 list[Msg] 序列化成 provider API 需要的 list[dict]。每个 provider 都有两个 formatter：

• XxxChatFormatter：单 Agent 场景（一个 user + 一个 agent）。直接用 API 的 name 字段区分消息来源， 每条 Msg 翻译成一条 API 消息。
• XxxMultiAgentFormatter：多 Agent 场景。把连续的非工具消息（agent 之间的对话）聚合成一段文本，包进 <history> 标签作为一条 user 消息发送；工具调用 / 工具结果序列仍按单 Agent formatter 处理。这样能避免 多个 agent 的 name 在不支持 name 字段的 API 上冲突。

Provider	单 Agent	多 Agent
DashScope	DashScopeChatFormatter	DashScopeMultiAgentFormatter
OpenAI (Chat)	OpenAIChatFormatter	OpenAIMultiAgentFormatter
OpenAI (Responses)	OpenAIResponseFormatter	OpenAIResponseMultiAgentFormatter
Anthropic	AnthropicChatFormatter	AnthropicMultiAgentFormatter
Gemini	GeminiChatFormatter	GeminiMultiAgentFormatter
DeepSeek	DeepSeekChatFormatter	DeepSeekMultiAgentFormatter
Moonshot	MoonshotChatFormatter	MoonshotMultiAgentFormatter
xAI	XAIChatFormatter	XAIMultiAgentFormatter
Ollama	OllamaChatFormatter	OllamaMultiAgentFormatter

注入方式

Formatter 通过模型构造参数 formatter= 注入。单 Agent 场景无需手动传入--模型会自动选用对应 provider 的 XxxChatFormatter；多 Agent 场景则需要显式注入 XxxMultiAgentFormatter：

import os

from agentscope.credential import DashScopeCredential
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.model import DashScopeChatModel

model = DashScopeChatModel(
    credential=DashScopeCredential(
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    model="qwen-plus",
    # 多 Agent 场景显式注入 MultiAgentFormatter
    formatter=DashScopeMultiAgentFormatter(),
)

切换 provider 时记得同步换 formatter，或干脆不传让模型用默认值。

自定义 Formatter

如果默认 formatter 不能满足需求（例如需要注入额外的 system reminder、改写历史、或对接非标准端点），可以 继承 FormatterBase 并实现 format 协程。FormatterBase 是 pydantic BaseModel，format 是 abstractmethod，签名为 async def format(self, msgs) -> list[dict]：

from agentscope.formatter import FormatterBase
from agentscope.message import Msg


class StripTextFormatter(FormatterBase):
    """只保留 role 与文本内容，丢弃工具调用与多模态块。"""

    async def format(self, msgs: list[Msg]) -> list[dict]:
        self.assert_list_of_msgs(msgs)
        return [
            {
                "role": msg.role,
                "content": msg.get_text_content() or "",
            }
            for msg in msgs
        ]

assert_list_of_msgs 是 FormatterBase 提供的静态方法，用于校验输入确实是 list[Msg]，建议在自定义 format 开头调用。注入方式与内置 formatter 一致：DashScopeChatModel(..., formatter=StripTextFormatter())。

结构化输出预览

模型层提供 generate_structured_output 方法，让模型按 Pydantic 模型或 JSON Schema 生成结构化结果。它 返回 StructuredResponse，其 content 是经过校验的 dict：

import asyncio
import os

from pydantic import BaseModel

from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg
from agentscope.model import DashScopeChatModel


class City(BaseModel):
    name: str
    country: str


model = DashScopeChatModel(
    credential=DashScopeCredential(
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    model="qwen-plus",
    stream=False,
)


async def main():
    response = await model.generate_structured_output(
        messages=[UserMsg(name="user", content="给出巴黎的结构化信息")],
        structured_model=City,
    )
    print(response.content)
    # 例如：{"name": "巴黎", "country": "法国"}


asyncio.run(main())

底层默认实现会构造一个名为 generate_structured_output 的工具并强制模型调用它，再把工具调用参数解析、 校验后包装成 StructuredResponse。部分 provider（如 DashScope 思考模式）不支持强制 tool_choice，会 自动降级为 auto 并靠注入的系统提示引导。完整的用法、边界与失败处理见 结构化输出 页。

ModelCard 与 list_models

ModelCard 描述一个候选模型的元信息：name、label、status（active / deprecated / sunset）、 context_size、output_size、input_types、output_types 等。每个模型类都有 list_models() 类方法， 从同目录的 _models/*.yaml 加载模型卡片：

from agentscope.model import DashScopeChatModel, ModelCard

cards: list[ModelCard] = DashScopeChatModel.list_models()
for card in cards:
    print(card.name, card.context_size, card.status)

也可以从凭据侧调用——CredentialBase.list_models() 会自动找到对应的模型类并委托执行：

from agentscope.credential import DashScopeCredential

cards = DashScopeCredential.list_models()

选择模型时，如果后续要接入 Agent（ReAct 循环），必须选支持 tool calling 的模型。ModelCard 本身 不直接暴露"是否支持工具调用"的布尔字段，需要结合 provider 文档与 output_types 判断；通常各 provider 的旗舰模型（qwen-plus、gpt-4.1、claude-sonnet 系列、qwen2.5:7b 等）都支持。

常见错误

模型不支持 tool calling

Agent 的 ReAct 循环依赖模型返回 ToolCallBlock。如果选了不支持工具调用的模型，Agent 会在第一轮就卡住 或反复返回纯文本。接入 Agent 前先确认模型支持 function calling，或在模型层单独测试 tools= 参数。

KeyError: 'DASHSCOPE_API_KEY'

凭据未设置。确认已在当前 shell export，或写在 .env 并由启动脚本加载。不要把 key 硬编码进源码或提交 到 Git。各 provider 对应的环境变量见上方 9 个 Provider 表。

Formatter 与 provider 不匹配

每个 formatter 是针对特定 API 格式写的。把 OpenAIChatFormatter 注入 AnthropicChatModel 会导致请求体 结构不符合 Anthropic API 规范而报错。换 provider 时记得同步换 formatter，或干脆不传 formatter= 让模型 用默认值。

把流式返回当成单响应

stream=True（默认）时 await model(messages) 返回的是异步生成器，不是 ChatResponse。如果直接 response = await model(messages) 后访问 response.content，会得到一个生成器对象而非响应。需要 async for chunk in response 消费，或显式设置 stream=False 走非流式路径。

先修

• 消息与事件 - 理解 Msg、ContentBlock 与 ChatResponse 的关系

下一步

• Agent 核心 - 把 Model + Formatter + Toolkit 装配进 ReAct 循环

官方参考

• Model
• Formatter
• Credential
• AgentScope GitHub - model 模块
• AgentScope GitHub - formatter 模块