Deploy Qwen Local API to Hugging Face

.gitignore

@@ -0,0 +1 @@
+.env

Dockerfile

@@ -0,0 +1,19 @@
+# 使用官方的 Python 3.10 slim 版本作为基础环境
+FROM python:3.10-slim
+
+# 设置工作目录
+WORKDIR /app
+
+# ！！！终极修正：使用 COPY . /app 来正确复制整个项目结构！！！
+# 这会把你的本地的 app 文件夹、requirements.txt 等，全部复制到容器的 /app 目录下
+COPY . /app
+
+# 设置环境变量，告诉 Python 模块的搜索路径
+ENV PYTHONPATH=/app
+
+# 安装所有 Python 依赖，使用国内源加速
+RUN pip install --no-cache-dir -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt
+
+# 容器启动时要执行的命令
+# 这个命令现在非常标准和稳定
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8082"]

README.md

@@ -1,12 +1,8 @@
 ---
-title: GPT3.5
-emoji: 💻
-colorFrom: green
-colorTo: indigo
+title: Qwen Local API
+emoji: 🚀
+colorFrom: blue
+colorTo: green
 sdk: docker
-pinned: false
-license: other
-short_description: GPT3.5  free API
+app_port: 8082
 ---
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app/core/config.py

@@ -0,0 +1,51 @@
+# app/core/config.py (v7.2 最终修正版)
+
+from pydantic_settings import BaseSettings
+# ！！！确保这一行包含了 Optional ！！！
+from typing import Dict, List, Optional
+
+class Settings(BaseSettings):
+    """
+    应用配置 (v7.2 最终修正版)
+    - 修正了因缺少 Optional 导入导致的启动崩溃问题。
+    """
+    # --- 服务监听端口 ---
+    LISTEN_PORT: int = 8082
+
+    # --- 应用元数据 ---
+    APP_NAME: str = "Qwen Multi-Account Local API"
+    APP_VERSION: str = "7.2.0"
+    DESCRIPTION: str = "一个支持根据模型名称动态切换账号并具备密钥认证功能的高性能通义千问本地代理。"
+
+    # --- 认证与安全 ---
+    API_MASTER_KEY: Optional[str] = None
+
+    # --- 模型与账号的映射关系 ---
+    MODEL_TO_ACCOUNT_MAP: Dict[str, int] = {
+        "Qwen3-Max-Preview": 2
+    }
+
+    # --- 全面更新的模型列表 (仅供参考，实际以你的账号支持为准) ---
+    SUPPORTED_MODELS: List[str] = [
+        "qwen-plus", "qwen-turbo", "qwen-max", "qwen-long", "qwen-vl-plus",
+        "Qwen3-Max-Preview",
+    ]
+
+    # --- 国内站账号 1 (默认) ---
+    CN_ACCOUNT_1_COOKIE: str = ""
+    CN_ACCOUNT_1_XSRF_TOKEN: str = ""
+
+    # --- 国内站账号 2 (专属) ---
+    CN_ACCOUNT_2_COOKIE: str = ""
+    CN_ACCOUNT_2_XSRF_TOKEN: str = ""
+
+    # --- 国际站账号 (可选) ---
+    INTL_COOKIE: str = ""
+    INTL_AUTHORIZATION: str = ""
+    INTL_BX_UA: str = ""
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = 'utf-8'
+
+settings = Settings()

app/providers/base.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any, AsyncGenerator, Union
+from fastapi import Request
+
+class BaseProvider(ABC):
+    """所有 Provider 的抽象基类"""
+
+    @abstractmethod
+    async def chat_completion(
+        self,
+        request_data: Dict[str, Any],
+        original_request: Request
+    ) -> Union[Dict[str, Any], AsyncGenerator[str, None]]:
+        """处理聊天补全请求的核心方法"""
+        pass

app/providers/text_provider.py

@@ -0,0 +1,250 @@
+# app/providers/text_provider.py (v11.0 终极版 - 健壮的增量转换)
+
+import httpx
+import json
+import uuid
+import time
+import traceback
+import asyncio
+from typing import Dict, Any, AsyncGenerator, Union, List
+
+from fastapi import Request
+from fastapi.responses import StreamingResponse, JSONResponse
+
+from app.providers.base import BaseProvider
+from app.core.config import settings
+
+import logging
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+
+class TextProvider(BaseProvider):
+    """
+    通义千问全能提供商 (v11.0 终极版)
+    - 采用最终修正的状态化解析器，逻辑清晰，确保将官网的“全量累积流”正确转换为客户端所需的“增量流”。
+    - 彻底解决所有场景下的内容重复问题。
+    - 保留了会话预热和国际站功能。
+    """
+
+    # --------------------------------------------------------------------------
+    # 核心入口
+    # --------------------------------------------------------------------------
+    async def chat_completion(self, request_data: Dict[str, Any], original_request: Request) -> Union[StreamingResponse, JSONResponse]:
+        model_name = request_data.get("model", "qwen-plus")
+        task_type = self._get_task_type(model_name, request_data)
+        try:
+            if task_type in ["image", "video"]:
+                logger.info(f"检测到 '{task_type}' 任务，强制使用国际站(INTL)模式...")
+                return await self._handle_long_polling_task(request_data)
+            else:
+                account_id = settings.MODEL_TO_ACCOUNT_MAP.get(model_name, 1)
+                logger.info(f"检测到模型 '{model_name}'，任务类型 '{task_type}'，将使用国内站账号 {account_id}...")
+                return await self._handle_stream_task(request_data, account_id)
+        except Exception as e:
+            logger.error(f"处理任务时出错: {type(e).__name__}: {e}")
+            traceback.print_exc()
+            return JSONResponse(content={"error": {"message": f"处理任务时出错: {e}", "type": "provider_error"}}, status_code=500)
+
+    # --------------------------------------------------------------------------
+    # 国内站流式任务处理
+    # --------------------------------------------------------------------------
+    async def _handle_stream_task(self, request_data: Dict[str, Any], account_id: int) -> StreamingResponse:
+        headers = self._prepare_cn_headers(account_id)
+        await self._prewarm_session(headers)
+        payload = self._prepare_cn_payload(request_data)
+        model_name_for_client = request_data.get("model", "qwen-plus")
+        url = "https://api.tongyi.com/dialog/conversation"
+        logger.info(f"   [CN-Account-{account_id}] 正在向模型 '{model_name_for_client}' 发送流式请求...")
+        return StreamingResponse(self._stream_generator(url, headers, payload, model_name_for_client), media_type="text/event-stream")
+
+    async def _prewarm_session(self, headers: Dict[str, Any]):
+        try:
+            logger.info("   [Pre-warm] 正在发送会话预热请求...")
+            url = "https://api.tongyi.com/assistant/api/record/list"
+            payload = {
+                "pageNo": 1, "terminal": "web", "pageSize": 10000, "module": "uploadhistory",
+                "fileTypes": ["file", "audio", "video"], "recordSources": ["chat", "zhiwen", "tingwu"],
+                "status": [20, 30, 40, 41], "taskTypes": ["local", "net_source", "doc_read", "paper_read", "book_read"]
+            }
+            async with httpx.AsyncClient() as client:
+                prewarm_headers = headers.copy()
+                prewarm_headers['Accept'] = 'application/json, text/plain, */*'
+                response = await client.post(url, headers=prewarm_headers, json=payload, timeout=10)
+                response.raise_for_status()
+                logger.info("   [Pre-warm] ✅ 会话预热成功！")
+        except Exception as e:
+            logger.warning(f"   [Pre-warm] ⚠️ 会话预热失败: {e}。继续尝试...")
+
+    # --------------------------------------------------------------------------
+    # 终极版高级流式解析器 (v11.0) - 核心修改
+    # --------------------------------------------------------------------------
+    async def _stream_generator(self, url: str, headers: Dict, payload: Dict, model_name: str) -> AsyncGenerator[str, None]:
+        """
+        健壮的状态化流式生成器，将通义千问的“全量累积流”转换为标准的“增量流”。
+        """
+        chat_id = f"chatcmpl-{uuid.uuid4().hex}"
+        is_first_chunk = True
+        full_content_so_far = "" # 关键(1) 🧠: 状态变量在生成器函数的顶层作用域，确保在循环中持久存在
+
+        try:
+            async with httpx.AsyncClient(timeout=60) as client:
+                async with client.stream("POST", url, headers=headers, json=payload) as response:
+                    response.raise_for_status()
+                    
+                    async for line in response.aiter_lines():
+                        if not line.startswith('data:'):
+                            continue
+                        
+                        raw_data_str = line.strip()[len('data:'):]
+                        if not raw_data_str or "[DONE]" in raw_data_str:
+                            continue
+                        
+                        try:
+                            qwen_data = json.loads(raw_data_str)
+                            
+                            # 从所有内容块中筛选出 'text' 类型的内容块
+                            text_blocks = [block for block in qwen_data.get("contents", []) if block.get("contentType") == "text"]
+                            if not text_blocks:
+                                continue
+
+                            # 通常我们只关心最后一个text块，因为它包含了最新的完整内容
+                            latest_text_block = text_blocks[-1]
+                            new_full_content = latest_text_block.get("content", "")
+                            
+                            if new_full_content is None:
+                                continue
+
+                            # 关键(2) 💡: 基于持久化的状态，精确计算增量
+                            delta_content = ""
+                            if new_full_content.startswith(full_content_so_far):
+                                delta_content = new_full_content[len(full_content_so_far):]
+                            else:
+                                logger.warning(f"   [Stream Reset] 流内容不连续，将发送全部新内容。")
+                                delta_content = new_full_content
+
+                            # 如果没有实际的新内容，就跳过
+                            if not delta_content:
+                                continue
+
+                            # 关键(3) ✅: 发送增量前，先发送角色信息（仅一次）
+                            if is_first_chunk:
+                                role_chunk = {
+                                    "id": chat_id, "object": "chat.completion.chunk", "created": int(time.time()), "model": model_name,
+                                    "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}]
+                                }
+                                yield f"data: {json.dumps(role_chunk, ensure_ascii=False)}\n\n"
+                                is_first_chunk = False
+
+                            # 发送真正的增量内容块
+                            openai_chunk = {
+                                "id": chat_id, "object": "chat.completion.chunk", "created": int(time.time()), "model": model_name,
+                                "choices": [{"index": 0, "delta": {"content": delta_content}, "finish_reason": None}]
+                            }
+                            yield f"data: {json.dumps(openai_chunk, ensure_ascii=False)}\n\n"
+                            
+                            # 关键(4) 🔄: 更新状态，为处理下一个 data: 消息做准备
+                            full_content_so_far = new_full_content
+                                
+                        except json.JSONDecodeError:
+                            logger.warning(f"   [Warning] JSON 解析失败: {raw_data_str}")
+                            continue
+            
+            # 流结束，发送终止块
+            final_chunk = {
+                "id": chat_id, "object": "chat.completion.chunk", "created": int(time.time()), "model": model_name,
+                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]
+            }
+            yield f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n"
+        
+        except Exception as e:
+            logger.error(f"   [Error] 流式生成器发生错误: {e}")
+            traceback.print_exc()
+        
+        finally:
+            logger.info("   [Stream] 流式传输结束。")
+            yield "data: [DONE]\n\n"
+
+    # --------------------------------------------------------------------------
+    # 辅助函数 (保持不变)
+    # --------------------------------------------------------------------------
+    def _get_task_type(self, model_name: str, request_data: Dict[str, Any]) -> str:
+        model_name_lower = model_name.lower()
+        if "wanx" in model_name_lower: return "image"
+        if "animate" in model_name_lower: return "video"
+        if "vl" in model_name_lower or "qvq" in model_name_lower: return "vision"
+        return "text"
+
+    def _prepare_cn_headers(self, account_id: int) -> Dict[str, str]:
+        try:
+            cookie = getattr(settings, f"CN_ACCOUNT_{account_id}_COOKIE")
+            xsrf_token = getattr(settings, f"CN_ACCOUNT_{account_id}_XSRF_TOKEN")
+        except AttributeError: raise ValueError(f"国内站账号 {account_id} 的配置不完整。")
+        if not cookie or not xsrf_token: raise ValueError(f"国内站账号 {account_id} 的认证信息为空。")
+        safe_cookie = cookie.encode('utf-8').decode('latin-1')
+        return {'Origin': 'https://www.tongyi.com', 'Referer': 'https://www.tongyi.com/', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36', 'Cookie': safe_cookie, 'x-xsrf-token': xsrf_token, 'x-platform': 'pc_tongyi', 'Accept': 'text/event-stream', 'Content-Type': 'application/json;charset=UTF-8'}
+
+    def _prepare_cn_payload(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
+        messages: List[Dict[str, Any]] = request_data.get("messages", [])
+        if not messages: messages = [{"role": "user", "content": request_data.get("prompt", "你好")}]
+        qwen_contents = []
+        for msg in messages:
+            content = msg.get("content")
+            if isinstance(content, str): qwen_contents.append({"role": msg.get("role"), "content": content, "contentType": "text"})
+        model_in_payload = request_data.get("model", "")
+        return {"action": "next", "contents": qwen_contents, "model": model_in_payload, "parentMsgId": "", "requestId": str(uuid.uuid4()), "sessionId": "", "sessionType": "text_chat", "userAction": "new_top", "feature_config": {"search_enabled": False, "thinking_enabled": False}}
+
+    # --------------------------------------------------------------------------
+    # 国际站相关函数 (保留)
+    # --------------------------------------------------------------------------
+    def _prepare_intl_headers(self) -> Dict[str, str]:
+        if not settings.INTL_AUTHORIZATION or not settings.INTL_COOKIE or not settings.INTL_BX_UA:
+            raise ValueError("国际站(intl)认证信息不完整，请检查.env文件。")
+        safe_cookie = settings.INTL_COOKIE.encode('utf-8').decode('latin-1')
+        return {'Origin': 'https://chat.qwen.ai', 'Referer': 'https://chat.qwen.ai/', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36', 'Authorization': settings.INTL_AUTHORIZATION, 'Cookie': safe_cookie, 'bx-ua': settings.INTL_BX_UA}
+
+    async def _handle_long_polling_task(self, request_data: Dict[str, Any]) -> JSONResponse:
+        headers = self._prepare_intl_headers()
+        headers['Accept'] = 'application/json, text/event-stream'
+        headers['Content-Type'] = 'application/json;charset=UTF-8'
+        completions_url = "https://chat.qwen.ai/api/v2/chat/completions"
+        task_status_url_template = "https://chat.qwen.ai/api/v1/tasks/status/{task_id}"
+        prompt = request_data.get("prompt", "一只猫")
+        model_name = "wanx-v1" if "wanx" in request_data.get("model", "") else "animate-v1"
+        msg_type = "t2i" if model_name == "wanx-v1" else "t2v"
+        payload = {"action": "next", "contents": [{"content": prompt, "contentType": "text", "role": "user"}], "msg_type": msg_type, "mode": "chat", "model": model_name, "parentMsgId": "", "requestId": str(uuid.uuid4())}
+        async with httpx.AsyncClient(timeout=60) as client:
+            logger.info(f"   [INTL] 正在启动 '{model_name}' 任务...")
+            response = await client.post(completions_url, headers=headers, json=payload)
+            response.raise_for_status()
+            task_id = None
+            async for line in response.aiter_lines():
+                if line.startswith('data:'):
+                    try:
+                        data = json.loads(line[len('data:'):])
+                        if data.get("taskIds"):
+                            task_id = data["taskIds"][0]
+                            break
+                    except json.JSONDecodeError: continue
+            if not task_id: raise ValueError(f"{model_name} 任务启动失败。")
+            logger.info(f"   [INTL] 成功获取任务 ID: {task_id}")
+            for i in range(120):
+                await asyncio.sleep(3)
+                status_url = task_status_url_template.format(task_id=task_id)
+                status_response = await client.get(status_url, headers=headers)
+                if status_response.status_code == 200:
+                    data = status_response.json()
+                    if data.get("status") == "succeeded":
+                        logger.info(f"   [INTL] {model_name} 任务成功！")
+                        return self._format_media_response(data, request_data, model_name)
+                    if data.get("status") == "failed":
+                        raise RuntimeError(f"任务失败: {data.get('result', '未知错误')}")
+            raise TimeoutError("任务超时。")
+
+    def _format_media_response(self, task_result: Dict[str, Any], request_data: Dict[str, Any], task_type: str) -> JSONResponse:
+        model_name = request_data.get("model")
+        items = task_result.get("result", {}).get("images" if "wanx" in task_type else "videos", [])
+        urls = [item.get("url") for item in items if item.get("url")]
+        content = "\n".join(f"!image({url})" for url in urls) if "wanx" in task_type else "\n".join(f"视频链接: {url}" for url in urls)
+        response_data = {"id": f"chatcmpl-{uuid.uuid4().hex}", "object": "chat.completion", "created": int(time.time()), "model": model_name, "choices": [{"index": 0, "message": {"role": "assistant", "content": content or "生成完成，但未能获取链接。"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}}
+        return JSONResponse(content=response_data)

docker-compose.yml

@@ -0,0 +1,38 @@
+# qwen-local/docker-compose.yml (终极架构修正版)
+
+services:
+  # 这是我们的“前台总机”服务
+  nginx:
+    image: nginx:latest
+    ports:
+      - "8082:80"
+    volumes:
+      # 关键修正: 直接覆盖主配置文件，而不是作为子配置被包含
+      - ./nginx.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      - qwen-local
+    networks:
+      - shared_network
+
+  # 这是我们的“工人”服务
+  qwen-local:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    restart: unless-stopped
+    # 工人不直接对外暴露端口
+    env_file:
+      - .env
+    environment:
+      - API_MASTER_KEY=1
+      - HTTP_PROXY=http://host.docker.internal:7890
+      - HTTPS_PROXY=http://host.docker.internal:7890
+      - NO_PROXY=localhost,127.0.0.1
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    networks:
+      - shared_network
+
+networks:
+  shared_network:
+    external: true

main.py

@@ -0,0 +1,77 @@
+# main.py (v7.1 密钥认证版)
+
+import traceback
+from typing import Optional
+
+from fastapi import FastAPI, Request, HTTPException, Depends, Header
+
+from app.core.config import settings
+from app.providers.text_provider import TextProvider
+
+app = FastAPI(
+    title=settings.APP_NAME,
+    version=settings.APP_VERSION,
+    description=settings.DESCRIPTION
+)
+
+# 只需一个全能的 Provider
+text_provider = TextProvider()
+
+
+# --- 认证依赖项 ---
+async def verify_api_key(authorization: Optional[str] = Header(None)):
+    """
+    检查 API 密钥的依赖项。
+    如果设置了 API_MASTER_KEY，则请求头中必须包含正确的密钥。
+    """
+    # 如果 .env 或 docker-compose.yml 中没有配置 API_MASTER_KEY，则跳过认证
+    if not settings.API_MASTER_KEY:
+        print("警告：未配置 API_MASTER_KEY，服务将对所有请求开放。")
+        return
+
+    # 如果配置了密钥，但请求头中没有 Authorization，则拒绝
+    if authorization is None:
+        raise HTTPException(
+            status_code=401,
+            detail="Unauthorized: Missing Authorization header.",
+        )
+    
+    # 检查认证方案和密钥是否正确
+    try:
+        scheme, token = authorization.split()
+        if scheme.lower() != "bearer":
+            raise ValueError("Invalid scheme")
+    except ValueError:
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid authentication scheme. Use 'Bearer <your_api_key>'.",
+        )
+    
+    # 核心验证：将传入的 token 与我们的主密钥进行安全比较
+    if token != settings.API_MASTER_KEY:
+        raise HTTPException(
+            status_code=403,
+            detail="Forbidden: Invalid API Key.",
+        )
+    # 认证通过，请求将继续被处理
+
+
+@app.post("/v1/chat/completions", dependencies=[Depends(verify_api_key)])
+async def chat_completions(request: Request):
+    """
+    终极路由：所有请求都交给全能的 TextProvider 处理。
+    在处理前会先通过 verify_api_key 进行认证。
+    """
+    try:
+        request_data = await request.json()
+        print("接收到请求，认证通过，路由到全能的 TextProvider...")
+        return await text_provider.chat_completion(request_data, request)
+    except Exception as e:
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"主路由发生内部服务器错误: {str(e)}")
+
+
+@app.get("/")
+def root():
+    """根路由，提供服务基本信息，无需认证。"""
+    return {"message": f"Welcome to {settings.APP_NAME}", "version": settings.APP_VERSION}

nginx.conf

@@ -0,0 +1,68 @@
+# =================================================================
+# Qwen Local API - 终极粘性会话与性能版 Nginx 配置
+# 核心: 绝对信任后端，零干预，极致吞吐，并采用最健壮的会话保持策略
+# =================================================================
+
+# --- 全局性能设置 ---
+worker_processes auto;
+worker_rlimit_nofile 102400;
+
+# --- 事件模型优化 ---
+events {
+    worker_connections 102400;
+    use epoll;
+    multi_accept on;
+}
+
+# --- HTTP 核心配置 ---
+http {
+    # --- 基础性能优化 ---
+    sendfile on;
+    tcp_nopush on;
+    tcp_nodelay on;
+    keepalive_timeout 15s;
+    client_body_timeout 10s;
+    client_header_timeout 10s;
+    server_tokens off;
+    access_log off;
+
+
+    # --- 上游服务器组 (我们的 AI 工人) ---
+    upstream qwen_backend {
+        # 关键修正 🚀: 使用更健壮的 hash 方法实现“终极粘性会话”
+        # 我们不再依赖可能不稳定的客户端 IP，而是使用 Authorization 请求头进行哈希。
+        # 因为来自同一个客户端的所有请求都包含相同的 API Key，这就像“人脸识别”，
+        # 确保了100%的会话保持，从根本上杜绝流式输出的混乱问题。
+        # `consistent` 关键字确保在工人数量变化时，尽可能少地重新映射会话。
+        hash $http_authorization consistent;
+
+        # 性能策略: 开启与工人的“VIP连接池”，实现极致连接复用
+        keepalive 128;
+
+        # 信任策略: 移除所有健康检查和熔断机制
+        server qwen-local:8082;
+    }
+
+
+    # --- 主服务器配置 (API 网关) ---
+    server {
+        listen 80;
+
+        location / {
+            # 性能策略: 移除所有请求限流
+            proxy_pass http://qwen_backend;
+
+            # --- 流式传输终极优化 ---
+            proxy_buffering off;
+            proxy_cache off;
+
+            # --- 协议与头信息设置 ---
+            proxy_http_version 1.1;
+            proxy_set_header Connection "";
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+            proxy_set_header X-Forwarded-Proto $scheme;
+        }
+    }
+}

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn[standard]
+httpx
+pydantic-settings
+python-dotenv