"通义千问 API 完整指南 2026：接入教程与最佳实践"

from openai import OpenAI

# 通过 Crazyrouter 访问通义千问（无需额外配置）
client = OpenAI(
    api_key="your-crazyrouter-api-key",
    base_url="https://crazyrouter.com/v1"
)

# 基础对话
response = client.chat.completions.create(
    model="qwen2.5-vl-72b-instruct",
    messages=[
        {
            "role": "system",
            "content": "你是一个专业的AI助手，擅长中文理解和代码生成。"
        },
        {
            "role": "user",
            "content": "用Python写一个计算斐波那契数列的函数，要求支持记忆化"
        }
    ],
    max_tokens=2048,
    temperature=0.1  # 代码任务用低temperature
)

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

# 流式输出，适合长文生成
stream = client.chat.completions.create(
    model="qwen2.5-vl-72b-instruct",
    messages=[
        {"role": "user", "content": "详细解释一下大型语言模型的工作原理"}
    ],
    stream=True,
    max_tokens=4096
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()  # 换行

import base64

def analyze_image(image_path: str, question: str) -> str:
    """使用 Qwen-VL 分析图像"""
    
    # 读取并编码图像
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
    
    # 获取图像格式
    ext = image_path.split(".")[-1].lower()
    mime_type = f"image/{ext}"
    
    response = client.chat.completions.create(
        model="qwen2.5-vl-72b-instruct",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{mime_type};base64,{image_data}"
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ],
        max_tokens=1024
    )
    
    return response.choices[0].message.content

# 使用示例
result = analyze_image(
    "product_screenshot.png",
    "请描述这张图片中的内容，并指出任何UI问题"
)
print(result)

import OpenAI from 'openai';

const client = new OpenAI({
  apiKey: process.env.CRAZYROUTER_API_KEY,
  baseURL: 'https://crazyrouter.com/v1',
});

// 中文问答
async function askQwen(question) {
  const response = await client.chat.completions.create({
    model: 'qwen2.5-vl-72b-instruct',
    messages: [
      {
        role: 'system',
        content: '你是一个专业助手，请用清晰的中文回答问题。'
      },
      {
        role: 'user',
        content: question
      }
    ],
    max_tokens: 2048,
    temperature: 0.7,
  });
  
  return response.choices[0].message.content;
}

// 使用示例
const answer = await askQwen('如何优化 Node.js 应用的内存使用？');
console.log(answer);

curl https://crazyrouter.com/v1/chat/completions \
  -H "Authorization: Bearer your-crazyrouter-key" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "qwen2.5-vl-72b-instruct",
    "messages": [
      {
        "role": "user",
        "content": "用Python实现一个简单的HTTP服务器"
      }
    ],
    "max_tokens": 2048,
    "temperature": 0.1
  }'

def generate_chinese_content(topic: str, style: str = "professional") -> str:
    """生成高质量中文内容"""
    
    style_prompts = {
        "professional": "请用专业、正式的语气撰写",
        "casual": "请用轻松、口语化的风格撰写",
        "technical": "请用技术性语言，包含具体数据和例子"
    }
    
    response = client.chat.completions.create(
        model="qwen2.5-vl-72b-instruct",
        messages=[
            {
                "role": "system",
                "content": f"{style_prompts.get(style)}，确保内容准确、流畅、符合中文表达习惯。"
            },
            {
                "role": "user",
                "content": f"请为我撰写一篇关于「{topic}」的文章，约800字。"
            }
        ],
        max_tokens=2000
    )
    
    return response.choices[0].message.content

# 使用示例
article = generate_chinese_content("人工智能在医疗领域的应用", "professional")

def translate_and_localize(text: str, target_lang: str) -> str:
    """翻译并本地化内容"""
    
    prompt_map = {
        "zh": f"请将以下英文内容翻译成自然流畅的中文，注意文化适配：\n\n{text}",
        "en": f"请将以下中文内容翻译成自然流畅的英文，注意文化适配：\n\n{text}"
    }
    
    response = client.chat.completions.create(
        model="qwen2.5-vl-72b-instruct",
        messages=[
            {"role": "user", "content": prompt_map.get(target_lang, prompt_map["en"])}
        ],
        temperature=0.3
    )
    
    return response.choices[0].message.content

def document_qa(document: str, question: str) -> str:
    """基于文档的智能问答"""
    
    response = client.chat.completions.create(
        model="qwen2.5-vl-72b-instruct",
        messages=[
            {
                "role": "system",
                "content": "你是一个专业的文档分析助手。请根据提供的文档内容回答用户问题，如果文档中没有相关信息，请明确说明。"
            },
            {
                "role": "user",
                "content": f"文档内容：\n{document}\n\n问题：{question}"
            }
        ],
        max_tokens=1024,
        temperature=0.1
    )
    
    return response.choices[0].message.content

# ✅ 好的系统提示
system_prompt = """你是一位专业的助手。请注意：
1. 优先使用简洁准确的中文表达
2. 避免过度翻译英文词汇，适当保留通用技术词汇
3. 回答要结构清晰，重点突出
4. 代码示例用注释说明关键步骤"""

# ❌ 较差的系统提示  
system_prompt = "Be helpful and answer in Chinese."

def chat_with_memory(history: list, new_message: str, max_context_tokens: int = 4096):
    """带历史的对话，自动管理上下文长度"""
    
    # 简单的历史截断策略
    messages = history[-10:]  # 保留最近10轮对话
    messages.append({"role": "user", "content": new_message})
    
    response = client.chat.completions.create(
        model="qwen2.5-vl-72b-instruct",
        messages=messages,
        max_tokens=2048
    )
    
    reply = response.choices[0].message.content
    history.append({"role": "user", "content": new_message})
    history.append({"role": "assistant", "content": reply})
    
    return reply, history

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key="your-crazyrouter-key",
    base_url="https://crazyrouter.com/v1"
)

async def batch_process(texts: list[str]) -> list[str]:
    tasks = [
        async_client.chat.completions.create(
            model="qwen2.5-vl-72b-instruct",
            messages=[{"role": "user", "content": text}],
            max_tokens=512
        )
        for text in texts
    ]
    responses = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in responses]