# 为 EDA Service 添加 Qwen 模型支持（仅阿里云集群）

## 📋 Context - 为什么需要这个改动

EDA Service 目前支持 OpenAI GPT-4/GPT-4o 和 Claude 3 (Sonnet) 模型用于 Text2SQL 任务。当数据库集群部署在阿里云时，使用阿里云的 Qwen 模型可以获得更好的网络性能和成本优势。

### 核心需求
1. **条件性集成**：仅当请求的数据库集群在阿里云（Alibaba Cloud）上时，自动使用 Qwen 模型
2. **强制策略**：阿里云集群强制使用 Qwen，覆盖用户的模型偏好设置
3. **透明切换**：对现有 API 调用者透明，不需要修改接口
4. **保持一致性**：失败时不降级，保持与现有 GPT/Claude 错误处理一致

### 用户决策
- ✅ **模型优先级**：强制使用 Qwen（阿里云集群自动切换）
- ✅ **API Key 管理**：双层配置（系统级环境变量 + 用户级UserPreference）
- ✅ **默认 Qwen 模型**：qwen-turbo（速度快，成本低）
- ✅ **失败处理**：抛出错误，不自动降级到其他模型

---

## 🏗️ 当前架构概述

### AI 模型调用架构
```
用户请求 → task_proxy() → ai_completion() → 路由到具体 API
                                            ├─ Claude → invoke_claude_3_with_text() → AWS Bedrock
                                            └─ GPT → OpenAI API
```

### 模型选择流程（当前）
1. **默认值**：`Claude3.SONNET`
2. **用户偏好**：`UserPreference.load(app_id).llm_model`
3. **线程本地覆盖**：`X-Override-AI-Model` 请求头

### API Key 配置方式（当前）

**双层配置策略**（Qwen将遵循相同模式）：

| 模型 | 系统级（环境变量） | 用户级（UserPreference.llm_api_key） | 格式 |
|------|-------------------|-----------------------------------|------|
| OpenAI | `OPENAI_API_KEY` | 直接存储API key | 字符串 |
| Claude | `BEDROCK_AWS_ACCESS_KEY_ID`<br>`BEDROCK_AWS_SECRET_ACCESS_KEY` | 存储AWS凭证 | `access_key:secret_key` |
| **Qwen（新增）** | `QWEN_API_KEY` | 存储Qwen API key | 字符串 |

**优先级**：用户级 > 系统级

### 集群信息获取
- `utils/credential.py::get_cluster_info(cluster_id)` 返回集群信息
- 包含字段：`cloud_provider_name`（如 "aws", "gcp", "azure", "alibaba"）
- 通过 `context.load_dataset_context()` 可获取 `cluster_id`

### 关键调用点统计
- **总计**：72 个 AI 调用点
- **主要文件**：
  - `svc/eda_task.py`: 39 个调用点（核心任务处理）
  - `svc/data_explore.py`: 7 个调用点（数据库架构分析）
  - `domain_info/index.py`: 2 个调用点（知识管理）

---

## 🎯 实施计划

### Phase 1: 创建 Qwen 模型集成模块

#### 1.1 新建 `svc/qwen.py`

创建 Qwen API 封装模块，参考 `svc/claude.py` 和 `svc/openai.py` 的设计模式。

**关键功能**：
- Qwen API 调用封装（使用阿里云 DashScope SDK）
- 双层API Key管理（系统级 + 用户级）
- 模型 ID 映射
- 响应解析和错误处理
- Token 计算和日志记录

**核心函数签名**：
```python
def invoke_qwen_with_text(
    prompt: str,
    output_format: str,
    message_list: list = [],
    dry_run: bool = False,
    ai_model: str = "qwen-turbo",
    extra_info: dict = {}
) -> Union[dict, str]:
    """
    调用 Qwen 模型的主函数

    Args:
        prompt: 提示词文本
        output_format: "json" 或其他格式
        message_list: 历史消息列表（支持多轮对话）
        dry_run: 是否干运行（仅检查 token 限制）
        ai_model: Qwen 模型 ID（qwen-turbo/qwen-plus/qwen-max）
        extra_info: 额外信息（module, session_id, query_id 等）

    Returns:
        解析后的 JSON 对象（output_format="json"）或文本字符串
    """
```

**实现要点**：

1. **模型ID映射**：
```python
qwen_model_id = {
    "qwen-turbo": "qwen-turbo",
    "qwen-plus": "qwen-plus",
    "qwen-max": "qwen-max"
}
```

2. **双层API Key管理**（参考 `svc/claude.py` 第50-68行）：
```python
import dashscope
from config import config
from models.user_preference import UserPreference
from utils.local import get_local_metrics

# 系统级全局client（使用环境变量）
dashscope.api_key = config.QWEN_API_KEY

def invoke_qwen_with_text(...):
    # 默认使用系统级API key
    api_key = config.QWEN_API_KEY

    # 检查用户自定义API key（优先级更高）
    metrics = get_local_metrics()
    app_id = metrics.get("app_id", 0)
    preference = UserPreference.load(app_id)

    if preference.llm_api_key:
        # 用户提供了自定义key，优先使用
        logging.info("Using custom Qwen API key for app_id = %s", app_id)
        api_key = preference.llm_api_key

    # 使用选定的API key初始化请求
    dashscope.api_key = api_key
```

3. **API调用实现**（使用DashScope SDK）：
```python
from dashscope import Generation

response = Generation.call(
    model=qwen_model_id[ai_model],
    messages=messages,
    result_format='message',
    stream=False,
    temperature=0,
    top_p=1.0,
    max_tokens=4096
)

# 解析响应
if response.status_code == 200:
    result = response.output
    input_tokens = response.usage.input_tokens
    output_tokens = response.usage.output_tokens
    resp_text = result.choices[0].message.content
else:
    raise AlgorithmException(f"Qwen API error: {response.message}")
```

4. **重试机制**（参考 `svc/claude.py`）：
```python
from tenacity import retry, stop_after_attempt, wait_random_exponential, before_sleep_log

@retry(
    reraise=True,
    wait=wait_random_exponential(min=60, max=120),
    stop=stop_after_attempt(3),
    before_sleep=before_sleep_log(logger, logging.INFO)
)
def _call_qwen_api(...):
    # API调用逻辑
```

5. **JSON格式修复**（复用 `svc/claude.py` 的 `extract_text()` 函数）：
```python
from svc.claude import extract_text

resp_json = extract_text(resp_text)
if output_format == "json":
    try:
        return json.loads(resp_json)
    except json.JSONDecodeError:
        # 尝试修复JSON
        from json_repair import repair_json
        return repair_json(resp_json)
```

6. **详细日志记录**：
```python
logging.info(
    "session_id = %s, query_id = %s, module = %s, start Qwen request, "
    "model = %s, prompt_length = %s",
    session_id, query_id, module, ai_model, len(prompt)
)

logging.info(
    "session_id = %s, query_id = %s, module = %s, end Qwen request, "
    "input_tokens = %s, output_tokens = %s, time_cost = %.2fs",
    session_id, query_id, module, input_tokens, output_tokens, time_cost
)
```

**错误处理**：
- API 认证失败 → `AuthenticationException`
- Token 超限 → `BadRequestException(GenSQLErrorMsg.TOKEN_EXCEED_ERROR)`
- 其他错误 → `AlgorithmException`

---

#### 1.2 更新 `consts/consts.py`

在 Claude3 类定义之后添加 Qwen 模型常量。

**位置**：第 ~186 行（Claude3 类之后）

```python
class Qwen:
    """阿里云通义千问模型"""
    DEFAULT = "qwen-turbo"
    TURBO = "qwen-turbo"
    PLUS = "qwen-plus"
    MAX = "qwen-max"
```

---

#### 1.3 更新 `config.py`

添加 Qwen API 配置（系统级）。

**位置**：第 ~56 行（AWS Bedrock 配置之后）

```python
# Alibaba Cloud Qwen Configuration
self.QWEN_API_KEY = os.getenv("QWEN_API_KEY", "")
```

---

#### 1.4 更新 `requirements.txt`

添加阿里云 DashScope SDK 依赖。

```
dashscope>=1.14.0  # 阿里云通义千问 SDK
```

---

### Phase 2: 实现模型路由逻辑

#### 2.1 修改 `svc/openai.py::ai_completion()`

在模型路由中添加 Qwen 分支。

**位置 1**：第 36-49 行 - 更新 Token 限制配置

```python
MAX_TOKEN = {
    GPT_4: 8000,
    GPT_4_TURBO: 128000,
    GPT_4O: 128000,
    GPT_4O_MINI: 128000,
    Qwen.TURBO: 8000,    # Qwen-turbo: 8K context
    Qwen.PLUS: 32000,    # Qwen-plus: 32K context
    Qwen.MAX: 32000      # Qwen-max: 32K context
}

PROMPT_MAX_TOKEN = {
    GPT_4: 6000,
    GPT_4_TURBO: 126000,
    GPT_4O: 126000,
    GPT_4O_MINI: 126000,
    Qwen.TURBO: 6000,    # 预留 2000 token 用于输出
    Qwen.PLUS: 30000,    # 预留 2000 token 用于输出
    Qwen.MAX: 30000      # 预留 2000 token 用于输出
}
```

**位置 2**：第 214-217 行 - 添加 Qwen 路由逻辑

**修改前**：
```python
def ai_completion(stage, main_task, output_format=None, ...):
    model = ai_model or os.getenv("AI_MODEL", GPT_DEFAULT)
    prompt = build_prompt(stage, main_task, output_format, output_schema)

    # Check for Claude models
    if model in (Claude3.DEFAULT, Claude3.SONNET, Claude3.HAIKU):
        return invoke_claude_3_with_text(...)

    # Check for GPT models
    if model not in (GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI):
        raise BadRequestException("Model not supported: {}".format(model))
```

**修改后**：
```python
def ai_completion(stage, main_task, output_format=None, ...):
    model = ai_model or os.getenv("AI_MODEL", GPT_DEFAULT)
    prompt = build_prompt(stage, main_task, output_format, output_schema)

    # Check for Claude models
    if model in (Claude3.DEFAULT, Claude3.SONNET, Claude3.HAIKU):
        return invoke_claude_3_with_text(...)

    # NEW: Check for Qwen models
    if model in (Qwen.DEFAULT, Qwen.TURBO, Qwen.PLUS, Qwen.MAX):
        from svc.qwen import invoke_qwen_with_text
        return invoke_qwen_with_text(prompt, output_format, dry_run=dry_run,
                                    ai_model=ai_model, extra_info=extra_info)

    # Check for GPT models
    if model not in (GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI):
        raise BadRequestException("Model not supported: {}".format(model))
```

**注意**：需要在文件开头添加 Qwen 常量的导入：
```python
from consts.consts import GPT_DEFAULT, GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI, MODE_COMPLETION, MODE_CHAT, DebugActions, GenSQLErrorMsg, Claude3, Qwen
```

---

#### 2.2 修改 `svc/eda_task.py::task_proxy()`

实现云平台检测和模型自动切换逻辑。

**位置 1**：第 138 行之后（加载用户偏好之后）

**修改前**：
```python
def task_proxy(self, app_id, query, ...):
    ai_model = Claude3.SONNET  # Default

    if app_id:
        preference = UserPreference.load(app_id)
        language = preference.language_code
        ai_model = preference.llm_model  # From user preference
        mode_type = preference.mode_type

    # 验证 Assistant API 支持
    if self.binded_thread_id:
        if ai_model not in (GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI):
            raise Exception("Conversation is not supported for model: %s" % ai_model)
```

**修改后**：
```python
def task_proxy(self, app_id, query, ...):
    ai_model = Claude3.SONNET  # Default

    if app_id:
        preference = UserPreference.load(app_id)
        language = preference.language_code
        ai_model = preference.llm_model  # From user preference
        mode_type = preference.mode_type

    # NEW: 根据云平台自动切换模型
    ai_model = self._determine_model_by_cloud_provider(ai_model)

    # 验证 Assistant API 支持
    if self.binded_thread_id:
        if ai_model not in (GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI):
            raise Exception("Conversation is not supported for model: %s" % ai_model)
```

**位置 2**：第 ~2700 行（EDATask 类末尾，添加新方法）

```python
def _determine_model_by_cloud_provider(self, ai_model):
    """
    根据云平台自动切换模型。
    仅当集群在阿里云时，强制使用 Qwen 模型。

    Args:
        ai_model: 用户偏好或默认的 AI 模型

    Returns:
        最终使用的 AI 模型（可能覆盖为 Qwen）

    实现逻辑:
        1. 从 context 获取 cluster_id
        2. 调用 get_cluster_info(cluster_id) 获取云平台信息
        3. 如果 cloud_provider_name in ["alibaba", "aliyun", "alibabacloud"]
           → 强制返回 Qwen.TURBO
        4. 其他云平台 → 返回原 ai_model

    错误处理:
        - cluster_id 缺失 → 返回原 ai_model（graceful degradation）
        - get_cluster_info() 失败 → 返回原 ai_model
        - 记录所有决策到日志
    """
    try:
        # 获取数据集上下文
        db_context = context.load_dataset_context()
        cluster_id = db_context.get("cluster_id")

        if not cluster_id:
            logging.info(
                "session_id = %s, no cluster_id found, using original model = %s",
                self.session_context_id, ai_model
            )
            return ai_model

        # 获取集群信息
        from utils.credential import get_cluster_info
        cluster_info = get_cluster_info(cluster_id)
        cloud_provider = cluster_info.get("cluster", {}).get("cloud_provider_name", "").lower()

        logging.info(
            "session_id = %s, cluster_id = %s, cloud_provider = %s, original_model = %s",
            self.session_context_id, cluster_id, cloud_provider, ai_model
        )

        # 阿里云集群强制使用 Qwen（支持多种可能的命名）
        if cloud_provider in ["alibaba", "aliyun", "alibabacloud"]:
            final_model = Qwen.TURBO  # 默认使用 qwen-turbo
            logging.info(
                "session_id = %s, Alibaba Cloud detected, overriding model: %s -> %s",
                self.session_context_id, ai_model, final_model
            )
            return final_model

        # 其他云平台保持原模型
        return ai_model

    except Exception as e:
        logging.warning(
            "session_id = %s, failed to determine cloud provider, error = %s, "
            "using original model = %s",
            self.session_context_id, str(e), ai_model
        )
        return ai_model  # 失败时使用原模型
```

**注意**：需要在文件开头添加 Qwen 常量的导入：
```python
from consts.consts import Qwen
```

---

### Phase 3: 日志和监控

#### 3.1 关键日志点

**在 `_determine_model_by_cloud_provider()` 中**（已包含在上述代码中）：
- 记录 cluster_id 和 cloud_provider
- 记录模型切换决策（original → final）
- 记录任何异常和降级情况

**在 `svc/qwen.py::invoke_qwen_with_text()` 中**：
```python
# 请求开始
logging.info(
    "session_id = %s, query_id = %s, module = %s, start Qwen request, "
    "model = %s, prompt_length = %s",
    session_id, query_id, module, ai_model, len(prompt)
)

# 请求结束
logging.info(
    "session_id = %s, query_id = %s, module = %s, end Qwen request, "
    "input_tokens = %s, output_tokens = %s, time_cost = %.2fs",
    session_id, query_id, module, input_tokens, output_tokens, time_cost
)

# 使用自定义API key时的日志
if preference.llm_api_key:
    logging.info(
        "session_id = %s, query_id = %s, module = %s, using custom Qwen API key for app_id = %s",
        session_id, query_id, module, app_id
    )
```

#### 3.2 指标收集

在 `svc/qwen.py` 中添加 Prometheus 指标（参考 `svc/openai.py` 的实现）：
- Qwen 请求次数（按模型分类）
- Qwen 请求延迟直方图
- Qwen 错误率
- Token 消耗统计

---

### Phase 4: 测试和验证

#### 4.1 单元测试

**创建 `tests/test_qwen.py`**：
```python
import unittest
from unittest.mock import patch, MagicMock
from svc.qwen import invoke_qwen_with_text

class TestQwenIntegration(unittest.TestCase):

    @patch('dashscope.Generation.call')
    def test_invoke_qwen_with_text_success(self, mock_call):
        """测试正常的Qwen API调用"""
        # Mock响应
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.output.choices[0].message.content = '{"result": "test"}'
        mock_response.usage.input_tokens = 100
        mock_response.usage.output_tokens = 50
        mock_call.return_value = mock_response

        # 调用函数
        result = invoke_qwen_with_text(
            prompt="test prompt",
            output_format="json",
            ai_model="qwen-turbo"
        )

        # 验证结果
        self.assertEqual(result, {"result": "test"})
        mock_call.assert_called_once()

    @patch('dashscope.Generation.call')
    def test_invoke_qwen_with_text_json_output(self, mock_call):
        """测试JSON格式输出"""
        pass

    def test_invoke_qwen_with_text_dry_run(self):
        """测试干运行模式"""
        result = invoke_qwen_with_text(
            prompt="test",
            output_format="json",
            dry_run=True
        )
        self.assertTrue(result)

    @patch('dashscope.Generation.call')
    def test_invoke_qwen_with_text_api_error(self, mock_call):
        """测试API错误处理"""
        mock_response = MagicMock()
        mock_response.status_code = 400
        mock_response.message = "API Error"
        mock_call.return_value = mock_response

        with self.assertRaises(AlgorithmException):
            invoke_qwen_with_text(
                prompt="test",
                output_format="json"
            )
```

**更新 `tests/test_chat2query_tasks.py`**：
```python
class TestCloudProviderDetection(unittest.TestCase):

    @patch('utils.credential.get_cluster_info')
    @patch('svc.context.load_dataset_context')
    def test_determine_model_by_cloud_provider_alibaba(self, mock_context, mock_cluster_info):
        """测试阿里云集群自动切换到Qwen"""
        # Mock返回阿里云集群
        mock_context.return_value = {"cluster_id": "12345"}
        mock_cluster_info.return_value = {
            "cluster": {"cloud_provider_name": "alibaba"}
        }

        eda_task = EDATask(session_context_id=1)
        result = eda_task._determine_model_by_cloud_provider(Claude3.SONNET)

        self.assertEqual(result, Qwen.TURBO)

    @patch('utils.credential.get_cluster_info')
    @patch('svc.context.load_dataset_context')
    def test_determine_model_by_cloud_provider_aws(self, mock_context, mock_cluster_info):
        """测试AWS集群保持原模型"""
        mock_context.return_value = {"cluster_id": "12345"}
        mock_cluster_info.return_value = {
            "cluster": {"cloud_provider_name": "aws"}
        }

        eda_task = EDATask(session_context_id=1)
        result = eda_task._determine_model_by_cloud_provider(Claude3.SONNET)

        self.assertEqual(result, Claude3.SONNET)

    @patch('svc.context.load_dataset_context')
    def test_determine_model_by_cloud_provider_no_cluster(self, mock_context):
        """测试无cluster_id时使用原模型"""
        mock_context.return_value = {}

        eda_task = EDATask(session_context_id=1)
        result = eda_task._determine_model_by_cloud_provider(GPT_4)

        self.assertEqual(result, GPT_4)
```

#### 4.2 集成测试

手动测试清单：

**场景 1：阿里云集群 + 系统级API Key**
- [ ] 用户偏好 = Claude3.SONNET → 实际使用 Qwen.TURBO
- [ ] 用户偏好 = GPT_4 → 实际使用 Qwen.TURBO
- [ ] 用户偏好 = 未设置（默认）→ 实际使用 Qwen.TURBO
- [ ] 验证日志记录模型切换决策
- [ ] 验证使用系统级 QWEN_API_KEY

**场景 2：阿里云集群 + 用户自定义API Key**
- [ ] UserPreference.llm_api_key = "custom_qwen_key"
- [ ] 验证使用用户自定义key而非系统key
- [ ] 验证日志记录使用自定义key

**场景 3：AWS 集群**
- [ ] 用户偏好 = Claude3.SONNET → 实际使用 Claude3.SONNET
- [ ] 用户偏好 = GPT_4 → 实际使用 GPT_4
- [ ] 验证不切换到 Qwen

**场景 4：GCP/Azure 集群**
- [ ] 用户偏好保持不变
- [ ] 不切换到 Qwen

**场景 5：错误处理**
- [ ] cluster_id 缺失 → 使用原模型
- [ ] get_cluster_info() 失败 → 使用原模型
- [ ] Qwen API 调用失败 → 抛出错误（不降级）
- [ ] QWEN_API_KEY 未配置 → 明确错误提示

**场景 6：完整流程**
- [ ] Chat2Data API 请求（阿里云集群）
- [ ] 数据库 Schema 分析（task_gen_db_schema）
- [ ] 问题推荐（task_suggest_questions）
- [ ] SQL 修正（task_sql_refine）

#### 4.3 验证命令

```bash
# 1. 安装依赖
pip install dashscope>=1.14.0

# 2. 设置环境变量（系统级）
export QWEN_API_KEY="your-qwen-api-key"

# 3. 运行单元测试
pytest tests/test_qwen.py -v
pytest tests/test_chat2query_tasks.py -k "cloud_provider" -v

# 4. 运行集成测试（需要真实的阿里云集群）
# 在测试环境中创建阿里云集群并测试完整流程
python scripts/e2etest.py --cluster-type alibaba
```

---

## 📂 关键文件清单

### 需要创建的新文件

| 文件路径 | 说明 | 行数估计 | 复杂度 |
|---------|------|---------|--------|
| `svc/qwen.py` | Qwen API 集成模块 | ~200 行 | ⭐⭐⭐ 核心 |
| `tests/test_qwen.py` | Qwen 模块单元测试 | ~100 行 | ⭐⭐ |

### 需要修改的现有文件

| 文件路径 | 修改位置 | 说明 | 复杂度 |
|---------|---------|------|--------|
| `consts/consts.py` | 第 ~186 行 | 添加 Qwen 模型常量（4行） | ⭐ 简单 |
| `config.py` | 第 ~57 行 | 添加 QWEN_API_KEY 配置（1行） | ⭐ 简单 |
| `svc/openai.py` | 第 12 行 | 导入 Qwen 常量 | ⭐ 简单 |
| `svc/openai.py` | 第 36-49 行 | 更新 MAX_TOKEN 和 PROMPT_MAX_TOKEN（6行） | ⭐ 简单 |
| `svc/openai.py` | 第 214-217 行 | 添加 Qwen 模型路由（4行） | ⭐⭐ |
| `svc/eda_task.py` | 文件开头 | 导入 Qwen 常量 | ⭐ 简单 |
| `svc/eda_task.py` | 第 138 行后 | 调用云平台检测方法（1行） | ⭐ 简单 |
| `svc/eda_task.py` | 第 ~2700 行 | 新增 `_determine_model_by_cloud_provider()` 方法（~60行） | ⭐⭐⭐ 核心 |
| `requirements.txt` | 末尾 | 添加 dashscope 依赖（1行） | ⭐ 简单 |
| `tests/test_chat2query_tasks.py` | 新增测试用例 | 添加云平台检测测试（~50行） | ⭐⭐ |

**代码修改总量估计**：
- 新增代码：~300 行
- 修改现有代码：~20 行
- 总计：~320 行

---

## 🔄 实施顺序

### Day 1: 基础设施（3-4 小时）
1. ✅ 更新 `consts/consts.py`（Qwen 模型常量）
2. ✅ 更新 `config.py`（QWEN_API_KEY 配置）
3. ✅ 更新 `requirements.txt`（添加 dashscope）
4. ✅ 创建 `svc/qwen.py` 基础结构（函数签名、导入）
5. ✅ 更新 `svc/openai.py` 的 MAX_TOKEN 字典

### Day 2: 核心逻辑实现（4-5 小时）
1. ✅ 完成 `svc/qwen.py` 的 API 调用逻辑
   - 双层API Key管理
   - DashScope SDK 集成
   - 错误处理和重试
   - JSON 格式修复
2. ✅ 修改 `svc/openai.py::ai_completion()` 添加 Qwen 路由
3. ✅ 在 `svc/eda_task.py` 中添加 `_determine_model_by_cloud_provider()` 方法
4. ✅ 修改 `svc/eda_task.py::task_proxy()` 调用云平台检测

### Day 3: 测试验证（4-5 小时）
1. ✅ 编写 `tests/test_qwen.py` 单元测试
2. ✅ 更新 `tests/test_chat2query_tasks.py` 集成测试
3. ✅ 本地测试（mock Qwen API）
4. ✅ 在测试环境验证（真实阿里云集群）
5. ✅ 日志和指标验证

### Day 4: 部署上线（2-3 小时）
1. ✅ 在 staging 环境部署
2. ✅ 设置 `QWEN_API_KEY` 环境变量
3. ✅ 验证日志和指标
4. ✅ 灰度发布到生产环境
5. ✅ 监控阿里云集群请求

---

## ⚠️ 注意事项和风险

### 关键风险及解决方案

1. **Assistant API 兼容性问题**
   - **问题**：Qwen 模型不支持 OpenAI Assistant API（多轮对话）
   - **影响**：如果用户在阿里云集群上使用会话模式（binded_thread_id 存在），会抛出异常
   - **当前验证逻辑**（`svc/eda_task.py` 第 172-174 行）：
     ```python
     if self.binded_thread_id:
         if ai_model not in (GPT_4, GPT_4_TURBO, GPT_4O, GPT_4O_MINI):
             raise Exception("Conversation is not supported for model: %s" % ai_model)
     ```
   - **影响评估**：阿里云集群用户将无法使用Assistant会话功能
   - **解决方案**：保持当前验证逻辑，Qwen 会触发异常，清楚告知用户不支持

2. **API Key 未配置**
   - **问题**：如果 `QWEN_API_KEY` 未设置且用户也没有自定义key，调用会失败
   - **影响**：阿里云集群上的所有请求都会失败
   - **解决方案**：
     ```python
     # 在 svc/qwen.py 开头验证
     if not config.QWEN_API_KEY:
         logging.warning("QWEN_API_KEY not configured in environment")

     # 在调用时检查
     if not api_key:
         raise BadRequestException(
             "Qwen API key not configured. Please set QWEN_API_KEY environment "
             "variable or provide custom key in user preferences."
         )
     ```

3. **云平台识别错误**
   - **问题**：`cloud_provider_name` 字段可能不是 "alibaba"，可能是 "aliyun" 或 "alibabacloud"
   - **影响**：无法正确识别阿里云集群，不会切换到 Qwen
   - **解决方案**：在 `_determine_model_by_cloud_provider()` 中使用多个匹配项
     ```python
     if cloud_provider in ["alibaba", "aliyun", "alibabacloud"]:
         return Qwen.TURBO
     ```

4. **Token 计算差异**
   - **问题**：Qwen 的 token 计算方式可能与 GPT 不同
   - **影响**：可能出现 token 超限或浪费
   - **解决方案**：
     - 参考 Qwen 官方文档设置 MAX_TOKEN
     - 实际测试中调整 PROMPT_MAX_TOKEN 值
     - 考虑实现 Qwen 特定的 token 计算函数

5. **用户自定义API Key格式问题**
   - **问题**：用户可能在 llm_api_key 中混合不同模型的key
   - **影响**：可能导致认证失败
   - **解决方案**：当前方案是直接使用 `preference.llm_api_key` 作为 Qwen key
   - **未来改进**：考虑引入前缀区分，如 `qwen:YOUR_API_KEY`

### 降级策略

如果发现 Qwen 集成有严重问题，可以通过以下方式快速回滚：

1. **环境变量开关**（推荐在实施时同步添加）：
   ```python
   # 在 config.py 中添加
   self.ENABLE_QWEN_AUTO_SWITCH = os.getenv("ENABLE_QWEN_AUTO_SWITCH", "true") == "true"

   # 在 _determine_model_by_cloud_provider() 中检查
   def _determine_model_by_cloud_provider(self, ai_model):
       if not config.ENABLE_QWEN_AUTO_SWITCH:
           return ai_model
       # 正常逻辑
   ```

2. **代码回滚**：
   - 注释掉 `task_proxy()` 中的 `ai_model = self._determine_model_by_cloud_provider(ai_model)` 调用
   - 保留 `svc/qwen.py` 模块（允许手动指定 Qwen，不自动切换）

3. **API Key 移除**：
   - 移除环境变量 `QWEN_API_KEY`
   - 所有阿里云集群请求将在尝试使用 Qwen 时失败，自动降级到原模型（由于 try-except）

---

## 📊 验证清单

### 功能验证
- [ ] 阿里云集群自动使用 Qwen 模型
- [ ] AWS/GCP/Azure 集群保持原有模型
- [ ] 用户偏好在非阿里云集群上生效
- [ ] 系统级 API Key 正常工作
- [ ] 用户自定义 API Key 优先级正确
- [ ] 日志正确记录云平台和模型切换决策
- [ ] 日志正确记录API Key使用情况（系统级 vs 用户级）
- [ ] Qwen API 调用成功返回结果
- [ ] JSON 格式输出正确解析

### 错误处理验证
- [ ] 缺少 cluster_id 时使用原模型
- [ ] get_cluster_info() 失败时使用原模型
- [ ] QWEN_API_KEY 未配置时明确报错
- [ ] Qwen API 调用失败时抛出明确错误
- [ ] Token 超限时正确报错
- [ ] 用户自定义 key 无效时正确报错

### 性能验证
- [ ] Qwen API 响应时间在可接受范围（< 10s）
- [ ] 云平台检测不显著增加延迟（< 100ms）
- [ ] Prometheus 指标正常收集

### 兼容性验证
- [ ] Chat2Data API 正常工作
- [ ] 数据库 Schema 分析正常
- [ ] 问题推荐功能正常
- [ ] SQL 修正功能正常
- [ ] Assistant API 在阿里云集群上正确报错（不支持）
- [ ] 所有现有单元测试通过

---

## 🎯 成功标准

1. ✅ **功能完整**：阿里云集群自动使用 Qwen，其他集群不受影响
2. ✅ **透明集成**：无需修改 API 调用方代码
3. ✅ **双层配置**：支持系统级和用户级 API Key，优先级正确
4. ✅ **可观测性**：日志和指标完整记录 Qwen 调用
5. ✅ **稳定性**：错误处理健壮，不影响其他云平台
6. ✅ **测试覆盖**：单元测试和集成测试通过率 100%
7. ✅ **文档完善**：代码注释清晰，便于后续维护

---

## 🔍 与现有架构的一致性检查

### API Key 管理 ✅
- **OpenAI**: 系统级（OPENAI_API_KEY）+ 用户级（llm_api_key 直接存储）
- **Claude**: 系统级（BEDROCK_AWS_*）+ 用户级（llm_api_key 格式：key:secret）
- **Qwen**: 系统级（QWEN_API_KEY）+ 用户级（llm_api_key 直接存储）
- **一致性**：完全遵循现有双层配置模式

### 错误处理 ✅
- **OpenAI**: 抛出 AlgorithmException / BadRequestException
- **Claude**: 抛出 AlgorithmException / Exception
- **Qwen**: 抛出相同类型的异常，不自动降级
- **一致性**：错误处理方式一致

### 模型路由 ✅
- **位置**：`svc/openai.py::ai_completion()`
- **逻辑**：检查 ai_model 所属类别，路由到对应 API
- **Qwen**: 在 Claude 检查之后，GPT 检查之前添加
- **一致性**：完全遵循现有路由模式

### 日志格式 ✅
- **现有格式**：`session_id = %s, query_id = %s, module = %s, ...`
- **Qwen**: 使用相同格式
- **一致性**：日志格式完全一致

### Token 管理 ✅
- **现有**: MAX_TOKEN 和 PROMPT_MAX_TOKEN 字典
- **Qwen**: 添加到相同字典
- **一致性**：Token 管理方式一致

---

## 📚 参考资料

- **阿里云 DashScope SDK 文档**: https://help.aliyun.com/zh/dashscope/
- **Qwen 模型文档**: https://help.aliyun.com/zh/dashscope/developer-reference/tongyi-qianwen-7b-14b-72b-api-detailes
- **EDA Service 现有架构文档**:
  - `docs/PROJECT_ARCHITECTURE.md` - 项目整体架构
  - `docs/QUICK_START.md` - 快速上手指南
  - `docs/TASK_PROXY_DEEP_DIVE.md` - task_proxy 深度解析
  - `docs/TEXT2SQL.md` - Text2SQL 完整流程

---

## 📝 实施后的文档更新

实施完成后，需要更新以下文档：

1. **`docs/PROJECT_ARCHITECTURE.md`**:
   - 更新 "AI 模型" 部分，添加 Qwen 模型说明
   - 更新模型选择流程，添加云平台检测逻辑

2. **`docs/QUICK_START.md`**:
   - 更新环境变量配置部分，添加 `QWEN_API_KEY`
   - 添加阿里云集群自动切换的说明

3. **`README.md`**（如果存在）:
   - 更新支持的 AI 模型列表
   - 添加阿里云特定配置说明

4. **API 文档**（`docs/swagger.json`）:
   - 更新模型参数的可选值，添加 Qwen 模型
   - 添加云平台相关的说明

---

**最后更新时间**: 2025-02-24
**文档版本**: v1.0
**作者**: Claude Code
**审核状态**: 待用户确认