Qwen AI - Multi-modal Platform

# Using Qwen API with Python
import requests
import json

class QwenClient:
    def __init__(self, api_key, base_url="https://dashscope.aliyuncs.com/api/v1"):
        self.api_key = api_key
        self.base_url = base_url
        
    def chat_completion(self, messages, model="qwen-turbo", temperature=0.7):
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "input": {
                "messages": messages
            },
            "parameters": {
                "temperature": temperature,
                "max_tokens": 2000
            }
        }
        
        response = requests.post(
            f"{self.base_url}/services/aigc/text-generation/generation",
            headers=headers,
            json=payload
        )
        
        if response.status_code == 200:
            return response.json()["output"]["text"]
        else:
            raise Exception(f"API Error: {response.status_code} - {response.text}")

# Example usage
client = QwenClient("your-api-key-here")

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Explain quantum computing in simple terms"}
]

response = client.chat_completion(messages, model="qwen-plus")
print(response)

# Qwen-VL for image understanding
class QwenVLClient:
    def __init__(self, api_key):
        self.api_key = api_key
        
    def analyze_image(self, image_url, question):
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "qwen-vl-plus",
            "input": {
                "messages": [
                    {
                        "role": "user",
                        "content": [
                            {
                                "image": image_url
                            },
                            {
                                "text": question
                            }
                        ]
                    }
                ]
            }
        }
        
        response = requests.post(
            "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation",
            headers=headers,
            json=payload
        )
        
        return response.json()

# Example: Analyze an image
client = QwenVLClient("your-api-key")
result = client.analyze_image(
    "https://example.com/image.jpg",
    "What's in this image and describe the scene?"
)
print(result['output']['choices'][0]['message']['content'])

# Specialized code generation example
class QwenCodeGenerator:
    def __init__(self, api_key):
        self.client = QwenClient(api_key)
        
    def generate_function(self, description, language="python"):
        prompt = f"""
        Write a {language} function based on this description:
        {description}
        
        Requirements:
        - Include proper documentation
        - Add type hints if applicable
        - Include error handling
        - Follow best practices for the language
        - Include example usage
        """
        
        messages = [
            {"role": "system", "content": "You are an expert software engineer."},
            {"role": "user", "content": prompt}
        ]
        
        return self.client.chat_completion(messages, model="qwen-code")
    
    def debug_code(self, code, error_message):
        prompt = f"""
        Debug this code and fix the error:
        
        Code:
        {code}
        
        Error:
        {error_message}
        
        Please:
        1. Identify the issue
        2. Provide the fixed code
        3. Explain what was wrong
        """
        
        messages = [
            {"role": "system", "content": "You are a debugging expert."},
            {"role": "user", "content": prompt}
        ]
        
        return self.client.chat_completion(messages, model="qwen-code")

# Usage examples
code_gen = QwenCodeGenerator("your-api-key")

# Generate a function
function_code = code_gen.generate_function(
    "A function to validate email addresses with regex",
    "python"
)
print(function_code)

# Debug existing code
debug_result = code_gen.debug_code(
    """
    def calculate_average(numbers):
        return sum(numbers) / len(numbers)
    """,
    "ZeroDivisionError: division by zero"
)
print(debug_result)

# Qwen-Math for complex mathematical problems
class MathSolver:
    def __init__(self, api_key):
        self.client = QwenClient(api_key)
        
    def solve_math_problem(self, problem, show_steps=True):
        prompt = f"""
        Solve this mathematical problem:
        {problem}
        
        {"Please show your step-by-step reasoning." if show_steps else "Provide the final answer."}
        """
        
        messages = [
            {"role": "system", "content": "You are a mathematics expert. Provide accurate solutions with clear reasoning."},
            {"role": "user", "content": prompt}
        ]
        
        return self.client.chat_completion(messages, model="qwen-math")
    
    def explain_concept(self, concept, level="beginner"):
        prompt = f"""
        Explain the mathematical concept: {concept}
        
        Target audience: {level}
        Include examples and practical applications.
        """
        
        messages = [
            {"role": "system", "content": "You are a mathematics educator."},
            {"role": "user", "content": prompt}
        ]
        
        return self.client.chat_completion(messages, model="qwen-math")

# Example usage
solver = MathSolver("your-api-key")

# Solve complex problem
solution = solver.solve_math_problem(
    "Find the integral of x^2 * e^x dx",
    show_steps=True
)
print(solution)

# Explain concept
explanation = solver.explain_concept("Fourier Transform", "undergraduate")
print(explanation)

# Running Qwen models locally
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class LocalQwen:
    def __init__(self, model_name="Qwen/Qwen-7B-Chat"):
        self.tokenizer = AutoTokenizer.from_pretrained(
            model_name, 
            trust_remote_code=True
        )
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto",
            trust_remote_code=True
        ).eval()
        
    def generate(self, prompt, max_length=1000):
        messages = [
            {"role": "user", "content": prompt}
        ]
        
        text = self.tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True
        )
        
        inputs = self.tokenizer(text, return_tensors="pt")
        with torch.no_grad():
            outputs = self.model.generate(
                inputs.input_ids.cuda(),
                max_new_tokens=max_length,
                do_sample=True,
                temperature=0.7
            )
            
        response = self.tokenizer.decode(
            outputs[0][len(inputs.input_ids[0]):],
            skip_special_tokens=True
        )
        
        return response

# Usage
local_qwen = LocalQwen("Qwen/Qwen-7B-Chat")
response = local_qwen.generate("Explain machine learning algorithms")
print(response)

# Dockerfile for Qwen deployment
FROM pytorch/pytorch:2.0.1-cuda11.7-cudnn8-devel

# Install dependencies
RUN pip install transformers accelerate bitsandbytes

# Create app directory
WORKDIR /app

# Download model (or mount volume)
RUN python -c "
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained('Qwen/Qwen-7B-Chat', trust_remote_code=True)
tokenizer = AutoTokenizer.from_pretrained('Qwen/Qwen-7B-Chat', trust_remote_code=True)
"

# Create API server
COPY app.py .

EXPOSE 8000
CMD ["python", "app.py"]

# app.py - FastAPI server
from fastapi import FastAPI
from pydantic import BaseModel
from LocalQwen import LocalQwen

app = FastAPI()
model = LocalQwen()

class ChatRequest(BaseModel):
    message: str
    max_tokens: int = 1000

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    response = model.generate(request.message, request.max_tokens)
    return {"response": response}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

class RobustQwenClient:
    def __init__(self, api_key, max_retries=3):
        self.client = QwenClient(api_key)
        self.max_retries = max_retries
        
    def generate_with_fallback(self, messages, primary_model, fallback_models=None):
        if fallback_models is None:
            fallback_models = ["qwen-turbo", "qwen-plus"]
            
        models_to_try = [primary_model] + fallback_models
        
        for model in models_to_try:
            for attempt in range(self.max_retries):
                try:
                    response = self.client.chat_completion(messages, model=model)
                    return response, model
                except Exception as e:
                    if "rate limit" in str(e).lower() and attempt < self.max_retries - 1:
                        # Exponential backoff for rate limits
                        wait_time = (2 ** attempt) * 1000
                        time.sleep(wait_time / 1000)
                        continue
                    elif "model not available" in str(e).lower():
                        # Try next model
                        break
                    else:
                        # For other errors, retry immediately
                        if attempt < self.max_retries - 1:
                            continue
                        raise e
                        
        raise Exception("All models failed after retries")
        
    def batch_process(self, prompts, batch_size=5):
        results = []
        for i in range(0, len(prompts), batch_size):
            batch = prompts[i:i + batch_size]
            batch_results = []
            
            for prompt in batch:
                try:
                    response, model_used = self.generate_with_fallback(
                        [{"role": "user", "content": prompt}],
                        "qwen-plus"
                    )
                    batch_results.append({
                        "prompt": prompt,
                        "response": response,
                        "model": model_used,
                        "success": True
                    })
                except Exception as e:
                    batch_results.append({
                        "prompt": prompt,
                        "response": None,
                        "error": str(e),
                        "success": False
                    })
                    
            results.extend(batch_results)
            # Small delay between batches
            time.sleep(0.1)
            
        return results