LLM

An LLM (called a chat model in AgentScope’s class names) drives an agent’s conversation and tool calls, accepting and producing multimodal content beyond plain text. AgentScope currently ships the following chat model classes:

Provider	Model Class
OpenAI	`OpenAIChatModel`
OpenAI (Responses API)	`OpenAIResponseModel`
Anthropic	`AnthropicChatModel`
DashScope	`DashScopeChatModel`
DeepSeek	`DeepSeekChatModel`
Gemini	`GeminiChatModel`
Moonshot	`MoonshotChatModel`
xAI	`XAIChatModel`
Ollama	`OllamaChatModel`

Create Chat Model

Every chat model takes a credential, a model name, and an optional provider-specific Parameters object. The three tabs below show typical setups for streaming, tool calling, and reasoning:

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

Common constructor arguments shared by every chat model:

Argument	Type	Description
`credential`	`CredentialBase`	Provider-specific credential
`model`	`str`	Model identifier (e.g. `"qwen-plus"`)
`parameters`	`Parameters \| None`	Provider-specific parameters such as `temperature`, `thinking_enable`, `parallel_tool_calls`
`stream`	`bool`	Whether to stream output
`max_retries`	`int`	Maximum API retries on failure
`context_size`	`int`	Context window used for context compression
`formatter`	`FormatterBase \| None`	Override message formatter (see Formatter)

Call Chat Model

Invoke the model by calling it with a list of Msg objects, plus optional tools and 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]:

The return type follows the model’s stream setting:

stream=False: awaits a single ChatResponse carrying the full output.
stream=True: awaits an AsyncGenerator[ChatResponse, None]. Intermediate chunks (is_last=False) carry only the delta generated in that step. The final chunk (is_last=True) carries the full accumulated content.

The two tabs below show both modes side by side:

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.")]

    # stream=True: iterate over an async generator of ChatResponse chunks
    async for chunk in await model(msgs):
        if chunk.is_last:
            print("Final:", chunk.content)   # full accumulated content
        else:
            print("Delta:", chunk.content)   # delta only

asyncio.run(main())

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=False,
    )
    msgs = [UserMsg(name="user", content="Count from 1 to 5.")]

    # stream=False: await a single ChatResponse with the full output
    response = await model(msgs)
    print(response.content)   # [TextBlock(text='1, 2, 3, 4, 5')]

asyncio.run(main())

A representative streaming trace, illustrating the delta-then-accumulated pattern:

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')]

Each ChatResponse carries content blocks (TextBlock, ThinkingBlock, ToolCallBlock, DataBlock), an is_last flag, a finished_reason (FinishedReason.COMPLETED or FinishedReason.INTERRUPTED), and a ChatUsage recording token counts and elapsed time.

Interrupt Model Call

Interruption at the model layer means asynchronous cancellation: a model call runs inside an asyncio task, and cancelling that task (raising asyncio.CancelledError) stops the in-flight API request, typically because the user interrupted the agent while it was reasoning. Instead of discarding the partial output, ChatModelBase.__call__ catches the cancellation and returns a final ChatResponse carrying the content accumulated so far, marked with finished_reason=FinishedReason.INTERRUPTED. Normal completions end with FinishedReason.COMPLETED, so downstream code can always distinguish a complete answer from a truncated one:

import asyncio
from agentscope.model import FinishedReason

async def call_model():
    async for chunk in await model(msgs):
        if chunk.is_last and chunk.finished_reason == FinishedReason.INTERRUPTED:
            # Partial content accumulated before the cancellation
            print("Interrupted:", chunk.content)

task = asyncio.create_task(call_model())
# Cancelling the task interrupts the in-flight model call
task.cancel()

This is the model-layer half of agent interruption: the Agent class builds on the same mechanism to stop a running reasoning-acting loop and keep its context consistent. See Interrupt Agent for the agent-level behavior.

Generate Structured Output

When you need a JSON object that conforms to a Pydantic model or JSON schema, call generate_structured_output instead of __call__. It returns a StructuredResponse whose content is a validated dict matching the schema:

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)  # validated dict matching WeatherInfo

asyncio.run(main())

generate_structured_output synthesizes a forced tool call from the schema, then validates and repairs the model’s response.

Formatter

A formatter translates AgentScope’s Msg objects into the list[dict] payload that each provider’s API expects. It is configured via the optional formatter argument on the chat model constructor. Every provider ships two built-in variants:

Variant	Use Case
ChatFormatter (default)	Standard single-agent dialog. Each `Msg` maps 1:1 to an API message, preserving native roles (`user`, `assistant`, `system`).
MultiAgentFormatter	Multi-agent scenarios such as debate or moderation. Consecutive agent messages are grouped and wrapped in `<history>` tags with the sender’s name, while tool call / result sequences keep their native API format.

Switch to multi-agent mode by passing the MultiAgent variant; no agent code changes are required:

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

For non-standard payload shapes (e.g. a provider whose API doesn’t follow the OpenAI or Anthropic conventions), subclass FormatterBase and pass an instance through the same formatter argument.

Custom Provider

You can extend AgentScope with your own model provider by implementing a credential and a chat model, then registering the credential.

Step 1: Define the Credential

Subclass CredentialBase with a unique type discriminator and implement 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

Step 2: Implement the Chat Model

Subclass ChatModelBase, define a Parameters inner class, and implement _call_api. The base class owns retry, streaming accumulation, and interruption handling, so _call_api only needs to translate provider responses into AgentScope ChatResponse chunks:

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,
        )
        # If your API follows the OpenAI format, reuse OpenAIChatFormatter;
        # otherwise implement your own FormatterBase subclass.
        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)
        # Call your provider's API using self.credential.api_key, etc.
        # Return one ChatResponse for stream=False, or an async generator
        # that yields delta ChatResponse chunks for stream=True.
        # Do not accumulate streaming chunks here; ChatModelBase.__call__
        # will append the final accumulated response and mark interruptions.
        ...

For streaming custom providers, yield provider deltas as ChatResponse(content=..., is_last=False). If _call_api ends without yielding a final is_last=True chunk, ChatModelBase.__call__ will accumulate the deltas with ChatResponse.append_chat_response() and emit the final response automatically. If the stream is cancelled, it emits the partial accumulated response with finished_reason=FinishedReason.INTERRUPTED.

Step 3: Add Model Cards (optional)

Drop YAML files into a _models/ directory next to your model implementation. Each file describes one model: its capabilities (input_types, output_types), limits (context_size, output_size), and any per-model parameter_overrides (see Integrate with Frontend for the full card format):

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() then loads every YAML in that directory. To pull cards from a different location (for example, a registry your application maintains separately), pass custom_yaml_dir:

cards = MyProviderChatModel.list_models(custom_yaml_dir="/path/to/cards")

Get Started

Building Blocks

Agent as Service

Others

Create Chat Model

Call Chat Model

Interrupt Model Call

Generate Structured Output

Formatter

Custom Provider

Step 1: Define the Credential

Step 2: Implement the Chat Model

Step 3: Add Model Cards (optional)

​Create Chat Model

​Call Chat Model

​Interrupt Model Call

​Generate Structured Output

​Formatter

​Custom Provider

​Step 1: Define the Credential

​Step 2: Implement the Chat Model

​Step 3: Add Model Cards (optional)

Create Chat Model

Call Chat Model

Interrupt Model Call

Generate Structured Output

Formatter

Custom Provider

Step 1: Define the Credential

Step 2: Implement the Chat Model

Step 3: Add Model Cards (optional)