File size: 4,159 Bytes
0b9dc2e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
# -*- 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())