Qwen Cloud Developer Tutorial

# Install the OpenAI Python SDK
!pip install -q openai

import os

# Option A: Read from environment (recommended)
# Make sure you have run: export DASHSCOPE_API_KEY="sk-your-api-key-here"

# Option B: Set directly (for quick testing only — never commit this!)
# os.environ["DASHSCOPE_API_KEY"] = "sk-your-api-key-here"

# Verify the key is set
api_key = os.getenv("DASHSCOPE_API_KEY")
if api_key:
    print(f"API key configured (starts with {api_key[:6]}...)")
else:
    print("API key not found. Please set DASHSCOPE_API_KEY.")

from openai import OpenAI

# Qwen Cloud uses the OpenAI-compatible endpoint
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1"
)

print("Client initialized successfully!")

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {"role": "user", "content": "Hello! Tell me a fun fact about AI."}
    ]
)

print(completion.choices[0].message.content)

import json

# Inspect the full API response
print(json.dumps(json.loads(completion.model_dump_json()), indent=2))

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "system",
            "content": "You are a helpful assistant. Answer clearly and concisely."
        },
        {
            "role": "user",
            "content": "Summarize the benefits of solar energy in three bullet points."
        }
    ]
)

print(completion.choices[0].message.content)

messages = [
    {"role": "system", "content": "You are a knowledgeable science tutor."},
    {"role": "user", "content": "What is photosynthesis?"},
]

# First turn
response1 = client.chat.completions.create(model="qwen3.6-plus", messages=messages)
assistant_reply = response1.choices[0].message.content
print("Turn 1:", assistant_reply)
print("---")

# Add assistant reply and ask a follow-up
messages.append({"role": "assistant", "content": assistant_reply})
messages.append({"role": "user", "content": "Can you explain the light-dependent reactions in simpler terms?"})

# Second turn
response2 = client.chat.completions.create(model="qwen3.6-plus", messages=messages)
print("Turn 2:", response2.choices[0].message.content)

# Creative mode — high temperature
creative = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[{"role": "user", "content": "Write a three-sentence story about a cat and sunlight."}],
    temperature=0.9,
    top_p=0.95
)
print("Creative:")
print(creative.choices[0].message.content)
print()

# Precise mode — low temperature
precise = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[{"role": "user", "content": "Write a three-sentence story about a cat and sunlight."}],
    temperature=0.1,
    top_p=0.7
)
print("Precise:")
print(precise.choices[0].message.content)

stream = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain quantum computing in simple terms."}
    ],
    stream=True  # Enable streaming
)

print("Streaming response:")
full_response = ""
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        content = chunk.choices[0].delta.content
        full_response += content
        print(content, end="", flush=True)

print("\n\n--- Stream complete ---")

stream = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[{"role": "user", "content": "If 3x + 7 = 22, what is x?"}],
    extra_body={"enable_thinking": True},  # Enable thinking mode
    stream=True
)

thinking_content = ""
answer_content = ""

print("Thinking process:")
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    
    # Phase 1: Reasoning / thinking
    if hasattr(delta, "reasoning_content") and delta.reasoning_content:
        thinking_content += delta.reasoning_content
        print(delta.reasoning_content, end="", flush=True)
    
    # Phase 2: Final answer
    if delta.content:
        if not answer_content:  # Print header on first answer token
            print("\n\nAnswer:")
        answer_content += delta.content
        print(delta.content, end="", flush=True)

print()

# Limit thinking to 500 tokens (faster, less detailed reasoning)
stream = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[{"role": "user", "content": "What is the integral of x^2 from 0 to 5?"}],
    extra_body={
        "enable_thinking": True,
        "thinking_budget": 500  # Max tokens for reasoning
    },
    stream=True
)

for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    if hasattr(delta, "reasoning_content") and delta.reasoning_content:
        print(delta.reasoning_content, end="", flush=True)
    if delta.content:
        print(delta.content, end="", flush=True)

print()

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
                    }
                },
                {
                    "type": "text",
                    "text": "Describe what you see in this image."
                }
            ]
        }
    ]
)

print(completion.choices[0].message.content)

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
                    }
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/tiger.png"
                    }
                },
                {
                    "type": "text",
                    "text": "Compare these two images. What do they depict?"
                }
            ]
        }
    ]
)

print(completion.choices[0].message.content)

import base64

def encode_image(image_path: str) -> str:
    """Convert a local image file to a Base64-encoded string."""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode("utf-8")

# Example usage (uncomment and set your image path):
# base64_image = encode_image("/path/to/your/image.png")
#
# completion = client.chat.completions.create(
#     model="qwen3.6-plus",
#     messages=[
#         {
#             "role": "user",
#             "content": [
#                 {
#                     "type": "image_url",
#                     "image_url": {"url": f"data:image/png;base64,{base64_image}"}
#                 },
#                 {"type": "text", "text": "Describe what you see in this image."}
#             ]
#         }
#     ]
# )
# print(completion.choices[0].message.content)

print("encode_image() helper function is ready to use.")
print("Uncomment the example above and set your image path to try it.")

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "video_url",
                    "video_url": {
                        "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241115/cqqkru/1.mp4"
                    },
                    "fps": 2  # Extract 2 frames per second
                },
                {
                    "type": "text",
                    "text": "Summarize what happens in this video."
                }
            ]
        }
    ]
)

print(completion.choices[0].message.content)

import requests
import time

API_KEY = os.getenv("DASHSCOPE_API_KEY")
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
    "X-DashScope-Async": "enable"  # Enable async mode
}

# Step 1: Submit the image generation task
payload = {
    "model": "wan2.7-image",
    "input": {
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "text": "A serene mountain lake at sunrise, with mist rolling over calm water, "
                                "pine trees reflected in the surface, photorealistic style"
                    }
                ]
            }
        ]
    },
    "parameters": {
        "size": "1024x1024",
        "n": 1
    }
}

response = requests.post(
    "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/image-generation/generation",
    headers=HEADERS,
    json=payload
)

result = response.json()
task_id = result["output"]["task_id"]
print(f"Task submitted! Task ID: {task_id}")
print(f"Status: {result['output']['task_status']}")

# Step 2: Poll the task until it completes
poll_headers = {
    "Authorization": f"Bearer {API_KEY}"
}

while True:
    status_response = requests.get(
        f"https://dashscope-intl.aliyuncs.com/api/v1/tasks/{task_id}",
        headers=poll_headers
    )
    status_result = status_response.json()
    task_status = status_result["output"]["task_status"]
    
    print(f"Status: {task_status}")
    
    if task_status in ["SUCCEEDED", "FAILED"]:
        break
    
    time.sleep(3)  # Wait 3 seconds before polling again

# Step 3: Get the generated image URL
if task_status == "SUCCEEDED":
    image_url = status_result["output"]["results"][0]["url"]
    print(f"\nImage generated successfully!")
    print(f"URL: {image_url}")
else:
    print(f"\nTask failed: {status_result}")

# Display the generated image inline (optional)
from IPython.display import Image, display

if task_status == "SUCCEEDED":
    display(Image(url=image_url, width=512))
else:
    print("No image to display — the task did not succeed.")

import json

# Define the tools the model can call
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather for a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name, e.g. Singapore or New York"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# Simulate the tool (in production, this would call a real weather API)
def get_current_weather(location: str) -> str:
    """Simulated weather lookup."""
    weather_data = {
        "Singapore": "Partly cloudy, 31C, humidity 78%",
        "New York": "Sunny, 22C, humidity 45%",
        "London": "Overcast, 15C, humidity 82%",
    }
    return weather_data.get(location, f"Weather data unavailable for {location}")

print("Tools and simulated function defined.")

messages = [
    {"role": "user", "content": "What is the weather like in Singapore today?"}
]

# Step 1: First model call — the model decides to call a tool
response = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=messages,
    tools=tools
)

assistant_message = response.choices[0].message
print("Step 1 — Model wants to call a tool:")

if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    print(f"   Tool: {function_name}")
    print(f"   Args: {function_args}")
    
    # Step 2: Execute the tool
    tool_result = get_current_weather(**function_args)
    print(f"\nStep 2 — Tool result: {tool_result}")
    
    # Step 3: Send the tool result back to the model
    messages.append(assistant_message.model_dump())  # Add assistant tool call message
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": tool_result
    })
    
    # Step 4: Second model call — the model generates the final answer
    final_response = client.chat.completions.create(
        model="qwen3.6-plus",
        messages=messages,
        tools=tools
    )
    
    print(f"\nStep 3 — Final response:")
    print(final_response.choices[0].message.content)
else:
    print("   Model responded directly (no tool call needed):")
    print(f"   {assistant_message.content}")

messages = [
    {"role": "user", "content": "What is the weather in Singapore and London?"}
]

response = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=messages,
    tools=tools,
    parallel_tool_calls=True  # Allow multiple tool calls in one response
)

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    print(f"Model requested {len(assistant_message.tool_calls)} parallel tool calls:\n")
    
    messages.append(assistant_message.model_dump())
    
    # Execute all tool calls
    for tool_call in assistant_message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = get_current_weather(**args)
        print(f"  Tool: {tool_call.function.name}({args}) -> {result}")
        
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result
        })
    
    # Get final response
    final = client.chat.completions.create(
        model="qwen3.6-plus",
        messages=messages,
        tools=tools
    )
    print(f"\nFinal answer:\n{final.choices[0].message.content}")

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "system",
            "content": "You extract contact information from text. Return a JSON object with keys: name, email, phone, company."
        },
        {
            "role": "user",
            "content": "Hi, I am Sarah Chen from TechCorp. You can reach me at sarah.chen@techcorp.com or call 555-0142."
        }
    ],
    response_format={"type": "json_object"}
)

result = json.loads(completion.choices[0].message.content)
print(json.dumps(result, indent=2))

completion = client.chat.completions.create(
    model="qwen3.6-plus",
    messages=[
        {
            "role": "system",
            "content": "You are a recipe analyzer. Extract the recipe details from the user text."
        },
        {
            "role": "user",
            "content": "To make a classic margherita pizza, you need pizza dough, 200g mozzarella, 150ml tomato sauce, fresh basil leaves, olive oil, and a pinch of salt. Preheat oven to 250C, spread sauce on dough, add cheese, bake for 10-12 minutes, then top with basil."
        }
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "recipe",
            "schema": {
                "type": "object",
                "properties": {
                    "dish_name": {"type": "string"},
                    "ingredients": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "steps": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "cooking_time_minutes": {"type": "integer"},
                    "temperature_celsius": {"type": "integer"}
                },
                "required": ["dish_name", "ingredients", "steps", "cooking_time_minutes", "temperature_celsius"]
            }
        }
    }
)

recipe = json.loads(completion.choices[0].message.content)
print(json.dumps(recipe, indent=2))

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1"
)

async def ask_question(question: str) -> str:
    """Send a single async request."""
    print(f"Sent: {question}")
    response = await async_client.chat.completions.create(
        model="qwen3.6-plus",
        messages=[{"role": "user", "content": question}]
    )
    answer = response.choices[0].message.content
    print(f"Received answer for: {question[:50]}...")
    return answer

async def main():
    questions = [
        "Summarize the benefits of solar energy in three bullet points.",
        "Write a subject line for a product launch email.",
        "Translate 'Welcome to our platform' into Spanish."
    ]
    
    # Send all questions concurrently
    results = await asyncio.gather(*[ask_question(q) for q in questions])
    
    print("\n" + "=" * 60)
    for q, a in zip(questions, results):
        print(f"\nQ: {q}")
        print(f"A: {a}")

# In Jupyter notebooks, use await directly
await main()