千问 / 通义千问

简介

通义千问（Qwen）是阿里云推出的大语言模型系列，通过 DashScope API 提供服务，完全兼容 OpenAI 接口规范，是国内金融 AI 项目的主流选择。

安装

pip install openai dashscope

快速开始

OpenAI 兼容接口（推荐）

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxx",  # DashScope API Key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 基础对话
response = client.chat.completions.create(
    model="qwen-turbo",
    messages=[
        {"role": "system", "content": "你是专业的金融分析师"},
        {"role": "user", "content": "分析当前 A 股市场的主要风险因素"}
    ],
    temperature=0.7,
    max_tokens=2048
)
print(response.choices[0].message.content)

流式输出

stream = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "详细解释 RAG 技术原理"}],
    stream=True
)

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

模型选型

模型	上下文	特点	适用场景
qwen-turbo	1M	速度快、成本低	高并发、简单问答
qwen-plus	128K	均衡性能	日常业务场景
qwen-max	32K	最强推理	复杂分析、代码生成
qwen-long	10M	超长文档	合同分析、长报告
qwen-vl-plus	32K	视觉理解	图表分析、OCR

Function Calling（工具调用）

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "获取指定股票的实时价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "股票代码，如 600036"
                    },
                    "market": {
                        "type": "string",
                        "enum": ["SH", "SZ"],
                        "description": "交易所"
                    }
                },
                "required": ["symbol"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "招商银行现在股价多少？"}],
    tools=tools,
    tool_choice="auto"
)

# 处理工具调用
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)
    print(f"调用工具: {func_name}, 参数: {func_args}")
    
    # 执行工具并返回结果
    tool_result = {"price": 38.52, "change": "+1.2%"}
    
    messages = [
        {"role": "user", "content": "招商银行现在股价多少？"},
        message,
        {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(tool_result, ensure_ascii=False)
        }
    ]
    
    final_response = client.chat.completions.create(
        model="qwen-plus",
        messages=messages
    )
    print(final_response.choices[0].message.content)

结构化输出（JSON Mode）

from pydantic import BaseModel
from typing import List

class RiskAnalysis(BaseModel):
    risk_level: str          # 高/中/低
    risk_factors: List[str]  # 风险因素列表
    recommendation: str      # 建议

response = client.chat.completions.create(
    model="qwen-plus",
    messages=[
        {
            "role": "system",
            "content": "你是风险分析师，以 JSON 格式输出分析结果"
        },
        {
            "role": "user",
            "content": "分析小微企业贷款的主要风险"
        }
    ],
    response_format={"type": "json_object"}
)

result = json.loads(response.choices[0].message.content)
analysis = RiskAnalysis(**result)
print(f"风险等级: {analysis.risk_level}")

多轮对话管理

class ConversationManager:
    def __init__(self, system_prompt: str, model: str = "qwen-turbo"):
        self.client = OpenAI(
            api_key="sk-xxx",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )
        self.model = model
        self.messages = [{"role": "system", "content": system_prompt}]
    
    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            temperature=0.7
        )
        
        assistant_msg = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": assistant_msg})
        
        # 控制历史长度，避免超出上下文窗口
        if len(self.messages) > 20:
            self.messages = self.messages[:1] + self.messages[-18:]
        
        return assistant_msg
    
    def clear(self):
        self.messages = self.messages[:1]  # 保留 system prompt

# 使用示例
bot = ConversationManager("你是专业的金融客服助手")
print(bot.chat("我想了解理财产品"))
print(bot.chat("有哪些低风险的选择？"))

异步调用（FastAPI 集成）

from openai import AsyncOpenAI
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
async_client = AsyncOpenAI(
    api_key="sk-xxx",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

@app.post("/chat/stream")
async def chat_stream(query: str):
    async def generate():
        stream = await async_client.chat.completions.create(
            model="qwen-turbo",
            messages=[{"role": "user", "content": query}],
            stream=True
        )
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

费用参考

模型	输入（元/百万 token）	输出（元/百万 token）
qwen-turbo	0.3	0.6
qwen-plus	0.8	2.0
qwen-max	40	120

金融场景最佳实践

• 合规敏感场景使用 qwen-max 保证准确性
• 高并发客服场景使用 qwen-turbo 控制成本
• 长合同/报告分析使用 qwen-long
• 始终在 system prompt 中明确角色和输出格式要求