feat: ProxyNode 代理节点管理系统与 OpenAI Responses API 解析增强

ProxyNode 系统：新增 aether-proxy（Rust）海外 VPS 代理组件，后端实现节点注册/心跳/ HMAC 认证/健康检测调度器/模块化集成，前端新增代理节点管理页面。ProxyConfig 支持 node_id 模式，http_client 支持 HMAC 签名代理 URL 构建与 TTL 缓存。 OpenAI CLI 解析器：适配 Responses API 格式，支持 input_tokens/output_tokens 提取、 output[].content[].text 文本解析、response.completed 流式事件 usage 嵌套结构。
2026-09-02 09:20:22 +08:00 · 2026-02-07 12:24:42 +08:00
parent 62f852b851
commit 1180634269
40 changed files with 4761 additions and 11 deletions
--- a/src/api/admin/init.py
+++ b/src/api/admin/init.py
@@ -44,5 +44,6 @@ router.include_router(video_tasks_router)
 # 注意：以下路由已迁移到模块系统，由 ModuleRegistry 动态注册
 # - ldap_router: 当 LDAP_AVAILABLE=true 时注册
 # - management_tokens_router: 当 MANAGEMENT_TOKENS_AVAILABLE=true 时注册
+# - proxy_nodes_router: 当 PROXY_NODES_AVAILABLE=true 时注册

 __all__ = ["router"]
--- a/src/api/admin/proxy_nodes/init.py
+++ b/src/api/admin/proxy_nodes/init.py
@@ -0,0 +1,5 @@
+"""Proxy node admin routes export."""
+
+from .routes import router
+
+__all__ = ["router"]
--- a/src/api/admin/proxy_nodes/routes.py
+++ b/src/api/admin/proxy_nodes/routes.py
@@ -0,0 +1,304 @@
+"""管理员代理节点（ProxyNode）管理端点
+
+用于 aether-proxy 在 VPS 上注册、心跳、注销节点，以及管理员查看/删除节点记录。
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query, Request
+from pydantic import BaseModel, Field, ValidationError, field_validator
+from sqlalchemy.orm import Session
+
+from src.api.base.admin_adapter import AdminApiAdapter
+from src.api.base.context import ApiRequestContext
+from src.api.base.pipeline import ApiRequestPipeline
+from src.core.exceptions import InvalidRequestException, NotFoundException
+from src.database import get_db
+from src.models.database import ProxyNode, ProxyNodeStatus
+
+router = APIRouter(prefix="/api/admin/proxy-nodes", tags=["Admin - Proxy Nodes"])
+pipeline = ApiRequestPipeline()
+
+
+def _node_to_dict(node: ProxyNode) -> dict[str, Any]:
+    return {
+        "id": node.id,
+        "name": node.name,
+        "ip": node.ip,
+        "port": node.port,
+        "region": node.region,
+        "status": node.status.value if node.status else None,
+        "registered_by": node.registered_by,
+        "last_heartbeat_at": node.last_heartbeat_at,
+        "heartbeat_interval": node.heartbeat_interval,
+        "active_connections": node.active_connections,
+        "total_requests": node.total_requests,
+        "avg_latency_ms": node.avg_latency_ms,
+        "created_at": node.created_at,
+        "updated_at": node.updated_at,
+    }
+
+
+class ProxyNodeRegisterRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=100, description="节点名")
+    ip: str = Field(..., description="公网 IP（IPv4/IPv6）")
+    port: int = Field(..., ge=1, le=65535, description="代理端口")
+    region: str | None = Field(None, max_length=100, description="区域标签")
+    heartbeat_interval: int = Field(30, ge=5, le=600, description="心跳间隔（秒）")
+
+    # 指标（可选）
+    active_connections: int | None = Field(None, ge=0, description="当前活跃连接数")
+    total_requests: int | None = Field(None, ge=0, description="累计请求数")
+    avg_latency_ms: float | None = Field(None, ge=0, description="平均延迟（毫秒）")
+
+    @field_validator("ip")
+    @classmethod
+    def validate_ip(cls, v: str) -> str:
+        v = v.strip()
+        try:
+            ipaddress.ip_address(v)
+        except ValueError as exc:
+            raise ValueError("ip 必须是合法的 IPv4/IPv6 地址") from exc
+        return v
+
+
+class ProxyNodeHeartbeatRequest(BaseModel):
+    node_id: str = Field(..., min_length=1, max_length=36, description="节点 ID")
+    heartbeat_interval: int | None = Field(None, ge=5, le=600, description="心跳间隔（秒）")
+
+    active_connections: int | None = Field(None, ge=0, description="当前活跃连接数")
+    total_requests: int | None = Field(None, ge=0, description="累计请求数")
+    avg_latency_ms: float | None = Field(None, ge=0, description="平均延迟（毫秒）")
+
+
+class ProxyNodeUnregisterRequest(BaseModel):
+    node_id: str = Field(..., min_length=1, max_length=36, description="节点 ID")
+
+
+@router.post("/register")
+async def register_proxy_node(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminRegisterProxyNodeAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.post("/heartbeat")
+async def heartbeat_proxy_node(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminHeartbeatProxyNodeAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.post("/unregister")
+async def unregister_proxy_node(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminUnregisterProxyNodeAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.get("")
+async def list_proxy_nodes(
+    request: Request,
+    status: str | None = Query(None, description="按状态筛选：online/unhealthy/offline"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    db: Session = Depends(get_db),
+) -> Any:
+    adapter = AdminListProxyNodesAdapter(status=status, skip=skip, limit=limit)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.delete("/{node_id}")
+async def delete_proxy_node(node_id: str, request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminDeleteProxyNodeAdapter(node_id=node_id)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+def _format_validation_error(exc: ValidationError) -> str:
+    parts: list[str] = []
+    for err in exc.errors():
+        field = " -> ".join(str(x) for x in err.get("loc", []))
+        msg = str(err.get("msg", "invalid"))
+        parts.append(f"{field}: {msg}")
+    return "; ".join(parts) or "输入验证失败"
+
+
+@dataclass
+class AdminRegisterProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_register_proxy_node"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        payload = context.ensure_json_body()
+        try:
+            req = ProxyNodeRegisterRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        now = datetime.now(timezone.utc)
+
+        node = (
+            context.db.query(ProxyNode)
+            .filter(ProxyNode.ip == req.ip, ProxyNode.port == req.port)
+            .first()
+        )
+        if node:
+            node.name = req.name
+            node.region = req.region
+            node.status = ProxyNodeStatus.ONLINE
+            node.last_heartbeat_at = now
+            node.heartbeat_interval = req.heartbeat_interval
+            if req.active_connections is not None:
+                node.active_connections = req.active_connections
+            if req.total_requests is not None:
+                node.total_requests = req.total_requests
+            if req.avg_latency_ms is not None:
+                node.avg_latency_ms = req.avg_latency_ms
+        else:
+            node = ProxyNode(
+                id=str(uuid.uuid4()),
+                name=req.name,
+                ip=req.ip,
+                port=req.port,
+                region=req.region,
+                status=ProxyNodeStatus.ONLINE,
+                registered_by=context.user.id if context.user else None,
+                last_heartbeat_at=now,
+                heartbeat_interval=req.heartbeat_interval,
+                active_connections=req.active_connections or 0,
+                total_requests=req.total_requests or 0,
+                avg_latency_ms=req.avg_latency_ms,
+                created_at=now,
+                updated_at=now,
+            )
+            context.db.add(node)
+
+        context.db.commit()
+        context.db.refresh(node)
+
+        context.add_audit_metadata(
+            action="proxy_node_register",
+            proxy_node_id=node.id,
+            proxy_node_ip=node.ip,
+            proxy_node_port=node.port,
+        )
+
+        return {"node_id": node.id, "node": _node_to_dict(node)}
+
+
+@dataclass
+class AdminHeartbeatProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_heartbeat_proxy_node"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        payload = context.ensure_json_body()
+        try:
+            req = ProxyNodeHeartbeatRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        node = context.db.query(ProxyNode).filter(ProxyNode.id == req.node_id).first()
+        if not node:
+            raise NotFoundException(f"ProxyNode {req.node_id} 不存在", "proxy_node")
+
+        now = datetime.now(timezone.utc)
+        node.status = ProxyNodeStatus.ONLINE
+        node.last_heartbeat_at = now
+        if req.heartbeat_interval is not None:
+            node.heartbeat_interval = req.heartbeat_interval
+        if req.active_connections is not None:
+            node.active_connections = req.active_connections
+        if req.total_requests is not None:
+            node.total_requests = req.total_requests
+        if req.avg_latency_ms is not None:
+            node.avg_latency_ms = req.avg_latency_ms
+
+        context.db.commit()
+        context.db.refresh(node)
+
+        context.add_audit_metadata(
+            action="proxy_node_heartbeat",
+            proxy_node_id=node.id,
+        )
+
+        return {"message": "heartbeat ok", "node": _node_to_dict(node)}
+
+
+@dataclass
+class AdminUnregisterProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_unregister_proxy_node"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        payload = context.ensure_json_body()
+        try:
+            req = ProxyNodeUnregisterRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        node = context.db.query(ProxyNode).filter(ProxyNode.id == req.node_id).first()
+        if not node:
+            raise NotFoundException(f"ProxyNode {req.node_id} 不存在", "proxy_node")
+
+        node.status = ProxyNodeStatus.OFFLINE
+        node.updated_at = datetime.now(timezone.utc)
+        context.db.commit()
+
+        context.add_audit_metadata(
+            action="proxy_node_unregister",
+            proxy_node_id=node.id,
+        )
+
+        return {"message": "unregistered", "node_id": node.id}
+
+
+@dataclass
+class AdminListProxyNodesAdapter(AdminApiAdapter):
+    name: str = "admin_list_proxy_nodes"
+    status: str | None = None
+    skip: int = 0
+    limit: int = 100
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        query = context.db.query(ProxyNode)
+        if self.status:
+            normalized = self.status.strip().lower()
+            allowed = {"online", "unhealthy", "offline"}
+            if normalized not in allowed:
+                raise InvalidRequestException(f"status 必须是以下之一: {sorted(allowed)}", "status")
+            query = query.filter(ProxyNode.status == ProxyNodeStatus(normalized))
+
+        total = query.count()
+        nodes = (
+            query.order_by(ProxyNode.updated_at.desc()).offset(self.skip).limit(self.limit).all()
+        )
+        return {
+            "items": [_node_to_dict(n) for n in nodes],
+            "total": total,
+            "skip": self.skip,
+            "limit": self.limit,
+        }
+
+
+@dataclass
+class AdminDeleteProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_delete_proxy_node"
+    node_id: str = ""
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        node = context.db.query(ProxyNode).filter(ProxyNode.id == self.node_id).first()
+        if not node:
+            raise NotFoundException(f"ProxyNode {self.node_id} 不存在", "proxy_node")
+
+        context.add_audit_metadata(
+            action="proxy_node_delete",
+            proxy_node_id=node.id,
+            proxy_node_ip=node.ip,
+            proxy_node_port=node.port,
+        )
+
+        context.db.delete(node)
+        context.db.commit()
+
+        return {"message": "deleted", "node_id": self.node_id}
--- a/src/api/handlers/base/cli_handler_base.py
+++ b/src/api/handlers/base/cli_handler_base.py
@@ -2182,7 +2182,15 @@ class CliMessageHandlerBase(BaseMessageHandler):
            message = evt.get("message", {})
            if isinstance(message, dict):
                usage = message.get("usage")
-        # OpenAI 格式: 直接在 chunk 中
+        # OpenAI Responses API (openai:cli) 格式: response.completed 中 usage 嵌套在 response 对象内
+        elif event_type == "response.completed":
+            resp_obj = evt.get("response")
+            if isinstance(resp_obj, dict):
+                usage = resp_obj.get("usage")
+            # 兼容: 部分实现可能在顶层也有 usage
+            if not usage:
+                usage = evt.get("usage")
+        # OpenAI Chat 格式: 直接在 chunk 中
        elif "usage" in evt:
            usage = evt.get("usage")
        # Gemini 格式: usageMetadata
--- a/src/api/handlers/base/parsers.py
+++ b/src/api/handlers/base/parsers.py
@@ -241,13 +241,145 @@ class OpenAIResponseParser(ResponseParser):


 class OpenAICliResponseParser(OpenAIResponseParser):
-    """OpenAI CLI 格式响应解析器"""
+    """OpenAI CLI / Responses API 格式响应解析器
+
+    OpenAI Responses API 与 Chat Completions API 的关键差异：
+    - Usage 字段: input_tokens/output_tokens（而非 prompt_tokens/completion_tokens）
+    - 响应结构: output[].content[].text（而非 choices[].message.content）
+    - 流式事件: response.completed 事件中 usage 嵌套在 response 对象内
+    """

    def __init__(self) -> None:
        super().__init__()
        self.name = "openai:cli"
        self.api_format = "openai:cli"

+    def parse_response(self, response: dict[str, Any], status_code: int) -> ParsedResponse:
+        result = ParsedResponse(
+            raw_response=response,
+            status_code=status_code,
+        )
+
+        # Responses API: 文本在 output[].content[].text 中
+        result.text_content = self._extract_responses_api_text(response)
+        result.response_id = response.get("id")
+
+        # Responses API usage: input_tokens / output_tokens
+        usage = self._extract_responses_api_usage(response)
+        result.input_tokens = usage.get("input_tokens", 0)
+        result.output_tokens = usage.get("output_tokens", 0)
+        result.cache_creation_tokens = usage.get("cache_creation_tokens", 0)
+        result.cache_read_tokens = usage.get("cache_read_tokens", 0)
+
+        # 检查错误（支持嵌套错误格式）
+        is_error, error_info = _check_nested_error(response)
+        if is_error and error_info:
+            result.is_error = True
+            result.error_type = error_info.get("type")
+            result.error_message = error_info.get("message")
+            result.embedded_status_code = _extract_embedded_status_code(error_info)
+
+        return result
+
+    def extract_usage_from_response(self, response: dict[str, Any]) -> dict[str, int]:
+        usage = self._extract_responses_api_usage(response)
+        return usage
+
+    def extract_text_content(self, response: dict[str, Any]) -> str:
+        return self._extract_responses_api_text(response)
+
+    @staticmethod
+    def _extract_responses_api_usage(response: dict[str, Any]) -> dict[str, int]:
+        """从 Responses API 响应或流式事件中提取 usage
+
+        支持多种结构：
+        1. 顶层 usage（非流式响应 / 部分转换后的响应）
+        2. response.usage（流式 response.completed 事件）
+        3. 兼容 Chat Completions 字段名（prompt_tokens/completion_tokens）
+        """
+        usage: dict[str, Any] = {}
+
+        # 优先从顶层 usage 提取
+        top_usage = response.get("usage")
+        if isinstance(top_usage, dict):
+            usage = top_usage
+        else:
+            # 流式事件: response.completed 中 usage 嵌套在 response 对象内
+            resp_obj = response.get("response")
+            if isinstance(resp_obj, dict):
+                nested_usage = resp_obj.get("usage")
+                if isinstance(nested_usage, dict):
+                    usage = nested_usage
+
+        if not usage:
+            return {
+                "input_tokens": 0,
+                "output_tokens": 0,
+                "cache_creation_tokens": 0,
+                "cache_read_tokens": 0,
+            }
+
+        # Responses API 使用 input_tokens/output_tokens
+        # 兼容 Chat Completions 的 prompt_tokens/completion_tokens（以防转换后的响应）
+        input_tokens = usage.get("input_tokens") or usage.get("prompt_tokens") or 0
+        output_tokens = usage.get("output_tokens") or usage.get("completion_tokens") or 0
+
+        return {
+            "input_tokens": int(input_tokens),
+            "output_tokens": int(output_tokens),
+            "cache_creation_tokens": int(
+                usage.get("cache_creation_input_tokens") or usage.get("cache_creation_tokens") or 0
+            ),
+            "cache_read_tokens": int(
+                usage.get("cache_read_input_tokens") or usage.get("cache_read_tokens") or 0
+            ),
+        }
+
+    @staticmethod
+    def _extract_responses_api_text(response: dict[str, Any]) -> str:
+        """从 Responses API 响应中提取文本内容
+
+        支持结构: output[].content[].text 或 output[].text
+        """
+        text_parts: list[str] = []
+
+        output = response.get("output")
+        if isinstance(output, list):
+            for item in output:
+                if not isinstance(item, dict):
+                    continue
+                # message 类型: output[].content[].text
+                if item.get("type") == "message":
+                    content = item.get("content")
+                    if isinstance(content, list):
+                        for part in content:
+                            if isinstance(part, dict):
+                                ptype = str(part.get("type") or "")
+                                if ptype in ("output_text", "text") and isinstance(
+                                    part.get("text"), str
+                                ):
+                                    text_parts.append(part["text"])
+                # 直接文本类型: output[].text
+                elif item.get("type") in ("output_text", "text") and isinstance(
+                    item.get("text"), str
+                ):
+                    text_parts.append(item["text"])
+
+        # 兼容: 部分实现可能直接给 output_text
+        if not text_parts and isinstance(response.get("output_text"), str):
+            text_parts.append(response["output_text"])
+
+        # 兼容: 如果是 Chat Completions 格式（可能来自转换后的响应），回退到 choices 结构
+        if not text_parts:
+            choices = response.get("choices", [])
+            if isinstance(choices, list) and choices:
+                message = choices[0].get("message", {}) if isinstance(choices[0], dict) else {}
+                content = message.get("content") if isinstance(message, dict) else None
+                if isinstance(content, str):
+                    text_parts.append(content)
+
+        return "".join(text_parts)
+

 class ClaudeResponseParser(ResponseParser):
    """Claude 格式响应解析器"""
--- a/src/clients/http_client.py
+++ b/src/clients/http_client.py
@@ -12,6 +12,7 @@ from __future__ import annotations

 import asyncio
 import hashlib
+import hmac
 import time
 from contextlib import asynccontextmanager
 from typing import Any
@@ -20,6 +21,7 @@ from urllib.parse import quote, urlparse
 import httpx

 from src.config import config
+from src.core.exceptions import ProxyNodeUnavailableError
 from src.core.logger import logger
 from src.utils.ssl_utils import get_ssl_context

@@ -27,6 +29,70 @@ from src.utils.ssl_utils import get_ssl_context
 _proxy_clients_lock = asyncio.Lock()
 _default_client_lock = asyncio.Lock()

+# ProxyNode 信息缓存（降低高频 DB 查询开销）
+_proxy_node_cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
+_PROXY_NODE_CACHE_TTL_SECONDS = 60.0
+_PROXY_NODE_CACHE_MAX_SIZE = 256
+
+
+def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
+    """
+    读取 ProxyNode 信息（带内存 TTL 缓存）
+
+    Returns:
+        {"ip": str, "port": int} 或 None（不存在/非在线）
+    """
+    now = time.time()
+    cached = _proxy_node_cache.get(node_id)
+    if cached:
+        value, expires_at = cached
+        if now < expires_at:
+            return value
+
+    # 防止无效 node_id 导致缓存无限膨胀
+    if len(_proxy_node_cache) >= _PROXY_NODE_CACHE_MAX_SIZE:
+        _proxy_node_cache.clear()
+
+    from src.database import create_session
+    from src.models.database import ProxyNode, ProxyNodeStatus
+
+    db = create_session()
+    try:
+        node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
+        if not node or node.status != ProxyNodeStatus.ONLINE:
+            _proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_TTL_SECONDS)
+            return None
+
+        value = {"ip": node.ip, "port": node.port}
+        _proxy_node_cache[node_id] = (value, now + _PROXY_NODE_CACHE_TTL_SECONDS)
+        return value
+    finally:
+        db.close()
+
+
+def _build_hmac_proxy_url(ip: str, port: int, node_id: str) -> str:
+    """
+    构建带 HMAC BasicAuth 的 httpx proxy URL
+
+    格式: http://hmac:{timestamp}.{signature}@{ip}:{port}
+    signature = HMAC-SHA256(PROXY_HMAC_KEY, "{timestamp}\\n{node_id}") 的 hex
+    """
+    if not config.proxy_hmac_key:
+        raise ProxyNodeUnavailableError(
+            "PROXY_HMAC_KEY 未配置，无法使用 ProxyNode 代理", node_id=node_id
+        )
+
+    timestamp = str(int(time.time()))
+    payload = f"{timestamp}\n{node_id}".encode("utf-8")
+    signature = hmac.new(
+        config.proxy_hmac_key.encode("utf-8"),
+        payload,
+        hashlib.sha256,
+    ).hexdigest()
+
+    host = f"[{ip}]" if ":" in ip else ip
+    return f"http://hmac:{timestamp}.{signature}@{host}:{int(port)}"
+

 def _compute_proxy_cache_key(proxy_config: dict[str, Any] | None) -> str:
    """
@@ -41,6 +107,16 @@ def _compute_proxy_cache_key(proxy_config: dict[str, Any] | None) -> str:
    if not proxy_config:
        return "__no_proxy__"

+    # enabled=False 时视为无代理（兼容旧数据）
+    if not proxy_config.get("enabled", True):
+        return "__no_proxy__"
+
+    # ProxyNode 模式：基于 node_id + 时间桶缓存，避免签名随时间变化导致 cache key 爆炸
+    node_id = proxy_config.get("node_id")
+    if isinstance(node_id, str) and node_id.strip():
+        time_bucket = int(time.time() / 120)  # 120 秒一个桶
+        return f"proxy_node:{node_id.strip()}:{time_bucket}"
+
    # 构建代理 URL 作为缓存键的基础
    proxy_url = build_proxy_url(proxy_config)
    if not proxy_url:
@@ -55,7 +131,9 @@ def build_proxy_url(proxy_config: dict[str, Any]) -> str | None:
    根据代理配置构建完整的代理 URL

    Args:
-        proxy_config: 代理配置字典，包含 url, username, password, enabled
+        proxy_config: 代理配置字典，支持两种模式：
+            - 手动 URL 模式: {url, username, password, enabled}
+            - ProxyNode 模式: {node_id, enabled}

    Returns:
        完整的代理 URL，如 socks5://user:pass@host:port
@@ -68,6 +146,15 @@ def build_proxy_url(proxy_config: dict[str, Any]) -> str | None:
    if not proxy_config.get("enabled", True):
        return None

+    # ProxyNode 模式（aether-proxy）
+    node_id = proxy_config.get("node_id")
+    if isinstance(node_id, str) and node_id.strip():
+        node_id = node_id.strip()
+        node_info = _get_proxy_node_info(node_id)
+        if not node_info:
+            raise ProxyNodeUnavailableError(f"代理节点 {node_id} 不可用", node_id=node_id)
+        return _build_hmac_proxy_url(node_info["ip"], node_info["port"], node_id)
+
    proxy_url: str | None = proxy_config.get("url")
    if not proxy_url:
        return None
@@ -307,9 +394,13 @@ class HTTPClientPool:
            client = httpx.AsyncClient(**client_config)  # type: ignore[arg-type]
            cls._proxy_clients[cache_key] = (client, time.time())

+            proxy_label = "none"
+            if proxy_config:
+                proxy_label = str(
+                    proxy_config.get("node_id") or proxy_config.get("url") or "unknown"
+                )
            logger.debug(
-                f"创建代理客户端(缓存): {proxy_config.get('url', 'unknown') if proxy_config else 'none'}, "
-                f"缓存数量: {len(cls._proxy_clients)}"
+                f"创建代理客户端(缓存): {proxy_label}, " f"缓存数量: {len(cls._proxy_clients)}"
            )

            return client
--- a/src/config/settings.py
+++ b/src/config/settings.py
@@ -3,6 +3,8 @@
 从环境变量或 .env 文件加载配置
 """

+import hashlib
+import hmac
 import os
 from pathlib import Path
 from typing import Any
@@ -46,6 +48,13 @@ class Config:
        # 加密密钥配置（独立于JWT密钥，用于敏感数据加密）
        self.encryption_key = os.getenv("ENCRYPTION_KEY", None)

+        # 代理节点 HMAC 密钥（用于 aether-proxy 认证）
+        proxy_hmac_key_env = os.getenv("PROXY_HMAC_KEY")
+        if proxy_hmac_key_env and proxy_hmac_key_env.strip():
+            self.proxy_hmac_key = proxy_hmac_key_env.strip()
+        else:
+            self.proxy_hmac_key = self._derive_proxy_hmac_key()
+
        # 环境配置 - 智能检测
        # Docker 部署默认为生产环境，本地开发默认为开发环境
        is_docker = (
@@ -307,6 +316,20 @@ class Config:
        # 验证连接池配置
        self._validate_pool_config()

+    def _derive_proxy_hmac_key(self) -> str:
+        """
+        从 ENCRYPTION_KEY 派生 PROXY_HMAC_KEY
+
+        目的：避免把 ENCRYPTION_KEY 直接下发到 VPS（aether-proxy）。
+        """
+        if not self.encryption_key:
+            return ""
+        return hmac.new(
+            self.encryption_key.encode("utf-8"),
+            b"aether-proxy-hmac-key-v1",
+            hashlib.sha256,
+        ).hexdigest()
+
    def _auto_pool_size(self) -> int:
        """
        智能计算连接池大小 - 根据 Worker 数量和 PostgreSQL 限制计算
--- a/src/core/exceptions.py
+++ b/src/core/exceptions.py
@@ -209,6 +209,17 @@ class ProviderNotAvailableException(ProviderException):
        self.upstream_response = upstream_response


+class ProxyNodeUnavailableError(ProviderException):
+    """代理节点不可用（ProxyNode 离线/不存在/不健康）"""
+
+    def __init__(self, message: str, node_id: str | None = None):
+        super().__init__(
+            message=message,
+            provider_name=None,
+            proxy_node_id=node_id,
+        )
+
+
 class ProviderTimeoutException(ProviderException):
    """提供商请求超时"""

--- a/src/models/admin_requests.py
+++ b/src/models/admin_requests.py
@@ -10,7 +10,7 @@ import re
 from datetime import datetime
 from typing import Any

-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator

 from src.core.enums import ProviderBillingType

@@ -18,18 +18,26 @@ from src.core.enums import ProviderBillingType
 class ProxyConfig(BaseModel):
    """代理配置"""

-    url: str = Field(..., description="代理 URL (http://, https://, socks5://)")
+    # 模式 1: 手动配置代理 URL（原有）
+    url: str | None = Field(None, description="代理 URL (http://, https://, socks5://)")
    username: str | None = Field(None, max_length=255, description="代理用户名")
    password: str | None = Field(None, max_length=500, description="代理密码")
+    # 模式 2: ProxyNode（aether-proxy 注册的节点）
+    node_id: str | None = Field(None, description="代理节点 ID")
    enabled: bool = Field(True, description="是否启用代理（false 时保留配置但不使用）")

    @field_validator("url")
    @classmethod
-    def validate_proxy_url(cls, v: str) -> str:
+    def validate_proxy_url(cls, v: str | None) -> str | None:
        """验证代理 URL 格式"""
+        if v is None:
+            return None
+
        from urllib.parse import urlparse

        v = v.strip()
+        if not v:
+            return None

        # 检查禁止的字符（防止注入）
        if "\n" in v or "\r" in v:
@@ -50,6 +58,27 @@ class ProxyConfig(BaseModel):

        return v

+    @field_validator("node_id")
+    @classmethod
+    def validate_node_id(cls, v: str | None) -> str | None:
+        if v is None:
+            return None
+        v = v.strip()
+        return v or None
+
+    @model_validator(mode="after")
+    def validate_proxy_mode(self) -> "ProxyConfig":
+        if not self.enabled:
+            return self
+
+        if not self.url and not self.node_id:
+            raise ValueError("启用代理时，必须提供 url 或 node_id")
+
+        if self.url and self.node_id:
+            raise ValueError("url 和 node_id 不能同时设置")
+
+        return self
+

 class CreateProviderRequest(BaseModel):
    """创建 Provider 请求"""
--- a/src/models/database.py
+++ b/src/models/database.py
@@ -792,6 +792,63 @@ class ProviderEndpoint(Base):
    )


+class ProxyNodeStatus(PyEnum):
+    """代理节点状态"""
+
+    ONLINE = "online"
+    UNHEALTHY = "unhealthy"
+    OFFLINE = "offline"
+
+
+class ProxyNode(Base):
+    """代理节点表（用于 aether-proxy 注册/心跳）"""
+
+    __tablename__ = "proxy_nodes"
+
+    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()), index=True)
+    name = Column(String(100), nullable=False)  # 节点名
+    ip = Column(String(45), nullable=False)  # 公网 IP（IPv6 最长 39 + 冗余）
+    port = Column(Integer, nullable=False)  # 代理端口
+    region = Column(String(100), nullable=True)  # 区域标签
+
+    status = Column(
+        Enum(
+            ProxyNodeStatus,
+            name="proxynodestatus",
+            create_type=False,
+            values_callable=lambda x: [e.value for e in x],
+        ),
+        default=ProxyNodeStatus.ONLINE,
+        nullable=False,
+    )
+
+    registered_by = Column(
+        String(36),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+        comment="注册该节点的管理员用户 ID（可空）",
+    )
+    last_heartbeat_at = Column(DateTime(timezone=True), nullable=True)
+    heartbeat_interval = Column(Integer, default=30, nullable=False)
+
+    # 性能指标（心跳上报）
+    active_connections = Column(Integer, default=0, nullable=False)
+    total_requests = Column(BigInteger, default=0, nullable=False)
+    avg_latency_ms = Column(Float, nullable=True)
+
+    created_at = Column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+        nullable=False,
+    )
+
+    __table_args__ = (UniqueConstraint("ip", "port", name="uq_proxy_node_ip_port"),)
+
+
 class GlobalModel(Base):
    """全局统一模型定义 - 包含价格和能力配置

--- a/src/modules/init.py
+++ b/src/modules/init.py
@@ -11,6 +11,7 @@ from src.modules.gemini_files import gemini_files_module
 from src.modules.ldap import ldap_module
 from src.modules.management_tokens import management_tokens_module
 from src.modules.oauth import oauth_module
+from src.modules.proxy_nodes import proxy_nodes_module

 # 所有模块列表
 ALL_MODULES: list[ModuleDefinition] = [
@@ -18,6 +19,7 @@ ALL_MODULES: list[ModuleDefinition] = [
    oauth_module,
    gemini_files_module,
    management_tokens_module,
+    proxy_nodes_module,
 ]

 __all__ = ["ALL_MODULES"]
--- a/src/modules/proxy_nodes/init.py
+++ b/src/modules/proxy_nodes/init.py
@@ -0,0 +1,112 @@
+"""
+代理节点模块
+
+提供海外 VPS 代理节点的注册、心跳、管理功能。
+aether-proxy 部署在海外 VPS 上自动注册节点，Aether 通过 HMAC 签名认证转发请求。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from src.core.modules.base import (
+    ModuleCategory,
+    ModuleDefinition,
+    ModuleHealth,
+    ModuleMetadata,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+
+def _get_router() -> Any:
+    """延迟导入路由"""
+    from src.api.admin.proxy_nodes import router
+
+    return router
+
+
+async def _on_startup() -> None:
+    """启动心跳检测调度器"""
+    import logging
+
+    from src.services.proxy_node.health_scheduler import get_proxy_node_health_scheduler
+    from src.utils.task_coordinator import StartupTaskCoordinator
+
+    logger = logging.getLogger("aether.modules.proxy_nodes")
+
+    from src.clients import get_redis_client
+
+    redis_client = await get_redis_client()
+    task_coordinator = StartupTaskCoordinator(redis_client)
+
+    proxy_node_health_scheduler = get_proxy_node_health_scheduler()
+    active = await task_coordinator.acquire("proxy_node_health")
+    if active:
+        logger.info("启动 ProxyNode 心跳检测调度器...")
+        await proxy_node_health_scheduler.start()
+    else:
+        logger.info("检测到其他 worker 已运行 ProxyNode 心跳检测，本实例跳过")
+
+
+async def _on_shutdown() -> None:
+    """停止心跳检测调度器"""
+    import logging
+
+    from src.services.proxy_node.health_scheduler import get_proxy_node_health_scheduler
+    from src.utils.task_coordinator import StartupTaskCoordinator
+
+    logger = logging.getLogger("aether.modules.proxy_nodes")
+
+    from src.clients import get_redis_client
+
+    redis_client = await get_redis_client()
+    task_coordinator = StartupTaskCoordinator(redis_client)
+
+    scheduler = get_proxy_node_health_scheduler()
+    if scheduler.running:
+        logger.info("停止 ProxyNode 心跳检测调度器...")
+        await scheduler.stop()
+        await task_coordinator.release("proxy_node_health")
+
+
+async def _health_check() -> ModuleHealth:
+    """健康检查 - 检查是否有在线节点"""
+    return ModuleHealth.HEALTHY
+
+
+def _validate_config(db: Session) -> tuple[bool, str]:
+    """
+    验证配置
+
+    代理节点模块需要 PROXY_HMAC_KEY 配置
+    """
+    from src.config.settings import config
+
+    if not config.proxy_hmac_key:
+        return False, "PROXY_HMAC_KEY 未配置（也未设置 ENCRYPTION_KEY 用于自动派生）"
+    return True, ""
+
+
+proxy_nodes_module = ModuleDefinition(
+    metadata=ModuleMetadata(
+        name="proxy_nodes",
+        display_name="代理节点",
+        description="海外 VPS 代理节点管理，通过 HMAC 签名认证转发 API 请求",
+        category=ModuleCategory.INTEGRATION,
+        env_key="PROXY_NODES_AVAILABLE",
+        default_available=True,
+        required_packages=[],
+        api_prefix="/api/admin/proxy-nodes",
+        admin_route="/admin/proxy-nodes",
+        admin_menu_icon="Server",
+        admin_menu_group="system",
+        admin_menu_order=60,
+    ),
+    router_factory=_get_router,
+    on_startup=_on_startup,
+    on_shutdown=_on_shutdown,
+    health_check=_health_check,
+    validate_config=_validate_config,
+)
--- a/src/services/proxy_node/init.py
+++ b/src/services/proxy_node/init.py
@@ -0,0 +1,5 @@
+"""Proxy node services."""
+
+from .health_scheduler import ProxyNodeHealthScheduler, get_proxy_node_health_scheduler
+
+__all__ = ["ProxyNodeHealthScheduler", "get_proxy_node_health_scheduler"]
--- a/src/services/proxy_node/health_scheduler.py
+++ b/src/services/proxy_node/health_scheduler.py
@@ -0,0 +1,103 @@
+"""
+ProxyNode 心跳检测调度器
+
+定期检查 proxy_nodes 的 last_heartbeat_at，更新节点状态：
+- elapsed > interval * 3  -> unhealthy
+- elapsed > interval * 10 -> offline
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from src.core.logger import logger
+from src.database import create_session
+from src.models.database import ProxyNode, ProxyNodeStatus
+from src.services.system.scheduler import get_scheduler
+
+
+class ProxyNodeHealthScheduler:
+    """代理节点心跳检测调度器"""
+
+    def __init__(self) -> None:
+        self.running = False
+
+    async def start(self) -> Any:
+        if self.running:
+            logger.warning("ProxyNodeHealthScheduler already running")
+            return
+
+        self.running = True
+        logger.info("ProxyNodeHealthScheduler started")
+
+        scheduler = get_scheduler()
+        scheduler.add_interval_job(
+            self._scheduled_check,
+            seconds=30,
+            job_id="proxy_node_health_check",
+            name="代理节点心跳检测",
+        )
+
+        # 启动时立即执行一次
+        await self._check_heartbeats()
+
+    async def stop(self) -> Any:
+        if not self.running:
+            return
+        self.running = False
+        logger.info("ProxyNodeHealthScheduler stopped")
+
+    async def _scheduled_check(self) -> None:
+        await self._check_heartbeats()
+
+    async def _check_heartbeats(self) -> None:
+        db = create_session()
+        try:
+            now = datetime.now(timezone.utc)
+            nodes = db.query(ProxyNode).filter(ProxyNode.status != ProxyNodeStatus.OFFLINE).all()
+            if not nodes:
+                return
+
+            changed = 0
+            for node in nodes:
+                interval = int(node.heartbeat_interval or 30)
+                last = node.last_heartbeat_at
+
+                if last is None:
+                    new_status = ProxyNodeStatus.OFFLINE
+                else:
+                    elapsed = (now - last).total_seconds()
+                    if elapsed > interval * 10:
+                        new_status = ProxyNodeStatus.OFFLINE
+                    elif elapsed > interval * 3:
+                        new_status = ProxyNodeStatus.UNHEALTHY
+                    else:
+                        new_status = ProxyNodeStatus.ONLINE
+
+                if node.status != new_status:
+                    node.status = new_status
+                    node.updated_at = now
+                    changed += 1
+
+            if changed:
+                db.commit()
+                logger.info("ProxyNode 心跳状态已更新: {} 个节点", changed)
+        except Exception as e:
+            try:
+                db.rollback()
+            except Exception:
+                pass
+            logger.exception("ProxyNode 心跳检测失败: {}", e)
+        finally:
+            db.close()
+
+
+_proxy_node_health_scheduler: ProxyNodeHealthScheduler | None = None
+
+
+def get_proxy_node_health_scheduler() -> ProxyNodeHealthScheduler:
+    global _proxy_node_health_scheduler
+    if _proxy_node_health_scheduler is None:
+        _proxy_node_health_scheduler = ProxyNodeHealthScheduler()
+    return _proxy_node_health_scheduler
--- a/src/services/task/service.py
+++ b/src/services/task/service.py
@@ -704,6 +704,7 @@ class TaskService:
        from src.core.exceptions import (
            ConcurrencyLimitError,
            EmbeddedErrorException,
+            ProxyNodeUnavailableError,
            ThinkingSignatureException,
            UpstreamClientException,
        )
@@ -744,6 +745,20 @@ class TaskService:
            )
            return "break"

+        if isinstance(cause, ProxyNodeUnavailableError):
+            # ProxyNode 不可用属于“配置明确指定但不可达/不可用”的情况，
+            # 在当前候选上重试通常没有意义，直接切换到下一个候选更合理。
+            logger.warning("  [{}] 代理节点不可用，切换候选: {}", request_id, str(cause))
+            RequestCandidateService.mark_candidate_failed(
+                db=self.db,
+                candidate_id=candidate_record_id,
+                error_type=type(cause).__name__,
+                error_message=extract_error_message(cause),
+                latency_ms=elapsed_ms,
+                concurrent_requests=captured_key_concurrent,
+            )
+            return "break"
+
        if isinstance(cause, EmbeddedErrorException):
            error_message = cause.error_message or ""
            embedded_status = cause.error_code or 200