chore: import upstream snapshot with attribution
K8s Workspace Integration Tests / k8s-workspace-tests (push) Has been cancelled
Pre-commit / run (ubuntu-latest) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.11) (push) Has been cancelled
Web UI / check (push) Has been cancelled

This commit is contained in:
wehub-resource-sync
2026-07-13 12:39:27 +08:00
commit c3bf08ac8d
755 changed files with 172616 additions and 0 deletions
+231
View File
@@ -0,0 +1,231 @@
# Model Call Examples
This directory contains example scripts for the major LLM providers supported by AgentScope, together with a unified test runner `run_tests.py`.
These scripts are designed to verify that AgentScope's chat model components function correctly across various input scenarios.
---
## Directory Layout
```
scripts/model_examples/
├── run_tests.py # Unified test runner
├── _utils.py # Shared helpers (stream_and_collect)
├── test.jpeg # Sample image for multimodal tests
├── openai_chat_call.py # OpenAI Chat Completions basic + tool call + structured output
├── openai_chat_multiagent.py # OpenAI Chat Completions multi-agent conversation
├── openai_chat_multimodal.py # OpenAI Chat Completions image/text multimodal
├── openai_chat_multiagent_multimodal.py
├── openai_response_call.py # OpenAI Responses API reasoning models (o1/o3)
├── openai_response_multiagent.py
├── openai_response_multimodal.py
├── openai_response_multiagent_multimodal.py
├── anthropic_call.py # Anthropic Claude
├── anthropic_multiagent.py
├── anthropic_multimodal.py
├── anthropic_multiagent_multimodal.py
├── dashscope_call.py # Alibaba DashScope / Qwen
├── dashscope_multiagent.py
├── dashscope_multimodal.py
├── dashscope_multiagent_multimodal.py
├── deepseek_call.py # DeepSeek (no multimodal support)
├── deepseek_multiagent.py
├── gemini_call.py # Google Gemini
├── gemini_multiagent.py
├── gemini_multimodal.py
├── gemini_multiagent_multimodal.py
├── moonshot_call.py # Moonshot AI (Kimi)
├── moonshot_multiagent.py
├── moonshot_multimodal.py
├── moonshot_multiagent_multimodal.py
├── xai_call.py # xAI Grok
├── xai_multiagent.py
├── xai_multimodal.py
├── xai_multiagent_multimodal.py
├── ollama_call.py # Ollama local models (requires a running server)
├── ollama_multiagent.py
├── ollama_multimodal.py
└── ollama_multiagent_multimodal.py
```
---
## Test Types
| Suffix | File Pattern | What it covers |
|---|---|---|
| `call` | `*_call.py` | Basic text call + two-round tool calling + structured output |
| `multiagent` | `*_multiagent.py` | Multi-agent scenario using `MultiAgentFormatter` |
| `multimodal` | `*_multimodal.py` | Image + text multimodal input (some providers also test audio/video) |
| `multiagent_multimodal` | `*_multiagent_multimodal.py` | Multi-agent + multimodal combined |
---
## Providers and Their Environment Variables
| Provider | Env Variable | Notes |
|---|---|---|
| `openai_chat` | `OPENAI_API_KEY` | Chat Completions API gpt-4.1, etc. |
| `openai_response` | `OPENAI_API_KEY` | Responses API o1, o3, o4-mini, etc. |
| `anthropic` | `ANTHROPIC_API_KEY` | Claude models, supports extended thinking |
| `dashscope` | `DASHSCOPE_API_KEY` | Qwen series, supports `thinking_enable` |
| `deepseek` | `DEEPSEEK_API_KEY` | Supports only `call` / `multiagent` (no multimodal) |
| `gemini` | `GEMINI_API_KEY` | Gemini models, supports `thinking_budget` |
| `moonshot` | `MOONSHOT_API_KEY` | Moonshot AI kimi-k2.6, etc. |
| `xai` | `XAI_API_KEY` | Grok models, supports `reasoning_effort` |
| `ollama` | *(none auto-detect)* | Local server, default `http://localhost:11434` |
---
## Quick Start
### 1. Export API Keys
Set the environment variables for the providers you want to test:
```bash
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export DASHSCOPE_API_KEY="sk-..."
export DEEPSEEK_API_KEY="sk-..."
export GEMINI_API_KEY="AIza..."
export MOONSHOT_API_KEY="sk-..."
export XAI_API_KEY="xai-..."
```
For Ollama, no API key is required. Just make sure the server is running:
```bash
ollama serve
ollama pull qwen3:14b # pull the default model used in the scripts
```
### 2. Check Provider Availability
```bash
python scripts/model_examples/run_tests.py --list
```
Sample output:
```
Provider Env Var Available Description
openai_chat OPENAI_API_KEY YES OpenAI Chat Completions API
anthropic ANTHROPIC_API_KEY NO Anthropic Claude models
...
```
### 3. Run All Available Tests
```bash
python scripts/model_examples/run_tests.py
```
The runner auto-detects which providers have credentials, skips those that do not, and runs all test types for the rest.
---
## `run_tests.py` Reference
```
usage: run_tests.py [-h] [--providers NAME[,NAME...]] [--tests TYPE[,TYPE...]]
[--timeout SECONDS] [--list] [--verbose]
```
### Options
| Option | Short | Default | Description |
|---|---|---|---|
| `--providers` | `-p` | all | Comma-separated list of providers to run |
| `--tests` | `-t` | all | Comma-separated list of test types to run |
| `--timeout` | | `120` | Per-script timeout in seconds |
| `--list` | `-l` | | Print provider status table and exit |
| `--verbose` | `-v` | | Stream each script's output in real time. By default output is suppressed and shown only when a test fails. |
### Examples
```bash
# Only test specific providers
python scripts/model_examples/run_tests.py --providers openai_chat,anthropic
# Only run a specific test type (across all available providers)
python scripts/model_examples/run_tests.py --tests call
# Combine: run call + multiagent tests for dashscope and deepseek
python scripts/model_examples/run_tests.py -p dashscope,deepseek -t call,multiagent
# Only run multimodal tests
python scripts/model_examples/run_tests.py --tests multimodal,multiagent_multimodal
# Increase per-script timeout
python scripts/model_examples/run_tests.py --timeout 180
# Check provider status
python scripts/model_examples/run_tests.py --list
```
### Summary Table
At the end of a run, a summary table is printed:
```
Provider Test Type Status Time
---------------------- ---------------------------- -------- -------
openai_chat call PASS 12.3s
openai_chat multiagent PASS 8.1s
anthropic call SKIP (env var ANTHROPIC_API_KEY not set)
deepseek call PASS 15.7s
deepseek multimodal SKIP (not supported)
Total: 12 | PASS: 8 | FAIL: 0 | SKIP: 4
```
| Status | Meaning |
|---|---|
| **PASS** | Script exited with code 0 |
| **FAIL** | Script exited with a non-zero code or timed out |
| **SKIP** | API key missing, test type not supported, or script file absent |
The runner exits with code `1` if any test fails.
---
## Running a Single Script
Every script can be executed independently once the relevant environment variable is set:
```bash
python scripts/model_examples/openai_chat_call.py
python scripts/model_examples/dashscope_multiagent.py
python scripts/model_examples/ollama_multimodal.py
```
Each script typically defines two or more async functions:
- `example_simple_call()` basic text call with streaming
- `example_tool_call()` two-round conversation with tool/function calling
- `example_structured_output()` force a Pydantic-validated JSON output (in `_call.py` variants, uses a thinking-enabled model)
- `example_image_url()` / `example_image_local_path()` / `example_image_base64()` image + text input (in `_multimodal.py` variants)
- `example_audio()` audio input (e.g. `openai_chat_multimodal.py`, `dashscope_multimodal.py`)
- `example_video()` video input (e.g. `dashscope_multimodal.py`)
---
## Ollama Notes
Ollama runs locally and requires no API key, but you must:
1. Start the service: `ollama serve`
2. Pull the model used by the scripts: `ollama pull qwen3:14b`
3. If the service runs on a non-default address, set: `export OLLAMA_HOST=http://your-host:11434`
`run_tests.py` pings the Ollama host before running any test. If the server is unreachable, all Ollama tests are automatically skipped.
+104
View File
@@ -0,0 +1,104 @@
# -*- coding: utf-8 -*-
"""Shared utility helpers for model-examples scripts."""
import base64 as _b64
from typing import AsyncGenerator
from agentscope.message import (
TextBlock,
ThinkingBlock,
DataBlock,
Base64Source,
)
from agentscope.model import ChatResponse
async def stream_and_collect(
gen: AsyncGenerator[ChatResponse, None],
) -> ChatResponse:
"""Stream delta chunks printing text in real-time; return the final chunk.
Only delta chunks (is_last=False) are printed. The final accumulated
chunk (is_last=True) is returned so callers can read ToolCallBlock objects
from it without re-printing the entire content. Text from the final chunk
is printed only when no text was streamed in any delta chunk (e.g. some
models batch the answer in the last chunk).
Streaming ``DataBlock`` chunks (e.g. omni audio output) are reported as
per-chunk size summaries while the stream is being consumed, so callers
can see the audio arriving incrementally rather than only in the final
cumulative chunk.
"""
final: ChatResponse | None = None
in_thinking = False
text_was_streamed = False
# Track per-block audio progress: block_id -> (media_type, chunk_count,
# total_bytes).
audio_progress: dict[str, list] = {}
async for chunk in gen:
if chunk.is_last:
final = chunk
continue # Skip printing; full content is in the final chunk
for block in chunk.content:
if isinstance(block, ThinkingBlock):
if not in_thinking:
print("[Thinking] ", end="", flush=True)
in_thinking = True
print(block.thinking, end="", flush=True)
elif isinstance(block, TextBlock):
if in_thinking:
print() # Newline after thinking content
print("--- Answer ---")
in_thinking = False
print(block.text, end="", flush=True)
text_was_streamed = True
elif isinstance(block, DataBlock) and isinstance(
block.source,
Base64Source,
):
# Streaming binary delta (e.g. omni audio output).
if in_thinking:
print()
print("--- Answer ---")
in_thinking = False
delta_bytes = len(_b64.b64decode(block.source.data))
state = audio_progress.setdefault(
block.id,
[block.source.media_type, 0, 0],
)
state[1] += 1
state[2] += delta_bytes
print(
f"\n[Audio chunk #{state[1]} ({state[0]}): "
f"+{delta_bytes}B, total={state[2]}B]",
flush=True,
)
# If text was not streamed in any delta, print it from the final chunk now.
if not text_was_streamed and final is not None:
final_text = "".join(
b.text
for b in final.content
if isinstance(b, TextBlock) and b.text
)
if final_text:
if in_thinking:
print()
print("--- Answer ---")
in_thinking = False
print(final_text)
if in_thinking:
print()
# Report cumulative audio output from the final chunk.
if final is not None:
for block in final.content:
if isinstance(block, DataBlock) and isinstance(
block.source,
Base64Source,
):
media = block.source.media_type
byte_size = len(_b64.b64decode(block.source.data))
print(
f"[Audio output (cumulative): {media}, {byte_size} bytes]",
)
print()
assert final is not None
return final
+191
View File
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""Examples of Anthropic Claude model calls."""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import AnthropicChatModel
from agentscope.credential import AnthropicCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the Anthropic model with a simple text message."""
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the Anthropic model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Shanghai?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the Anthropic model and force a structured (JSON) output."""
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""Example of Anthropic Claude model calls with AnthropicMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import AnthropicMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import AnthropicChatModel
from agentscope.credential import AnthropicCredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let claude-opus-4-5
summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = AnthropicMultiAgentFormatter()
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,99 @@
# -*- coding: utf-8 -*-
"""Example of Anthropic Claude model calls with MultiAgentFormatter and
image input.
Demonstrates combining AnthropicMultiAgentFormatter with multimodal (vision)
content: Alice shares an image in a multi-agent conversation, and the model
is asked to summarize what everyone is looking at.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import AnthropicMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import AnthropicChatModel
from agentscope.credential import AnthropicCredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = AnthropicMultiAgentFormatter()
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
@@ -0,0 +1,154 @@
# -*- coding: utf-8 -*-
"""Example of Anthropic Claude model multimodal (vision) calls using
DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import AnthropicChatModel
from agentscope.credential import AnthropicCredential
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call claude-opus-4-5 with an image URL and ask what is in the image."""
model = AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> AnthropicChatModel:
"""Build and return an AnthropicChatModel instance."""
return AnthropicChatModel(
credential=AnthropicCredential(
api_key=os.environ["ANTHROPIC_API_KEY"],
),
model="claude-opus-4-5",
stream=True,
context_size=1_000_000,
parameters=AnthropicChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
async def example_image_local_path() -> None:
"""Call claude-opus-4-5 with a local image using a ``file://`` URL.
The formatter automatically reads the file and converts it to base64.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
image_block = DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
image_block,
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call claude-opus-4-5 with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
image_block = DataBlock(
source=Base64Source(
data=data,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
image_block,
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
+185
View File
@@ -0,0 +1,185 @@
# -*- coding: utf-8 -*-
"""Examples of DashScope (Alibaba) model calls."""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
TextBlock,
)
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the DashScope model with a simple text message."""
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the DashScope model with tool calling enabled.
Uses qwen3-max which supports both thinking mode and tool calling.
"""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Beijing?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the DashScope model and force a structured (JSON) output."""
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-
"""Example of DashScope model calls with DashScopeMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let qwen3.5-plus summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = DashScopeMultiAgentFormatter()
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,95 @@
# -*- coding: utf-8 -*-
"""Example of DashScope model calls with MultiAgentFormatter and image input.
Demonstrates combining DashScopeMultiAgentFormatter with multimodal (vision)
content: Alice shares an image in a multi-agent conversation, and the model
is asked to summarize what everyone is looking at.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = DashScopeMultiAgentFormatter()
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
@@ -0,0 +1,231 @@
# -*- coding: utf-8 -*-
"""Example of DashScope model multimodal (vision) calls using DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
# A publicly accessible test image
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
# A publicly accessible test video
TEST_VIDEO_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241115/cqqkru/1.mp4"
)
# A publicly accessible test audio
TEST_AUDIO_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20250211/tixcef/cherry.wav"
)
async def example_image_url() -> None:
"""Call qwen3.5-plus with an image URL and ask what is in the image."""
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> DashScopeChatModel:
"""Build and return a DashScopeChatModel instance."""
return DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-plus",
stream=True,
context_size=1_000_000,
parameters=DashScopeChatModel.Parameters(thinking_enable=True),
)
async def example_image_local_path() -> None:
"""Call qwen3.5-plus with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to a base64 data
URI.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
image_block = DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
image_block,
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call qwen3.5-plus with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
image_block = DataBlock(
source=Base64Source(
data=data,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
image_block,
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
async def example_video() -> None:
"""Call qwen3.5-plus with a video URL and ask what is in the video."""
model = _build_model()
video_block = DataBlock(
source=URLSource(
url=TEST_VIDEO_URL,
media_type="video/mp4",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this video? "
"Describe it briefly.",
),
video_block,
],
role="user",
),
]
print("=== Multimodal Call (Video URL) ===")
await stream_and_collect(await model(msgs))
async def example_audio() -> None:
"""Call qwen3.5-omni-plus with an audio URL.
Audio understanding requires an Omni model (qwen3.5-omni-plus or
Qwen3-Omni-Flash). Omni models also require stream=True and the
``modalities`` parameter.
"""
model = DashScopeChatModel(
credential=DashScopeCredential(
api_key=os.environ["DASHSCOPE_API_KEY"],
),
model="qwen3.5-omni-plus",
stream=True,
context_size=1_000_000,
)
audio_block = DataBlock(
source=URLSource(
url=TEST_AUDIO_URL,
media_type="audio/wav",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(text="What is being said in this audio clip?"),
audio_block,
],
role="user",
),
]
print("=== Multimodal Call (Audio URL - Omni) ===")
await stream_and_collect(
await model(msgs, modalities=["text", "audio"]),
)
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
asyncio.run(example_video())
asyncio.run(example_audio())
+189
View File
@@ -0,0 +1,189 @@
# -*- coding: utf-8 -*-
"""Examples of DeepSeek model calls.
For DeepSeek, thinking (chain-of-thought) is controlled by the
``thinking_enable`` parameter on ``deepseek-v4-flash``. The legacy
``deepseek-chat`` / ``deepseek-reasoner`` model names will be deprecated;
they correspond to the non-thinking / thinking modes of
``deepseek-v4-flash`` respectively.
"""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import DeepSeekChatModel
from agentscope.credential import DeepSeekCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming, with chain-of-thought)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call DeepSeek with thinking enabled on a simple text message."""
model = DeepSeekChatModel(
credential=DeepSeekCredential(
api_key=os.environ["DEEPSEEK_API_KEY"],
),
model="deepseek-v4-flash",
stream=True,
context_size=1_000_000,
parameters=DeepSeekChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call DeepSeek with thinking + tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = DeepSeekChatModel(
credential=DeepSeekCredential(
api_key=os.environ["DEEPSEEK_API_KEY"],
),
model="deepseek-v4-flash",
stream=True,
context_size=1_000_000,
parameters=DeepSeekChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Shenzhen?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call DeepSeek with thinking enabled and force a structured output."""
model = DeepSeekChatModel(
credential=DeepSeekCredential(
api_key=os.environ["DEEPSEEK_API_KEY"],
),
model="deepseek-v4-flash",
stream=True,
context_size=1_000_000,
parameters=DeepSeekChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
"""Example of DeepSeek model calls with DeepSeekMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import DeepSeekMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import DeepSeekChatModel
from agentscope.credential import DeepSeekCredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let deepseek-v4-flash
summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = DeepSeekMultiAgentFormatter()
model = DeepSeekChatModel(
credential=DeepSeekCredential(
api_key=os.environ["DEEPSEEK_API_KEY"],
),
model="deepseek-v4-flash",
stream=True,
context_size=1_000_000,
parameters=DeepSeekChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
+191
View File
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""Examples of Google Gemini model calls."""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import GeminiChatModel
from agentscope.credential import GeminiCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the Gemini model with a simple text message."""
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the Gemini model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Guangzhou?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the Gemini model and force a structured (JSON) output."""
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
+105
View File
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""Example of Gemini model calls with GeminiMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import GeminiMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import GeminiChatModel
from agentscope.credential import GeminiCredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let gemini-2.5-flash
summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = GeminiMultiAgentFormatter()
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
"""Example of Gemini model calls with MultiAgentFormatter and image input."""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import GeminiMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import GeminiChatModel
from agentscope.credential import GeminiCredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = GeminiMultiAgentFormatter()
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
+143
View File
@@ -0,0 +1,143 @@
# -*- coding: utf-8 -*-
"""Example of Gemini model multimodal (vision) calls using DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import GeminiChatModel
from agentscope.credential import GeminiCredential
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call gemini-2.5-flash with an image URL and ask what is in the image."""
model = GeminiChatModel(
credential=GeminiCredential(
api_key=os.environ["GEMINI_API_KEY"],
),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> GeminiChatModel:
return GeminiChatModel(
credential=GeminiCredential(api_key=os.environ["GEMINI_API_KEY"]),
model="gemini-2.5-flash",
stream=True,
context_size=1_048_576,
parameters=GeminiChatModel.Parameters(
thinking_enable=True,
thinking_budget=1024,
),
)
async def example_image_local_path() -> None:
"""Call gemini-2.5-flash with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to base64.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call gemini-2.5-flash with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
+180
View File
@@ -0,0 +1,180 @@
# -*- coding: utf-8 -*-
"""Examples of Moonshot AI model calls."""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import MoonshotChatModel
from agentscope.credential import MoonshotCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the Moonshot model with a simple text message."""
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the Moonshot model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Xi'an?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the Moonshot model and force a structured (JSON) output."""
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
parameters=MoonshotChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-
"""Example of Moonshot model calls with MoonshotMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags and preserves reasoning_content for Moonshot's
Preserved Thinking feature in multi-turn conversations.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import MoonshotMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import MoonshotChatModel
from agentscope.credential import MoonshotCredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let kimi-k2.6 summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = MoonshotMultiAgentFormatter()
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
parameters=MoonshotChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,91 @@
# -*- coding: utf-8 -*-
"""Example of Moonshot model calls with MoonshotMultiAgentFormatter and
image input."""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import MoonshotMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import MoonshotChatModel
from agentscope.credential import MoonshotCredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = MoonshotMultiAgentFormatter()
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
parameters=MoonshotChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
"""Example of Moonshot model multimodal (vision) calls using DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import MoonshotChatModel
from agentscope.credential import MoonshotCredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call kimi-k2.6 with an image URL and ask what is in the image."""
model = MoonshotChatModel(
credential=MoonshotCredential(
api_key=os.environ["MOONSHOT_API_KEY"],
),
model="kimi-k2.6",
stream=True,
context_size=262_144,
parameters=MoonshotChatModel.Parameters(thinking_enable=True),
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> MoonshotChatModel:
return MoonshotChatModel(
credential=MoonshotCredential(api_key=os.environ["MOONSHOT_API_KEY"]),
model="kimi-k2.6",
stream=True,
context_size=262_144,
parameters=MoonshotChatModel.Parameters(thinking_enable=True),
)
async def example_image_local_path() -> None:
"""Call kimi-k2.6 with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to a base64 data
URI.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call kimi-k2.6 with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
+188
View File
@@ -0,0 +1,188 @@
# -*- coding: utf-8 -*-
"""Examples of Ollama (local) model calls.
Make sure Ollama is running locally before running these examples:
ollama serve
Pull the model first if you haven't already:
ollama pull qwen3:14b
"""
import asyncio
import json
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import OllamaChatModel
from agentscope.credential import OllamaCredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the Ollama model with a simple text message."""
model = OllamaChatModel(
credential=OllamaCredential(
host="http://localhost:11434",
),
model="qwen3:14b",
stream=True,
context_size=40_960,
parameters=OllamaChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the Ollama model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = OllamaChatModel(
credential=OllamaCredential(
host="http://localhost:11434",
),
model="qwen3:14b",
stream=True,
context_size=40_960,
parameters=OllamaChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Nanjing?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the Ollama model and force a structured (JSON) output."""
model = OllamaChatModel(
credential=OllamaCredential(
host="http://localhost:11434",
),
model="qwen3:14b",
stream=True,
context_size=40_960,
parameters=OllamaChatModel.Parameters(thinking_enable=True),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,96 @@
# -*- coding: utf-8 -*-
"""Example of Ollama model calls with OllamaMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
from _utils import stream_and_collect
from agentscope.formatter import OllamaMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import OllamaChatModel
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let qwen3:14b summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = OllamaMultiAgentFormatter()
model = OllamaChatModel(
model="qwen3:14b",
stream=True,
context_size=40_960,
parameters=OllamaChatModel.Parameters(thinking_enable=True),
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,91 @@
# -*- coding: utf-8 -*-
"""Example of Ollama model calls with MultiAgentFormatter and image input.
Requires a multimodal Ollama model such as llava. Run `ollama pull llava`
first.
"""
import asyncio
from _utils import stream_and_collect
from agentscope.formatter import OllamaMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import OllamaChatModel
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group.
Requires `ollama pull llava` to have been run first.
"""
formatter = OllamaMultiAgentFormatter()
model = OllamaChatModel(
model="llava:7b",
stream=True,
context_size=4_096,
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
+136
View File
@@ -0,0 +1,136 @@
# -*- coding: utf-8 -*-
"""Example of Ollama model multimodal (vision) calls using DataBlock.
Note: Vision input requires a multimodal Ollama model such as llava, bakllava,
or moondream. Run `ollama pull llava` before executing this example.
"""
import asyncio
import base64
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import OllamaChatModel
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call llava:7b with an image URL and ask what is in the image.
Requires `ollama pull llava` to be run first.
"""
model = OllamaChatModel(
model="llava:7b",
stream=True,
context_size=4_096,
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> OllamaChatModel:
return OllamaChatModel(
model="llava:7b",
stream=True,
context_size=4_096,
)
async def example_image_local_path() -> None:
"""Call llava:7b with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to base64.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call llava:7b with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
+191
View File
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""Examples of OpenAI Chat Completions model calls."""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the OpenAI Chat model with a simple text message."""
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o4-mini",
stream=True,
context_size=200_000,
parameters=OpenAIChatModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the OpenAI Chat model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o4-mini",
stream=True,
context_size=200_000,
parameters=OpenAIChatModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Chengdu?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the OpenAI Chat model and force a structured (JSON) output."""
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o4-mini",
stream=True,
context_size=200_000,
parameters=OpenAIChatModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Chat model calls with OpenAIMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import OpenAIMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let gpt-4.1 summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = OpenAIMultiAgentFormatter()
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,90 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Chat model calls with MultiAgentFormatter and image
input."""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import OpenAIMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = OpenAIMultiAgentFormatter()
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
@@ -0,0 +1,194 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Chat model multimodal (vision) calls using DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import OpenAIChatModel
from agentscope.credential import OpenAICredential
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
# A publicly accessible test audio
TEST_AUDIO_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20250211/tixcef/cherry.wav"
)
async def example_image_url() -> None:
"""Call gpt-4.1 with an image URL and ask what is in the image."""
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> OpenAIChatModel:
return OpenAIChatModel(
credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
)
async def example_image_local_path() -> None:
"""Call gpt-4.1 with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to a base64 data
URI.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call gpt-4.1 with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
async def example_audio() -> None:
"""Call gpt-audio-mini with an audio URL.
Audio understanding requires an audio-capable model such as
``gpt-audio-mini``. The formatter converts the audio source
to the ``input_audio`` format expected by the Chat Completions API.
"""
model = OpenAIChatModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-audio-mini",
stream=True,
)
audio_block = DataBlock(
source=URLSource(
url=TEST_AUDIO_URL,
media_type="audio/wav",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(text="What is being said in this audio clip?"),
audio_block,
],
role="user",
),
]
print("=== Multimodal Call (Audio Input and Output) ===")
response = await stream_and_collect(
await model(
msgs,
modalities=["text", "audio"],
audio={"voice": "alloy", "format": "pcm16"},
),
)
# Save audio if present
for block in response.content:
if isinstance(block, DataBlock) and block.source.media_type.startswith(
"audio/",
):
audio_bytes = base64.b64decode(block.source.data)
print(f" Audio received: {len(audio_bytes)} bytes")
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
asyncio.run(example_audio())
@@ -0,0 +1,199 @@
# -*- coding: utf-8 -*-
"""Examples of OpenAI Responses API model calls.
The Responses API (vs Chat Completions API) provides first-class streaming
events for reasoning/thinking content, making it a natural fit for reasoning
models such as o3 and o4-mini.
"""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import OpenAIResponseModel
from agentscope.credential import OpenAICredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the OpenAI Response model with a simple text message."""
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o1",
stream=True,
context_size=200_000,
parameters=OpenAIResponseModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the OpenAI Response model with tool calling enabled."""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o1",
stream=True,
context_size=200_000,
parameters=OpenAIResponseModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Hangzhou?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
# Responses API: use call_id (call_xxx) so that the
# function_call_output.call_id matches
# function_call.call_id
id=tool_call.call_id or tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the OpenAI Response model and force a structured (JSON) output."""
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="o1",
stream=True,
context_size=200_000,
parameters=OpenAIResponseModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
@@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Responses API model calls with
OpenAIResponseMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags, enabling the model to handle multi-agent
conversations where more than one non-user agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import OpenAIResponseMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import OpenAIResponseModel
from agentscope.credential import OpenAICredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let gpt-4.1 (Responses API)
summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = OpenAIResponseMultiAgentFormatter()
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,90 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Responses API model calls with MultiAgentFormatter and
image input."""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import OpenAIResponseMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import OpenAIResponseModel
from agentscope.credential import OpenAICredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = OpenAIResponseMultiAgentFormatter()
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
@@ -0,0 +1,138 @@
# -*- coding: utf-8 -*-
"""Example of OpenAI Responses API model multimodal (vision) calls using
DataBlock."""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import OpenAIResponseModel
from agentscope.credential import OpenAICredential
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call gpt-4.1 (Responses API) with an image URL and ask what is in
the image."""
model = OpenAIResponseModel(
credential=OpenAICredential(
api_key=os.environ["OPENAI_API_KEY"],
),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> OpenAIResponseModel:
return OpenAIResponseModel(
credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
model="gpt-4.1",
stream=True,
context_size=1_047_576,
)
async def example_image_local_path() -> None:
"""Call gpt-4.1 (Responses API) with a local image using a ``file://`` URL.
The formatter reads the file from disk and converts it to a base64 data
URI.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call gpt-4.1 (Responses API) with a local image using explicit base64.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())
+585
View File
@@ -0,0 +1,585 @@
# -*- coding: utf-8 -*-
"""Unified test runner for AgentScope model scripts.
Automatically reads API keys from environment variables and only runs tests
for providers whose keys are present. Supports fine-grained control over
which providers and test types to run.
Usage examples:
# Run all available tests (auto-detected from env vars)
python scripts/model_examples/run_tests.py
# Run only specific providers
python scripts/model_examples/run_tests.py --providers openai_chat,
anthropic,gemini
# Run only specific test types
python scripts/model_examples/run_tests.py --tests call,multiagent
# Combine: only call tests for openai and dashscope
python scripts/model_examples/run_tests.py --providers openai_chat,
dashscope --tests call
# List all providers and their env var / availability status
python scripts/model_examples/run_tests.py --list
# Include ollama even if server may not be running
python scripts/model_examples/run_tests.py --providers ollama
# Set a per-test timeout (seconds, default 120)
python scripts/model_examples/run_tests.py --timeout 60
# Stream each script's output to the terminal (default: suppressed,
shown only on failure)
python scripts/model_examples/run_tests.py --verbose
"""
import argparse
import os
import subprocess
import sys
import textwrap
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
# ---------------------------------------------------------------------------
# Provider definitions
# ---------------------------------------------------------------------------
SCRIPTS_DIR = Path(__file__).parent
@dataclass
class Provider:
"""Metadata for a model provider."""
name: str
env_var: Optional[str] # None for local providers (e.g. Ollama)
file_prefix: str # e.g. "openai_chat_model"
supported_tests: list[str] = field(default_factory=list)
description: str = ""
def is_available(self) -> bool:
"""Return True if the provider's credentials are present."""
if self.env_var is None:
# Ollama: check if server is reachable
return _ollama_is_running()
return bool(os.environ.get(self.env_var, "").strip())
def script_path(self, test_type: str) -> Optional[Path]:
"""Return the script path for the given test type, or None if it
doesn't exist."""
path = SCRIPTS_DIR / f"{self.file_prefix}_{test_type}.py"
return path if path.exists() else None
def _ollama_is_running() -> bool:
"""Check whether an Ollama server is reachable."""
import urllib.request
import urllib.error
host = os.environ.get("OLLAMA_HOST", "http://localhost:11434")
try:
with urllib.request.urlopen(f"{host}/api/tags", timeout=3):
pass
return True
except Exception:
return False
ALL_PROVIDERS: list[Provider] = [
Provider(
name="openai_chat",
env_var="OPENAI_API_KEY",
file_prefix="openai_chat",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="OpenAI Chat Completions API (gpt-4.1, etc.)",
),
Provider(
name="openai_response",
env_var="OPENAI_API_KEY",
file_prefix="openai_response",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="OpenAI Responses API (o1, o3, etc.)",
),
Provider(
name="anthropic",
env_var="ANTHROPIC_API_KEY",
file_prefix="anthropic",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="Anthropic Claude models",
),
Provider(
name="dashscope",
env_var="DASHSCOPE_API_KEY",
file_prefix="dashscope",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="Alibaba DashScope / Qwen models",
),
Provider(
name="deepseek",
env_var="DEEPSEEK_API_KEY",
file_prefix="deepseek",
supported_tests=["call", "multiagent"],
description="DeepSeek models (no multimodal support)",
),
Provider(
name="gemini",
env_var="GEMINI_API_KEY",
file_prefix="gemini",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="Google Gemini models",
),
Provider(
name="moonshot",
env_var="MOONSHOT_API_KEY",
file_prefix="moonshot",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="Moonshot AI (Kimi) models",
),
Provider(
name="xai",
env_var="XAI_API_KEY",
file_prefix="xai",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="xAI Grok models",
),
Provider(
name="ollama",
env_var=None,
file_prefix="ollama",
supported_tests=[
"call",
"multiagent",
"multimodal",
"multiagent_multimodal",
],
description="Ollama local models (requires running server)",
),
]
PROVIDER_MAP: dict[str, Provider] = {p.name: p for p in ALL_PROVIDERS}
ALL_TEST_TYPES = ["call", "multiagent", "multimodal", "multiagent_multimodal"]
# ---------------------------------------------------------------------------
# Result tracking
# ---------------------------------------------------------------------------
PASS = "PASS"
FAIL = "FAIL"
SKIP = "SKIP"
COLOR_GREEN = "\033[92m"
COLOR_RED = "\033[91m"
COLOR_YELLOW = "\033[93m"
COLOR_BLUE = "\033[94m"
COLOR_RESET = "\033[0m"
COLOR_BOLD = "\033[1m"
def _color(text: str, color: str) -> str:
"""Wrap *text* in ANSI *color* escape codes when stdout is a TTY."""
if sys.stdout.isatty():
return f"{color}{text}{COLOR_RESET}"
return text
@dataclass
class TestResult:
"""Result of a single test run."""
provider: str
test_type: str
status: str # PASS / FAIL / SKIP
reason: str = ""
duration: float = 0.0
output: str = "" # captured stdout+stderr (only when not streaming)
# ---------------------------------------------------------------------------
# Core runner
# ---------------------------------------------------------------------------
def run_script(
script_path: Path,
timeout: int,
verbose: bool,
) -> tuple[str, float, str]:
"""Run a script as a subprocess; return (status, elapsed_seconds, output).
In verbose mode the subprocess output streams directly to the terminal and
the returned output string is empty. In quiet mode (default) output is
captured; it is printed only when the test fails.
"""
start = time.monotonic()
try:
if verbose:
result = subprocess.run(
[sys.executable, str(script_path)],
timeout=timeout,
text=True,
check=False,
)
captured = ""
else:
result = subprocess.run(
[sys.executable, str(script_path)],
timeout=timeout,
capture_output=True,
text=True,
check=False,
)
captured = (result.stdout or "") + (result.stderr or "")
elapsed = time.monotonic() - start
status = PASS if result.returncode == 0 else FAIL
return status, elapsed, captured
except subprocess.TimeoutExpired:
elapsed = time.monotonic() - start
return FAIL, elapsed, f"[Timed out after {timeout}s]"
def print_header(text: str) -> None:
"""Print a prominent section header separated by a full-width rule."""
width = 72
print()
print(_color("=" * width, COLOR_BLUE))
print(_color(f" {text}", COLOR_BOLD))
print(_color("=" * width, COLOR_BLUE))
def print_section(text: str) -> None:
"""Print a lightweight subsection heading."""
print()
print(_color(f"--- {text} ---", COLOR_BLUE))
def run_all(
providers: list[str],
test_types: list[str],
timeout: int,
verbose: bool,
) -> list[TestResult]:
"""Run every requested test type for each provider and return all results.
Providers whose credentials are absent are skipped automatically.
For each (provider, test_type) pair the corresponding script file is
located and executed as a subprocess.
Args:
providers: Ordered list of provider names to evaluate.
test_types: List of test-type suffixes to run per provider.
timeout: Per-script timeout in seconds.
verbose: When True, stream subprocess output directly to the terminal.
When False, capture it and print only on failure.
Returns:
A list of :class:`TestResult` objects, one per (provider, test_type).
"""
results: list[TestResult] = []
for pname in providers:
provider = PROVIDER_MAP[pname]
available = provider.is_available()
if not available:
if provider.env_var:
reason = f"env var {provider.env_var} not set"
else:
reason = "Ollama server not reachable"
for ttype in test_types:
results.append(TestResult(pname, ttype, SKIP, reason))
print_section(
f"{pname.upper()} [{_color('SKIP', COLOR_YELLOW)}] —"
f" {reason}",
)
continue
print_section(
f"{pname.upper()} — running tests: {', '.join(test_types)}",
)
for ttype in test_types:
if ttype not in provider.supported_tests:
results.append(
TestResult(
pname,
ttype,
SKIP,
f"not supported by {pname}",
),
)
print(
f" [{_color('SKIP', COLOR_YELLOW)}] {ttype:30s} (not "
f"supported)",
)
continue
script = provider.script_path(ttype)
if script is None:
results.append(
TestResult(pname, ttype, SKIP, "script not found"),
)
print(
f" [{_color('SKIP', COLOR_YELLOW)}] {ttype:30s} ("
f"script not found)",
)
continue
print(f"\n >>> {script.name}")
status, elapsed, captured = run_script(script, timeout, verbose)
elapsed_str = f"{elapsed:.1f}s"
if status == PASS:
label = _color(PASS, COLOR_GREEN)
else:
label = _color(FAIL, COLOR_RED)
# Always show captured output on failure (even in quiet mode)
if captured and not verbose:
print(captured, end="")
print(f" [{label}] {ttype:30s} {elapsed_str}")
results.append(
TestResult(
pname,
ttype,
status,
duration=elapsed,
output=captured,
),
)
return results
# ---------------------------------------------------------------------------
# Summary table
# ---------------------------------------------------------------------------
def print_summary(results: list[TestResult]) -> None:
"""Print a formatted table summarising every test result.
Args:
results: All :class:`TestResult` objects produced by :func:`run_all`.
"""
print_header("TEST SUMMARY")
passes = [r for r in results if r.status == PASS]
fails = [r for r in results if r.status == FAIL]
skips = [r for r in results if r.status == SKIP]
col_p = f"{len(passes):>3}"
col_f = f"{len(fails):>3}"
col_s = f"{len(skips):>3}"
print(f" {'Provider':<22} {'Test Type':<28} {'Status':<8} {'Time':>7}")
print(f" {'-'*22} {'-'*28} {'-'*8} {'-'*7}")
for r in results:
if r.status == PASS:
status_str = _color(r.status, COLOR_GREEN)
elif r.status == FAIL:
status_str = _color(r.status, COLOR_RED)
else:
status_str = _color(r.status, COLOR_YELLOW)
time_str = (
f"{r.duration:.1f}s"
if r.duration
else (r.reason[:20] if r.reason else "")
)
print(
f" {r.provider:<22} {r.test_type:<28} {status_str:<18}"
f" {time_str:>7}",
)
print()
total = len(results)
summary_line = (
f" Total: {total} | "
f"{_color('PASS', COLOR_GREEN)}: {col_p} | "
f"{_color('FAIL', COLOR_RED)}: {col_f} | "
f"{_color('SKIP', COLOR_YELLOW)}: {col_s}"
)
print(summary_line)
print()
if fails:
print(_color(" Failed tests:", COLOR_RED))
for r in fails:
print(f" - {r.provider} / {r.test_type}")
print()
# ---------------------------------------------------------------------------
# --list mode
# ---------------------------------------------------------------------------
def print_provider_list() -> None:
"""Print a status table of all registered providers and their
availability."""
print_header("PROVIDER STATUS")
print(f" {'Provider':<22} {'Env Var':<25} {'Available':<12} Description")
print(f" {'-'*22} {'-'*25} {'-'*12} {'-'*30}")
for p in ALL_PROVIDERS:
avail = p.is_available()
avail_str = (
_color("YES", COLOR_GREEN) if avail else _color("NO", COLOR_RED)
)
env_str = p.env_var or "(local — ping server)"
tests_str = ", ".join(p.supported_tests)
print(f" {p.name:<22} {env_str:<25} {avail_str:<21} {p.description}")
print(f" {'':22} {'Supported tests:':<25} {tests_str}")
print()
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
"""Construct and return the CLI argument parser."""
parser = argparse.ArgumentParser(
description=textwrap.dedent(__doc__),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--providers",
"-p",
metavar="NAME[,NAME...]",
default=None,
help=(
"Comma-separated list of providers to test "
f"(default: all). Available: {', '.join(PROVIDER_MAP)}"
),
)
parser.add_argument(
"--tests",
"-t",
metavar="TYPE[,TYPE...]",
default=None,
help=(
"Comma-separated list of test types to run "
f"(default: all). Available: {', '.join(ALL_TEST_TYPES)}"
),
)
parser.add_argument(
"--timeout",
type=int,
default=120,
metavar="SECONDS",
help="Per-script timeout in seconds (default: 120)",
)
parser.add_argument(
"--list",
"-l",
action="store_true",
help="List all providers with their env var and availability "
"status, then exit",
)
parser.add_argument(
"--verbose",
"-v",
action="store_true",
help=(
"Stream each script's output to the terminal in real time. "
"By default output is suppressed and only shown when a test fails."
),
)
return parser
def main() -> int:
"""Entry point: parse arguments, run tests, and return an exit code.
Returns:
0 if all executed tests passed (or were skipped); 1 if any test failed.
"""
parser = build_parser()
args = parser.parse_args()
if args.list:
print_provider_list()
return 0
# Resolve providers
if args.providers:
requested = [p.strip() for p in args.providers.split(",") if p.strip()]
unknown = [p for p in requested if p not in PROVIDER_MAP]
if unknown:
print(f"Unknown providers: {', '.join(unknown)}")
print(f"Available: {', '.join(PROVIDER_MAP)}")
return 1
providers = requested
else:
providers = list(PROVIDER_MAP)
# Resolve test types
if args.tests:
requested_tests = [
t.strip() for t in args.tests.split(",") if t.strip()
]
unknown_tests = [t for t in requested_tests if t not in ALL_TEST_TYPES]
if unknown_tests:
print(f"Unknown test types: {', '.join(unknown_tests)}")
print(f"Available: {', '.join(ALL_TEST_TYPES)}")
return 1
test_types = requested_tests
else:
test_types = ALL_TEST_TYPES
print_header(
f"AgentScope Model Tests | providers: {len(providers)} | test "
f"types: {len(test_types)}",
)
print(f" Providers : {', '.join(providers)}")
print(f" Test types: {', '.join(test_types)}")
print(f" Timeout : {args.timeout}s per script")
results = run_all(providers, test_types, args.timeout, args.verbose)
print_summary(results)
# Exit with non-zero if any test failed
failed = any(r.status == FAIL for r in results)
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())
Binary file not shown.

After

Width:  |  Height:  |  Size: 485 KiB

+195
View File
@@ -0,0 +1,195 @@
# -*- coding: utf-8 -*-
"""Examples of xAI Grok model calls using the official xai_sdk gRPC client.
Unlike the OpenAI-compatible approach, the xai_sdk provides native access to
xAI-specific features such as server-side agentic tools (web search, X search,
code execution) and extended reasoning control.
"""
import asyncio
import json
import os
from pydantic import BaseModel, Field
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
)
from agentscope.model import XAIChatModel
from agentscope.credential import XAICredential
from agentscope.tool import Toolkit, ToolChoice, FunctionTool
# ---------------------------------------------------------------------------
# Example 1: Simple user message (streaming, with reasoning)
# ---------------------------------------------------------------------------
async def example_simple_call() -> None:
"""Call the xAI reasoning model with a simple text message."""
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-3-mini",
stream=True,
context_size=131_072,
parameters=XAIChatModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is 1 + 1? Answer briefly.")],
role="user",
),
]
print("=== Simple Call ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 2: Tool calling (streaming)
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to query the weather for.
Returns:
A description of the current weather.
"""
return f"The weather in {city} is sunny and 25°C."
async def example_tool_call() -> None:
"""Call the Grok model with tool calling enabled.
Uses grok-4.3, xAI's flagship mainstream model.
"""
toolkit = Toolkit(tools=[FunctionTool(get_weather)])
tools = await toolkit.get_tool_schemas()
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-4.3",
stream=True,
context_size=1_000_000,
)
msgs = [
Msg(
name="user",
content=[TextBlock(text="What is the weather in Wuhan?")],
role="user",
),
]
# First call: model decides to call a tool
print("=== Tool Call - Round 1 ===")
response = await stream_and_collect(
await model(msgs, tools=tools, tool_choice=ToolChoice(mode="auto")),
)
print(response)
tool_calls = [b for b in response.content if isinstance(b, ToolCallBlock)]
if tool_calls:
tool_result_blocks = []
for tool_call in tool_calls:
args = json.loads(tool_call.input)
result = get_weather(**args)
tool_result_blocks.append(
ToolResultBlock(
id=tool_call.id,
name=tool_call.name,
output=result,
state=ToolResultState.SUCCESS,
),
)
assistant_msg = Msg(
name="assistant",
content=response.content,
role="assistant",
)
tool_result_msg = Msg(
name="tool",
content=tool_result_blocks,
role="assistant",
)
msgs = msgs + [assistant_msg, tool_result_msg]
print("=== Tool Call - Round 2 (Final) ===")
await stream_and_collect(await model(msgs))
# ---------------------------------------------------------------------------
# Example 3: Structured output
# ---------------------------------------------------------------------------
class MathSolution(BaseModel):
"""Structured solution to a math problem."""
problem: str = Field(description="The original problem statement")
answer: float = Field(description="The final numeric answer")
steps: list[str] = Field(
description="Step-by-step reasoning leading to the answer",
)
async def example_structured_output() -> None:
"""Call the xAI reasoning model and force a structured (JSON) output."""
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-3-mini",
stream=True,
context_size=131_072,
parameters=XAIChatModel.Parameters(
thinking_enable=True,
reasoning_effort="low",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text=(
"Solve this: A train travels at 60 km/h for "
"2.5 hours. How far does it travel in km?"
),
),
],
role="user",
),
]
print("=== Structured Output ===")
response = await model.generate_structured_output(
msgs,
structured_model=MathSolution,
)
print(response.content)
if __name__ == "__main__":
asyncio.run(example_simple_call())
asyncio.run(example_tool_call())
asyncio.run(example_structured_output())
+101
View File
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-
"""Example of Grok (xAI) model calls with XAIMultiAgentFormatter.
The multi-agent formatter wraps prior conversation history in
<history></history> tags within a user protobuf message, enabling the xAI
Grok model to handle multi-agent conversations where more than one non-user
agent is involved.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import XAIMultiAgentFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import XAIChatModel
from agentscope.credential import XAICredential
async def example_multiagent() -> None:
"""Simulate a multi-agent conversation and let grok-4.3 summarize it.
Alice and Bob discuss the weather, then a moderator (the model) is asked
to summarize the conversation.
"""
formatter = XAIMultiAgentFormatter()
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-4.3",
stream=True,
context_size=1_000_000,
formatter=formatter,
)
# Multi-agent conversation history between Alice and Bob
msgs = [
Msg(
name="system",
content=[
TextBlock(
text="You are a helpful moderator. Summarize the "
"conversation.",
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hi Bob! What do you think about the weather today?",
),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="It's quite sunny and warm, Alice. Perfect for a "
"walk!",
),
],
role="assistant",
),
Msg(
name="alice",
content=[
TextBlock(text="Agreed! I might head to the park later."),
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(
text="Great idea. I'll join you if I finish work early.",
),
],
role="assistant",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the conversation above in one "
"sentence.",
),
],
role="user",
),
]
print("=== Multi-Agent Formatter Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent())
@@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
"""Example of Grok (xAI) model calls with XAIMultiAgentFormatter and image
input.
Note: Vision (image) input requires grok-4.3 or above.
"""
import asyncio
import os
from _utils import stream_and_collect
from agentscope.formatter import XAIMultiAgentFormatter
from agentscope.message import Msg, TextBlock, DataBlock, URLSource
from agentscope.model import XAIChatModel
from agentscope.credential import XAICredential
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_multiagent_image_url() -> None:
"""Multi-agent conversation where Alice shares an image for the group."""
formatter = XAIMultiAgentFormatter()
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-4.3",
stream=True,
context_size=1_000_000,
formatter=formatter,
)
image_block = DataBlock(
source=URLSource(url=TEST_IMAGE_URL, media_type="image/jpeg"),
)
msgs = [
Msg(
name="system",
content=[
TextBlock(
text=(
"You are a helpful moderator in a group chat. "
"Summarize what the image shows and what the "
"participants said."
),
),
],
role="system",
),
Msg(
name="alice",
content=[
TextBlock(
text="Hey everyone, look at this cute photo I took!",
),
image_block,
],
role="user",
),
Msg(
name="bob",
content=[
TextBlock(text="Aww, that's adorable! Where was this taken?"),
],
role="assistant",
),
Msg(
name="alice",
content=[TextBlock(text="At the local park yesterday.")],
role="user",
),
Msg(
name="moderator",
content=[
TextBlock(
text="Please summarize the image content and the "
"conversation in one paragraph.",
),
],
role="user",
),
]
print("=== Multi-Agent + Multimodal Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_multiagent_image_url())
+140
View File
@@ -0,0 +1,140 @@
# -*- coding: utf-8 -*-
"""Example of Grok model multimodal (vision) calls using DataBlock.
Note: Vision (image) input is supported by grok-4.3 and above.
grok-3-mini does NOT support image input.
"""
import asyncio
import base64
import os
from pathlib import Path
from _utils import stream_and_collect
from agentscope.message import (
Msg,
TextBlock,
DataBlock,
URLSource,
Base64Source,
)
from agentscope.model import XAIChatModel
from agentscope.credential import XAICredential
# A publicly accessible test image (a simple cat photo)
TEST_IMAGE_URL = (
"https://help-static-aliyun-doc.aliyuncs.com/file-manage"
"-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
)
async def example_image_url() -> None:
"""Call grok-4.3 with an image URL and ask what is in the image."""
model = XAIChatModel(
credential=XAICredential(
api_key=os.environ["XAI_API_KEY"],
),
model="grok-4.3",
stream=True,
context_size=1_000_000,
)
image_block = DataBlock(
source=URLSource(
url=TEST_IMAGE_URL,
media_type="image/jpeg",
),
)
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What animal is in this image? Describe it briefly.",
),
image_block,
],
role="user",
),
]
print("=== Multimodal Call (Image URL) ===")
await stream_and_collect(await model(msgs))
def _build_model() -> XAIChatModel:
return XAIChatModel(
credential=XAICredential(api_key=os.environ["XAI_API_KEY"]),
model="grok-4.3",
stream=True,
context_size=1_000_000,
)
async def example_image_local_path() -> None:
"""Call grok-4.3 with a local image using a ``file://`` URL.
The XAI formatter automatically reads the file from disk and converts
it to a base64 data URI.
"""
model = _build_model()
abs_path = str(Path(__file__).parent / "test.jpeg")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=URLSource(
url=f"file://{abs_path}",
media_type="image/jpeg",
),
),
],
role="user",
),
]
print("=== Local Path Call (file://) ===")
await stream_and_collect(await model(msgs))
async def example_image_base64() -> None:
"""Call grok-4.3 with a local image using explicit base64 encoding.
Use ``Base64Source`` when you already have the binary data in memory or
want full control over the encoding step.
"""
model = _build_model()
with open(Path(__file__).parent / "test.jpeg", "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
msgs = [
Msg(
name="user",
content=[
TextBlock(
text="What is happening in this image? Describe it "
"briefly.",
),
DataBlock(
source=Base64Source(data=data, media_type="image/jpeg"),
),
],
role="user",
),
]
print("=== Explicit Base64 Call ===")
await stream_and_collect(await model(msgs))
if __name__ == "__main__":
asyncio.run(example_image_url())
asyncio.run(example_image_local_path())
asyncio.run(example_image_base64())