Anthropic兼容-Messages

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

stream = client.messages.create(
    model="qwen3.8-max",
    max_tokens=2048,
    stream=True,
    thinking={
        "type": "enabled",
        "budget_tokens": 1024
    },
    messages=[
        {
            "role": "user",
            "content": "分析一下量子计算的发展前景。"
        }
    ]
)

for chunk in stream:
    if chunk.type == "content_block_delta":
        if hasattr(chunk.delta, 'thinking'):
            print(chunk.delta.thinking, end="", flush=True)
        elif hasattr(chunk.delta, 'text'):
            print(chunk.delta.text, end="", flush=True)

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

stream = client.messages.create(
    model="qwen3.8-max",
    max_tokens=1024,
    stream=True,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "url",
                        "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250414/mqqmiy/animal_01.jpg",
                    },
                },
                {
                    "type": "text",
                    "text": "描述这张图片的内容。"
                },
            ],
        }
    ],
    thinking={"type": "disabled"},
)

for chunk in stream:
    if chunk.type == "content_block_delta":
        if hasattr(chunk.delta, 'text'):
            print(chunk.delta.text, end="", flush=True)

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

stream = client.messages.create(
    model="qwen3.8-max",
    max_tokens=1024,
    stream=True,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "video",
                    "source": {
                        "type": "url",
                        "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251208/zpupby/3e81ef38-98f0-4d55-bbb6-259334ca18d0.mp4",
                    },
                },
                {
                    "type": "text",
                    "text": "描述这段视频的内容。"
                },
            ],
        }
    ],
    thinking={"type": "disabled"},
)

for chunk in stream:
    if chunk.type == "content_block_delta":
        if hasattr(chunk.delta, 'text'):
            print(chunk.delta.text, end="", flush=True)

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                }
            },
            "required": ["city"]
        }
    }
]

message = client.messages.create(
    model="qwen3.8-max",
    max_tokens=1024,
    tools=tools,
    messages=[
        {
            "role": "user",
            "content": "杭州今天天气怎么样？"
        }
    ]
)

print(message.content)

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

# 模拟代码仓库内容，需达到最小可缓存长度（1024 Token）
long_text_content = "<Your Code Here>" * 400

def get_completion(user_input):
    response = client.messages.create(
        # 选择支持显式缓存的模型
        model="qwen3.8-max",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": long_text_content,
                # 在 text 块上添加 cache_control 即标记缓存断点；也可放在 messages 数组的 content 块上
                "cache_control": {"type": "ephemeral"},
            }
        ],
        messages=[
            {"role": "user", "content": user_input},
        ],
    )
    return response

# 第一次请求：创建缓存
first = get_completion("这段代码的内容是什么")
print(f"创建缓存 Token：{first.usage.cache_creation_input_tokens}")
print(f"命中缓存 Token：{first.usage.cache_read_input_tokens}")
print("=" * 20)
# 第二次请求：长内容相同，仅修改提问 → 命中缓存
second = get_completion("这段代码可以怎么优化")
print(f"创建缓存 Token：{second.usage.cache_creation_input_tokens}")
print(f"命中缓存 Token：{second.usage.cache_read_input_tokens}")

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/apps/anthropic",
)

message = client.messages.create(
    model="deepseek-v4-pro",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "提取以下邮件的关键信息：张三 (zhangsan@example.com) 对企业版方案感兴趣，希望预约下周二下午 2 点的演示。"
        }
    ],
    extra_body={
        "output_config": {
            "format": {
                "type": "json_schema",
                "schema": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "email": {"type": "string"},
                        "plan_interest": {"type": "string"},
                        "demo_requested": {"type": "boolean"}
                    },
                    "required": ["name", "email", "plan_interest", "demo_requested"],
                    "additionalProperties": False
                }
            }
        }
    },
)

# deepseek-v4-pro 模型会返回 thinking 块，需要找到 type='text' 的内容块
text_block = next(block for block in message.content if block.type == "text")
print(text_block.text)