大模型API使用实战#

1

pip install dashscope

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import dashscope
from dashscope import Generation

# 设置API Key
dashscope.api_key = "your-api-key"

def simple_chat(prompt):
    """基础对话调用"""
    response = Generation.call(
        model='qwen-turbo',  # 可选：qwen-turbo, qwen-plus, qwen-max
        prompt=prompt,
        temperature=0.7,     # 控制输出随机性，0-1之间
        max_tokens=1500,     # 最大输出长度
        top_p=0.9,          # 核采样参数
    )
    
    if response.status_code == 200:
        return response.output.text
    else:
        return f"Error: {response.code} - {response.message}"

# 使用示例
result = simple_chat("请介绍一下机器学习的基本概念")
print(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def stream_chat(prompt):
    """流式输出，实时显示生成内容"""
    responses = Generation.call(
        model='qwen-turbo',
        prompt=prompt,
        stream=True,  # 开启流式输出
        temperature=0.7
    )
    
    full_response = ""
    for response in responses:
        if response.status_code == 200:
            content = response.output.text
            print(content, end='', flush=True)
            full_response += content
        else:
            print(f"Error: {response.code}")
            break
    
    return full_response

# 使用示例
stream_chat("写一个Python快速排序算法")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

def multi_turn_chat():
    """多轮对话示例"""
    messages = []
    
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ['退出', 'quit', 'exit']:
            break
            
        # 添加用户消息
        messages.append({"role": "user", "content": user_input})
        
        # 调用API
        response = Generation.call(
            model='qwen-turbo',
            messages=messages,  # 使用消息格式而不是prompt
            temperature=0.7
        )
        
        if response.status_code == 200:
            ai_response = response.output.message.content
            print(f"AI: {ai_response}")
            
            # 添加AI回复到对话历史
            messages.append({"role": "assistant", "content": ai_response})
        else:
            print(f"Error: {response.code}")

# 运行多轮对话
multi_turn_chat()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

def chat_with_system_prompt(user_message, system_prompt="你是一个专业的AI助手"):
    """使用系统提示定制AI行为"""
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_message}
    ]
    
    response = Generation.call(
        model='qwen-plus',
        messages=messages,
        temperature=0.7
    )
    
    return response.output.message.content

# 示例：编程助手
programming_system = """你是一个专业的Python编程助手。
- 总是提供完整、可运行的代码
- 包含详细的注释说明
- 遵循PEP 8代码规范
- 如果有多种解决方案，提供最佳实践
"""

code_result = chat_with_system_prompt(
    "写一个函数来计算列表中所有偶数的平方和",
    programming_system
)
print(code_result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

def optimized_generation(prompt, task_type="general"):
    """根据任务类型优化参数"""
    
    # 不同任务的参数配置
    configs = {
        "creative": {
            "temperature": 0.9,
            "top_p": 0.95,
            "repetition_penalty": 1.1
        },
        "analytical": {
            "temperature": 0.3,
            "top_p": 0.8,
            "repetition_penalty": 1.0
        },
        "code": {
            "temperature": 0.1,
            "top_p": 0.95,
            "repetition_penalty": 1.0
        },
        "general": {
            "temperature": 0.7,
            "top_p": 0.9,
            "repetition_penalty": 1.05
        }
    }
    
    config = configs.get(task_type, configs["general"])
    
    response = Generation.call(
        model='qwen-plus',
        prompt=prompt,
        **config
    )
    
    return response.output.text

# 使用示例
creative_text = optimized_generation(
    "创作一首关于秋天的诗", 
    "creative"
)

analytical_text = optimized_generation(
    "分析人工智能对就业市场的影响", 
    "analytical"
)

code_text = optimized_generation(
    "实现一个二叉搜索树的插入和查找方法", 
    "code"
)

1
2

# Function Call的工作流程：
# 1. 用户提问 -> 2. 模型分析需要调用什么函数 -> 3. 调用函数获取数据 -> 4. 将结果返回给模型 -> 5. 模型整理后回复用户

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

import json
import requests

# 定义天气查询函数
def get_weather(location, date=None):
    """
    获取指定地点的天气信息
    
    Args:
        location (str): 城市名称
        date (str, optional): 日期，格式YYYY-MM-DD
    
    Returns:
        dict: 天气信息
    """
    # 这里使用模拟数据，实际应该调用真实的天气API
    weather_data = {
        "北京": {"temperature": "15°C", "condition": "晴天", "humidity": "45%"},
        "上海": {"temperature": "20°C", "condition": "多云", "humidity": "60%"},
        "广州": {"temperature": "25°C", "condition": "小雨", "humidity": "80%"},
        "深圳": {"temperature": "26°C", "condition": "晴天", "humidity": "55%"}
    }
    
    result = weather_data.get(location, {
        "temperature": "未知",
        "condition": "无数据", 
        "humidity": "未知"
    })
    
    return {
        "location": location,
        "date": date or "今天",
        "weather": result
    }

# 定义函数描述，让模型知道如何使用这个函数
weather_function_schema = {
    "name": "get_weather",
    "description": "获取指定城市的天气信息",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "城市名称，例如：北京、上海、广州"
            },
            "date": {
                "type": "string",
                "description": "日期，格式YYYY-MM-DD，可选参数"
            }
        },
        "required": ["location"]
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

def handle_function_call(user_message):
    """处理带Function Call的对话"""
    
    # 第一步：向模型发送消息，包含function描述
    messages = [
        {"role": "user", "content": user_message}
    ]
    
    response = Generation.call(
        model='qwen-plus',
        messages=messages,
        tools=[{
            "type": "function",
            "function": weather_function_schema
        }],
        tool_choice="auto"  # 让模型自动决定是否调用函数
    )
    
    assistant_message = response.output.message
    
    # 检查模型是否要调用函数
    if hasattr(assistant_message, 'tool_calls') and 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"模型决定调用函数: {function_name}")
        print(f"函数参数: {function_args}")
        
        # 执行函数调用
        if function_name == "get_weather":
            function_result = get_weather(**function_args)
        else:
            function_result = {"error": "未知函数"}
        
        # 将函数结果返回给模型
        messages.extend([
            assistant_message,  # 模型的函数调用请求
            {
                "role": "tool",
                "content": json.dumps(function_result, ensure_ascii=False),
                "tool_call_id": tool_call.id
            }
        ])
        
        # 让模型基于函数结果生成最终回复
        final_response = Generation.call(
            model='qwen-plus',
            messages=messages
        )
        
        return final_response.output.message.content
    else:
        # 模型没有调用函数，直接返回回复
        return assistant_message.content

# 使用示例
user_queries = [
    "北京今天天气怎么样？",
    "帮我查一下上海的天气",
    "广州和深圳的天气对比一下",
    "明天要去旅行，穿什么衣服合适？"  # 这个查询模型可能不会调用函数
]

for query in user_queries:
    print(f"\n用户: {query}")
    response = handle_function_call(query)
    print(f"AI: {response}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

# 定义多个函数
def get_stock_price(symbol):
    """获取股票价格"""
    # 模拟股票数据
    stock_data = {
        "AAPL": {"price": 175.43, "change": "+2.15%"},
        "GOOGL": {"price": 2847.31, "change": "-0.82%"},
        "TSLA": {"price": 248.50, "change": "+3.27%"}
    }
    return stock_data.get(symbol, {"error": "股票代码不存在"})

def calculate_tip(bill_amount, tip_percentage=15):
    """计算小费"""
    tip = bill_amount * tip_percentage / 100
    total = bill_amount + tip
    return {
        "bill_amount": bill_amount,
        "tip_percentage": tip_percentage,
        "tip_amount": round(tip, 2),
        "total_amount": round(total, 2)
    }

# 函数描述列表
function_schemas = [
    {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "城市名称"}
            },
            "required": ["location"]
        }
    },
    {
        "name": "get_stock_price",
        "description": "获取股票价格信息",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码，如AAPL、GOOGL"}
            },
            "required": ["symbol"]
        }
    },
    {
        "name": "calculate_tip",
        "description": "计算小费和总金额",
        "parameters": {
            "type": "object",
            "properties": {
                "bill_amount": {"type": "number", "description": "账单金额"},
                "tip_percentage": {"type": "number", "description": "小费百分比，默认15%"}
            },
            "required": ["bill_amount"]
        }
    }
]

# 函数映射
function_map = {
    "get_weather": get_weather,
    "get_stock_price": get_stock_price,
    "calculate_tip": calculate_tip
}

def advanced_function_call(user_message):
    """支持多函数的高级Function Call"""
    messages = [{"role": "user", "content": user_message}]
    
    tools = [{"type": "function", "function": schema} for schema in function_schemas]
    
    response = Generation.call(
        model='qwen-plus',
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    
    assistant_message = response.output.message
    
    if hasattr(assistant_message, 'tool_calls') and assistant_message.tool_calls:
        # 处理函数调用
        messages.append(assistant_message)
        
        for tool_call in assistant_message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)
            
            # 执行对应的函数
            if function_name in function_map:
                function_result = function_map[function_name](**function_args)
            else:
                function_result = {"error": f"未知函数: {function_name}"}
            
            # 添加函数结果到消息历史
            messages.append({
                "role": "tool",
                "content": json.dumps(function_result, ensure_ascii=False),
                "tool_call_id": tool_call.id
            })
        
        # 生成最终回复
        final_response = Generation.call(
            model='qwen-plus',
            messages=messages
        )
        
        return final_response.output.message.content
    else:
        return assistant_message.content

# 测试多函数功能
test_queries = [
    "北京天气如何？",
    "AAPL股票价格是多少？",
    "账单是120元，小费按20%计算，总共多少钱？",
    "帮我查一下上海天气，还有TSLA的股价"
]

for query in test_queries:
    print(f"\n用户: {query}")
    response = advanced_function_call(query)
    print(f"AI: {response}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

from dashscope import MultiModalConversation

def analyze_image(image_path, question):
    """分析图像并回答问题"""
    
    # 读取图像文件
    with open(image_path, 'rb') as f:
        image_data = f.read()
    
    # 构建多模态消息
    messages = [
        {
            "role": "user",
            "content": [
                {"text": question},
                {"image": image_data}
            ]
        }
    ]
    
    response = MultiModalConversation.call(
        model='qwen-vl-plus',
        messages=messages
    )
    
    return response.output.choices[0].message.content

# 使用示例
image_analysis_questions = [
    "这张图片中有什么内容？",
    "图片中的人在做什么？",
    "这张图片的主要颜色是什么？",
    "图片中有多少个物体？"
]

# 注意：需要提供实际的图片路径
# result = analyze_image("path/to/your/image.jpg", "描述这张图片的内容")
# print(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

def extract_table_data(image_path):
    """从图像中提取表格数据"""
    
    with open(image_path, 'rb') as f:
        image_data = f.read()
    
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "text": """请分析这张图片中的表格，并将数据提取为JSON格式。
                    要求：
                    1. 识别表格的行和列
                    2. 提取所有文本内容
                    3. 按照表格结构组织数据
                    4. 返回标准的JSON格式"""
                },
                {"image": image_data}
            ]
        }
    ]
    
    response = MultiModalConversation.call(
        model='qwen-vl-plus',
        messages=messages
    )
    
    return response.output.choices[0].message.content

# 使用示例
# table_json = extract_table_data("path/to/table_image.jpg")
# print("提取的表格数据：")
# print(table_json)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

def analyze_chart(image_path, analysis_type="general"):
    """分析图表数据"""
    
    analysis_prompts = {
        "general": "请分析这个图表，说明其类型、主要数据趋势和关键发现。",
        "trend": "分析这个图表中的数据趋势，识别上升、下降或稳定的模式。",
        "comparison": "比较图表中不同类别或时间段的数据差异。",
        "insights": "从这个图表中提取3-5个关键商业洞察。"
    }
    
    with open(image_path, 'rb') as f:
        image_data = f.read()
    
    messages = [
        {
            "role": "user",
            "content": [
                {"text": analysis_prompts.get(analysis_type, analysis_prompts["general"])},
                {"image": image_data}
            ]
        }
    ]
    
    response = MultiModalConversation.call(
        model='qwen-vl-plus',
        messages=messages
    )
    
    return response.output.choices[0].message.content

# 使用示例
# chart_analysis = analyze_chart("path/to/chart.png", "trend")
# print("图表分析结果：")
# print(chart_analysis)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import time
from typing import Optional

def robust_api_call(prompt: str, max_retries: int = 3, retry_delay: float = 1.0) -> Optional[str]:
    """健壮的API调用，包含重试机制"""
    
    for attempt in range(max_retries):
        try:
            response = Generation.call(
                model='qwen-turbo',
                prompt=prompt,
                temperature=0.7
            )
            
            if response.status_code == 200:
                return response.output.text
            else:
                print(f"API错误 (尝试 {attempt + 1}/{max_retries}): {response.code} - {response.message}")
                
        except Exception as e:
            print(f"请求异常 (尝试 {attempt + 1}/{max_retries}): {str(e)}")
            
        if attempt < max_retries - 1:
            time.sleep(retry_delay * (2 ** attempt))  # 指数退避
    
    return None

# 使用示例
result = robust_api_call("写一个Python递归函数来计算阶乘")
if result:
    print("成功获取结果：", result)
else:
    print("API调用失败")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

def cost_optimized_call(prompt: str, priority: str = "balanced"):
    """根据优先级选择合适的模型以优化成本"""
    
    model_configs = {
        "economy": {
            "model": "qwen-turbo",
            "max_tokens": 1000,
            "temperature": 0.7
        },
        "balanced": {
            "model": "qwen-plus", 
            "max_tokens": 2000,
            "temperature": 0.7
        },
        "premium": {
            "model": "qwen-max",
            "max_tokens": 4000,
            "temperature": 0.8
        }
    }
    
    config = model_configs.get(priority, model_configs["balanced"])
    
    response = Generation.call(
        prompt=prompt,
        **config
    )
    
    return response.output.text

# 使用示例
economy_result = cost_optimized_call("简单问题", "economy")
premium_result = cost_optimized_call("复杂分析任务", "premium")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

import hashlib
import json
import os

class APICache:
    """简单的API响应缓存"""
    
    def __init__(self, cache_dir: str = "api_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_key(self, prompt: str, **kwargs) -> str:
        """生成缓存键"""
        cache_data = {"prompt": prompt, **kwargs}
        cache_str = json.dumps(cache_data, sort_keys=True)
        return hashlib.md5(cache_str.encode()).hexdigest()
    
    def get(self, prompt: str, **kwargs) -> Optional[str]:
        """从缓存获取结果"""
        cache_key = self._get_cache_key(prompt, **kwargs)
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.json")
        
        if os.path.exists(cache_path):
            with open(cache_path, 'r', encoding='utf-8') as f:
                cached_data = json.load(f)
                return cached_data.get("response")
        return None
    
    def set(self, prompt: str, response: str, **kwargs):
        """保存结果到缓存"""
        cache_key = self._get_cache_key(prompt, **kwargs)
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.json")
        
        cache_data = {
            "prompt": prompt,
            "response": response,
            "timestamp": time.time(),
            **kwargs
        }
        
        with open(cache_path, 'w', encoding='utf-8') as f:
            json.dump(cache_data, f, ensure_ascii=False, indent=2)

def cached_api_call(prompt: str, use_cache: bool = True) -> str:
    """带缓存的API调用"""
    cache = APICache()
    
    if use_cache:
        cached_result = cache.get(prompt)
        if cached_result:
            print("从缓存返回结果")
            return cached_result
    
    # 调用API
    response = Generation.call(
        model='qwen-turbo',
        prompt=prompt,
        temperature=0.7
    )
    
    if response.status_code == 200:
        result = response.output.text
        if use_cache:
            cache.set(prompt, result)
        return result
    else:
        raise Exception(f"API调用失败: {response.code}")

# 使用示例
result1 = cached_api_call("什么是深度学习？")  # 第一次调用API
result2 = cached_api_call("什么是深度学习？")  # 从缓存返回