pip install openai

from openai import OpenAI

client = OpenAI(
    api_key="your-moonshot-api-key",  # Get from platform.moonshot.ai
    base_url="https://api.moonshot.cn/v1"
)

response = client.chat.completions.create(
    model="kimi-k2-thinking",
    messages=[
        {
            "role": "system",
            "content": "You are Kimi, an AI assistant by Moonshot AI."
        },
        {
            "role": "user",
            "content": "Explain what makes K2.5 different from other models."
        }
    ],
    temperature=0.6,
)

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

import base64
from openai import OpenAI

client = OpenAI(
    api_key="your-moonshot-api-key",
    base_url="https://api.moonshot.cn/v1"
)

def encode_image(image_path):
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

base64_image = encode_image("screenshot.png")

response = client.chat.completions.create(
    model="moonshotai/Kimi-K2.5",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Analyze this UI mockup and generate the HTML/CSS to recreate it. Focus on responsive design and modern CSS features."
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{base64_image}"
                    }
                }
            ]
        }
    ],
    max_tokens=4096
)

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

response = client.chat.completions.create(
    model="moonshotai/Kimi-K2.5",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Watch this screen recording and extract the user flow. Then create a step-by-step implementation guide."
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/demo-video.mp4"
                    }
                }
            ]
        }
    ],
    temperature=0.3
)

def research_with_swarm(client, topic):
    """
    K2.5 will automatically spawn multiple agents to:
    1. Search for recent papers
    2. Analyze market trends
    3. Identify key competitors
    4. Synthesize findings
    """
    prompt = f"""
    Research the topic: {topic}

    Break this down into parallel subtasks and have multiple agents work on them simultaneously.
    Then merge all findings into a comprehensive report with citations.
    """

    response = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[
            {
                "role": "system",
                "content": "You are an expert research coordinator. Use agent swarm mode to parallelize tasks."
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.4,
        max_tokens=8192
    )

    return response.choices[0].message.content

# Usage
report = research_with_swarm(
    client,
    "Latest developments in multimodal AI agents for 2026"
)
print(report)

def visual_debugger(client, screenshot_path, bug_description):
    """
    K2.5 analyzes a screenshot, identifies UI issues,
    and generates fixes with pixel-perfect precision.
    """
    base64_image = encode_image(screenshot_path)

    response = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[
            {
                "role": "system",
                "content": """You are a visual debugging expert.
                Analyze screenshots, identify UI/layout issues, and provide fixes.
                Include specific CSS/HTML changes with line numbers."""
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"Bug Report: {bug_description}\n\nAnalyze this screenshot and provide a fix."
                    },
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/png;base64,{base64_image}"}
                    }
                ]
            }
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

# Usage
fix = visual_debugger(
    client,
    "button-alignment-bug.png",
    "The 'Submit' button is misaligned on mobile devices"
)
print(fix)

/* Fix for button alignment issue */
.button-container {
    display: flex;
    justify-content: center;
    align-items: center;
    gap: 1rem;
    padding: 1rem;
}

@media (max-width: 768px) {
    .button-container {
        flex-direction: column;
        width: 100%;
    }
}

def design_to_code(client, design_image_path):
    """Convert UI mockups directly to production code"""
    base64_image = encode_image(design_image_path)

    response = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": """Generate a complete React component for this design.
                        Requirements:
                        - Use Tailwind CSS
                        - Include responsive design
                        - Add accessibility attributes
                        - Include TypeScript types
                        - Write clean, production-ready code"""
                    },
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/png;base64,{base64_image}"}
                    }
                ]
            }
        ],
        temperature=0.3,
        max_tokens=4096
    )

    return response.choices[0].message.content

def analyze_document(client, document_path, extraction_goal):
    """Analyze complex documents (PDFs, spreadsheets, images)"""
    base64_doc = encode_image(document_path)

    response = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[
            {
                "role": "system",
                "content": "You are an expert document analyst. Extract structured data from unstructured inputs."
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"""Extract the following from this document:
                        {extraction_goal}

                        Return results in JSON format."""
                    },
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/png;base64,{base64_doc}"}
                    }
                ]
            }
        ],
        temperature=0.1
    )

    return response.choices[0].message.content

# Usage
financial_data = analyze_document(
    client,
    "financial_report.png",
    "Extract all revenue figures, quarter-over-quarter growth, and key metrics"
)

def automate_workflow(client, workflow_description):
    """Let K2.5 design and execute a complex workflow"""
    response = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[
            {
                "role": "system",
                "content": """You are a workflow automation expert.
                Design efficient workflows, break them into parallel tasks,
                and coordinate agent swarms for execution."""
            },
            {
                "role": "user",
                "content": f"""
                Workflow: {workflow_description}

                Please:
                1. Break this down into optimal subtasks
                2. Identify which can run in parallel
                3. Design the agent coordination strategy
                4. Execute and report results
                """
            }
        ],
        temperature=0.4,
        max_tokens=8192
    )

    return response.choices[0].message.content

# Example: Automating a content pipeline
results = automate_workflow(
    client,
    """
    Create a content publishing pipeline that:
    1. Monitors 50 tech news sources
    2. Summarizes top 10 stories daily
    3. Generates social media posts
    4. Creates newsletter drafts
    5. Schedules posts for optimal times
    """
)

import time
from openai import OpenAI, APITimeoutError, RateLimitError

client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.moonshot.cn/v1",
    timeout=30.0,
    max_retries=2
)

def robust_completion(client, messages, max_retries=3):
    """Robust completion with exponential backoff"""
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="moonshotai/Kimi-K2.5",
                messages=messages,
                temperature=0.6
            )
            return response.choices[0].message.content

        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
                continue
            raise

        except APITimeoutError:
            if attempt < max_retries - 1:
                time.sleep(1)
                continue
            raise

    raise Exception("Max retries exceeded")

def choose_model(task_complexity, token_budget):
    """Select the right model variant based on task"""

    if token_budget < 1000:
        return "kimi-k2-turbo"  # Fast, cheap
    elif task_complexity == "high":
        return "kimi-k2-thinking"  # Best reasoning
    else:
        return "moonshotai/Kimi-K2.5"  # Full capabilities

# Usage
model = choose_model(
    task_complexity="high",
    token_budget=5000
)

response = client.chat.completions.create(
    model=model,
    messages=[...],
    max_tokens=token_budget
)

import hashlib
import json

def cache_key(messages, model):
    """Generate cache key from messages"""
    content = json.dumps(messages, sort_keys=True) + model
    return hashlib.sha256(content.encode()).hexdigest()

# Simple in-memory cache (use Redis in production)
response_cache = {}

def cached_completion(client, messages, model):
    key = cache_key(messages, model)

    if key in response_cache:
        print("✓ Using cached response")
        return response_cache[key]

    response = client.chat.completions.create(
        model=model,
        messages=messages
    )

    result = response.choices[0].message.content
    response_cache[key] = result

    return result

def stream_completion(client, messages):
    """Stream responses for real-time user feedback"""
    stream = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=messages,
        stream=True
    )

    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            full_response += content
            print(content, end="", flush=True)  # Real-time output

    return full_response

def estimate_tokens(text):
    """Rough token estimation (1 token ≈ 0.75 words)"""
    return len(text.split()) * 1.3

def truncate_context(messages, max_tokens=100000):
    """Truncate messages to fit context window"""
    total_tokens = sum(estimate_tokens(m['content']) for m in messages)

    if total_tokens <= max_tokens:
        return messages

    # Keep system prompt and recent messages
    system_msg = [m for m in messages if m['role'] == 'system']
    user_msgs = [m for m in messages if m['role'] != 'system']

    # Trim from oldest
    while user_msgs and sum(estimate_tokens(m['content']) for m in system_msg + user_msgs) > max_tokens:
        user_msgs.pop(0)

    return system_msg + user_msgs

def optimize_image_size(image_path, max_dimension=1024):
    """Resize images to reduce token cost"""
    from PIL import Image

    img = Image.open(image_path)
    img.thumbnail((max_dimension, max_dimension))

    optimized_path = image_path.replace('.', '_optimized.')
    img.save(optimized_path, quality=85)

    return optimized_path

def smart_task_execution(client, tasks):
    """Let K2.5 plan optimal execution strategy"""

    planning_prompt = f"""
    I have these tasks to complete:
    {json.dumps(tasks, indent=2)}

    Plan the optimal execution strategy:
    1. Which tasks can run in parallel?
    2. What's the optimal order?
    3. How should agent swarms be coordinated?

    Return a JSON execution plan.
    """

    plan = client.chat.completions.create(
        model="moonshotai/Kimi-K2.5",
        messages=[{"role": "user", "content": planning_prompt}],
        temperature=0.2
    )

    return json.loads(plan.choices[0].message.content)

def test_visual_analysis():
    """Test vision capabilities"""
    result = visual_debugger(
        client,
        "test_screenshot.png",
        "Test bug report"
    )

    assert "fix" in result.lower()
    assert "css" in result.lower() or "html" in result.lower()
    print("✓ Visual analysis test passed")

def test_agent_swarm():
    """Test parallel execution"""
    start_time = time.time()

    result = research_with_swarm(
        client,
        "Test research topic"
    )

    elapsed = time.time() - start_time
    assert elapsed < 60  # Should complete in under 60 seconds
    assert len(result) > 500  # Substantial output
    print("✓ Agent swarm test passed")

import time

def benchmark_model(client, test_cases):
    """Benchmark latency and quality"""
    results = []

    for test_case in test_cases:
        start = time.time()

        response = client.chat.completions.create(
            model="moonshotai/Kimi-K2.5",
            messages=test_case["messages"]
        )

        elapsed = time.time() - start
        results.append({
            "test": test_case["name"],
            "latency": elapsed,
            "tokens": response.usage.total_tokens,
            "tokens_per_second": response.usage.total_tokens / elapsed
        })

    return results